handled DEL command as a special optimized case for disk store

src/db.c

@@ -205,10 +205,22 @@ void delCommand(redisClient *c) {
     int deleted = 0, j;
 
     for (j = 1; j < c->argc; j++) {
+        if (server.ds_enabled) {
+            lookupKeyRead(c->db,c->argv[j]);
+            /* FIXME: this can be optimized a lot, no real need to load
+             * a possibly huge value. */
+        }
         if (dbDelete(c->db,c->argv[j])) {
             signalModifiedKey(c->db,c->argv[j]);
             server.dirty++;
             deleted++;
+        } else if (server.ds_enabled) {
+            if (cacheKeyMayExist(c->db,c->argv[j]) &&
+                dsExists(c->db,c->argv[j]))
+            {
+                cacheScheduleForFlush(c->db,c->argv[j]);
+                deleted = 1;
+            }
         }
     }
     addReplyLongLong(c,deleted);

src/diskstore.c

@@ -262,6 +262,10 @@ int dsDel(redisDb *db, robj *key) {
 }
 
 int dsExists(redisDb *db, robj *key) {
+    char buf[1024];
+
+    dsKeyToPath(db,buf,key);
+    return access(buf,R_OK) == 0;
 }
 
 int dsFlushDb(int dbid) {

src/dscache.c

@@ -304,8 +304,10 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             handleClientsBlockedOnSwappedKey(j->db,j->key);
             freeIOJob(j);
         } else if (j->type == REDIS_IOJOB_SAVE) {
-            redisAssert(j->val->storage == REDIS_DS_SAVING);
+            if (j->val) {
-            j->val->storage = REDIS_DS_MEMORY;
+                redisAssert(j->val->storage == REDIS_DS_SAVING);
+                j->val->storage = REDIS_DS_MEMORY;
+            }
             freeIOJob(j);
         }
         processed++;
@@ -362,11 +364,12 @@ void *IOThreadEntryPoint(void *arg) {
             j->val = dsGet(j->db,j->key,&expire);
             if (j->val) j->expire = expire;
         } else if (j->type == REDIS_IOJOB_SAVE) {
-            redisAssert(j->val->storage == REDIS_DS_SAVING);
+            if (j->val) {
-            if (j->val)
+                redisAssert(j->val->storage == REDIS_DS_SAVING);
                 dsSet(j->db,j->key,j->val);
-            else
+            } else {
                 dsDel(j->db,j->key);
+            }
         }
 
         /* Done: insert the job into the processed queue */
@@ -540,6 +543,15 @@ void cacheCron(void) {
     }
 }
 
+/* ============ Negative caching for diskstore objects ====================== */
+/* Since accesses to keys that don't exist with disk store cost us a disk
+ * access, we need to cache names of keys that do not exist but are frequently
+ * accessed. */
+int cacheKeyMayExist(redisDb *db, robj *key) {
+    /* FIXME: for now we just always return true. */
+    return 1;
+}
+
 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
 
 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.

src/redis.h

@@ -808,6 +808,7 @@ void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key);
 int cacheFreeOneEntry(void);
 void cacheScheduleForFlush(redisDb *db, robj *key);
 void cacheCron(void);
+int cacheKeyMayExist(redisDb *db, robj *key);
 
 /* Set data type */
 robj *setTypeCreate(robj *value);