WL#5419 "LOCK_open scalability: make tdc_refresh_version

an atomic counter"

Split the large LOCK_open section in open_table(). 
Do not call open_table_from_share() under LOCK_open.
Remove thd->version.

This fixes
Bug#50589 "Server hang on a query evaluated using a temporary 
table"
Bug#51557 "LOCK_open and kernel_mutex are not happy together"
Bug#49463 "LOCK_table and innodb are not nice when handler 
instances are created".

This patch has effect on storage engines that rely on
ha_open() PSEA method being called under LOCK_open.
In particular:

1) NDB is broken and left unfixed. NDB relies on LOCK_open
being kept as part of ha_open(), since it uses auto-discovery.
While previously the NDB open code was race-prone, now
it simply fails on asserts.

2) HEAP engine had a race in ha_heap::open() when
a share for the same table could be added twice
to the list of shares, or a dangling reference to a share
stored in HEAP handler. This patch aims to address this
problem by 'pinning' the newly created share in the 
internal HEAP engine share list until at least one
handler instance is created using that share.


include/heap.h:
  Add members to HP_CREATE_INFO.
  Declare heap_release_share().
sql/lock.cc:
  Remove thd->version, use thd->open_tables->s->version instead.
sql/repl_failsafe.cc:
  Remove thd->version.
sql/sql_base.cc:
  - close_thread_table(): move handler cleanup code outside the critical section protected by LOCK_open.
  - remove thd->version
  - split the large critical section in open_table() that
  opens a new table from share and is protected by LOCK_open
  into 2 critical sections, thus reducing the critical path.
  - make check_if_table_exists() acquire LOCK_open internally.
  - use thd->open_tables->s->version instead of thd->refresh_version to make sure that all tables in
  thd->open_tables are in the same refresh series.
sql/sql_base.h:
  Add declaration for check_if_table_exists().
sql/sql_class.cc:
  Remove init_open_tables_state(), it's now equal to
  reset_open_tables_state().
sql/sql_class.h:
  Remove thd->version, THD::init_open_tables_state().
sql/sql_plugin.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
sql/sql_udf.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
sql/table.h:
  Remove an unused variable.
sql/tztime.cc:
  Use table->m_needs_reopen to mark the table as stale
  rather than manipulate with thd->version, which is no more.
storage/heap/CMakeLists.txt:
  Add heap tests to cmake build files.
storage/heap/ha_heap.cc:
  Fix a race when ha_heap::ha_open() could insert two 
  HP_SHARE objects into the internal share list or store
  a dangling reference to a share in ha_heap instance,
  or wrongly set implicit_emptied.
storage/heap/hp_create.c:
  Optionally pin a newly created share in the list of shares
  by increasing its open_count. This is necessary to 
  make sure that a newly created share doesn't disappear while
  a HP_INFO object is being created to reference it.
storage/heap/hp_open.c:
  When adding a new HP_INFO object to the list of objects
  in the heap share, make sure the open_count is not increased
  twice.
storage/heap/hp_test1.c:
  Adjust the test to new function signatures.
storage/heap/hp_test2.c:
  Adjust the test to new function signatures.

6 files changed, 137 insertions, 61 deletions

diff --git a/storage/heap/CMakeLists.txt b/storage/heap/CMakeLists.txt
index 32359759abc..32510ba0c79 100755
--- a/storage/heap/CMakeLists.txt
+++ b/storage/heap/CMakeLists.txt
@@ -23,3 +23,9 @@ SET(HEAP_SOURCES  _check.c _rectest.c hp_block.c hp_clear.c hp_close.c hp_create
 				hp_rrnd.c hp_rsame.c hp_scan.c hp_static.c hp_update.c hp_write.c)
 
 MYSQL_ADD_PLUGIN(heap ${HEAP_SOURCES} STORAGE_ENGINE MANDATORY RECOMPILE_FOR_EMBEDDED)
+
+ADD_EXECUTABLE(hp_test1 hp_test1.c)
+TARGET_LINK_LIBRARIES(hp_test1 mysys heap)
+
+ADD_EXECUTABLE(hp_test2 hp_test2.c)
+TARGET_LINK_LIBRARIES(hp_test2 mysys heap)
diff --git a/storage/heap/ha_heap.cc b/storage/heap/ha_heap.cc
index 3abffc7087f..541650bd5e8 100644
--- a/storage/heap/ha_heap.cc
+++ b/storage/heap/ha_heap.cc
@@ -29,6 +29,10 @@
 static handler *heap_create_handler(handlerton *hton,
                                     TABLE_SHARE *table, 
                                     MEM_ROOT *mem_root);
