DB API refactoring. The changes were designed together with Pieter Noordhuis.

src/cluster.c

@@ -1390,7 +1390,7 @@ void restoreCommand(redisClient *c) {
     long ttl;
 
     /* Make sure this key does not already exist here... */
-    if (dbExists(c->db,c->argv[1])) {
+    if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
         addReplyError(c,"Target key name is busy.");
         return;
     }

src/db.c

@@ -86,8 +86,7 @@ robj *lookupKey(redisDb *db, robj *key) {
             redisLog(REDIS_DEBUG,"Force loading key %s via lookup", key->ptr);
             val = dsGet(db,key,&expire);
             if (val) {
-                int retval = dbAdd(db,key,val);
-                redisAssert(retval == REDIS_OK);
+                dbAdd(db,key,val);
                 if (expire != -1) setExpire(db,key,expire);
                 server.stat_keyspace_hits++;
                 return val;
@@ -122,42 +121,47 @@ robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) {
     return o;
 }
 
-/* Add the key to the DB. If the key already exists REDIS_ERR is returned,
- * otherwise REDIS_OK is returned, and the caller should increment the
- * refcount of 'val'. */
-int dbAdd(redisDb *db, robj *key, robj *val) {
-    /* Perform a lookup before adding the key, as we need to copy the
-     * key value. */
-    if (dictFind(db->dict, key->ptr) != NULL) {
-        return REDIS_ERR;
-    } else {
-        sds copy = sdsdup(key->ptr);
-        dictAdd(db->dict, copy, val);
-        if (server.ds_enabled) cacheSetKeyMayExist(db,key);
-        if (server.cluster_enabled) SlotToKeyAdd(key);
-        return REDIS_OK;
-    }
+/* Add the key to the DB. It's up to the caller to increment the reference
+ * counte of the value if needed.
+ *
+ * The program is aborted if the key already exists. */
+void dbAdd(redisDb *db, robj *key, robj *val) {
+    sds copy = sdsdup(key->ptr);
+    int retval = dictAdd(db->dict, copy, val);
+
+    redisAssert(retval == REDIS_OK);
+    if (server.ds_enabled) cacheSetKeyMayExist(db,key);
+    if (server.cluster_enabled) SlotToKeyAdd(key);
+ }
+
+/* Overwrite an existing key with a new value. Incrementing the reference
+ * count of the new value is up to the caller.
+ * This function does not modify the expire time of the existing key.
+ *
+ * The program is aborted if the key was not already present. */
+void dbOverwrite(redisDb *db, robj *key, robj *val) {
+    struct dictEntry *de = dictFind(db->dict,key->ptr);
+    
+    redisAssert(de != NULL);
+    dictReplace(db->dict, key->ptr, val);
+    if (server.ds_enabled) cacheSetKeyMayExist(db,key);
 }
 
-/* If the key does not exist, this is just like dbAdd(). Otherwise
- * the value associated to the key is replaced with the new one.
+/* High level Set operation. This function can be used in order to set
+ * a key, whatever it was existing or not, to a new object.
  *
- * On update (key already existed) 0 is returned. Otherwise 1. */
-int dbReplace(redisDb *db, robj *key, robj *val) {
-    robj *oldval;
-    int retval;
-
-    if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) {
-        sds copy = sdsdup(key->ptr);
-        dictAdd(db->dict, copy, val);
-        if (server.cluster_enabled) SlotToKeyAdd(key);
-        retval = 1;
+ * 1) The ref count of the value object is incremented.
+ * 2) clients WATCHing for the destination key notified.
+ * 3) The expire time of the key is reset (the key is made persistent). */
+void setKey(redisDb *db, robj *key, robj *val) {
+    if (lookupKeyWrite(db,key) == NULL) {
+        dbAdd(db,key,val);
     } else {
-        dictReplace(db->dict, key->ptr, val);
-        retval = 0;
+        dbOverwrite(db,key,val);
     }
-    if (server.ds_enabled) cacheSetKeyMayExist(db,key);
-    return retval;
+    incrRefCount(val);
+    removeExpire(db,key);
+    touchWatchedKey(db,key);
 }
 
 int dbExists(redisDb *db, robj *key) {
@@ -414,13 +418,15 @@ void renameGenericCommand(redisClient *c, int nx) {
         return;
 
     incrRefCount(o);
-    if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) {
+    if (lookupKeyWrite(c->db,c->argv[2]) != NULL) {
         if (nx) {
             decrRefCount(o);
             addReply(c,shared.czero);
             return;
         }
-        dbReplace(c->db,c->argv[2],o);
+        dbOverwrite(c->db,c->argv[2],o);
+    } else {
+        dbAdd(c->db,c->argv[2],o);
     }
     dbDelete(c->db,c->argv[1]);
     signalModifiedKey(c->db,c->argv[1]);
@@ -471,11 +477,12 @@ void moveCommand(redisClient *c) {
         return;
     }
 
-    /* Try to add the element to the target DB */
-    if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) {
+    /* Return zero if the key already exists in the target DB */
+    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
         addReply(c,shared.czero);
         return;
     }
+    dbAdd(dst,c->argv[1],o);
     incrRefCount(o);
 
     /* OK! key moved, free the entry in the source DB */

src/dscache.c

@@ -362,7 +362,8 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             if (j->val != NULL) {
                 /* Note: it's possible that the key is already in memory
                  * due to a blocking load operation. */
-                if (dbAdd(j->db,j->key,j->val) == REDIS_OK) {
+                if (dictFind(j->db->dict,j->key->ptr) == NULL) {
+                    dbAdd(j->db,j->key,j->val);
                     incrRefCount(j->val);
                     if (j->expire != -1) setExpire(j->db,j->key,j->expire);
                 }

src/rdb.c

@@ -918,7 +918,7 @@ void stopLoading(void) {
 int rdbLoad(char *filename) {
     FILE *fp;
     uint32_t dbid;
-    int type, retval, rdbver;
+    int type, rdbver;
     redisDb *db = server.db+0;
     char buf[1024];
     time_t expiretime, now = time(NULL);
@@ -981,11 +981,8 @@ int rdbLoad(char *filename) {
             continue;
         }
         /* Add the new object in the hash table */
-        retval = dbAdd(db,key,val);
-        if (retval == REDIS_ERR) {
-            redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);
-            exit(1);
-        }
+        dbAdd(db,key,val);
+
         /* Set the expire time if needed */
         if (expiretime != -1) setExpire(db,key,expiretime);
 

src/redis.h

@@ -1058,8 +1058,9 @@ robj *lookupKeyRead(redisDb *db, robj *key);
 robj *lookupKeyWrite(redisDb *db, robj *key);
 robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply);
 robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply);
-int dbAdd(redisDb *db, robj *key, robj *val);
-int dbReplace(redisDb *db, robj *key, robj *val);
+void dbAdd(redisDb *db, robj *key, robj *val);
+void dbOverwrite(redisDb *db, robj *key, robj *val);
+void setKey(redisDb *db, robj *key, robj *val);
 int dbExists(redisDb *db, robj *key);
 robj *dbRandomKey(redisDb *db);
 int dbDelete(redisDb *db, robj *key);

src/sort.c

@@ -366,12 +366,12 @@ void sortCommand(redisClient *c) {
                 }
             }
         }
-        dbReplace(c->db,storekey,sobj);
+        setKey(c->db,storekey,sobj);
+        decrRefCount(sobj);
         /* Note: we add 1 because the DB is dirty anyway since even if the
          * SORT result is empty a new key is set and maybe the old content
          * replaced. */
         server.dirty += 1+outputlen;
-        signalModifiedKey(c->db,storekey);
         addReplyLongLong(c,outputlen);
     }
 

