Merge eherman@bk-internal.mysql.com:/home/bk/mysql-5.1-new

into  mysql.com:/home/eric/fed-txn-mysql-5.1-new

BUILD/compile-sap

-#! /bin/sh
-
-path=`dirname $0`
-. "$path/SETUP.sh"
-
-extra_flags="$pentium_cflags"
-extra_configs="$pentium_configs --without-berkeley-db"
-
-. "$path/FINISH.sh"

BUILD/compile-sap-debug

-#! /bin/sh
-
-path=`dirname $0`
-. "$path/SETUP.sh"
-
-extra_flags="$pentium_cflags $debug_cflags"
-extra_configs="$pentium_configs $debug_configs --without-berkeley-db $static_link"
-
-. "$path/FINISH.sh"

mysql-test/r/show_check.result

@@ -564,3 +564,7 @@ DROP TABLE tyt2;
 DROP TABLE urkunde;
 SHOW TABLES FROM non_existing_database;
 ERROR 42000: Unknown database 'non_existing_database'
+SHOW AUTHORS;
+Name	Location	Comment
+Brian "Krow" Aker	Seattle, WA. USA	Architecture, archive, federated, buncha of little stuff :)
+David Axmark	Uppsala, Sweden	Small stuff long time ago, Monty ripped it out!

mysql-test/t/show_check.test

@@ -424,3 +424,7 @@ DROP TABLE urkunde;
 #
 --error 1049
 SHOW TABLES FROM non_existing_database;
+
+# End of 4.1 tests
+#
+SHOW AUTHORS;

sql/authors.h

+/***************************************************************************
+** Output from "SHOW AUTHORS"
+** If you can update it, you get to be in it :)
+** Dont be offended if your name is not in here, just add it!
+***************************************************************************/
+
+struct show_table_authors_st {
+  const char *name;
+  const char *location;
+  const char *comment;
+};
+
+struct show_table_authors_st show_table_authors[]= {
+  { "Brian \"Krow\" Aker", "Seattle, WA. USA", 
+    "Architecture, archive, federated, buncha of little stuff :)" },
+  { "David Axmark", "Uppsala, Sweden", "Small stuff long time ago, Monty ripped it out!"},
+  {NULL, NULL, NULL}
+};

sql/field.h

@@ -1081,7 +1081,7 @@ public:
   int cmp_max(const char *, const char *, uint max_length);
   int cmp(const char *a,const char*b)
   {
-    return cmp_max(a, b, ~0);
+    return cmp_max(a, b, ~0L);
   }
   void sort_string(char *buff,uint length);
   void get_key_image(char *buff,uint length, imagetype type);
@@ -1141,7 +1141,7 @@ public:
   my_decimal *val_decimal(my_decimal *);
   int cmp_max(const char *, const char *, uint max_length);
   int cmp(const char *a,const char*b)
-    { return cmp_max(a, b, ~0); }
+    { return cmp_max(a, b, ~0L); }
   int cmp(const char *a, uint32 a_length, const char *b, uint32 b_length);
   int cmp_binary(const char *a,const char *b, uint32 max_length=~0L);
   int key_cmp(const byte *,const byte*);

sql/ha_berkeley.cc

@@ -78,9 +78,11 @@ const char *ha_berkeley_ext=".db";
 bool berkeley_shared_data=0;
 u_int32_t berkeley_init_flags= DB_PRIVATE | DB_RECOVER, berkeley_env_flags=0,
           berkeley_lock_type=DB_LOCK_DEFAULT;
-ulong berkeley_cache_size, berkeley_log_buffer_size, berkeley_log_file_size=0;
+ulong berkeley_log_buffer_size=0 , berkeley_log_file_size=0;
+ulonglong berkeley_cache_size= 0;
 char *berkeley_home, *berkeley_tmpdir, *berkeley_logdir;
 long berkeley_lock_scan_time=0;
+ulong berkeley_region_size=0, berkeley_cache_parts=1;
 ulong berkeley_trans_retry=1;
 ulong berkeley_max_lock;
 pthread_mutex_t bdb_mutex;
@@ -89,9 +91,12 @@ static DB_ENV *db_env;
 static HASH bdb_open_tables;
 
 const char *berkeley_lock_names[] =
-{ "DEFAULT", "OLDEST","RANDOM","YOUNGEST",0 };
+{ "DEFAULT", "OLDEST", "RANDOM", "YOUNGEST", "EXPIRE", "MAXLOCKS",
+  "MAXWRITE", "MINLOCKS", "MINWRITE", 0 };
 u_int32_t berkeley_lock_types[]=
-{ DB_LOCK_DEFAULT, DB_LOCK_OLDEST, DB_LOCK_RANDOM };
+{ DB_LOCK_DEFAULT, DB_LOCK_OLDEST, DB_LOCK_RANDOM, DB_LOCK_YOUNGEST,
+  DB_LOCK_EXPIRE, DB_LOCK_MAXLOCKS, DB_LOCK_MAXWRITE, DB_LOCK_MINLOCKS,
+  DB_LOCK_MINWRITE };
 TYPELIB berkeley_lock_typelib= {array_elements(berkeley_lock_names)-1,"",
 				berkeley_lock_names, NULL};
 
