MDEV-33627 refactor threading in mariadb-import

Use threadpool, instead of one-thread-and-connection-per-table

client/CMakeLists.txt

@@ -57,7 +57,8 @@ MYSQL_ADD_EXECUTABLE(mariadb-dump mysqldump.cc ../sql-common/my_user.c connectio
 TARGET_LINK_LIBRARIES(mariadb-dump ${CLIENT_LIB})
 
 MYSQL_ADD_EXECUTABLE(mariadb-import mysqlimport.cc)
-TARGET_LINK_LIBRARIES(mariadb-import ${CLIENT_LIB})
+target_include_directories(mariadb-import PRIVATE ${CMAKE_SOURCE_DIR}/tpool)
+target_link_libraries(mariadb-import PRIVATE tpool ${CLIENT_LIB})
 
 MYSQL_ADD_EXECUTABLE(mariadb-upgrade mysql_upgrade.c COMPONENT Server)
 TARGET_LINK_LIBRARIES(mariadb-upgrade ${CLIENT_LIB})

client/mysqlimport.cc

@@ -36,12 +36,8 @@
 
 #include <welcome_copyright_notice.h>   /* ORACLE_WELCOME_COPYRIGHT_NOTICE */
 
-
-/* Global Thread counter */
-uint counter= 0;
-pthread_mutex_t init_mutex;
-pthread_mutex_t counter_mutex;
-pthread_cond_t count_threshhold;
+#include <tpool.h>
+tpool::thread_pool *thread_pool;
 
 static void db_error_with_table(MYSQL *mysql, char *table);
 static void db_error(MYSQL *mysql);
@@ -445,13 +441,10 @@ static MYSQL *db_connect(char *host, char *database,
     fprintf(stdout, "Connecting to %s\n", host ? host : "localhost");
   if (opt_use_threads && !lock_tables)
   {
-    pthread_mutex_lock(&init_mutex);
     if (!(mysql= mysql_init(NULL)))
     {
-      pthread_mutex_unlock(&init_mutex);
       return 0;
     }
-    pthread_mutex_unlock(&init_mutex);
   }
   else
     if (!(mysql= mysql_init(NULL)))
@@ -496,6 +489,8 @@ static MYSQL *db_connect(char *host, char *database,
   if (ignore_foreign_keys)
     mysql_query(mysql, "set foreign_key_checks= 0;");
 
+  if (mysql_query(mysql, "/*!40101 set @@character_set_database=binary */;"))
+    db_error(mysql);
   return mysql;
 }
 
@@ -514,14 +509,10 @@ static void safe_exit(int error, MYSQL *mysql)
   if (error && ignore_errors)
     return;
 
-  /* in multi-threaded mode protect from concurrent safe_exit's */
-  if (counter)
-    pthread_mutex_lock(&counter_mutex);
-
   if (mysql)
     mysql_close(mysql);
 
-  if (counter)
+  if (thread_pool)
   {
     /* dirty exit. some threads are running,
        memory is not freed, openssl not deinitialized */
@@ -603,49 +594,42 @@ static char *field_escape(char *to,const char *from,uint length)
   return to;
 }
 
-int exitcode= 0;
-
-pthread_handler_t worker_thread(void *arg)
+std::atomic<int> exitcode;
+void set_exitcode(int code)
 {
-  int error;
-  char *raw_table_name= (char *)arg;
-  MYSQL *mysql= 0;
+  int expected= 0;
+  exitcode.compare_exchange_strong(expected,code);
+}
 
-  if (mysql_thread_init())
-    goto error;
-  
-  if (!(mysql= db_connect(current_host,current_db,current_user,opt_password)))
-  {
-    goto error;
-  }
+thread_local MYSQL *thread_local_mysql;
 
-  if (mysql_query(mysql, "/*!40101 set @@character_set_database=binary */;"))
-  {
-    db_error(mysql); /* We shall continue here, if --force was given */
-    goto error;
-  }
 
+void load_single_table(void *arg)
+{
+  int error;
+  char *raw_table_name= (char *)arg;
+  MYSQL *mysql= thread_local_mysql;
   /*
     We are not currently catching the error here.
   */
   if((error= write_to_table(raw_table_name, mysql)))
-    if (exitcode == 0)
-      exitcode= error;
-
-error:
-  if (mysql)
-    db_disconnect(current_host, mysql);
+    set_exitcode(error);
+}
 
-  pthread_mutex_lock(&counter_mutex);
-  counter--;
-  pthread_cond_signal(&count_threshhold);
-  pthread_mutex_unlock(&counter_mutex);
+static void tpool_thread_init(void)
+{
+  mysql_thread_init();
+  thread_local_mysql= db_connect(current_host,current_db,current_user,opt_password);
+}
+static void tpool_thread_exit(void)
+{
+  if (thread_local_mysql)
+   db_disconnect(current_host,thread_local_mysql);
   mysql_thread_end();
-  pthread_exit(0);
-  return 0;
 }
 
 
+#include <vector>
 int main(int argc, char **argv)
 {
   int error=0;
@@ -668,102 +652,33 @@ int main(int argc, char **argv)
 
   if (opt_use_threads && !lock_tables)
   {
-    char **save_argv;
-    uint worker_thread_count= 0, table_count= 0, i= 0;
-    pthread_t *worker_threads;       /* Thread descriptor */
-    pthread_attr_t attr;             /* Thread attributes */
-    pthread_attr_init(&attr);
-    pthread_attr_setdetachstate(&attr,
-                                PTHREAD_CREATE_JOINABLE);
-
-    pthread_mutex_init(&init_mutex, NULL);
-    pthread_mutex_init(&counter_mutex, NULL);
-    pthread_cond_init(&count_threshhold, NULL);
-
-    /* Count the number of tables. This number denotes the total number
-       of threads spawn.
-    */
-    save_argv= argv;
-    for (table_count= 0; *argv != NULL; argv++)
-      table_count++;
-    argv= save_argv;
-
-    if (!(worker_threads= (pthread_t*) my_malloc(PSI_NOT_INSTRUMENTED,
-                               table_count * sizeof(*worker_threads), MYF(0))))
-      return -2;
-
-    for (; *argv != NULL; argv++) /* Loop through tables */
-    {
-      pthread_mutex_lock(&counter_mutex);
-      while (counter == opt_use_threads)
-      {
-        struct timespec abstime;
-
-        set_timespec(abstime, 3);
-        pthread_cond_timedwait(&count_threshhold, &counter_mutex, &abstime);
-      }
-      /* Before exiting the lock we set ourselves up for the next thread */
-      counter++;
-      pthread_mutex_unlock(&counter_mutex);
-      /* now create the thread */
-      if (pthread_create(&worker_threads[worker_thread_count], &attr,
-                         worker_thread, (void *)*argv) != 0)
-      {
-        pthread_mutex_lock(&counter_mutex);
-        counter--;
-        pthread_mutex_unlock(&counter_mutex);
-        fprintf(stderr,"%s: Could not create thread\n", my_progname);
-        continue;
-      }
-      worker_thread_count++;
-    }
+    thread_pool= tpool::create_thread_pool_generic(opt_use_threads,opt_use_threads);
+    thread_pool->set_thread_callbacks(tpool_thread_init,tpool_thread_exit);
 
-    /*
-      We loop until we know that all children have cleaned up.
-    */
-    pthread_mutex_lock(&counter_mutex);
-    while (counter)
-    {
-      struct timespec abstime;
-
-      set_timespec(abstime, 3);
-      pthread_cond_timedwait(&count_threshhold, &counter_mutex, &abstime);
-    }
-    pthread_mutex_unlock(&counter_mutex);
-    pthread_mutex_destroy(&init_mutex);
-    pthread_mutex_destroy(&counter_mutex);
-    pthread_cond_destroy(&count_threshhold);
-    pthread_attr_destroy(&attr);
+    std::vector<tpool::task> all_tasks;
+    for (int i=0; argv[i]; i++)
+      all_tasks.push_back(tpool::task(load_single_table, argv[i]));
 
-    for(i= 0; i < worker_thread_count; i++)
-    {
-      if (pthread_join(worker_threads[i], NULL))
-        fprintf(stderr,"%s: Could not join worker thread.\n", my_progname);
-    }
+    for (auto &t: all_tasks)
+      thread_pool->submit_task(&t);
 
-    my_free(worker_threads);
+    delete thread_pool;
+    thread_pool= nullptr;
   }
   else
   {
-    MYSQL *mysql= 0;
-    if (!(mysql= db_connect(current_host,current_db,current_user,opt_password)))
+    MYSQL *mysql= db_connect(current_host,current_db,current_user,opt_password);
+    if (!mysql)
     {
       free_defaults(argv_to_free);
       return(1); /* purecov: dead code */
     }
 
-    if (mysql_query(mysql, "/*!40101 set @@character_set_database=binary */;"))
-    {
-      db_error(mysql); /* We shall continue here, if --force was given */
-      return(1);
-    }
-
     if (lock_tables)
       lock_table(mysql, argc, argv);
     for (; *argv != NULL; argv++)
       if ((error= write_to_table(*argv, mysql)))
-        if (exitcode == 0)
-          exitcode= error;
+          set_exitcode(error);
     db_disconnect(current_host, mysql);
   }
   safe_exit(0, 0);