Use threadpool for async scs

Signed-off-by: mdouglas47 <[email protected]>

Author: morgando

db/comdb2.c

@@ -128,6 +128,7 @@ void berk_memp_sync_alarm_ms(int);
 #include <cdb2_constants.h>
 #include <bb_oscompat.h>
 #include <schemachange.h>
+#include <sc_global.h>
 #include "comdb2_atomic.h"
 #include "cron.h"
 #include "metrics.h"
@@ -5849,6 +5850,10 @@ int main(int argc, char **argv)
     */
     gbl_tunables->freeze = 1;
 
+    if (init_sc_globals()) {
+        logmsg(LOGMSG_FATAL, "Failed to init sc globals\n");
+        exit(1);
+    }
     handle_resume_sc();
 
     create_marker_file();

db/osqlblockproc.c

@@ -1305,6 +1305,18 @@ int bplog_schemachange_wait(struct ireq *iq, int rc)
     iq->sc_pending = NULL;
 
     while (sc != NULL) {
+        if (!sc->nothrevent) {
+            // If sc is running asynchronously, we must first hold mtxStart and
+            // wait on condStart until sc->started is set to true—otherwise we risk
+            // locking sc->mtx before the sc has begun. 
+            Pthread_mutex_lock(&sc->mtxStart);
+            while (!sc->started) {
+                Pthread_cond_wait(&sc->condStart, &sc->mtxStart);
+            }
+            Pthread_mutex_unlock(&sc->mtxStart);
+        }
+        // Now that sc is guaranteed to be running, we can safely lock its main mutex,
+        // which the sc will release once it completes
         Pthread_mutex_lock(&sc->mtx);
         sc->nothrevent = 1;
         iq->sc = sc->sc_next;

db/osqlcomm.c

@@ -6563,8 +6563,7 @@ static int _process_single_table_sc_partitioning(struct ireq *iq)
         arg.start = 1; /* first shard is already there */
         arg.pos = 0; /* reset this so we do not set publish on additional shards */
     }
-    /* should we serialize ? */
-    arg.s->nothrevent = sc->partition.u.tpt.retention > gbl_dohsql_sc_max_threads;
+    arg.s->nothrevent = 0;
     rc = timepart_foreach_shard_lockless(
             sc->newpartition, start_schema_change_tran_wrapper, &arg);
 

schemachange/sc_global.c

@@ -55,7 +55,6 @@ struct schema_change_type *sc_resuming = NULL;
 /* async ddl sc */
 pthread_mutex_t sc_async_mtx = PTHREAD_MUTEX_INITIALIZER;
 pthread_cond_t sc_async_cond = PTHREAD_COND_INITIALIZER;
