ndb dd -

  create RWPool
  update bench_pool

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)
 {

storage/ndb/src/kernel/vm/Makefile.am

@@ -20,7 +20,7 @@ libkernel_a_SOURCES = \
         Mutex.cpp SafeCounter.cpp \
         Rope.cpp \
 	ndbd_malloc.cpp ndbd_malloc_impl.cpp \
-        Pool.cpp WOPool.cpp
+        Pool.cpp WOPool.cpp RWPool.cpp
 
 INCLUDES_LOC = -I$(top_srcdir)/storage/ndb/src/mgmapi
 
@@ -53,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 \

storage/ndb/src/kernel/vm/Pool.hpp

@@ -304,9 +304,10 @@ inline
 void
 RecordPool<T, P>::release(Uint32 i)
 {
-  Ptr<T> p;
-  getPtr(p, i);
-  m_pool.release(p);
+  Ptr<void> ptr;
+  ptr.i = i;
+  ptr.p = m_pool.getPtr(i);
+  m_pool.release(ptr);
 }
 
 template <typename T, typename P>

storage/ndb/src/kernel/vm/RWPool.cpp

+/* 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);
+}

storage/ndb/src/kernel/vm/RWPool.hpp

+/* 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);
+}

storage/ndb/src/kernel/vm/SimulatedBlock.cpp

@@ -2032,3 +2032,6 @@ SimulatedBlock::create_distr_key(Uint32 tableId,
   }
   return dstPos;
 }
+
+CArray<KeyDescriptor> g_key_descriptor_pool;
+

storage/ndb/src/kernel/vm/WOPool.hpp

@@ -20,8 +20,7 @@ struct WOPage
 {
   Uint32 m_type_id;
   Uint32 m_ref_count;
-  Uint32 m_next_page;
-  Uint32 m_data[GLOBAL_PAGE_SIZE_WORDS - 3];
+  Uint32 m_data[GLOBAL_PAGE_SIZE_WORDS - 2];
 };
 
 /**