Make RDB types/opcodes explicit; load/save object type

5 files changed, 119 insertions, 99 deletions

diff --git a/src/cluster.c b/src/cluster.c
index 1ae84e1ec..1c0b1c23e 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -1383,10 +1383,10 @@ void clusterCommand(redisClient *c) {
 
 /* RESTORE key ttl serialized-value */
 void restoreCommand(redisClient *c) {
-    robj *o;
     long ttl;
     rio payload;
-    unsigned char *data;
+    int type;
+    robj *obj;
 
     /* Make sure this key does not already exist here... */
     if (dbExists(c->db,c->argv[1])) {
@@ -1402,23 +1402,16 @@ void restoreCommand(redisClient *c) {
         return;
     }
 
-    /* Temporary hack to get RDB-aligned payload. */
-    payload = rioInitWithBuffer(sdsnewlen(c->argv[3]->ptr+1, sdslen(c->argv[3]->ptr)-1));
-    data = c->argv[3]->ptr;
-
-    /* Create the object from the serialized dump. */
-    if ((data[0] > 4 && data[0] < 9) ||
-         data[0] > 11 ||
-        (o = rdbLoadObject(data[0],&payload)) == NULL)
+    payload = rioInitWithBuffer(c->argv[3]->ptr);
+    if (((type = rdbLoadObjectType(&payload)) == -1) ||
+        ((obj = rdbLoadObject(type,&payload)) == NULL))
     {
-        addReplyError(c,"Bad data format.");
-        sdsfree(payload.io.buffer.ptr);
+        addReplyError(c,"Bad data format");
         return;
     }
-    sdsfree(payload.io.buffer.ptr);
 
     /* Create the key and set the TTL if any */
-    dbAdd(c->db,c->argv[1],o);
+    dbAdd(c->db,c->argv[1],obj);
     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
     addReply(c,shared.ok);
 }
@@ -1430,7 +1423,6 @@ void migrateCommand(redisClient *c) {
     long dbid;
     time_t ttl;
     robj *o;
-    unsigned char type;
     rio cmd, payload;
 
     /* Sanity check */
@@ -1467,16 +1459,6 @@ void migrateCommand(redisClient *c) {
     redisAssert(rioWriteBulkLongLong(&cmd,dbid));
 
     ttl = getExpire(c->db,c->argv[3]);
-    type = o->type;
-    if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
-        type = REDIS_LIST_ZIPLIST;
-    else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
-        type = REDIS_HASH_ZIPMAP;
-    else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
-        type = REDIS_SET_INTSET;
-    else
-        type = o->type;
-
     redisAssert(rioWriteBulkCount(&cmd,'*',4));
     redisAssert(rioWriteBulkString(&cmd,"RESTORE",7));
     redisAssert(c->argv[3]->encoding == REDIS_ENCODING_RAW);
@@ -1486,7 +1468,7 @@ void migrateCommand(redisClient *c) {
     /* Finally the last argument that is the serailized object payload
      * in the form: <type><rdb-serialized-object>. */
     payload = rioInitWithBuffer(sdsempty());
-    redisAssert(rioWrite(&payload,&type,1));
+    redisAssert(rdbSaveObjectType(&payload,o));
     redisAssert(rdbSaveObject(&payload,o) != -1);
     redisAssert(rioWriteBulkString(&cmd,payload.io.buffer.ptr,sdslen(payload.io.buffer.ptr)));
     sdsfree(payload.io.buffer.ptr);
@@ -1552,38 +1534,22 @@ socket_rd_err:
  * complement of RESTORE and can be useful for different applications. */
 void dumpCommand(redisClient *c) {
     robj *o, *dumpobj;
-    sds dump = NULL;
     rio payload;
-    unsigned int type;
 
     /* Check if the key is here. */
     if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
         addReply(c,shared.nullbulk);
         return;
     }
-    
-    /* Dump the serailized object and read it back in memory. We prefix it
-     * with a one byte containing the type ID.  This is the serialization
-     * format understood by RESTORE. */
-    payload = rioInitWithBuffer(sdsempty());
-    redisAssert(rdbSaveObject(&payload,o)); /* always write >= 1 bytes. */
-    dump = sdsnewlen(NULL,sdslen(payload.io.buffer.ptr)+1);
-    memcpy(dump+1,payload.io.buffer.ptr,sdslen(payload.io.buffer.ptr));
-    sdsfree(payload.io.buffer.ptr);
 
-    type = o->type;
-    if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
-        type = REDIS_LIST_ZIPLIST;
-    else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
-        type = REDIS_HASH_ZIPMAP;
-    else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
-        type = REDIS_SET_INTSET;
-    else
-        type = o->type;
-    dump[0] = type;
+    /* Serialize the object in a RDB-like format. It consist of an object type
+     * byte followed by the serialized object. This is understood by RESTORE. */
+    payload = rioInitWithBuffer(sdsempty());
+    redisAssert(rdbSaveObjectType(&payload,o));
+    redisAssert(rdbSaveObject(&payload,o));
 
     /* Transfer to the client */
-    dumpobj = createObject(REDIS_STRING,dump);
+    dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
     addReplyBulk(c,dumpobj);
     decrRefCount(dumpobj);
     return;
diff --git a/src/diskstore.c b/src/diskstore.c
index 9e45305f8..046e476db 100644
--- a/src/diskstore.c
+++ b/src/diskstore.c
@@ -242,7 +242,7 @@ robj *dsGet(redisDb *db, robj *key, time_t *expire) {
 
     rdb = rioInitWithFile(fp);
     if ((type = rdbLoadType(&rdb)) == -1) goto readerr;
-    if (type == REDIS_EXPIRETIME) {
+    if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
         if ((expiretime = rdbLoadTime(&rdb)) == -1) goto readerr;
         /* We read the time so we need to read the object type again */
         if ((type = rdbLoadType(&rdb)) == -1) goto readerr;
@@ -416,7 +416,7 @@ void *dsRdbSave_thread(void *arg) {
                 dbid = dsGetDbidFromFilename(dp->d_name);
                 if (dbid != last_dbid) {
                     last_dbid = dbid;
-                    if (rdbSaveType(&rdb,REDIS_SELECTDB) == -1) goto werr;
+                    if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
                     if (rdbSaveLen(&rdb,dbid) == -1) goto werr;
                 }
 
@@ -453,7 +453,7 @@ void *dsRdbSave_thread(void *arg) {
     }
     
     /* Output the end of file opcode */
-    if (rdbSaveType(&rdb,REDIS_EOF) == -1) goto werr;
+    if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
 
     /* Make sure data will not remain on the OS's output buffers */
     fflush(fp);
diff --git a/src/rdb.c b/src/rdb.c
index 1e1df2ad9..fdb04754f 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -366,6 +366,53 @@ off_t rdbSavedObjectLen(robj *o) {
     return len;
 }
 
+/* Save the object type of object "o". */
+int rdbSaveObjectType(rio *rdb, robj *o) {
+    switch (o->type) {
+    case REDIS_STRING:
+        return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
+    case REDIS_LIST:
+        if (o->encoding == REDIS_ENCODING_ZIPLIST)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_ZIPLIST);
+        else if (o->encoding == REDIS_ENCODING_LINKEDLIST)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST);
+        else
+            redisPanic("Unknown list encoding");
+    case REDIS_SET:
+        if (o->encoding == REDIS_ENCODING_INTSET)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET);
+        else if (o->encoding == REDIS_ENCODING_HT)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_SET);
+        else
+            redisPanic("Unknown set encoding");
+    case REDIS_ZSET:
+        if (o->encoding == REDIS_ENCODING_ZIPLIST)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST);
+        else if (o->encoding == REDIS_ENCODING_SKIPLIST)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET);
+        else
+            redisPanic("Unknown sorted set encoding");
+    case REDIS_HASH:
+        if (o->encoding == REDIS_ENCODING_ZIPMAP)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPMAP);
+        else if (o->encoding == REDIS_ENCODING_HT)
+            return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH);
+        else
+            redisPanic("Unknown hash encoding");
+    default:
+        redisPanic("Unknown object type");
+    }
+    return -1; /* avoid warning */
+}
+
+/* Load object type. Return -1 when the byte doesn't contain an object type. */
+int rdbLoadObjectType(rio *rdb) {
+    int type;
+    if ((type = rdbLoadType(rdb)) == -1) return -1;
+    if (!rdbIsObjectType(type)) return -1;
+    return type;
+}
+
 /* Save a key-value pair, with expire time, type, key, value.
  * On error -1 is returned.
  * On success if the key was actaully saved 1 is returned, otherwise 0
@@ -373,28 +420,16 @@ off_t rdbSavedObjectLen(robj *o) {
 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
                         time_t expiretime, time_t now)
 {
-    int vtype;
-
     /* Save the expire time */
     if (expiretime != -1) {
         /* If this key is already expired skip it */
         if (expiretime < now) return 0;
-        if (rdbSaveType(rdb,REDIS_EXPIRETIME) == -1) return -1;
+        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME) == -1) return -1;
         if (rdbSaveTime(rdb,expiretime) == -1) return -1;
     }
