cleanup: fix and generalize handle_manager thread

* provide an argument to the callback
* don't ignore a callback request if it's already present in the queue
* initialize mutex/cond/in_use flag before starting the thread,
  in case the first callback queueing request arrives before
  handle_manager had time to initialize
* set/check abort_manager under a mutex, otherwise handle_manager
  thread might destroy LOCK_manager before stop_handle_manager
  released it
* signal COND on queueing a callback, stop cond_wait on callback request
* always start the thread, even if flush_time is 0
* but keep the old behavior in embedded (no replication, no galera)
* style cleanups (e.g. remove volatile for a variable protected by a mutex)

libmysqld/lib_sql.cc

@@ -623,6 +623,7 @@ int init_embedded_server(int argc, char **argv, char **groups)
 
   (void) thr_setconcurrency(concurrency);	// 10 by default
 
+  if (flush_time && flush_time != ~(ulong) 0L)
     start_handle_manager();
 
   // FIXME initialize binlog_filter and rpl_filter if not already done

sql/sql_manager.cc

@@ -26,8 +26,8 @@
 #include "sql_manager.h"
 #include "sql_base.h"                           // flush_tables
 
-static bool volatile manager_thread_in_use;
-static bool abort_manager;
+static bool volatile manager_thread_in_use = 0;
+static bool abort_manager = false;
 
 pthread_t manager_thread;
 mysql_mutex_t LOCK_manager;
@@ -35,22 +35,21 @@ mysql_cond_t COND_manager;
 
 struct handler_cb {
    struct handler_cb *next;
-   void (*action)(void);
+   void (*action)(void *);
+   void *data;
 };
 
-static struct handler_cb * volatile cb_list;
+static struct handler_cb *cb_list; // protected by LOCK_manager
 
-bool mysql_manager_submit(void (*action)())
+bool mysql_manager_submit(void (*action)(void *), void *data)
 {
   bool result= FALSE;
   DBUG_ASSERT(manager_thread_in_use);
-  struct handler_cb * volatile *cb;
+  struct handler_cb **cb;
   mysql_mutex_lock(&LOCK_manager);
   cb= &cb_list;
-  while (*cb && (*cb)->action != action)
+  while (*cb)
     cb= &(*cb)->next;
-  if (!*cb)
-  {
   *cb= (struct handler_cb *)my_malloc(sizeof(struct handler_cb), MYF(MY_WME));
   if (!*cb)
     result= TRUE;
@@ -58,8 +57,9 @@ bool mysql_manager_submit(void (*action)())
   {
     (*cb)->next= NULL;
     (*cb)->action= action;
+    (*cb)->data= data;
   }
-  }
+  mysql_cond_signal(&COND_manager);
   mysql_mutex_unlock(&LOCK_manager);
   return result;
 }
@@ -69,18 +69,14 @@ pthread_handler_t handle_manager(void *arg __attribute__((unused)))
   int error = 0;
   struct timespec abstime;
   bool reset_flush_time = TRUE;
-  struct handler_cb *cb= NULL;
   my_thread_init();
   DBUG_ENTER("handle_manager");
 
   pthread_detach_this_thread();
   manager_thread = pthread_self();
-  mysql_cond_init(key_COND_manager, &COND_manager,NULL);
-  mysql_mutex_init(key_LOCK_manager, &LOCK_manager, NULL);
-  manager_thread_in_use = 1;
-  for (;;)
-  {
   mysql_mutex_lock(&LOCK_manager);
+  while (!abort_manager)
+  {
     /* XXX: This will need to be made more general to handle different
      * polling needs. */
     if (flush_time)
@@ -90,23 +86,8 @@ pthread_handler_t handle_manager(void *arg __attribute__((unused)))
 	set_timespec(abstime, flush_time);
         reset_flush_time = FALSE;
       }
-      while ((!error || error == EINTR) && !abort_manager)
+      while ((!error || error == EINTR) && !abort_manager && !cb_list)
         error= mysql_cond_timedwait(&COND_manager, &LOCK_manager, &abstime);
-    }
-    else
-    {
-      while ((!error || error == EINTR) && !abort_manager)
-        error= mysql_cond_wait(&COND_manager, &LOCK_manager);
-    }
-    if (cb == NULL)
-    {
-      cb= cb_list;
-      cb_list= NULL;
-    }
-    mysql_mutex_unlock(&LOCK_manager);
-
-    if (abort_manager)
-      break;
 
       if (error == ETIMEDOUT || error == ETIME)
       {
@@ -114,16 +95,28 @@ pthread_handler_t handle_manager(void *arg __attribute__((unused)))
         error = 0;
         reset_flush_time = TRUE;
       }
+    }
+    else
+    {
+      while ((!error || error == EINTR) && !abort_manager && !cb_list)
+        error= mysql_cond_wait(&COND_manager, &LOCK_manager);
+    }
+
+    struct handler_cb *cb= cb_list;
+    cb_list= NULL;
+    mysql_mutex_unlock(&LOCK_manager);
 
     while (cb)
     {
       struct handler_cb *next= cb->next;
-      cb->action();
+      cb->action(cb->data);
       my_free(cb);
       cb= next;
     }
+    mysql_mutex_lock(&LOCK_manager);
   }
   manager_thread_in_use = 0;
+  mysql_mutex_unlock(&LOCK_manager);
   mysql_mutex_destroy(&LOCK_manager);
   mysql_cond_destroy(&COND_manager);
   DBUG_LEAVE; // Can't use DBUG_RETURN after my_thread_end
@@ -137,15 +130,15 @@ void start_handle_manager()
 {
   DBUG_ENTER("start_handle_manager");
   abort_manager = false;
-  if (flush_time && flush_time != ~(ulong) 0L)
   {
     pthread_t hThread;
-    int error;
-    if ((error= mysql_thread_create(key_thread_handle_manager,
-                                    &hThread, &connection_attrib,
-                                    handle_manager, 0)))
-      sql_print_warning("Can't create handle_manager thread (errno= %d)",
-                        error);
+    int err;
+    manager_thread_in_use = 1;
+    mysql_cond_init(key_COND_manager, &COND_manager,NULL);
+    mysql_mutex_init(key_LOCK_manager, &LOCK_manager, NULL);
+    if ((err= mysql_thread_create(key_thread_handle_manager, &hThread,
+                                  &connection_attrib, handle_manager, 0)))
+      sql_print_warning("Can't create handle_manager thread (errno: %M)", err);
   }
   DBUG_VOID_RETURN;
 }
@@ -155,10 +148,10 @@ void start_handle_manager()
 void stop_handle_manager()
 {
   DBUG_ENTER("stop_handle_manager");
-  abort_manager = true;
   if (manager_thread_in_use)
   {
     mysql_mutex_lock(&LOCK_manager);
+    abort_manager = true;
     DBUG_PRINT("quit", ("initiate shutdown of handle manager thread: %lu",
                         (ulong)manager_thread));
     mysql_cond_signal(&COND_manager);

sql/sql_manager.h

@@ -18,6 +18,6 @@
 
 void start_handle_manager();
 void stop_handle_manager();
-bool mysql_manager_submit(void (*action)());
+bool mysql_manager_submit(void (*action)(void *), void *data);
 
 #endif /* SQL_MANAGER_INCLUDED */