minor changes to doc and comments

[redis.git] / src / diskstore.c

diff --git a/src/diskstore.c b/src/diskstore.c
index 043adc4b0875338930a9d469c246af647c107ad2..06d4ae85b15a8b49207bdfb0f04a01a5d6a5e76a 100644 (file)
--- a/src/diskstore.c
+++ b/src/diskstore.c
@@ -70,6 +70,7 @@
  */
 
 #include "redis.h"
+#include "sha1.h"
 
 #include <fcntl.h>
 #include <sys/stat.h>
@@ -138,16 +139,135 @@ int dsClose(void) {
     return REDIS_OK;
 }
 
+/* Convert key into full path for this object. Dirty but hopefully
+ * is fast enough. */
+void dsKeyToPath(redisDb *db, char *buf, robj *key) {
+    SHA1_CTX ctx;
+    unsigned char hash[20];
+    char hex[40], digits[] = "0123456789abcdef";
+    int j, l;
+
+    SHA1Init(&ctx);
+    SHA1Update(&ctx,key->ptr,sdslen(key->ptr));
+    SHA1Final(hash,&ctx);
+
+    /* Convert the hash into hex format */
+    for (j = 0; j < 20; j++) {
+        hex[j*2] = digits[(hash[j]&0xF0)>>4];
+        hex[(j*2)+1] = digits[hash[j]&0x0F];
+    }
+
+    /* Create the object path. Start with server.ds_path that's the root dir */
+    l = sdslen(server.ds_path);
+    memcpy(buf,server.ds_path,l);
+    buf += l;
+    *buf++ = '/';
+
+    /* Then add xx/yy/ that is the two level directories */
+    buf[0] = hex[0];
+    buf[1] = hex[1];
+    buf[2] = '/';
+    buf[3] = hex[2];
+    buf[4] = hex[3];
+    buf[5] = '/';
+    buf += 6;
+
+    /* Add the database number followed by _ and finall the SHA1 hex */
+    l = ll2string(buf,64,db->id);
+    buf += l;
+    buf[0] = '_';
+    memcpy(buf+1,hex,40);
+    buf[41] = '\0';
+}
+
 int dsSet(redisDb *db, robj *key, robj *val) {
+    char buf[1024];
+    FILE *fp;
+    int retval;
+
+    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. Unlink failing not critical */
+    return REDIS_OK;
 }
 
-robj *dsGet(redisDb *db, robj *key) {
+robj *dsGet(redisDb *db, robj *key, time_t *expire) {
+    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:
+    redisLog(REDIS_WARNING,"Read error reading reading %s. Corrupted key?",
+        buf);
+    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) {
+    char buf[1024];
+
+    dsKeyToPath(db,buf,key);
+    return access(buf,R_OK) == 0;
 }
 
 int dsFlushDb(int dbid) {