Bug#21464787 THE CHANNEL SERVICE INTERFACE LACKS A METHOD TO CHECK THE RECEIVER THREAD ID.

Currently the channel service interface lacks a method to check the
receiver thread id while having this method for the applier thread. In
fact the interface should be generic and always allow to execute the
same methods on both threads whenever it makes sense to.

The method is now generic and allows to check either the receiver or
the applier thread. It shall be useful to identify if the thread that
executed some hook is the thread from channel A or B.

Author: pedrolgomes

mysql-test/suite/rpl/r/rpl_replication_observers_example_plugin_channels_receiver_thread.result

@@ -0,0 +1,20 @@
+include/have_replication_observers_example_plugin.inc
+include/rpl_init.inc [topology=1->2]
+CHANGE MASTER TO MASTER_HOST="127.0.0.1", MASTER_USER="root", MASTER_PASSWORD="", MASTER_PORT=SERVER_1_PORT, MASTER_AUTO_POSITION=1 FOR CHANNEL "example_channel";
+Warnings:
+Note	1759	Sending passwords in plain text without SSL/TLS is extremely insecure.
+Note	1760	Storing MySQL user name or password information in the master info repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START SLAVE; see the 'START SLAVE Syntax' in the MySQL Manual for more information.
+include/start_slave.inc
+include/assert.inc ['The example_channel is ON']
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+include/sync_slave_sql_with_master.inc
+include/install_replication_observers_example.inc
+SET @debug_saved= @@GLOBAL.DEBUG;
+SET @@GLOBAL.DEBUG= '+d,validate_replication_observers_plugin_server_channel_io_thread';
+INSERT INTO t1 VALUES(1);
+include/assert.inc ['The example_channel is OFF']
+SET @@GLOBAL.DEBUG= @debug_saved;
+RESET SLAVE ALL;
+DROP TABLE t1;
+include/uninstall_replication_observers_example.inc
+DROP TABLE t1;

mysql-test/suite/rpl/t/rpl_replication_observers_example_plugin_channels_receiver_thread-master.opt

@@ -0,0 +1 @@
+$RPL_OBS_EXAMPLE_OPT

mysql-test/suite/rpl/t/rpl_replication_observers_example_plugin_channels_receiver_thread-slave.opt

@@ -0,0 +1 @@
+$RPL_OBS_EXAMPLE_OPT

mysql-test/suite/rpl/t/rpl_replication_observers_example_plugin_channels_receiver_thread.cnf

@@ -0,0 +1,5 @@
+!include ../my.cnf
+
+[mysqld.2]
+master-info-repository=TABLE
+relay-log-info-repository=TABLE

mysql-test/suite/rpl/t/rpl_replication_observers_example_plugin_channels_receiver_thread.test

@@ -0,0 +1,63 @@
+# This test verifies the basic functionalities of the channel service interface.
+# Using the example plugin, we execute several of the interface methods on a
+# channel's receiver thread.
+
+--source include/have_replication_observers_example_plugin.inc
+--source include/not_group_replication_plugin.inc
+--source include/not_embedded.inc
+--source include/have_debug.inc
+--source include/have_binlog_format_mixed.inc
+--source include/have_gtid.inc
+--let $rpl_topology= 1->2
+--let $rpl_skip_change_master= 1
+--let $rpl_skip_start_slave= 1
+--source include/rpl_init.inc
+
+# Establish a connection to a slave with an example channel
+
+--connection server_2
+--replace_result $SERVER_MYPORT_1 SERVER_1_PORT
+--eval CHANGE MASTER TO MASTER_HOST="127.0.0.1", MASTER_USER="root", MASTER_PASSWORD="", MASTER_PORT=$SERVER_MYPORT_1, MASTER_AUTO_POSITION=1 FOR CHANNEL "example_channel"
+--source include/start_slave.inc
+
+--let $assert_text= 'The example_channel is ON'
+--let $assert_cond= [SELECT COUNT(*) AS count FROM performance_schema.replication_connection_status where CHANNEL_NAME="example_channel" and SERVICE_STATE="ON", count, 1] = 1
+--source include/assert.inc
+
+--connection server_1
+
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+
+--let $wait_for_executed_gtid_set= 1
+--let $rpl_channel_name= example_channel
+--let $sync_slave_connection= server_2
+--source include/sync_slave_sql_with_master.inc
+
+# Install the replication observers example plugin on the slave
+
+--source include/install_replication_observers_example.inc
+
+# Execute a query that on a server hook will execute a test against the several
+# method of the channel interface on the receiver thread of the created channel
+
+SET @debug_saved= @@GLOBAL.DEBUG;
+SET @@GLOBAL.DEBUG= '+d,validate_replication_observers_plugin_server_channel_io_thread';
+--eval INSERT INTO t1 VALUES(1)
+
+# The tests will stop the channel.
+
+--let $assert_text= 'The example_channel is OFF'
+--let $assert_cond= [SELECT COUNT(*) AS count FROM performance_schema.replication_connection_status where CHANNEL_NAME="example_channel" and SERVICE_STATE="OFF", count, 1] = 1
+--source include/assert.inc
+
+# Clean
+
+SET @@GLOBAL.DEBUG= @debug_saved;
+
+RESET SLAVE ALL;
+DROP TABLE t1;
+
+--source include/uninstall_replication_observers_example.inc
+
+--connection server_1
+DROP TABLE t1;