+static int
+heap_prepare_hp_create_info(TABLE *table_arg, bool internal_table,
+                            HP_CREATE_INFO *hp_create_info);
+
 
 int heap_panic(handlerton *hton, ha_panic_function flag)
 {
@@ -96,43 +100,48 @@ const char **ha_heap::bas_ext() const
 
 int ha_heap::open(const char *name, int mode, uint test_if_locked)
 {
-  if ((test_if_locked & HA_OPEN_INTERNAL_TABLE) ||
-      (!(file= heap_open(name, mode)) && my_errno == ENOENT))
+  internal_table= test(test_if_locked & HA_OPEN_INTERNAL_TABLE);
+  if (internal_table || (!(file= heap_open(name, mode)) && my_errno == ENOENT))
   {
-    HA_CREATE_INFO create_info;
-    internal_table= test(test_if_locked & HA_OPEN_INTERNAL_TABLE);
-    bzero(&create_info, sizeof(create_info));
+    HP_CREATE_INFO create_info;
+    my_bool created_new_share;
+    int rc;
     file= 0;
-    if (!create(name, table, &create_info))
+    if (heap_prepare_hp_create_info(table, internal_table, &create_info))
+      goto end;
+    create_info.pin_share= TRUE;
+
+    rc= heap_create(name, &create_info, &internal_share, &created_new_share);
+    my_free((uchar*) create_info.keydef, MYF(0));
+    if (rc)
+      goto end;
+
+    implicit_emptied= test(created_new_share);
+    if (internal_table)
+      file= heap_open_from_share(internal_share, mode);
+    else
+      file= heap_open_from_share_and_register(internal_share, mode);
+
+    if (!file)
     {
-        file= internal_table ?
-          heap_open_from_share(internal_share, mode) :
-          heap_open_from_share_and_register(internal_share, mode);
-      if (!file)
-      {
-         /* Couldn't open table; Remove the newly created table */
-        mysql_mutex_lock(&THR_LOCK_heap);
-        hp_free(internal_share);
-        mysql_mutex_unlock(&THR_LOCK_heap);
-      }
-      implicit_emptied= 1;
+      heap_release_share(internal_share, internal_table);
+      goto end;
     }
   }
+
   ref_length= sizeof(HEAP_PTR);
-  if (file)
-  {
-    /* Initialize variables for the opened table */
-    set_keys_for_scanning();
-    /*
-      We cannot run update_key_stats() here because we do not have a
-      lock on the table. The 'records' count might just be changed
-      temporarily at this moment and we might get wrong statistics (Bug
-      #10178). Instead we request for update. This will be done in
-      ha_heap::info(), which is always called before key statistics are
-      used.
+  /* Initialize variables for the opened table */
+  set_keys_for_scanning();
+  /*
+    We cannot run update_key_stats() here because we do not have a
+    lock on the table. The 'records' count might just be changed
+    temporarily at this moment and we might get wrong statistics (Bug
+    #10178). Instead we request for update. This will be done in
+    ha_heap::info(), which is always called before key statistics are
+    used.
     */
-    key_stat_version= file->s->key_stat_version-1;
-  }
+  key_stat_version= file->s->key_stat_version-1;
+end:
   return (file ? 0 : 1);
 }
 
@@ -624,18 +633,20 @@ ha_rows ha_heap::records_in_range(uint inx, key_range *min_key,
 }
 
 
-int ha_heap::create(const char *name, TABLE *table_arg,
-		    HA_CREATE_INFO *create_info)
+static int
+heap_prepare_hp_create_info(TABLE *table_arg, bool internal_table,
+                            HP_CREATE_INFO *hp_create_info)
 {
   uint key, parts, mem_per_row= 0, keys= table_arg->s->keys;
   uint auto_key= 0, auto_key_type= 0;
   ha_rows max_rows;
   HP_KEYDEF *keydef;
   HA_KEYSEG *seg;
-  int error;
   TABLE_SHARE *share= table_arg->s;
   bool found_real_auto_increment= 0;
 
+  bzero(hp_create_info, sizeof(*hp_create_info));
+
   for (key= parts= 0; key < keys; key++)
     parts+= table_arg->key_info[key].key_parts;
 
@@ -715,29 +726,45 @@ int ha_heap::create(const char *name, TABLE *table_arg,
     }
   }
   mem_per_row+= MY_ALIGN(share->reclength + 1, sizeof(char*));
-  max_rows = (ha_rows) (table_arg->in_use->variables.max_heap_table_size /
-			(ulonglong) mem_per_row);
   if (table_arg->found_next_number_field)
   {
     keydef[share->next_number_index].flag|= HA_AUTO_KEY;
     found_real_auto_increment= share->next_number_key_offset == 0;
   }
+  hp_create_info->auto_key= auto_key;
+  hp_create_info->auto_key_type= auto_key_type;
+  hp_create_info->max_table_size=current_thd->variables.max_heap_table_size;
+  hp_create_info->with_auto_increment= found_real_auto_increment;
+  hp_create_info->internal_table= internal_table;
+
+  max_rows= (ha_rows) (hp_create_info->max_table_size / mem_per_row);
+  if (share->max_rows && share->max_rows < max_rows)
+    max_rows= share->max_rows;
+
+  hp_create_info->max_records= (ulong) max_rows;
+  hp_create_info->min_records= (ulong) share->min_rows;
+  hp_create_info->keys= share->keys;
+  hp_create_info->reclength= share->reclength;
+  hp_create_info->keydef= keydef;
+  return 0;
+}
+
+
+int ha_heap::create(const char *name, TABLE *table_arg,
+		    HA_CREATE_INFO *create_info)
+{
+  int error;
+  my_bool created;
   HP_CREATE_INFO hp_create_info;
-  hp_create_info.auto_key= auto_key;
-  hp_create_info.auto_key_type= auto_key_type;
+
+  error= heap_prepare_hp_create_info(table_arg, internal_table,
+                                     &hp_create_info);
+  if (error)
+    return error;
   hp_create_info.auto_increment= (create_info->auto_increment_value ?
 				  create_info->auto_increment_value - 1 : 0);
-  hp_create_info.max_table_size=current_thd->variables.max_heap_table_size;
-  hp_create_info.with_auto_increment= found_real_auto_increment;
-  hp_create_info.internal_table= internal_table;
-  max_rows = (ha_rows) (hp_create_info.max_table_size / mem_per_row);
-  error= heap_create(name,
-		     keys, keydef, share->reclength,
-		     (ulong) ((share->max_rows < max_rows &&
-			       share->max_rows) ? 
-			      share->max_rows : max_rows),
-		     (ulong) share->min_rows, &hp_create_info, &internal_share);
-  my_free((uchar*) keydef, MYF(0));
+  error= heap_create(name, &hp_create_info, &internal_share, &created);
+  my_free((uchar*) hp_create_info.keydef, MYF(0));
   DBUG_ASSERT(file == 0);
   return (error);
 }
