added dictFetchValue() to dict.c to make hash table API a bit less verbose in the...

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index bb54cce02062b714b302d2c689561227cc1a65ba..4264b7e23510b131c989f772f08c7f068d6eb9b3 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -91,7 +91,7 @@
 #define REDIS_CONFIGLINE_MAX    1024
 #define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */
 #define REDIS_MAX_SYNC_TIME     60      /* Slave can't take more to sync */
-#define REDIS_EXPIRELOOKUPS_PER_CRON    10 /* try to expire 10 keys/loop */
+#define REDIS_EXPIRELOOKUPS_PER_CRON    10 /* lookup 10 expires per loop */
 #define REDIS_MAX_WRITE_PER_EVENT (1024*64)
 #define REDIS_REQUEST_MAX_SIZE (1024*1024*256) /* max bytes in inline command */
 
@@ -379,6 +379,7 @@ struct redisServer {
     char *requirepass;
     int shareobjects;
     int rdbcompression;
+    int activerehashing;
     /* Replication related */
     int isslave;
     char *masterauth;
@@ -1201,16 +1202,28 @@ static void tryResizeHashTables(void) {
     int j;
 
     for (j = 0; j < server.dbnum; j++) {
-        if (htNeedsResize(server.db[j].dict)) {
-            redisLog(REDIS_VERBOSE,"The hash table %d is too sparse, resize it...",j);
+        if (htNeedsResize(server.db[j].dict))
             dictResize(server.db[j].dict);
-            redisLog(REDIS_VERBOSE,"Hash table %d resized.",j);
-        }
         if (htNeedsResize(server.db[j].expires))
             dictResize(server.db[j].expires);
     }
 }
 
+/* Our hash table implementation performs rehashing incrementally while
+ * we write/read from the hash table. Still if the server is idle, the hash
+ * table will use two tables for a long time. So we try to use 1 millisecond
+ * of CPU time at every serverCron() loop in order to rehash some key. */
+static void incrementallyRehash(void) {
+    int j;
+
+    for (j = 0; j < server.dbnum; j++) {
+        if (dictIsRehashing(server.db[j].dict)) {
+            dictRehashMilliseconds(server.db[j].dict,1);
+            break; /* already used our millisecond for this loop... */
+        }
+    }
+}
+
 /* A background saving child (BGSAVE) terminated its work. Handle this. */
 void backgroundSaveDoneHandler(int statloc) {
     int exitcode = WEXITSTATUS(statloc);
@@ -1340,10 +1353,9 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
      * if we resize the HT while there is the saving child at work actually
      * a lot of memory movements in the parent will cause a lot of pages
      * copied. */
-    if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1 &&
-        !(loops % 10))
-    {
-        tryResizeHashTables();
+    if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) {
+        if (!(loops % 10)) tryResizeHashTables();
+        if (server.activerehashing) incrementallyRehash();
     }
 
     /* Show information about connected clients */
@@ -1571,6 +1583,7 @@ static void initServerConfig() {
     server.requirepass = NULL;
     server.shareobjects = 0;
     server.rdbcompression = 1;
+    server.activerehashing = 1;
     server.maxclients = 0;
     server.blpop_blocked_clients = 0;
     server.maxmemory = 0;
@@ -1804,6 +1817,10 @@ static void loadServerConfig(char *filename) {
             if ((server.rdbcompression = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"activerehashing") && argc == 2) {
+            if ((server.activerehashing = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {
             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
@@ -2304,7 +2321,8 @@ static int processCommand(redisClient *c) {
     }
 
     /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
-    if (dictSize(c->pubsub_channels) > 0 &&
+    if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0)
+        &&
         cmd->proc != subscribeCommand && cmd->proc != unsubscribeCommand &&
         cmd->proc != psubscribeCommand && cmd->proc != punsubscribeCommand) {
         addReplySds(c,sdsnew("-ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context\r\n"));
@@ -7054,7 +7072,7 @@ static void expireGenericCommand(redisClient *c, robj *key, robj *param, long of
         addReply(c,shared.czero);
         return;
     }
-    if (seconds < 0) {
+    if (seconds <= 0) {
         if (deleteKey(c->db,key)) server.dirty++;
         addReply(c, shared.cone);
         return;