ndb dd

  - add resource limit to buddy allocator


storage/ndb/src/kernel/SimBlockList.cpp:
  Add emulator data to args
storage/ndb/src/kernel/blocks/record_types.hpp:
  Update record types
storage/ndb/src/kernel/main.cpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/Emulator.cpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/Emulator.hpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/Makefile.am:
  Add emulator data to args
storage/ndb/src/kernel/vm/Pool.hpp:
  Update
storage/ndb/src/kernel/vm/SimBlockList.hpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/SimulatedBlock.hpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp:
  use resource_limit in buddy
storage/ndb/src/kernel/vm/ndbd_malloc_impl.hpp:
  Add emulator data to args
storage/ndb/src/kernel/vm/Pool.cpp:
  New BitKeeper file ``storage/ndb/src/kernel/vm/Pool.cpp''

12 files changed, 476 insertions, 145 deletions

diff --git a/storage/ndb/src/kernel/SimBlockList.cpp b/storage/ndb/src/kernel/SimBlockList.cpp
index 0f0439e9773..8467690966c 100644
--- a/storage/ndb/src/kernel/SimBlockList.cpp
+++ b/storage/ndb/src/kernel/SimBlockList.cpp
@@ -15,6 +15,7 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 #include "SimBlockList.hpp"
+#include "EmulatorData.hpp"
 #include <SimulatedBlock.hpp>
 #include <Cmvmi.hpp>
 #include <Ndbfs.hpp>
