Allow dynamic modification of io-threads num

Signed-off-by: Ayush Sharma <mrayushs933@gmail.com>

Author: ayush933

src/config.c

@@ -28,6 +28,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include "io_threads.h"
 #include "server.h"
 #include "cluster.h"
 #include "connection.h"
@@ -3264,8 +3265,8 @@ standardConfig static_configs[] = {
 
     /* Integer configs */
     createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.dbnum, 16, INTEGER_CONFIG, NULL, NULL),
-    createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.port, 6379, INTEGER_CONFIG, NULL, updatePort),                                   /* TCP port. */
-    createIntConfig("io-threads", NULL, DEBUG_CONFIG | IMMUTABLE_CONFIG, 1, IO_THREADS_MAX_NUM, server.io_threads_num, 1, INTEGER_CONFIG, NULL, NULL), /* Single threaded by default */
+    createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.port, 6379, INTEGER_CONFIG, NULL, updatePort),                                               /* TCP port. */
+    createIntConfig("io-threads", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1, IO_THREADS_MAX_NUM, server.io_threads_num, 1, INTEGER_CONFIG, NULL, updateIOThreads), /* Single threaded by default */
     createIntConfig("events-per-io-thread", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.events_per_io_thread, 2, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("prefetch-batch-max-size", NULL, MODIFIABLE_CONFIG, 0, 128, server.prefetch_batch_max_size, 16, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("auto-aof-rewrite-percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.aof_rewrite_perc, 100, INTEGER_CONFIG, NULL, NULL),

src/io_threads.c

@@ -227,6 +227,8 @@ static void *IOThreadMain(void *myid) {
     size_t jobs_to_process = 0;
     IOJobQueue *jq = &io_jobs[id];
     while (1) {
+        /* Cancellation point so that pthread_cancel() from main thread is honored. */
+        pthread_testcancel();
         /* Wait for jobs */
         for (int j = 0; j < 1000000; j++) {
             jobs_to_process = IOJobQueue_availableJobs(jq);
@@ -297,6 +299,41 @@ void killIOThreads(void) {
     }
 }
 
+int updateIOThreads(const char **err) {
+    serverAssert(inMainThread());
+    UNUSED(err);
+    int prev_threads_num = 1;
+    for (int i = IO_THREADS_MAX_NUM - 1; i > 0; i--) {
+        if (io_threads[i]) {
+            prev_threads_num = i + 1;
+            break;
+        }
+    }
+    if (prev_threads_num == server.io_threads_num) return 1;
+    serverLog(LL_DEBUG, "Changing number of IO threads from %d to %d.", prev_threads_num, server.io_threads_num);
+    drainIOThreadsQueue();
+    // Create new threads.
+    if (server.io_threads_num > prev_threads_num) {
+        for (int i = prev_threads_num; i < server.io_threads_num; i++) {
+            createIOThread(i);
+        }
+    }
+    // Decrease the number of threads.
+    else {
+        for (int i = prev_threads_num - 1; i >= server.io_threads_num; i--) {
+            // Unblock inactive thread.
+            if (i >= server.active_io_threads_num)
+                pthread_mutex_unlock(&io_threads_mutex[i]);
+            shutdownIOThread(i);
+            io_threads[i] = 0;
+        }
+    }
+    /* Set active threads to 1, will be adjusted based on workload later. */
+    for (int i = 1; i < server.io_threads_num; i++) pthread_mutex_lock(&io_threads_mutex[i]);
+    server.active_io_threads_num = 1;
+    return 1;
+}
+
 /* Initialize the data structures needed for I/O threads. */
 void initIOThreads(void) {
     server.active_io_threads_num = 1; /* We start with threads not active. */

src/io_threads.h

@@ -14,5 +14,6 @@ void adjustIOThreadsByEventLoad(int numevents, int increase_only);
 void drainIOThreadsQueue(void);
 void trySendPollJobToIOThreads(void);
 int trySendAcceptToIOThreads(connection *conn);
+int updateIOThreads(const char **err);
 
 #endif /* IO_THREADS_H */