-volatile int sc_async_threads = 0;
 
 /* Throttle settings, which you can change with message traps.  Note that if
  * you have gbl_sc_usleep=0, the important live writer threads never get to
@@ -688,3 +687,18 @@ void sc_alter_latency(int counter)
         }
     }
 }
+
+struct thdpool *gbl_sc_thdpool = NULL;
+int init_sc_globals()
+{
+    gbl_sc_thdpool = thdpool_create("sc_thdpool", 0);
+    if (!gbl_sc_thdpool) {
+        logmsg(LOGMSG_ERROR, "%s: Failed to allocate threadpool\n", __func__);
+        return 1;
+    }
+
+    thdpool_set_maxthds(gbl_sc_thdpool, bdb_attr_get(thedb->bdb_attr, BDB_ATTR_SC_ASYNC_MAXTHREADS));
+    thdpool_set_maxqueue(gbl_sc_thdpool, 1000);
+
+    return 0;
+}

schemachange/sc_global.h

@@ -30,7 +30,6 @@ extern int gbl_default_sc_scanmode;
 
 extern pthread_mutex_t sc_async_mtx;
 extern pthread_cond_t sc_async_cond;
-extern volatile int sc_async_threads;
 
 /* Throttle settings, which you can change with message traps.  Note that if
  * you have gbl_sc_usleep=0, the important live writer threads never get to
@@ -86,5 +85,6 @@ struct schema_change_type *preempt_ongoing_alter(char *table, int action);
 void clear_ongoing_alter();
 int get_stopsc(const char *func, int line);
 void sc_alter_latency(int counter);
+int init_sc_globals();
 
 #endif

schemachange/sc_logic.c

@@ -66,10 +66,9 @@ static enum thrtype prepare_sc_thread(struct schema_change_type *s)
             thrman_change_type(thr_self, THRTYPE_SCHEMACHANGE);
         } else
             thr_self = thrman_register(THRTYPE_SCHEMACHANGE);
-        if (!s->nothrevent) {
-            backend_thread_event(thedb, COMDB2_THR_EVENT_START_RDWR);
-            logmsg(LOGMSG_INFO, "Preparing schema change read write thread\n");
-        }
+
+        backend_thread_event(thedb, COMDB2_THR_EVENT_START_RDWR);
+        logmsg(LOGMSG_INFO, "Preparing schema change read write thread\n");
     }
     return oldtype;
 }
@@ -728,12 +727,6 @@ static int do_schema_change_tran_int(sc_arg_t *arg)
     else
         s->sc_rc = rc;
 
-    if (!s->nothrevent) {
-        Pthread_mutex_lock(&sc_async_mtx);
-        sc_async_threads--;
-        Pthread_cond_broadcast(&sc_async_cond);
-        Pthread_mutex_unlock(&sc_async_mtx);
-    }
     if (rc == SC_COMMIT_PENDING && (s->preempted == SC_ACTION_RESUME ||
                                     s->kind == SC_ALTERTABLE_PENDING)) {
         int bdberr = 0;
@@ -795,6 +788,13 @@ int do_schema_change_tran_thd(sc_arg_t *arg)
     return rc;
 }
 
+void do_schema_change_tran_thd_thdpool_wrapper(struct thdpool *pool, void *work, void *thddata, int op)
+{
+    bdb_thread_event(thedb->bdb_env, 1);
+    do_schema_change_tran_int(work);
+    bdb_thread_event(thedb->bdb_env, 0);
+}
+
 int do_schema_change_locked(struct schema_change_type *s, void *tran)
 {
     comdb2_name_thread(__func__);
@@ -825,6 +825,11 @@ int do_schema_change_locked(struct schema_change_type *s, void *tran)
     return rc;
 }
 
+void do_schema_change_locked_thdpool_wrapper(struct thdpool *pool, void *work, void *thddata, int op)
+{
+    do_schema_change_locked(work, NULL);
+}
+
 int finalize_schema_change(struct ireq *iq, tran_type *trans)
 {
     if (iq == NULL || iq->sc == NULL) abort();

schemachange/sc_logic.h

@@ -35,5 +35,7 @@ int verify_constraints_exist(struct dbtable *from_db, struct dbtable *to_db,
 
 int do_schema_change_tran(sc_arg_t *);
 int do_schema_change_tran_thd(sc_arg_t *);
+void do_schema_change_tran_thd_thdpool_wrapper(struct thdpool *pool, void *work, void *thddata, int op);
 int do_schema_change_locked(struct schema_change_type *, void *tran);
+void do_schema_change_locked_thdpool_wrapper(struct thdpool *pool, void *work, void *thddata, int op);
 #endif

schemachange/schemachange.c

@@ -41,6 +41,34 @@ const char *get_hostname_with_crc32(bdb_state_type *bdb_state,
                                     unsigned int hash);
 
 extern int gbl_test_sc_resume_race;
+extern struct thdpool *gbl_sc_thdpool;
+
+static int do_schema_change_async(struct ireq * iq, sc_arg_t *arg, struct schema_change_type *s)
+{
+    int rc = 0;
+
+    iq->sc_locked = 0;
+    if (s->kind == SC_ALTERTABLE_PENDING ||
+        s->preempted == SC_ACTION_RESUME) {
+        free(arg);
+        arg = NULL;
+        rc = thdpool_enqueue(gbl_sc_thdpool, (thdpool_work_fn) do_schema_change_locked_thdpool_wrapper, s, 0, NULL, THDPOOL_FORCE_QUEUE);
+    } else {
+        rc = thdpool_enqueue(gbl_sc_thdpool, (thdpool_work_fn) do_schema_change_tran_thd_thdpool_wrapper, arg, 0, NULL, THDPOOL_FORCE_QUEUE);
+    }
+    if (rc) {
+        logmsg(LOGMSG_ERROR, "%s:thdpool_enqueue rc %d\n", __func__, rc);
+        if (arg) { free(arg); }
+        if (!s->is_osql) {
+            sc_set_running(iq, s, s->tablename, 0, gbl_myhostname,
+                           time(NULL), __func__, __LINE__);
+            free_schema_change_type(s);
+        }
+        rc = SC_ASYNC_FAILED;
+    }
+
+    return rc;
+}
 
 /* If this is successful, it increments */
 int start_schema_change_tran(struct ireq *iq, tran_type *trans)
