FLUSHALL / FLUSHDB for diskstore implemented

src/db.c

@@ -200,8 +200,6 @@ void signalModifiedKey(redisDb *db, robj *key) {
 
 void signalFlushedDb(int dbid) {
     touchWatchedKeysOnFlush(dbid);
-    if (server.ds_enabled)
-        dsFlushDb(dbid);
 }
 
 /*-----------------------------------------------------------------------------
@@ -213,6 +211,7 @@ void flushdbCommand(redisClient *c) {
     signalFlushedDb(c->db->id);
     dictEmpty(c->db->dict);
     dictEmpty(c->db->expires);
+    if (server.ds_enabled) dsFlushDb(c->db->id);
     addReply(c,shared.ok);
 }
 
@@ -224,7 +223,10 @@ void flushallCommand(redisClient *c) {
         kill(server.bgsavechildpid,SIGKILL);
         rdbRemoveTempFile(server.bgsavechildpid);
     }
-    rdbSave(server.dbfilename);
+    if (server.ds_enabled)
+        dsFlushDb(-1);
+    else
+        rdbSave(server.dbfilename);
     server.dirty++;
 }
 

src/debug.c

@@ -262,10 +262,12 @@ void _redisAssert(char *estr, char *file, int line) {
 }
 
 void _redisPanic(char *msg, char *file, int line) {
+    redisLog(REDIS_WARNING,"------------------------------------------------");
     redisLog(REDIS_WARNING,"!!! Software Failure. Press left mouse button to continue");
     redisLog(REDIS_WARNING,"Guru Meditation: %s #%s:%d",msg,file,line);
 #ifdef HAVE_BACKTRACE
     redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)");
+    redisLog(REDIS_WARNING,"------------------------------------------------");
     *((char*)-1) = 'x';
 #endif
 }

src/diskstore.c

@@ -74,6 +74,7 @@
 
 #include <fcntl.h>
 #include <sys/stat.h>
+#include <dirent.h>
 
 int create256dir(char *prefix) {
     char buf[1024];
@@ -270,5 +271,51 @@ int dsExists(redisDb *db, robj *key) {
     return access(buf,R_OK) == 0;
 }
 
-int dsFlushDb(int dbid) {
+void dsFlushOneDir(char *path, int dbid) {
+    DIR *dir;
+    struct dirent *dp, de;
+
+    dir = opendir(path);
+    if (dir == NULL) {
+        redisLog(REDIS_WARNING,"Disk store can't open dir %s: %s",
+            path, strerror(errno));
+        redisPanic("Unrecoverable Disk store errore. Existing.");
+    }
+    while(1) {
+        readdir_r(dir,&de,&dp);
+        if (dp == NULL) break;
+        if (dp->d_name[0] == '.') continue;
+
+        /* Check if we need to remove this entry accordingly to the
+         * DB number */
+        if (dbid != -1) {
+            char id[64];
+            char *p = strchr(dp->d_name,'_');
+            int len = (p - dp->d_name);
+
+            redisAssert(p != NULL && len < 64);
+            memcpy(id,dp->d_name,len);
+            id[len] = '\0';
+            if (atoi(id) != dbid) continue; /* skip this file */
+        }
+        if (unlink(path) == -1) {
+            redisLog(REDIS_WARNING,
+                "Can't unlink %s: %s", path, strerror(errno));
+            redisPanic("Unrecoverable Disk store errore. Existing.");
+        }
+    }
+    closedir(dir);
+}
+
+void dsFlushDb(int dbid) {
+    char buf[1024];
+    int j, i;
+
+    redisLog(REDIS_NOTICE,"Flushing diskstore DB (%d)",dbid);
+    for (j = 0; j < 256; j++) {
+        for (i = 0; i < 256; i++) {
+            snprintf(buf,1024,"%s/%02x/%02x",server.ds_path,j,i);
+            dsFlushOneDir(buf,dbid);
+        }
+    }
 }

src/redis.h

@@ -791,7 +791,7 @@ int dsSet(redisDb *db, robj *key, robj *val);
 robj *dsGet(redisDb *db, robj *key, time_t *expire);
 int dsDel(redisDb *db, robj *key);
 int dsExists(redisDb *db, robj *key);
-int dsFlushDb(int dbid);
+void dsFlushDb(int dbid);
 
 /* Disk Store Cache */
 void dsInit(void);