in mysql_unlock_tables(), do thr_unlock() AFTER external_unlock().

it means, {update,restore}_status() should be called in external_lock, 
not in thr_unlock. Only affects storage engines that support
TL_WRITE_CONCURRENT.

mysys/thr_lock.c

@@ -24,7 +24,7 @@ Locks are prioritized according to:
 
 The current lock types are:
 
-TL_READ	 		# Low priority read
+TL_READ                 # Low priority read
 TL_READ_WITH_SHARED_LOCKS
 TL_READ_HIGH_PRIORITY	# High priority read
 TL_READ_NO_INSERT	# Read without concurrent inserts
@@ -57,8 +57,12 @@ should put a pointer to the following functions in the lock structure:
 	 In MyISAM this is a simple check if the insert can be done
 	 at the end of the datafile.
 update_status:
-	Before a write lock is released, this function is called.
-	In MyISAM this functions updates the count and length of the datafile
+        in thr_reschedule_write_lock(), when an insert delayed thread
+        downgrades TL_WRITE lock to TL_WRITE_DELAYED, to allow SELECT
+        threads to proceed.
+        A storage engine should also call update_status internally
+        in the ::external_lock(F_UNLCK) method.
+        In MyISAM and CSV this functions updates the length of the datafile.
 get_status:
 	When one gets a lock this functions is called.
 	In MyISAM this stores the number of rows and size of the datafile
@@ -762,16 +766,6 @@ void thr_unlock(THR_LOCK_DATA *data)
   }
   else
     lock->write.last=data->prev;
-  if (lock_type >= TL_WRITE_CONCURRENT_INSERT)
-  {
-    if (lock->update_status)
-      (*lock->update_status)(data->status_param);
-  }
-  else
-  {
-    if (lock->restore_status)
-      (*lock->restore_status)(data->status_param);
-  }
   if (lock_type == TL_READ_NO_INSERT)
     lock->read_no_write_count--;
   data->type=TL_UNLOCK;				/* Mark unlocked */

sql/lock.cc

@@ -281,10 +281,10 @@ static int lock_external(THD *thd, TABLE **tables, uint count)
 void mysql_unlock_tables(THD *thd, MYSQL_LOCK *sql_lock)
 {
   DBUG_ENTER("mysql_unlock_tables");
-  if (sql_lock->lock_count)
-    thr_multi_unlock(sql_lock->locks,sql_lock->lock_count);
   if (sql_lock->table_count)
     VOID(unlock_external(thd,sql_lock->table,sql_lock->table_count));
+  if (sql_lock->lock_count)
+    thr_multi_unlock(sql_lock->locks,sql_lock->lock_count);
   my_free((gptr) sql_lock,MYF(0));
   DBUG_VOID_RETURN;
 }

storage/csv/ha_tina.cc

@@ -441,7 +441,7 @@ ha_tina::ha_tina(handlerton *hton, TABLE_SHARE *table_arg)
   */
   current_position(0), next_position(0), local_saved_data_file_length(0),
   file_buff(0), chain_alloced(0), chain_size(DEFAULT_CHAIN_LENGTH),
-  records_is_known(0)
+  records_is_known(0), curr_lock_type(F_UNLCK)
 {
   /* Set our original buffers from pre-allocated memory */
   buffer.set((char*)byte_buffer, IO_SIZE, system_charset_info);
@@ -1394,6 +1394,14 @@ int ha_tina::delete_all_rows()
   DBUG_RETURN(rc);
 }
 
+int ha_tina::external_lock(THD *thd __attribute__((unused)), int lock_type)
+{
+  if (lock_type==F_UNLCK && curr_lock_type == F_WRLCK)
+    update_status();
+  curr_lock_type= lock_type;
+  return 0;
+}
+
 /*
   Called by the database to lock the table. Keep in mind that this
   is an internal lock.
@@ -1408,7 +1416,7 @@ THR_LOCK_DATA **ha_tina::store_lock(THD *thd,
   return to;
 }
 
-/* 
+/*
   Create a table. You do not want to leave the table open after a call to
   this (the database will call ::open() if it needs to).
 */

storage/csv/ha_tina.h

@@ -81,6 +81,8 @@ class ha_tina: public handler
   bool records_is_known;
 
 private:
+  int curr_lock_type;
+
   bool get_write_pos(off_t *end_pos, tina_set *closest_hole);
   int open_update_temp_file_if_needed();
   int init_tina_writer();
@@ -153,6 +155,8 @@ class ha_tina: public handler
   bool check_if_incompatible_data(HA_CREATE_INFO *info,
                                   uint table_changes);
 
+  int external_lock(THD *thd, int lock_type);
+
   THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
       enum thr_lock_type lock_type);
 

storage/maria/ma_locking.c

@@ -55,9 +55,15 @@ int maria_lock_database(MARIA_HA *info, int lock_type)
     case F_UNLCK:
       maria_ftparser_call_deinitializer(info);
       if (info->lock_type == F_RDLCK)