src/t_string.c

@@ -13,7 +13,6 @@ static int checkStringLength(redisClient *c, long long size) {
 }
 
 void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expire) {
-    int retval;
     long seconds = 0; /* initialized to avoid an harmness warning */
 
     if (expire) {
@@ -25,21 +24,12 @@ void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expir
         }
     }
 
-    retval = dbAdd(c->db,key,val);
-    if (retval == REDIS_ERR) {
-        if (!nx) {
-            dbReplace(c->db,key,val);
-            incrRefCount(val);
-        } else {
-            addReply(c,shared.czero);
-            return;
-        }
-    } else {
-        incrRefCount(val);
+    if (lookupKeyWrite(c->db,key) != NULL && nx) {
+        addReply(c,shared.czero);
+        return;
     }
-    signalModifiedKey(c->db,key);
+    setKey(c->db,key,val);
     server.dirty++;
-    removeExpire(c->db,key);
     if (expire) setExpire(c->db,key,time(NULL)+seconds);
     addReply(c, nx ? shared.cone : shared.ok);
 }
@@ -81,9 +71,7 @@ void getCommand(redisClient *c) {
 void getsetCommand(redisClient *c) {
     if (getGenericCommand(c) == REDIS_ERR) return;
     c->argv[2] = tryObjectEncoding(c->argv[2]);
-    dbReplace(c->db,c->argv[1],c->argv[2]);
-    incrRefCount(c->argv[2]);
-    signalModifiedKey(c->db,c->argv[1]);
+    setKey(c->db,c->argv[1],c->argv[2]);
     server.dirty++;
     removeExpire(c->db,c->argv[1]);
 }
@@ -138,7 +126,7 @@ void setbitCommand(redisClient *c) {
             robj *decoded = getDecodedObject(o);
             o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
-            dbReplace(c->db,c->argv[1],o);
+            dbOverwrite(c->db,c->argv[1],o);
         }
     }
 
