save peak memory usage as statistic and show it in INFO. Also a new INFO field was...

[redis.git] / src / db.c

diff --git a/src/db.c b/src/db.c
index 7d2bf4af49be33d66e6ba1faac182bb6bc531b46..85a220626237ad41c4d1e322f6c95739b7cdf651 100644 (file)
--- a/src/db.c
+++ b/src/db.c
@@ -50,7 +50,7 @@ robj *lookupKey(redisDb *db, robj *key) {
             cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG)
         {
             /* Need to wait for the key to get unbusy */
-            redisLog(REDIS_WARNING,"Lookup found a key in SAVEINPROG state. Waiting. (Key was in the cache)");
+            redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting. (Key was in the cache)");
             lookupWaitBusyKey(db,key);
         }
         server.stat_keyspace_hits++;
@@ -76,7 +76,7 @@ robj *lookupKey(redisDb *db, robj *key) {
             /* At this point we need to blocking load the key in memory.
              * The first thing we do is waiting here if the key is busy. */
             if (flags & REDIS_IO_SAVEINPROG) {
-                redisLog(REDIS_WARNING,"Lookup found a key in SAVEINPROG state. Waiting (while force loading).");
+                redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting (while force loading).");
                 lookupWaitBusyKey(db,key);
             }
 
@@ -201,7 +201,8 @@ int dbDelete(redisDb *db, robj *key) {
     return dictDelete(db->dict,key->ptr) == DICT_OK;
 }
 
-/* Empty the whole database */
+/* Empty the whole database.
+ * If diskstore is enabled this function will just flush the in-memory cache. */
 long long emptyDb() {
     int j;
     long long removed = 0;
@@ -210,6 +211,7 @@ long long emptyDb() {
         removed += dictSize(server.db[j].dict);
         dictEmpty(server.db[j].dict);
         dictEmpty(server.db[j].expires);
+        if (server.ds_enabled) dictEmpty(server.db[j].io_negcache);
     }
     return removed;
 }
@@ -305,6 +307,10 @@ void existsCommand(redisClient *c) {
 void selectCommand(redisClient *c) {
     int id = atoi(c->argv[1]->ptr);
 
+    if (server.cluster_enabled) {
+        addReplyError(c,"SELECT is not allowed in cluster mode");
+        return;
+    }
     if (selectDb(c,id) == REDIS_ERR) {
         addReplyError(c,"invalid DB index");
     } else {
@@ -379,30 +385,6 @@ void typeCommand(redisClient *c) {
     addReplyStatus(c,type);
 }
 
-void saveCommand(redisClient *c) {
-    if (server.bgsavechildpid != -1) {
-        addReplyError(c,"Background save already in progress");
-        return;
-    }
-    if (rdbSave(server.dbfilename) == REDIS_OK) {
-        addReply(c,shared.ok);
-    } else {
-        addReply(c,shared.err);
-    }
-}
-
-void bgsaveCommand(redisClient *c) {
-    if (server.bgsavechildpid != -1) {
-        addReplyError(c,"Background save already in progress");
-        return;
-    }
-    if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
-        addReplyStatus(c,"Background saving started");
-    } else {
-        addReply(c,shared.err);
-    }
-}
-
 void shutdownCommand(redisClient *c) {
     if (prepareForShutdown() == REDIS_OK)
         exit(0);
@@ -450,6 +432,11 @@ void moveCommand(redisClient *c) {
     redisDb *src, *dst;
     int srcid;
 
+    if (server.cluster_enabled) {
+        addReplyError(c,"MOVE is not allowed in cluster mode");
+        return;
+    }
+
     /* Obtain source and target DB pointers */
     src = c->db;
     srcid = c->db->id;
@@ -549,6 +536,8 @@ void propagateExpire(redisDb *db, robj *key) {
 int expireIfNeeded(redisDb *db, robj *key) {
     time_t when = getExpire(db,key);
 
+    if (when < 0) return 0; /* No expire for this key */
+
     /* If we are running in the context of a slave, return ASAP:
      * the slave key expiration is controlled by the master that will
      * send us synthesized DEL operations for expired keys.
@@ -560,8 +549,6 @@ int expireIfNeeded(redisDb *db, robj *key) {
         return time(NULL) > when;
     }
 
-    if (when < 0) return 0;
-
     /* Return when this key has not expired */
     if (time(NULL) <= when) return 0;
 
@@ -614,6 +601,7 @@ void expireatCommand(redisClient *c) {
 void ttlCommand(redisClient *c) {
     time_t expire, ttl = -1;
 
+    if (server.ds_enabled) lookupKeyRead(c->db,c->argv[1]);
     expire = getExpire(c->db,c->argv[1]);
     if (expire != -1) {
         ttl = (expire-time(NULL));
@@ -637,3 +625,76 @@ void persistCommand(redisClient *c) {
         }
     }
 }
+
+/* -----------------------------------------------------------------------------
+ * API to get key arguments from commands
+ * ---------------------------------------------------------------------------*/
+
+int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) {
+    int j, i = 0, last, *keys;
+    REDIS_NOTUSED(argv);
+
+    if (cmd->firstkey == 0) {
+        *numkeys = 0;
+        return NULL;
+    }
+    last = cmd->lastkey;
+    if (last < 0) last = argc+last;
+    keys = zmalloc(sizeof(int)*((last - cmd->firstkey)+1));
+    for (j = cmd->firstkey; j <= last; j += cmd->keystep) {
+        redisAssert(j < argc);
+        keys[i++] = j;
+    }
+    *numkeys = i;
+    return keys;
+}
+
+int *getKeysFromCommand(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
+    if (cmd->getkeys_proc) {
+        return cmd->getkeys_proc(cmd,argv,argc,numkeys,flags);
+    } else {
+        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
+    }
+}
+
+void getKeysFreeResult(int *result) {
+    zfree(result);
+}
+
+int *noPreloadGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
+    if (flags & REDIS_GETKEYS_PRELOAD) {
+        *numkeys = 0;
+        return NULL;
+    } else {
+        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
+    }
+}
+
+int *renameGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
+    if (flags & REDIS_GETKEYS_PRELOAD) {
+        int *keys = zmalloc(sizeof(int));
+        *numkeys = 1;
+        keys[0] = 1;
+        return keys;
+    } else {
+        return getKeysUsingCommandTable(cmd,argv,argc,numkeys);
+    }
+}
+
+int *zunionInterGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) {
+    int i, num, *keys;
+    REDIS_NOTUSED(cmd);
+    REDIS_NOTUSED(flags);
+
+    num = atoi(argv[2]->ptr);
+    /* Sanity check. Don't return any key if the command is going to
+     * reply with syntax error. */
+    if (num > (argc-3)) {
+        *numkeys = 0;
+        return NULL;
+    }
+    keys = zmalloc(sizeof(int)*num);
+    for (i = 0; i < num; i++) keys[i] = 3+i;
+    *numkeys = num;
+    return keys;
+}