-    /* Fix the object type if needed, to support saving zipmaps, ziplists,
-     * and intsets, directly as blobs of bytes: they are already serialized. */
-    vtype = val->type;
-    if (vtype == REDIS_HASH && val->encoding == REDIS_ENCODING_ZIPMAP)
-        vtype = REDIS_HASH_ZIPMAP;
-    else if (vtype == REDIS_LIST && val->encoding == REDIS_ENCODING_ZIPLIST)
-        vtype = REDIS_LIST_ZIPLIST;
-    else if (vtype == REDIS_SET && val->encoding == REDIS_ENCODING_INTSET)
-        vtype = REDIS_SET_INTSET;
-    else if (vtype == REDIS_ZSET && val->encoding == REDIS_ENCODING_ZIPLIST)
-        vtype = REDIS_ZSET_ZIPLIST;
+
     /* Save type, key, value */
-    if (rdbSaveType(rdb,vtype) == -1) return -1;
+    if (rdbSaveObjectType(rdb,val) == -1) return -1;
     if (rdbSaveStringObject(rdb,key) == -1) return -1;
     if (rdbSaveObject(rdb,val) == -1) return -1;
     return 1;
@@ -437,7 +472,7 @@ int rdbSave(char *filename) {
         }
 
         /* Write the SELECT DB opcode */
