From: antirez <antirez@gmail.com>
Date: Tue, 9 Mar 2010 15:22:50 +0000 (+0100)
Subject: Merged ZREMBYRANK from Pietern
X-Git-Url: https://git.saurik.com/redis.git/commitdiff_plain/ee5cfe385305071ed0ab293be3ad18b13f2f9c86?hp=841053366f12c27dcd225f83c8ae7312326be32e

Merged ZREMBYRANK from Pietern
---

diff --git a/Makefile b/Makefile
index 9f7dde3a..15ae660f 100644
--- a/Makefile
+++ b/Makefile
@@ -14,7 +14,7 @@ endif
 CCOPT= $(CFLAGS) $(CCLINK) $(ARCH) $(PROF)
 DEBUG?= -g -rdynamic -ggdb 
 
-OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o
+OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o
 BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o
 CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o
 
@@ -39,8 +39,9 @@ redis-benchmark.o: redis-benchmark.c fmacros.h ae.h anet.h sds.h adlist.h \
   zmalloc.h
 redis-cli.o: redis-cli.c fmacros.h anet.h sds.h adlist.h zmalloc.h
 redis.o: redis.c fmacros.h config.h redis.h ae.h sds.h anet.h dict.h \
-  adlist.h zmalloc.h lzf.h pqsort.h staticsymbols.h
+  adlist.h zmalloc.h lzf.h pqsort.h zipmap.h staticsymbols.h
 sds.o: sds.c sds.h zmalloc.h
+zipmap.o: zipmap.c zmalloc.h
 zmalloc.o: zmalloc.c config.h
 
 redis-server: $(OBJ)
diff --git a/TODO b/TODO
index d40d5585..f9fa8f2a 100644
--- a/TODO
+++ b/TODO
@@ -59,6 +59,8 @@ BIG ONES:
 
 SMALL ONES:
 
+* If sizeof(double) == sizeof(void*) we could store the double value of sorted sets directly in place of the pointer instead of allocating it in the heap.
+* Delete on writes against expire policy should only happen after argument parsing for commands doing their own arg parsing stuff.
 * Give errors when incrementing a key that does not look like an integer, when providing as a sorted set score something can't be parsed as a double, and so forth.
 * MSADD (n keys) (n values). See this thread in the Redis google group: http://groups.google.com/group/redis-db/browse_thread/thread/e766d84eb375cd41
 * Don't save empty lists / sets / zsets on disk with snapshotting.