+      {
 	count= --share->r_locks;
+        _ma_restore_status(info);
+      }
       else
+      {
 	count= --share->w_locks;
+        _ma_update_status(info);
+      }
       --share->tot_locks;
       if (info->lock_type == F_WRLCK && !share->w_locks)
       {

storage/myisam/mi_locking.c

@@ -56,9 +56,15 @@ int mi_lock_database(MI_INFO *info, int lock_type)
     case F_UNLCK:
       ftparser_call_deinitializer(info);
       if (info->lock_type == F_RDLCK)
+      {
 	count= --share->r_locks;
+        mi_restore_status(info);
+      }
       else
+      {
 	count= --share->w_locks;
+        mi_update_status(info);
+      }
       --share->tot_locks;
       if (info->lock_type == F_WRLCK && !share->w_locks &&
 	  !share->delay_key_write && flush_key_blocks(share->key_cache,
@@ -84,16 +90,16 @@ int mi_lock_database(MI_INFO *info, int lock_type)
 	if (share->changed && !share->w_locks)
 	{
 #ifdef HAVE_MMAP
-    if ((info->s->mmaped_length != info->s->state.state.data_file_length) &&
-        (info->s->nonmmaped_inserts > MAX_NONMAPPED_INSERTS))
-    {
-      if (info->s->concurrent_insert)
-        rw_wrlock(&info->s->mmap_lock);
-      mi_remap_file(info, info->s->state.state.data_file_length);
-      info->s->nonmmaped_inserts= 0;
-      if (info->s->concurrent_insert)
-        rw_unlock(&info->s->mmap_lock);
-    }
+          if ((info->s->mmaped_length != info->s->state.state.data_file_length) &&
+              (info->s->nonmmaped_inserts > MAX_NONMAPPED_INSERTS))
+          {
+            if (info->s->concurrent_insert)
+              rw_wrlock(&info->s->mmap_lock);
+            mi_remap_file(info, info->s->state.state.data_file_length);
+            info->s->nonmmaped_inserts= 0;
+            if (info->s->concurrent_insert)
+              rw_unlock(&info->s->mmap_lock);
+          }
 #endif
 	  share->state.process= share->last_process=share->this_process;
 	  share->state.unique=   info->last_unique=  info->this_unique;
@@ -300,6 +306,7 @@ void mi_get_status(void* param, int concurrent_insert)
 void mi_update_status(void* param)
 {
   MI_INFO *info=(MI_INFO*) param;
+  DBUG_ENTER("mi_update_status");
   /*
     Because someone may have closed the table we point at, we only
     update the state if its our own state.  This isn't a problem as
@@ -336,20 +343,32 @@ void mi_update_status(void* param)
     }
     info->opt_flag&= ~WRITE_CACHE_USED;
   }
+  DBUG_VOID_RETURN;
 }
 
 
 void mi_restore_status(void *param)
 {
   MI_INFO *info= (MI_INFO*) param;
+  DBUG_ENTER("mi_restore_status");
+  DBUG_PRINT("info",("key_file: %ld  data_file: %ld",
+		     (long) info->s->state.state.key_file_length,
+		     (long) info->s->state.state.data_file_length));
   info->state= &info->s->state.state;
   info->append_insert_at_end= 0;
+  DBUG_VOID_RETURN;
 }
 
 
 void mi_copy_status(void* to,void *from)
 {
-  ((MI_INFO*) to)->state= &((MI_INFO*) from)->save_state;
+  MI_INFO *info= (MI_INFO*) to;
+  DBUG_ENTER("mi_copy_status");
+  info->state= &((MI_INFO*) from)->save_state;
+  DBUG_PRINT("info",("key_file: %ld  data_file: %ld",
+		     (long) info->state->key_file_length,
+		     (long) info->state->data_file_length));
+  DBUG_VOID_RETURN;
 }
 
 
@@ -377,17 +396,18 @@ void mi_copy_status(void* to,void *from)
 my_bool mi_check_status(void *param)
 {
   MI_INFO *info=(MI_INFO*) param;
+  DBUG_ENTER("mi_check_status");
+  DBUG_PRINT("info",("dellink: %ld  r_locks: %u  w_locks: %u",
+                     (long) info->s->state.dellink, (uint) info->s->r_locks,
+                     (uint) info->s->w_locks));
   /*
     The test for w_locks == 1 is here because this thread has already done an
     external lock (in other words: w_locks == 1 means no other threads has
     a write lock)
   */
-  DBUG_PRINT("info",("dellink: %ld  r_locks: %u  w_locks: %u",
-                     (long) info->s->state.dellink, (uint) info->s->r_locks,
-                     (uint) info->s->w_locks));
-  return (my_bool) !(info->s->state.dellink == HA_OFFSET_ERROR ||
+  DBUG_RETURN((my_bool) !(info->s->state.dellink == HA_OFFSET_ERROR ||
                      (myisam_concurrent_insert == 2 && info->s->r_locks &&
-                      info->s->w_locks == 1));
+                      info->s->w_locks == 1)));
 }