Initial support for key expire times with millisecond resolution. RDB version...

Initial support for key expire times with millisecond resolution. RDB version is now 3, new opcoded added for high resolution times. Redis is still able to correctly load RDB version 2. Tests passing but still a work in progress. API to specify milliseconds expires still missing, but the precision of normal expires is now already improved and working.

src/aof.c

@@ -613,7 +613,7 @@ int rewriteAppendOnlyFile(char *filename) {
                 if (expiretime < now) continue;
                 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
-                if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
+                if (rioWriteBulkLongLong(&aof,expiretime/1000) == 0) goto werr;
             }
         }
         dictReleaseIterator(di);

src/db.c

 #include "redis.h"
 
 #include <signal.h>
+#include <ctype.h>
 
 void SlotToKeyAdd(robj *key);
 void SlotToKeyDel(robj *key);
@@ -334,7 +335,7 @@ void shutdownCommand(redisClient *c) {
 
 void renameGenericCommand(redisClient *c, int nx) {
     robj *o;
-    time_t expire;
+    long long expire;
 
     /* To use the same key as src and dst is probably an error */
     if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) {
@@ -432,18 +433,19 @@ int removeExpire(redisDb *db, robj *key) {
     return dictDelete(db->expires,key->ptr) == DICT_OK;
 }
 
-void setExpire(redisDb *db, robj *key, time_t when) {
-    dictEntry *de;
+void setExpire(redisDb *db, robj *key, long long when) {
+    dictEntry *kde, *de;
 
     /* Reuse the sds from the main dict in the expire dict */
-    de = dictFind(db->dict,key->ptr);
-    redisAssertWithInfo(NULL,key,de != NULL);
-    dictReplace(db->expires,dictGetKey(de),(void*)when);
+    kde = dictFind(db->dict,key->ptr);
+    redisAssertWithInfo(NULL,key,kde != NULL);
+    de = dictReplaceRaw(db->expires,dictGetKey(kde));
+    dictSetSignedIntegerVal(de,when);
 }
 
 /* Return the expire time of the specified key, or -1 if no expire
  * is associated with this key (i.e. the key is non volatile) */
-time_t getExpire(redisDb *db, robj *key) {
+long long getExpire(redisDb *db, robj *key) {
     dictEntry *de;
 
     /* No expire? return ASAP */
@@ -453,7 +455,7 @@ time_t getExpire(redisDb *db, robj *key) {
     /* The entry was found in the expire dict, this means it should also
      * be present in the main dict (safety check). */
     redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL);
-    return (time_t) dictGetVal(de);
+    return dictGetSignedIntegerVal(de);
 }
 
 /* Propagate expires into slaves and the AOF file.
@@ -481,7 +483,7 @@ void propagateExpire(redisDb *db, robj *key) {
 }
 
 int expireIfNeeded(redisDb *db, robj *key) {
-    time_t when = getExpire(db,key);
+    long long when = getExpire(db,key);
 
     if (when < 0) return 0; /* No expire for this key */
 
@@ -500,7 +502,7 @@ int expireIfNeeded(redisDb *db, robj *key) {
     }
 
     /* Return when this key has not expired */
-    if (time(NULL) <= when) return 0;
+    if (mstime() <= when) return 0;
 
     /* Delete the key */
     server.stat_expiredkeys++;
@@ -512,13 +514,28 @@ int expireIfNeeded(redisDb *db, robj *key) {
  * Expires Commands
  *----------------------------------------------------------------------------*/
 
-void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
+void expireGenericCommand(redisClient *c, long long offset) {
     dictEntry *de;
-    long seconds;
+    robj *key = c->argv[1], *param = c->argv[2];
+    long long milliseconds;
+    int time_in_seconds = 1;
 
-    if (getLongFromObjectOrReply(c, param, &seconds, NULL) != REDIS_OK) return;
+    if (getLongLongFromObjectOrReply(c, param, &milliseconds, NULL) != REDIS_OK)
+        return;
 
-    seconds -= offset;
+    /* If no "ms" argument was passed the time is in second, so we need
+     * to multilpy it by 1000 */
+    if (c->argc == 4) {
+        char *arg = c->argv[3]->ptr;
+
+        if (tolower(arg[0]) != 'm' || tolower(arg[1]) != 's' || arg[2]) {
+            addReply(c,shared.syntaxerr);
+            return;
+        }
+        time_in_seconds = 0; /* "ms" argument passed. */
+    }
+    if (time_in_seconds) milliseconds *= 1000;
+    milliseconds -= offset;
 
     de = dictFind(c->db->dict,key->ptr);
     if (de == NULL) {
@@ -531,7 +548,7 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
      *
      * Instead we take the other branch of the IF statement setting an expire
      * (possibly in the past) and wait for an explicit DEL from the master. */
-    if (seconds <= 0 && !server.loading && !server.masterhost) {
+    if (milliseconds <= 0 && !server.loading && !server.masterhost) {
         robj *aux;
 
         redisAssertWithInfo(c,key,dbDelete(c->db,key));
@@ -545,7 +562,7 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
         addReply(c, shared.cone);
         return;
     } else {
-        time_t when = time(NULL)+seconds;
+        long long when = mstime()+milliseconds;
         setExpire(c->db,key,when);
         addReply(c,shared.cone);
         signalModifiedKey(c->db,key);
@@ -555,22 +572,26 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) {
 }
 
 void expireCommand(redisClient *c) {
-    expireGenericCommand(c,c->argv[1],c->argv[2],0);
+    expireGenericCommand(c,0);
 }
 
 void expireatCommand(redisClient *c) {
-    expireGenericCommand(c,c->argv[1],c->argv[2],time(NULL));
+    expireGenericCommand(c,mstime());
 }
 
 void ttlCommand(redisClient *c) {
-    time_t expire, ttl = -1;
+    long long expire, ttl = -1;
 
     expire = getExpire(c->db,c->argv[1]);
     if (expire != -1) {
-        ttl = (expire-time(NULL));
+        ttl = expire-mstime();
         if (ttl < 0) ttl = -1;
     }
-    addReplyLongLong(c,(long long)ttl);
+    if (ttl == -1) {
+        addReplyLongLong(c,-1);
+    } else {
+        addReplyLongLong(c,(ttl+500)/1000);
+    }
 }
 
 void persistCommand(redisClient *c) {

src/rdb.c

@@ -36,6 +36,17 @@ time_t rdbLoadTime(rio *rdb) {
     return (time_t)t32;
 }
 
+int rdbSaveMillisecondTime(rio *rdb, time_t t) {
+    int64_t t64 = (int64_t) t;
+    return rdbWriteRaw(rdb,&t64,8);
+}
+
+long long rdbLoadMillisecondTime(rio *rdb) {
+    int64_t t64;
+    if (rioRead(rdb,&t64,8) == 0) return -1;
+    return (long long)t64;
+}
+
 /* Saves an encoded length. The first two bits in the first byte are used to
  * hold the encoding type. See the REDIS_RDB_* definitions for more information
  * on the types of encoding. */
@@ -563,14 +574,14 @@ off_t rdbSavedObjectLen(robj *o) {
  * On success if the key was actaully saved 1 is returned, otherwise 0
  * is returned (the key was already expired). */
 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
-                        time_t expiretime, time_t now)
+                        long long expiretime, long long now)
 {
     /* Save the expire time */
     if (expiretime != -1) {
         /* If this key is already expired skip it */
         if (expiretime < now) return 0;
-        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME) == -1) return -1;
-        if (rdbSaveTime(rdb,expiretime) == -1) return -1;
+        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
+        if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
     }
 
     /* Save type, key, value */
@@ -586,7 +597,7 @@ int rdbSave(char *filename) {
     dictEntry *de;
     char tmpfile[256];
     int j;
-    time_t now = time(NULL);
+    time_t now = mstime();
     FILE *fp;
     rio rdb;
 
@@ -599,7 +610,7 @@ int rdbSave(char *filename) {
     }
 
     rioInitWithFile(&rdb,fp);
-    if (rdbWriteRaw(&rdb,"REDIS0002",9) == -1) goto werr;
+    if (rdbWriteRaw(&rdb,"REDIS0003",9) == -1) goto werr;
 
     for (j = 0; j < server.dbnum; j++) {
         redisDb *db = server.db+j;
@@ -619,7 +630,7 @@ int rdbSave(char *filename) {
         while((de = dictNext(di)) != NULL) {
             sds keystr = dictGetKey(de);
             robj key, *o = dictGetVal(de);
-            time_t expire;
+            long long expire;
             
             initStaticStringObject(key,keystr);
             expire = getExpire(db,&key);
@@ -942,7 +953,7 @@ int rdbLoad(char *filename) {
     int type, rdbver;
     redisDb *db = server.db+0;
     char buf[1024];
-    time_t expiretime, now = time(NULL);
+    long long expiretime, now = mstime();
     long loops = 0;
     FILE *fp;
     rio rdb;
@@ -962,7 +973,7 @@ int rdbLoad(char *filename) {
         return REDIS_ERR;
     }
     rdbver = atoi(buf+5);
-    if (rdbver < 1 || rdbver > 2) {
+    if (rdbver < 1 || rdbver > 3) {
         fclose(fp);
         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
         errno = EINVAL;
@@ -986,6 +997,15 @@ int rdbLoad(char *filename) {
             if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
             /* We read the time so we need to read the object type again. */
             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+            /* the EXPIREITME opcode specifies time in seconds, so convert
+             * into milliesconds. */
+            expiretime *= 1000;
+        } else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) {
+            /* Milliseconds precision expire times introduced with RDB
+             * version 3. */
+            if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
+            /* We read the time so we need to read the object type again. */
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
         }
 
         if (type == REDIS_RDB_OPCODE_EOF)

src/rdb.h

@@ -52,6 +52,7 @@
 #define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 12))
 
 /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
+#define REDIS_RDB_OPCODE_EXPIRETIME_MS 252
 #define REDIS_RDB_OPCODE_EXPIRETIME 253
 #define REDIS_RDB_OPCODE_SELECTDB   254
 #define REDIS_RDB_OPCODE_EOF        255
@@ -76,7 +77,7 @@ off_t rdbSavedObjectLen(robj *o);
 off_t rdbSavedObjectPages(robj *o);
 robj *rdbLoadObject(int type, rio *rdb);
 void backgroundSaveDoneHandler(int exitcode, int bysignal);
-int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, time_t expireitme, time_t now);
+int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expireitme, long long now);
 robj *rdbLoadStringObject(rio *rdb);
 
 #endif

src/redis.h

@@ -95,11 +95,6 @@
 #define REDIS_ENCODING_INTSET 6  /* Encoded as intset */
 #define REDIS_ENCODING_SKIPLIST 7  /* Encoded as skiplist */
 
-/* Object types only used for dumping to disk */
-#define REDIS_EXPIRETIME 253
-#define REDIS_SELECTDB 254
-#define REDIS_EOF 255
-
 /* Defines related to the dump file format. To store 32 bits lengths for short
  * keys requires a lot of space, so we check the most significant 2 bits of
  * the first byte to interpreter the length:
@@ -941,8 +936,8 @@ void resetServerSaveParams();
 int removeExpire(redisDb *db, robj *key);
 void propagateExpire(redisDb *db, robj *key);
 int expireIfNeeded(redisDb *db, robj *key);
-time_t getExpire(redisDb *db, robj *key);
-void setExpire(redisDb *db, robj *key, time_t when);
+long long getExpire(redisDb *db, robj *key);
+void setExpire(redisDb *db, robj *key, long long when);
 robj *lookupKey(redisDb *db, robj *key);
 robj *lookupKeyRead(redisDb *db, robj *key);
 robj *lookupKeyWrite(redisDb *db, robj *key);

src/t_string.c

@@ -30,7 +30,7 @@ void setGenericCommand(redisClient *c, int nx, robj *key, robj *val, robj *expir
     }
     setKey(c->db,key,val);
     server.dirty++;
-    if (expire) setExpire(c->db,key,time(NULL)+seconds);
+    if (expire) setExpire(c->db,key,(time(NULL)+seconds)*1000);
     addReply(c, nx ? shared.cone : shared.ok);
 }