Fix data races in connection lifecycle code (TSAN findings, #460)

libgearman-server/connection.cc

@@ -427,11 +427,9 @@ gearman_server_con_st * gearman_server_con_to_be_freed_next(gearman_server_threa
 {
   gearman_server_con_st *con;
 
-  if (thread->to_be_freed_list == NULL)
-  {
-    return NULL;
-  }
-
+  /* thread->to_be_freed_list is mutated under thread->lock by other threads
+     (e.g. gearman_server_con_to_be_freed_add()), so it must not be read
+     before taking the lock below. */
   int lock_error;
   if ((lock_error= pthread_mutex_lock(&thread->lock)) == 0)
   {

libgearman-server/struct/io.h

@@ -41,6 +41,8 @@
 
 #include "libgearman/ssl.h"
 
+#include <atomic>
+
 struct gearmand_io_st
 {
   struct {
@@ -146,14 +148,19 @@ struct gearman_server_con_st
   gearmand_io_st con;
   bool is_sleeping{};
   bool is_exceptions{};
-  bool is_dead{};
+  // These flags are used to hand connection ownership off between the IO,
+  // proc, and main threads (e.g. gearman_server_con_attempt_free() sets
+  // is_dead from an IO thread while _proc() reads it from the proc thread
+  // with no other synchronization), so they must be atomic rather than
+  // plain bool.
+  std::atomic<bool> is_dead{false};
   bool is_noop_sent{};
   bool is_cleaned_up{};
   gearmand_error_t ret{};
-  bool io_list{};
-  bool proc_list{};
-  bool proc_removed{};
-  bool to_be_freed_list{};
+  std::atomic<bool> io_list{false};
+  std::atomic<bool> proc_list{false};
+  std::atomic<bool> proc_removed{false};
+  std::atomic<bool> to_be_freed_list{false};
   uint32_t io_packet_count{};
   uint32_t proc_packet_count{};
   uint32_t worker_count{};

libgearman-server/thread.cc

@@ -189,7 +189,7 @@ gearman_server_thread_run(gearman_server_thread_st *thread,
         gearman_server_con_free(server_con);
       else
         gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM, "con %p isn't dead %d or proc removed %d, but is in to_be_freed_list",
-                           server_con, server_con->is_dead, server_con->proc_removed);
+                           server_con, bool(server_con->is_dead), bool(server_con->proc_removed));
     }
 
     while ((server_con= gearman_server_con_io_next(thread)) != NULL)
@@ -344,18 +344,45 @@ static gearmand_error_t _thread_packet_flush(gearman_server_con_st *con)
     return GEARMAND_IO_WAIT;
   }
 
-  while (con->io_packet_list)
+  while (1)
   {
-    gearmand_error_t ret= gearman_io_send(con, &(con->io_packet_list->packet),
-                                          con->io_packet_list->next == NULL ? true : false);
+    /* con->io_packet_list is appended to under con->thread->lock by other
+       threads (e.g. gearman_server_io_packet_add() from the proc thread), so
+       the head pointer and its ->next must be read under the same lock
+       rather than dereferenced directly here. */
+    gearman_server_packet_st *packet;
+    bool is_last;
+
+    int lock_error;
+    if ((lock_error= pthread_mutex_lock(&con->thread->lock)) == 0)
+    {
+      packet= con->io_packet_list;
+      is_last= (packet != NULL) and (packet->next == NULL);
+      if ((lock_error= pthread_mutex_unlock(&con->thread->lock)))
+      {
+        gearmand_log_fatal_perror(GEARMAN_DEFAULT_LOG_PARAM, lock_error, "pthread_mutex_unlock");
+      }
+    }
+    else
+    {
+      gearmand_log_fatal_perror(GEARMAN_DEFAULT_LOG_PARAM, lock_error, "pthread_mutex_lock");
+      return GEARMAND_ERRNO;
+    }
+
+    if (packet == NULL)
+    {
+      break;
+    }
+
+    gearmand_error_t ret= gearman_io_send(con, &(packet->packet), is_last);
     if (gearmand_failed(ret))
     {
       return ret;
     }
 
-    gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, 
+    gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
                        "Sent %s",
-                       gearman_strcommand(con->io_packet_list->packet.command));
+                       gearman_strcommand(packet->packet.command));
 
     gearman_server_io_packet_remove(con);
   }