Fixed memory loss detected on P8. This can happen when we call after_flush but...

Fixed memory loss detected on P8. This can happen when we call after_flush but never call after_rollback() or after_commit().

The old code used pthread_setspecific() to store temporary data used by the thread.
This is not safe when used with thread pool, as the thread may change for the transaction.

The fix is to save the data in THD, which is guaranteed to be properly freed.
I also fixed the code so that we don't do a malloc() for every transaction.

sql/rpl_handler.cc

@@ -38,8 +38,6 @@ typedef struct Trans_binlog_info {
   char log_file[FN_REFLEN];
 } Trans_binlog_info;
 
-static pthread_key(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO);
-
 int get_user_var_int(const char *name,
                      long long int *value, int *null_value)
 {
@@ -143,13 +141,6 @@ int delegates_init()
   }
 #endif
 
-  if (pthread_key_create(&RPL_TRANS_BINLOG_INFO, NULL))
-  {
-    sql_print_error("Error while creating pthread specific data key for replication. "
-                    "Please report a bug.");
-    return 1;
-  }
-
   return 0;
 }
 
@@ -195,27 +186,27 @@ void delegates_destroy()
 int Trans_delegate::after_commit(THD *thd, bool all)
 {
   Trans_param param;
+  Trans_binlog_info *log_info;
   bool is_real_trans= (all || thd->transaction.all.ha_list == 0);
+  int ret= 0;
 
   param.flags = is_real_trans ? TRANS_IS_REAL_TRANS : 0;
 
-  Trans_binlog_info *log_info=
-    my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO);
+  log_info= thd->semisync_info;
 
-  param.log_file= log_info ? log_info->log_file : 0;
+  param.log_file= log_info && log_info->log_file[0] ? log_info->log_file : 0;
   param.log_pos= log_info ? log_info->log_pos : 0;
 
-  int ret= 0;
   FOREACH_OBSERVER(ret, after_commit, false, (&param));
 
   /*
     This is the end of a real transaction or autocommit statement, we
-    can free the memory allocated for binlog file and position.
+    can mark the memory unused.
   */
   if (is_real_trans && log_info)
   {
-    my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, NULL);
-    my_free(log_info);
+    log_info->log_file[0]= 0;
+    log_info->log_pos= 0;
   }
   return ret;
 }
@@ -223,27 +214,27 @@ int Trans_delegate::after_commit(THD *thd, bool all)
 int Trans_delegate::after_rollback(THD *thd, bool all)
 {
   Trans_param param;
+  Trans_binlog_info *log_info;
   bool is_real_trans= (all || thd->transaction.all.ha_list == 0);
+  int ret= 0;
 
   param.flags = is_real_trans ? TRANS_IS_REAL_TRANS : 0;
 
-  Trans_binlog_info *log_info=
-    my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO);
-    
-  param.log_file= log_info ? log_info->log_file : 0;
+  log_info= thd->semisync_info;
+
+  param.log_file= log_info && log_info->log_file[0] ? log_info->log_file : 0;
   param.log_pos= log_info ? log_info->log_pos : 0;
 
-  int ret= 0;
   FOREACH_OBSERVER(ret, after_rollback, false, (&param));
 
   /*
     This is the end of a real transaction or autocommit statement, we
-    can free the memory allocated for binlog file and position.
+    can mark the memory unused.
   */
   if (is_real_trans && log_info)
   {
-    my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, NULL);
-    my_free(log_info);
+    log_info->log_file[0]= 0;
+    log_info->log_pos= 0;
   }
   return ret;
 }
@@ -254,25 +245,24 @@ int Binlog_storage_delegate::after_flush(THD *thd,
                                          bool synced)
 {
   Binlog_storage_param param;
+  Trans_binlog_info *log_info;
   uint32 flags=0;
+  int ret= 0;
+
   if (synced)
     flags |= BINLOG_STORAGE_IS_SYNCED;
 
-  Trans_binlog_info *log_info=
-    my_pthread_getspecific_ptr(Trans_binlog_info*, RPL_TRANS_BINLOG_INFO);
-    
-  if (!log_info)
+  if (!(log_info= thd->semisync_info))
   {
     if(!(log_info=
-         (Trans_binlog_info *)my_malloc(sizeof(Trans_binlog_info), MYF(0))))
+         (Trans_binlog_info*) my_malloc(sizeof(Trans_binlog_info), MYF(0))))
       return 1;
-    my_pthread_setspecific_ptr(RPL_TRANS_BINLOG_INFO, log_info);
+    thd->semisync_info= log_info;
   }
-    
+
   strcpy(log_info->log_file, log_file+dirname_length(log_file));
   log_info->log_pos = log_pos;
   
-  int ret= 0;
   FOREACH_OBSERVER(ret, after_flush, false,
                    (&param, log_info->log_file, log_info->log_pos, flags));
   return ret;

sql/sql_class.cc

@@ -877,6 +877,7 @@ THD::THD()
   file_id = 0;
   query_id= 0;
   query_name_consts= 0;
+  semisync_info= 0;
   db_charset= global_system_variables.collation_database;
   bzero(ha_data, sizeof(ha_data));
   mysys_var=0;
@@ -1493,6 +1494,7 @@ THD::~THD()
   mysql_audit_free_thd(this);
   if (rli_slave)
     rli_slave->cleanup_after_session();
+  my_free(semisync_info);
 #endif
 
   free_root(&main_mem_root, MYF(0));

sql/sql_class.h

@@ -47,7 +47,6 @@
 
 class Reprepare_observer;
 class Relay_log_info;
-
 class Query_log_event;
 class Load_log_event;
 class Slave_log_event;
@@ -59,6 +58,7 @@ class Rows_log_event;
 class Sroutine_hash_entry;
 class User_level_lock;
 class user_var_entry;
+class Trans_binlog_info;
 
 enum enum_enable_or_disable { LEAVE_AS_IS, ENABLE, DISABLE };
 enum enum_ha_read_modes { RFIRST, RNEXT, RPREV, RLAST, RKEY, RNEXT_SAME };
@@ -1670,6 +1670,9 @@ public:
   */
   const char *where;
 
+  /* Needed by MariaDB semi sync replication */
+  Trans_binlog_info *semisync_info;
+
   ulong client_capabilities;		/* What the client supports */
   ulong max_client_packet_length;