Merge perch.ndb.mysql.com:/home/jonas/src/mysql-5.1-new

into  perch.ndb.mysql.com:/home/jonas/src/51-ndb


storage/ndb/include/kernel/ndb_limits.h:
  Auto merged
storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/dbtup/DbtupIndex.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/lgman.cpp:
  Auto merged
storage/ndb/src/kernel/blocks/lgman.hpp:
  Auto merged
storage/ndb/src/kernel/vm/SimulatedBlock.hpp:
  Auto merged
storage/ndb/src/kernel/SimBlockList.cpp:
  merge

41 files changed, 2496 insertions, 591 deletions

diff --git a/storage/ndb/include/kernel/ndb_limits.h b/storage/ndb/include/kernel/ndb_limits.h
index 6bf9256bf95..5985fbe5d9b 100644
--- a/storage/ndb/include/kernel/ndb_limits.h
+++ b/storage/ndb/include/kernel/ndb_limits.h
@@ -150,10 +150,4 @@
  */
 #define LCP_RESTORE_BUFFER (4*32)
 
-/*
- * Log buffer pages
- *   8M
- */
-#define LGMAN_LOG_BUFFER (8*32)
-
 #endif
diff --git a/storage/ndb/include/kernel/signaldata/FsReadWriteReq.hpp b/storage/ndb/include/kernel/signaldata/FsReadWriteReq.hpp
index c2825a77b21..7b50bcf2d49 100644
--- a/storage/ndb/include/kernel/signaldata/FsReadWriteReq.hpp
+++ b/storage/ndb/include/kernel/signaldata/FsReadWriteReq.hpp
@@ -60,9 +60,10 @@ public:
     fsFormatArrayOfPages=1,
     fsFormatListOfMemPages=2,
     fsFormatGlobalPage=3,
+    fsFormatSharedPage=4,
     fsFormatMax
   };
-
+  
   /**
    * Length of signal
    */
diff --git a/storage/ndb/include/mgmapi/mgmapi_config_parameters.h b/storage/ndb/include/mgmapi/mgmapi_config_parameters.h
index d0f72c28209..938706db331 100644
--- a/storage/ndb/include/mgmapi/mgmapi_config_parameters.h
+++ b/storage/ndb/include/mgmapi/mgmapi_config_parameters.h
@@ -88,11 +88,12 @@
 #define CFG_DB_STRING_MEMORY          161 /* used from 5.1 */
 #define CFG_DB_INITIAL_OPEN_FILES     162 /* used from 5.1 */
 
-#define CFG_DB_DATA_MEM_2             199 /* used in special build in 5.1 */
-
 #define CFG_DB_DISK_PAGE_BUFFER_MEMORY 160
 #define CFG_DB_STRING_MEMORY          161
 
+#define CFG_DB_SGA                    198 /* super pool mem */
+#define CFG_DB_DATA_MEM_2             199 /* used in special build in 5.1 */
+
 #define CFG_NODE_ARBIT_RANK           200
 #define CFG_NODE_ARBIT_DELAY          201
 
diff --git a/storage/ndb/src/common/debugger/signaldata/FsReadWriteReq.cpp b/storage/ndb/src/common/debugger/signaldata/FsReadWriteReq.cpp
index 9270417c3ed..57acdea7063 100644
--- a/storage/ndb/src/common/debugger/signaldata/FsReadWriteReq.cpp
+++ b/storage/ndb/src/common/debugger/signaldata/FsReadWriteReq.cpp
@@ -49,6 +49,8 @@ printFSREADWRITEREQ(FILE * output, const Uint32 * theData,
     break;
   case FsReadWriteReq::fsFormatGlobalPage:
     fprintf(output, "List of global pages)\n");
+  case FsReadWriteReq::fsFormatSharedPage:
+    fprintf(output, "List of shared pages)\n");
     break;
   default:
     fprintf(output, "fsFormatMax not handled\n");
diff --git a/storage/ndb/src/kernel/SimBlockList.cpp b/storage/ndb/src/kernel/SimBlockList.cpp
index 7964acd23bf..785be12a235 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 <Emulator.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;
@@ -78,12 +79,13 @@ SimBlockList::load(Configuration & conf){
   Lgman* lg = 0;
   Tsman* ts = 0;
 
-  Block_context ctx(conf, * (Ndbd_mem_manager*)0);
+  Block_context ctx(conf, *data.m_mem_manager);
   
   SimulatedBlock * fs = 0;
   {
     Uint32 dl;
-    const ndb_mgm_configuration_iterator * p = conf.getOwnConfigIterator();
+    const ndb_mgm_configuration_iterator * p = 
+      ctx.m_config.getOwnConfigIterator();
     if(p && !ndb_mgm_get_int_parameter(p, CFG_DB_DISCLESS, &dl) && dl){
       fs = NEW_BLOCK(VoidFs)(ctx);
     } else { 
diff --git a/storage/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp b/storage/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
index ab356acb715..02c3abe0e80 100644
--- a/storage/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
+++ b/storage/ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp
@@ -131,6 +131,7 @@ Cmvmi::Cmvmi(Block_context& ctx) :
 
 Cmvmi::~Cmvmi()
 {
+  m_shared_page_pool.clear();
 }
 
 
@@ -324,13 +325,28 @@ Cmvmi::execREAD_CONFIG_REQ(Signal* signal)
   Uint64 page_buffer = 64*1024*1024;
   ndb_mgm_get_int64_parameter(p, CFG_DB_DISK_PAGE_BUFFER_MEMORY, &page_buffer);
   
-  page_buffer /= GLOBAL_PAGE_SIZE; // in pages
-  if (page_buffer > 0)
+  Uint32 pages = 0;
+  pages += page_buffer / GLOBAL_PAGE_SIZE; // in pages
+  pages += LCP_RESTORE_BUFFER;
+  m_global_page_pool.setSize(pages + 64, true);
+  
+  Uint64 shared_mem = 8*1024*1024;
+  ndb_mgm_get_int64_parameter(p, CFG_DB_SGA, &shared_mem);
+  shared_mem /= GLOBAL_PAGE_SIZE;
+  if (shared_mem)
   {
-    page_buffer += LGMAN_LOG_BUFFER;
+    Resource_limit rl;
+    rl.m_min = 0;
+    rl.m_max = shared_mem;
+    rl.m_resource_id = 0;
+    m_ctx.m_mm.set_resource_limit(rl);
+  }
+  
+  ndbrequire(m_ctx.m_mm.init());
+  {
+    void* ptr = m_ctx.m_mm.get_memroot();
+    m_shared_page_pool.set((GlobalPage*)ptr, ~0);
   }
-  page_buffer += LCP_RESTORE_BUFFER;
-  m_global_page_pool.setSize(page_buffer + 64, true);
   
   ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
   conf->senderRef = reference();
@@ -1083,6 +1099,37 @@ Cmvmi::execDUMP_STATE_ORD(Signal* signal)
 	      g_sectionSegmentPool.getSize(),
 	      g_sectionSegmentPool.getNoOfFree());
   }
+
+  if (dumpState->args[0] == 1000)
+  {
+    Uint32 len = signal->getLength();
+    if (signal->getLength() == 1)
+    {
+      signal->theData[1] = 0;
+      signal->theData[2] = ~0;
+      sendSignal(reference(), GSN_DUMP_STATE_ORD, signal, 3, JBB);
+      return;
+    }
+    Uint32 id = signal->theData[1];
+    Resource_limit rl;
+    if (!m_ctx.m_mm.get_resource_limit(id, rl))
+      len = 2;
+    else
+    {
+      if (rl.m_min || rl.m_curr || rl.m_max)
+	infoEvent("Resource %d min: %d max: %d curr: %d",
+		  id, rl.m_min, rl.m_max, rl.m_curr);
+    }
+
+    if (len == 3)
+    {
+      signal->theData[0] = 1000;
+      signal->theData[1] = id+1;
+      signal->theData[2] = ~0;
+      sendSignal(reference(), GSN_DUMP_STATE_ORD, signal, 3, JBB);
+    }
+    return;
+  }
   
   if (dumpState->args[0] == DumpStateOrd::CmvmiSetRestartOnErrorInsert){
     if(signal->getLength() == 1)
diff --git a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
index ba5920415e9..deac87bff8c 100644
--- a/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
+++ b/storage/ndb/src/kernel/blocks/dbdict/Dbdict.cpp
@@ -15325,8 +15325,6 @@ Dbdict::create_file_abort_complete(Signal* signal, SchemaOp* op)
   execute(signal, op->m_callback, 0);
 }
 
-CArray<KeyDescriptor> g_key_descriptor_pool;
-
 void
 Dbdict::drop_file_prepare_start(Signal* signal, SchemaOp* op)
 {
diff --git a/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp b/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
index cc558261a5e..8955a204800 100644
--- a/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtc/DbtcMain.cpp
@@ -12381,7 +12381,7 @@ void Dbtc::executeTriggers(Signal* signal, ApiConnectRecordPtr* transPtr)
 	  tmp2.release();
 	  LocalDataBuffer<11> tmp3(pool, trigPtr.p->afterValues);
 	  tmp3.release();
-          regApiPtr->theFiredTriggers.release(trigPtr.i);
+          regApiPtr->theFiredTriggers.release(trigPtr);
         }
 	trigPtr = nextTrigPtr;
       }
diff --git a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
index 0c0c0d702c3..6f004bc5154 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
@@ -469,20 +469,21 @@ typedef Ptr<Fragoperrec> FragoperrecPtr;
 
   struct Page_request 
   {
-    Page_request() {}
     Local_key m_key;
-    Uint16 m_estimated_free_space; // in bytes/records
-    Uint16 m_list_index; // in Disk_alloc_info.m_page_requests
     Uint32 m_frag_ptr_i;
     Uint32 m_extent_info_ptr;
-    Uint16 m_ref_count; // Waiters for page
+    Uint16 m_estimated_free_space; // in bytes/records
+    Uint16 m_list_index;           // in Disk_alloc_info.m_page_requests
+    Uint16 m_ref_count;            // Waiters for page
     Uint16 m_uncommitted_used_space;
-    union {
-      Uint32 nextList;
-      Uint32 nextPool;
-    };
+    Uint32 nextList;
     Uint32 prevList;
+    Uint32 m_magic;
   }; // 32 bytes
+  
+  typedef RecordPool<Page_request, WOPool> Page_request_pool;
+  typedef DLFifoListImpl<Page_request_pool, Page_request> Page_request_list;
+  typedef LocalDLFifoListImpl<Page_request_pool, Page_request> Local_page_request_list;
 
   STATIC_CONST( EXTENT_SEARCH_MATRIX_COLS = 4 ); // Guarantee size
   STATIC_CONST( EXTENT_SEARCH_MATRIX_ROWS = 5 ); // Total size
@@ -518,7 +519,6 @@ typedef Ptr<Fragoperrec> FragoperrecPtr;
   }; // 40 bytes
   
   typedef LocalDLList<Extent_info> Extent_list;
-  typedef LocalDLList<Page_request> Page_request_list;
 
   struct Tablerec;
   struct Disk_alloc_info 
@@ -553,7 +553,7 @@ typedef Ptr<Fragoperrec> FragoperrecPtr;
      * Requests (for update) that have sufficient space left after request
      *   these are currently being "mapped"
      */