diff --git a/storage/heap/hp_create.c b/storage/heap/hp_create.c
index 85e632e5aad..cf0f5d5ba6d 100644
--- a/storage/heap/hp_create.c
+++ b/storage/heap/hp_create.c
@@ -21,24 +21,30 @@ static void init_block(HP_BLOCK *block,uint reclength,ulong min_records,
 
 /* Create a heap table */
 
-int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
-		uint reclength, ulong max_records, ulong min_records,
-		HP_CREATE_INFO *create_info, HP_SHARE **res)
+int heap_create(const char *name, HP_CREATE_INFO *create_info,
+                HP_SHARE **res, my_bool *created_new_share)
 {
   uint i, j, key_segs, max_length, length;
   HP_SHARE *share= 0;
   HA_KEYSEG *keyseg;
+  HP_KEYDEF *keydef= create_info->keydef;
+  uint reclength= create_info->reclength;
+  uint keys= create_info->keys;
+  ulong min_records= create_info->min_records;
+  ulong max_records= create_info->max_records;
   DBUG_ENTER("heap_create");
 
   if (!create_info->internal_table)
   {
     mysql_mutex_lock(&THR_LOCK_heap);
-    if ((share= hp_find_named_heap(name)) && share->open_count == 0)
+    share= hp_find_named_heap(name);
+    if (share && share->open_count == 0)
     {
       hp_free(share);
       share= 0;
     }
-  }  
+  }
+  *created_new_share= (share == NULL);
 
   if (!share)
   {
@@ -200,7 +206,11 @@ int heap_create(const char *name, uint keys, HP_KEYDEF *keydef,
       share->delete_on_close= 1;
   }
   if (!create_info->internal_table)
+  {
+    if (create_info->pin_share)
+      ++share->open_count;
     mysql_mutex_unlock(&THR_LOCK_heap);
+  }
 
   *res= share;
   DBUG_RETURN(0);
diff --git a/storage/heap/hp_open.c b/storage/heap/hp_open.c
index feafa5d5cf1..ef2ce15f9b3 100644
--- a/storage/heap/hp_open.c
+++ b/storage/heap/hp_open.c
@@ -74,12 +74,33 @@ HP_INFO *heap_open_from_share_and_register(HP_SHARE *share, int mode)
   {
     info->open_list.data= (void*) info;
     heap_open_list= list_add(heap_open_list,&info->open_list);
+    /* Unpin the share, it is now pinned by the file. */
+    share->open_count--;
   }
   mysql_mutex_unlock(&THR_LOCK_heap);
   DBUG_RETURN(info);
 }
 
 
