Bug#26241

  "Blackhole tables don't honor table locks"
  Implement neccessary shared lock structure for table locks.
  Imported test case created by Giuseppe Maxia

storage/blackhole/ha_blackhole.cc

@@ -31,6 +31,14 @@ static handler *blackhole_create_handler(handlerton *hton,
 }
 
 
+/* Static declarations for shared structures */
+
+static pthread_mutex_t blackhole_mutex;
+static HASH blackhole_open_tables;
+
+static st_blackhole_share *get_share(const char *table_name);
+static void free_share(st_blackhole_share *share);
+
 /*****************************************************************************
 ** BLACKHOLE tables
 *****************************************************************************/
@@ -53,15 +61,18 @@ const char **ha_blackhole::bas_ext() const
 int ha_blackhole::open(const char *name, int mode, uint test_if_locked)
 {
   DBUG_ENTER("ha_blackhole::open");
-  thr_lock_init(&thr_lock);
-  thr_lock_data_init(&thr_lock,&lock,NULL);
+
+  if (!(share= get_share(name)))
+    DBUG_RETURN(HA_ERR_OUT_OF_MEM);
+
+  thr_lock_data_init(&share->lock, &lock, NULL);
   DBUG_RETURN(0);
 }
 
 int ha_blackhole::close(void)
 {
   DBUG_ENTER("ha_blackhole::close");
-  thr_lock_delete(&thr_lock);
+  free_share(share);
   DBUG_RETURN(0);
 }
 
@@ -136,17 +147,39 @@ int ha_blackhole::external_lock(THD *thd, int lock_type)
 }
 
 
-uint ha_blackhole::lock_count(void) const
-{
-  DBUG_ENTER("ha_blackhole::lock_count");
-  DBUG_RETURN(0);
-}
-
 THR_LOCK_DATA **ha_blackhole::store_lock(THD *thd,
                                          THR_LOCK_DATA **to,
                                          enum thr_lock_type lock_type)
 {
   DBUG_ENTER("ha_blackhole::store_lock");
+  if (lock_type != TL_IGNORE && lock.type == TL_UNLOCK)
+  {
+    /*
+      Here is where we get into the guts of a row level lock.
+      If TL_UNLOCK is set
+      If we are not doing a LOCK TABLE or DISCARD/IMPORT
+      TABLESPACE, then allow multiple writers
+    */
+
+    if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
+         lock_type <= TL_WRITE) && !thd_in_lock_tables(thd)
+        && !thd_tablespace_op(thd))
+      lock_type = TL_WRITE_ALLOW_WRITE;
+
+    /*
+      In queries of type INSERT INTO t1 SELECT ... FROM t2 ...
+      MySQL would use the lock TL_READ_NO_INSERT on t2, and that
+      would conflict with TL_WRITE_ALLOW_WRITE, blocking all inserts
+      to t2. Convert the lock to a normal read lock to allow
+      concurrent inserts to t2.
+    */
+
+    if (lock_type == TL_READ_NO_INSERT && !thd_in_lock_tables(thd))
+      lock_type = TL_READ;
+
+    lock.type= lock_type;
+  }
+  *to++= &lock;
   DBUG_RETURN(to);
 }
 
@@ -204,6 +237,63 @@ int ha_blackhole::index_last(byte * buf)
   DBUG_RETURN(HA_ERR_END_OF_FILE);
 }
 
+
+static st_blackhole_share *get_share(const char *table_name)
+{
+  st_blackhole_share *share;
+  uint length;
+
+  length= (uint) strlen(table_name);
+  pthread_mutex_lock(&blackhole_mutex);
+    
+  if (!(share= (st_blackhole_share*) hash_search(&blackhole_open_tables,
+                                                 (byte*) table_name, length)))
+  {
+    if (!(share= (st_blackhole_share*) my_malloc(sizeof(st_blackhole_share) +
+                                                 length,
+                                                 MYF(MY_WME | MY_ZEROFILL))))
+      goto error;
+
+    share->table_name_length= length;
+    strmov(share->table_name, table_name);
+    
+    if (my_hash_insert(&blackhole_open_tables, (byte*) share))
+    {
+      my_free((gptr) share, MYF(0));
+      share= NULL;
+      goto error;
+    }
+    
+    thr_lock_init(&share->lock);
+  }
+  share->use_count++;
+  
+error:
+  pthread_mutex_unlock(&blackhole_mutex);
+  return share;
+}
+
+static void free_share(st_blackhole_share *share)
+{
+  pthread_mutex_lock(&blackhole_mutex);
+  if (!--share->use_count)
+    hash_delete(&blackhole_open_tables, (byte*) share);
+  pthread_mutex_unlock(&blackhole_mutex);
+}
+
+static void blackhole_free_key(st_blackhole_share *share)
+{
+  thr_lock_delete(&share->lock);
+  my_free((gptr) share, MYF(0));
+}
+
+static byte* blackhole_get_key(st_blackhole_share *share, uint *length,
+                               my_bool not_used __attribute__((unused)))
+{
+  *length= share->table_name_length;
+  return (byte*) share->table_name;
+}
+
 static int blackhole_init(void *p)
 {
   handlerton *blackhole_hton;
@@ -212,6 +302,20 @@ static int blackhole_init(void *p)
   blackhole_hton->db_type= DB_TYPE_BLACKHOLE_DB;
   blackhole_hton->create= blackhole_create_handler;
   blackhole_hton->flags= HTON_CAN_RECREATE;
+  
+  VOID(pthread_mutex_init(&blackhole_mutex, MY_MUTEX_INIT_FAST));
+  (void) hash_init(&blackhole_open_tables, system_charset_info,32,0,0,
+                   (hash_get_key) blackhole_get_key,
+                   (hash_free_key) blackhole_free_key, 0);
+
+  return 0;
+}
+
+static int blackhole_fini(void *p)
+{
+  hash_free(&blackhole_open_tables);
+  pthread_mutex_destroy(&blackhole_mutex);
+
   return 0;
 }
 
@@ -227,7 +331,7 @@ mysql_declare_plugin(blackhole)
   "/dev/null storage engine (anything you write to it disappears)",
   PLUGIN_LICENSE_GPL,
   blackhole_init, /* Plugin Init */
-  NULL, /* Plugin Deinit */
+  blackhole_fini, /* Plugin Deinit */
   0x0100 /* 1.0 */,
   NULL,                       /* status variables                */
   NULL,                       /* system variables                */

storage/blackhole/ha_blackhole.h

@@ -17,6 +17,17 @@
 #pragma interface			/* gcc class implementation */
 #endif
 
+/*
+  Shared structure for correct LOCK operation
+*/
+struct st_blackhole_share {
+  THR_LOCK lock;
+  uint use_count;
+  uint table_name_length;
+  char table_name[1];
+};
+
+
 /*
   Class definition for the blackhole storage engine
   "Dumbest named feature ever"
@@ -24,7 +35,7 @@
 class ha_blackhole: public handler
 {
   THR_LOCK_DATA lock;      /* MySQL lock */
-  THR_LOCK thr_lock;
+  st_blackhole_share *share;
 
 public:
   ha_blackhole(handlerton *hton, TABLE_SHARE *table_arg);
@@ -76,7 +87,6 @@ public:
   void position(const byte *record);
   int info(uint flag);
   int external_lock(THD *thd, int lock_type);
-  uint lock_count(void) const;
   int create(const char *name, TABLE *table_arg,
              HA_CREATE_INFO *create_info);
   THR_LOCK_DATA **store_lock(THD *thd,