-    DLList<Page_request>::Head m_page_requests[MAX_FREE_LIST];
+    Page_request_list::Head m_page_requests[MAX_FREE_LIST];
 
     /**
      * Current extent
@@ -1010,7 +1010,7 @@ ArrayPool<TupTriggerData> c_triggerPool;
   };
   
   ArrayPool<Extent_info> c_extent_pool;
-  ArrayPool<Page_request> c_page_request_pool;
+  Page_request_pool c_page_request_pool;
   DLHashTable<Extent_info> c_extent_hash;
 
   typedef Ptr<Tablerec> TablerecPtr;
@@ -2609,13 +2609,6 @@ private:
   void disk_page_free(Signal*, 
 		      Tablerec*, Fragrecord*, Local_key*, PagePtr, Uint32);
   
-  void disk_page_update_free_space(Fragrecord*, Ptr<Page_request>,
-				   DLList<Page_request>::Head list[],
-				   Uint32 i, Uint32 sz);
-  void disk_page_update_free_space(Fragrecord*, PagePtr, Uint32 i,
-				   Int32 uncommitted_delta, 
-				   Int32 extent_delta);
-  
   void disk_page_commit_callback(Signal*, Uint32 opPtrI, Uint32 page_id);  
   
   void disk_page_log_buffer_callback(Signal*, Uint32 opPtrI, Uint32); 
@@ -2632,6 +2625,7 @@ private:
 			     Uint32 gci, Uint32 logfile_group_id);
 
   void undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused);
+  void undo_createtable_logsync_callback(Signal* signal, Uint32, Uint32);
 
   void disk_page_set_dirty(Ptr<Page>);
   void restart_setup_page(Disk_alloc_info&, Ptr<Page>);
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
index 54166a67493..056baae0cdc 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
@@ -89,8 +89,8 @@ Dbtup::dump_disk_alloc(Dbtup::Disk_alloc_info & alloc)
   {
     printf("  %d : ", i);
     Ptr<Page_request> ptr;
-    LocalDLList<Page_request> list(c_page_request_pool, 
-				   alloc.m_page_requests[i]);
+    Local_page_request_list list(c_page_request_pool, 
+				 alloc.m_page_requests[i]);
     for(list.first(ptr); !ptr.isNull(); list.next(ptr))
     {
       ndbout << ptr << " ";
@@ -388,8 +388,6 @@ Dbtup::disk_page_prealloc(Signal* signal,
     return -err;
   }
 
-  new (req.p) Page_request();
-
   req.p->m_ref_count= 1;
   req.p->m_frag_ptr_i= fragPtr.i;
   req.p->m_uncommitted_used_space= sz;
@@ -510,8 +508,8 @@ Dbtup::disk_page_prealloc(Signal* signal,
   // And put page request in correct free list
   idx= alloc.calc_page_free_bits(new_size);
   {
-    LocalDLList<Page_request> list(c_page_request_pool, 
-				   alloc.m_page_requests[idx]);
+    Local_page_request_list list(c_page_request_pool, 
+				 alloc.m_page_requests[idx]);
     
     list.add(req);
   }
@@ -605,9 +603,9 @@ Dbtup::disk_page_prealloc_transit_page(Disk_alloc_info& alloc,
   
   if (old_idx != new_idx)
   {
-    DLList<Page_request>::Head *lists = alloc.m_page_requests;
-    LocalDLList<Page_request> old_list(c_page_request_pool, lists[old_idx]);
-    LocalDLList<Page_request> new_list(c_page_request_pool, lists[new_idx]);
+    Page_request_list::Head *lists = alloc.m_page_requests;
+    Local_page_request_list old_list(c_page_request_pool, lists[old_idx]);
+    Local_page_request_list new_list(c_page_request_pool, lists[new_idx]);
     old_list.remove(req);
     new_list.add(req);
 
@@ -761,8 +759,8 @@ Dbtup::disk_page_prealloc_callback_common(Signal* signal,
   }
   
   {
-    Page_request_list list(c_page_request_pool, 
-			   alloc.m_page_requests[old_idx]);
+    Local_page_request_list list(c_page_request_pool, 
+				 alloc.m_page_requests[old_idx]);
     list.release(req);
   }
 }
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
index 715f53ac95e..41d29c4ce11 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
@@ -309,7 +309,10 @@ void Dbtup::execREAD_CONFIG_REQ(Signal* signal)
 
   c_extent_pool.setSize(8192);
   c_extent_hash.setSize(1024); // 4k
-  c_page_request_pool.setSize(1000);
+  
+  Pool_context pc;
+  pc.m_block = this;
+  c_page_request_pool.wo_pool_init(RT_DBTUP_PAGE_REQUEST, pc);
   
   Uint32 nScanOp;       // use TUX config for now
   ndbrequire(!ndb_mgm_get_int_parameter(p, CFG_TUX_SCAN_OP, &nScanOp));
diff --git a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
index 1426389c9ca..5aa37aaade3 100644
--- a/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
+++ b/storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
@@ -548,8 +548,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
   if (regTabPtr.p->m_no_of_disk_attributes)
   {
     ljam();
-    if(!(getNodeState().getSystemRestartInProgress() && 
-	 getNodeState().startLevel == NodeState::SL_STARTING && 
+    if(!(getNodeState().startLevel == NodeState::SL_STARTING && 
 	 getNodeState().starting.startPhase <= 4))
     {
       Callback cb;
@@ -605,7 +604,7 @@ Dbtup::undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused)
   getFragmentrec(regFragPtr, fragOperPtr.p->fragidFrag, regTabPtr.p);
   ndbrequire(regFragPtr.i != RNIL);
   
-  Logfile_client lsman(this, c_lgman, regFragPtr.p->m_logfile_group_id);
+  Logfile_client lgman(this, c_lgman, regFragPtr.p->m_logfile_group_id);
 
   Disk_undo::Create create;
   create.m_type_length= Disk_undo::UNDO_CREATE << 16 | (sizeof(create) >> 2);
@@ -613,14 +612,39 @@ Dbtup::undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused)
   
   Logfile_client::Change c[1] = {{ &create, sizeof(create) >> 2 } };
   
-  Uint64 lsn= lsman.add_entry(c, 1);
+  Uint64 lsn= lgman.add_entry(c, 1);
+
+  Logfile_client::Request req;
+  req.m_callback.m_callbackData= fragOperPtr.i;
+  req.m_callback.m_callbackFunction = 
+    safe_cast(&Dbtup::undo_createtable_logsync_callback);
+  
+  int ret = lgman.sync_lsn(signal, lsn, &req, 0);
+  switch(ret){
+  case 0:
+    return;
+  default:
+    ndbout_c("ret: %d", ret);
+    ndbrequire(false);
+  }
+  
+}
+
+void
+Dbtup::undo_createtable_logsync_callback(Signal* signal, Uint32 ptrI, 
+					 Uint32 res)
+{
+  jamEntry();
+  FragoperrecPtr fragOperPtr;
+  fragOperPtr.i= ptrI;
+  ptrCheckGuard(fragOperPtr, cnoOfFragoprec, fragoperrec);
   
   signal->theData[0] = fragOperPtr.p->lqhPtrFrag;
   signal->theData[1] = 1;
   sendSignal(fragOperPtr.p->lqhBlockrefFrag, GSN_TUP_ADD_ATTCONF, 
 	     signal, 2, JBB);
-
-  releaseFragoperrec(fragOperPtr);
+  
+  releaseFragoperrec(fragOperPtr);  
 }
 
 /*
diff --git a/storage/ndb/src/kernel/blocks/lgman.cpp b/storage/ndb/src/kernel/blocks/lgman.cpp
index 879d09689c4..bfc0521c0a3 100644
--- a/storage/ndb/src/kernel/blocks/lgman.cpp
+++ b/storage/ndb/src/kernel/blocks/lgman.cpp
@@ -33,6 +33,8 @@
 #include <EventLogger.hpp>
 extern EventLogger g_eventLogger;
 
+#include <record_types.hpp>
+
 /**
  * ---<a>-----<b>-----<c>-----<d>---> (time)
  *
@@ -90,12 +92,11 @@ Lgman::Lgman(Block_context & ctx) :
 
   addRecSignal(GSN_GET_TABINFOREQ, &Lgman::execGET_TABINFOREQ);
 
-  m_last_lsn = 0;
+  m_last_lsn = 1;
   m_logfile_group_pool.setSize(10);
   m_logfile_group_hash.setSize(10);
   m_file_pool.setSize(10);
   m_data_buffer_pool.setSize(10);
-  m_log_waiter_pool.setSize(10000);
 }
   
 Lgman::~Lgman()
@@ -118,6 +119,10 @@ Lgman::execREAD_CONFIG_REQ(Signal* signal)
     m_ctx.m_config.getOwnConfigIterator();
   ndbrequire(p != 0);
 
+  Pool_context pc;
+  pc.m_block = this;
+  m_log_waiter_pool.wo_pool_init(RT_LGMAN_LOG_WAITER, pc);
+
   ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
   conf->senderRef = reference();
   conf->senderData = senderData;
@@ -271,9 +276,9 @@ Lgman::execDUMP_STATE_ORD(Signal* signal){
       if (!ptr.p->m_log_buffer_waiters.isEmpty())
       {
 	Uint32 free_buffer= ptr.p->m_free_buffer_words;
-	LocalDLFifoList<Log_waiter> 
-	  list(m_log_waiter_pool, ptr.p->m_log_buffer_waiters);
 	Ptr<Log_waiter> waiter;
+	Local_log_waiter_list 
+	  list(m_log_waiter_pool, ptr.p->m_log_buffer_waiters);
 	list.first(waiter);
 	infoEvent("  free_buffer_words: %d head(waiters).sz: %d %d",
 		  ptr.p->m_free_buffer_words,
@@ -282,13 +287,21 @@ Lgman::execDUMP_STATE_ORD(Signal* signal){
       }
       if (!ptr.p->m_log_sync_waiters.isEmpty())
       {
-	LocalDLFifoList<Log_waiter> 
-	  list(m_log_waiter_pool, ptr.p->m_log_sync_waiters);
 	Ptr<Log_waiter> waiter;
+	Local_log_waiter_list 
+	  list(m_log_waiter_pool, ptr.p->m_log_sync_waiters);
 	list.first(waiter);
-	infoEvent("  m_last_synced_lsn: %lld: %d head(waiters).m_sync_lsn: %lld",
+	infoEvent("  m_last_synced_lsn: %lld head(waiters %x).m_sync_lsn: %lld",
 		  ptr.p->m_last_synced_lsn,
+		  waiter.i,
 		  waiter.p->m_sync_lsn);
+	
+	while(!waiter.isNull())
+	{
+	  ndbout_c("ptr: %x %p lsn: %lld next: %x", 
+		   waiter.i, waiter.p, waiter.p->m_sync_lsn, waiter.p->nextList);
+	  list.next(waiter);
+	}
       }
       m_logfile_group_list.next(ptr);
     }
@@ -859,9 +872,6 @@ Lgman::Logfile_group::Logfile_group(const CreateFilegroupImplReq* req)
 bool
 Lgman::alloc_logbuffer_memory(Ptr<Logfile_group> ptr, Uint32 bytes)
 {
-  /**
-   * TODO use buddy allocator
-   */
   Uint32 pages= (((bytes + 3) >> 2) + File_formats::NDB_PAGE_SIZE_WORDS - 1)
     / File_formats::NDB_PAGE_SIZE_WORDS;
   Uint32 requested= pages;