+/**
+  Dereference a HEAP share and free it if it's not referenced.
+  We don't check open_count for internal tables since they
+  are always thread-local, i.e. referenced by a single thread.
+*/
+void heap_release_share(HP_SHARE *share, my_bool internal_table)
+{
+  /* Couldn't open table; Remove the newly created table */
+  if (internal_table)
+    hp_free(share);
+  else
+  {
+    mysql_mutex_lock(&THR_LOCK_heap);
+    if (--share->open_count == 0)
+      hp_free(share);
+    mysql_mutex_unlock(&THR_LOCK_heap);
+  }
+}
+
 /*
   Open heap table based on name
 
diff --git a/storage/heap/hp_test1.c b/storage/heap/hp_test1.c
index 911e3a285a2..535db60e237 100644
--- a/storage/heap/hp_test1.c
+++ b/storage/heap/hp_test1.c
@@ -38,6 +38,7 @@ int main(int argc, char **argv)
   HA_KEYSEG keyseg[4];
   HP_CREATE_INFO hp_create_info;
   HP_SHARE *tmp_share;
+  my_bool unused;
   MY_INIT(argv[0]);
 
   filename= "test1";
@@ -45,6 +46,11 @@ int main(int argc, char **argv)
 
   bzero(&hp_create_info, sizeof(hp_create_info));
   hp_create_info.max_table_size= 1024L*1024L;
+  hp_create_info.keys= 1;
+  hp_create_info.keydef= keyinfo;
+  hp_create_info.reclength= 30;
+  hp_create_info.max_records= (ulong) flag*100000L;
+  hp_create_info.min_records= 10UL;
 
   keyinfo[0].keysegs=1;
   keyinfo[0].seg=keyseg;
@@ -55,13 +61,12 @@ int main(int argc, char **argv)
   keyinfo[0].seg[0].charset= &my_charset_latin1;
   keyinfo[0].seg[0].null_bit= 0;
   keyinfo[0].flag = HA_NOSAME;
-  
+
   deleted=0;
   bzero((uchar*) flags,sizeof(flags));
 
   printf("- Creating heap-file\n");
-  if (heap_create(filename,1,keyinfo,30,(ulong) flag*100000L,10L,
-		  &hp_create_info, &tmp_share) ||
+  if (heap_create(filename, &hp_create_info, &tmp_share, &unused) ||
       !(file= heap_open(filename, 2)))
     goto err;
   printf("- Writing records:s\n");
diff --git a/storage/heap/hp_test2.c b/storage/heap/hp_test2.c
index 8216c7360b4..733dff40c1a 100644
--- a/storage/heap/hp_test2.c
+++ b/storage/heap/hp_test2.c
@@ -65,15 +65,21 @@ int main(int argc, char *argv[])
   HEAP_PTR UNINIT_VAR(position);
   HP_CREATE_INFO hp_create_info;
   CHARSET_INFO *cs= &my_charset_latin1;
+  my_bool unused;
   MY_INIT(argv[0]);		/* init my_sys library & pthreads */
 
   filename= "test2";
   filename2= "test2_2";
   file=file2=0;
   get_options(argc,argv);
-  
+
   bzero(&hp_create_info, sizeof(hp_create_info));
   hp_create_info.max_table_size= 1024L*1024L;
+  hp_create_info.keys= keys;
+  hp_create_info.keydef= keyinfo;
+  hp_create_info.reclength= reclength;
+  hp_create_info.max_records= (ulong) flag*100000L;
+  hp_create_info.min_records= (ulong) recant/2;
 
   write_count=update=opt_delete=0;
   key_check=0;
@@ -125,8 +131,7 @@ int main(int argc, char *argv[])
   bzero((char*) key3,sizeof(key3));
 
   printf("- Creating heap-file\n");
-  if (heap_create(filename,keys,keyinfo,reclength,(ulong) flag*100000L, 
-                  (ulong) recant/2, &hp_create_info, &tmp_share) ||
+  if (heap_create(filename, &hp_create_info, &tmp_share, &unused) ||
       !(file= heap_open(filename, 2)))
     goto err;
   signal(SIGINT,endprog);
@@ -563,8 +568,10 @@ int main(int argc, char *argv[])
   heap_close(file2);
 
   printf("- Creating output heap-file 2\n");
-  if (heap_create(filename2, 1, keyinfo, reclength, 0L, 0L, &hp_create_info,
-                  &tmp_share) ||
+  hp_create_info.keys= 1;
+  hp_create_info.max_records= 0;
+  hp_create_info.min_records= 0;
+  if (heap_create(filename2, &hp_create_info, &tmp_share, &unused) ||
       !(file2= heap_open_from_share_and_register(tmp_share, 2)))
     goto err;