diff --git a/redis-benchmark.c b/redis-benchmark.c
index 06057b0d..123d8118 100644
--- a/redis-benchmark.c
+++ b/redis-benchmark.c
@@ -464,7 +464,7 @@ void parseOptions(int argc, char **argv) {
             printf(" -n <requests>      Total number of requests (default 10000)\n");
             printf(" -d <size>          Data size of SET/GET value in bytes (default 2)\n");
             printf(" -k <boolean>       1=keep alive 0=reconnect (default 1)\n");
-            printf(" -r <keyspacelen>   Use random keys for SET/GET/INCR\n");
+            printf(" -r <keyspacelen>   Use random keys for SET/GET/INCR, random values for SADD\n");
             printf("  Using this option the benchmark will get/set keys\n");
             printf("  in the form mykey_rand000000012456 instead of constant\n");
             printf("  keys, the <keyspacelen> argument determines the max\n");
@@ -595,6 +595,24 @@ int main(int argc, char **argv) {
         aeMain(config.el);
         endBenchmark("LPOP");
 
+        prepareForBenchmark();
+        c = createClient();
+        if (!c) exit(1);
+        c->obuf = sdscat(c->obuf,"SADD myset 24\r\ncounter_rand000000000000\r\n");
+        prepareClientForReply(c,REPLY_RETCODE);
+        createMissingClients(c);
+        aeMain(config.el);
+        endBenchmark("SADD");
+
+        prepareForBenchmark();
+        c = createClient();
+        if (!c) exit(1);
+        c->obuf = sdscat(c->obuf,"SPOP myset\r\n");
+        prepareClientForReply(c,REPLY_BULK);
+        createMissingClients(c);
+        aeMain(config.el);
+        endBenchmark("SPOP");
+
         prepareForBenchmark();
         c = createClient();
         if (!c) exit(1);
diff --git a/redis-cli.c b/redis-cli.c
index bbcac0c5..807f676d 100644
--- a/redis-cli.c
+++ b/redis-cli.c
@@ -141,8 +141,10 @@ static struct redisCommand cmdTable[] = {
     {"msetnx",-3,REDIS_CMD_MULTIBULK},
     {"monitor",1,REDIS_CMD_INLINE},
     {"multi",1,REDIS_CMD_INLINE},
-    {"exec",1,REDIS_CMD_MULTIBULK},
+    {"exec",1,REDIS_CMD_INLINE},
     {"discard",1,REDIS_CMD_INLINE},
+    {"hset",4,REDIS_CMD_MULTIBULK},
+    {"hget",3,REDIS_CMD_BULK},
     {NULL,0,0}
 };
 
diff --git a/redis.c b/redis.c
index ab0b9be4..8658bc19 100644
--- a/redis.c
+++ b/redis.c
@@ -75,6 +75,7 @@
 #include "zmalloc.h" /* total memory usage aware version of malloc/free */
 #include "lzf.h"    /* LZF compression library */
 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
+#include "zipmap.h"
 
 /* Error codes */
 #define REDIS_OK                0
@@ -118,9 +119,13 @@
 #define REDIS_ZSET 3
 #define REDIS_HASH 4
 
-/* Objects encoding */
+/* 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. */
 #define REDIS_ENCODING_RAW 0    /* Raw representation */
 #define REDIS_ENCODING_INT 1    /* Encoded as integer */
+#define REDIS_ENCODING_ZIPMAP 2 /* Encoded as zipmap */
+#define REDIS_ENCODING_HT 3     /* Encoded as an hash table */
 
 /* Object types only used for dumping to disk */
 #define REDIS_EXPIRETIME 253
@@ -221,6 +226,10 @@
 #define APPENDFSYNC_ALWAYS 1
 #define APPENDFSYNC_EVERYSEC 2
 
+/* Hashes related defaults */
+#define REDIS_HASH_MAX_ZIPMAP_ENTRIES 64
+#define REDIS_HASH_MAX_ZIPMAP_VALUE 512
+
 /* We can print the stacktrace, so our assert is defined this way: */
 #define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
 static void _redisAssert(char *estr, char *file, int line);
@@ -386,6 +395,9 @@ struct redisServer {
     off_t vm_page_size;
     off_t vm_pages;
     unsigned long long vm_max_memory;
+    /* Hashes config */
+    size_t hash_max_zipmap_entries;
+    size_t hash_max_zipmap_value;
     /* Virtual memory state */
     FILE *vm_fp;
     int vm_fd;
@@ -578,6 +590,7 @@ static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mas
 static struct redisCommand *lookupCommand(char *name);
 static void call(redisClient *c, struct redisCommand *cmd);
 static void resetClient(redisClient *c);
+static void convertToRealHash(robj *o);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -661,6 +674,9 @@ static void brpopCommand(redisClient *c);
 static void appendCommand(redisClient *c);
 static void substrCommand(redisClient *c);
 static void zrankCommand(redisClient *c);
+static void zrevrankCommand(redisClient *c);
+static void hsetCommand(redisClient *c);
+static void hgetCommand(redisClient *c);
 static void zremrangebyrankCommand(redisClient *c);
 
 /*================================= Globals ================================= */
@@ -717,6 +733,9 @@ static struct redisCommand cmdTable[] = {
     {"zcard",zcardCommand,2,REDIS_CMD_INLINE,1,1,1},
     {"zscore",zscoreCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
     {"zrank",zrankCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"zrevrank",zrevrankCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"hset",hsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"hget",hgetCommand,3,REDIS_CMD_BULK,1,1,1},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,1,1},
     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,1,1},
     {"getset",getsetCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
@@ -1013,7 +1032,7 @@ static dictType zsetDictType = {
 };
 
 /* Db->dict */
-static dictType hashDictType = {
+static dictType dbDictType = {
     dictObjHash,                /* hash function */
     NULL,                       /* key dup */
     NULL,                       /* val dup */
@@ -1032,6 +1051,16 @@ static dictType keyptrDictType = {
     NULL                       /* val destructor */
 };
 
+/* Hash type hash table (note that small hashes are represented with zimpaps) */
+static dictType hashDictType = {
+    dictEncObjHash,             /* hash function */
+    NULL,                       /* key dup */
+    NULL,                       /* val dup */
+    dictEncObjKeyCompare,       /* key compare */
+    dictRedisObjectDestructor,  /* key destructor */
+    dictRedisObjectDestructor   /* val destructor */
+};
+
 /* Keylist hash table type has unencoded redis objects as keys and
  * lists as values. It's used for blocking operations (BLPOP) and to
  * map swapped keys to a list of clients waiting for this keys to be loaded. */
@@ -1448,6 +1477,8 @@ static void initServerConfig() {
     server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */
     server.vm_max_threads = 4;
     server.vm_blocked_clients = 0;
+    server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;
+    server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE;
 
     resetServerSaveParams();
 
@@ -1495,7 +1526,7 @@ static void initServer() {
         exit(1);
     }
     for (j = 0; j < server.dbnum; j++) {
-        server.db[j].dict = dictCreate(&hashDictType,NULL);
+        server.db[j].dict = dictCreate(&dbDictType,NULL);
         server.db[j].expires = dictCreate(&keyptrDictType,NULL);
         server.db[j].blockingkeys = dictCreate(&keylistDictType,NULL);
         if (server.vm_enabled)
@@ -1708,6 +1739,12 @@ static void loadServerConfig(char *filename) {
             server.vm_pages = strtoll(argv[1], NULL, 10);
         } else if (!strcasecmp(argv[0],"vm-max-threads") && argc == 2) {
             server.vm_max_threads = strtoll(argv[1], NULL, 10);
+        } else if (!strcasecmp(argv[0],"hash-max-zipmap-entries") && argc == 2){
+            server.hash_max_zipmap_entries = strtol(argv[1], NULL, 10);
+        } else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2){
+            server.hash_max_zipmap_value = strtol(argv[1], NULL, 10);
+        } else if (!strcasecmp(argv[0],"vm-max-threads") && argc == 2) {
+            server.vm_max_threads = strtoll(argv[1], NULL, 10);
         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
@@ -2545,6 +2582,16 @@ static robj *createSetObject(void) {
     return createObject(REDIS_SET,d);
 }
 
+static robj *createHashObject(void) {
+    /* All the Hashes start as zipmaps. Will be automatically converted
+     * into hash tables if there are enough elements or big elements
+     * inside. */
+    unsigned char *zm = zipmapNew();
+    robj *o = createObject(REDIS_HASH,zm);
+    o->encoding = REDIS_ENCODING_ZIPMAP;
+    return o;
+}
+
 static robj *createZsetObject(void) {
     zset *zs = zmalloc(sizeof(*zs));
 
@@ -2576,7 +2623,17 @@ static void freeZsetObject(robj *o) {
 }
 
 static void freeHashObject(robj *o) {
-    dictRelease((dict*) o->ptr);
+    switch (o->encoding) {
+    case REDIS_ENCODING_HT:
+        dictRelease((dict*) o->ptr);
+        break;
+    case REDIS_ENCODING_ZIPMAP:
+        zfree(o->ptr);
+        break;
+    default:
+        redisAssert(0);
+        break;
+    }
 }
 
 static void incrRefCount(robj *o) {
@@ -2881,7 +2938,7 @@ static int rdbSaveLen(FILE *fp, uint32_t len) {
 /* String objects in the form "2391" "-100" without any space and with a
  * range of values that can fit in an 8, 16 or 32 bit signed value can be
  * encoded as integers to save space */
-static int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
+static int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
     long long value;
     char *endptr, buf[32];
 
@@ -2892,7 +2949,7 @@ static int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
 
     /* If the number converted back into a string is not identical
      * then it's not possible to encode the string as integer */
-    if (strlen(buf) != sdslen(s) || memcmp(buf,s,sdslen(s))) return 0;
+    if (strlen(buf) != len || memcmp(buf,s,len)) return 0;
 
     /* Finally check if it fits in our ranges */
     if (value >= -(1<<7) && value <= (1<<7)-1) {
@@ -2916,16 +2973,16 @@ static int rdbTryIntegerEncoding(sds s, unsigned char *enc) {
     }
 }
 
-static int rdbSaveLzfStringObject(FILE *fp, robj *obj) {
-    unsigned int comprlen, outlen;
+static int rdbSaveLzfStringObject(FILE *fp, unsigned char *s, size_t len) {
+    size_t comprlen, outlen;
     unsigned char byte;
     void *out;
 
     /* We require at least four bytes compression for this to be worth it */
-    outlen = sdslen(obj->ptr)-4;
-    if (outlen <= 0) return 0;
+    if (len <= 4) return 0;
+    outlen = len-4;
     if ((out = zmalloc(outlen+1)) == NULL) return 0;
-    comprlen = lzf_compress(obj->ptr, sdslen(obj->ptr), out, outlen);
+    comprlen = lzf_compress(s, len, out, outlen);
     if (comprlen == 0) {
         zfree(out);
         return 0;
@@ -2934,7 +2991,7 @@ static int rdbSaveLzfStringObject(FILE *fp, robj *obj) {
     byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
     if (fwrite(&byte,1,1,fp) == 0) goto writeerr;
     if (rdbSaveLen(fp,comprlen) == -1) goto writeerr;
-    if (rdbSaveLen(fp,sdslen(obj->ptr)) == -1) goto writeerr;
+    if (rdbSaveLen(fp,len) == -1) goto writeerr;
     if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
     zfree(out);
     return comprlen;
@@ -2946,16 +3003,13 @@ writeerr:
 
 /* Save a string objet as [len][data] on disk. If the object is a string
  * representation of an integer value we try to safe it in a special form */
-static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
-    size_t len;
+static int rdbSaveRawString(FILE *fp, unsigned char *s, size_t len) {
     int enclen;
 
-    len = sdslen(obj->ptr);
-
     /* Try integer encoding */
     if (len <= 11) {
         unsigned char buf[5];
-        if ((enclen = rdbTryIntegerEncoding(obj->ptr,buf)) > 0) {
+        if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
             if (fwrite(buf,enclen,1,fp) == 0) return -1;
             return 0;
         }
@@ -2966,7 +3020,7 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
     if (server.rdbcompression && len > 20) {
         int retval;
 
-        retval = rdbSaveLzfStringObject(fp,obj);
+        retval = rdbSaveLzfStringObject(fp,s,len);
         if (retval == -1) return -1;
         if (retval > 0) return 0;
         /* retval == 0 means data can't be compressed, save the old way */
@@ -2974,7 +3028,7 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
 
     /* Store verbatim */
     if (rdbSaveLen(fp,len) == -1) return -1;
-    if (len && fwrite(obj->ptr,len,1,fp) == 0) return -1;
+    if (len && fwrite(s,len,1,fp) == 0) return -1;
     return 0;
 }
 
@@ -2989,10 +3043,10 @@ static int rdbSaveStringObject(FILE *fp, robj *obj) {
      * this in order to avoid bugs) */
     if (obj->encoding != REDIS_ENCODING_RAW) {
         obj = getDecodedObject(obj);
-        retval = rdbSaveStringObjectRaw(fp,obj);
+        retval = rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
         decrRefCount(obj);
     } else {
-        retval = rdbSaveStringObjectRaw(fp,obj);
+        retval = rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
     }
     return retval;
 }
@@ -3070,6 +3124,33 @@ static int rdbSaveObject(FILE *fp, robj *o) {
             if (rdbSaveDoubleValue(fp,*score) == -1) return -1;
         }
         dictReleaseIterator(di);
+    } else if (o->type == REDIS_HASH) {
+        /* Save a hash value */
+        if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+            unsigned char *p = zipmapRewind(o->ptr);
+            unsigned int count = zipmapLen(o->ptr);
+            unsigned char *key, *val;
+            unsigned int klen, vlen;
+
+            if (rdbSaveLen(fp,count) == -1) return -1;
+            while((p = zipmapNext(p,&key,&klen,&val,&vlen)) != NULL) {
+                if (rdbSaveRawString(fp,key,klen) == -1) return -1;
+                if (rdbSaveRawString(fp,val,vlen) == -1) return -1;
+            }
+        } else {
+            dictIterator *di = dictGetIterator(o->ptr);
+            dictEntry *de;
+
+            if (rdbSaveLen(fp,dictSize((dict*)o->ptr)) == -1) return -1;
+            while((de = dictNext(di)) != NULL) {
+                robj *key = dictGetEntryKey(de);
+                robj *val = dictGetEntryVal(de);
+
+                if (rdbSaveStringObject(fp,key) == -1) return -1;
+                if (rdbSaveStringObject(fp,val) == -1) return -1;
+            }
+            dictReleaseIterator(di);
+        }
     } else {
         redisAssert(0 != 0);
     }
@@ -3394,7 +3475,7 @@ static robj *rdbLoadObject(int type, FILE *fp) {
         }
     } else if (type == REDIS_ZSET) {
         /* Read list/set value */
-        uint32_t zsetlen;
+        size_t zsetlen;
         zset *zs;
 
         if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
@@ -3412,6 +3493,46 @@ static robj *rdbLoadObject(int type, FILE *fp) {
             zslInsert(zs->zsl,*score,ele);
             incrRefCount(ele); /* added to skiplist */
         }
+    } else if (type == REDIS_HASH) {
+        size_t hashlen;
+
+        if ((hashlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        o = createHashObject();
+        /* Too many entries? Use an hash table. */
+        if (hashlen > server.hash_max_zipmap_entries)
+            convertToRealHash(o);
+        /* Load every key/value, then set it into the zipmap or hash
+         * table, as needed. */
+        while(hashlen--) {
+            robj *key, *val;
+
+            if ((key = rdbLoadStringObject(fp)) == NULL) return NULL;
+            if ((val = rdbLoadStringObject(fp)) == NULL) return NULL;
+            /* If we are using a zipmap and there are too big values
+             * the object is converted to real hash table encoding. */
+            if (o->encoding != REDIS_ENCODING_HT &&
+               (sdslen(key->ptr) > server.hash_max_zipmap_value ||
+                sdslen(val->ptr) > server.hash_max_zipmap_value))
+            {
+                    convertToRealHash(o);
+            }
+
+            if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+                unsigned char *zm = o->ptr;
+
+                zm = zipmapSet(zm,key->ptr,sdslen(key->ptr),
+                                  val->ptr,sdslen(val->ptr),NULL);
+                o->ptr = zm;
+                decrRefCount(key);
+                decrRefCount(val);
+            } else {
+                tryObjectEncoding(key);
+                tryObjectEncoding(val);
+                dictAdd((dict*)o->ptr,key,val);
+                incrRefCount(key);
+                incrRefCount(val);
+            }
+        }
     } else {
         redisAssert(0 != 0);
     }
@@ -3919,7 +4040,8 @@ static void typeCommand(redisClient *c) {
         case REDIS_LIST: type = "+list"; break;
         case REDIS_SET: type = "+set"; break;
         case REDIS_ZSET: type = "+zset"; break;
-        default: type = "unknown"; break;
+        case REDIS_HASH: type = "+hash"; break;
+        default: type = "+unknown"; break;
         }
     }
     addReplySds(c,sdsnew(type));
@@ -5575,7 +5697,7 @@ static void zscoreCommand(redisClient *c) {
     }
 }
 
-static void zrankCommand(redisClient *c) {
+static void zrankGenericCommand(redisClient *c, int reverse) {
     robj *o;
     o = lookupKeyRead(c->db,c->argv[1]);
     if (o == NULL) {
@@ -5599,13 +5721,119 @@ static void zrankCommand(redisClient *c) {
         double *score = dictGetEntryVal(de);
         rank = zslGetRank(zsl, *score, c->argv[2]);
         if (rank) {
-            addReplyLong(c, rank-1);
+            if (reverse) {
+                addReplyLong(c, zsl->length - rank);
+            } else {
+                addReplyLong(c, rank-1);
+            }
         } else {
             addReply(c,shared.nullbulk);
         }
     }
 }
 
+static void zrankCommand(redisClient *c) {
+    zrankGenericCommand(c, 0);
+}
+
+static void zrevrankCommand(redisClient *c) {
+    zrankGenericCommand(c, 1);
+}
+
+/* =================================== Hashes =============================== */
+static void hsetCommand(redisClient *c) {
+    int update = 0;
+    robj *o = lookupKeyWrite(c->db,c->argv[1]);
+
+    if (o == NULL) {
+        o = createHashObject();
+        dictAdd(c->db->dict,c->argv[1],o);
+        incrRefCount(c->argv[1]);
+    } else {
+        if (o->type != REDIS_HASH) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+    }
+    if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+        unsigned char *zm = o->ptr;
+        robj *valobj = getDecodedObject(c->argv[3]);
+
+        zm = zipmapSet(zm,c->argv[2]->ptr,sdslen(c->argv[2]->ptr),
+            valobj->ptr,sdslen(valobj->ptr),&update);
+        decrRefCount(valobj);
+        o->ptr = zm;
+    } else {
+        if (dictAdd(o->ptr,c->argv[2],c->argv[3]) == DICT_OK) {
+            incrRefCount(c->argv[2]);
+        } else {
+            update = 1;
+        }
+        incrRefCount(c->argv[3]);
+    }
+    server.dirty++;
+    addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",update == 0));
+}
+
+static void hgetCommand(redisClient *c) {
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+
+    if (o == NULL) {
+        addReply(c,shared.nullbulk);
+        return;
+    } else {
+        if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+            unsigned char *zm = o->ptr;
+            unsigned char *val;
+            unsigned int vlen;
+
+            if (zipmapGet(zm,c->argv[2]->ptr,sdslen(c->argv[2]->ptr), &val,&vlen)) {
+                addReplySds(c,sdscatprintf(sdsempty(),"$%u\r\n", vlen));
+                addReplySds(c,sdsnewlen(val,vlen));
+                addReply(c,shared.crlf);
+                return;
+            } else {
+                addReply(c,shared.nullbulk);
+                return;
+            }
+        } else {
+            struct dictEntry *de;
+
+            de = dictFind(o->ptr,c->argv[2]);
+            if (de == NULL) {
+                addReply(c,shared.nullbulk);
+            } else {
+                robj *e = dictGetEntryVal(de);
+
+                addReplyBulkLen(c,e);
+                addReply(c,e);
+                addReply(c,shared.crlf);
+            }
+        }
+    }
+}
+
+static void convertToRealHash(robj *o) {
+    unsigned char *key, *val, *p, *zm = o->ptr;
+    unsigned int klen, vlen;
+    dict *dict = dictCreate(&hashDictType,NULL);
+
+    assert(o->type == REDIS_HASH && o->encoding != REDIS_ENCODING_HT);
+    p = zipmapRewind(zm);
+    while((p = zipmapNext(p,&key,&klen,&val,&vlen)) != NULL) {
+        robj *keyobj, *valobj;
+
+        keyobj = createStringObject((char*)key,klen);
+        valobj = createStringObject((char*)val,vlen);
+        tryObjectEncoding(keyobj);
+        tryObjectEncoding(valobj);
+        dictAdd(dict,keyobj,valobj);
+    }
+    o->encoding = REDIS_ENCODING_HT;
+    o->ptr = dict;
+    zfree(zm);
+}
+
 /* ========================= Non type-specific commands  ==================== */
 
 static void flushdbCommand(redisClient *c) {
@@ -6746,7 +6974,7 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
 
 static int syncWithMaster(void) {
     char buf[1024], tmpfile[256], authcmd[1024];
-    int dumpsize;
+    long dumpsize;
     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
     int dfd;
 
@@ -6798,8 +7026,8 @@ static int syncWithMaster(void) {
         redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
         return REDIS_ERR;
     }
-    dumpsize = atoi(buf+1);
-    redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
+    dumpsize = strtol(buf+1,NULL,10);
+    redisLog(REDIS_NOTICE,"Receiving %ld bytes data dump from MASTER",dumpsize);
     /* Read the bulk write data on a temp file */
     snprintf(tmpfile,256,"temp-%d.%ld.rdb",(int)time(NULL),(long int)random());
     dfd = open(tmpfile,O_CREAT|O_WRONLY,0644);
diff --git a/redis.conf b/redis.conf
index 2923a3aa..0ed593d2 100644
--- a/redis.conf
+++ b/redis.conf
@@ -271,3 +271,10 @@ glueoutputbuf yes
 # your development environment so that we can test it better.
 shareobjects no
 shareobjectspoolsize 1024
+
+# Hashes are encoded in a special way (much more memory efficient) when they
+# have at max a given numer of elements, and the biggest element does not
+# exceed a given threshold. You can configure this limits with the following
+# configuration directives.
+hash-max-zipmap-entries 64
+hash-max-zipmap-value 512
diff --git a/staticsymbols.h b/staticsymbols.h
index 446a5dc4..85f07d32 100644
--- a/staticsymbols.h
+++ b/staticsymbols.h
@@ -5,8 +5,10 @@ static struct redisFunctionSym symsTable[] = {
 {"addReply",(unsigned long)addReply},
 {"addReplyBulkLen",(unsigned long)addReplyBulkLen},
 {"addReplyDouble",(unsigned long)addReplyDouble},
+{"addReplyLong",(unsigned long)addReplyLong},
 {"addReplySds",(unsigned long)addReplySds},
 {"aofRemoveTempFile",(unsigned long)aofRemoveTempFile},
+{"appendCommand",(unsigned long)appendCommand},
 {"appendServerSaveParams",(unsigned long)appendServerSaveParams},
 {"authCommand",(unsigned long)authCommand},
 {"beforeSleep",(unsigned long)beforeSleep},
@@ -23,6 +25,7 @@ static struct redisFunctionSym symsTable[] = {
 {"compareStringObjects",(unsigned long)compareStringObjects},
 {"computeObjectSwappability",(unsigned long)computeObjectSwappability},
 {"createClient",(unsigned long)createClient},
+{"createHashObject",(unsigned long)createHashObject},
 {"createListObject",(unsigned long)createListObject},
 {"createObject",(unsigned long)createObject},
 {"createSetObject",(unsigned long)createSetObject},
@@ -45,6 +48,7 @@ static struct redisFunctionSym symsTable[] = {
 {"dictObjKeyCompare",(unsigned long)dictObjKeyCompare},
 {"dictRedisObjectDestructor",(unsigned long)dictRedisObjectDestructor},
 {"dictVanillaFree",(unsigned long)dictVanillaFree},
+{"discardCommand",(unsigned long)discardCommand},
 {"dontWaitForSwappedKey",(unsigned long)dontWaitForSwappedKey},
 {"dupClientReplyValue",(unsigned long)dupClientReplyValue},
 {"dupStringObject",(unsigned long)dupStringObject},
@@ -75,6 +79,7 @@ static struct redisFunctionSym symsTable[] = {
 {"fwriteBulkDouble",(unsigned long)fwriteBulkDouble},
 {"fwriteBulkLong",(unsigned long)fwriteBulkLong},
 {"genRedisInfoString",(unsigned long)genRedisInfoString},
+{"genericZrangebyscoreCommand",(unsigned long)genericZrangebyscoreCommand},
 {"getCommand",(unsigned long)getCommand},
 {"getDecodedObject",(unsigned long)getDecodedObject},
 {"getExpire",(unsigned long)getExpire},
@@ -84,6 +89,8 @@ static struct redisFunctionSym symsTable[] = {
 {"glueReplyBuffersIfNeeded",(unsigned long)glueReplyBuffersIfNeeded},
 {"handleClientsBlockedOnSwappedKey",(unsigned long)handleClientsBlockedOnSwappedKey},
 {"handleClientsWaitingListPush",(unsigned long)handleClientsWaitingListPush},
+{"hgetCommand",(unsigned long)hgetCommand},
+{"hsetCommand",(unsigned long)hsetCommand},
 {"htNeedsResize",(unsigned long)htNeedsResize},
 {"incrCommand",(unsigned long)incrCommand},
 {"incrDecrCommand",(unsigned long)incrDecrCommand},
@@ -143,8 +150,8 @@ static struct redisFunctionSym symsTable[] = {
 {"rdbSaveLen",(unsigned long)rdbSaveLen},
 {"rdbSaveLzfStringObject",(unsigned long)rdbSaveLzfStringObject},
 {"rdbSaveObject",(unsigned long)rdbSaveObject},
+{"rdbSaveRawString",(unsigned long)rdbSaveRawString},
 {"rdbSaveStringObject",(unsigned long)rdbSaveStringObject},
-{"rdbSaveStringObjectRaw",(unsigned long)rdbSaveStringObjectRaw},
 {"rdbSaveTime",(unsigned long)rdbSaveTime},
 {"rdbSaveType",(unsigned long)rdbSaveType},
 {"rdbSavedObjectLen",(unsigned long)rdbSavedObjectLen},
@@ -196,6 +203,7 @@ static struct redisFunctionSym symsTable[] = {
 {"srandmemberCommand",(unsigned long)srandmemberCommand},
 {"sremCommand",(unsigned long)sremCommand},
 {"stringObjectLen",(unsigned long)stringObjectLen},
+{"substrCommand",(unsigned long)substrCommand},
 {"sunionCommand",(unsigned long)sunionCommand},
 {"sunionDiffGenericCommand",(unsigned long)sunionDiffGenericCommand},
 {"sunionstoreCommand",(unsigned long)sunionstoreCommand},
@@ -240,10 +248,12 @@ static struct redisFunctionSym symsTable[] = {
 {"zaddCommand",(unsigned long)zaddCommand},
 {"zaddGenericCommand",(unsigned long)zaddGenericCommand},
 {"zcardCommand",(unsigned long)zcardCommand},
+{"zcountCommand",(unsigned long)zcountCommand},
 {"zincrbyCommand",(unsigned long)zincrbyCommand},
 {"zrangeCommand",(unsigned long)zrangeCommand},
 {"zrangeGenericCommand",(unsigned long)zrangeGenericCommand},
 {"zrangebyscoreCommand",(unsigned long)zrangebyscoreCommand},
+{"zrankCommand",(unsigned long)zrankCommand},
 {"zremCommand",(unsigned long)zremCommand},
 {"zremrangebyscoreCommand",(unsigned long)zremrangebyscoreCommand},
 {"zrevrangeCommand",(unsigned long)zrevrangeCommand},
diff --git a/test-redis.tcl b/test-redis.tcl
index b1c4da01..9139f544 100644
--- a/test-redis.tcl
+++ b/test-redis.tcl
@@ -1204,6 +1204,10 @@ proc main {server port} {
         list [$r zrank zranktmp x] [$r zrank zranktmp y] [$r zrank zranktmp z]
     } {0 1 2}
 
+    test {ZREVRANK basics} {
+        list [$r zrevrank zranktmp x] [$r zrevrank zranktmp y] [$r zrevrank zranktmp z]
+    } {2 1 0}
+
     test {ZRANK - after deletion} {
         $r zrem zranktmp y
         list [$r zrank zranktmp x] [$r zrank zranktmp z]
diff --git a/zipmap.c b/zipmap.c
index cfdfd56a..5f024dfa 100644
--- a/zipmap.c
+++ b/zipmap.c
@@ -97,6 +97,11 @@
  * comments above, that is, the max number of trailing bytes in a value. */
 #define ZIPMAP_VALUE_MAX_FREE 5
 
+/* The following macro returns the number of bytes needed to encode the length
+ * for the integer value _l, that is, 1 byte for lengths < ZIPMAP_BIGLEN and
+ * 5 bytes for all the other lengths. */
+#define ZIPMAP_LEN_BYTES(_l) (((_l) < ZIPMAP_BIGLEN) ? 1 : sizeof(unsigned int)+1)
+
 /* Create a new empty zipmap. */
 unsigned char *zipmapNew(void) {
     unsigned char *zm = zmalloc(2);
@@ -119,7 +124,7 @@ static unsigned int zipmapDecodeLength(unsigned char *p) {
  * the amount of bytes required to encode such a length. */
 static unsigned int zipmapEncodeLength(unsigned char *p, unsigned int len) {
     if (p == NULL) {
-        return (len < ZIPMAP_BIGLEN) ? 1 : 1+sizeof(unsigned int);
+        return ZIPMAP_LEN_BYTES(len);
     } else {
         if (len < ZIPMAP_BIGLEN) {
             p[0] = len;
@@ -219,23 +224,24 @@ static unsigned int zipmapRawEntryLength(unsigned char *p) {
     return l + zipmapRawValueLength(p+l);
 }
 
-/* Set key to value, creating the key if it does not already exist. */
-unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char *val, unsigned int vlen) {
+/* Set key to value, creating the key if it does not already exist.
+ * If 'update' is not NULL, *update is set to 1 if the key was
+ * already preset, otherwise to 0. */
+unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char *val, unsigned int vlen, int *update) {
     unsigned int oldlen = 0, freeoff = 0, freelen;
     unsigned int reqlen = zipmapRequiredLength(klen,vlen);
     unsigned int empty, vempty;
     unsigned char *p;
    
     freelen = reqlen;
+    if (update) *update = 0;
     p = zipmapLookupRaw(zm,key,klen,&oldlen,&freeoff,&freelen);
     if (p == NULL && freelen == 0) {
-        printf("HERE oldlen:%u required:%u\n",oldlen,reqlen);
         /* Key not found, and not space for the new key. Enlarge */
         zm = zrealloc(zm,oldlen+reqlen);
         p = zm+oldlen-1;
         zm[oldlen+reqlen-1] = ZIPMAP_END;
         freelen = reqlen;
-        printf("New total length is: %u\n", oldlen+reqlen);
     } else if (p == NULL) {
         /* Key not found, but there is enough free space. */
         p = zm+freeoff;
@@ -245,6 +251,7 @@ unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int kle
 
         /* Key found. Is there enough space for the new value? */
         /* Compute the total length: */
+        if (update) *update = 1;
         freelen = zipmapRawKeyLength(b);
         b += freelen;
         freelen += zipmapRawValueLength(b);
@@ -253,7 +260,7 @@ unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int kle
             p[0] = ZIPMAP_EMPTY;
             zipmapEncodeLength(p+1,freelen);
             zm[0] |= ZIPMAP_STATUS_FRAGMENTED;
-            return zipmapSet(zm,key,klen,val,vlen);
+            return zipmapSet(zm,key,klen,val,vlen,NULL);
         }
     }
 
@@ -304,6 +311,67 @@ unsigned char *zipmapDel(unsigned char *zm, unsigned char *key, unsigned int kle
     return zm;
 }
 
+/* Call it before to iterate trought elements via zipmapNext() */
+unsigned char *zipmapRewind(unsigned char *zm) {
+    return zm+1;
+}
+
+/* This function is used to iterate through all the zipmap elements.
+ * In the first call the first argument is the pointer to the zipmap + 1.
+ * In the next calls what zipmapNext returns is used as first argument.
+ * Example:
+ *
+ * unsigned char *i = zipmapRewind(my_zipmap);
+ * while((i = zipmapNext(i,&key,&klen,&value,&vlen)) != NULL) {
+ *     printf("%d bytes key at $p\n", klen, key);
+ *     printf("%d bytes value at $p\n", vlen, value);
+ * }
+ */
+unsigned char *zipmapNext(unsigned char *zm, unsigned char **key, unsigned int *klen, unsigned char **value, unsigned int *vlen) {
+    while(zm[0] == ZIPMAP_EMPTY)
+        zm += zipmapDecodeLength(zm+1);
+    if (zm[0] == ZIPMAP_END) return NULL;
+    if (key) {
+        *key = zm;
+        *klen = zipmapDecodeLength(zm);
+        *key += ZIPMAP_LEN_BYTES(*klen);
+    }
+    zm += zipmapRawKeyLength(zm);
+    if (value) {
+        *value = zm+1;
+        *vlen = zipmapDecodeLength(zm);
+        *value += ZIPMAP_LEN_BYTES(*vlen);
+    }
+    zm += zipmapRawValueLength(zm);
+    return zm;
+}
+
+/* Search a key and retrieve the pointer and len of the associated value.
+ * If the key is found the function returns 1, otherwise 0. */
+int zipmapGet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char **value, unsigned int *vlen) {
+    unsigned char *p;
+
+    if ((p = zipmapLookupRaw(zm,key,klen,NULL,NULL,NULL)) == NULL) return 0;
+    p += zipmapRawKeyLength(p);
+    *vlen = zipmapDecodeLength(p);
+    *value = p + ZIPMAP_LEN_BYTES(*vlen) + 1;
+    return 1;
+}
+
+/* Return 1 if the key exists, otherwise 0 is returned. */
+int zipmapExists(unsigned char *zm, unsigned char *key, unsigned int klen) {
+    return zipmapLookupRaw(zm,key,klen,NULL,NULL,NULL) != NULL;
+}
+
+/* Return the number of entries inside a zipmap */
+unsigned int zipmapLen(unsigned char *zm) {
+    unsigned char *p = zipmapRewind(zm);
+    unsigned int len = 0;
+
+    while((p = zipmapNext(p,NULL,NULL,NULL,NULL)) != NULL) len++;
+    return len;
+}
+
 void zipmapRepr(unsigned char *p) {
     unsigned int l;
 
@@ -341,19 +409,49 @@ void zipmapRepr(unsigned char *p) {
     printf("\n");
 }
 
+#ifdef ZIPMAP_TEST_MAIN
 int main(void) {
     unsigned char *zm;
 
     zm = zipmapNew();
-    zm = zipmapSet(zm,(unsigned char*) "hello",5, (unsigned char*) "world!",6);
-    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "bar",3);
-    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "!",1);
+
+    zm = zipmapSet(zm,(unsigned char*) "name",4, (unsigned char*) "foo",3,NULL);
+    zm = zipmapSet(zm,(unsigned char*) "surname",7, (unsigned char*) "foo",3,NULL);
+    zm = zipmapSet(zm,(unsigned char*) "age",3, (unsigned char*) "foo",3,NULL);
     zipmapRepr(zm);
-    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "12345",5);
+    exit(1);
+
+    zm = zipmapSet(zm,(unsigned char*) "hello",5, (unsigned char*) "world!",6,NULL);
+    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "bar",3,NULL);
+    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "!",1,NULL);
+    zipmapRepr(zm);
+    zm = zipmapSet(zm,(unsigned char*) "foo",3, (unsigned char*) "12345",5,NULL);
     zipmapRepr(zm);
-    zm = zipmapSet(zm,(unsigned char*) "new",3, (unsigned char*) "xx",2);
+    zm = zipmapSet(zm,(unsigned char*) "new",3, (unsigned char*) "xx",2,NULL);
+    zm = zipmapSet(zm,(unsigned char*) "noval",5, (unsigned char*) "",0,NULL);
     zipmapRepr(zm);
     zm = zipmapDel(zm,(unsigned char*) "new",3,NULL);
     zipmapRepr(zm);
+    printf("\nPerform a direct lookup:\n");
+    {
+        unsigned char *value;
+        unsigned int vlen;
+
+        if (zipmapGet(zm,(unsigned char*) "foo",3,&value,&vlen)) {
+            printf("  foo is associated to the %d bytes value: %.*s\n",
+                vlen, vlen, value);
+        }
+    }
+    printf("\nIterate trought elements:\n");
+    {
+        unsigned char *i = zipmapRewind(zm);
+        unsigned char *key, *value;
+        unsigned int klen, vlen;
+
+        while((i = zipmapNext(i,&key,&klen,&value,&vlen)) != NULL) {
+            printf("  %d:%.*s => %d:%.*s\n", klen, klen, key, vlen, vlen, value);
+        }
+    }
     return 0;
 }
+#endif
diff --git a/zipmap.h b/zipmap.h
new file mode 100644
index 00000000..089472ed
--- /dev/null
+++ b/zipmap.h
@@ -0,0 +1,47 @@
+/* String -> String Map data structure optimized for size.
+ *
+ * See zipmap.c for more info.
+ *
+ * --------------------------------------------------------------------------
+ *
+ * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef _ZIMMAP_H
+#define _ZIPMAP_H
+
+unsigned char *zipmapNew(void);
+unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char *val, unsigned int vlen, int *update);
+unsigned char *zipmapDel(unsigned char *zm, unsigned char *key, unsigned int klen, int *deleted);
+unsigned char *zipmapRewind(unsigned char *zm);
+unsigned char *zipmapNext(unsigned char *zm, unsigned char **key, unsigned int *klen, unsigned char **value, unsigned int *vlen);
+int zipmapGet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char **value, unsigned int *vlen);
+int zipmapExists(unsigned char *zm, unsigned char *key, unsigned int klen);
+unsigned int zipmapLen(unsigned char *zm);
+
+#endif