MOD-13142 Option to resize mr thpool after creation, but not after started

src/mr.c

@@ -1564,6 +1564,24 @@ int MR_Init(RedisModuleCtx* ctx, size_t numThreads, char *password) {
     return REDISMODULE_OK;
 }
 
+int MR_ResizeExecutionThreadPoolIfUnstarted(size_t numThreads) {
+    mr_threadpool tp = mrCtx.executionsThreadPool;
+
+    /* MR_Init() not called yet. */
+    if (!tp) {
+        return REDISMODULE_OK;
+    }
+
+    if (mr_thpool_workers_started(tp)) {
+        return REDISMODULE_ERR;
+    }
+
+    if (mr_thpool_resize_unstarted(tp, numThreads) != 0) {
+        return REDISMODULE_ERR;
+    }
+    return REDISMODULE_OK;
+}
+
 int MR_RegisterObject(MRObjectType* t) {
     mrCtx.objectTypesDict = array_append(mrCtx.objectTypesDict, t);
     t->id = array_len(mrCtx.objectTypesDict) - 1;

src/mr.h

@@ -181,6 +181,9 @@ LIBMR_API void MR_FreeExecution(Execution* e);
 /* Initialize mr library */
 LIBMR_API int MR_Init(struct RedisModuleCtx* ctx, size_t numThreads, char *password);
 
+/* Resize the execution thread pool with a new size if worker threads were never started. */
+LIBMR_API int MR_ResizeExecutionThreadPoolIfUnstarted(size_t numThreads);
+
 /* Register a new object type */
 LIBMR_API int MR_RegisterObject(MRObjectType* t);
 

src/utils/thpool.c

@@ -141,8 +141,9 @@ struct mr_thpool_* mr_thpool_init(int num_threads) {
   threads_on_hold = 0;
   threads_keepalive = 1;
 
-  if (num_threads < 0) {
-    num_threads = 0;
+  if (num_threads <= 0) {
+    err("thpool_init(): num_threads must be greater than 0\n");
+    return NULL;
   }
 
   /* Make new thread pool */
@@ -180,6 +181,42 @@ struct mr_thpool_* mr_thpool_init(int num_threads) {
   return thpool_p;
 }
 
+int mr_thpool_workers_started(mr_threadpool thpool) {
+  return thpool != NULL && thpool->is_threads_started;
+}
+
+int mr_thpool_resize_unstarted(mr_threadpool thpool, int num_threads) {
+
+  if (thpool == NULL) {
+    return -1;
+  }
+
+  if (num_threads <= 0) {
+    err("mr_thpool_resize_unstarted(): num_threads must be greater than 0\n");
+    return -1;
+  }
+
+  pthread_mutex_lock(&thpool->is_threads_started_lock);
+  if (thpool->is_threads_started) {
+    pthread_mutex_unlock(&thpool->is_threads_started_lock);
+    err("mr_thpool_resize_unstarted(): workers already started\n");
+    return -1;
+  }
+
+  mr_thread** new_pool =
+      (mr_thread**)MR_REALLOC(thpool->threads, (size_t)num_threads * sizeof(struct mr_thread*));
+  if (new_pool == NULL) {
+    pthread_mutex_unlock(&thpool->is_threads_started_lock);
+    err("mr_thpool_resize_unstarted(): realloc failed\n");
+    return -1;
+  }
+
+  thpool->threads = new_pool;
+  thpool->total_num_of_threads = num_threads;
+  pthread_mutex_unlock(&thpool->is_threads_started_lock);
+  return 0;
+}
+
 /* Add work to the thread pool */
 int mr_thpool_add_work(mr_thpool_* thpool_p, void (*function_p)(void*), void* arg_p) {
   mr_job* newjob;

src/utils/thpool.h

@@ -36,6 +36,15 @@ typedef struct mr_thpool_* mr_threadpool;
  */
 mr_threadpool mr_thpool_init(int num_threads);
 
+/** @return Non-zero if worker pthreads have been started, 0 otherwise. */
+int mr_thpool_workers_started(mr_threadpool thpool);
+
+/**
+ * Resize an unstarted pool: realloc the thread pointer table and update
+ * total_num_of_threads; job queue and sync primitives are unchanged.
+ * @return 0 on success, -1 on failure (pool unchanged on failure).
+ */
+int mr_thpool_resize_unstarted(mr_threadpool thpool, int num_threads);
 
 /**
  * @brief Add work to the job queue