-        if (rdbSaveType(&rdb,REDIS_SELECTDB) == -1) goto werr;
+        if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
         if (rdbSaveLen(&rdb,j) == -1) goto werr;
 
         /* Iterate this DB writing every entry */
@@ -453,7 +488,7 @@ int rdbSave(char *filename) {
         dictReleaseIterator(di);
     }
     /* EOF opcode */
-    if (rdbSaveType(&rdb,REDIS_EOF) == -1) goto werr;
+    if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
 
     /* Make sure data will not remain on the OS's output buffers */
     fflush(fp);
@@ -672,17 +707,17 @@ int rdbLoadDoubleValue(rio *rdb, double *val) {
 
 /* Load a Redis object of the specified type from the specified file.
  * On success a newly allocated object is returned, otherwise NULL. */
-robj *rdbLoadObject(int type, rio *rdb) {
+robj *rdbLoadObject(int rdbtype, rio *rdb) {
     robj *o, *ele, *dec;
     size_t len;
     unsigned int i;
 
-    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,rdb->tell(rdb));
-    if (type == REDIS_STRING) {
+    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",rdbtype,rdb->tell(rdb));
+    if (rdbtype == REDIS_RDB_TYPE_STRING) {
         /* Read string value */
         if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
         o = tryObjectEncoding(o);
-    } else if (type == REDIS_LIST) {
+    } else if (rdbtype == REDIS_RDB_TYPE_LIST) {
         /* Read list value */
         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 
@@ -714,7 +749,7 @@ robj *rdbLoadObject(int type, rio *rdb) {
                 listAddNodeTail(o->ptr,ele);
             }
         }
-    } else if (type == REDIS_SET) {
+    } else if (rdbtype == REDIS_RDB_TYPE_SET) {
         /* Read list/set value */
         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 
@@ -753,7 +788,7 @@ robj *rdbLoadObject(int type, rio *rdb) {
                 decrRefCount(ele);
             }
         }
-    } else if (type == REDIS_ZSET) {
+    } else if (rdbtype == REDIS_RDB_TYPE_ZSET) {
         /* Read list/set value */
         size_t zsetlen;
         size_t maxelelen = 0;
@@ -787,7 +822,7 @@ robj *rdbLoadObject(int type, rio *rdb) {
         if (zsetLength(o) <= server.zset_max_ziplist_entries &&
             maxelelen <= server.zset_max_ziplist_value)
                 zsetConvert(o,REDIS_ENCODING_ZIPLIST);
-    } else if (type == REDIS_HASH) {
+    } else if (rdbtype == REDIS_RDB_TYPE_HASH) {
         size_t hashlen;
 
         if ((hashlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
@@ -833,10 +868,10 @@ robj *rdbLoadObject(int type, rio *rdb) {
                 dictAdd((dict*)o->ptr,key,val);
             }
         }
-    } else if (type == REDIS_HASH_ZIPMAP ||
-               type == REDIS_LIST_ZIPLIST ||
-               type == REDIS_SET_INTSET ||
-               type == REDIS_ZSET_ZIPLIST)
+    } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP  ||
+               rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
+               rdbtype == REDIS_RDB_TYPE_SET_INTSET   ||
+               rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST)
     {
         robj *aux = rdbLoadStringObject(rdb);
 
@@ -852,26 +887,26 @@ robj *rdbLoadObject(int type, rio *rdb) {
          * type. Note that we only check the length and not max element
          * size as this is an O(N) scan. Eventually everything will get
          * converted. */
-        switch(type) {
-            case REDIS_HASH_ZIPMAP:
+        switch(rdbtype) {
+            case REDIS_RDB_TYPE_HASH_ZIPMAP:
                 o->type = REDIS_HASH;
                 o->encoding = REDIS_ENCODING_ZIPMAP;
                 if (zipmapLen(o->ptr) > server.hash_max_zipmap_entries)
                     convertToRealHash(o);
                 break;
-            case REDIS_LIST_ZIPLIST:
+            case REDIS_RDB_TYPE_LIST_ZIPLIST:
                 o->type = REDIS_LIST;
                 o->encoding = REDIS_ENCODING_ZIPLIST;
                 if (ziplistLen(o->ptr) > server.list_max_ziplist_entries)
                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
                 break;
-            case REDIS_SET_INTSET:
+            case REDIS_RDB_TYPE_SET_INTSET:
                 o->type = REDIS_SET;
                 o->encoding = REDIS_ENCODING_INTSET;
                 if (intsetLen(o->ptr) > server.set_max_intset_entries)
                     setTypeConvert(o,REDIS_ENCODING_HT);
                 break;
-            case REDIS_ZSET_ZIPLIST:
+            case REDIS_RDB_TYPE_ZSET_ZIPLIST:
                 o->type = REDIS_ZSET;
                 o->encoding = REDIS_ENCODING_ZIPLIST;
                 if (zsetLength(o) > server.zset_max_ziplist_entries)
@@ -952,14 +987,17 @@ int rdbLoad(char *filename) {
 
         /* Read type. */
         if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
-        if (type == REDIS_EXPIRETIME) {
+        if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
             if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
-            /* We read the time so we need to read the object type again */
+            /* We read the time so we need to read the object type again. */
             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
         }
-        if (type == REDIS_EOF) break;
+
+        if (type == REDIS_RDB_OPCODE_EOF)
+            break;
+
         /* Handle SELECT DB opcode as a special case */
-        if (type == REDIS_SELECTDB) {
+        if (type == REDIS_RDB_OPCODE_SELECTDB) {
             if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
                 goto eoferr;
             if (dbid >= (unsigned)server.dbnum) {
diff --git a/src/rdb.h b/src/rdb.h
index a4a8c096f..93185fc37 100644
--- a/src/rdb.h
+++ b/src/rdb.h
@@ -7,6 +7,31 @@
 /* TBD: include only necessary headers. */
 #include "redis.h"
 
+/* Dup object types to RDB object types. Only reason is readability (are we
+ * dealing with RDB types or with in-memory object types?). */
+#define REDIS_RDB_TYPE_STRING 0
+#define REDIS_RDB_TYPE_LIST   1
+#define REDIS_RDB_TYPE_SET    2
+#define REDIS_RDB_TYPE_ZSET   3
+#define REDIS_RDB_TYPE_HASH   4
+
+/* Object types for encoded objects. */
+#define REDIS_RDB_TYPE_HASH_ZIPMAP    9
+#define REDIS_RDB_TYPE_LIST_ZIPLIST  10
+#define REDIS_RDB_TYPE_SET_INTSET    11
+#define REDIS_RDB_TYPE_ZSET_ZIPLIST  12
+
+/* Test if a type is an object type. */
+#define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 12))
+
+/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
+#define REDIS_RDB_OPCODE_EXPIRETIME 253
+#define REDIS_RDB_OPCODE_SELECTDB   254
+#define REDIS_RDB_OPCODE_EOF        255
+
+/* Test if a type is an opcode. */
+#define rdbIsOpcode(t) (t >= 253 && t <= 255)
+
 int rdbLoad(char *filename);
 int rdbSaveBackground(char *filename);
 void rdbRemoveTempFile(pid_t childpid);
@@ -22,5 +47,7 @@ time_t rdbLoadTime(rio *rdb);
 robj *rdbLoadStringObject(rio *rdb);
 int rdbSaveType(rio *rdb, unsigned char type);
 int rdbSaveLen(rio *rdb, uint32_t len);
+int rdbSaveObjectType(rio *rdb, robj *o);
+int rdbLoadObjectType(rio *rdb);
 
 #endif
diff --git a/src/redis.h b/src/redis.h
index a172bca4a..5cf9dcad6 100644
--- a/src/redis.h
+++ b/src/redis.h
@@ -73,12 +73,6 @@
 #define REDIS_HASH 4
 #define REDIS_VMPOINTER 8
 
-/* Object types only used for persistence in .rdb files */
-#define REDIS_HASH_ZIPMAP 9
-#define REDIS_LIST_ZIPLIST 10
-#define REDIS_SET_INTSET 11
-#define REDIS_ZSET_ZIPLIST 12
-
 /* Objects encoding. Some kind of objects like Strings and Hashes can be
  * internally represented in multiple ways. The 'encoding' field of the object
  * is set to one of this fields for this object. */
@@ -91,11 +85,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: