Bug#22894429 Unable to uninstall mysqlx plugin

Issue: Unable to uninstall mysqlx plugin when the limit of
available threads has been achived.

Reviewed-by: Lukasz Kotula <[email protected]>
Reviewed-by: Andrzej Religa <[email protected]>
RB: 12072
(cherry picked from commit 41f85a0c5478e327f37cafa63ef1bb195d218671)

Author: Grzegorz Szwarc

mysql-test/suite/xplugin/r/performance_schema.result

@@ -2,8 +2,8 @@ install plugin mysqlx soname "mysqlx.so";
 call mtr.add_suppression("Plugin mysqlx reported: .Failed at SSL configuration: .SSL context is not usable without certificate and private key..");
 call mtr.add_suppression("Plugin mysqlx reported: .SSL_CTX_load_verify_locations failed.");
 
-6 rows affected
-Rows matched: 6  Changed: 0  Warnings: 0
+7 rows affected
+Rows matched: 7  Changed: 0  Warnings: 0
 
 1 rows affected
 Rows matched: 1  Changed: 1  Warnings: 0
@@ -25,6 +25,7 @@ thread/mysqlx/acceptor_network
 0 rows affected
 name
 wait/synch/mutex/mysqlx/lock_list_access
+wait/synch/mutex/mysqlx/scheduler_dynamic_post
 wait/synch/mutex/mysqlx/scheduler_dynamic_task_pending
 wait/synch/mutex/mysqlx/scheduler_dynamic_thread_exit
 0 rows affected
@@ -39,6 +40,7 @@ NAME	ENABLED	TIMED
 wait/synch/mutex/mysqlx/lock_list_access	NO	NO
 wait/synch/mutex/mysqlx/scheduler_dynamic_task_pending	NO	NO
 wait/synch/mutex/mysqlx/scheduler_dynamic_thread_exit	YES	NO
+wait/synch/mutex/mysqlx/scheduler_dynamic_post	YES	NO
 wait/synch/rwlock/mysqlx/client_list_clients	YES	NO
 wait/synch/cond/mysqlx/scheduler_dynamic_task_pending	NO	NO
 wait/synch/cond/mysqlx/scheduler_dynamic_thread_exit	YES	NO
@@ -171,6 +173,7 @@ thread/mysqlx/acceptor_network
 command ok
 name
 wait/synch/mutex/mysqlx/lock_list_access
+wait/synch/mutex/mysqlx/scheduler_dynamic_post
 wait/synch/mutex/mysqlx/scheduler_dynamic_task_pending
 wait/synch/mutex/mysqlx/scheduler_dynamic_thread_exit
 command ok

rapid/plugin/x/ngs/include/ngs/scheduler.h

@@ -52,7 +52,7 @@ namespace ngs
     virtual ~Scheduler_dynamic();
 
     virtual void launch();
-    virtual void set_num_workers(unsigned int n);
+    virtual unsigned int set_num_workers(unsigned int n);
     virtual void stop();
     void set_idle_worker_timeout(unsigned long long milliseconds);
     bool post(Task* task);
@@ -65,6 +65,7 @@ namespace ngs
     void set_monitor(Monitor *monitor);
 
     bool is_worker_thread(my_thread_t thread_id);
+    bool is_running();
 
   private:
     template<typename Element_type>
@@ -113,7 +114,7 @@ namespace ngs
     void* worker();
 
     void create_thread();
-    bool is_running();
+    void create_min_num_workers();
 
     int32 increase_workers_count();
     int32 decrease_workers_count();
@@ -125,6 +126,7 @@ namespace ngs
     Cond m_task_pending_cond;
     Mutex m_thread_exit_mutex;
     Cond m_thread_exit_cond;
+    Mutex m_post_mutex;
     volatile int32 m_is_running;
     volatile int32 m_min_workers_count;
     volatile int32 m_workers_count;

rapid/plugin/x/ngs/include/ngs/thread.h

@@ -46,9 +46,9 @@ namespace ngs
   typedef my_start_routine Start_routine_t;
 #endif
 
-  int thread_create(PSI_thread_key key, Thread_t *thread,
-                    const Thread_attr_t *attr,
-                    Start_routine_t func, void *arg);
+  void thread_create(PSI_thread_key key, Thread_t *thread,
+                     const Thread_attr_t *attr,
+                     Start_routine_t func, void *arg);
   int thread_join(Thread_t *thread, void *   *ret);
 
   class Mutex

rapid/plugin/x/ngs/src/scheduler.cc

@@ -45,6 +45,7 @@ Scheduler_dynamic::Scheduler_dynamic(const char* name)
   m_task_pending_cond(KEY_cond_x_scheduler_dynamic_task_pending),
   m_thread_exit_mutex(KEY_mutex_x_scheduler_dynamic_thread_exit),
   m_thread_exit_cond(KEY_cond_x_scheduler_dynamic_thread_exit),
+  m_post_mutex(KEY_mutex_x_scheduler_dynamic_post),
   m_is_running(0),
   m_min_workers_count(1),
   m_workers_count(0),
@@ -65,27 +66,42 @@ void Scheduler_dynamic::launch()
   int32 int_0 = 0;
   if (my_atomic_cas32(&m_is_running, &int_0, 1))
   {
-    set_num_workers(my_atomic_load32(&m_min_workers_count));
+    create_min_num_workers();
     log_info("Scheduler \"%s\" started.", m_name.c_str());
   }
 }
 
 
-void Scheduler_dynamic::set_num_workers(unsigned int n)
+void Scheduler_dynamic::create_min_num_workers()
 {
-  my_atomic_store32(&m_min_workers_count, n);
-
   while (is_running() &&
-         my_atomic_load32(&m_workers_count) <
-           my_atomic_load32(&m_min_workers_count))
+         my_atomic_load32(&m_workers_count) < my_atomic_load32(&m_min_workers_count))
   {
     create_thread();
   }
 }
 
 
-void Scheduler_dynamic::set_idle_worker_timeout(
-  unsigned long long milliseconds)
+unsigned int Scheduler_dynamic::set_num_workers(unsigned int n)
+{
+  my_atomic_store32(&m_min_workers_count, n);
+  try
+  {
+    create_min_num_workers();
+  }
+  catch (std::exception &e)
+  {
+    log_debug("Exception in set minimal number of workers \"%s\"", e.what());
+    const int32 m = my_atomic_load32(&m_workers_count);
+    log_warning("Unable to set minimal number of workers to %u; actual value is %i", n, m);
+    my_atomic_store32(&m_min_workers_count, m);
+    return m;
+  }
+  return n;
+}
+
+
+void Scheduler_dynamic::set_idle_worker_timeout(unsigned long long milliseconds)
 {
   my_atomic_store64(&m_idle_worker_timeout, milliseconds);
   m_task_pending_cond.broadcast(m_task_pending_mutex);
@@ -131,45 +147,53 @@ bool Scheduler_dynamic::post(Task* task)
   if (is_running() == false || task == NULL)
     return false;
 
+  {
+    Mutex_lock lock(m_post_mutex);
+
+    if (increase_tasks_count() >= my_atomic_load32(&m_workers_count))
+    {
+      try { create_thread(); }
+      catch (std::exception &e)
+      {
+        log_error("Exception in post: %s", e.what());
+        decrease_tasks_count();
+        return false;
+      }
+    }
+  }
+
   while (m_tasks.push(task) == false) {}
   m_task_pending_cond.signal(m_task_pending_mutex);
 
-  if (increase_tasks_count() >= my_atomic_load32(&m_workers_count))
-    create_thread();
-
   return true;
 }
 
 
 bool Scheduler_dynamic::post(const Task& task)
 {
-  if (is_running() == false || task == NULL)
-    return false;
-
   Task *copy_task = new (std::nothrow) Task(task);
 
-  if (NULL == copy_task)
-    return false;
-
-  while (m_tasks.push(copy_task) == false) {}
-  m_task_pending_cond.signal(m_task_pending_mutex);
-
-  if (increase_tasks_count() >= my_atomic_load32(&m_workers_count))
-    create_thread();
+  if (post(copy_task))
+    return true;
 
-  return true;
+  delete copy_task;
+  return false;
 }
 