@@ -204,10 +209,17 @@ bool berkeley_init(void)
 			DB_VERB_DEADLOCK | DB_VERB_RECOVERY,
 			1);
 
-  db_env->set_cachesize(db_env, 0, berkeley_cache_size, 0);
+  if (berkeley_cache_size > (uint) ~0)
+    db_env->set_cachesize(db_env, berkeley_cache_size / (1024*1024L*1024L),
+                          berkeley_cache_size % (1024L*1024L*1024L),
+                          berkeley_cache_parts);
+  else
+    db_env->set_cachesize(db_env, 0, berkeley_cache_size, berkeley_cache_parts);
+
   db_env->set_lg_max(db_env, berkeley_log_file_size);
   db_env->set_lg_bsize(db_env, berkeley_log_buffer_size);
   db_env->set_lk_detect(db_env, berkeley_lock_type);
+  db_env->set_lg_regionmax(db_env, berkeley_region_size);
   if (berkeley_max_lock)
     db_env->set_lk_max(db_env, berkeley_max_lock);
 

sql/ha_berkeley.h

@@ -163,7 +163,9 @@ extern const u_int32_t bdb_DB_PRIVATE;
 extern bool berkeley_shared_data;
 extern u_int32_t berkeley_init_flags,berkeley_env_flags, berkeley_lock_type,
                  berkeley_lock_types[];
-extern ulong berkeley_cache_size, berkeley_max_lock, berkeley_log_buffer_size;
+extern ulong berkeley_max_lock, berkeley_log_buffer_size;
+extern ulonglong berkeley_cache_size;
+extern ulong berkeley_region_size, berkeley_cache_parts;
 extern char *berkeley_home, *berkeley_tmpdir, *berkeley_logdir;
 extern long berkeley_lock_scan_time;
 extern TYPELIB berkeley_lock_typelib;

sql/lex.h

@@ -74,6 +74,7 @@ static SYMBOL symbols[] = {
   { "ASC",		SYM(ASC)},
   { "ASCII",		SYM(ASCII_SYM)},
   { "ASENSITIVE",       SYM(ASENSITIVE_SYM)},
+  { "AUTHORS",	        SYM(AUTHORS_SYM)},
   { "AUTO_INCREMENT",	SYM(AUTO_INC)},
   { "AVG",		SYM(AVG_SYM)},
   { "AVG_ROW_LENGTH",	SYM(AVG_ROW_LENGTH)},

sql/mysql_priv.h

@@ -857,6 +857,7 @@ int mysqld_show_variables(THD *thd,const char *wild);
 int mysql_find_files(THD *thd,List<char> *files, const char *db,
                 const char *path, const char *wild, bool dir);
 bool mysqld_show_storage_engines(THD *thd);
+bool mysqld_show_authors(THD *thd);
 bool mysqld_show_privileges(THD *thd);
 bool mysqld_show_column_types(THD *thd);
 bool mysqld_help (THD *thd, const char *text);

sql/mysqld.cc

@@ -394,7 +394,9 @@ extern const u_int32_t bdb_DB_TXN_NOSYNC, bdb_DB_RECOVER, bdb_DB_PRIVATE;
 extern bool berkeley_shared_data;
 extern u_int32_t berkeley_init_flags,berkeley_env_flags, berkeley_lock_type,
                  berkeley_lock_types[];
-extern ulong berkeley_cache_size, berkeley_max_lock, berkeley_log_buffer_size;
+extern ulong berkeley_max_lock, berkeley_log_buffer_size;
+extern ulonglong berkeley_cache_size;
+extern ulong berkeley_region_size, berkeley_cache_parts;
 extern char *berkeley_home, *berkeley_tmpdir, *berkeley_logdir;
 extern long berkeley_lock_scan_time;
 extern TYPELIB berkeley_lock_typelib;
@@ -4582,8 +4584,10 @@ enum options_mysqld
   OPT_INNODB_CONCURRENCY_TICKETS,
   OPT_INNODB_THREAD_SLEEP_DELAY,
   OPT_BDB_CACHE_SIZE,
+  OPT_BDB_CACHE_PARTS,
   OPT_BDB_LOG_BUFFER_SIZE,
   OPT_BDB_MAX_LOCK,
+  OPT_BDB_REGION_SIZE,
   OPT_ERROR_LOG_FILE,
   OPT_DEFAULT_WEEK_FORMAT,
   OPT_RANGE_ALLOC_BLOCK_SIZE, OPT_ALLOW_SUSPICIOUS_UDFS,
@@ -5398,8 +5402,12 @@ log and this option does nothing anymore.",
 #ifdef WITH_BERKELEY_STORAGE_ENGINE
   { "bdb_cache_size", OPT_BDB_CACHE_SIZE,
     "The buffer that is allocated to cache index and rows for BDB tables.",
-    (gptr*) &berkeley_cache_size, (gptr*) &berkeley_cache_size, 0, GET_ULONG,
-    REQUIRED_ARG, KEY_CACHE_SIZE, 20*1024, (long) ~0, 0, IO_SIZE, 0},
+    (gptr*) &berkeley_cache_size, (gptr*) &berkeley_cache_size, 0, GET_ULL,
+    REQUIRED_ARG, KEY_CACHE_SIZE, 20*1024, (ulonglong) ~0, 0, IO_SIZE, 0},
+  { "bdb_cache_parts", OPT_BDB_CACHE_PARTS,
+    "Number of parts to use for BDB cache.",
+    (gptr*) &berkeley_cache_parts, (gptr*) &berkeley_cache_parts, 0, GET_ULONG,
+    REQUIRED_ARG, 1, 1, 1024, 0, 1, 0},
   /* QQ: The following should be removed soon! (bdb_max_lock preferred) */
   {"bdb_lock_max", OPT_BDB_MAX_LOCK, "Synonym for bdb_max_lock.",
    (gptr*) &berkeley_max_lock, (gptr*) &berkeley_max_lock, 0, GET_ULONG,
@@ -5412,6 +5420,10 @@ log and this option does nothing anymore.",
    "The maximum number of locks you can have active on a BDB table.",
    (gptr*) &berkeley_max_lock, (gptr*) &berkeley_max_lock, 0, GET_ULONG,
    REQUIRED_ARG, 10000, 0, (long) ~0, 0, 1, 0},
