filesystem based KV store mostly implemented in diskstore.c

src/diskstore.c

@@ -141,10 +141,10 @@ int dsClose(void) {
 
 /* Convert key into full path for this object. Dirty but hopefully
  * is fast enough. */
-void dsKeyToPath(redisDb *db, unsigned char *buf, robj *key) {
+void dsKeyToPath(redisDb *db, char *buf, robj *key) {
     SHA1_CTX ctx;
     unsigned char hash[20];
-    char *hex, digits[] = "0123456789abcdef";
+    char hex[40], digits[] = "0123456789abcdef";
     int j, l;
 
     SHA1Init(&ctx);
@@ -185,20 +185,80 @@ int dsSet(redisDb *db, robj *key, robj *val) {
     FILE *fp;
     int retval;
 
-    dsKeyToPath(buf,key);
+    dsKeyToPath(db,buf,key);
     fp = fopen(buf,"w");
     if ((retval = rdbSaveKeyValuePair(fp,db,key,val,time(NULL))) == -1)
         return REDIS_ERR;
     fclose(fp);
-    if (retval == 0) unlink(buf); /* Expired key */
+    if (retval == 0) unlink(buf); /* Expired key. Unlink failing not critical */
     return REDIS_OK;
 }
 
 robj *dsGet(redisDb *db, robj *key, time_t *expire) {
-    return createStringObject("foo",3);
+    char buf[1024];
+    int type;
+    time_t expiretime = -1; /* -1 means: no expire */
+    robj *dskey; /* Key as loaded from disk. */
+    robj *val;
+    FILE *fp;
+
+    dsKeyToPath(db,buf,key);
+    fp = fopen(buf,"r");
+    if (fp == NULL && errno == ENOENT) return NULL; /* No such key */
+    if (fp == NULL) {
+        redisLog(REDIS_WARNING,"Disk store failed opening %s: %s",
+            buf, strerror(errno));
+        goto readerr;
+    }
+
+    if ((type = rdbLoadType(fp)) == -1) goto readerr;
+    if (type == REDIS_EXPIRETIME) {
+        if ((expiretime = rdbLoadTime(fp)) == -1) goto readerr;
+        /* We read the time so we need to read the object type again */
+        if ((type = rdbLoadType(fp)) == -1) goto readerr;
+    }
+    /* Read key */
+    if ((dskey = rdbLoadStringObject(fp)) == NULL) goto readerr;
+    /* Read value */
+    if ((val = rdbLoadObject(type,fp)) == NULL) goto readerr;
+    fclose(fp);
+
+    /* The key we asked, and the key returned, must be the same */
+    redisAssert(equalStringObjects(key,dskey));
+
+    /* Check if the key already expired */
+    decrRefCount(dskey);
+    if (expiretime != -1 && expiretime < time(NULL)) {
+        decrRefCount(val);
+        unlink(buf); /* This failing is non critical here */
+        return NULL;
+    }
+
+    /* Everything ok... */
+    *expire = expiretime;
+    return val;
+
+readerr:
+    redisPanic("Unrecoverable error reading from disk store");
+    return NULL; /* unreached */
 }
 
 int dsDel(redisDb *db, robj *key) {
+    char buf[1024];
+
+    dsKeyToPath(db,buf,key);
+    if (unlink(buf) == -1) {
+        if (errno == ENOENT) {
+            return REDIS_ERR;
+        } else {
+            redisLog(REDIS_WARNING,"Disk store can't remove %s: %s",
+                buf, strerror(errno));
+            redisPanic("Unrecoverable Disk store errore. Existing.");
+            return REDIS_ERR; /* unreached */
+        }
+    } else {
+        return REDIS_OK;
+    }
 }
 
 int dsExists(redisDb *db, robj *key) {

src/dscache.c

@@ -291,7 +291,7 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             if (j->val != NULL) {
                 dbAdd(j->db,j->key,j->val);
                 incrRefCount(j->val);
-                setExpire(j->db,j->key,j->expire);
+                if (j->expire != -1) setExpire(j->db,j->key,j->expire);
             } else {
                 /* The key does not exist. Create a negative cache entry
                  * for this key. */

src/rdb.c

@@ -404,7 +404,7 @@ int rdbSaveKeyValuePair(FILE *fp, redisDb *db, robj *key, robj *val,
 {
     time_t expiretime;
     
-    expiretime = getExpire(db,&key);
+    expiretime = getExpire(db,key);
 
     /* Save the expire time */
     if (expiretime != -1) {
@@ -415,7 +415,7 @@ int rdbSaveKeyValuePair(FILE *fp, redisDb *db, robj *key, robj *val,
     }
     /* Save type, key, value */
     if (rdbSaveType(fp,val->type) == -1) return -1;
-    if (rdbSaveStringObject(fp,&key) == -1) return -1;
+    if (rdbSaveStringObject(fp,key) == -1) return -1;
     if (rdbSaveObject(fp,val) == -1) return -1;
     return 1;
 }
@@ -459,7 +459,7 @@ int rdbSave(char *filename) {
             robj key, *o = dictGetEntryVal(de);
             
             initStaticStringObject(key,keystr);
-            if (rdbSaveKeyValuePair(fp,db,key,o,now) == -1) goto werr;
+            if (rdbSaveKeyValuePair(fp,db,&key,o,now) == -1) goto werr;
         }
         dictReleaseIterator(di);
     }

src/redis.h

@@ -748,6 +748,9 @@ off_t rdbSavedObjectPages(robj *o);
 robj *rdbLoadObject(int type, FILE *fp);
 void backgroundSaveDoneHandler(int statloc);
 int rdbSaveKeyValuePair(FILE *fp, redisDb *db, robj *key, robj *val, time_t now);
+int rdbLoadType(FILE *fp);
+time_t rdbLoadTime(FILE *fp);
+robj *rdbLoadStringObject(FILE *fp);
 
 /* AOF persistence */
 void flushAppendOnlyFile(void);
@@ -783,7 +786,7 @@ void populateCommandTable(void);
 int dsOpen(void);
 int dsClose(void);
 int dsSet(redisDb *db, robj *key, robj *val);
-robj *dsGet(redisDb *db, robj *key);
+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);