Bug #33482064 rpl.rpl_semi_sync_turn_on_off_optimize_for_static_plugin_config fails on asan

Problem
-------

Ack Receiver thread is reading empty pointer which is pointed out by ASAN
(heap-use-after-free).

Analysis / Root-cause analysis
------------------------------

Problem happens when semi-synchronous replication is turned on and the replica
is disconnecting. These conditions may cause Ack Receiver to encounter error
when reading communication packets (ER_NET_READ_ERROR).
It may happen that Binary log dump thread resources are released before Ack
Receiver will remove connection information while AckReceiver wrongfully
iterates over it. Let's consider the following flow:
1. Removing of thread X is requested from thread S. S changes `is_leaving` to
true and waits for the confirmation.
2. Function `init_replica_sockets` is called from the Ack Receiver thread A,
therefore X is allowed to leave. A changes `is_leaving` to false.
3. A is listening on sockets, possibly `listen_on_sockets` returns some error,
but not critical, or just A processes the data from sockets quickly.
4. Other threads are requested to leave - `m_slaves_changed` is set to true
5. A is woken up again (S is still waiting for its turn), and calls
`init_replica_sockets again`. Listener checks the status of X (still not
removed from the Ack Receiver container), it is false, therefore X is
copied into the Listener internal container.
6. S is woken up. It sees that status of the X thread is false, therefore it
was allowed to leave. S thinks that X was removed from the listener container,
therefore it is removing X from the Ack Receiver container.
7. A is woken up to process data from sockets. In the meantime, replica shuts
down and S resources are released. A gots ER_NET_READ_ERROR and checks the
while condition (net.vio->has_data(net.vio)).
Vio pointer is NULL at this point. A SIGSEGV signal is issued.

Implemented solution
--------------------

'Slave' 'is_leaving' internal variable type is changed to newly introduced
enumeration class 'enum_status'. 'enum_status' class contains three
states: up, leaving and down. Leaving thread is changing status to
'leaving'. Ack Receiver notices that 'status' value of the leaving
thread changed to 'leaving' and changes the status to 'down'. If
'Slave' object state is 'down' or 'leaving', Ack Receiver thread is
ommiting the 'Slave' object while creating the 'Listener' vector.

Signed-off-by: Karolina Szczepankiewicz <karolina.szczepankiewicz@oracle.com>
Change-Id: Ic59191479473c86d4b97da060457930a52e79f0a

Author: Karolina Szczepankiewicz

plugin/semisync/semisync_source_ack_receiver.cc

@@ -151,7 +151,6 @@ bool Ack_receiver::add_slave(THD *thd) {
           &slave.compress_ctx, algorithm,
           thd->get_protocol_classic()->get_compression_level());
   }
-  slave.is_leaving = false;
   slave.vio = thd->get_protocol_classic()->get_vio();
   slave.vio->mysql_socket.m_psi = nullptr;
 
@@ -186,7 +185,7 @@ void Ack_receiver::remove_slave(THD *thd) {
   */
   for (it = m_slaves.begin(); it != m_slaves.end(); ++it) {
     if (it->thread_id == thd->thread_id()) {
-      it->is_leaving = true;
+      it->m_status = Slave::EnumStatus::leaving;
       m_slaves_changed = true;
       break;
     }
@@ -196,7 +195,8 @@ void Ack_receiver::remove_slave(THD *thd) {
     Wait till Ack_receiver::run() is done reading from the
     slave's socket.
   */
-  while ((it != m_slaves.end()) && (it->is_leaving) && (m_status == ST_UP)) {
+  while ((it != m_slaves.end()) &&
+         (it->m_status == Slave::EnumStatus::leaving) && (m_status == ST_UP)) {
     mysql_cond_wait(&m_cond, &m_mutex);
     /*
       In above cond_wait, we release and reacquire m_mutex.
@@ -311,8 +311,9 @@ void Ack_receiver::run() {
           if (likely(len != packet_error))
             repl_semisync->reportReplyPacket(slave_obj.server_id, net.read_pos,
                                              len);
-          else if (net.last_errno == ER_NET_READ_ERROR)
+          else if (net.last_errno == ER_NET_READ_ERROR) {
             listener.clear_socket_info(i);
+          }
         } while (net.vio->has_data(net.vio) && m_status == ST_UP);
       }
       i++;

plugin/semisync/semisync_source_ack_receiver.h

@@ -34,11 +34,12 @@
 #include "sql/sql_class.h"
 
 struct Slave {
+  enum class EnumStatus { up, leaving, down };
   uint32_t thread_id;
   Vio *vio;
   uint server_id;
   mysql_compress_context compress_ctx;
-  bool is_leaving;
+  EnumStatus m_status = EnumStatus::up;
 
   my_socket sock_fd() const { return vio->mysql_socket.fd; }
 };

plugin/semisync/semisync_source_socket_listener.h

@@ -51,8 +51,8 @@ class Poll_socket_listener {
         Do not consider the slave's socket
         if the slave is in the process of leaving.
       */
-      if (slaves[i].is_leaving) {
-        slaves[i].is_leaving = false;
+      if (slaves[i].m_status != Slave::EnumStatus::up) {
+        slaves[i].m_status = Slave::EnumStatus::down;
         continue;
       }
       pollfd poll_fd;
@@ -105,8 +105,8 @@ class Select_socket_listener {
         Do not consider the slave's socket
         if the slave is in the process of leaving.
       */
-      if (slaves[i].is_leaving) {
-        slaves[i].is_leaving = false;
+      if (slaves[i].m_status != Slave::EnumStatus::up) {
+        slaves[i].m_status = Slave::EnumStatus::down;
         continue;
       }
       my_socket socket_id = slaves[i].sock_fd();