@@ -236,7 +224,7 @@ void setrangeCommand(redisClient *c) {
             robj *decoded = getDecodedObject(o);
             o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
-            dbReplace(c->db,c->argv[1],o);
+            dbOverwrite(c->db,c->argv[1],o);
         }
     }
 
@@ -319,18 +307,15 @@ void msetGenericCommand(redisClient *c, int nx) {
                 busykeys++;
             }
         }
-    }
-    if (busykeys) {
-        addReply(c, shared.czero);
-        return;
+        if (busykeys) {
+            addReply(c, shared.czero);
+            return;
+        }
     }
 
     for (j = 1; j < c->argc; j += 2) {
         c->argv[j+1] = tryObjectEncoding(c->argv[j+1]);
-        dbReplace(c->db,c->argv[j],c->argv[j+1]);
-        incrRefCount(c->argv[j+1]);
-        removeExpire(c->db,c->argv[j]);
-        signalModifiedKey(c->db,c->argv[j]);
+        setKey(c->db,c->argv[j],c->argv[j+1]);
     }
     server.dirty += (c->argc-1)/2;
     addReply(c, nx ? shared.cone : shared.ok);
@@ -346,7 +331,7 @@ void msetnxCommand(redisClient *c) {
 
 void incrDecrCommand(redisClient *c, long long incr) {
     long long value, oldvalue;
-    robj *o;
+    robj *o, *new;
 
     o = lookupKeyWrite(c->db,c->argv[1]);
     if (o != NULL && checkType(c,o,REDIS_STRING)) return;
@@ -358,12 +343,15 @@ void incrDecrCommand(redisClient *c, long long incr) {
         addReplyError(c,"increment or decrement would overflow");
         return;
     }
-    o = createStringObjectFromLongLong(value);
-    dbReplace(c->db,c->argv[1],o);
+    new = createStringObjectFromLongLong(value);
+    if (o)
+        dbOverwrite(c->db,c->argv[1],new);
+    else
+        dbAdd(c->db,c->argv[1],new);
     signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;
     addReply(c,shared.colon);
-    addReply(c,o);
+    addReply(c,new);
     addReply(c,shared.crlf);
 }
 
@@ -416,7 +404,7 @@ void appendCommand(redisClient *c) {
             robj *decoded = getDecodedObject(o);
             o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
-            dbReplace(c->db,c->argv[1],o);
+            dbOverwrite(c->db,c->argv[1],o);
         }
 
         /* Append the value */