+  {"bdb_region_size", OPT_BDB_REGION_SIZE,
+    "The size of the underlying logging area of the Berkeley DB environment.",
+    (gptr*) &berkeley_region_size, (gptr*) &berkeley_region_size, 0, GET_ULONG,
+    OPT_ARG, 60*1024L, 60*1024L, (long) ~0, 0, 1, 0},
 #endif /* WITH_BERKELEY_STORAGE_ENGINE */
   {"binlog_cache_size", OPT_BINLOG_CACHE_SIZE,
    "The size of the cache to hold the SQL statements for the binary log during a transaction. If you often use big, multi-statement transactions you can increase this to get more performance.",
@@ -7410,7 +7422,9 @@ SHOW_COMP_OPTION have_blackhole_db= SHOW_OPTION_NO;
 
 #ifndef WITH_BERKELEY_STORAGE_ENGINE
 bool berkeley_shared_data;
-ulong berkeley_cache_size, berkeley_max_lock, berkeley_log_buffer_size;
+ulong berkeley_max_lock, berkeley_log_buffer_size;
+ulonglong berkeley_cache_size;
+ulong berkeley_region_size, berkeley_cache_parts;
 char *berkeley_home, *berkeley_tmpdir, *berkeley_logdir;
 #endif
 

sql/set_var.cc

@@ -61,7 +61,9 @@
 
 /* WITH_BERKELEY_STORAGE_ENGINE */
 extern bool berkeley_shared_data;
-extern ulong berkeley_cache_size, berkeley_max_lock, berkeley_log_buffer_size;
+extern ulong berkeley_max_lock, berkeley_log_buffer_size;
+extern ulonglong berkeley_cache_size;
+extern ulong berkeley_region_size, berkeley_cache_parts;
 extern char *berkeley_home, *berkeley_tmpdir, *berkeley_logdir;
 
 /* WITH_INNOBASE_STORAGE_ENGINE */
@@ -622,11 +624,13 @@ struct show_var_st init_vars[]= {
   {sys_automatic_sp_privileges.name,(char*) &sys_automatic_sp_privileges,       SHOW_SYS},
   {"back_log",                (char*) &back_log,                    SHOW_LONG},
   {"basedir",                 mysql_home,                           SHOW_CHAR},
-  {"bdb_cache_size",          (char*) &berkeley_cache_size,         SHOW_LONG},
+  {"bdb_cache_size",          (char*) &berkeley_cache_size,         SHOW_LONGLONG},
+  {"bdb_cache_parts",         (char*) &berkeley_cache_parts,        SHOW_LONG},
   {"bdb_home",                (char*) &berkeley_home,               SHOW_CHAR_PTR},
   {"bdb_log_buffer_size",     (char*) &berkeley_log_buffer_size,    SHOW_LONG},
   {"bdb_logdir",              (char*) &berkeley_logdir,             SHOW_CHAR_PTR},
   {"bdb_max_lock",            (char*) &berkeley_max_lock,	    SHOW_LONG},
+  {"bdb_region_size",         (char*) &berkeley_region_size,	    SHOW_LONG},
   {"bdb_shared_data",	      (char*) &berkeley_shared_data,	    SHOW_BOOL},
   {"bdb_tmpdir",              (char*) &berkeley_tmpdir,             SHOW_CHAR_PTR},
   {sys_binlog_cache_size.name,(char*) &sys_binlog_cache_size,	    SHOW_SYS},

sql/sql_cache.h

@@ -215,6 +215,8 @@ struct Query_cache_memory_bin
 
 struct Query_cache_memory_bin_step
 {
+public:
+  Query_cache_memory_bin_step() {}
   ulong size;
   ulong increment;
   uint idx;

sql/sql_lex.h

@@ -92,6 +92,7 @@ enum enum_sql_command {
   SQLCOM_XA_START, SQLCOM_XA_END, SQLCOM_XA_PREPARE,
   SQLCOM_XA_COMMIT, SQLCOM_XA_ROLLBACK, SQLCOM_XA_RECOVER,
   SQLCOM_INSTALL_PLUGIN, SQLCOM_UNINSTALL_PLUGIN,
+  SQLCOM_SHOW_AUTHORS,
   /* This should be the last !!! */
 
   SQLCOM_END

sql/sql_parse.cc

@@ -3409,6 +3409,9 @@ end_with_restore_list:
   case SQLCOM_SHOW_STORAGE_ENGINES:
     res= mysqld_show_storage_engines(thd);
     break;
+  case SQLCOM_SHOW_AUTHORS:
+    res= mysqld_show_authors(thd);
+    break;
   case SQLCOM_SHOW_PRIVILEGES:
     res= mysqld_show_privileges(thd);
     break;

sql/sql_show.cc

@@ -23,6 +23,7 @@
 #include "sp.h"
 #include "sp_head.h"
 #include "sql_trigger.h"
+#include "authors.h"
 #include <my_dir.h>
 
 
@@ -56,6 +57,9 @@ bool mysqld_show_storage_engines(THD *thd)
   field_list.push_back(new Item_empty_string("Engine",10));
   field_list.push_back(new Item_empty_string("Support",10));
   field_list.push_back(new Item_empty_string("Comment",80));
+  field_list.push_back(new Item_empty_string("Transactions",3));
+  field_list.push_back(new Item_empty_string("XA",3));
+  field_list.push_back(new Item_empty_string("Savepoints",3));
 
   if (protocol->send_fields(&field_list,
                             Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
@@ -76,6 +80,9 @@ bool mysqld_show_storage_engines(THD *thd)
       option_name= "DEFAULT";
     protocol->store(option_name, system_charset_info);
     protocol->store((*types)->comment, system_charset_info);
+    protocol->store((*types)->commit ? "YES" : "NO", system_charset_info);
+    protocol->store((*types)->prepare ? "YES" : "NO", system_charset_info);
+    protocol->store((*types)->savepoint_set ? "YES" : "NO", system_charset_info);
     if (protocol->write())
       DBUG_RETURN(TRUE);
   }
@@ -83,6 +90,38 @@ bool mysqld_show_storage_engines(THD *thd)
   DBUG_RETURN(FALSE);
 }
 
+/***************************************************************************
+** List all Authors.
+** If you can update it, you get to be in it :)
+***************************************************************************/
+
+bool mysqld_show_authors(THD *thd)
+{
+  List<Item> field_list;
+  Protocol *protocol= thd->protocol;
+  DBUG_ENTER("mysqld_show_authors");
+
+  field_list.push_back(new Item_empty_string("Name",40));
+  field_list.push_back(new Item_empty_string("Location",40));
+  field_list.push_back(new Item_empty_string("Comment",80));
+
+  if (protocol->send_fields(&field_list,
+                            Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
+    DBUG_RETURN(TRUE);
+
+  show_table_authors_st *authors;
+  for (authors= show_table_authors; authors->name; authors++)
+  {
+    protocol->prepare_for_resend();
+    protocol->store(authors->name, system_charset_info);
+    protocol->store(authors->location, system_charset_info);
+    protocol->store(authors->comment, system_charset_info);
+    if (protocol->write())
+      DBUG_RETURN(TRUE);
+  }
+  send_eof(thd);
+  DBUG_RETURN(FALSE);
+}
 
 /***************************************************************************
  List all privileges supported

sql/sql_yacc.yy

@@ -137,6 +137,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  ASCII_SYM
 %token  ASENSITIVE_SYM
 %token  ATAN
+%token  AUTHORS_SYM
 %token  AUTO_INC
 %token  AVG_ROW_LENGTH
 %token  AVG_SYM
@@ -7174,6 +7175,11 @@ show_param:
 	    LEX *lex=Lex;
 	    lex->sql_command= SQLCOM_SHOW_STORAGE_ENGINES;
 	  }
+	| AUTHORS_SYM
+	  {
+	    LEX *lex=Lex;
+	    lex->sql_command= SQLCOM_SHOW_AUTHORS;
+	  }
 	| PRIVILEGES
 	  {
 	    LEX *lex=Lex;
@@ -8171,6 +8177,7 @@ user:
 keyword:
 	keyword_sp		{}
 	| ASCII_SYM		{}
+	| AUTHORS_SYM		{}
 	| BACKUP_SYM		{}
 	| BEGIN_SYM		{}
 	| BYTE_SYM		{}

storage/ndb/src/kernel/vm/LinearPool.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 */
+
+#ifndef LINEAR_POOL_HPP
+#define LINEAR_POOL_HPP
+
+#include <SuperPool.hpp>
+
+/*
+ * LinearPool - indexed record pool
+ *
+ * LinearPool implements a pool where each record has a 0-based index.
+ * Any index value (up to 2^32-1) is allowed.  Normal efficient usage is
+ * to assign index values in sequence and to re-use any values which
+ * have become free.  This is default seize/release behaviour.
+ *
+ * LinearPool has 2 internal RecordPool instances:
+ *
+ * (a) record pool of T (the template argument class)
+ * (b) record pool of "maps" (array of Uint32)
+ *
+ * The maps translate an index into an i-value in (a).  Each map has
+ * a level.  Level 0 maps point to i-values.  Level N+1 maps point to
+ * level N maps.  There is a unique "root map" at top.
+ *
+ * This works exactly like numbers in a given base.  Each map has base
+ * size entries.  For implementation convenience the base must be power
+ * of 2 and is given as its log2 value.
+ *
+ * There is a doubly linked list of available maps (some free entries)
+ * on each level.  There is a singly linked free list within each map.
+ *
+ * Level 0 free entry has space for one record.  Level N free entry
+ * implies space for base^N records.  The implied levels are created and
+ * removed on demand.  Completely empty maps are removed.
+ *
+ * Default base is 256 (log2 = 8) which requires maximum 4 levels or
+ * "digits" (similar to ip address).
+ *
+ * TODO
+ *
+ * - move most of the inline code to LinearPool.cpp
+ * - add methods to check / seize user-specified index
+ */
+
+#include "SuperPool.hpp"
+
+template <class T, Uint32 LogBase = 8>
+class LinearPool {
+  typedef SuperPool::PtrI PtrI;
+
+  // Base.
+  STATIC_CONST( Base = 1 << LogBase );
+
+  // Digit mask.
+  STATIC_CONST( DigitMask = Base - 1 );
+
+  // Max possible levels (0 to max root level).
+  STATIC_CONST( MaxLevels = (32 + LogBase - 1) / LogBase );
+
+  // Map.
+  struct Map {
+    Uint32 m_level;
+    Uint32 m_occup;     // number of used entries
+    Uint32 m_firstfree; // position of first free entry
+    PtrI m_parent;      // parent map
+    Uint32 m_index;     // from root to here
+    PtrI m_nextavail;
+    PtrI m_prevavail;
+    PtrI m_entry[Base];
+  };
+
+public:
+
+  // Constructor.
+  LinearPool(GroupPool& gp);
+
+  // Destructor.
+  ~LinearPool();
+
+  // Update pointer ptr.p according to index value ptr.i.
+  void getPtr(Ptr<T>& ptr);
+
+  // Allocate record from the pool.  Reuses free index if possible.
+  bool seize(Ptr<T>& ptr);
+
+  // Return record to the pool.
+  void release(Ptr<T>& ptr);
+
+  // Verify (debugging).
+  void verify();
+
+private:
+
+  // Given index find the bottom map.
+  void get_map(Ptr<Map>& map_ptr, Uint32 index);
+
+  // Add new root map and increase level
+  bool add_root();
+
+  // Add new non-root map.
+  bool add_map(Ptr<Map>& map_ptr, Ptr<Map> parent_ptr, Uint32 digit);
+
+  // Remove entry and map if it becomes empty.
+  void remove_entry(Ptr<Map> map_ptr, Uint32 digit);
+
+  // Remove map and all parents which become empty.
+  void remove_map(Ptr<Map> map_ptr);
+
+  // Add map to available list.
+  void add_avail(Ptr<Map> map_ptr);
+
+  // Remove map from available list.
+  void remove_avail(Ptr<Map> map_ptr);
+
+  // Verify map (recursive).
+  void verify(Ptr<Map> map_ptr, Uint32 level);
+
+  RecordPool<T> m_records;
+  RecordPool<Map> m_maps;
+  Uint32 m_levels;              // 0 means empty pool
+  PtrI m_root;
+  PtrI m_avail[MaxLevels];
+};
+
+template <class T, Uint32 LogBase>
+inline
+LinearPool<T, LogBase>::LinearPool(GroupPool& gp) :
+  m_records(gp),
+  m_maps(gp),
+  m_levels(0),
+  m_root(RNIL)
+{
+  Uint32 n;
+  for (n = 0; n < MaxLevels; n++)
+    m_avail[n] = RNIL;
+}
+
+template <class T, Uint32 LogBase>
+inline
+LinearPool<T, LogBase>::~LinearPool()
+{
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::getPtr(Ptr<T>& ptr)
+{
+  Uint32 index = ptr.i;
+  // get level 0 map
+  Ptr<Map> map_ptr;
+  get_map(map_ptr, index);
+  // get record
+  Ptr<T> rec_ptr;
+  Uint32 digit = index & DigitMask;
+  rec_ptr.i = map_ptr.p->m_entry[digit];
+  m_records.getPtr(rec_ptr);
+  ptr.p = rec_ptr.p;
+}
+
+template <class T, Uint32 LogBase>
+inline bool
+LinearPool<T, LogBase>::seize(Ptr<T>& ptr)
+{
+  // look for free list on some level
+  Ptr<Map> map_ptr;
+  map_ptr.i = RNIL;
+  Uint32 n = 0;
+  while (n < m_levels) {
+    if ((map_ptr.i = m_avail[n]) != RNIL)
+      break;
+    n++;
+  }
+  if (map_ptr.i == RNIL) {
+    // add new level with available maps
+    if (! add_root())
+      return false;
+    assert(n < m_levels);
+    map_ptr.i = m_avail[n];
+  }
+  m_maps.getPtr(map_ptr);
+  // walk down creating missing levels and using an entry on each
+  Uint32 firstfree;
+  while (true) {
+    assert(map_ptr.p->m_occup < Base);
+    map_ptr.p->m_occup++;
+    firstfree = map_ptr.p->m_firstfree;
+    assert(firstfree < Base);
+    map_ptr.p->m_firstfree = map_ptr.p->m_entry[firstfree];
+    if (map_ptr.p->m_occup == Base) {
+      assert(map_ptr.p->m_firstfree == Base);
+      // remove from available list
+      remove_avail(map_ptr);
+    }
+    if (n == 0)
+      break;
+    Ptr<Map> child_ptr;
+    if (! add_map(child_ptr, map_ptr, firstfree)) {
+      remove_entry(map_ptr, firstfree);
+      return false;
+    }
+    map_ptr.p->m_entry[firstfree] = child_ptr.i;
+    map_ptr = child_ptr;
+    n--;
+  }
+  // now on level 0
+  assert(map_ptr.p->m_level == 0);
+  Ptr<T> rec_ptr;
+  if (! m_records.seize(rec_ptr)) {
+    remove_entry(map_ptr, firstfree);
+    return false;
+  }
+  map_ptr.p->m_entry[firstfree] = rec_ptr.i;
+  ptr.i = firstfree + (map_ptr.p->m_index << LogBase);
+  ptr.p = rec_ptr.p;
+  return true;
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::release(Ptr<T>& ptr)
+{
+  Uint32 index = ptr.i;
+  // get level 0 map
+  Ptr<Map> map_ptr;
+  get_map(map_ptr, index);
+  // release record
+  Ptr<T> rec_ptr;
+  Uint32 digit = index & DigitMask;
+  rec_ptr.i = map_ptr.p->m_entry[digit];
+  m_records.release(rec_ptr);
+  // remove entry
+  remove_entry(map_ptr, digit);
+  // null pointer
+  ptr.i = RNIL;
+  ptr.p = 0;
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::verify()
+{
+  if (m_root == RNIL)
+    return;
+  assert(m_levels != 0);
+  Ptr<Map> map_ptr;
+  map_ptr.i = m_root;
+  m_maps.getPtr(map_ptr);
+  verify(map_ptr, m_levels - 1);
+}
+
+// private methods
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::get_map(Ptr<Map>& map_ptr, Uint32 index)
+{
+  // root map must exist
+  Ptr<Map> tmp_ptr;
+  tmp_ptr.i = m_root;
+  m_maps.getPtr(tmp_ptr);
+  assert(tmp_ptr.p->m_level + 1 == m_levels);
+  // extract index digits up to current root level
+  Uint32 digits[MaxLevels];
+  Uint32 n = 0;
+  do {
+    digits[n] = index & DigitMask;
+    index >>= LogBase;
+  } while (++n < m_levels);
+  assert(index == 0);
+  // walk down indirect levels
+  while (--n > 0) {
+    tmp_ptr.i = tmp_ptr.p->m_entry[digits[n]];
+    m_maps.getPtr(tmp_ptr);
+  }
+  // level 0 map
+  assert(tmp_ptr.p->m_level == 0);
+  map_ptr = tmp_ptr;
+}
+
+template <class T, Uint32 LogBase>
+inline bool
+LinearPool<T, LogBase>::add_root()
+{
+  // new root
+  Ptr<Map> map_ptr;
+  if (! m_maps.seize(map_ptr))
+    return false;
+  Uint32 n = m_levels++;
+  assert(n < MaxLevels);
+  // set up
+  map_ptr.p->m_level = n;
+  if (n == 0) {
+    map_ptr.p->m_occup = 0;
+    map_ptr.p->m_firstfree = 0;
+  } else {
+    // on level > 0 digit 0 points to old root
+    map_ptr.p->m_occup = 1;
+    map_ptr.p->m_firstfree = 1;
+    Ptr<Map> old_ptr;
+    old_ptr.i = m_root;
+    m_maps.getPtr(old_ptr);
+    assert(old_ptr.p->m_parent == RNIL);
+    old_ptr.p->m_parent = map_ptr.i;
+    map_ptr.p->m_entry[0] = old_ptr.i;
+  }
+  // set up free list with Base as terminator
+  for (Uint32 j = map_ptr.p->m_firstfree; j < Base; j++)
+    map_ptr.p->m_entry[j] = j + 1;
+  map_ptr.p->m_parent = RNIL;
+  map_ptr.p->m_index = 0;
+  add_avail(map_ptr);
+  // set new root
+  m_root = map_ptr.i;
+  return true;
+}
+
+template <class T, Uint32 LogBase>
+inline bool
+LinearPool<T, LogBase>::add_map(Ptr<Map>& map_ptr, Ptr<Map> parent_ptr, Uint32 digit)
+{
+  if (! m_maps.seize(map_ptr))
+    return false;
+  assert(parent_ptr.p->m_level != 0);
+  // set up
+  map_ptr.p->m_level = parent_ptr.p->m_level - 1;
+  map_ptr.p->m_occup = 0;
+  map_ptr.p->m_firstfree = 0;
+  // set up free list with Base as terminator
+  for (Uint32 j = map_ptr.p->m_firstfree; j < Base; j++)
+    map_ptr.p->m_entry[j] = j + 1;
+  map_ptr.p->m_parent = parent_ptr.i;
+  map_ptr.p->m_index = digit + (parent_ptr.p->m_index << LogBase);
+  add_avail(map_ptr);
+  return true;
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::remove_entry(Ptr<Map> map_ptr, Uint32 digit)
+{
+  assert(map_ptr.p->m_occup != 0 && digit < Base);
+  map_ptr.p->m_occup--;
+  map_ptr.p->m_entry[digit] = map_ptr.p->m_firstfree;
+  map_ptr.p->m_firstfree = digit;
+  if (map_ptr.p->m_occup + 1 == Base)
+    add_avail(map_ptr);
+  else if (map_ptr.p->m_occup == 0)
+    remove_map(map_ptr);
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::remove_map(Ptr<Map> map_ptr)
+{
+  assert(map_ptr.p->m_occup == 0);
+  remove_avail(map_ptr);
+  Ptr<Map> parent_ptr;
+  parent_ptr.i = map_ptr.p->m_parent;
+  Uint32 digit = map_ptr.p->m_index & DigitMask;
+  PtrI map_ptr_i = map_ptr.i;
+  m_maps.release(map_ptr);
+  if (m_root == map_ptr_i) {
+    assert(parent_ptr.i == RNIL);
+    Uint32 used = m_maps.m_superPool.getRecUseCount(m_maps.m_recInfo);
+    assert(used == 0);
+    m_root = RNIL;
+    m_levels = 0;
+  }
+  if (parent_ptr.i != RNIL) {
+    m_maps.getPtr(parent_ptr);
+    // remove child entry (recursive)
+    remove_entry(parent_ptr, digit);
+  }
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::add_avail(Ptr<Map> map_ptr)
+{
+  Uint32 n = map_ptr.p->m_level;
+  assert(n < m_levels);
+  map_ptr.p->m_nextavail = m_avail[n];
+  if (map_ptr.p->m_nextavail != RNIL) {
+    Ptr<Map> next_ptr;
+    next_ptr.i = map_ptr.p->m_nextavail;
+    m_maps.getPtr(next_ptr);
+    next_ptr.p->m_prevavail = map_ptr.i;
+  }
+  map_ptr.p->m_prevavail = RNIL;
+  m_avail[n] = map_ptr.i;
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::remove_avail(Ptr<Map> map_ptr)
+{
+  Uint32 n = map_ptr.p->m_level;
+  assert(n < m_levels);
+  if (map_ptr.p->m_nextavail != RNIL) {
+    Ptr<Map> next_ptr;
+    next_ptr.i = map_ptr.p->m_nextavail;
+    m_maps.getPtr(next_ptr);
+    next_ptr.p->m_prevavail = map_ptr.p->m_prevavail;
+  }
+  if (map_ptr.p->m_prevavail != RNIL) {
+    Ptr<Map> prev_ptr;
+    prev_ptr.i = map_ptr.p->m_prevavail;
+    m_maps.getPtr(prev_ptr);
+    prev_ptr.p->m_nextavail = map_ptr.p->m_nextavail;
+  }
+  if (map_ptr.p->m_prevavail == RNIL) {
+    m_avail[n] = map_ptr.p->m_nextavail;
+  }
+  map_ptr.p->m_nextavail = RNIL;
+  map_ptr.p->m_prevavail = RNIL;
+}
+
+template <class T, Uint32 LogBase>
+inline void
+LinearPool<T, LogBase>::verify(Ptr<Map> map_ptr, Uint32 level)
+{
+  assert(level < MaxLevels);
+  assert(map_ptr.p->m_level == level);
+  Uint32 j = 0;
+  while (j < Base) {
+    bool free = false;
+    Uint32 j2 = map_ptr.p->m_firstfree;
+    while (j2 != Base) {
+      if (j2 == j) {
+        free = true;
+        break;
+      }
+      assert(j2 < Base);
+      j2 = map_ptr.p->m_entry[j2];
+    }
+    if (! free) {
+      if (level != 0) {
+        Ptr<Map> child_ptr;
+        child_ptr.i = map_ptr.p->m_entry[j];
+        m_maps.getPtr(child_ptr);
+        assert(child_ptr.p->m_parent == map_ptr.i);
+        assert(child_ptr.p->m_index == j + (map_ptr.p->m_index << LogBase));
+        verify(child_ptr, level - 1);
+      } else {
+        Ptr<T> rec_ptr;
+        rec_ptr.i = map_ptr.p->m_entry[j];
+        m_records.getPtr(rec_ptr);
+      }
+      Ptr<Map> avail_ptr;
+      avail_ptr.i = m_avail[map_ptr.p->m_level];
+      bool found = false;
+      while (avail_ptr.i != RNIL) {
+        if (avail_ptr.i == map_ptr.i) {
+          found = true;
+          break;
+        }
+        m_maps.getPtr(avail_ptr);
+        avail_ptr.i = avail_ptr.p->m_nextavail;
+      }
+      assert(found == (map_ptr.p->m_occup < Base));
+    }
+    j++;
+  }
+}
+
+#endif

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

@@ -23,6 +23,7 @@ exit $?
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
 
 #include "SuperPool.hpp"
+#include "LinearPool.hpp"
 #include <NdbOut.hpp>
 
 template <Uint32 sz>
@@ -198,6 +199,74 @@ sp_test(GroupPool& gp)
   delete [] ptrList;
 }
 
+template <class T>
+static void
+lp_test(GroupPool& gp)
+{
+  SuperPool& sp = gp.m_superPool;
+  LinearPool<T, 5> lp(gp);
+  ndbout << "linear pool test" << endl;
+  Ptr<T> ptr;
+  Uint32 loop;
+  for (loop = 0; loop < 3 * loopcount; loop++) {
+    int count = 0;
+    while (1) {
+      bool ret = lp.seize(ptr);
+      lp.verify();
+      if (! ret)
+        break;
+      assert(ptr.i == count);
+      Ptr<T> ptr2;
+      ptr2.i = ptr.i;
+      ptr2.p = 0;
+      lp.getPtr(ptr2);
+      assert(ptr.p == ptr2.p);
+      count++;
+    }
+    ndbout << "seized " << count << endl;
+    switch (loop % 3) {
+    case 0:
+      {
+        int n = 0;
+        while (n < count) {
+          ptr.i = n;
+          lp.release(ptr);
+          lp.verify();
+          n++;
+        }
+        ndbout << "released in order" << endl;
+      }
+      break;
+    case 1:
+      {
+        int n = count;
+        while (n > 0) {
+          n--;
+          ptr.i = n;
+          lp.release(ptr);
+          lp.verify();
+        }
+        ndbout << "released in reverse" << endl;
+      }
+      break;
+    default:
+      {
+        int coprime = random_coprime(count);
+        int n = 0;
+        while (n < count) {
+          int m = (coprime * n) % count;
+          ptr.i = m;
+          lp.release(ptr);
+          lp.verify();
+          n++;
+        }
+        ndbout << "released at random" << endl;
+      }
+      break;
+    }
+  }
+}
+
 static Uint32 pageSize = 32768;
 static Uint32 pageBits = 17;
 
@@ -218,6 +287,8 @@ template static void sp_test<T2>(GroupPool& sp);
 template static void sp_test<T3>(GroupPool& sp);
 template static void sp_test<T4>(GroupPool& sp);
 template static void sp_test<T5>(GroupPool& sp);
+//
+template static void lp_test<T3>(GroupPool& sp);
 
 int
 main()
@@ -231,13 +302,18 @@ main()
   Uint16 s = (Uint16)getpid();
   srandom(s);
   ndbout << "rand " << s << endl;
-  int count = 0;
-  while (++count <= 1) {
+  int count;
+  count = 0;
+  while (++count <= 0) {
     sp_test<T1>(gp);
     sp_test<T2>(gp);
     sp_test<T3>(gp);
     sp_test<T4>(gp);
     sp_test<T5>(gp);
   }
+  count = 0;
+  while (++count <= 1) {
+    lp_test<T3>(gp);
+  }
   return 0;
 }

support-files/my-large.cnf.sh

@@ -115,7 +115,7 @@ server-id	= 1
 #log-update 	= /path-to-dedicated-directory/hostname
 
 # Uncomment the following if you are using BDB tables
-#bdb_cache_size = 64M
+#bdb_cache_size = 256M
 #bdb_max_lock = 100000
 
 # Uncomment the following if you are using InnoDB tables

support-files/my-medium.cnf.sh

@@ -113,7 +113,7 @@ server-id	= 1
 #log-update 	= /path-to-dedicated-directory/hostname
 
 # Uncomment the following if you are using BDB tables
-#bdb_cache_size = 4M
+#bdb_cache_size = 16M
 #bdb_max_lock = 10000
 
 # Uncomment the following if you are using InnoDB tables