@@ -869,14 +879,14 @@ Lgman::alloc_logbuffer_memory(Ptr<Logfile_group> ptr, Uint32 bytes)
     Page_map map(m_data_buffer_pool, ptr.p->m_buffer_pages);
     while(pages)
     {
-      Ptr<GlobalPage> page;
-      if(m_global_page_pool.seize(page))
+      Uint32 ptrI;
+      Uint32 cnt = pages > 64 ? 64 : pages;
+      m_ctx.m_mm.alloc(&ptrI, &cnt, 1);
+      if (cnt)
       {
 	Buffer_idx range;
-	range.m_ptr_i= page.i;
-	range.m_idx = 1;
-	while(pages >range.m_idx && m_global_page_pool.seizeId(page, page.i+1))
-	  range.m_idx++;
+	range.m_ptr_i= ptrI;
+	range.m_idx = cnt;
 	
 	ndbrequire(map.append((Uint32*)&range, 2));
 	pages -= range.m_idx;
@@ -989,12 +999,8 @@ Lgman::free_logbuffer_memory(Ptr<Logfile_group> ptr)
     tmp[0] = *it.data;
     ndbrequire(map.next(it));
     tmp[1] = *it.data;
-    while(range.m_idx)
-    {
-      m_global_page_pool.release(range.m_ptr_i);
-      range.m_ptr_i++;
-      range.m_idx--;
-    }
+    
+    m_ctx.m_mm.release(range.m_ptr_i, range.m_idx);
     map.next(it);
   }
   map.release();
@@ -1039,7 +1045,7 @@ Logfile_client::sync_lsn(Signal* signal,
     bool empty= false;
     Ptr<Lgman::Log_waiter> wait;
     {
-      LocalDLFifoList<Lgman::Log_waiter> 
+      Lgman::Local_log_waiter_list
 	list(m_lgman->m_log_waiter_pool, ptr.p->m_log_sync_waiters);
       
       empty= list.isEmpty();
@@ -1076,8 +1082,7 @@ Lgman::force_log_sync(Signal* signal,
 		      Ptr<Logfile_group> ptr, 
 		      Uint32 lsn_hi, Uint32 lsn_lo)
 {
-  LocalDLFifoList<Lgman::Log_waiter> 
-    list(m_log_waiter_pool, ptr.p->m_log_sync_waiters);
+  Local_log_waiter_list list(m_log_waiter_pool, ptr.p->m_log_sync_waiters);
   Uint64 force_lsn = lsn_hi; force_lsn <<= 32; force_lsn += lsn_lo;
 
   if(ptr.p->m_last_sync_req_lsn < force_lsn)
@@ -1086,7 +1091,7 @@ Lgman::force_log_sync(Signal* signal,
      * Do force
      */
     Buffer_idx pos= ptr.p->m_pos[PRODUCER].m_current_pos;
-    GlobalPage *page = m_global_page_pool.getPtr(pos.m_ptr_i);
+    GlobalPage *page = m_shared_page_pool.getPtr(pos.m_ptr_i);
   
     Uint32 free= File_formats::UNDO_PAGE_WORDS - pos.m_idx;
     if(pos.m_idx) // don't flush empty page...
@@ -1136,7 +1141,7 @@ Lgman::force_log_sync(Signal* signal,
 void
 Lgman::process_log_sync_waiters(Signal* signal, Ptr<Logfile_group> ptr)
 {
-  LocalDLFifoList<Log_waiter> 
+  Local_log_waiter_list 
     list(m_log_waiter_pool, ptr.p->m_log_sync_waiters);
 
   if(list.isEmpty())
@@ -1155,7 +1160,7 @@ Lgman::process_log_sync_waiters(Signal* signal, Ptr<Logfile_group> ptr)
     SimulatedBlock* b = globalData.getBlock(block);
     b->execute(signal, waiter.p->m_callback, 0);
 
-    list.release(waiter);
+    list.releaseFirst(waiter);
   }
   
   if(removed && !list.isEmpty())
@@ -1176,7 +1181,7 @@ Uint32*
 Lgman::get_log_buffer(Ptr<Logfile_group> ptr, Uint32 sz)
 {
   GlobalPage *page;
-  page=m_global_page_pool.getPtr(ptr.p->m_pos[PRODUCER].m_current_pos.m_ptr_i);
+  page=m_shared_page_pool.getPtr(ptr.p->m_pos[PRODUCER].m_current_pos.m_ptr_i);
   
   Uint32 total_free= ptr.p->m_free_buffer_words;
   assert(total_free >= sz);
@@ -1214,7 +1219,7 @@ next:
   pos= 0;
   assert(total_free >= free);
   total_free -= free;
-  page= m_global_page_pool.getPtr(next_page(ptr.p, PRODUCER));
+  page= m_shared_page_pool.getPtr(next_page(ptr.p, PRODUCER));
   goto next;
 }
 
@@ -1271,7 +1276,7 @@ Logfile_client::get_log_buffer(Signal* signal, Uint32 sz,
     bool empty= false;
     {
       Ptr<Lgman::Log_waiter> wait;
-      LocalDLFifoList<Lgman::Log_waiter> 
+      Lgman::Local_log_waiter_list
 	list(m_lgman->m_log_waiter_pool, ptr.p->m_log_buffer_waiters);
       
       empty= list.isEmpty();
@@ -1352,7 +1357,7 @@ Lgman::flush_log(Signal* signal, Ptr<Logfile_group> ptr, Uint32 force)
       else
       {
 	Buffer_idx pos= producer.m_current_pos;
-	GlobalPage *page = m_global_page_pool.getPtr(pos.m_ptr_i);
+	GlobalPage *page = m_shared_page_pool.getPtr(pos.m_ptr_i);
 	
 	Uint32 free= File_formats::UNDO_PAGE_WORDS - pos.m_idx;
 
@@ -1497,7 +1502,7 @@ void
 Lgman::process_log_buffer_waiters(Signal* signal, Ptr<Logfile_group> ptr)
 {
   Uint32 free_buffer= ptr.p->m_free_buffer_words;
-  LocalDLFifoList<Log_waiter> 
+  Local_log_waiter_list 
     list(m_log_waiter_pool, ptr.p->m_log_buffer_waiters);
 
   if(list.isEmpty())
@@ -1516,7 +1521,7 @@ Lgman::process_log_buffer_waiters(Signal* signal, Ptr<Logfile_group> ptr)
     SimulatedBlock* b = globalData.getBlock(block);
     b->execute(signal, waiter.p->m_callback, 0);
 
-    list.release(waiter);
+    list.releaseFirst(waiter);
   }
   
   if(removed && !list.isEmpty())
@@ -1571,7 +1576,7 @@ Lgman::write_log_pages(Signal* signal, Ptr<Logfile_group> ptr,
   req->data.pageData[0] = pageId;
   req->operationFlag = 0;
   FsReadWriteReq::setFormatFlag(req->operationFlag, 
-				FsReadWriteReq::fsFormatGlobalPage);
+				FsReadWriteReq::fsFormatSharedPage);
 
   if(max > pages)
   {
@@ -1592,7 +1597,7 @@ Lgman::write_log_pages(Signal* signal, Ptr<Logfile_group> ptr,
     filePtr.p->m_state |= Undofile::FS_OUTSTANDING;
 
     File_formats::Undofile::Undo_page *page= (File_formats::Undofile::Undo_page*)
-      m_global_page_pool.getPtr(pageId + max - 1);
+      m_shared_page_pool.getPtr(pageId + max - 1);
     Uint64 lsn = 0;
     lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32;
     lsn += page->m_page_header.m_page_lsn_lo;
@@ -1618,7 +1623,7 @@ Lgman::write_log_pages(Signal* signal, Ptr<Logfile_group> ptr,
     filePtr.p->m_state |= Undofile::FS_OUTSTANDING;
 
     File_formats::Undofile::Undo_page *page= (File_formats::Undofile::Undo_page*)
-      m_global_page_pool.getPtr(pageId + max - 1);
+      m_shared_page_pool.getPtr(pageId + max - 1);
     Uint64 lsn = 0;
     lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32;
     lsn += page->m_page_header.m_page_lsn_lo;
@@ -2099,7 +2104,7 @@ Lgman::find_log_head(Signal* signal, Ptr<Logfile_group> ptr)
     req->data.pageData[0] = page_id;
     req->operationFlag = 0;
     FsReadWriteReq::setFormatFlag(req->operationFlag, 
-				  FsReadWriteReq::fsFormatGlobalPage);
+				  FsReadWriteReq::fsFormatSharedPage);
     
     sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 
 	       FsReadWriteReq::FixedLength + 1, JBA);
@@ -2140,7 +2145,7 @@ Lgman::find_log_head(Signal* signal, Ptr<Logfile_group> ptr)
     req->data.pageData[0] = page_id;
     req->operationFlag = 0;
     FsReadWriteReq::setFormatFlag(req->operationFlag, 
-				  FsReadWriteReq::fsFormatGlobalPage);
+				  FsReadWriteReq::fsFormatSharedPage);
     
     sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 
 	       FsReadWriteReq::FixedLength + 1, JBA);
@@ -2201,7 +2206,7 @@ Lgman::execFSREADCONF(Signal* signal)
   lg_ptr.p->m_outstanding_fs = cnt - 1;
 
   Ptr<GlobalPage> page_ptr;
-  m_global_page_pool.getPtr(page_ptr, ptr.p->m_online.m_outstanding);
+  m_shared_page_pool.getPtr(page_ptr, ptr.p->m_online.m_outstanding);
   ptr.p->m_online.m_outstanding= 0;
   
   File_formats::Undofile::Undo_page* page = 
@@ -2333,7 +2338,7 @@ Lgman::find_log_head_in_file(Signal* signal,
     req->data.pageData[0] = page_id;
     req->operationFlag = 0;
     FsReadWriteReq::setFormatFlag(req->operationFlag, 
-				  FsReadWriteReq::fsFormatGlobalPage);
+				  FsReadWriteReq::fsFormatSharedPage);
     
     sendSignal(NDBFS_REF, GSN_FSREADREQ, signal, 
 	       FsReadWriteReq::FixedLength + 1, JBA);
@@ -2438,7 +2443,7 @@ Lgman::init_run_undo_log(Signal* signal)
       
       Uint32 page = ptr.p->m_pos[CONSUMER].m_current_pos.m_ptr_i;
       File_formats::Undofile::Undo_page* pageP = 
-	(File_formats::Undofile::Undo_page*)m_global_page_pool.getPtr(page);
+	(File_formats::Undofile::Undo_page*)m_shared_page_pool.getPtr(page);
       
       ptr.p->m_pos[CONSUMER].m_current_pos.m_idx = pageP->m_words_used;
       ptr.p->m_pos[PRODUCER].m_current_pos.m_idx = 1;
@@ -2578,7 +2583,7 @@ Lgman::read_undo_pages(Signal* signal, Ptr<Logfile_group> ptr,
   req->userPointer = filePtr.i;
   req->operationFlag = 0;
   FsReadWriteReq::setFormatFlag(req->operationFlag, 
-				FsReadWriteReq::fsFormatGlobalPage);
+				FsReadWriteReq::fsFormatSharedPage);
 
 
   if(max > pages)
@@ -2717,7 +2722,7 @@ Lgman::get_next_undo_record(Uint64 * this_lsn)
   Uint32 page = consumer.m_current_pos.m_ptr_i;
   
   File_formats::Undofile::Undo_page* pageP=(File_formats::Undofile::Undo_page*)
-    m_global_page_pool.getPtr(page);
+    m_shared_page_pool.getPtr(page);
 
   if(pos == 0)
   {
@@ -2795,7 +2800,7 @@ Lgman::get_next_undo_record(Uint64 * this_lsn)
       ndbout_c("reading from %d", consumer.m_current_pos.m_ptr_i);
 
     pageP=(File_formats::Undofile::Undo_page*)
-      m_global_page_pool.getPtr(consumer.m_current_pos.m_ptr_i);
+      m_shared_page_pool.getPtr(consumer.m_current_pos.m_ptr_i);
     
     pos= consumer.m_current_pos.m_idx= pageP->m_words_used;
     
diff --git a/storage/ndb/src/kernel/blocks/lgman.hpp b/storage/ndb/src/kernel/blocks/lgman.hpp
index 2a20bc01df1..72cef179e08 100644
--- a/storage/ndb/src/kernel/blocks/lgman.hpp
+++ b/storage/ndb/src/kernel/blocks/lgman.hpp
@@ -28,6 +28,9 @@
 #include "diskpage.hpp"
 #include <signaldata/GetTabInfo.hpp>
 
+#include <WOPool.hpp>
+#include <SLFifoList.hpp>
+
 class Lgman : public SimulatedBlock
 {
 public:
@@ -77,18 +80,19 @@ protected:
 public:
   struct Log_waiter
   {
+    Callback m_callback;
     union {
       Uint32 m_size;
       Uint64 m_sync_lsn;
     };
     Uint32 m_block;
-    Callback m_callback;
-    union {
-      Uint32 nextPool;
-      Uint32 nextList;
-    };
-    Uint32 prevList;
+    Uint32 nextList;
+    Uint32 m_magic;
   };
+
+  typedef RecordPool<Log_waiter, WOPool> Log_waiter_pool;
+  typedef SLFifoListImpl<Log_waiter_pool, Log_waiter> Log_waiter_list;
+  typedef LocalSLFifoListImpl<Log_waiter_pool, Log_waiter> Local_log_waiter_list;
   
   struct Undofile
   {
@@ -190,7 +194,7 @@ public:
       Uint64 m_last_read_lsn;
       Uint64 m_last_lcp_lsn;
     };
-    DLFifoList<Log_waiter>::Head m_log_sync_waiters;
+    Log_waiter_list::Head m_log_sync_waiters;
     
     Buffer_idx m_tail_pos[3]; // 0 is cut, 1 is saved, 2 is current
     Buffer_idx m_file_pos[2]; // 0 tail, 1 head = { file_ptr_i, page_no }
@@ -200,7 +204,7 @@ public:
     DLFifoList<Undofile>::Head m_meta_files;// Files being created or dropped
     
     Uint32 m_free_buffer_words;    // Free buffer page words
-    DLFifoList<Log_waiter>::Head m_log_buffer_waiters;
+    Log_waiter_list::Head m_log_buffer_waiters;
     Page_map::Head m_buffer_pages; // Pairs of { ptr.i, count }
     struct Position {
       Buffer_idx m_current_page;   // { m_buffer_pages.i, left in range }
@@ -235,7 +239,7 @@ private:
   friend class Logfile_client;
   ArrayPool<Undofile> m_file_pool;
   ArrayPool<Logfile_group> m_logfile_group_pool;
-  ArrayPool<Log_waiter> m_log_waiter_pool;
+  Log_waiter_pool m_log_waiter_pool;
 
   Page_map::DataBufferPool m_data_buffer_pool;
 
diff --git a/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp b/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
index 91ee8d5aa47..eced400d480 100644
--- a/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
+++ b/storage/ndb/src/kernel/blocks/ndbcntr/NdbcntrMain.cpp
@@ -90,6 +90,7 @@ static BlockInfo ALL_BLOCKS[] = {
 static const Uint32 ALL_BLOCKS_SZ = sizeof(ALL_BLOCKS)/sizeof(BlockInfo);
 
 static BlockReference readConfigOrder[ALL_BLOCKS_SZ] = {
+  CMVMI_REF,
   DBTUP_REF,
   DBACC_REF,
   DBTC_REF,
@@ -100,7 +101,6 @@ static BlockReference readConfigOrder[ALL_BLOCKS_SZ] = {
   NDBFS_REF,
   NDBCNTR_REF,
   QMGR_REF,
-  CMVMI_REF,
   TRIX_REF,
   BACKUP_REF,
   DBUTIL_REF,
diff --git a/storage/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp b/storage/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp
index feffc071d47..138b519631c 100644
--- a/storage/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp
+++ b/storage/ndb/src/kernel/blocks/ndbfs/Ndbfs.cpp
@@ -325,14 +325,17 @@ Ndbfs::readWriteRequest(int action, Signal * signal)
   request->action = (Request::Action) action;
   request->theTrace = signal->getTrace();
 
+  Uint32 format = fsRWReq->getFormatFlag(fsRWReq->operationFlag);
+
   if (fsRWReq->numberOfPages == 0) { //Zero pages not allowed
     jam();
     errorCode = FsRef::fsErrInvalidParameters;
     goto error;
   }
 
-  if(fsRWReq->getFormatFlag(fsRWReq->operationFlag) != 
-     FsReadWriteReq::fsFormatGlobalPage){
+  if(format != FsReadWriteReq::fsFormatGlobalPage &&
+     format != FsReadWriteReq::fsFormatSharedPage)
+  {     
     if (fsRWReq->varIndex >= getBatSize(blockNumber)) {
       jam();// Ensure that a valid variable is used    
       errorCode = FsRef::fsErrInvalidParameters;
@@ -353,7 +356,7 @@ Ndbfs::readWriteRequest(int action, Signal * signal)
     tNRR = myBaseAddrRef->nrr;
     tWA = (char*)myBaseAddrRef->WA;
     
-    switch (fsRWReq->getFormatFlag(fsRWReq->operationFlag)) {
+    switch (format) {
       
       // List of memory and file pages pairs
     case FsReadWriteReq::fsFormatListOfPairs: { 
@@ -422,7 +425,9 @@ Ndbfs::readWriteRequest(int action, Signal * signal)
       goto error;
     }//default
     }//switch
-  } else {
+  } 
+  else if (format == FsReadWriteReq::fsFormatGlobalPage)
+  {
     Ptr<GlobalPage> ptr;
     m_global_page_pool.getPtr(ptr, fsRWReq->data.pageData[0]);
     request->par.readWrite.pages[0].buf = (char*)ptr.p;
@@ -430,10 +435,20 @@ Ndbfs::readWriteRequest(int action, Signal * signal)
     request->par.readWrite.pages[0].offset= ((UintPtr)GLOBAL_PAGE_SIZE)*fsRWReq->varIndex;
     request->par.readWrite.numberOfPages = 1;
   }
+  else
+  {
+    ndbrequire(format == FsReadWriteReq::fsFormatSharedPage);
+    Ptr<GlobalPage> ptr;
+    m_shared_page_pool.getPtr(ptr, fsRWReq->data.pageData[0]);
+    request->par.readWrite.pages[0].buf = (char*)ptr.p;
+    request->par.readWrite.pages[0].size = ((UintPtr)GLOBAL_PAGE_SIZE)*fsRWReq->numberOfPages;
+    request->par.readWrite.pages[0].offset= ((UintPtr)GLOBAL_PAGE_SIZE)*fsRWReq->varIndex;
+    request->par.readWrite.numberOfPages = 1;
+  }
   
   ndbrequire(forward(openFile, request));
   return;
-
+  
 error:
   theRequestPool->put(request);
   FsRef * const fsRef = (FsRef *)&signal->theData[0];
diff --git a/storage/ndb/src/kernel/blocks/pgman.cpp b/storage/ndb/src/kernel/blocks/pgman.cpp
index 3769b6aaf44..fb47a16665d 100644
--- a/storage/ndb/src/kernel/blocks/pgman.cpp
+++ b/storage/ndb/src/kernel/blocks/pgman.cpp
@@ -125,10 +125,13 @@ Pgman::execREAD_CONFIG_REQ(Signal* signal)
   {
     page_buffer /= GLOBAL_PAGE_SIZE; // in pages
     m_page_entry_pool.setSize(100*page_buffer);
-    m_page_request_pool.setSize(10000);
     m_param.m_max_pages = page_buffer;
     m_param.m_max_hot_pages = (page_buffer * 9) / 10;
   }
+
+  Pool_context pc;
+  pc.m_block = this;
+  m_page_request_pool.wo_pool_init(RT_PGMAN_PAGE_REQUEST, pc);
   
   ReadConfigConf * conf = (ReadConfigConf*)signal->getDataPtrSend();
   conf->senderRef = reference();
@@ -977,8 +980,7 @@ Pgman::process_callback(Signal* signal, Ptr<Page_entry> ptr)
        * Make sure list is in own scope if callback will access this
        * list again (destructor restores list head).
        */
-      LocalDLFifoList<Page_request>
-        req_list(m_page_request_pool, ptr.p->m_requests);
+      Local_page_request_list req_list(m_page_request_pool, ptr.p->m_requests);
       Ptr<Page_request> req_ptr;
 
       req_list.first(req_ptr);
@@ -988,8 +990,6 @@ Pgman::process_callback(Signal* signal, Ptr<Page_entry> ptr)
       b = globalData.getBlock(req_ptr.p->m_block);
       callback = req_ptr.p->m_callback;
       
-      req_list.release(req_ptr);
-
       if (req_ptr.p->m_flags & DIRTY_FLAGS)
       {
         jam();
@@ -997,6 +997,8 @@ Pgman::process_callback(Signal* signal, Ptr<Page_entry> ptr)
 	ndbassert(ptr.p->m_dirty_count);
 	ptr.p->m_dirty_count --;
       }
+
+      req_list.releaseFirst(req_ptr);
     }
     ndbrequire(state & Page_entry::BOUND);
     ndbrequire(state & Page_entry::MAPPED);
@@ -1602,12 +1604,11 @@ Pgman::get_page(Signal* signal, Ptr<Page_entry> ptr, Page_request page_req)
   // queue the request
   Ptr<Pgman::Page_request> req_ptr;
   {
-    LocalDLFifoList<Page_request>
-      req_list(m_page_request_pool, ptr.p->m_requests);
+    Local_page_request_list req_list(m_page_request_pool, ptr.p->m_requests);
     if (! (req_flags & Page_request::ALLOC_REQ))
-      req_list.seize(req_ptr);
+      req_list.seizeLast(req_ptr);
     else
-      req_list.seizeFront(req_ptr);
+      req_list.seizeFirst(req_ptr);
   }
   
   if (req_ptr.i == RNIL)
@@ -2171,8 +2172,8 @@ operator<<(NdbOut& out, Ptr<Pgman::Page_entry> ptr)
   out << " busy_count=" << dec << pe.m_busy_count;
 #ifdef VM_TRACE
   {
-    LocalDLFifoList<Pgman::Page_request>
-      req_list(ptr.p->m_this->m_page_request_pool, ptr.p->m_requests);
+    Local_page_request_list req_list(ptr.p->m_this->m_page_request_pool, 
+				     ptr.p->m_requests);
     if (! req_list.isEmpty())
     {
       Ptr<Pgman::Page_request> req_ptr;
diff --git a/storage/ndb/src/kernel/blocks/pgman.hpp b/storage/ndb/src/kernel/blocks/pgman.hpp
index d9c75b06e7d..17ffdd47d0a 100644
--- a/storage/ndb/src/kernel/blocks/pgman.hpp
+++ b/storage/ndb/src/kernel/blocks/pgman.hpp
@@ -262,13 +262,14 @@ private:
     Uint16 m_flags;
     SimulatedBlock::Callback m_callback;
 
-    union {
-      Uint32 nextPool;
-      Uint32 nextList;
-    };
-    Uint32 prevList;
+    Uint32 nextList;
+    Uint32 m_magic;
   };
 
+  typedef RecordPool<Page_request, WOPool> Page_request_pool;
+  typedef SLFifoListImpl<Page_request_pool, Page_request> Page_request_list;
+  typedef LocalSLFifoListImpl<Page_request_pool, Page_request> Local_page_request_list;
+  
   struct Page_entry_stack_ptr {
     Uint32 nextList;
     Uint32 prevList;
@@ -338,7 +339,7 @@ private:
       Uint32 nextPool;
     };
     
-    DLFifoList<Page_request>::Head m_requests;
+    Page_request_list::Head m_requests;
     
     Uint32 nextHash;
     Uint32 prevHash;
@@ -384,7 +385,7 @@ private:
   File_map::DataBufferPool m_data_buffer_pool;
 
   // page entries and requests
-  ArrayPool<Page_request> m_page_request_pool;
+  Page_request_pool m_page_request_pool;
   ArrayPool<Page_entry> m_page_entry_pool;
   Page_hashlist m_page_hashlist;
   Page_stack m_page_stack;
diff --git a/storage/ndb/src/kernel/blocks/record_types.hpp b/storage/ndb/src/kernel/blocks/record_types.hpp
index 6d2857bd4e2..3917c897fda 100644
--- a/storage/ndb/src/kernel/blocks/record_types.hpp
+++ b/storage/ndb/src/kernel/blocks/record_types.hpp
@@ -18,25 +18,13 @@
 #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
  */
 
 /**
  * Operations for dd
  *    PGMAN_PAGE_REQUEST
- *    LGMAN_LOG_BUFFER_WAITER
- *    LGMAN_LOG_SYNC_WAITER
+ *    LGMAN_LOG_WAITER
  *    DBTUP_PAGE_REQUEST
  */
 #define RG_DISK_OPERATIONS      1
@@ -47,4 +35,20 @@
  */
 #define RG_DISK_RECORDS         2
 
+/**
+ * 
+ */
+#define RG_RESERVED             0
+#define RG_COUNT                3
+
+/**
+ * Record types
+ */
+#define RT_PGMAN_PAGE_REQUEST      MAKE_TID(1, RG_DISK_OPERATIONS)
+
+#define RT_LGMAN_LOG_WAITER        MAKE_TID(2, RG_DISK_OPERATIONS)
+
+#define RT_DBTUP_PAGE_REQUEST      MAKE_TID(3, RG_DISK_OPERATIONS)
+#define RT_DBTUP_EXTENT_INFO       MAKE_TID(4, 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/ArrayPool.hpp b/storage/ndb/src/kernel/vm/ArrayPool.hpp
index 49dd394eee4..6a5bb948dd8 100644
--- a/storage/ndb/src/kernel/vm/ArrayPool.hpp
+++ b/storage/ndb/src/kernel/vm/ArrayPool.hpp
@@ -45,6 +45,7 @@ public:
    */
   bool setSize(Uint32 noOfElements, bool align = false, bool exit_on_error = true, bool guard = true);
   bool set(T*, Uint32 cnt, bool align = false);
+  void clear() { theArray = 0; }
 
   inline Uint32 getNoOfFree() const {
     return noOfFree;
diff --git a/storage/ndb/src/kernel/vm/DLFifoList.hpp b/storage/ndb/src/kernel/vm/DLFifoList.hpp
index 9d9374426ca..2a10c0303ea 100644
--- a/storage/ndb/src/kernel/vm/DLFifoList.hpp
+++ b/storage/ndb/src/kernel/vm/DLFifoList.hpp
@@ -17,20 +17,23 @@
 #ifndef DLFIFOLIST_HPP
 #define DLFIFOLIST_HPP
 
-#include "ArrayPool.hpp"
-#include <NdbOut.hpp>
+#include <ndb_global.h>
+#include <kernel_types.h>
+#include "Pool.hpp"
 
 /**
  * Template class used for implementing an
  *   list of object retreived from a pool
  */
-template <class T, class U = T>
-class DLFifoList {
+template <typename P, typename T, typename U = T>
+class DLFifoListImpl 
+{
 public:
   /**
    * List head
    */
-  struct Head {
+  struct Head 
+  {
     Head();
     Uint32 firstItem;
     Uint32 lastItem;
@@ -42,62 +45,27 @@ public:
     inline bool isEmpty() const { return firstItem == RNIL;}
   };
   
-  DLFifoList(ArrayPool<T> & thePool);
+  DLFifoListImpl(P & thePool);
   
-  /**
-   * Allocate an object from pool - update Ptr
-   *
-   * Return i
-   */
-  bool seize(Ptr<T> &);
-
-  /**
-   * Allocate an object from pool - update Ptr - put in front of list
-   *
-   * Return i
-   */
-  bool seizeFront(Ptr<T> &);
-
-  /**
-   * Allocate object <b>i</b> from pool - update Ptr
-   *
-   * Return i
-   */
-  bool seizeId(Ptr<T> &, Uint32 i);
+  bool seizeFirst(Ptr<T> &);
+  bool seizeLast(Ptr<T> &);
+  bool seize(Ptr<T> & ptr) { return seizeLast(ptr);}
   
-  /**
-   * Add object to list 
-   * 
-   * @NOTE MUST be seized from correct pool
-   */
-  void add(Ptr<T> &);
+  void release(Ptr<T> &);
+  void release(); // release all
+  
+  void addFirst(Ptr<T> &);
+  void addLast(Ptr<T> &);
+  void add(Ptr<T> & ptr) { addLast(ptr);}
 
   /**
    * Insert object <em>ptr</ptr> _before_ <em>loc</em>
    */
   void insert(Ptr<T> & ptr, Ptr<T>& loc);
   
-  /**
-   * Remove from list 
-   */
   void remove(Ptr<T> &);
-
-  /**
-   * Return an object to pool
-   */
-  void release(Uint32 i);
   
   /**
-   * Return an object to pool
-   */
-  void release(Ptr<T> &);
-
-  /**
-   * Return all objects to the pool
-   */
-  void release();
-
-  /**
    *  Update i & p value according to <b>i</b>
    */
   void getPtr(Ptr<T> &, Uint32 i) const;
@@ -119,7 +87,6 @@ public:
    */
   bool first(Ptr<T> &) const ;
 
-
   /**
    * Update ptr to first element in list
    *
@@ -134,7 +101,6 @@ public:
    */
   bool next(Ptr<T> &) const ;
   
-
   /**
    * Get next element
    *
@@ -148,61 +114,37 @@ public:
    * NOTE ptr must be both p & i
    */
   bool hasNext(const Ptr<T> &) const;
-  
+
   /**
-   * Check if prev exists i.e. this is not first
+   * Check if next exists i.e. this is not last
    *
    * NOTE ptr must be both p & i
    */
   bool hasPrev(const Ptr<T> &) const;
-
-  Uint32 noOfElements() const {
-    Uint32 c = 0;
-    Uint32 i = head.firstItem;
-    while(i != RNIL){
-      c++;
-      const T * t = thePool.getPtr(i);
-      i = t->U::nextList;
-    }
-    return c;
-  }
-
-  /**
-   * Print
-   * (Run operator NdbOut<< on every element)
-   */
-  void print(NdbOut & out) {
-    Uint32 i = head.firstItem;
-    while(i != RNIL){
-      T * t = thePool.getPtr(i);
-      out << (unsigned int) t << "[" << i << "]:"; 
-      t->print(out); out << " ";
-      i = t->U::nextList;
-    }
-  }
-
+  
   inline bool isEmpty() const { return head.firstItem == RNIL;}
 
   /**
    * Copy list (head)
    *   Will construct to identical lists
    */
-  DLFifoList<T>& operator=(const DLFifoList<T>& src){
+  DLFifoListImpl<P,T,U>& operator=(const DLFifoListImpl<P,T,U>& src){
     assert(&thePool == &src.thePool);
     this->head = src.head;
     return * this;
   }
-
+  
 protected:
   Head head;
-  ArrayPool<T> & thePool;
+  P & thePool;
 };
 
-template <class T, class U = T>
-class LocalDLFifoList : public DLFifoList<T,U> {
+template <typename P, typename T, typename U = T>
+class LocalDLFifoListImpl : public DLFifoListImpl<P,T,U> 
+{
 public:
-  LocalDLFifoList(ArrayPool<T> & thePool, typename DLFifoList<T,U>::Head &_src)
-    : DLFifoList<T,U>(thePool), src(_src)
+  LocalDLFifoListImpl(P & thePool, typename DLFifoListImpl<P,T,U>::Head &_src)
+    : DLFifoListImpl<P,T,U>(thePool), src(_src)
   {
     this->head = src;
 #ifdef VM_TRACE
@@ -211,25 +153,27 @@ public:
 #endif
   }
   
-  ~LocalDLFifoList(){
+  ~LocalDLFifoListImpl(){
 #ifdef VM_TRACE
     assert(src.in_use == true);
 #endif
     src = this->head;
   }
 private:
-  typename DLFifoList<T,U>::Head & src;
+  typename DLFifoListImpl<P,T,U>::Head & src;
 };
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
-DLFifoList<T,U>::DLFifoList(ArrayPool<T> & _pool):
-  thePool(_pool){
+DLFifoListImpl<P,T,U>::DLFifoListImpl(P & _pool):
+  thePool(_pool)
+{
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
-DLFifoList<T,U>::Head::Head(){
+DLFifoListImpl<P,T,U>::Head::Head()
+{
   firstItem = RNIL;
   lastItem = RNIL;
 #ifdef VM_TRACE
@@ -237,94 +181,83 @@ DLFifoList<T,U>::Head::Head(){
 #endif
 }
 
-/**
- * Allocate an object from pool - update Ptr
- *
- * Return i
- */
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::seize(Ptr<T> & p){
-  thePool.seize(p);
-  if (p.i != RNIL) {
-    add(p);
+DLFifoListImpl<P,T,U>::seizeFirst(Ptr<T> & p)
+{
+  if (likely(thePool.seize(p)))
+  {
+    addFirst(p);
     return true;
   }
   p.p = NULL;
   return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::seizeFront(Ptr<T> & p){
-  Uint32 ff = head.firstItem;
-  thePool.seize(p);
-  if (p.i != RNIL) 
+DLFifoListImpl<P,T,U>::seizeLast(Ptr<T> & p)
+{
+  if (likely(thePool.seize(p)))
   {
-    p.p->U::prevList = RNIL;
-    p.p->U::nextList = ff;
-    head.firstItem = p.i;
-    if (ff == RNIL)
-    {
-      head.lastItem = p.i;
-    }
-    else
-    {
-      T * t2 = thePool.getPtr(ff);
-      t2->U::prevList = p.i;
-    }
+    addLast(p);
     return true;
   }
   p.p = NULL;
   return false;
 }
 
-/**
- * Allocate an object from pool - update Ptr
- *
- * Return i
- */
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
-bool
-DLFifoList<T,U>::seizeId(Ptr<T> & p, Uint32 ir){
-  thePool.seizeId(p, ir);
-  if(p.i != RNIL){
-    add(p);
-    return true;
+void
+DLFifoListImpl<P,T,U>::addFirst(Ptr<T> & p)
+{
+  Uint32 ff = head.firstItem;
+  
+  p.p->U::prevList = RNIL;
+  p.p->U::nextList = ff;
+  head.firstItem = p.i;
+  if (ff == RNIL)
+  {
+    head.lastItem = p.i;
+  }
+  else
+  {
+    T * t2 = thePool.getPtr(ff);
+    t2->U::prevList = p.i;
   }
-  p.p = NULL;
-  return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 void
-DLFifoList<T,U>::add(Ptr<T> & p){
+DLFifoListImpl<P,T,U>::addLast(Ptr<T> & p)
+{
   T * t = p.p;
   Uint32 last = head.lastItem;
-
-  if(p.i == RNIL)
-    ErrorReporter::handleAssert("DLFifoList<T,U>::add", __FILE__, __LINE__);
+  head.lastItem = p.i;    
   
   t->U::nextList = RNIL;
   t->U::prevList = last;
-  if (head.firstItem == RNIL)
-    head.firstItem = p.i;
-  head.lastItem = p.i;    
   
-  if(last != RNIL){
+  if(last != RNIL)
+  {
     T * t2 = thePool.getPtr(last);
     t2->U::nextList = p.i;
   }
+  else
+  {
+    head.firstItem = p.i;
+  }
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 void
-DLFifoList<T,U>::insert(Ptr<T> & ptr, Ptr<T> & loc){
+DLFifoListImpl<P,T,U>::insert(Ptr<T> & ptr, Ptr<T> & loc)
+{
   Uint32 prev= loc.p->U::prevList;
   if(loc.i == head.firstItem)
   {
@@ -342,88 +275,85 @@ DLFifoList<T,U>::insert(Ptr<T> & ptr, Ptr<T> & loc){
   ptr.p->U::nextList = loc.i;
 }
 
-/**
- * Return an object to pool
- */
-template <class T, class U>
-inline
-void 
-DLFifoList<T,U>::release(Uint32 i){
-  Ptr<T> p;
-  p.i = i;
-  p.p = thePool.getPtr(i);
-  release(p);
-}
-
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
-void 
-DLFifoList<T,U>::remove(Ptr<T> & p){
+void
+DLFifoListImpl<P,T,U>::remove(Ptr<T> & p)
+{
   T * t = p.p;
   Uint32 ni = t->U::nextList;
   Uint32 pi = t->U::prevList;
 
-  if(ni != RNIL){
+  if(ni != RNIL)
+  {
     T * t = thePool.getPtr(ni);
     t->U::prevList = pi;
-  } else {
+  } 
+  else 
+  {
     // We are releasing last
     head.lastItem = pi;
   }
   
-  if(pi != RNIL){
+  if(pi != RNIL)
+  {
     T * t = thePool.getPtr(pi);
     t->U::nextList = ni;
-  } else {
+  } 
+  else 
+  {
     // We are releasing first
     head.firstItem = ni;
   }
 }
-  
-/**
- * Return an object to pool
- */
-template <class T, class U>
-inline
-void 
-DLFifoList<T,U>::release(Ptr<T> & p){
-  remove(p);
-  thePool.release(p.i);
-}  
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 void 
-DLFifoList<T,U>::release(){
+DLFifoListImpl<P,T,U>::release()
+{
   Ptr<T> p;
-  while(head.firstItem != RNIL){
+  while(head.firstItem != RNIL)
+  {
     p.i = head.firstItem;  
     p.p = thePool.getPtr(head.firstItem);    
     T * t = p.p;
     head.firstItem = t->U::nextList;
     release(p);
   }
-}  
+}
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 void 
-DLFifoList<T,U>::getPtr(Ptr<T> & p, Uint32 i) const {
+DLFifoListImpl<P,T,U>::release(Ptr<T> & p)
+{
+  remove(p);
+  thePool.release(p);
+}
+
+template <typename P, typename T, typename U>
+inline
+void 
+DLFifoListImpl<P,T,U>::getPtr(Ptr<T> & p, Uint32 i) const 
+{
   p.i = i;
   p.p = thePool.getPtr(i);
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 void 
-DLFifoList<T,U>::getPtr(Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::getPtr(Ptr<T> & p) const 
+{
   thePool.getPtr(p);
 }
   
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 T * 
-DLFifoList<T,U>::getPtr(Uint32 i) const {
+DLFifoListImpl<P,T,U>::getPtr(Uint32 i) const 
+{
   return thePool.getPtr(i);
 }
 
@@ -432,12 +362,14 @@ DLFifoList<T,U>::getPtr(Uint32 i) const {
  *
  * Return i
  */
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::first(Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::first(Ptr<T> & p) const 
+{
   p.i = head.firstItem;
-  if(p.i != RNIL){
+  if(p.i != RNIL)
+  {
     p.p = thePool.getPtr(p.i);
     return true;
   }
@@ -445,12 +377,14 @@ DLFifoList<T,U>::first(Ptr<T> & p) const {
   return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::last(Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::last(Ptr<T> & p) const 
+{
   p.i = head.lastItem;
-  if(p.i != RNIL){
+  if(p.i != RNIL)
+  {
     p.p = thePool.getPtr(p.i);
     return true;
   }
@@ -458,12 +392,14 @@ DLFifoList<T,U>::last(Ptr<T> & p) const {
   return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::next(Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::next(Ptr<T> & p) const 
+{
   p.i = p.p->U::nextList;
-  if(p.i != RNIL){
+  if(p.i != RNIL)
+  {
     p.p = thePool.getPtr(p.i);
     return true;
   }
@@ -471,12 +407,14 @@ DLFifoList<T,U>::next(Ptr<T> & p) const {
   return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::prev(Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::prev(Ptr<T> & p) const 
+{
   p.i = p.p->U::prevList;
-  if(p.i != RNIL){
+  if(p.i != RNIL)
+  {
     p.p = thePool.getPtr(p.i);
     return true;
   }
@@ -484,18 +422,36 @@ DLFifoList<T,U>::prev(Ptr<T> & p) const {
   return false;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::hasNext(const Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::hasNext(const Ptr<T> & p) const 
+{
   return p.p->U::nextList != RNIL;
 }
 
-template <class T, class U>
+template <typename P, typename T, typename U>
 inline
 bool
-DLFifoList<T,U>::hasPrev(const Ptr<T> & p) const {
+DLFifoListImpl<P,T,U>::hasPrev(const Ptr<T> & p) const 
+{
   return p.p->U::prevList != RNIL;
 }
 
+// Specializations
+
+template <typename T, typename U = T>
+class DLFifoList : public DLFifoListImpl<ArrayPool<T>, T, U>
+{
+public:
+  DLFifoList(ArrayPool<T> & p) : DLFifoListImpl<ArrayPool<T>, T, U>(p) {}
+};
+
+template <typename T, typename U = T>
+class LocalDLFifoList : public LocalDLFifoListImpl<ArrayPool<T>,T,U> {
+public:
+  LocalDLFifoList(ArrayPool<T> & p, typename DLFifoList<T,U>::Head & _src)
+    : LocalDLFifoListImpl<ArrayPool<T>,T,U>(p, _src) {}
+};
+
 #endif
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/GlobalData.hpp b/storage/ndb/src/kernel/vm/GlobalData.hpp
index 6d48d96711f..2761edb0571 100644
--- a/storage/ndb/src/kernel/vm/GlobalData.hpp
+++ b/storage/ndb/src/kernel/vm/GlobalData.hpp
@@ -81,6 +81,7 @@ private:
   SimulatedBlock* blockTable[NO_OF_BLOCKS]; // Owned by Dispatcher::
 public:
   ArrayPool<GlobalPage> m_global_page_pool;
+  ArrayPool<GlobalPage> m_shared_page_pool;
 };
 
 extern GlobalData globalData;
diff --git a/storage/ndb/src/kernel/vm/Makefile.am b/storage/ndb/src/kernel/vm/Makefile.am
index 64739b7459e..b348c995552 100644
--- a/storage/ndb/src/kernel/vm/Makefile.am
+++ b/storage/ndb/src/kernel/vm/Makefile.am
@@ -19,7 +19,8 @@ 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 WOPool.cpp RWPool.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/mgmapi
 
@@ -52,9 +53,9 @@ ndbd_malloc_impl_test_LDFLAGS = @ndb_bin_am_ldflags@ \
   $(top_builddir)/dbug/libdbug.a \
   $(top_builddir)/strings/libmystrings.a
 
-bench_pool_SOURCES = bench_pool.cpp ndbd_malloc.cpp \
-                     SuperPool.cpp NdbdSuperPool.cpp ndbd_malloc_impl.cpp
+bench_pool_SOURCES = bench_pool.cpp ../SimBlockList.o
 bench_pool_LDFLAGS = @ndb_bin_am_ldflags@ \
+  libkernel.a ../error/liberror.a \
   $(top_builddir)/storage/ndb/src/libndbclient.la \
   $(top_builddir)/mysys/libmysys.a \
   $(top_builddir)/dbug/libdbug.a \
diff --git a/storage/ndb/src/kernel/vm/Pool.cpp b/storage/ndb/src/kernel/vm/Pool.cpp
new file mode 100644
index 00000000000..844c3ec34ab
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/Pool.cpp
@@ -0,0 +1,43 @@
+/* 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)
+{
+  m_block->m_ctx.m_mm.release_page(type_id, i);
+}
+
+void*
+Pool_context::get_memroot()
+{
+  return m_block->m_ctx.m_mm.get_memroot();
+}
+
+void
+Pool_context::handleAbort(int err, const char * msg)
+{
+  m_block->progError(__LINE__, err, msg);
+}
diff --git a/storage/ndb/src/kernel/vm/Pool.hpp b/storage/ndb/src/kernel/vm/Pool.hpp
index 52f7df0ead0..f4149c7da65 100644
--- a/storage/ndb/src/kernel/vm/Pool.hpp
+++ b/storage/ndb/src/kernel/vm/Pool.hpp
@@ -17,8 +17,33 @@
 #ifndef NDB_POOL_HPP
 #define NDB_POOL_HPP
 
+#include <ndb_global.h>
 #include <kernel_types.h>
 
+/**
+ * Type bits
+ *
+ * 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)
+
+/**
+ * Page bits
+ */
+#define POOL_RECORD_BITS 13
+#define POOL_RECORD_MASK ((1 << POOL_RECORD_BITS) - 1)
+
+/**
+ * Record_info
+ *
+ */
 struct Record_info
 {
   Uint16 m_size;
@@ -27,6 +52,9 @@ struct Record_info
   Uint16 m_offset_magic;
 };
 
+/**
+ * Resource_limit
+ */
 struct Resource_limit
 {
   Uint32 m_min;
@@ -38,7 +66,11 @@ struct Resource_limit
 struct Pool_context
 {
   class SimulatedBlock* m_block;
-  struct Resource_limit* m_resource_limit;
+
+  /**
+   * Get mem root
+   */
+  void* get_memroot();
   
   /**
    * Alloc consekutive pages
@@ -48,7 +80,7 @@ struct Pool_context
    *
    * Will handle resource limit 
    */
-  void* alloc_page(Uint32 *i);
+  void* alloc_page(Uint32 type_id, Uint32 *i);
   
   /**
    * Release pages
@@ -56,8 +88,8 @@ 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);
+  
   /**
    * Alloc consekutive pages
    *
@@ -70,7 +102,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,21 +111,12 @@ 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, Uint32 cnt);
+
   /**
-   * Pool abort
-   *   Only know issue is getPtr with invalid i-value.
-   *   If other emerges, we will add argument to this method
+   * Abort
    */
-  struct AbortArg
-  {
-    Uint32 m_expected_magic;
-    Uint32 m_found_magic;
-    Uint32 i;
-    void * p;
-  };
-  void handle_abort(const AbortArg &);
+  void handleAbort(int code, const char* msg);
 };
 
 template <typename T>
@@ -118,19 +141,181 @@ struct ConstPtr
 /**
  * Any pool should implement the following
  */
-struct Pool
+struct PoolImpl
 {
-public:
-  Pool();
+  Pool_context m_ctx;
+  Record_info m_record_info;
   
   void init(const Record_info& ri, const Pool_context& pc);
-
-  bool seize(Uint32*, void**);
-  void* seize(Uint32*);
-
-  void release(Uint32 i, void* p);
+  void init(const Record_info& ri, const Pool_context& pc);
+  
+  bool seize(Ptr<void>&);
+  void release(Ptr<void>);
   void * getPtr(Uint32 i);
 };
 #endif
 
+template <typename T, typename P>
+class RecordPool {
+public:
+  RecordPool();
+  ~RecordPool();
+  
+  void init(Uint32 type_id, const Pool_context& pc);
+  void wo_pool_init(Uint32 type_id, const Pool_context& pc);
+  
+  /**
+   * Update p value for ptr according to i value 
+   */
+  void getPtr(Ptr<T> &);
+  void getPtr(ConstPtr<T> &) const;
+  
+  /**
+   * Get pointer for i value
+   */
+  T * getPtr(Uint32 i);
+  const T * getConstPtr(Uint32 i) const;
+
+  /**
+   * Update p & i value for ptr according to <b>i</b> value 
+   */
+  void getPtr(Ptr<T> &, Uint32 i);
+  void getPtr(ConstPtr<T> &, Uint32 i) const;
+
+  /**
+   * Allocate an object from pool - update Ptr
+   *
+   * Return i
+   */
+  bool seize(Ptr<T> &);
+
+  /**
+   * Return an object to pool
+   */
+  void release(Uint32 i);
+
+  /**
+   * Return an object to pool
+   */
+  void release(Ptr<T> &);
+private:
+  P m_pool;
+};
+
+template <typename T, typename P>
+inline
+RecordPool<T, P>::RecordPool()
+{
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::init(Uint32 type_id, const Pool_context& pc)
+{
+  Record_info ri;
+  ri.m_size = sizeof(T);
+  ri.m_offset_next_pool = offsetof(T, nextPool);
+  ri.m_offset_magic = offsetof(T, m_magic);
+  ri.m_type_id = type_id;
+  m_pool.init(ri, pc);
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::wo_pool_init(Uint32 type_id, const Pool_context& pc)
+{
+  Record_info ri;
+  ri.m_size = sizeof(T);
+  ri.m_offset_next_pool = 0;
+  ri.m_offset_magic = offsetof(T, m_magic);
+  ri.m_type_id = type_id;
+  m_pool.init(ri, pc);
+}
+
+template <typename T, typename P>
+inline
+RecordPool<T, P>::~RecordPool()
+{
+}
+
+  
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::getPtr(Ptr<T> & ptr)
+{
+  ptr.p = static_cast<T*>(m_pool.getPtr(ptr.i));
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::getPtr(ConstPtr<T> & ptr) const 
+{
+  ptr.p = static_cast<const T*>(m_pool.getPtr(ptr.i));
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::getPtr(Ptr<T> & ptr, Uint32 i)
+{
+  ptr.i = i;
+  ptr.p = static_cast<T*>(m_pool.getPtr(ptr.i));  
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::getPtr(ConstPtr<T> & ptr, Uint32 i) const 
+{
+  ptr.i = i;
+  ptr.p = static_cast<const T*>(m_pool.getPtr(ptr.i));  
+}
+  
+template <typename T, typename P>
+inline
+T * 
+RecordPool<T, P>::getPtr(Uint32 i)
+{
+  return static_cast<T*>(m_pool.getPtr(i));  
+}
+
+template <typename T, typename P>
+inline
+const T * 
+RecordPool<T, P>::getConstPtr(Uint32 i) const 
+{
+  return static_cast<const T*>(m_pool.getPtr(i)); 
+}
+  
+template <typename T, typename P>
+inline
+bool
+RecordPool<T, P>::seize(Ptr<T> & ptr)
+{
+  return m_pool.seize(*(Ptr<void>*)&ptr);
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::release(Uint32 i)
+{
+  Ptr<void> ptr;
+  ptr.i = i;
+  ptr.p = m_pool.getPtr(i);
+  m_pool.release(ptr);
+}
+
+template <typename T, typename P>
+inline
+void
+RecordPool<T, P>::release(Ptr<T> & ptr)
+{
+  m_pool.release(*(Ptr<void>*)&ptr);
+}
+
 #endif
diff --git a/storage/ndb/src/kernel/vm/RWPool.cpp b/storage/ndb/src/kernel/vm/RWPool.cpp
new file mode 100644
index 00000000000..c56dcc3c6b9
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/RWPool.cpp
@@ -0,0 +1,225 @@
+/* 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 "RWPool.hpp"
+#include <ndbd_exit_codes.h>
+#include <NdbOut.hpp>
+
+#define REC_NIL GLOBAL_PAGE_SIZE_WORDS
+
+RWPool::RWPool() 
+{
+  bzero(this, sizeof(* this));
+  m_current_pos = GLOBAL_PAGE_SIZE_WORDS;
+  m_current_first_free = REC_NIL;
+  m_first_free_page = RNIL;
+}
+
+void
+RWPool::init(const Record_info& ri, const Pool_context& pc)
+{
+  m_ctx = pc;
+  m_record_info = ri;
+  m_record_info.m_size = ((ri.m_size + 3) >> 2); // Align to word boundary
+  m_record_info.m_offset_magic = ((ri.m_offset_magic + 3) >> 2);
+  m_record_info.m_offset_next_pool = ((ri.m_offset_next_pool + 3) >> 2);
+  m_memroot = (RWPage*)m_ctx.get_memroot();
+}
+
+bool
+RWPool::seize(Ptr<void>& ptr)
+{
+  Uint32 pos = m_current_pos;
+  Uint32 size = m_record_info.m_size;
+  Uint32 off = m_record_info.m_offset_magic;
+  RWPage *pageP = m_current_page;
+  if (likely(m_current_first_free != REC_NIL))
+  {
+seize_free:
+    pos = m_current_first_free;
+    ptr.i = (m_current_page_no << POOL_RECORD_BITS) + pos;
+    ptr.p = pageP->m_data + pos;
+    pageP->m_data[pos+off] = ~(Uint32)m_record_info.m_type_id;
+    m_current_ref_count++;
+    m_current_first_free = pageP->m_data[pos+m_record_info.m_offset_next_pool];
+    return true;
+  }
+  else if (pos + size < GLOBAL_PAGE_SIZE_WORDS)
+  {
+seize_first:
+    ptr.i = (m_current_page_no << POOL_RECORD_BITS) + pos;
+    ptr.p = (pageP->m_data + pos);
+    pageP->m_data[pos+off] = ~(Uint32)m_record_info.m_type_id;
+    m_current_ref_count++;
+    m_current_pos = pos + size;
+    return true;
+  }
+
+  if (m_current_page)
+  {
+    m_current_page->m_first_free = REC_NIL;
+    m_current_page->m_next_page = RNIL;
+    m_current_page->m_prev_page = RNIL;
+    m_current_page->m_type_id = m_record_info.m_type_id;
+    m_current_page->m_ref_count = m_current_ref_count;
+  }
+
+  if (m_first_free_page != RNIL)
+  {
+    pageP = m_current_page = m_memroot + m_first_free_page;
+    m_current_page_no = m_first_free_page;
+    m_current_pos = GLOBAL_PAGE_SIZE_WORDS;
+    m_current_first_free = m_current_page->m_first_free;
+    m_first_free_page = m_current_page->m_next_page;
+    m_current_ref_count = m_current_page->m_ref_count;
+    (m_memroot + m_first_free_page)->m_prev_page = RNIL;
+    goto seize_free;
+  }
+
+  m_current_ref_count = 0;
+  
+  RWPage* page;
+  Uint32 page_no = RNIL;
+  if ((page = (RWPage*)m_ctx.alloc_page(m_record_info.m_type_id, &page_no)))
+  {
+    pos = 0;
+    m_current_page_no = page_no;
+    pageP = m_current_page = page;
+    m_current_first_free = REC_NIL;
+    page->m_type_id = m_record_info.m_type_id;
+    goto seize_first;
+  }
+
+  m_current_page = 0;
+  m_current_page_no = RNIL;
+  m_current_pos = GLOBAL_PAGE_SIZE_WORDS;
+  m_current_first_free = REC_NIL;
+  
+  return false;
+}
+
+void
+RWPool::release(Ptr<void> ptr)
+{
+  Uint32 cur_page = m_current_page_no;
+  Uint32 ptr_page = ptr.i >> POOL_RECORD_BITS;
+  Uint32 *record_ptr = (Uint32*)ptr.p;
+  Uint32 magic_val = * (record_ptr + m_record_info.m_offset_magic);
+  
+  if (likely(magic_val == ~(Uint32)m_record_info.m_type_id))
+  {
+    * (record_ptr + m_record_info.m_offset_magic) = 0;
+    if (cur_page == ptr_page)
+    {
+      * (record_ptr + m_record_info.m_offset_next_pool) = m_current_first_free;
+      assert(m_current_ref_count);
+      m_current_ref_count--;
+      m_current_first_free = ptr.i & POOL_RECORD_MASK;
+      return;
+    }
+
+    // Cache miss on page...
+    RWPage* page = m_memroot + ptr_page;
+    Uint32 ref_cnt = page->m_ref_count;
+    Uint32 ff = page->m_first_free;
+
+    * (record_ptr + m_record_info.m_offset_next_pool) = ff;
+    page->m_first_free = ptr.i;
+    page->m_ref_count = ref_cnt - 1;
+    
+    if (ff == REC_NIL)
+    {
+      /**
+       * It was full...add to free page list
+       */
+      Uint32 ffp = m_first_free_page;
+      if (ffp != RNIL)
+      {
+	RWPage* next = (m_memroot + ffp);
+	assert(next->m_prev_page == RNIL);
+	next->m_prev_page = ptr_page;
+      }
+      page->m_next_page = ffp;
+      page->m_prev_page = RNIL;
+      return;
+    }
+    else if(ref_cnt == 1)
+    {
+      /**
+       * It's now empty...release it
+       */
+      Uint32 prev = page->m_prev_page;
+      Uint32 next = page->m_next_page;
+      if (prev != RNIL)
+      {
+	(m_memroot + prev)->m_next_page = next;
+      }
+      else
+      {
+	assert(m_first_free_page == ptr_page);
+	m_first_free_page = next;
+      }
+      
+      if (next != RNIL)
+      {
+	(m_memroot + next)->m_prev_page = prev;
+      }
+      m_ctx.release_page(m_record_info.m_type_id, ptr_page);
+      return;
+    }
+    return;
+  }
+  handle_invalid_release(ptr);
+}
+
+void
+RWPool::handle_invalid_release(Ptr<void> ptr)
+{
+  char buf[255];
+
+  Uint32 pos = ptr.i & POOL_RECORD_MASK;
+  Uint32 pageI = ptr.i >> POOL_RECORD_BITS;
+  Uint32 * record_ptr_p = (Uint32*)ptr.p;
+  Uint32 * record_ptr_i = (m_memroot+pageI)->m_data + pos;
+  
+  Uint32 magic = * (record_ptr_p + m_record_info.m_offset_magic);
+  snprintf(buf, sizeof(buf), 
+	   "Invalid memory release: ptr (%x %p %p) magic: (%.8x %.8x) memroot: %p page: %x",
+	   ptr.i, ptr.p, record_ptr_i, magic, m_record_info.m_type_id,
+	   m_memroot,
+	   (m_memroot+pageI)->m_type_id);
+  
+  m_ctx.handleAbort(NDBD_EXIT_PRGERR, buf);
+}
+
+void
+RWPool::handle_invalid_get_ptr(Uint32 ptrI)
+{
+  char buf[255];
+
+  Uint32 pos = ptrI & POOL_RECORD_MASK;
+  Uint32 pageI = ptrI >> POOL_RECORD_BITS;
+  Uint32 * record_ptr_i = (m_memroot+pageI)->m_data + pos;
+  
+  Uint32 magic = * (record_ptr_i + m_record_info.m_offset_magic);
+  snprintf(buf, sizeof(buf), 
+	   "Invalid memory access: ptr (%x %p) magic: (%.8x %.8x) memroot: %p page: %x",
+	   ptrI, record_ptr_i, magic, m_record_info.m_type_id,
+	   m_memroot,
+	   (m_memroot+pageI)->m_type_id);
+  
+  m_ctx.handleAbort(NDBD_EXIT_PRGERR, buf);
+}
diff --git a/storage/ndb/src/kernel/vm/RWPool.hpp b/storage/ndb/src/kernel/vm/RWPool.hpp
new file mode 100644
index 00000000000..2902f0a5d7e
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/RWPool.hpp
@@ -0,0 +1,70 @@
+/* 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"
+
+struct RWPage
+{
+  Uint32 m_type_id;
+  Uint16 m_first_free;
+  Uint16 m_ref_count;
+  Uint32 m_next_page;
+  Uint32 m_prev_page;
+  Uint32 m_data[GLOBAL_PAGE_SIZE_WORDS - 4];
+};
+
+/**
+ * Read Write  Pool
+ */
+struct RWPool
+{
+  Record_info m_record_info;
+  RWPage* m_memroot;
+  RWPage* m_current_page;
+  Pool_context m_ctx;
+  Uint32 m_first_free_page;
+  Uint32 m_current_page_no;
+  Uint16 m_current_pos;
+  Uint16 m_current_first_free;
+  Uint16 m_current_ref_count;
+public:
+  RWPool();
+  
+  void init(const Record_info& ri, const Pool_context& pc);
+  bool seize(Ptr<void>&);
+  void release(Ptr<void>);
+  void * getPtr(Uint32 i);
+  
+private:  
+  void handle_invalid_release(Ptr<void>);
+  void handle_invalid_get_ptr(Uint32 i);
+};
+
+inline
+void*
+RWPool::getPtr(Uint32 i)
+{
+  Uint32 page_no = i >> POOL_RECORD_BITS;
+  Uint32 page_idx = i & POOL_RECORD_MASK;
+  RWPage * page = m_memroot + page_no;
+  Uint32 * record = page->m_data + page_idx;
+  Uint32 magic_val = * (record + m_record_info.m_offset_magic);
+  if (likely(magic_val == ~(Uint32)m_record_info.m_type_id))
+  {
+    return record;
+  }
+  handle_invalid_get_ptr(i);
+}
diff --git a/storage/ndb/src/kernel/vm/SLFifoList.hpp b/storage/ndb/src/kernel/vm/SLFifoList.hpp
new file mode 100644
index 00000000000..f222715dd6d
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/SLFifoList.hpp
@@ -0,0 +1,343 @@
+/* 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 */
+
+#ifndef SLFIFOLIST_HPP
+#define SLFIFOLIST_HPP
+
+#include <ndb_global.h>
+#include <kernel_types.h>
+#include "Pool.hpp"
+
+/**
+ * Template class used for implementing an
+ *   list of object retreived from a pool
+ */
+template <typename P, typename T, typename U = T>
+class SLFifoListImpl 
+{
+public:
+  /**
+   * List head
+   */
+  struct Head 
+  {
+    Head();
+    Uint32 firstItem;
+    Uint32 lastItem;
+
+#ifdef VM_TRACE
+    bool in_use;
+#endif
+
+    inline bool isEmpty() const { return firstItem == RNIL;}
+  };
+  
+  SLFifoListImpl(P & thePool);
+  
+  bool seizeFirst(Ptr<T> &);
+  bool seizeLast(Ptr<T> &);
+  bool seize(Ptr<T> & ptr) { return seizeLast(ptr);}
+  
+  void releaseFirst(Ptr<T> &);
+  
+  void addFirst(Ptr<T> &);
+  void addLast(Ptr<T> &);
+  
+  void removeFirst(Ptr<T> &);
+  
+  /**
+   *  Update i & p value according to <b>i</b>
+   */
+  void getPtr(Ptr<T> &, Uint32 i) const;
+  
+  /**
+   * Update p value for ptr according to i value 
+   */
+  void getPtr(Ptr<T> &) const ;
+  
+  /**
+   * Get pointer for i value
+   */
+  T * getPtr(Uint32 i) const ;
+
+  /**
+   * Update ptr to first element in list
+   *
+   * Return i
+   */
+  bool first(Ptr<T> &) const ;
+
+  /**
+   * Update ptr to first element in list
+   *
+   * Return i
+   */
+  bool last(Ptr<T> &) const ;
+
+  /**
+   * Get next element
+   *
+   * NOTE ptr must be both p & i
+   */
+  bool next(Ptr<T> &) const ;
+  
+  /**
+   * Check if next exists i.e. this is not last
+   *
+   * NOTE ptr must be both p & i
+   */
+  bool hasNext(const Ptr<T> &) const;
+  
+  inline bool isEmpty() const { return head.firstItem == RNIL;}
+  
+protected:
+  Head head;
+  P & thePool;
+};
+
+template <typename P, typename T, typename U = T>
+class LocalSLFifoListImpl : public SLFifoListImpl<P,T,U> 
+{
+public:
+  LocalSLFifoListImpl(P & thePool, typename SLFifoListImpl<P,T,U>::Head &_src)
+    : SLFifoListImpl<P,T,U>(thePool), src(_src)
+  {
+    this->head = src;
+#ifdef VM_TRACE
+    assert(src.in_use == false);
+    src.in_use = true;
+#endif
+  }
+  
+  ~LocalSLFifoListImpl(){
+#ifdef VM_TRACE
+    assert(src.in_use == true);
+#endif
+    src = this->head;
+  }
+private:
+  typename SLFifoListImpl<P,T,U>::Head & src;
+};
+
+template <typename P, typename T, typename U>
+inline
+SLFifoListImpl<P,T,U>::SLFifoListImpl(P & _pool):
+  thePool(_pool)
+{
+}
+
+template <typename P, typename T, typename U>
+inline
+SLFifoListImpl<P,T,U>::Head::Head()
+{
+  firstItem = RNIL;
+  lastItem = RNIL;
+#ifdef VM_TRACE
+  in_use = false;
+#endif
+}
+
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::seizeFirst(Ptr<T> & p)
+{
+  if (likely(thePool.seize(p)))
+  {
+    addFirst(p);
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::seizeLast(Ptr<T> & p)
+{
+  if (likely(thePool.seize(p)))
+  {
+    addLast(p);
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
+template <typename P, typename T, typename U>
+inline
+void
+SLFifoListImpl<P,T,U>::addFirst(Ptr<T> & p)
+{
+  Uint32 first = head.firstItem;
+  head.firstItem = p.i;
+  if (first == RNIL)
+  {
+    head.lastItem = p.i;
+  }
+  p.p->U::nextList = first;
+}
+
+template <typename P, typename T, typename U>
+inline
+void
+SLFifoListImpl<P,T,U>::addLast(Ptr<T> & p)
+{
+  T * t = p.p;
+  Uint32 last = head.lastItem;
+  
+  t->U::nextList = RNIL;
+  head.lastItem = p.i;    
+  
+  if(last != RNIL)
+  {
+    T * t2 = thePool.getPtr(last);
+    t2->U::nextList = p.i;
+  }
+  else
+  {
+    head.firstItem = p.i;
+  }
+}
+
+template <typename P, typename T, typename U>
+inline
+void
+SLFifoListImpl<P,T,U>::removeFirst(Ptr<T> & p)
+{
+  Uint32 first = head.firstItem;
+  Uint32 last = head.lastItem;
+  assert(p.i == first);
+  if (first != last)
+  {
+    head.firstItem = p.p->U::nextList;
+  }
+  else
+  {
+    head.firstItem = head.lastItem = RNIL;
+  }
+}
+
+template <typename P, typename T, typename U>
+inline
+void 
+SLFifoListImpl<P,T,U>::releaseFirst(Ptr<T> & p)
+{
+  removeFirst(p);
+  thePool.release(p);
+}
+
+template <typename P, typename T, typename U>
+inline
+void 
+SLFifoListImpl<P,T,U>::getPtr(Ptr<T> & p, Uint32 i) const 
+{
+  p.i = i;
+  p.p = thePool.getPtr(i);
+}
+
+template <typename P, typename T, typename U>
+inline
+void 
+SLFifoListImpl<P,T,U>::getPtr(Ptr<T> & p) const 
+{
+  thePool.getPtr(p);
+}
+  
+template <typename P, typename T, typename U>
+inline
+T * 
+SLFifoListImpl<P,T,U>::getPtr(Uint32 i) const 
+{
+  return thePool.getPtr(i);
+}
+
+/**
+ * Update ptr to first element in list
+ *
+ * Return i
+ */
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::first(Ptr<T> & p) const 
+{
+  p.i = head.firstItem;
+  if(p.i != RNIL)
+  {
+    p.p = thePool.getPtr(p.i);
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::last(Ptr<T> & p) const 
+{
+  p.i = head.lastItem;
+  if(p.i != RNIL)
+  {
+    p.p = thePool.getPtr(p.i);
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::next(Ptr<T> & p) const 
+{
+  p.i = p.p->U::nextList;
+  if(p.i != RNIL)
+  {
+    p.p = thePool.getPtr(p.i);
+    return true;
+  }
+  p.p = NULL;
+  return false;
+}
+
+template <typename P, typename T, typename U>
+inline
+bool
+SLFifoListImpl<P,T,U>::hasNext(const Ptr<T> & p) const 
+{
+  return p.p->U::nextList != RNIL;
+}
+
+// Specializations
+
+template <typename T, typename U = T>
+class SLFifoList : public SLFifoListImpl<ArrayPool<T>, T, U>
+{
+public:
+  SLFifoList(ArrayPool<T> & p) : SLFifoListImpl<ArrayPool<T>, T, U>(p) {}
+};
+
+template <typename T, typename U = T>
+class LocalSLFifoList : public LocalSLFifoListImpl<ArrayPool<T>,T,U> {
+public:
+  LocalSLFifoList(ArrayPool<T> & p, typename SLFifoList<T,U>::Head & _src)
+    : LocalSLFifoListImpl<ArrayPool<T>,T,U>(p, _src) {}
+};
+
+#endif
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.cpp b/storage/ndb/src/kernel/vm/SimulatedBlock.cpp
index 15c7027afd8..1de47197867 100644
--- a/storage/ndb/src/kernel/vm/SimulatedBlock.cpp
+++ b/storage/ndb/src/kernel/vm/SimulatedBlock.cpp
@@ -52,6 +52,7 @@ SimulatedBlock::SimulatedBlock(BlockNumber blockNumber,
     theReference(numberToRef(blockNumber, globalData.ownId)),
     m_ctx(ctx),
     m_global_page_pool(globalData.m_global_page_pool),
+    m_shared_page_pool(globalData.m_shared_page_pool),
     c_fragmentInfoHash(c_fragmentInfoPool),
     c_linearFragmentSendList(c_fragmentSendPool),
     c_segmentedFragmentSendList(c_fragmentSendPool),
@@ -2031,3 +2032,6 @@ SimulatedBlock::create_distr_key(Uint32 tableId,
   }
   return dstPos;
 }
+
+CArray<KeyDescriptor> g_key_descriptor_pool;
+
diff --git a/storage/ndb/src/kernel/vm/SimulatedBlock.hpp b/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
index 5901ae4da0b..a567650a6c7 100644
--- a/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
+++ b/storage/ndb/src/kernel/vm/SimulatedBlock.hpp
@@ -53,6 +53,7 @@
 #include <signaldata/ReadConfig.hpp>
 #include <signaldata/UpgradeStartup.hpp>
 #include "ndbd_malloc_impl.hpp"
+#include <blocks/record_types.hpp>
 
 /**
  * Something for filesystem access
@@ -93,6 +94,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();
@@ -427,7 +429,8 @@ private:
 
 protected:  
   ArrayPool<GlobalPage>& m_global_page_pool;
-
+  ArrayPool<GlobalPage>& m_shared_page_pool;
+  
 private:
   /**
    * Node state
diff --git a/storage/ndb/src/kernel/vm/WOPool.cpp b/storage/ndb/src/kernel/vm/WOPool.cpp
new file mode 100644
index 00000000000..b97b3d62d3a
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/WOPool.cpp
@@ -0,0 +1,137 @@
+/* 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 "WOPool.hpp"
+#include <ndbd_exit_codes.h>
+#include <NdbOut.hpp>
+
+WOPool::WOPool() 
+{
+  bzero(this, sizeof(* this));
+  m_current_pos = GLOBAL_PAGE_SIZE_WORDS;
+}
+
+void
+WOPool::init(const Record_info& ri, const Pool_context& pc)
+{
+  m_ctx = pc;
+  m_record_info = ri;
+  m_record_info.m_size = ((ri.m_size + 3) >> 2); // Align to word boundary
+  m_record_info.m_offset_magic = ((ri.m_offset_magic + 3) >> 2);
+  m_memroot = (WOPage*)m_ctx.get_memroot();
+  ndbout_c("WOPool::init(%x, %d)",ri.m_type_id, m_record_info.m_size);
+}
+
+bool
+WOPool::seize_new_page(Ptr<void>& ptr)
+{
+  WOPage* page;
+  Uint32 page_no = RNIL;
+  if ((page = (WOPage*)m_ctx.alloc_page(m_record_info.m_type_id, &page_no)))
+  {
+    if (m_current_page)
+    {
+      m_current_page->m_ref_count = m_current_ref_count;
+    }
+    
+    m_current_pos = 0;
+    m_current_ref_count = 0;
+    m_current_page_no = page_no;
+    m_current_page = page;
+    page->m_type_id = m_record_info.m_type_id;
+    bool ret = seize(ptr);
+    assert(ret);
+    return true;
+  }
+  return false;
+}
+
+void
+WOPool::release_not_current(Ptr<void> ptr)
+{
+  WOPage* page = (WOPage*)(UintPtr(ptr.p) & ~(GLOBAL_PAGE_SIZE - 1));
+  Uint32 cnt = page->m_ref_count;
+  Uint32 type = page->m_type_id;
+  Uint32 ri_type = m_record_info.m_type_id;
+  if (likely(cnt && type == ri_type))
+  {
+    if (cnt == 1)
+    {
+      m_ctx.release_page(ri_type, ptr.i >> POOL_RECORD_BITS);
+      return;
+    }
+    page->m_ref_count = cnt - 1;
+    return;
+  }
+  
+  handle_inconsistent_release(ptr);
+}
+
+void
+WOPool::handle_invalid_release(Ptr<void> ptr)
+{
+  char buf[255];
+
+  Uint32 pos = ptr.i & POOL_RECORD_MASK;
+  Uint32 pageI = ptr.i >> POOL_RECORD_BITS;
+  Uint32 * record_ptr_p = (Uint32*)ptr.p;
+  Uint32 * record_ptr_i = (m_memroot+pageI)->m_data + pos;
+  
+  Uint32 magic = * (record_ptr_p + m_record_info.m_offset_magic);
+  snprintf(buf, sizeof(buf), 
+	   "Invalid memory release: ptr (%x %p %p) magic: (%.8x %.8x) memroot: %p page: %x",
+	   ptr.i, ptr.p, record_ptr_i, magic, m_record_info.m_type_id,
+	   m_memroot,
+	   (m_memroot+pageI)->m_type_id);
+  
+  m_ctx.handleAbort(NDBD_EXIT_PRGERR, buf);
+}
+
+void
+WOPool::handle_invalid_get_ptr(Uint32 ptrI)
+{
+  char buf[255];
+
+  Uint32 pos = ptrI & POOL_RECORD_MASK;
+  Uint32 pageI = ptrI >> POOL_RECORD_BITS;
+  Uint32 * record_ptr_i = (m_memroot+pageI)->m_data + pos;
+  
+  Uint32 magic = * (record_ptr_i + m_record_info.m_offset_magic);
+  snprintf(buf, sizeof(buf), 
+	   "Invalid memory access: ptr (%x %p) magic: (%.8x %.8x) memroot: %p page: %x",
+	   ptrI, record_ptr_i, magic, m_record_info.m_type_id,
+	   m_memroot,
+	   (m_memroot+pageI)->m_type_id);
+  
+  m_ctx.handleAbort(NDBD_EXIT_PRGERR, buf);
+}
+
+void
+WOPool::handle_inconsistent_release(Ptr<void> ptr)
+{
+  WOPage* page = (WOPage*)(UintPtr(ptr.p) & ~(GLOBAL_PAGE_SIZE - 1));
+  Uint32 cnt = page->m_ref_count;
+  Uint32 type = page->m_type_id;
+  Uint32 ri_type = m_record_info.m_type_id;
+
+  char buf[255];
+  
+  snprintf(buf, sizeof(buf), 
+	   "Memory corruption: ptr (%x %p) page (%d %x %x)",
+	   ptr.i, ptr.p, cnt, type, ri_type);
+  
+  m_ctx.handleAbort(NDBD_EXIT_PRGERR, buf);
+}
diff --git a/storage/ndb/src/kernel/vm/WOPool.hpp b/storage/ndb/src/kernel/vm/WOPool.hpp
new file mode 100644
index 00000000000..e05958d65a3
--- /dev/null
+++ b/storage/ndb/src/kernel/vm/WOPool.hpp
@@ -0,0 +1,115 @@
+/* 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"
+
+struct WOPage
+{
+  Uint32 m_type_id;
+  Uint32 m_ref_count;
+  Uint32 m_data[GLOBAL_PAGE_SIZE_WORDS - 2];
+};
+
+/**
+ * Write Once Pool
+ */
+struct WOPool
+{
+  Record_info m_record_info;
+  WOPage* m_memroot;
+  WOPage* m_current_page;
+  Pool_context m_ctx;
+  Uint32 m_current_page_no;
+  Uint16 m_current_pos;
+  Uint16 m_current_ref_count;
+public:
+  WOPool();
+  
+  void init(const Record_info& ri, const Pool_context& pc);
+  bool seize(Ptr<void>&);
+  void release(Ptr<void>);
+  void * getPtr(Uint32 i);
+  
+private:  
+  bool seize_new_page(Ptr<void>&);
+  void release_not_current(Ptr<void>);
+
+  void handle_invalid_release(Ptr<void>);
+  void handle_invalid_get_ptr(Uint32 i);
+  void handle_inconsistent_release(Ptr<void>);
+};
+
+inline
+bool
+WOPool::seize(Ptr<void>& ptr)
+{
+  Uint32 pos = m_current_pos;
+  Uint32 size = m_record_info.m_size;
+  WOPage *pageP = m_current_page;
+  if (likely(pos + size < GLOBAL_PAGE_SIZE_WORDS))
+  {
+    ptr.i = (m_current_page_no << POOL_RECORD_BITS) + pos;
+    ptr.p = (pageP->m_data + pos);
+    pageP->m_data[pos+m_record_info.m_offset_magic] = ~(Uint32)m_record_info.m_type_id;
+    m_current_pos = pos + size;
+    m_current_ref_count++;
+    return true;
+  }
+  
+  return seize_new_page(ptr);
+}
+
+inline
+void
+WOPool::release(Ptr<void> ptr)
+{
+  Uint32 cur_page = m_current_page_no;
+  Uint32 ptr_page = ptr.i >> POOL_RECORD_BITS;
+  Uint32 *magic_ptr = (((Uint32*)ptr.p)+m_record_info.m_offset_magic);
+  Uint32 magic_val = *magic_ptr;
+  
+  if (likely(magic_val == ~(Uint32)m_record_info.m_type_id))
+  {
+    * magic_ptr = 0;
+    if (cur_page == ptr_page)
+    {
+      if (m_current_ref_count == 1)
+      {
+	m_current_pos = 0;
+      }
+      m_current_ref_count--;
+      return;
+    }
+    return release_not_current(ptr);
+  }
+  handle_invalid_release(ptr);
+}
+
+inline
+void*
+WOPool::getPtr(Uint32 i)
+{
+  Uint32 page_no = i >> POOL_RECORD_BITS;
+  Uint32 page_idx = i & POOL_RECORD_MASK;
+  WOPage * page = m_memroot + page_no;
+  Uint32 * record = page->m_data + page_idx;
+  Uint32 magic_val = * (record + m_record_info.m_offset_magic);
+  if (likely(magic_val == ~(Uint32)m_record_info.m_type_id))
+  {
+    return record;
+  }
+  handle_invalid_get_ptr(i);
+}
diff --git a/storage/ndb/src/kernel/vm/bench_pool.cpp b/storage/ndb/src/kernel/vm/bench_pool.cpp
index 1042059b0c5..d728710da61 100644
--- a/storage/ndb/src/kernel/vm/bench_pool.cpp
+++ b/storage/ndb/src/kernel/vm/bench_pool.cpp
@@ -15,66 +15,150 @@
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 
-#include "NdbdSuperPool.hpp"
 #include "ArrayPool.hpp"
+#include "WOPool.hpp"
+#include "RWPool.hpp"
 #include <NdbTick.h>
 #include "ndbd_malloc_impl.hpp"
+#include "SimulatedBlock.hpp"
+
+#ifdef USE_CALLGRIND
+#include <valgrind/callgrind.h>
+#else
+#define CALLGRIND_TOGGLE_COLLECT()
+#endif
+
+#define T_TEST_AP   (1 << 0)
+#define T_TEST_WO   (1 << 1)
+#define T_TEST_RW   (1 << 2)
+
+#define T_SEIZE       (1 << 0)
+#define T_RELEASE     (1 << 1)
+#define T_G_RELEASE   (1 << 2)
+#define T_R_RELEASE   (1 << 3)
+#define T_R_G_RELEASE (1 << 4)
+#define T_MIX         (1 << 5)
+#define T_GETPTR      (1 << 6)
+#define T_FIFO        (1 << 7)
+
+const char *test_names[] = {
+  "seize",
+  "release",
+  "get+rel",
+  "r-rel",
+  "r-get+rel",
+  "mix",
+  "getptr",
+  "fifo",
+  0
+};
+
+Uint32 pools = ~0;
+Uint32 tests = ~0;
+Uint32 records = ~0;
+Uint32 sizes = 7;
+unsigned int seed;
+Ndbd_mem_manager mm;
+Configuration cfg;
+Block_context ctx = { cfg, mm };
+struct BB : public SimulatedBlock
+{
+  BB(int no, Block_context& ctx) : SimulatedBlock(no, ctx) {}
+};
+
+BB block(DBACC, ctx);
 
 template <typename T>
-inline
 void
 init(ArrayPool<T> & pool, Uint32 cnt)
 {
-  pool.setSize(cnt + 1);
+  pool.setSize(cnt + 1, true);
 }
 
 template <typename T>
-inline
 void
-init(RecordPool<T> & pool, Uint32 cnt)
+init(RecordPool<T, WOPool> & pool, Uint32 cnt)
 {
+  Pool_context pc;
+  pc.m_block = &block;
+  pool.wo_pool_init(0x2001, pc);
 }
 
+template <typename T>
+void
+init(RecordPool<T, RWPool> & pool, Uint32 cnt)
+{
+  Pool_context pc;
+  pc.m_block = &block;
+  pool.init(0x2001, pc);
+}
 
-template<typename T, typename R>
-inline
+template <typename T, typename R>
 void 
 test_pool(R& pool, Uint32 cnt, Uint32 loops)
 {
-  init(pool, cnt);
   Ptr<T> ptr;
   Uint32 *arr = (Uint32*)alloca(cnt * sizeof(Uint32));
+  bzero(arr, cnt * sizeof(Uint32));
+  if (tests & T_SEIZE)
   {
-    printf(" ; seize "); fflush(stdout);
     Uint64 sum = 0;
     for(Uint32 i = 0; i<loops; i++)
-    {
+    { // seize
       Uint64 start = NdbTick_CurrentMillisecond();
+      CALLGRIND_TOGGLE_COLLECT();
       for(Uint32 j = 0; j<cnt; j++)
       {
 	bool b = pool.seize(ptr);
 	arr[j] = ptr.i;
+	ptr.p->do_stuff();
 	assert(b);
       }
+      CALLGRIND_TOGGLE_COLLECT();
       Uint64 stop = NdbTick_CurrentMillisecond();      
 
       for(Uint32 j = 0; j<cnt; j++)
       {
 	ptr.i = arr[j];
 	pool.getPtr(ptr);
-	pool.release(ptr);
+	ptr.p->do_stuff();
+	pool.release(ptr.i);
 	arr[j] = RNIL;
       }
-      
       sum += (stop - start);
-      if (i == 0)
-	printf("; first ; %lld", (stop - start));
     }
-    printf(" ; avg ; %lld ; tot ; %lld", sum/loops, sum);fflush(stdout);
+    printf(" ; %lld", sum); fflush(stdout);
   }
+  
+  if (tests & T_RELEASE)
+  { // release
+    Uint64 sum = 0;
+    for(Uint32 i = 0; i<loops; i++)
+    {
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	bool b = pool.seize(ptr);
+	arr[j] = ptr.i;
+	ptr.p->do_stuff();
+      }
 
-  {
-    printf(" ; release "); fflush(stdout);
+      Uint64 start = NdbTick_CurrentMillisecond();
+      CALLGRIND_TOGGLE_COLLECT();
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	pool.release(arr[j]);
+	arr[j] = RNIL;
+      }
+      CALLGRIND_TOGGLE_COLLECT();
+      Uint64 stop = NdbTick_CurrentMillisecond();      
+      
+      sum += (stop - start);
+    }
+    printf(" ; %lld", sum); fflush(stdout);
+  }
+  
+  if (tests & T_G_RELEASE)
+  { // getptr + release
     Uint64 sum = 0;
     for(Uint32 i = 0; i<loops; i++)
     {
@@ -82,44 +166,108 @@ test_pool(R& pool, Uint32 cnt, Uint32 loops)
       {
 	bool b = pool.seize(ptr);
 	arr[j] = ptr.i;
-	assert(b);
+	ptr.p->do_stuff();
       }
 
       Uint64 start = NdbTick_CurrentMillisecond();
+      CALLGRIND_TOGGLE_COLLECT();
       for(Uint32 j = 0; j<cnt; j++)
       {
-	ptr.i = arr[j];
+	pool.getPtr(ptr, arr[j]);
+	ptr.p->do_stuff();
 	pool.release(ptr);
 	arr[j] = RNIL;
       }
+      CALLGRIND_TOGGLE_COLLECT();
+      Uint64 stop = NdbTick_CurrentMillisecond();      
+      
+      sum += (stop - start);
+    }
+    printf(" ; %lld", sum); fflush(stdout);
+  }
+
+  if (tests & T_R_RELEASE)
+  { // release reverse
+    Uint64 sum = 0;
+    for(Uint32 i = 0; i<loops; i++)
+    {
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	bool b = pool.seize(ptr);
+	arr[j] = ptr.i;
+	ptr.p->do_stuff();
+      }
+
+      Uint64 start = NdbTick_CurrentMillisecond();
+      CALLGRIND_TOGGLE_COLLECT();
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	pool.release(arr[cnt - j - 1]);
+	arr[cnt - j - 1] = RNIL;
+      }
+      CALLGRIND_TOGGLE_COLLECT();
       Uint64 stop = NdbTick_CurrentMillisecond();      
       
       sum += (stop - start);
     }
-    printf("; avg ; %lld ; tot ; %lld", sum/loops, sum); fflush(stdout);
+    printf(" ; %lld", sum); fflush(stdout);
   }
+  
+  if (tests & T_R_G_RELEASE)
+  { // getptr + release
+    Uint64 sum = 0;
+    for(Uint32 i = 0; i<loops; i++)
+    {
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	bool b = pool.seize(ptr);
+	arr[j] = ptr.i;
+	ptr.p->do_stuff();
+      }
 
+      Uint64 start = NdbTick_CurrentMillisecond();
+      CALLGRIND_TOGGLE_COLLECT();
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	pool.getPtr(ptr, arr[cnt - j - 1]);
+	ptr.p->do_stuff();
+	pool.release(ptr);
+	arr[cnt - j - 1] = RNIL;
+      }
+      CALLGRIND_TOGGLE_COLLECT();
+      Uint64 stop = NdbTick_CurrentMillisecond();      
+      
+      sum += (stop - start);
+    }
+    printf(" ; %lld", sum); fflush(stdout);
+  }
+  
+  if (tests & T_MIX)
   {
-    printf(" ; mix"); fflush(stdout);
-    
     Uint64 sum = 0;
     Uint64 start = NdbTick_CurrentMillisecond();
+    Uint32 lseed = seed;
+    CALLGRIND_TOGGLE_COLLECT();
     for(Uint32 i = 0; i<loops * cnt; i++)
     {
-      int pos = rand() % cnt;
+      int pos = rand_r(&lseed) % cnt;
       ptr.i = arr[pos];
       if (ptr.i == RNIL)
       {
 	pool.seize(ptr);
 	arr[pos] = ptr.i;
 	assert(ptr.i != RNIL);
+	ptr.p->do_stuff();
       }
       else
       {
+	pool.getPtr(ptr);
+	ptr.p->do_stuff();
 	pool.release(ptr);
 	arr[pos] = RNIL;
       }
     }
+    CALLGRIND_TOGGLE_COLLECT();
     Uint64 stop = NdbTick_CurrentMillisecond();    
     
     for(Uint32 j = 0; j<cnt; j++)
@@ -128,7 +276,7 @@ test_pool(R& pool, Uint32 cnt, Uint32 loops)
       if (ptr.i != RNIL)
       {
 	pool.getPtr(ptr);
-	pool.release(ptr);
+	pool.release(ptr.i);
       }
       arr[j] = RNIL;
     }
@@ -137,113 +285,322 @@ test_pool(R& pool, Uint32 cnt, Uint32 loops)
     printf(" ; %lld", sum); fflush(stdout);
   }
   
+  if (tests & T_GETPTR)
   {
-    printf(" ; getPtr"); fflush(stdout);
-
+    Uint32 lseed = seed;
     for(Uint32 j = 0; j<cnt; j++)
     {
       bool b = pool.seize(ptr);
       arr[j] = ptr.i;
+      ptr.p->do_stuff();
       assert(b);
     }
 
     Uint64 sum = 0;
     Uint64 start = NdbTick_CurrentMillisecond();
+    CALLGRIND_TOGGLE_COLLECT();
     for(Uint32 i = 0; i<loops * cnt; i++)
     {
-      int pos = rand() % cnt;
+      int pos = rand_r(&lseed) % cnt;
       ptr.i = arr[pos];
       pool.getPtr(ptr);
+      ptr.p->do_stuff();
     }
+    CALLGRIND_TOGGLE_COLLECT();
     Uint64 stop = NdbTick_CurrentMillisecond();    
-
+    
     for(Uint32 j = 0; j<cnt; j++)
     {
       ptr.i = arr[j];
       pool.getPtr(ptr);
-      pool.release(ptr);
+      ptr.p->do_stuff();
+      pool.release(ptr.i);
       arr[j] = RNIL;
     }
     
     sum += (stop - start);
     printf(" ; %lld", sum); fflush(stdout);
   }
+
+  if (tests & T_FIFO)
+  { // fifo
+    Uint64 sum = 0;
+    Uint64 start = NdbTick_CurrentMillisecond();
+    CALLGRIND_TOGGLE_COLLECT();
+    for(Uint32 i = 0; i<loops; i++)
+    {
+      Uint32 head = RNIL;
+      Uint32 last = RNIL;
+      
+      Uint64 sum = 0;
+      for(Uint32 j = 0; j<cnt; j++)
+      {
+	pool.seize(ptr);
+	ptr.p->do_stuff();
+	ptr.p->m_nextList = RNIL;
+	if (head == RNIL)
+	{
+	  head = ptr.i;
+	}
+	else
+	{
+	  T* t = pool.getPtr(last);
+	  t->m_nextList = ptr.i;
+	}
+	last = ptr.i;
+      }
+
+      while (head != RNIL)
+      {
+	pool.getPtr(ptr, head);
+	ptr.p->do_stuff();
+	head = ptr.p->m_nextList;
+	pool.release(ptr);
+      }
+    }      
+    CALLGRIND_TOGGLE_COLLECT();
+    Uint64 stop = NdbTick_CurrentMillisecond();    
+    sum += (stop - start);
+    printf(" ; %lld", sum); fflush(stdout);
+  }
+  
   ndbout_c("");
 }
 
-template <Uint32 sz> struct Rec { char data[sz-4]; Uint32 nextPool; };
+template <Uint32 sz> 
+struct Rec { 
+  Uint32 m_data;
+  Uint32 m_magic; 
+  Uint32 nextPool; 
+  Uint32 m_nextList;
+  char m_cdata[sz-16];
+
+  void do_stuff() { 
+    Uint32 sum = 0; 
+    Uint32 *ptr = (Uint32*)this; 
+    for(Uint32 i = 0; i<(sz >> 2); i++)
+      sum += * ptr ++;
+    m_data = sum;
+  }
+};
+
 typedef Rec<32> Rec32;
 typedef Rec<36> Rec36;
-typedef Rec<256> Rec256;
-typedef Rec<260> Rec260;
+typedef Rec<56> Rec56;
+typedef Rec<224> Rec224;
 
-Ndbd_mem_manager mem;
+template <typename T>
+void test_ap(Uint32 cnt, Uint32 loop)
+{
+  printf("AP ; %d ; %d", sizeof(T), (cnt * sizeof(T))>>10); fflush(stdout);
+  ArrayPool<T> pool;
+  init(pool, cnt);
+  test_pool<T, ArrayPool<T> >(pool, cnt, loop);
+}
 
 template <typename T>
-inline
-void test_rp(Uint32 cnt, Uint32 loop, Uint32 pgsz)
+void test_rw(Uint32 cnt, Uint32 loop)
 {
-  printf("RP ; %d ; ws ; %d ; page ; %d", 
-	 sizeof(T), (sizeof(T)*cnt) >> 10, pgsz >> 10);
-  NdbdSuperPool sp(mem, pgsz, 19);
-  GroupPool gp(sp);
-  sp.init_1();
-  sp.init_2();
-  
-  sp.setInitPages(4);
-  sp.setIncrPages(4);
-  sp.setMaxPages(~0);
-  sp.allocMemory();
-  
-  RecordPool<T> pool(gp);
-  test_pool<T, RecordPool<T> >(pool, cnt, loop);
+  printf("RW ; %d ; %d", sizeof(T), (cnt * sizeof(T))>>10); fflush(stdout);
+  RecordPool<T, RWPool> pool;
+  init(pool, cnt);
+  test_pool<T, RecordPool<T, RWPool> >(pool, cnt, loop);
 }
 
 template <typename T>
-inline
-void test_ap(Uint32 cnt, Uint32 loop)
+void test_wo(Uint32 cnt, Uint32 loop)
 {
-  printf("AP ; %d ; ws ; %d ; page ; n/a", sizeof(T), (cnt * sizeof(T))>>10);
-  ArrayPool<T> pool;
-  test_pool<T, ArrayPool<T> >(pool, cnt, loop);
+  printf("WO ; %d ; %d", sizeof(T), (cnt * sizeof(T))>>10); fflush(stdout);
+  RecordPool<T, WOPool> pool;
+  init(pool, cnt);
+  test_pool<T, RecordPool<T, WOPool> >(pool, cnt, loop);
 }
 
+#include <EventLogger.hpp>
+extern EventLogger g_eventLogger;
+
 int
 main(int argc, char **argv)
 {
-  mem.init(10000);
+  Uint32 loops = 300000;
+  for (Uint32 i = 1 ; i<argc ; i++)
+  {
+    if (argc > i+1 && strcmp(argv[i], "-pools") == 0)
+    {
+      pools = 0;
+      for (Uint32 j = 0; j<strlen(argv[i+1]); j++)
+      {
+	char c = argv[i+1][j];
+	if (c >= '0' && c <= '9')
+	  pools |= 1 << (c - '0');
+	else 
+	  pools |= 1 << (10 + (c - 'a'));
+      }
+    }
+    else if (argc > i+1 && strcmp(argv[i], "-tests") == 0)
+    {
+      tests = 0;
+      for (Uint32 j = 0; j<strlen(argv[i+1]); j++)
+      {
+	char c = argv[i+1][j];
+	if (c >= '0' && c <= '9')
+	  tests |= 1 << (c - '0');
+	else 
+	  tests |= 1 << (10 + (c - 'a'));
+      }
+    }
+    else if (argc > i+1 && strcmp(argv[i], "-sizes") == 0)
+    {
+      sizes = 0;
+      for (Uint32 j = 0; j<strlen(argv[i+1]); j++)
+      {
+	char c = argv[i+1][j];
+	if (c >= '0' && c <= '9')
+	  sizes |= 1 << (c - '0');
+	else
+	  sizes |= 1 << (10 + (c - 'a'));
+      }
+    }
+    else if (argc > i+1 && strcmp(argv[i], "-records") == 0)
+    {
+      records = 0;
+      for (Uint32 j = 0; j<strlen(argv[i+1]); j++)
+      {
+	char c = argv[i+1][j];
+	if (c >= '0' && c <= '9')
+	  records |= 1 << (c - '0');
+	else
+	  records |= 1 << (10 + (c - 'a'));
+      }
+    }
+    else if (argc > i+1 && strcmp(argv[i], "-loop") == 0)
+    {
+      loops = atoi(argv[i+1]);
+    }
+  }
+  
+  Resource_limit rl;
+  rl.m_min = 0;
+  rl.m_max = 10000;
+  rl.m_resource_id = 0;
+  mm.set_resource_limit(rl);
+  if(!mm.init())
+  {
+    abort();
+  }
+
+  seed = time(0);
+  Uint32 sz = 0;
+  Uint32 cnt = 256;
 
-  Uint32 cnt = 100;
-  Uint32 loop = 300000;
+  printf("pool ; rs ; ws");
+  for (Uint32 i = 0; test_names[i] && i<31; i++)
+    if (tests & (1 << i))
+      printf(" ; %s", test_names[i]);
+  ndbout_c("");
 
   while(cnt <= 1000000)
   {
-    test_rp<Rec32>(cnt, loop, 8192);
-    test_rp<Rec32>(cnt, loop, 32768);
-    test_ap<Rec32>(cnt, loop);
-    
-    test_rp<Rec36>(cnt, loop, 8192);
-    test_rp<Rec36>(cnt, loop, 32768);
-    test_ap<Rec36>(cnt, loop);
+    Uint32 loop = 768 * loops / cnt;
+    if (sizes & (1 << sz))
+    {
+      if (records & 1)
+      {
+	if (pools & T_TEST_AP)
+	  test_ap<Rec32>(cnt, loop);
+	if (pools & T_TEST_WO)
+	  test_wo<Rec32>(cnt, loop);
+	if (pools & T_TEST_RW)
+	  test_rw<Rec32>(cnt, loop);
+      }
+      if (records & 2)
+      {
+	if (pools & T_TEST_AP)
+	  test_ap<Rec36>(cnt, loop);
+	if (pools & T_TEST_WO)
+	  test_wo<Rec36>(cnt, loop);
+	if (pools & T_TEST_RW)
+	  test_rw<Rec36>(cnt, loop);
+      }
+      if (records & 4)
+      {
+	if (pools & T_TEST_AP)
+	  test_ap<Rec56>(cnt, loop);
+	if (pools & T_TEST_WO)
+	  test_wo<Rec56>(cnt, loop);
+	if (pools & T_TEST_RW)
+	  test_rw<Rec56>(cnt, loop);
+      }
+      if (records & 8)
+      {
+	if (pools & T_TEST_AP)
+	  test_ap<Rec224>(cnt, loop);
+	if (pools & T_TEST_WO)
+	  test_wo<Rec224>(cnt, loop);
+	if (pools & T_TEST_RW)
+	  test_rw<Rec224>(cnt, loop);
+      }
+    }
 
-    test_rp<Rec256>(cnt, loop, 8192);
-    test_rp<Rec256>(cnt, loop, 32768);
-    test_ap<Rec256>(cnt, loop);
+    cnt += (512 << sz);
+    sz++;
+  }
+}
 
-    test_rp<Rec260>(cnt, loop, 8192);
-    test_rp<Rec260>(cnt, loop, 32768);
-    test_ap<Rec260>(cnt, loop);
+Uint32 g_currentStartPhase;
 
-    cnt *= 100;
-    loop /= 100;
-  }
+void childExit(int code, Uint32 currentStartPhase)
+{
+  abort();
 }
 
-void
-ErrorReporter::handleAssert(const char * msg, const char * file, 
-			    int line, int)
+void childAbort(int code, Uint32 currentStartPhase)
+{
+  abort();
+}
+
+void childReportError(int error)
 {
-  ndbout << "ErrorReporter::handleAssert activated - " 
-	 << " line= " << line << endl;
   abort();
 }
+
+void
+UpgradeStartup::sendCmAppChg(Ndbcntr& cntr, Signal* signal, Uint32 startLevel){
+}
+
+void
+UpgradeStartup::execCM_APPCHG(SimulatedBlock & block, Signal* signal){
+}
+
+void
+UpgradeStartup::sendCntrMasterReq(Ndbcntr& cntr, Signal* signal, Uint32 n){
+}
+
+void
+UpgradeStartup::execCNTR_MASTER_REPLY(SimulatedBlock & block, Signal* signal){
+}
+
+#include <SimBlockList.hpp>
+
+void
+SimBlockList::unload()
+{
+
+}
+
+#define INSTANCE(X) \
+template void test_ap<X>(unsigned, unsigned);\
+template void test_wo<X>(unsigned, unsigned);\
+template void test_rw<X>(unsigned, unsigned);\
+template void test_pool<X, ArrayPool<X> >(ArrayPool<X>&, unsigned, unsigned);\
+template void test_pool<X, RecordPool<X, RWPool> >(RecordPool<X, RWPool>&, unsigned, unsigned); \
+template void test_pool<X, RecordPool<X, WOPool> >(RecordPool<X, WOPool>&, unsigned, unsigned);\
+template void init<X>(ArrayPool<X>&, unsigned);\
+template void init<X>(RecordPool<X, RWPool>&, unsigned);\
+template void init<X>(RecordPool<X, WOPool>&, unsigned)
+
+INSTANCE(Rec32);
+INSTANCE(Rec36);
+INSTANCE(Rec56);
+INSTANCE(Rec224);
+
diff --git a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
index e78a1cc1194..7b8795f7ecb 100644
--- a/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
+++ b/storage/ndb/src/kernel/vm/ndbd_malloc_impl.cpp
@@ -18,6 +18,112 @@
 
 #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)
+{
+  pages += 1;
+  void * ptr = 0;
+  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);
+  }
+
+#ifdef UNIT_TEST
+  ndbout_c("do_malloc(%d) -> %p %d", pages, ptr, chunk->m_cnt);
+  if (1)
+  {
+    Uint32 sum = 0;
+    Alloc_page* page = chunk->m_ptr;
+    for (Uint32 i = 0; i<chunk->m_cnt; i++, page++)
+    {
+      page->m_data[0*1024] = 0;
+      page->m_data[1*1024] = 0;
+      page->m_data[2*1024] = 0;
+      page->m_data[3*1024] = 0;
+      page->m_data[4*1024] = 0;
+      page->m_data[5*1024] = 0;
+      page->m_data[6*1024] = 0;
+      page->m_data[7*1024] = 0;
+    }
+  }
+#endif
+  
+  return true;
+}
 
 Uint32
 Ndbd_mem_manager::log2(Uint32 input)
@@ -33,67 +139,224 @@ 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::get_resource_limit(Uint32 id, Resource_limit& rl) const
+{
+  if (id < XX_RL_COUNT)
+  {
+    rl = m_resource_limit[id];
+    return true;
+  }
+  return false;
+}
+
 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;
+  Uint32 reserved = m_resource_limit[0].m_min;
+  if (m_resource_limit[0].m_max)
+  {
+    pages = m_resource_limit[0].m_max;
+  } 
+  else
+  {
+    pages = m_resource_limit[0].m_min; // reserved
+  }
   
-  m_base = malloc((2 + pages) * sizeof(Alloc_page));
-  UintPtr ptr = (UintPtr)m_base;
-  UintPtr rem = ptr % sizeof(Alloc_page);
-  if (rem)
+  if (m_resource_limit[0].m_min == 0)
   {
-    ptr += sizeof(Alloc_page) - rem;
+    m_resource_limit[0].m_min = pages;
   }
-  else
+  
+  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)*pages)>>20);
+  
+  if (pages == 0)
   {
-    pages++;
+    return 0;
   }
-  m_base_page = (Alloc_page*)ptr;
-  m_pages_alloc += pages;
-  m_pages_used += pages;
 
-  Uint32 bmp = (pages + (1 << BPP_2LOG) - 1) >> BPP_2LOG;
-  for(Uint32 i = 0; i < bmp; i++)
+  /**
+   * 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;
+    }
+  }
+  
+  if (allocated < m_resource_limit[0].m_min)
+  {
+    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 if (allocated < pages)
   {
-    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);
+    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 = 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);
     
-    release(start+1, end - 1 - start);    
+    if (last > max_page)
+      max_page = last;
   }
 
+  m_resource_limit[0].m_resource_id = max_page;
+  m_resource_limit[0].m_min = reserved;
+  m_resource_limit[0].m_max = 0;
+  
+  for (Uint32 i = 0; i<cnt; i++)
+  {
+    grow(chunks[i].m_start, chunks[i].m_cnt);
+  }
+  
   return true;
 }
 
+#include <NdbOut.hpp>
+
+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))
+  {
+    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;
+    }
+
+#ifdef UNIT_TEST
+    ndbout_c("creating bitmap page %d", start_bmp);
+#endif
+    
+    Alloc_page* bmp = m_base_page + start;
+    memset(bmp, 0, sizeof(Alloc_page));
+    m_used_bitmap_pages.set(start_bmp);
+    cnt--;
+    start++;
+  }
+
+  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 +394,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 +424,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,8 +435,8 @@ Ndbd_mem_manager::alloc(Uint32* ret, Uint32 *pages, Uint32 min)
    *   search in other lists...
    */
 
-  Uint32 min_list = log2(min - 1);
-  assert(list >= min_list);
+  Int32 min_list = log2(min - 1);
+  assert((Int32)list >= min_list);
   for (i = list - 1; i >= min_list; i--) 
   {
     if ((start = m_buddy_lists[i]))
@@ -192,33 +456,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 +529,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,32 +547,139 @@ 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)
+{
+  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;
+  m_resource_limit[idx].m_curr = rl.m_curr - 1;
+}
+
 #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;}
+
+  struct timeval st;
+
+  void start() {
+    gettimeofday(&st, 0);
+  }
+
+  Uint64 calc_diff() {
+    struct timeval st2;
+    gettimeofday(&st2, 0);
+    Uint64 diff = st2.tv_sec;
+    diff -= st.tv_sec;
+    diff *= 1000000;
+    diff += st2.tv_usec;
+    diff -= st.tv_usec;
+    return diff;
+  }
+  
+  void stop() {
+    add(calc_diff());
+  }
+  
+  void add(Uint64 diff) { sum += diff; cnt++;}
+
+  void print(const char * title) const {
+    float ps = sum;
+    ps /= cnt;
+    printf("%s %fus/call %lld %d\n", title, ps, sum, cnt);
+  }
+};
+
 int 
-main(void)
+main(int argc, char** argv)
 {
-  printf("Startar modul test av Page Manager\n");
+  int sz = 1*32768;
+  int run_time = 30;
+  if (argc > 1)
+    sz = 32*atoi(argv[1]);
+
+  if (argc > 2)
+    run_time = atoi(argv[2]);
+
+  char buf[255];
+  Timer timer[4];
+  printf("Startar modul test av Page Manager %dMb %ds\n", 
+	 (sz >> 5), run_time);
+  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 = sz;
+  rl.m_curr = 0;
+  rl.m_resource_id = 0;
+  mem.set_resource_limit(rl);
+  rl.m_min = sz < 16384 ? sz : 16384;
+  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;
-  for(Uint32 i = 0; i<LOOPS; i++){
+  time_t stop = time(0) + run_time;
+  for(Uint32 i = 0; time(0) < stop; i++){
     //mem.dump();
     
     // Case
     Uint32 c = (rand() % 100);
-    const Uint32 free = mem.get_no_allocated_pages() - mem.get_no_used_pages();
-    if (c < 60)
+    if (c < 50)
     {
       c = 0;
     } 
@@ -359,17 +692,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 +706,52 @@ main(void)
       const int ch = rand() % chunks.size();
       Chunk chunk = chunks[ch];
       chunks.erase(ch);
+      timer[0].start();
+      Uint64 start = NdbTick_CurrentMillisecond();      
       mem.release(chunk.pageId, chunk.pageCount);
+      timer[0].stop();
       if(DEBUG)
 	printf(" release %d %d\n", chunk.pageId, chunk.pageCount);
     }
       break;
     case 2: { // Seize(n) - fail
-      alloc += free;
+      alloc += sz;
       // Fall through
     }
     case 1: { // Seize(n) (success)
       Chunk chunk;
       chunk.pageCount = alloc;
+      if (DEBUG)
+      {
+	printf(" alloc %d -> ", alloc); fflush(stdout);
+      }
+      timer[0].start();
       mem.alloc(&chunk.pageId, &chunk.pageCount, 1);
+      Uint64 diff = timer[0].calc_diff();
+
       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(diff);
 	  if (DEBUG)
 	    printf(" -  Tried to allocate %d - only allocated %d - free: %d",
-		   alloc, chunk.pageCount, free);
+		   alloc, chunk.pageCount, 0);
+	}
+	else
+	{
+	  timer[1].add(diff);
 	}
       } else {
+	timer[3].add(diff);
 	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 +762,17 @@ 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]);
+
+  mem.dump();
 }
 
 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..4fe7507e7ae 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,25 @@ 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 get_resource_limit(Uint32 id, Resource_limit& rl) const;
 
+  bool init(bool allow_alloc_less_than_requested = true);
+  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* alloc_pages(Uint32 type, Uint32* i, Uint32 *cnt, Uint32 min = 1);
+  void release_pages(Uint32 type, Uint32 i, void*p, Uint32 cnt);
+  
   /**
    * Compute 2log of size 
    * @note size = 0     -> 0
@@ -78,18 +79,17 @@ public:
   static Uint32 log2(Uint32 size);
 
 private:
+  void grow(Uint32 start, Uint32 cnt);
+
+#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 +121,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 +139,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 +157,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 +177,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;
diff --git a/storage/ndb/src/mgmsrv/ConfigInfo.cpp b/storage/ndb/src/mgmsrv/ConfigInfo.cpp
index 7fef28c3a92..214a3b41e62 100644
--- a/storage/ndb/src/mgmsrv/ConfigInfo.cpp
+++ b/storage/ndb/src/mgmsrv/ConfigInfo.cpp
@@ -771,6 +771,18 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     "64M",
     "4M",
     "1024G" },
+
+  {
+    CFG_DB_SGA,
+    "SharedGlobalMemory",
+    DB_TOKEN,
+    "Total number bytes on each "DB_TOKEN_PRINT" node allocated for any use",
+    ConfigInfo::CI_USED,
+    false,
+    ConfigInfo::CI_INT64,
+    "20M",
+    "0",
+    "65536G" }, // 32k pages * 32-bit i value
   
   {
     CFG_DB_START_PARTIAL_TIMEOUT,
diff --git a/storage/ndb/src/mgmsrv/InitConfigFileParser.cpp b/storage/ndb/src/mgmsrv/InitConfigFileParser.cpp
index f937daf8620..68a5f02f4c5 100644
--- a/storage/ndb/src/mgmsrv/InitConfigFileParser.cpp
+++ b/storage/ndb/src/mgmsrv/InitConfigFileParser.cpp
@@ -800,6 +800,7 @@ InitConfigFileParser::parse_mycnf()
   /**
    * Add ndbd, ndb_mgmd, api/mysqld
    */
+  Uint32 idx = options.size();
   {
     struct my_option opt;
     bzero(&opt, sizeof(opt));
@@ -809,7 +810,6 @@ InitConfigFileParser::parse_mycnf()
     opt.var_type = GET_STR;
     opt.arg_type = REQUIRED_ARG;
     options.push_back(opt);
-    ndbd = &options.back();
 
     opt.name = "ndb_mgmd";
     opt.id = 256;
@@ -817,7 +817,6 @@ InitConfigFileParser::parse_mycnf()
     opt.var_type = GET_STR;
     opt.arg_type = REQUIRED_ARG;
     options.push_back(opt);
-    ndb_mgmd = &options.back();
 
     opt.name = "mysqld";
     opt.id = 256;
@@ -825,7 +824,6 @@ InitConfigFileParser::parse_mycnf()
     opt.var_type = GET_STR;
     opt.arg_type = REQUIRED_ARG;
     options.push_back(opt);
-    mysqld = &options.back();
 
     opt.name = "api";
     opt.id = 256;
@@ -833,10 +831,14 @@ InitConfigFileParser::parse_mycnf()
     opt.var_type = GET_STR;
     opt.arg_type = REQUIRED_ARG;
     options.push_back(opt);
-    api = &options.back();
 
     bzero(&opt, sizeof(opt));
     options.push_back(opt);
+
+    ndbd = &options[idx];
+    ndb_mgmd = &options[idx+1];
+    mysqld = &options[idx+2];
+    api = &options[idx+3];
   }