Bug#36778475 Issues with I_S.processlist due to missing synchronization

The protocol of a non-current THD was accessed fron the code path of
I_S processlist. This caused a race condition to occur which resulted
in invalid memory being accessed. The fix is to cache information being
accessed and to make sure protocol objects of non-current THDs are not
referenced from within the I_S processlist implementation.

Change-Id: I0ce332854f0bc63879da68702846c0ca52787ca3

Author: ssorumgard

sql-common/net_serv.cc

@@ -81,7 +81,6 @@ using std::min;
 */
 
 #ifdef MYSQL_SERVER
-
 /*
   The following variables/functions should really not be declared
   extern, but as it's hard to include sql_class.h here, we have to
@@ -92,6 +91,13 @@ extern void thd_increment_bytes_received(size_t length);
 
 /* Additional instrumentation hooks for the server */
 #include "mysql_com_server.h"
+
+/* Refresh cached values in the THD for the server. */
+static void server_store_cached_values(const NET *net) {
+  // Refresh only if this net is the net of the current_thd
+  if (current_thd != nullptr && current_thd->get_net() == net)
+    current_thd->store_cached_properties(THD::cached_properties::RW_STATUS);
+}
 #endif
 
 static bool net_write_buff(NET *, const uchar *, size_t);
@@ -167,6 +173,7 @@ bool my_net_init(NET *net, Vio *vio) {
   net->last_errno = 0;
 #ifdef MYSQL_SERVER
   net->extension = nullptr;
+  server_store_cached_values(net);
 #else
   NET_EXTENSION *ext = net_extension_init();
   ext->net_async_context->cur_pos = net->buff + net->where_b;
@@ -1303,6 +1310,9 @@ bool net_write_packet(NET *net, const uchar *packet, size_t length) {
     return true;
 
   net->reading_or_writing = 2;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
 
   const bool do_compress = net->compress;
   if (do_compress) {
@@ -1311,6 +1321,9 @@ bool net_write_packet(NET *net, const uchar *packet, size_t length) {
       net->last_errno = ER_OUT_OF_RESOURCES;
       /* In the server, allocation failure raises a error. */
       net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+      server_store_cached_values(net);
+#endif
       return true;
     }
   }
@@ -1324,6 +1337,9 @@ bool net_write_packet(NET *net, const uchar *packet, size_t length) {
   if (do_compress) my_free(const_cast<uchar *>(packet));
 
   net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
 
   /* Socket can't be used any more */
   if (net->error == NET_ERROR_SOCKET_NOT_READABLE) {
@@ -1712,6 +1728,9 @@ static net_async_status net_read_packet_nonblocking(NET *net, ulong *ret) {
     case NET_ASYNC_PACKET_READ_IDLE:
       net_async->async_packet_read_state = NET_ASYNC_PACKET_READ_HEADER;
       net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+      server_store_cached_values(net);
+#endif
       [[fallthrough]];
     case NET_ASYNC_PACKET_READ_HEADER:
       /*
@@ -1788,6 +1807,9 @@ static net_async_status net_read_packet_nonblocking(NET *net, ulong *ret) {
 
   net->read_pos[*ret] = 0;
   net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
   if (net->compress) {
     mysql_compress_context *mysql_compress_ctx = compress_context(net);
     if (my_uncompress(mysql_compress_ctx, net->buff + net->where_b,
@@ -1808,6 +1830,9 @@ static net_async_status net_read_packet_nonblocking(NET *net, ulong *ret) {
 error:
   *ret = packet_error;
   net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
   return NET_ASYNC_COMPLETE;
 }
 
@@ -2093,6 +2118,9 @@ static size_t net_read_packet(NET *net, size_t *complen) {
   *complen = 0;
 
   net->reading_or_writing = 1;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
 
   /*
     We should reset compress_packet_nr even before reading the header because
@@ -2140,12 +2168,18 @@ static size_t net_read_packet(NET *net, size_t *complen) {
     net->error = NET_ERROR_SOCKET_UNUSABLE;
   DBUG_DUMP("net read", net->buff + net->where_b, pkt_len);
   net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
   return pkt_len;
 
 error:
   if (net->error == NET_ERROR_SOCKET_NOT_WRITABLE)
     net->error = NET_ERROR_SOCKET_UNUSABLE;
   net->reading_or_writing = 0;
+#ifdef MYSQL_SERVER
+  server_store_cached_values(net);
+#endif
   return packet_error;
 }
 

sql/protocol_classic.cc

@@ -1385,6 +1385,7 @@ uchar Protocol_classic::get_error() { return m_thd->net.error; }
 
 void Protocol_classic::wipe_net() {
   memset(&m_thd->net, 0, sizeof(m_thd->net));
+  m_thd->store_cached_properties(THD::cached_properties::RW_STATUS);
 }
 
 void Protocol_classic::set_max_packet_size(ulong max_packet_size) {

sql/sql_class.cc

@@ -608,6 +608,7 @@ void THD::enter_stage(const PSI_stage_info *new_stage,
 
     m_current_stage_key = new_stage->m_key;
     set_proc_info(msg);
+    store_cached_properties(cached_properties::RW_STATUS);
 
     m_stage_progress_psi =
         MYSQL_SET_STAGE(m_current_stage_key, calling_file, calling_line);
@@ -870,7 +871,7 @@ THD::THD(bool enable_plugins)
   protocol_text->init(this);
   protocol_binary->init(this);
   protocol_text->set_client_capabilities(0);  // minimalistic client
-  m_cached_is_connection_alive.store(m_protocol->connection_alive());
+  store_cached_properties();
 
   /*
     Make sure thr_lock_info_init() is called for threads which do not get
@@ -907,6 +908,22 @@ THD::THD(bool enable_plugins)
   }
 }
 
+void THD::store_cached_properties(cached_properties prop_mask) {
+  DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                  { assert(current_thd == this); });
+
+  auto is_selected = [this, prop_mask](cached_properties property) -> bool {
+    return (this->m_protocol != nullptr &&
+            static_cast<int>(prop_mask) & static_cast<int>(property));
+  };
+
+  if (is_selected(cached_properties::IS_ALIVE))
+    m_cached_is_connection_alive.store(m_protocol->connection_alive());
+
+  if (is_selected(cached_properties::RW_STATUS))
+    m_cached_rw_status.store(m_protocol->get_rw_status());
+}
+
 void THD::copy_table_access_properties(THD *thd) {
   thread_stack = thd->thread_stack;
   variables.option_bits = thd->variables.option_bits & OPTION_BIN_LOG;
@@ -1427,7 +1444,7 @@ void THD::release_resources() {
   if (is_classic_protocol() && get_protocol_classic()->get_vio()) {
     vio_delete(get_protocol_classic()->get_vio());
     get_protocol_classic()->end_net();
-    m_cached_is_connection_alive.store(m_protocol->connection_alive());
+    store_cached_properties();
   }
 
   /* modification plan for UPDATE/DELETE should be freed. */
@@ -1732,6 +1749,8 @@ void THD::disconnect(bool server_shutdown) {
     /* Disconnect even if a active vio is not associated. */
     if (is_classic_protocol() && get_protocol_classic()->get_vio() != vio &&
         get_protocol_classic()->connection_alive()) {
+      DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                      { assert(current_thd == this); });
       m_protocol->shutdown(server_shutdown);
     }
   }
@@ -2341,6 +2360,8 @@ void THD::reset_sub_statement_state(Sub_statement_state *backup,
   backup->examined_row_count = m_examined_row_count;
   backup->sent_row_count = m_sent_row_count;
   backup->num_truncated_fields = num_truncated_fields;
+  DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                  { assert(current_thd == this); });
   backup->client_capabilities = m_protocol->get_client_capabilities();
   backup->savepoints = get_transaction()->m_savepoints;
   backup->first_successful_insert_id_in_prev_stmt =
@@ -2913,6 +2934,8 @@ bool THD::send_result_metadata(const mem_root_deque<Item *> &list, uint flags) {
   DBUG_TRACE;
   uchar buff[MAX_FIELD_WIDTH];
   String tmp((char *)buff, sizeof(buff), &my_charset_bin);
+  DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                  { assert(current_thd == this); });
 
   if (m_protocol->start_result_metadata(CountVisibleFields(list), flags,
                                         variables.character_set_results))
@@ -2953,6 +2976,8 @@ bool THD::send_result_set_row(const mem_root_deque<Item *> &row_items) {
   String str_buffer(buffer, sizeof(buffer), &my_charset_bin);
 
   DBUG_TRACE;
+  DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                  { assert(current_thd == this); });
 
   for (Item *item : VisibleFields(row_items)) {
     if (item->send(m_protocol, &str_buffer) || is_error()) return true;
@@ -2970,6 +2995,8 @@ void THD::send_statement_status() {
   assert(!get_stmt_da()->is_sent());
   bool error = false;
   Diagnostics_area *da = get_stmt_da();
+  DBUG_EXECUTE_IF("assert_only_current_thd_protocol_access",
+                  { assert(current_thd == this); });
 
   /* Can not be true, but do not take chances in production. */
   if (da->is_sent()) return;
@@ -3270,14 +3297,16 @@ bool THD::is_connected(bool use_cached_connection_alive) {
   return get_protocol()->connection_alive();
 }
 
+uint THD::get_protocol_rw_status() { return m_cached_rw_status.load(); }
+
 Protocol *THD::get_protocol() {
-  m_cached_is_connection_alive.store(m_protocol->connection_alive());
+  store_cached_properties();
   return m_protocol;
 }
 
 Protocol_classic *THD::get_protocol_classic() {
   assert(is_classic_protocol());
-  m_cached_is_connection_alive.store(m_protocol->connection_alive());
+  store_cached_properties();
   return pointer_cast<Protocol_classic *>(m_protocol);
 }
 
@@ -3286,14 +3315,14 @@ void THD::push_protocol(Protocol *protocol) {
   assert(protocol != nullptr);
   m_protocol->push_protocol(protocol);
   m_protocol = protocol;
-  m_cached_is_connection_alive.store(m_protocol->connection_alive());
+  store_cached_properties();
 }
 
 void THD::pop_protocol() {
   assert(m_protocol != nullptr);
   m_protocol = m_protocol->pop_protocol();
   assert(m_protocol != nullptr);
-  m_cached_is_connection_alive.store(m_protocol->connection_alive());
+  store_cached_properties();
 }
 
 void THD::set_time() {

sql/sql_class.h

@@ -1074,6 +1074,17 @@ class THD : public MDL_context_owner,
   handlerton *m_eligible_secondary_engine_handlerton;
 
  public:
+  /* Store a thread safe copy of protocol properties. */
+  enum class cached_properties : int {
+    NONE = 0,         // No properties
+    IS_ALIVE = 1,     // protocol->is_connection_alive()
+    RW_STATUS = 2,    // protocol->get_rw_status()
+    LAST = 4,         // Next unused power of 2.
+    ALL = (LAST - 1)  // Mask selecting all properties.
+  };
+  void store_cached_properties(
+      cached_properties prop_mask = cached_properties::ALL);
+
   /* Used to execute base64 coded binlog events in MySQL server */
   Relay_log_info *rli_fake;
   /* Slave applier execution context */
@@ -1299,12 +1310,13 @@ class THD : public MDL_context_owner,
   mysql_mutex_t LOCK_query_plan;
 
   /**
-    Keep a cached value saying whether the connection is alive. Update when
+    Keep cached values of "connection alive" and "rw status". Update when
     pushing, popping or getting the protocol. Used by
     information_schema.processlist to avoid locking mutexes that might
     affect performance.
   */
   std::atomic<bool> m_cached_is_connection_alive;
+  std::atomic<uint> m_cached_rw_status;
 
  public:
   /// Locks the query plan of this THD
@@ -3291,6 +3303,9 @@ class THD : public MDL_context_owner,
   /** Return false if connection to client is broken. */
   bool is_connected(bool use_cached_connection_alive = false) final;
 
+  /** Return the cached protocol rw status. */
+  uint get_protocol_rw_status();
+
   /**
     Mark the current error as fatal. Warning: this does not
     set any error, it sets a property of the error, so must be

sql/sql_show.cc

@@ -2791,8 +2791,8 @@ class thread_info_compare {
 
 static const char *thread_state_info(THD *invoking_thd, THD *inspected_thd) {
   DBUG_TRACE;
-  if (inspected_thd->get_protocol()->get_rw_status()) {
-    if (inspected_thd->get_protocol()->get_rw_status() == 2)
+  if (inspected_thd->get_protocol_rw_status()) {
+    if (inspected_thd->get_protocol_rw_status() == 2)
       return "Sending to client";
     if (inspected_thd->get_command() == COM_SLEEP) return "";
     return "Receiving from client";

sql/sql_thd_api.cc

@@ -256,6 +256,7 @@ uint thd_get_net_read_write(THD *thd) {
 
 void thd_set_net_read_write(THD *thd, uint val) {
   thd->get_protocol_classic()->get_net()->reading_or_writing = val;
+  thd->store_cached_properties(THD::cached_properties::RW_STATUS);
 }
 
 /**