@@ -324,66 +352,19 @@ int start_schema_change_tran(struct ireq *iq, tran_type *trans)
     *forward
     ** in order to produce minimal spew
     */
-    if (s->nothrevent) {
-        if (s->kind != SC_PARTIALUPRECS)
-            logmsg(LOGMSG_INFO, "Executing SYNCHRONOUSLY\n");
-        rc = do_schema_change_tran(arg);
-    } else {
-        int max_threads =
-            bdb_attr_get(thedb->bdb_attr, BDB_ATTR_SC_ASYNC_MAXTHREADS);
-        Pthread_mutex_lock(&sc_async_mtx);
-        while (!s->must_resume && !s->resume && max_threads > 0 &&
-               sc_async_threads >= max_threads) {
-            logmsg(LOGMSG_INFO, "Waiting for avaiable schema change threads\n");
-            Pthread_cond_wait(&sc_async_cond, &sc_async_mtx);
-        }
-        sc_async_threads++;
-        Pthread_mutex_unlock(&sc_async_mtx);
-
-        if (s->kind != SC_PARTIALUPRECS)
-            logmsg(LOGMSG_INFO, "Executing ASYNCHRONOUSLY\n");
-        pthread_t tid;
-
-        if (s->kind == SC_ALTERTABLE_PENDING ||
-            s->preempted == SC_ACTION_RESUME) {
-            free(arg);
-            arg = NULL;
-            rc = pthread_create(&tid, &gbl_pthread_attr_detached,
-                                (void *(*)(void *))do_schema_change_locked, s);
-        } else {
-            Pthread_mutex_lock(&s->mtxStart);
-            rc = pthread_create(&tid, &gbl_pthread_attr_detached,
-                                (void *(*)(void *))do_schema_change_tran_thd,
-                                arg);
-            if (rc == 0) {
-                while (!s->started) {
-                    Pthread_cond_wait(&s->condStart, &s->mtxStart);
-                }
-            }
-            Pthread_mutex_unlock(&s->mtxStart);
-        }
-        if (rc) {
-            logmsg(LOGMSG_ERROR,
-                   "start_schema_change:pthread_create rc %d %s\n", rc,
-                   strerror(errno));
-
-            Pthread_mutex_lock(&sc_async_mtx);
-            sc_async_threads--;
-            Pthread_mutex_unlock(&sc_async_mtx);
-
-            if (arg)
-                free(arg);
-            if (!s->is_osql) {
-                sc_set_running(iq, s, s->tablename, 0, gbl_myhostname,
-                               time(NULL), __func__, __LINE__);
-                free_schema_change_type(s);
-            }
-            rc = SC_ASYNC_FAILED;
-        }
-    }
+    if (s->kind != SC_PARTIALUPRECS)
+        logmsg(LOGMSG_INFO, "Executing %s\n", s->nothrevent ?
+            "SYNCHRONOUSLY" : "ASYNCHRONOUSLY");
+
+    rc = s->nothrevent
+        ? do_schema_change_tran(arg)
+        : do_schema_change_async(iq, arg, s);
+
     /* SC_COMMIT_PENDING is SC_OK for the upper layers */
     if (rc == SC_COMMIT_PENDING) {
         rc = s->sc_rc = SC_OK;
+    } else if (rc) {
+        logmsg(LOGMSG_ERROR, "%s:%d Schema change failed\n", __func__, __LINE__);
     }
 
     return rc;

tests/comdb2sys.test/comdb2sys.expected

@@ -349,6 +349,7 @@
 (name='pgcompactpool')
 (name='recovery_processors')
 (name='recovery_workers')
+(name='sc_thdpool')
 (name='sqlenginepool')
 (name='udppfaultpool')
 [SELECT name FROM comdb2_threadpools ORDER BY name] rc 0