+
 bool Scheduler_dynamic::post_and_wait(const Task& task_to_be_posted)
 {
   Wait_for_signal future;
 
   {
     ngs::Scheduler_dynamic::Task task = boost::bind(&Wait_for_signal::Signal_when_done::execute,
-        boost::make_shared<ngs::Wait_for_signal::Signal_when_done>(boost::ref(future), task_to_be_posted));
+                                                    boost::make_shared<ngs::Wait_for_signal::Signal_when_done>(boost::ref(future), task_to_be_posted));
 
     if (!post(task))
+    {
+      log_error("Internal error scheduling task");
       return false;
+    }
   }
 
   future.wait();
@@ -190,6 +214,7 @@ void *Scheduler_dynamic::worker_proxy(void *data)
   return reinterpret_cast<Scheduler_dynamic*>(data)->worker();
 }
 
+
 void *Scheduler_dynamic::worker()
 {
   bool worker_timed_out = false;
@@ -259,15 +284,8 @@ void Scheduler_dynamic::create_thread()
   if (is_running())
   {
     Thread_t thread;
-
+    ngs::thread_create(0, &thread, NULL, worker_proxy, this);
     increase_workers_count();
-
-    if (ngs::thread_create(0, &thread, NULL, worker_proxy, this) < 0)
-    {
-      decrease_workers_count();
-      throw std::runtime_error("Could not create a worker thread.");
-    }
-
     m_threads.push(thread);
   }
 }

rapid/plugin/x/ngs/src/thread.cc

@@ -25,11 +25,12 @@
 #include "ngs/thread.h"
 
 
-int ngs::thread_create(PSI_thread_key key, Thread_t *thread,
-                       const Thread_attr_t *attr,
-                       Start_routine_t func, void *arg)
+void ngs::thread_create(PSI_thread_key key, Thread_t *thread,
+                        const Thread_attr_t *attr,
+                        Start_routine_t func, void *arg)
 {
-  return mysql_thread_create(key, thread, attr, func, arg);
+  if (mysql_thread_create(key, thread, attr, func, arg))
+    throw std::runtime_error("Could not create a thread");
 }
 
 

rapid/plugin/x/src/xpl_performance_schema.cc

@@ -36,11 +36,13 @@ static PSI_thread_info all_x_threads[] = {
 PSI_mutex_key KEY_mutex_x_lock_list_access = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_scheduler_dynamic_task_pending = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_scheduler_dynamic_thread_exit = PSI_NOT_INSTRUMENTED;
+PSI_mutex_key KEY_mutex_x_scheduler_dynamic_post = PSI_NOT_INSTRUMENTED;
 
 static PSI_mutex_info all_x_mutexes[] = {
   { &KEY_mutex_x_lock_list_access, "lock_list_access", 0 },
   { &KEY_mutex_x_scheduler_dynamic_task_pending, "scheduler_dynamic_task_pending", 0 },
   { &KEY_mutex_x_scheduler_dynamic_thread_exit, "scheduler_dynamic_thread_exit", 0 },
+  { &KEY_mutex_x_scheduler_dynamic_post, "scheduler_dynamic_post", 0 }
 };
 
 PSI_cond_key KEY_cond_x_scheduler_dynamic_task_pending = PSI_NOT_INSTRUMENTED;

rapid/plugin/x/src/xpl_performance_schema.h

@@ -33,6 +33,7 @@ extern PSI_thread_key KEY_thread_x_acceptor;
 extern PSI_mutex_key KEY_mutex_x_lock_list_access;
 extern PSI_mutex_key KEY_mutex_x_scheduler_dynamic_task_pending;
 extern PSI_mutex_key KEY_mutex_x_scheduler_dynamic_thread_exit;
+extern PSI_mutex_key KEY_mutex_x_scheduler_dynamic_post;
 
 extern PSI_cond_key KEY_cond_x_scheduler_dynamic_task_pending;
 extern PSI_cond_key KEY_cond_x_scheduler_dynamic_thread_exit;

rapid/plugin/x/src/xpl_server.cc

@@ -158,10 +158,16 @@ bool xpl::Server::on_verify_server_state()
 
     // closing clients has been moved to other thread
     // this thread have to gracefully shutdown io operations
-    typedef ngs::Scheduler_dynamic::Task Task;
-    Task *task = new Task(boost::bind(&ngs::Server::close_all_clients, &m_server));
-    if (!m_wscheduler->post(task))
-      delete task;
+    if (m_wscheduler->is_running())
+    {
+      typedef ngs::Scheduler_dynamic::Task Task;
+      Task *task = new Task(boost::bind(&ngs::Server::close_all_clients, &m_server));
+      if (!m_wscheduler->post(task))
+      {
+        log_debug("Unable to schedule closing all clients ");
+        delete task;
+      }
+    }
 
     // stop the server, making the event loop stop looping around
     m_server.stop();
@@ -171,6 +177,7 @@ bool xpl::Server::on_verify_server_state()
   return true;
 }
 
