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.

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)

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);
 }

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

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
 

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");

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;