@@ -69,7 +70,7 @@ void * operator new (size_t sz, SIMBLOCKLIST_DUMMY dummy){
 #endif
 
 void 
-SimBlockList::load(Configuration & conf){
+SimBlockList::load(EmulatorData& data){
   noOfBlocks = NO_OF_BLOCKS;
   theList = new SimulatedBlock * [noOfBlocks];
   Dbdict* dbdict = 0;
@@ -79,7 +80,7 @@ SimBlockList::load(Configuration & conf){
   Tsman* ts = 0;
 
   Block_context ctx = 
-    { conf, * (Ndbd_mem_manager*)0 };
+    { *data.theConfiguration, *data.m_mem_manager };
   
   SimulatedBlock * fs = 0;
   {
diff --git a/storage/ndb/src/kernel/blocks/record_types.hpp b/storage/ndb/src/kernel/blocks/record_types.hpp
index 6d2857bd4e2..bb8da685b05 100644
--- a/storage/ndb/src/kernel/blocks/record_types.hpp
+++ b/storage/ndb/src/kernel/blocks/record_types.hpp
@@ -18,17 +18,6 @@
 #define KERNEL_RECORDS_HPP
 
 /**
- * Record types
- */
-#define PGMAN_PAGE_REQUEST      1
-
-#define LGMAN_LOG_BUFFER_WAITER 2
-#define LGMAN_LOG_SYNC_WAITER   3
-
-#define DBTUP_PAGE_REQUEST      4
-#define DBTUP_EXTENT_INFO       5
-
-/**
  * Resource groups
  */
 
@@ -47,4 +36,21 @@
  */
 #define RG_DISK_RECORDS         2
 
+/**
+ * 
+ */
+#define RG_RESERVED             0
+#define RG_COUNT                3
+
+/**
+ * Record types
+ */
+#define PGMAN_PAGE_REQUEST      MAKE_TID(1, RG_DISK_OPERATIONS)
+
+#define LGMAN_LOG_BUFFER_WAITER MAKE_TID(2, RG_DISK_OPERATIONS)
+#define LGMAN_LOG_SYNC_WAITER   MAKE_TID(3, RG_DISK_OPERATIONS)
+
+#define DBTUP_PAGE_REQUEST      MAKE_TID(4, RG_DISK_OPERATIONS)
+#define DBTUP_EXTENT_INFO       MAKE_TID(5, RG_DISK_RECORDS)
+
 #endif
diff --git a/storage/ndb/src/kernel/main.cpp b/storage/ndb/src/kernel/main.cpp
index 7c1763485ce..b380a4e74c3 100644
--- a/storage/ndb/src/kernel/main.cpp
+++ b/storage/ndb/src/kernel/main.cpp
@@ -406,7 +406,7 @@ int main(int argc, char** argv)
   systemInfo(* theConfig, * theConfig->m_logLevel); 
   
     // Load blocks
-  globalEmulatorData.theSimBlockList->load(* theConfig);
+  globalEmulatorData.theSimBlockList->load(globalEmulatorData);
     
   // Set thread concurrency for Solaris' light weight processes
   int status;
diff --git a/storage/ndb/src/kernel/vm/Emulator.cpp b/storage/ndb/src/kernel/vm/Emulator.cpp
index 2105b7ddb5e..d05fc5fa97c 100644
--- a/storage/ndb/src/kernel/vm/Emulator.cpp
+++ b/storage/ndb/src/kernel/vm/Emulator.cpp
@@ -28,6 +28,7 @@
 #include "SimBlockList.hpp"
 
 #include <NodeState.hpp>
+#include "ndbd_malloc_impl.hpp"
 
 #include <NdbMem.h>
 #include <NdbMutex.h>
@@ -77,6 +78,7 @@ EmulatorData::EmulatorData(){
   theSimBlockList  = 0;
   theShutdownMutex = 0;
   m_socket_server = 0;
+  m_mem_manager = 0;
 }
 
 void
@@ -93,6 +95,7 @@ EmulatorData::create(){
   theThreadConfig  = new ThreadConfig();
   theSimBlockList  = new SimBlockList();
   m_socket_server  = new SocketServer();
+  m_mem_manager    = new Ndbd_mem_manager();
 
   theShutdownMutex = NdbMutex_Create();
 
@@ -111,6 +114,9 @@ EmulatorData::destroy(){
     delete theSimBlockList; theSimBlockList = 0;
   if(m_socket_server)
     delete m_socket_server; m_socket_server = 0;
+  if (m_mem_manager)
+    delete m_mem_manager; m_mem_manager = 0;
+  
   NdbMem_Destroy();
 }
 
diff --git a/storage/ndb/src/kernel/vm/Emulator.hpp b/storage/ndb/src/kernel/vm/Emulator.hpp
index cd194202d85..f8f55406ffc 100644
--- a/storage/ndb/src/kernel/vm/Emulator.hpp
+++ b/storage/ndb/src/kernel/vm/Emulator.hpp
@@ -56,6 +56,7 @@ struct EmulatorData {
   class ThreadConfig  * theThreadConfig;
   class SimBlockList  * theSimBlockList;
   class SocketServer  * m_socket_server;
+  class Ndbd_mem_manager * m_mem_manager;
 
   /**
    * Constructor
diff --git a/storage/ndb/src/kernel/vm/Makefile.am b/storage/ndb/src/kernel/vm/Makefile.am
index 64739b7459e..740e6783fcd 100644
--- a/storage/ndb/src/kernel/vm/Makefile.am
+++ b/storage/ndb/src/kernel/vm/Makefile.am
@@ -19,7 +19,7 @@ libkernel_a_SOURCES = \
         SectionReader.cpp \
         Mutex.cpp SafeCounter.cpp \
         Rope.cpp \
-	ndbd_malloc.cpp ndbd_malloc_impl.cpp
+	ndbd_malloc.cpp ndbd_malloc_impl.cpp Pool.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/mgmapi
 
diff --git a/storage/ndb/src/kernel/vm/Pool.cpp b/storage/ndb/src/kernel/vm/Pool.cpp
new file mode 100644
index 00000000000..e64c18d4b4f
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/Pool.cpp
@@ -0,0 +1,37 @@
+/* Copyright (C) 2003 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+
+#include "Pool.hpp"
+#include "SimulatedBlock.hpp"
+
+void*
+Pool_context::alloc_page(Uint32 type_id, Uint32 *i)
+{
+  return m_block->m_ctx.m_mm.alloc_page(type_id, i);
+}
+  
+void 
+Pool_context::release_page(Uint32 type_id, Uint32 i, void* p)
+{
+  m_block->m_ctx.m_mm.release_page(type_id, i, p);
+}
+
+void
+Pool_context::handle_abort(const AbortArg &)
+{
+  
+}
diff --git a/storage/ndb/src/kernel/vm/Pool.hpp b/storage/ndb/src/kernel/vm/Pool.hpp
index 52f7df0ead0..fb5e2961a99 100644
--- a/storage/ndb/src/kernel/vm/Pool.hpp
+++ b/storage/ndb/src/kernel/vm/Pool.hpp
@@ -19,6 +19,22 @@
 
 #include <kernel_types.h>
 
+/**
+ * Type id is 11 bits record type, and 5 bits resource id
+ *   -> 2048 different kind of records and 32 different resource groups
+ * 
+ * Resource id is used to handle configuration parameters
+ *
+ * see blocks/records_types.hpp
+ */
+#define RG_BITS 5
+#define RG_MASK ((1 << RG_BITS) - 1)
+#define MAKE_TID(TID,RG) ((TID << RG_BITS) | RG)
+
+/**
+ * Record_info
+ *
+ */
 struct Record_info
 {
   Uint16 m_size;
@@ -27,6 +43,9 @@ struct Record_info
   Uint16 m_offset_magic;
 };
 
+/**
+ * Resource_limit
+ */
 struct Resource_limit
 {
   Uint32 m_min;
@@ -38,7 +57,6 @@ struct Resource_limit
 struct Pool_context
 {
   class SimulatedBlock* m_block;
-  struct Resource_limit* m_resource_limit;
   
   /**
    * Alloc consekutive pages
@@ -48,7 +66,7 @@ struct Pool_context
    *
    * Will handle resource limit 
    */
-  void* alloc_page(Uint32 *i);
+  void* alloc_page(Uint32 type_id, Uint32 *i);
   
   /**
    * Release pages
@@ -56,7 +74,7 @@ struct Pool_context
    *   @param i   : in : i value of first page
    *   @param p   : in : pointer to first page
    */
-  void release_page(Uint32 i, void* p);
+  void release_page(Uint32 type_id, Uint32 i, void* p);
 
   /**
    * Alloc consekutive pages
@@ -70,7 +88,7 @@ struct Pool_context
    *
    * Will handle resource limit 
    */
-  void* alloc_pages(Uint32 *i, Uint32 *cnt, Uint32 min = 1);
+  void* alloc_pages(Uint32 type_id, Uint32 *i, Uint32 *cnt, Uint32 min =1);
   
   /**
    * Release pages
@@ -79,7 +97,7 @@ struct Pool_context
    *   @param p   : in : pointer to first page
    *   @param cnt : in : no of pages to release
    */
-  void release_pages(Uint32 i, void* p, Uint32 cnt);
+  void release_pages(Uint32 type_id, Uint32 i, void* p, Uint32 cnt);
   
   /**
    * Pool abort
diff --git a/storage/ndb/src/kernel/vm/SimBlockList.hpp b/storage/ndb/src/kernel/vm/SimBlockList.hpp
index 3204a50b03d..8ba4ba4e2d1 100644
--- a/storage/ndb/src/kernel/vm/SimBlockList.hpp
+++ b/storage/ndb/src/kernel/vm/SimBlockList.hpp
@@ -19,7 +19,7 @@
 
 #include <SimulatedBlock.hpp>
 
-class Configuration;
+class EmulatorData;
 
 class SimBlockList 
 {
@@ -27,7 +27,7 @@ public:
   SimBlockList();
   ~SimBlockList();
   
-  void load(Configuration & conf);
+  void load(EmulatorData&);
   void unload();
 private:
   int noOfBlocks;
diff --git a/storage/ndb/src/kernel/vm/SimulatedBlock.hpp b/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
index 477e75349eb..f71f79b777a 100644
--- a/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
+++ b/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
@@ -91,6 +91,7 @@ class SimulatedBlock {
   friend class Page_cache_client;
   friend class Lgman;
   friend class Logfile_client;
+  friend struct Pool_context;
 public:
   friend class BlockComponent;
   virtual ~SimulatedBlock();
diff --git a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
index e78a1cc1194..a6067528db8 100644
--- a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
+++ b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
@@ -18,6 +18,92 @@
 
 #include "ndbd_malloc_impl.hpp"
 #include <ndb_global.h>
+#include <EventLogger.hpp>
+
+#ifndef UNIT_TEST
+extern EventLogger g_eventLogger;
+#else
+extern EventLogger g_eventLogger;
+#endif
+
+#ifdef NDBD_MALLOC_METHOD
+#if NDBD_MALLOC_METHOD == sbrk
+static const char * f_method = "sbrk";
+#else
+static const char * f_method = "malloc";
+#endif
+#elif SIZEOF_CHARP == 8
+static const char * f_method = "sbrk";
+#else
+static const char * f_method = "malloc";
+#endif
+#define MAX_CHUNKS 10
+
+struct InitChunk
+{
+  Uint32 m_cnt;
+  Uint32 m_start;
+  Alloc_page* m_ptr;
+};
+
+#include <NdbOut.hpp>
+
+static
+bool
+do_malloc(Uint32 pages, InitChunk* chunk)
+{
+  void * ptr;
+  pages += 1;
+  Uint32 sz = pages;
+  if (strcmp(f_method, "sbrk") == 0)
+  {
+    ptr = 0;
+    while (ptr == 0)
+    {
+      ptr = sbrk(sizeof(Alloc_page) * sz);
+      if (ptr == (void*)-1)
+      {
+	ptr = 0;
+	sz = 1 + (9 * sz) / 10;
+	if (pages >= 32 && sz < 32)
+	{
+	  sz = pages;
+	  f_method = "malloc";
+	  g_eventLogger.info("sbrk(%lld) failed, trying malloc",
+			     (Uint64)(sizeof(Alloc_page) * sz));
+	  break;
+	}
+      }
+    }
+  }
+  if (strcmp(f_method, "malloc") == 0)
+  {
+    ptr = 0;
+    while (ptr == 0)
+    {
+      ptr = malloc(sizeof(Alloc_page) * sz);
+      if (ptr == 0)
+      {
+	sz = 1 + (9 * sz) / 10;
+	if (pages >= 32 && sz < 32)
+	{
+	  return false;
+	}
+      }
+    }
+  }
+  
+  chunk->m_cnt = sz;
+  chunk->m_ptr = (Alloc_page*)ptr;
+  const UintPtr align = sizeof(Alloc_page) - 1;
+  if (UintPtr(ptr) & align)
+  {
+    chunk->m_cnt--;
+    chunk->m_ptr = (Alloc_page*)((UintPtr(ptr) + align) & ~align);
+  }
+  
+  return true;
+}
 
 Uint32
 Ndbd_mem_manager::log2(Uint32 input)
@@ -33,67 +119,196 @@ Ndbd_mem_manager::log2(Uint32 input)
   return output;
 }
 
-Ndbd_mem_manager::Ndbd_mem_manager(Uint32 default_grow)
+Ndbd_mem_manager::Ndbd_mem_manager()
 {
-  m_grow_size = default_grow;
-  bzero(m_buddy_lists, sizeof(m_buddy_lists));
-  m_base = 0;
   m_base_page = 0;
-  
-  m_pages_alloc = 0;
-  m_pages_used = 0;
+  bzero(m_buddy_lists, sizeof(m_buddy_lists));
+  bzero(m_resource_limit, sizeof(m_resource_limit));
   
   if (sizeof(Free_page_data) != (4 * (1 << FPD_2LOG)))
   {
+    g_eventLogger.error("Invalid build, ndbd_malloc_impl.cpp:%d", __LINE__);
     abort();
   }
 }
 
+void
+Ndbd_mem_manager::set_resource_limit(const Resource_limit& rl)
+{
+  Uint32 id = rl.m_resource_id;
+  assert(id < XX_RL_COUNT);
+  
+  Uint32 reserve = id ? rl.m_min : 0;
+  Uint32 current_reserved = m_resource_limit[0].m_min;
+  
+  m_resource_limit[id] = rl;
+  m_resource_limit[id].m_curr = 0;
+  m_resource_limit[0].m_min = current_reserved + reserve;
+}
+
 bool
-Ndbd_mem_manager::init(Uint32 pages)
+Ndbd_mem_manager::init(bool alloc_less_memory)
 {
-  assert(m_base == 0);
   assert(m_base_page == 0);
-  assert(m_pages_alloc == 0);
-  pages = pages ? pages : m_grow_size;
+
+  Uint32 pages = 0;
+  Uint32 max_page = 0;
+  if (m_resource_limit[0].m_max)
+  {
+    pages = m_resource_limit[0].m_max;
+  } 
+  else
+  {
+    pages = m_resource_limit[0].m_min; // reserved
+  }
+  assert(pages);
+  
+  g_eventLogger.info("Ndbd_mem_manager::init(%d) min: %dMb initial: %dMb",
+		     alloc_less_memory,
+		     (sizeof(Alloc_page)*m_resource_limit[0].m_min)>>20,
+		     (sizeof(Alloc_page)*m_resource_limit[0].m_max)>>20);
+  
+  /**
+   * Do malloc
+   */
+
+  Uint32 cnt = 0;
+  struct InitChunk chunks[MAX_CHUNKS];
+  bzero(chunks, sizeof(chunks));
+
+  Uint32 allocated = 0;
+  while (cnt < MAX_CHUNKS && allocated < pages)
+  {
+    InitChunk chunk; 
+    Uint32 remaining = pages - allocated;
+    
+    if (do_malloc(pages - allocated, &chunk))
+    {
+      Uint32 i = 0;
+      for(; i<cnt ; i++)
+      {
+	if (chunk.m_ptr < chunks[i].m_ptr)
+	{
+	  for (Uint32 j = cnt; j > i; j--)
+	    chunks[j] = chunks[j-1];
+	  break;
+	}
+      }
+      cnt++;
+      chunks[i] = chunk;
+      allocated += chunk.m_cnt;
+    }
+    else
+    {
+      break;
+    }
+  }
   
-  m_base = malloc((2 + pages) * sizeof(Alloc_page));
-  UintPtr ptr = (UintPtr)m_base;
-  UintPtr rem = ptr % sizeof(Alloc_page);
-  if (rem)
+  if (allocated < m_resource_limit[0].m_min)
   {
-    ptr += sizeof(Alloc_page) - rem;
+    g_eventLogger.
+      error("Unable to alloc min memory from OS: min: %lldMb "
+	    " allocated: %lldMb", 
+	    (Uint64)(sizeof(Alloc_page)*m_resource_limit[0].m_min) >> 20,
+	    (Uint64)(sizeof(Alloc_page)*allocated) >> 20);
+    return false;
   }
-  else
+  else if (allocated < pages)
   {
-    pages++;
+    g_eventLogger.
+      warning("Unable to alloc requested memory from OS: min: %lldMb"
+	      " requested: %lldMb allocated: %lldMb",
+	      (Uint64)(sizeof(Alloc_page)*m_resource_limit[0].m_min)>>20,
+	      (Uint64)(sizeof(Alloc_page)*m_resource_limit[0].m_max)>>20,
+	      (Uint64)(sizeof(Alloc_page)*allocated)>>20);
+    if (!alloc_less_memory)
+      return false;
   }
-  m_base_page = (Alloc_page*)ptr;
-  m_pages_alloc += pages;
-  m_pages_used += pages;
+  
+  m_base_page = chunks[0].m_ptr;
+  
+  for (Uint32 i = 0; i<cnt; i++)
+  {
+    UintPtr start = UintPtr(chunks[i].m_ptr) - UintPtr(m_base_page);
+    start >>= (2 + BMW_2LOG);
+    UintPtr last = start + chunks[i].m_cnt;
+    chunks[i].m_start = start;
+    assert((Uint64(start) >> 32) == 0);
+    
+    if (last > max_page)
+      max_page = last;
+  }
+  m_resource_limit[0].m_resource_id = max_page;
+  
+  for (Uint32 i = 0; i<cnt; i++)
+  {
+    grow(chunks[i].m_start, chunks[i].m_cnt);
+  }
+  
+  return true;
+}
+
+#include <NdbOut.hpp>
 
-  Uint32 bmp = (pages + (1 << BPP_2LOG) - 1) >> BPP_2LOG;
-  for(Uint32 i = 0; i < bmp; i++)
+void
+Ndbd_mem_manager::grow(Uint32 start, Uint32 cnt)
+{
+  assert(cnt);
+  Uint32 start_bmp = start >> BPP_2LOG;
+  Uint32 last_bmp = (start + cnt - 1) >> BPP_2LOG;
+  
+#if SIZEOF_CHARP == 4
+  assert(start_bmp == 0 && last_bmp == 0);
+#endif
+  
+  if (start_bmp != last_bmp)
+  {
+    Uint32 tmp = ((start_bmp + 1) << BPP_2LOG) - start;
+    grow(start, tmp);
+    grow((start_bmp + 1) << BPP_2LOG, cnt - tmp);
+    return;
+  }
+  
+  if ((start + cnt) == ((start_bmp + 1) << BPP_2LOG))
+  {
+    cnt--; // last page is always marked as empty
+  }
+  
+  if (!m_used_bitmap_pages.get(start_bmp))
   {
-    Uint32 start = i * (1 << BPP_2LOG);
-    Uint32 end = start + (1 << BPP_2LOG);
-    end = end > m_pages_alloc ? m_pages_alloc : end - 1;
-    Alloc_page *ptr = m_base_page + start;
-    BitmaskImpl::clear(BITMAP_WORDS, ptr->m_data);
+    if (start != (start_bmp << BPP_2LOG))
+    {
+      ndbout_c("ndbd_malloc_impl.cpp:%d:grow(%d, %d) %d!=%d"
+	       " - Unable to use due to bitmap pages missaligned!!",
+	       __LINE__, start, cnt, start, (start_bmp << BPP_2LOG));
+      g_eventLogger.error("ndbd_malloc_impl.cpp:%d:grow(%d, %d)"
+			  " - Unable to use due to bitmap pages missaligned!!",
+			  __LINE__, start, cnt);
+      return;
+    }
     
-    release(start+1, end - 1 - start);    
+    Alloc_page* bmp = m_base_page + start;
+    memset(bmp, 0, sizeof(Alloc_page));
+    m_used_bitmap_pages.set(start_bmp);
+    cnt--;
+    start++;
   }
 
-  return true;
+  if (cnt)
+  {
+    m_resource_limit[0].m_curr += cnt;
+    m_resource_limit[0].m_max += cnt;
+    release(start, cnt);
+  }
 }
 
 void
 Ndbd_mem_manager::release(Uint32 start, Uint32 cnt)
 {
-  assert(m_pages_used >= cnt);
+  assert(m_resource_limit[0].m_curr >= cnt);
   assert(start);
-  m_pages_used -= cnt;
-
+  m_resource_limit[0].m_curr -= cnt;
+  
   set(start, start+cnt-1);
   
   release_impl(start, cnt);
@@ -131,7 +346,8 @@ Ndbd_mem_manager::release_impl(Uint32 start, Uint32 cnt)
 void
 Ndbd_mem_manager::alloc(Uint32* ret, Uint32 *pages, Uint32 min)
 {
-  Uint32 start, i;
+  Int32 i;
+  Uint32 start;
   Uint32 cnt = * pages;
   Uint32 list = log2(cnt - 1);
   
@@ -160,8 +376,8 @@ Ndbd_mem_manager::alloc(Uint32* ret, Uint32 *pages, Uint32 min)
 	clear(start, start+cnt-1);
       }
       * ret = start;
-      m_pages_used += cnt;
-      assert(m_pages_used <= m_pages_alloc);
+      m_resource_limit[0].m_curr += cnt;
+      assert(m_resource_limit[0].m_curr <= m_resource_limit[0].m_max);
       return;
     }
   }
@@ -171,7 +387,7 @@ Ndbd_mem_manager::alloc(Uint32* ret, Uint32 *pages, Uint32 min)
    *   search in other lists...
    */
 
-  Uint32 min_list = log2(min - 1);
+  Int32 min_list = log2(min - 1);
   assert(list >= min_list);
   for (i = list - 1; i >= min_list; i--) 
   {
@@ -192,33 +408,14 @@ Ndbd_mem_manager::alloc(Uint32* ret, Uint32 *pages, Uint32 min)
 
       * ret = start;
       * pages = sz;
-      m_pages_used += sz;
-      assert(m_pages_used <= m_pages_alloc);
+      m_resource_limit[0].m_curr += sz;
+      assert(m_resource_limit[0].m_curr <= m_resource_limit[0].m_max);
       return;
     }
   }
   * pages = 0;
 }
 
-void*
-Ndbd_mem_manager::alloc(Uint32 *pages, Uint32 min)
-{
-  Uint32 ret;
-  alloc(&ret, pages, min);
-  if (pages)
-  {
-    return m_base_page + ret;
-  }
-  return 0;
-}
-
-void
-Ndbd_mem_manager::release(void* ptr, Uint32 cnt)
-{
-  Uint32 page = ((Alloc_page*)ptr) - m_base_page;
-  release(page, cnt);
-}
-
 void
 Ndbd_mem_manager::insert_free_list(Uint32 start, Uint32 size)
 {
@@ -284,25 +481,6 @@ Ndbd_mem_manager::remove_free_list(Uint32 start, Uint32 list)
   return size;
 }
 
-Uint32
-Ndbd_mem_manager::get_no_allocated_pages() const
-{
-  return m_pages_alloc;
-}
-
-Uint32
-Ndbd_mem_manager::get_no_used_pages() const
-{
-  return m_pages_used;
-}
-
-Uint32
-Ndbd_mem_manager::get_no_free_pages() const
-{
-  return m_pages_alloc - m_pages_used;
-}
-
-
 void
 Ndbd_mem_manager::dump() const
 {
@@ -321,31 +499,105 @@ Ndbd_mem_manager::dump() const
   }
 }
 
+void*
+Ndbd_mem_manager::alloc_page(Uint32 type, Uint32* i)
+{
+  Uint32 idx = type & RG_MASK;
+  assert(idx && idx < XX_RL_COUNT);
+  Resource_limit tot = m_resource_limit[0];
+  Resource_limit rl = m_resource_limit[idx];
+
+  Uint32 add = (rl.m_curr < rl.m_min) ? 0 : 1; // Over min ?
+  Uint32 limit = (rl.m_max == 0 || rl.m_curr < rl.m_max) ? 0 : 1; // Over limit
+  Uint32 free = (tot.m_min + tot.m_curr < tot.m_max) ? 1 : 0; // Has free
+  
+  if (likely(add == 0 || (limit == 0 && free == 1)))
+  {
+    Uint32 cnt = 1;
+    alloc(i, &cnt, 1);
+    assert(cnt);
+    m_resource_limit[0].m_curr = tot.m_curr + add;
+    m_resource_limit[idx].m_curr = rl.m_curr + 1;
+    return m_base_page + *i;
+  }
+  return 0;
+}
+
+void
+Ndbd_mem_manager::release_page(Uint32 type, Uint32 i, void * p)
+{
+  Uint32 idx = type & RG_MASK;
+  assert(idx && idx < XX_RL_COUNT);
+  Resource_limit tot = m_resource_limit[0];
+  Resource_limit rl = m_resource_limit[idx];
+
+  Uint32 sub = (rl.m_curr < rl.m_min) ? 0 : 1; // Over min ?
+  release(i, 1);
+  m_resource_limit[0].m_curr = tot.m_curr - sub;
+}
+
 #ifdef UNIT_TEST
 
 #include <Vector.hpp>
+#include <NdbTick.h>
 
 struct Chunk {
   Uint32 pageId;
   Uint32 pageCount;
 };
 
+struct Timer
+{
+  Uint64 sum;
+  Uint32 cnt;
+
+  Timer() { sum = cnt = 0;}
+  
+  void add(Uint64 diff) { sum += diff; cnt++;}
+
+  void print(const char * title) const {
+    printf("%s %lld %d -> %d/s\n", title, sum, cnt, Uint32(1000*cnt/sum));
+  }
+};
+
 int 
 main(void)
 {
+  char buf[255];
+  Timer timer[4];
   printf("Startar modul test av Page Manager\n");
+  g_eventLogger.createConsoleHandler();
+  g_eventLogger.setCategory("keso");
+  g_eventLogger.enable(Logger::LL_ON, Logger::LL_INFO);
+  g_eventLogger.enable(Logger::LL_ON, Logger::LL_CRITICAL);
+  g_eventLogger.enable(Logger::LL_ON, Logger::LL_ERROR);
+  g_eventLogger.enable(Logger::LL_ON, Logger::LL_WARNING);
+
 #define DEBUG 0
 
   Ndbd_mem_manager mem;
-  mem.init(32000);
+  Resource_limit rl;
+  rl.m_min = 0;
+  rl.m_max = 2*32768 + 2*16384;
+  rl.m_curr = 0;
+  rl.m_resource_id = 0;
+  mem.set_resource_limit(rl);
+  rl.m_min = 32768;
+  rl.m_max = 0;
+  rl.m_resource_id = 1;
+  mem.set_resource_limit(rl);
+  
+  mem.init();
+  mem.dump();
+  printf("pid: %d press enter to continue\n", getpid());
+  fgets(buf, sizeof(buf), stdin);
   Vector<Chunk> chunks;
-  const Uint32 LOOPS = 100000;
+  const Uint32 LOOPS = 1000000;
   for(Uint32 i = 0; i<LOOPS; i++){
     //mem.dump();
     
     // Case
     Uint32 c = (rand() % 100);
-    const Uint32 free = mem.get_no_allocated_pages() - mem.get_no_used_pages();
     if (c < 60)
     {
       c = 0;
@@ -359,17 +611,8 @@ main(void)
       c = 2;
     }
     
-    Uint32 alloc = 0;
-    if(free <= 1)
-    {
-      c = 0;
-      alloc = 1;
-    } 
-    else 
-    {
-      alloc = 1 + (rand() % (free - 1));
-    }  
-
+    Uint32 alloc = 1 + rand() % 3200;
+    
     if(chunks.size() == 0 && c == 0)
     {
       c = 1 + rand() % 2;
@@ -382,38 +625,52 @@ main(void)
       const int ch = rand() % chunks.size();
       Chunk chunk = chunks[ch];
       chunks.erase(ch);
+      Uint64 start = NdbTick_CurrentMillisecond();      
       mem.release(chunk.pageId, chunk.pageCount);
+      Uint64 stop = NdbTick_CurrentMillisecond();
+      timer[0].add(stop-start);
       if(DEBUG)
 	printf(" release %d %d\n", chunk.pageId, chunk.pageCount);
     }
       break;
     case 2: { // Seize(n) - fail
-      alloc += free;
+      alloc += 32000;
       // Fall through
     }
     case 1: { // Seize(n) (success)
       Chunk chunk;
       chunk.pageCount = alloc;
+      if (DEBUG)
+      {
+	printf(" alloc %d -> ", alloc); fflush(stdout);
+      }
+      Uint64 start = NdbTick_CurrentMillisecond();      
       mem.alloc(&chunk.pageId, &chunk.pageCount, 1);
+      Uint64 stop = NdbTick_CurrentMillisecond();      
+
       if (DEBUG)
-	printf(" alloc %d -> %d %d", alloc, chunk.pageId, chunk.pageCount);
+	printf("%d %d", chunk.pageId, chunk.pageCount);
       assert(chunk.pageCount <= alloc);
       if(chunk.pageCount != 0){
 	chunks.push_back(chunk);
 	if(chunk.pageCount != alloc) {
+	  timer[2].add(stop-start);
 	  if (DEBUG)
 	    printf(" -  Tried to allocate %d - only allocated %d - free: %d",
-		   alloc, chunk.pageCount, free);
+		   alloc, chunk.pageCount, 0);
+	}
+	else
+	{
+	  timer[1].add(stop-start);
 	}
       } else {
+	timer[3].add(stop-start);
 	if (DEBUG)
 	  printf("  Failed to alloc %d pages with %d pages free",
-		 alloc, free);
+		 alloc, 0);
       }
       if (DEBUG)
 	printf("\n");
-      if(alloc == 1 && free > 0)
-	assert(chunk.pageCount == alloc);
     }
       break;
     }
@@ -424,6 +681,15 @@ main(void)
       mem.release(chunk.pageId, chunk.pageCount);      
       chunks.erase(chunks.size() - 1);
     }
+
+  const char *title[] = {
+    "release   ",
+    "alloc full",
+    "alloc part",
+    "alloc fail"
+  };
+  for(Uint32 i = 0; i<4; i++)
+    timer[i].print(title[i]);
 }
 
 template class Vector<Chunk>;
diff --git a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.hpp b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.hpp
index 15b5c9c390e..c1b3985ba42 100644
--- a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.hpp
+++ b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.hpp
@@ -20,6 +20,7 @@
 #include <kernel_types.h>
 #include <Bitmask.hpp>
 #include <assert.h>
+#include "Pool.hpp"
 
 /**
  * 13 -> 8192 words -> 32768 bytes
@@ -51,25 +52,22 @@ struct Free_page_data
 class Ndbd_mem_manager 
 {
 public:
-  Ndbd_mem_manager(Uint32 default_grow = 32);
-
+  Ndbd_mem_manager();
+  
+  void set_resource_limit(const Resource_limit& rl);
+  
+  bool init(bool allow_alloc_less_than_requested = true);
+  void grow(Uint32 start, Uint32 cnt);
+  void* get_memroot() const { return (void*)m_base_page;}
+  
   void alloc(Uint32* ret, Uint32 *pages, Uint32 min_requested);
   void release(Uint32 start, Uint32 cnt);
-
-  Uint32 get_no_allocated_pages() const;
-  Uint32 get_no_used_pages() const;
-  Uint32 get_no_free_pages() const;
-
-  bool init(Uint32 pages = 0);
-  bool grow(Uint32 pages = 0);
-
+  
   void dump() const ;
-
-  void* get_memroot() const { return (void*)m_base_page;}
-
-  void* alloc(Uint32 * pages, Uint32 min_requested);
-  void release(void* ptr, Uint32 cnt);
-
+  
+  void* alloc_page(Uint32 type, Uint32* i);
+  void release_page(Uint32 type, Uint32 i, void * p);
+  
   /**
    * Compute 2log of size 
    * @note size = 0     -> 0
@@ -78,18 +76,15 @@ public:
   static Uint32 log2(Uint32 size);
 
 private:
+#define XX_RL_COUNT 3
   /**
    * Return pointer to free page data on page
    */
   static Free_page_data* get_free_page_data(Alloc_page*, Uint32 idx);
+  Bitmask<1> m_used_bitmap_pages;
   
-  Uint32 m_pages_alloc;
-  Uint32 m_pages_used;
-  
-  Uint32 m_grow_size;
   Uint32 m_buddy_lists[16];
-  
-  void * m_base;
+  Resource_limit m_resource_limit[XX_RL_COUNT]; // RG_COUNT in record_types.hpp
   Alloc_page * m_base_page;
   
   void release_impl(Uint32 start, Uint32 cnt);  
@@ -121,7 +116,7 @@ Ndbd_mem_manager::set(Uint32 first, Uint32 last)
 #if ((SPACE_PER_BMP_2LOG < 32) && (SIZEOF_CHARP == 4)) || (SIZEOF_CHARP == 8)
   Uint32 bmp = first & ~((1 << BPP_2LOG) - 1);
   assert((first >> BPP_2LOG) == (last >> BPP_2LOG));
-  assert(bmp < m_pages_alloc);
+  assert(bmp < m_resource_limit[0].m_resource_id);
 
   first -= bmp;
   last -= bmp;
@@ -139,7 +134,7 @@ Ndbd_mem_manager::clear(Uint32 first, Uint32 last)
 #if ((SPACE_PER_BMP_2LOG < 32) && (SIZEOF_CHARP == 4)) || (SIZEOF_CHARP == 8)
   Uint32 bmp = first & ~((1 << BPP_2LOG) - 1);
   assert((first >> BPP_2LOG) == (last >> BPP_2LOG));
-  assert(bmp < m_pages_alloc);
+  assert(bmp < m_resource_limit[0].m_resource_id);
   
   first -= bmp;
   last -= bmp;
@@ -157,7 +152,7 @@ Ndbd_mem_manager::clear_and_set(Uint32 first, Uint32 last)
 #if ((SPACE_PER_BMP_2LOG < 32) && (SIZEOF_CHARP == 4)) || (SIZEOF_CHARP == 8)
   Uint32 bmp = first & ~((1 << BPP_2LOG) - 1);
   assert((first >> BPP_2LOG) == (last >> BPP_2LOG));
-  assert(bmp < m_pages_alloc);
+  assert(bmp < m_resource_limit[0].m_resource_id);
   
   first -= bmp;
   last -= bmp;
@@ -177,7 +172,7 @@ Ndbd_mem_manager::check(Uint32 first, Uint32 last)
 #if ((SPACE_PER_BMP_2LOG < 32) && (SIZEOF_CHARP == 4)) || (SIZEOF_CHARP == 8)
   Uint32 bmp = first & ~((1 << BPP_2LOG) - 1);
   assert((first >> BPP_2LOG) == (last >> BPP_2LOG));
-  assert(bmp < m_pages_alloc);
+  assert(bmp < m_resource_limit[0].m_resource_id);
   
   first -= bmp;
   last -= bmp;