+
 boost::shared_ptr<ngs::Client> xpl::Server::create_client(ngs::Connection_ptr connection)
 {
   return boost::make_shared<xpl::Client>(connection, &m_server, ++m_client_id, new xpl::Protocol_monitor());
@@ -216,6 +223,7 @@ void xpl::Server::did_accept_client(boost::shared_ptr<ngs::Client> client)
   Global_status_variables::instance().increment_accepted_connections_count();
 }
 
+
 void xpl::Server::did_reject_client(ngs::Server_delegate::Reject_reason reason)
 {
   switch (reason)
@@ -233,7 +241,10 @@ void xpl::Server::did_reject_client(ngs::Server_delegate::Reject_reason reason)
 
 void xpl::Server::plugin_system_variables_changed()
 {
-  m_wscheduler->set_num_workers(Plugin_system_variables::min_worker_threads);
+  const unsigned int min = m_wscheduler->set_num_workers(Plugin_system_variables::min_worker_threads);
+  if (min < Plugin_system_variables::min_worker_threads)
+    Plugin_system_variables::min_worker_threads = min;
+
   m_wscheduler->set_idle_worker_timeout(Plugin_system_variables::idle_worker_thread_timeout * 1000);
 
   m_config->max_message_size = Plugin_system_variables::max_allowed_packet;
@@ -246,11 +257,13 @@ bool xpl::Server::is_terminating() const
   return mysqld::is_terminating();
 }
 
+
 bool xpl::Server::is_exiting()
 {
   return mysqld::is_terminating() || xpl::Server::exiting;
 }
 
+
 int xpl::Server::main(MYSQL_PLUGIN p)
 {
   xpl::plugin_handle = p;

rapid/unittest/gunit/xplugin/mock/session.h

@@ -59,7 +59,7 @@ class Mock_scheduler_dynamic : public Scheduler_dynamic
   MOCK_METHOD0(stop, void());
   MOCK_METHOD0(thread_init, bool ());
   MOCK_METHOD0(thread_end, void ());
-  MOCK_METHOD1(set_num_workers, void(unsigned int n));
+  MOCK_METHOD1(set_num_workers, unsigned int(unsigned int n));
 };
 
 

rapid/unittest/gunit/xplugin/stubs/misc.cc

@@ -34,6 +34,7 @@ PSI_mutex_key KEY_mutex_x_obuffer = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_lock_list_access = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_scheduler_dynamic_task_pending = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_scheduler_dynamic_thread_exit = PSI_NOT_INSTRUMENTED;
+PSI_mutex_key KEY_mutex_x_scheduler_dynamic_post = PSI_NOT_INSTRUMENTED;
 PSI_mutex_key KEY_mutex_x_queue = PSI_NOT_INSTRUMENTED;
 
 PSI_cond_key KEY_cond_x_scheduler_dynamic_task_pending = PSI_NOT_INSTRUMENTED;