plugin/replication_observers_example/replication_observers_example.cc

@@ -30,6 +30,7 @@ static MYSQL_PLUGIN plugin_info_ptr;
 int validate_plugin_server_requirements(Trans_param *param);
 int test_channel_service_interface_initialization();
 int test_channel_service_interface();
+int test_channel_service_interface_io_thread();
 
 /*
   Will register the number of calls to each method of Server state
@@ -198,6 +199,8 @@ int trans_before_dml(Trans_param *param, int& out_val)
                   out_val= 1;);
   DBUG_EXECUTE_IF("validate_replication_observers_plugin_server_channels",
                   test_channel_service_interface(););
+  DBUG_EXECUTE_IF("validate_replication_observers_plugin_server_channel_io_thread",
+                  test_channel_service_interface_io_thread(););
   DBUG_EXECUTE_IF("validate_replication_observers_plugin_server_channels_init",
                   test_channel_service_interface_initialization(););
   return 0;
@@ -677,8 +680,9 @@ int test_channel_service_interface()
 
     //Extract the applier id
     long unsigned int * applier_id= NULL;
-    channel_get_appliers_thread_id(interface_channel,
-                                   &applier_id);
+    channel_get_thread_id(interface_channel,
+                          CHANNEL_APPLIER_THREAD,
+                          &applier_id);
     DBUG_ASSERT(*applier_id > 0);
     my_free(applier_id);
 
@@ -708,6 +712,45 @@ int test_channel_service_interface()
     return (error && exists && running && gno);
 }
 
+int test_channel_service_interface_io_thread()
+{
+  //The initialization method should return OK
+  int error= initialize_channel_service_interface();
+  DBUG_ASSERT(!error);
+
+  char interface_channel[]= "example_channel";
+
+  //Assert the channel exists
+  bool exists= channel_is_active(interface_channel, CHANNEL_NO_THD);
+  DBUG_ASSERT(exists);
+
+  //Assert that the applier receiver is running
+  bool running= channel_is_active(interface_channel, CHANNEL_RECEIVER_THREAD);
+  DBUG_ASSERT(running);
+
+  //Extract the receiver id
+  long unsigned int * thread_id= NULL;
+  int num_threads= channel_get_thread_id(interface_channel,
+                                         CHANNEL_APPLIER_THREAD,
+                                         &thread_id);
+  DBUG_ASSERT(num_threads == 1);
+  DBUG_ASSERT(*thread_id > 0);
+  my_free(thread_id);
+
+
+  //Stop the channel
+  error= channel_stop(interface_channel,
+                      3,
+                      10000);
+  DBUG_ASSERT(!error);
+
+  //Assert that the receiver thread is not running
+  running= channel_is_active(interface_channel, CHANNEL_RECEIVER_THREAD);
+  DBUG_ASSERT(!running);
+
+  return (error && exists && running && num_threads);
+}
+
 /*
   Initialize the Replication Observer example at server start or plugin
   installation.

sql/rpl_channel_service_interface.cc

@@ -519,12 +519,13 @@ bool channel_is_active(const char* channel, enum_channel_thread_types thd_type)
   DBUG_RETURN(false);
 }
 
-int channel_get_appliers_thread_id(const char* channel,
-                                   unsigned long** appliers_id)
+int channel_get_thread_id(const char* channel,
+                          enum_channel_thread_types thd_type,
+                          unsigned long** thread_id)
 {
-  DBUG_ENTER("channel_get_appliers_thread_id(channel, *appliers_id");
+  DBUG_ENTER("channel_get_thread_id(channel, thread_type ,*thread_id");
 
-  int number_appliers= -1;
+  int number_threads= -1;
 
   Master_info *mi= msr_map.get_mi(channel);
 
@@ -533,45 +534,64 @@ int channel_get_appliers_thread_id(const char* channel,
     DBUG_RETURN(RPL_CHANNEL_SERVICE_CHANNEL_DOES_NOT_EXISTS_ERROR);
   }
 
-  if (mi->rli != NULL)
+  switch(thd_type)
   {
-    mysql_mutex_lock(&mi->rli->run_lock);
-
-    int num_workers= mi->rli->slave_parallel_workers;
-    if (num_workers > 1)
-    {
-      *appliers_id=
-          (unsigned long*) my_malloc(PSI_NOT_INSTRUMENTED,
-                                     num_workers * sizeof(unsigned long),
-                                     MYF(MY_WME));
-      unsigned long *appliers_id_pointer= *appliers_id;
-
-      for (int i= 0; i < num_workers; i++, appliers_id_pointer++)
+    case CHANNEL_RECEIVER_THREAD:
+      mysql_mutex_lock(&mi->info_thd_lock);
+      if (mi->info_thd != NULL)
       {
-        mysql_mutex_lock(&mi->rli->workers.at(i)->info_thd_lock);
-        *appliers_id_pointer= mi->rli->workers.at(i)->info_thd->thread_id();
-        mysql_mutex_unlock(&mi->rli->workers.at(i)->info_thd_lock);
+        *thread_id= (unsigned long*) my_malloc(PSI_NOT_INSTRUMENTED,
+                                               sizeof(unsigned long),
+                                               MYF(MY_WME));
+        **thread_id= mi->info_thd->thread_id();
+        number_threads= 1;
       }
-
-      number_appliers= num_workers;
-    }
-    else
-    {
-      if (mi->rli->info_thd != NULL)
+      mysql_mutex_unlock(&mi->info_thd_lock);
+      break;
+    case CHANNEL_APPLIER_THREAD:
+      if (mi->rli != NULL)
       {
-        *appliers_id= (unsigned long*) my_malloc(PSI_NOT_INSTRUMENTED,
-                                                 sizeof(unsigned long),
-                                                 MYF(MY_WME));
-        mysql_mutex_lock(&mi->rli->info_thd_lock);
-        **appliers_id= mi->rli->info_thd->thread_id();
-        mysql_mutex_unlock(&mi->rli->info_thd_lock);
-        number_appliers= 1;
+        mysql_mutex_lock(&mi->rli->run_lock);
+
+        int num_workers= mi->rli->slave_parallel_workers;
+        if (num_workers > 1)
+        {
+          *thread_id=
+              (unsigned long*) my_malloc(PSI_NOT_INSTRUMENTED,
+                                         num_workers * sizeof(unsigned long),
+                                         MYF(MY_WME));
+          unsigned long *appliers_id_pointer= *thread_id;
+
+          for (int i= 0; i < num_workers; i++, appliers_id_pointer++)
+          {
+            mysql_mutex_lock(&mi->rli->workers.at(i)->info_thd_lock);
+            *appliers_id_pointer= mi->rli->workers.at(i)->info_thd->thread_id();
+            mysql_mutex_unlock(&mi->rli->workers.at(i)->info_thd_lock);
+          }
+
+          number_threads= num_workers;
+        }
+        else
+        {
+          if (mi->rli->info_thd != NULL)
+          {
+            *thread_id= (unsigned long*) my_malloc(PSI_NOT_INSTRUMENTED,
+                                                     sizeof(unsigned long),
+                                                     MYF(MY_WME));
+            mysql_mutex_lock(&mi->rli->info_thd_lock);
+            **thread_id= mi->rli->info_thd->thread_id();
+            mysql_mutex_unlock(&mi->rli->info_thd_lock);
+            number_threads= 1;
+          }
+        }
+        mysql_mutex_unlock(&mi->rli->run_lock);
       }
-    }
-    mysql_mutex_unlock(&mi->rli->run_lock);
+      break;
+    default:
+      DBUG_RETURN(number_threads);
   }
 
-  DBUG_RETURN(number_appliers);
+  DBUG_RETURN(number_threads);
 }
 
 long long channel_get_last_delivered_gno(const char* channel, int sidno)

sql/rpl_channel_service_interface.h

@@ -216,18 +216,20 @@ int channel_purge_queue(const char* channel, bool reset_all);
 bool channel_is_active(const char* channel, enum_channel_thread_types type);
 
 /**
-  Returns the ids of the channel appliers.
-  If more than one applier exists, a channel is returned
+  Returns the id(s) of the channel threads: receiver or applier.
+  If more than one applier exists, an array is returned
 
   @param[in]  channel      The channel name
-  @param[out] appliers_id  The array of id(s)
+  @param[in]  thread_type  The thread type (receiver or applier)
+  @param[out] thread_id    The array of id(s)
 
   @return the number of returned ids
-    @retval <=0  the channel does no exists, or the applier is not present
-    @retval >0 the number of applier ids returned.
+    @retval -1  the channel does no exists, or the thread is not present
+    @retval >0 the number of thread ids returned.
 */
-int channel_get_appliers_thread_id(const char* channel,
-                                   unsigned long** appliers_id);
+int channel_get_thread_id(const char* channel,
+                          enum_channel_thread_types thread_type,
+                          unsigned long** thread_id);
 
 /**
   Returns last GNO from applier from a given UUID.