Make threadpool-size a configurable parameter with an auto option

(default), in which case the size of the threadpool would be equal to
the number of CPU's. (hopefully this is cross-platform)

redis.conf

@@ -36,6 +36,11 @@ port 6379
 # unixsocket /tmp/redis.sock
 # unixsocketperm 755
 
+# size of the threadpool, default is auto. can be a number between
+# 1 and REDIS_THREADPOOL_MAX_SIZE
+#
+# threadpool-size auto
+
 # Close the connection after a client is idle for N seconds (0 to disable)
 timeout 0
 

src/config.c

@@ -384,6 +384,15 @@ void loadServerConfigFromString(char *config) {
             }
         } else if (!strcasecmp(argv[0],"slave-priority") && argc == 2) {
             server.slave_priority = atoi(argv[1]);
+        } else if (!strcasecmp(argv[0],"threadpool-size") && argc == 2) {
+            if (!strcasecmp(argv[1], "auto")) {
+                server.threadpool_size = -1;
+            } else {
+                server.threadpool_size = atoi(argv[1]);
+                if ((server.threadpool_size <= 0) || (server.threadpool_size > REDIS_THREADPOOL_MAX_SIZE)) {
+                    err = "Invalid threadpool-size"; goto loaderr;
+                }
+            }
         } else if (!strcasecmp(argv[0],"sentinel")) {
             /* argc == 1 is handled by main() as we need to enter the sentinel
              * mode ASAP. */
@@ -800,6 +809,7 @@ void configGetCommand(redisClient *c) {
     config_get_numerical_field("maxclients",server.maxclients);
     config_get_numerical_field("watchdog-period",server.watchdog_period);
     config_get_numerical_field("slave-priority",server.slave_priority);
+    config_get_numerical_field("threadpool-size",server.threadpool_size);
 
     /* Bool (yes/no) values */
     config_get_bool_field("no-appendfsync-on-rewrite",

src/redis.c

@@ -1153,6 +1153,7 @@ void initServerConfig() {
     server.repl_ping_slave_period = REDIS_REPL_PING_SLAVE_PERIOD;
     server.repl_timeout = REDIS_REPL_TIMEOUT;
     server.lua_time_limit = REDIS_LUA_TIME_LIMIT;
+    server.threadpool_size = -1;
 
     updateLRUClock();
     resetServerSaveParams();
@@ -1264,7 +1265,6 @@ int getNumCPUs() {
 
 void initServer() {
     int j;
-    int numcpu;
 
     signal(SIGHUP, SIG_IGN);
     signal(SIGPIPE, SIG_IGN);
@@ -1365,10 +1365,11 @@ void initServer() {
         }
     }
 
-    if ((numcpu = getNumCPUs()) <= 0)
-        numcpu = REDIS_THREADPOOL_DEFAULT_SIZE;
-    redisLog(REDIS_NOTICE,"Starting %d worker threads with a threadpool queue of size %d.", numcpu, REDIS_THREADPOOL_DEFAULT_QUEUE_SIZE);
-    server.tpool = threadpool_create(numcpu, REDIS_THREADPOOL_DEFAULT_QUEUE_SIZE, 0); // THREDIS TODO - this should be configurable
+    if (server.threadpool_size == -1)
+        if ((server.threadpool_size = getNumCPUs()) <= 0)
+            server.threadpool_size = REDIS_THREADPOOL_DEFAULT_SIZE;
+    redisLog(REDIS_NOTICE,"Starting %d worker threads with a threadpool queue of size %d.", server.threadpool_size, REDIS_THREADPOOL_DEFAULT_QUEUE_SIZE);
+    server.tpool = threadpool_create(server.threadpool_size, REDIS_THREADPOOL_DEFAULT_QUEUE_SIZE, 0); // THREDIS TODO - queue size should be configurable
     server.lock = zmalloc(sizeof(pthread_mutex_t));
     pthread_mutex_init(server.lock, NULL);
 

src/redis.h

@@ -292,6 +292,7 @@
 
 /* Threads */
 #define REDIS_THREADPOOL_DEFAULT_SIZE 4
+#define REDIS_THREADPOOL_MAX_SIZE 1024
 #define REDIS_THREADPOOL_DEFAULT_QUEUE_SIZE 1024
 
 /* Using the following macro you can run code inside serverCron() with the
@@ -670,6 +671,7 @@ struct redisServer {
     int watchdog_period;  /* Software watchdog period in ms. 0 = off */
 
     threadpool_t *tpool;
+    int threadpool_size;
     pthread_mutex_t *lock;   /* only used for blocking BL* command stuff */
 };