zipmap to hash conversion in HSET

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index 21c547c0560e908898772940948ed9b85c87c7f4..b45b86184a51b46afed6a4b19d6803ceff7359c7 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -1,5 +1,5 @@
 /*
 /*

- * Copyright (c) 2006-2009, Salvatore Sanfilippo <antirez at gmail dot com>
+ * 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
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without


@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
  * POSSIBILITY OF SUCH DAMAGE.
  */
 

-#define REDIS_VERSION "1.3.3"
+#define REDIS_VERSION "1.3.4"

 
 #include "fmacros.h"
 #include "config.h"
 
 #include "fmacros.h"
 #include "config.h"


@@ -38,6 +38,7 @@
 #include <time.h>
 #include <unistd.h>
 #define __USE_POSIX199309
 #include <time.h>
 #include <unistd.h>
 #define __USE_POSIX199309

+#define __USE_UNIX98

 #include <signal.h>
 
 #ifdef HAVE_BACKTRACE
 #include <signal.h>
 
 #ifdef HAVE_BACKTRACE


@@ -74,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 "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
 
 /* Error codes */
 #define REDIS_OK                0


@@ -117,9 +119,17 @@
 #define REDIS_ZSET 3
 #define REDIS_HASH 4
 
 #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_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 */
+
+static char* strencoding[] = {
+    "raw", "int", "zipmap", "hashtable"
+};

 
 /* Object types only used for dumping to disk */
 #define REDIS_EXPIRETIME 253
 
 /* Object types only used for dumping to disk */
 #define REDIS_EXPIRETIME 253


@@ -172,13 +182,12 @@
 #define REDIS_MAX_COMPLETED_JOBS_PROCESSED 1
 
 /* Client flags */
 #define REDIS_MAX_COMPLETED_JOBS_PROCESSED 1
 
 /* Client flags */

-#define REDIS_CLOSE 1       /* This client connection should be closed ASAP */
-#define REDIS_SLAVE 2       /* This client is a slave server */
-#define REDIS_MASTER 4      /* This client is a master server */
-#define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */
-#define REDIS_MULTI 16      /* This client is in a MULTI context */
-#define REDIS_BLOCKED 32    /* The client is waiting in a blocking operation */
-#define REDIS_IO_WAIT 64    /* The client is waiting for Virtual Memory I/O */
+#define REDIS_SLAVE 1       /* This client is a slave server */
+#define REDIS_MASTER 2      /* This client is a master server */
+#define REDIS_MONITOR 4     /* This client is a slave monitor, see MONITOR */
+#define REDIS_MULTI 8       /* This client is in a MULTI context */
+#define REDIS_BLOCKED 16    /* The client is waiting in a blocking operation */
+#define REDIS_IO_WAIT 32    /* The client is waiting for Virtual Memory I/O */

 
 /* Slave replication state - slave side */
 #define REDIS_REPL_NONE 0   /* No active replication */
 
 /* Slave replication state - slave side */
 #define REDIS_REPL_NONE 0   /* No active replication */


@@ -221,8 +230,12 @@
 #define APPENDFSYNC_ALWAYS 1
 #define APPENDFSYNC_EVERYSEC 2
 
 #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: */
 /* We can print the stacktrace, so our assert is defined this way: */

-#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),exit(1)))
+#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))

 static void _redisAssert(char *estr, char *file, int line);
 
 /*================================= Data types ============================== */
 static void _redisAssert(char *estr, char *file, int line);
 
 /*================================= Data types ============================== */


@@ -269,6 +282,7 @@ typedef struct redisDb {
     dict *dict;                 /* The keyspace for this DB */
     dict *expires;              /* Timeout of keys with a timeout set */
     dict *blockingkeys;         /* Keys with clients waiting for data (BLPOP) */
     dict *dict;                 /* The keyspace for this DB */
     dict *expires;              /* Timeout of keys with a timeout set */
     dict *blockingkeys;         /* Keys with clients waiting for data (BLPOP) */

+    dict *io_keys;              /* Keys with clients waiting for VM I/O */

     int id;
 } redisDb;
 
     int id;
 } redisDb;
 


@@ -298,8 +312,7 @@ typedef struct redisClient {
     list *reply;
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */
     list *reply;
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */

-    int flags;              /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
-                            /* REDIS_MULTI */
+    int flags;              /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */

     int slaveseldb;         /* slave selected db, if this client is a slave */
     int authenticated;      /* when requirepass is non-NULL */
     int replstate;          /* replication state if this is a slave */
     int slaveseldb;         /* slave selected db, if this client is a slave */
     int authenticated;      /* when requirepass is non-NULL */
     int replstate;          /* replication state if this is a slave */


@@ -307,7 +320,7 @@ typedef struct redisClient {
     long repldboff;         /* replication DB file offset */
     off_t repldbsize;       /* replication DB file size */
     multiState mstate;      /* MULTI/EXEC state */
     long repldboff;         /* replication DB file offset */
     off_t repldbsize;       /* replication DB file size */
     multiState mstate;      /* MULTI/EXEC state */

-    robj **blockingkeys;    /* The key we waiting to terminate a blocking
+    robj **blockingkeys;    /* The key we are waiting to terminate a blocking

                              * operation such as BLPOP. Otherwise NULL. */
     int blockingkeysnum;    /* Number of blocking keys */
     time_t blockingto;      /* Blocking operation timeout. If UNIX current time
                              * operation such as BLPOP. Otherwise NULL. */
     int blockingkeysnum;    /* Number of blocking keys */
     time_t blockingto;      /* Blocking operation timeout. If UNIX current time


@@ -373,7 +386,8 @@ struct redisServer {
     int replstate;
     unsigned int maxclients;
     unsigned long long maxmemory;
     int replstate;
     unsigned int maxclients;
     unsigned long long maxmemory;

-    unsigned int blockedclients;
+    unsigned int blpop_blocked_clients;
+    unsigned int vm_blocked_clients;

     /* Sort parameters - qsort_r() is only available under BSD so we
      * have to take this state global, in order to pass it to sortCompare() */
     int sort_desc;
     /* Sort parameters - qsort_r() is only available under BSD so we
      * have to take this state global, in order to pass it to sortCompare() */
     int sort_desc;


@@ -385,6 +399,9 @@ struct redisServer {
     off_t vm_page_size;
     off_t vm_pages;
     unsigned long long vm_max_memory;
     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;
     /* Virtual memory state */
     FILE *vm_fp;
     int vm_fd;


@@ -399,7 +416,7 @@ struct redisServer {
     list *io_newjobs; /* List of VM I/O jobs yet to be processed */
     list *io_processing; /* List of VM I/O jobs being processed */
     list *io_processed; /* List of VM I/O jobs already processed */
     list *io_newjobs; /* List of VM I/O jobs yet to be processed */
     list *io_processing; /* List of VM I/O jobs being processed */
     list *io_processed; /* List of VM I/O jobs already processed */

-    list *io_clients; /* All the clients waiting for SWAP I/O operations */
+    list *io_ready_clients; /* Clients ready to be unblocked. All keys loaded */

     pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
     pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
     pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
     pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
     pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
     pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */


@@ -426,6 +443,10 @@ struct redisCommand {
     redisCommandProc *proc;
     int arity;
     int flags;
     redisCommandProc *proc;
     int arity;
     int flags;

+    /* What keys should be loaded in background when calling this command? */
+    int vm_firstkey; /* The first argument that's a key (0 = no keys) */
+    int vm_lastkey;  /* THe last argument that's a key */
+    int vm_keystep;  /* The step between first and last key */

 };
 
 struct redisFunctionSym {
 };
 
 struct redisFunctionSym {


@@ -451,6 +472,7 @@ typedef struct _redisSortOperation {
 typedef struct zskiplistNode {
     struct zskiplistNode **forward;
     struct zskiplistNode *backward;
 typedef struct zskiplistNode {
     struct zskiplistNode **forward;
     struct zskiplistNode *backward;

+    unsigned int *span;

     double score;
     robj *obj;
 } zskiplistNode;
     double score;
     robj *obj;
 } zskiplistNode;


@@ -487,7 +509,7 @@ static double R_Zero, R_PosInf, R_NegInf, R_Nan;
 #define REDIS_IOJOB_LOAD 0          /* Load from disk to memory */
 #define REDIS_IOJOB_PREPARE_SWAP 1  /* Compute needed pages */
 #define REDIS_IOJOB_DO_SWAP 2       /* Swap from memory to disk */
 #define REDIS_IOJOB_LOAD 0          /* Load from disk to memory */
 #define REDIS_IOJOB_PREPARE_SWAP 1  /* Compute needed pages */
 #define REDIS_IOJOB_DO_SWAP 2       /* Swap from memory to disk */

-typedef struct iojon {
+typedef struct iojob {

     int type;   /* Request type, REDIS_IOJOB_* */
     redisDb *db;/* Redis database */
     robj *key;  /* This I/O request is about swapping this key */
     int type;   /* Request type, REDIS_IOJOB_* */
     redisDb *db;/* Redis database */
     robj *key;  /* This I/O request is about swapping this key */


@@ -565,6 +587,14 @@ static robj *vmReadObjectFromSwap(off_t page, int type);
 static void waitEmptyIOJobsQueue(void);
 static void vmReopenSwapFile(void);
 static int vmFreePage(off_t page);
 static void waitEmptyIOJobsQueue(void);
 static void vmReopenSwapFile(void);
 static int vmFreePage(off_t page);

+static int blockClientOnSwappedKeys(struct redisCommand *cmd, redisClient *c);
+static int dontWaitForSwappedKey(redisClient *c, robj *key);
+static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key);
+static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
+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);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);


@@ -634,6 +664,7 @@ static void zaddCommand(redisClient *c);
 static void zincrbyCommand(redisClient *c);
 static void zrangeCommand(redisClient *c);
 static void zrangebyscoreCommand(redisClient *c);
 static void zincrbyCommand(redisClient *c);
 static void zrangeCommand(redisClient *c);
 static void zrangebyscoreCommand(redisClient *c);

+static void zcountCommand(redisClient *c);

 static void zrevrangeCommand(redisClient *c);
 static void zcardCommand(redisClient *c);
 static void zremCommand(redisClient *c);
 static void zrevrangeCommand(redisClient *c);
 static void zcardCommand(redisClient *c);
 static void zremCommand(redisClient *c);


@@ -641,93 +672,116 @@ static void zscoreCommand(redisClient *c);
 static void zremrangebyscoreCommand(redisClient *c);
 static void multiCommand(redisClient *c);
 static void execCommand(redisClient *c);
 static void zremrangebyscoreCommand(redisClient *c);
 static void multiCommand(redisClient *c);
 static void execCommand(redisClient *c);

+static void discardCommand(redisClient *c);

 static void blpopCommand(redisClient *c);
 static void brpopCommand(redisClient *c);
 static void blpopCommand(redisClient *c);
 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 hdelCommand(redisClient *c);
+static void zremrangebyrankCommand(redisClient *c);
+static void zunionCommand(redisClient *c);
+static void zinterCommand(redisClient *c);

 
 /*================================= Globals ================================= */
 
 /* Global vars */
 static struct redisServer server; /* server global state */
 static struct redisCommand cmdTable[] = {
 
 /*================================= Globals ================================= */
 
 /* Global vars */
 static struct redisServer server; /* server global state */
 static struct redisCommand cmdTable[] = {

-    {"get",getCommand,2,REDIS_CMD_INLINE},
-    {"set",setCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"setnx",setnxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"del",delCommand,-2,REDIS_CMD_INLINE},
-    {"exists",existsCommand,2,REDIS_CMD_INLINE},
-    {"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"decr",decrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"mget",mgetCommand,-2,REDIS_CMD_INLINE},
-    {"rpush",rpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"rpop",rpopCommand,2,REDIS_CMD_INLINE},
-    {"lpop",lpopCommand,2,REDIS_CMD_INLINE},
-    {"brpop",brpopCommand,-3,REDIS_CMD_INLINE},
-    {"blpop",blpopCommand,-3,REDIS_CMD_INLINE},
-    {"llen",llenCommand,2,REDIS_CMD_INLINE},
-    {"lindex",lindexCommand,3,REDIS_CMD_INLINE},
-    {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
-    {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
-    {"lrem",lremCommand,4,REDIS_CMD_BULK},
-    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"srem",sremCommand,3,REDIS_CMD_BULK},
-    {"smove",smoveCommand,4,REDIS_CMD_BULK},
-    {"sismember",sismemberCommand,3,REDIS_CMD_BULK},
-    {"scard",scardCommand,2,REDIS_CMD_INLINE},
-    {"spop",spopCommand,2,REDIS_CMD_INLINE},
-    {"srandmember",srandmemberCommand,2,REDIS_CMD_INLINE},
-    {"sinter",sinterCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sunion",sunionCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sdiff",sdiffCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
-    {"zadd",zaddCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"zincrby",zincrbyCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"zrem",zremCommand,3,REDIS_CMD_BULK},
-    {"zremrangebyscore",zremrangebyscoreCommand,4,REDIS_CMD_INLINE},
-    {"zrange",zrangeCommand,-4,REDIS_CMD_INLINE},
-    {"zrangebyscore",zrangebyscoreCommand,-4,REDIS_CMD_INLINE},
-    {"zrevrange",zrevrangeCommand,-4,REDIS_CMD_INLINE},
-    {"zcard",zcardCommand,2,REDIS_CMD_INLINE},
-    {"zscore",zscoreCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"getset",getsetCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"mset",msetCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"msetnx",msetnxCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
-    {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE},
-    {"select",selectCommand,2,REDIS_CMD_INLINE},
-    {"move",moveCommand,3,REDIS_CMD_INLINE},
-    {"rename",renameCommand,3,REDIS_CMD_INLINE},
-    {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE},
-    {"expire",expireCommand,3,REDIS_CMD_INLINE},
-    {"expireat",expireatCommand,3,REDIS_CMD_INLINE},
-    {"keys",keysCommand,2,REDIS_CMD_INLINE},
-    {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE},
-    {"auth",authCommand,2,REDIS_CMD_INLINE},
-    {"ping",pingCommand,1,REDIS_CMD_INLINE},
-    {"echo",echoCommand,2,REDIS_CMD_BULK},
-    {"save",saveCommand,1,REDIS_CMD_INLINE},
-    {"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE},
-    {"bgrewriteaof",bgrewriteaofCommand,1,REDIS_CMD_INLINE},
-    {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE},
-    {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE},
-    {"type",typeCommand,2,REDIS_CMD_INLINE},
-    {"multi",multiCommand,1,REDIS_CMD_INLINE},
-    {"exec",execCommand,1,REDIS_CMD_INLINE},
-    {"sync",syncCommand,1,REDIS_CMD_INLINE},
-    {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE},
-    {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
-    {"sort",sortCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
-    {"info",infoCommand,1,REDIS_CMD_INLINE},
-    {"monitor",monitorCommand,1,REDIS_CMD_INLINE},
-    {"ttl",ttlCommand,2,REDIS_CMD_INLINE},
-    {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE},
-    {"debug",debugCommand,-2,REDIS_CMD_INLINE},
-    {NULL,NULL,0,0}
+    {"get",getCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"set",setCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,0,0,0},
+    {"setnx",setnxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,0,0,0},
+    {"append",appendCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"substr",substrCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"del",delCommand,-2,REDIS_CMD_INLINE,0,0,0},
+    {"exists",existsCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,1,1},
+    {"decr",decrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,1,1},
+    {"mget",mgetCommand,-2,REDIS_CMD_INLINE,1,-1,1},
+    {"rpush",rpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"rpop",rpopCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"lpop",lpopCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"brpop",brpopCommand,-3,REDIS_CMD_INLINE,1,1,1},
+    {"blpop",blpopCommand,-3,REDIS_CMD_INLINE,1,1,1},
+    {"llen",llenCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"lindex",lindexCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"lrange",lrangeCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"lrem",lremCommand,4,REDIS_CMD_BULK,1,1,1},
+    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,2,1},
+    {"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"srem",sremCommand,3,REDIS_CMD_BULK,1,1,1},
+    {"smove",smoveCommand,4,REDIS_CMD_BULK,1,2,1},
+    {"sismember",sismemberCommand,3,REDIS_CMD_BULK,1,1,1},
+    {"scard",scardCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"spop",spopCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"srandmember",srandmemberCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"sinter",sinterCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,-1,1},
+    {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,2,-1,1},
+    {"sunion",sunionCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,-1,1},
+    {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,2,-1,1},
+    {"sdiff",sdiffCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,-1,1},
+    {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,2,-1,1},
+    {"smembers",sinterCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"zadd",zaddCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"zincrby",zincrbyCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,1,1},
+    {"zrem",zremCommand,3,REDIS_CMD_BULK,1,1,1},
+    {"zremrangebyscore",zremrangebyscoreCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"zremrangebyrank",zremrangebyrankCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"zunion",zunionCommand,-4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,0,0,0},
+    {"zinter",zinterCommand,-4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,0,0,0},
+    {"zrange",zrangeCommand,-4,REDIS_CMD_INLINE,1,1,1},
+    {"zrangebyscore",zrangebyscoreCommand,-4,REDIS_CMD_INLINE,1,1,1},
+    {"zcount",zcountCommand,4,REDIS_CMD_INLINE,1,1,1},
+    {"zrevrange",zrevrangeCommand,-4,REDIS_CMD_INLINE,1,1,1},
+    {"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},
+    {"hdel",hdelCommand,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},
+    {"mset",msetCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,-1,2},
+    {"msetnx",msetnxCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,1,-1,2},
+    {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"select",selectCommand,2,REDIS_CMD_INLINE,0,0,0},
+    {"move",moveCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"rename",renameCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE,1,1,1},
+    {"expire",expireCommand,3,REDIS_CMD_INLINE,0,0,0},
+    {"expireat",expireatCommand,3,REDIS_CMD_INLINE,0,0,0},
+    {"keys",keysCommand,2,REDIS_CMD_INLINE,0,0,0},
+    {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"auth",authCommand,2,REDIS_CMD_INLINE,0,0,0},
+    {"ping",pingCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"echo",echoCommand,2,REDIS_CMD_BULK,0,0,0},
+    {"save",saveCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"bgrewriteaof",bgrewriteaofCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"type",typeCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"multi",multiCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"exec",execCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"discard",discardCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"sync",syncCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"flushall",flushallCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"sort",sortCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,1,1,1},
+    {"info",infoCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"monitor",monitorCommand,1,REDIS_CMD_INLINE,0,0,0},
+    {"ttl",ttlCommand,2,REDIS_CMD_INLINE,1,1,1},
+    {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE,0,0,0},
+    {"debug",debugCommand,-2,REDIS_CMD_INLINE,0,0,0},
+    {NULL,NULL,0,0,0,0,0}

 };
 
 /*============================ Utility functions ============================ */
 };
 
 /*============================ Utility functions ============================ */


@@ -864,7 +918,7 @@ static void redisLog(int level, const char *fmt, ...) {
 
     va_start(ap, fmt);
     if (level >= server.verbosity) {
 
     va_start(ap, fmt);
     if (level >= server.verbosity) {

-        char *c = ".-*";
+        char *c = ".-*#";

         char buf[64];
         time_t now;
 
         char buf[64];
         time_t now;
 


@@ -947,10 +1001,24 @@ static int dictEncObjKeyCompare(void *privdata, const void *key1,
 static unsigned int dictEncObjHash(const void *key) {
     robj *o = (robj*) key;
 
 static unsigned int dictEncObjHash(const void *key) {
     robj *o = (robj*) key;
 

-    o = getDecodedObject(o);
-    unsigned int hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
-    decrRefCount(o);
-    return hash;
+    if (o->encoding == REDIS_ENCODING_RAW) {
+        return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
+    } else {
+        if (o->encoding == REDIS_ENCODING_INT) {
+            char buf[32];
+            int len;
+
+            len = snprintf(buf,32,"%ld",(long)o->ptr);
+            return dictGenHashFunction((unsigned char*)buf, len);
+        } else {
+            unsigned int hash;
+
+            o = getDecodedObject(o);
+            hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
+            decrRefCount(o);
+            return hash;
+        }
+    }

 }
 
 /* Sets type and expires */
 }
 
 /* Sets type and expires */


@@ -974,7 +1042,7 @@ static dictType zsetDictType = {
 };
 
 /* Db->dict */
 };
 
 /* Db->dict */

-static dictType hashDictType = {
+static dictType dbDictType = {

     dictObjHash,                /* hash function */
     NULL,                       /* key dup */
     NULL,                       /* val dup */
     dictObjHash,                /* hash function */
     NULL,                       /* key dup */
     NULL,                       /* val dup */


@@ -993,8 +1061,19 @@ static dictType keyptrDictType = {
     NULL                       /* val destructor */
 };
 
     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
 /* Keylist hash table type has unencoded redis objects as keys and

- * lists as values. It's used for blocking operations (BLPOP) */
+ * 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. */

 static dictType keylistDictType = {
     dictObjHash,                /* hash function */
     NULL,                       /* key dup */
 static dictType keylistDictType = {
     dictObjHash,                /* hash function */
     NULL,                       /* key dup */


@@ -1195,7 +1274,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
     }
 
     /* Close connections of timedout clients */
     }
 
     /* Close connections of timedout clients */

-    if ((server.maxidletime && !(loops % 10)) || server.blockedclients)
+    if ((server.maxidletime && !(loops % 10)) || server.blpop_blocked_clients)

         closeTimedoutClients();
 
     /* Check if a background saving or AOF rewrite in progress terminated */
         closeTimedoutClients();
 
     /* Check if a background saving or AOF rewrite in progress terminated */


@@ -1294,6 +1373,38 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
     return 1000;
 }
 
     return 1000;
 }
 

+/* This function gets called every time Redis is entering the
+ * main loop of the event driven library, that is, before to sleep
+ * for ready file descriptors. */
+static void beforeSleep(struct aeEventLoop *eventLoop) {
+    REDIS_NOTUSED(eventLoop);
+
+    if (server.vm_enabled && listLength(server.io_ready_clients)) {
+        listIter li;
+        listNode *ln;
+
+        listRewind(server.io_ready_clients,&li);
+        while((ln = listNext(&li))) {
+            redisClient *c = ln->value;
+            struct redisCommand *cmd;
+
+            /* Resume the client. */
+            listDelNode(server.io_ready_clients,ln);
+            c->flags &= (~REDIS_IO_WAIT);
+            server.vm_blocked_clients--;
+            aeCreateFileEvent(server.el, c->fd, AE_READABLE,
+                readQueryFromClient, c);
+            cmd = lookupCommand(c->argv[0]->ptr);
+            assert(cmd != NULL);
+            call(c,cmd);
+            resetClient(c);
+            /* There may be more data to process in the input buffer. */
+            if (c->querybuf && sdslen(c->querybuf) > 0)
+                processInputBuffer(c);
+        }
+    }
+}
+

 static void createSharedObjects(void) {
     shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
     shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
 static void createSharedObjects(void) {
     shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
     shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));


@@ -1367,7 +1478,7 @@ static void initServerConfig() {
     server.rdbcompression = 1;
     server.sharingpoolsize = 1024;
     server.maxclients = 0;
     server.rdbcompression = 1;
     server.sharingpoolsize = 1024;
     server.maxclients = 0;

-    server.blockedclients = 0;
+    server.blpop_blocked_clients = 0;

     server.maxmemory = 0;
     server.vm_enabled = 0;
     server.vm_swap_file = zstrdup("/tmp/redis-%p.vm");
     server.maxmemory = 0;
     server.vm_enabled = 0;
     server.vm_swap_file = zstrdup("/tmp/redis-%p.vm");


@@ -1375,6 +1486,9 @@ static void initServerConfig() {
     server.vm_pages = 1024*1024*100;    /* 104 millions of pages */
     server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */
     server.vm_max_threads = 4;
     server.vm_pages = 1024*1024*100;    /* 104 millions of pages */
     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();
 
 
     resetServerSaveParams();
 


@@ -1422,9 +1536,11 @@ static void initServer() {
         exit(1);
     }
     for (j = 0; j < server.dbnum; j++) {
         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);
         server.db[j].expires = dictCreate(&keyptrDictType,NULL);
         server.db[j].blockingkeys = dictCreate(&keylistDictType,NULL);

+        if (server.vm_enabled)
+            server.db[j].io_keys = dictCreate(&keylistDictType,NULL);

         server.db[j].id = j;
     }
     server.cronloops = 0;
         server.db[j].id = j;
     }
     server.cronloops = 0;


@@ -1633,6 +1749,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);
             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;
         }
         } else {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }


@@ -1685,11 +1807,17 @@ static void freeClient(redisClient *c) {
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);

-    /* Remove from the list of clients waiting for VM operations */
-    if (server.vm_enabled && listLength(c->io_keys)) {
-        ln = listSearchKey(server.io_clients,c);
-        if (ln) listDelNode(server.io_clients,ln);
-        listRelease(c->io_keys);
+    /* Remove from the list of clients waiting for swapped keys */
+    if (c->flags & REDIS_IO_WAIT && listLength(c->io_keys) == 0) {
+        ln = listSearchKey(server.io_ready_clients,c);
+        if (ln) {
+            listDelNode(server.io_ready_clients,ln);
+            server.vm_blocked_clients--;
+        }
+    }
+    while (server.vm_enabled && listLength(c->io_keys)) {
+        ln = listFirst(c->io_keys);
+        dontWaitForSwappedKey(c,ln->value);

     }
     listRelease(c->io_keys);
     /* Other cleanup */
     }
     listRelease(c->io_keys);
     /* Other cleanup */


@@ -2002,6 +2130,9 @@ static int processCommand(redisClient *c) {
         freeClient(c);
         return 0;
     }
         freeClient(c);
         return 0;
     }

+
+    /* Now lookup the command and check ASAP about trivial error conditions
+     * such wrong arity, bad command name and so forth. */

     cmd = lookupCommand(c->argv[0]->ptr);
     if (!cmd) {
         addReplySds(c,
     cmd = lookupCommand(c->argv[0]->ptr);
     if (!cmd) {
         addReplySds(c,


@@ -2022,6 +2153,7 @@ static int processCommand(redisClient *c) {
         resetClient(c);
         return 1;
     } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
         resetClient(c);
         return 1;
     } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {

+        /* This is a bulk command, we have to read the last argument yet. */

         int bulklen = atoi(c->argv[c->argc-1]->ptr);
 
         decrRefCount(c->argv[c->argc-1]);
         int bulklen = atoi(c->argv[c->argc-1]->ptr);
 
         decrRefCount(c->argv[c->argc-1]);


@@ -2043,6 +2175,8 @@ static int processCommand(redisClient *c) {
             c->argc++;
             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
         } else {
             c->argc++;
             c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
         } else {

+            /* Otherwise return... there is to read the last argument
+             * from the socket. */

             return 1;
         }
     }
             return 1;
         }
     }


@@ -2064,18 +2198,16 @@ static int processCommand(redisClient *c) {
     }
 
     /* Exec the command */
     }
 
     /* Exec the command */

-    if (c->flags & REDIS_MULTI && cmd->proc != execCommand) {
+    if (c->flags & REDIS_MULTI && cmd->proc != execCommand && cmd->proc != discardCommand) {

         queueMultiCommand(c,cmd);
         addReply(c,shared.queued);
     } else {
         queueMultiCommand(c,cmd);
         addReply(c,shared.queued);
     } else {

+        if (server.vm_enabled && server.vm_max_threads > 0 &&
+            blockClientOnSwappedKeys(cmd,c)) return 1;

         call(c,cmd);
     }
 
     /* Prepare the client for the next command */
         call(c,cmd);
     }
 
     /* Prepare the client for the next command */

-    if (c->flags & REDIS_CLOSE) {
-        freeClient(c);
-        return 0;
-    }

     resetClient(c);
     return 1;
 }
     resetClient(c);
     return 1;
 }


@@ -2259,7 +2391,8 @@ static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mas
     } else {
         return;
     }
     } else {
         return;
     }

-    processInputBuffer(c);
+    if (!(c->flags & REDIS_BLOCKED))
+        processInputBuffer(c);

 }
 
 static int selectDb(redisClient *c, int id) {
 }
 
 static int selectDb(redisClient *c, int id) {


@@ -2271,7 +2404,7 @@ static int selectDb(redisClient *c, int id) {
 
 static void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);
 
 static void *dupClientReplyValue(void *o) {
     incrRefCount((robj*)o);

-    return 0;
+    return o;

 }
 
 static redisClient *createClient(int fd) {
 }
 
 static redisClient *createClient(int fd) {


@@ -2339,6 +2472,14 @@ static void addReplyDouble(redisClient *c, double d) {
         (unsigned long) strlen(buf),buf));
 }
 
         (unsigned long) strlen(buf),buf));
 }
 

+static void addReplyLong(redisClient *c, long l) {
+    char buf[128];
+    size_t len;
+
+    len = snprintf(buf,sizeof(buf),":%ld\r\n",l);
+    addReplySds(c,sdsnewlen(buf,len));
+}
+

 static void addReplyBulkLen(redisClient *c, robj *obj) {
     size_t len;
 
 static void addReplyBulkLen(redisClient *c, robj *obj) {
     size_t len;
 


@@ -2451,6 +2592,16 @@ static robj *createSetObject(void) {
     return createObject(REDIS_SET,d);
 }
 
     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));
 
 static robj *createZsetObject(void) {
     zset *zs = zmalloc(sizeof(*zs));
 


@@ -2482,7 +2633,17 @@ static void freeZsetObject(robj *o) {
 }
 
 static void freeHashObject(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) {
 }
 
 static void incrRefCount(robj *o) {


@@ -2550,10 +2711,16 @@ static robj *lookupKey(redisDb *db, robj *key) {
                 /* Update the access time of the key for the aging algorithm. */
                 key->vm.atime = server.unixtime;
             } else {
                 /* Update the access time of the key for the aging algorithm. */
                 key->vm.atime = server.unixtime;
             } else {

+                int notify = (key->storage == REDIS_VM_LOADING);
+

                 /* Our value was swapped on disk. Bring it at home. */
                 redisAssert(val == NULL);
                 val = vmLoadObject(key);
                 dictGetEntryVal(de) = val;
                 /* Our value was swapped on disk. Bring it at home. */
                 redisAssert(val == NULL);
                 val = vmLoadObject(key);
                 dictGetEntryVal(de) = val;

+
+                /* Clients blocked by the VM subsystem may be waiting for
+                 * this key... */
+                if (notify) handleClientsBlockedOnSwappedKey(db,key);

             }
         }
         return val;
             }
         }
         return val;


@@ -2781,7 +2948,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 */
 /* 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];
 
     long long value;
     char *endptr, buf[32];
 


@@ -2792,7 +2959,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 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) {
 
     /* Finally check if it fits in our ranges */
     if (value >= -(1<<7) && value <= (1<<7)-1) {


@@ -2816,16 +2983,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 */
     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;
     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;
     if (comprlen == 0) {
         zfree(out);
         return 0;


@@ -2834,7 +3001,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;
     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;
     if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
     zfree(out);
     return comprlen;


@@ -2846,16 +3013,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 */
 
 /* 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;
 
     int enclen;
 

-    len = sdslen(obj->ptr);
-

     /* Try integer encoding */
     if (len <= 11) {
         unsigned char buf[5];
     /* 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;
         }
             if (fwrite(buf,enclen,1,fp) == 0) return -1;
             return 0;
         }


@@ -2866,7 +3030,7 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
     if (server.rdbcompression && len > 20) {
         int retval;
 
     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 */
         if (retval == -1) return -1;
         if (retval > 0) return 0;
         /* retval == 0 means data can't be compressed, save the old way */


@@ -2874,7 +3038,7 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
 
     /* Store verbatim */
     if (rdbSaveLen(fp,len) == -1) return -1;
 
     /* 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;
 }
 
     return 0;
 }
 


@@ -2889,10 +3053,10 @@ static int rdbSaveStringObject(FILE *fp, robj *obj) {
      * this in order to avoid bugs) */
     if (obj->encoding != REDIS_ENCODING_RAW) {
         obj = getDecodedObject(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 {
         decrRefCount(obj);
     } else {

-        retval = rdbSaveStringObjectRaw(fp,obj);
+        retval = rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));

     }
     return retval;
 }
     }
     return retval;
 }


@@ -2970,6 +3134,33 @@ static int rdbSaveObject(FILE *fp, robj *o) {
             if (rdbSaveDoubleValue(fp,*score) == -1) return -1;
         }
         dictReleaseIterator(di);
             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);
     }
     } else {
         redisAssert(0 != 0);
     }


@@ -3104,9 +3295,9 @@ static int rdbSaveBackground(char *filename) {
         if (server.vm_enabled) vmReopenSwapFile();
         close(server.fd);
         if (rdbSave(filename) == REDIS_OK) {
         if (server.vm_enabled) vmReopenSwapFile();
         close(server.fd);
         if (rdbSave(filename) == REDIS_OK) {

-            exit(0);
+            _exit(0);

         } else {
         } else {

-            exit(1);
+            _exit(1);

         }
     } else {
         /* Parent */
         }
     } else {
         /* Parent */


@@ -3266,6 +3457,7 @@ static int rdbLoadDoubleValue(FILE *fp, double *val) {
 static robj *rdbLoadObject(int type, FILE *fp) {
     robj *o;
 
 static robj *rdbLoadObject(int type, FILE *fp) {
     robj *o;
 

+    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));

     if (type == REDIS_STRING) {
         /* Read string value */
         if ((o = rdbLoadStringObject(fp)) == NULL) return NULL;
     if (type == REDIS_STRING) {
         /* Read string value */
         if ((o = rdbLoadStringObject(fp)) == NULL) return NULL;


@@ -3276,6 +3468,10 @@ static robj *rdbLoadObject(int type, FILE *fp) {
 
         if ((listlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
         o = (type == REDIS_LIST) ? createListObject() : createSetObject();
 
         if ((listlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
         o = (type == REDIS_LIST) ? createListObject() : createSetObject();

+        /* It's faster to expand the dict to the right size asap in order
+         * to avoid rehashing */
+        if (type == REDIS_SET && listlen > DICT_HT_INITIAL_SIZE)
+            dictExpand(o->ptr,listlen);

         /* Load every single element of the list/set */
         while(listlen--) {
             robj *ele;
         /* Load every single element of the list/set */
         while(listlen--) {
             robj *ele;


@@ -3290,7 +3486,7 @@ static robj *rdbLoadObject(int type, FILE *fp) {
         }
     } else if (type == REDIS_ZSET) {
         /* Read list/set value */
         }
     } 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;
         zset *zs;
 
         if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;


@@ -3308,6 +3504,46 @@ static robj *rdbLoadObject(int type, FILE *fp) {
             zslInsert(zs->zsl,*score,ele);
             incrRefCount(ele); /* added to skiplist */
         }
             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);
     }
     } else {
         redisAssert(0 != 0);
     }


@@ -3434,7 +3670,7 @@ static void setGenericCommand(redisClient *c, int nx) {
              * to overwrite the old. So we delete the old key in the database.
              * This will also make sure that swap pages about the old object
              * will be marked as free. */
              * to overwrite the old. So we delete the old key in the database.
              * This will also make sure that swap pages about the old object
              * will be marked as free. */

-            if (deleteIfSwapped(c->db,c->argv[1]))
+            if (server.vm_enabled && deleteIfSwapped(c->db,c->argv[1]))

                 incrRefCount(c->argv[1]);
             dictReplace(c->db->dict,c->argv[1],c->argv[2]);
             incrRefCount(c->argv[2]);
                 incrRefCount(c->argv[1]);
             dictReplace(c->db->dict,c->argv[1],c->argv[2]);
             incrRefCount(c->argv[2]);


@@ -3618,6 +3854,96 @@ static void decrbyCommand(redisClient *c) {
     incrDecrCommand(c,-incr);
 }
 
     incrDecrCommand(c,-incr);
 }
 

+static void appendCommand(redisClient *c) {
+    int retval;
+    size_t totlen;
+    robj *o;
+
+    o = lookupKeyWrite(c->db,c->argv[1]);
+    if (o == NULL) {
+        /* Create the key */
+        retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
+        incrRefCount(c->argv[1]);
+        incrRefCount(c->argv[2]);
+        totlen = stringObjectLen(c->argv[2]);
+    } else {
+        dictEntry *de;
+       
+        de = dictFind(c->db->dict,c->argv[1]);
+        assert(de != NULL);
+
+        o = dictGetEntryVal(de);
+        if (o->type != REDIS_STRING) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+        /* If the object is specially encoded or shared we have to make
+         * a copy */
+        if (o->refcount != 1 || o->encoding != REDIS_ENCODING_RAW) {
+            robj *decoded = getDecodedObject(o);
+
+            o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
+            decrRefCount(decoded);
+            dictReplace(c->db->dict,c->argv[1],o);
+        }
+        /* APPEND! */
+        if (c->argv[2]->encoding == REDIS_ENCODING_RAW) {
+            o->ptr = sdscatlen(o->ptr,
+                c->argv[2]->ptr, sdslen(c->argv[2]->ptr));
+        } else {
+            o->ptr = sdscatprintf(o->ptr, "%ld",
+                (unsigned long) c->argv[2]->ptr);
+        }
+        totlen = sdslen(o->ptr);
+    }
+    server.dirty++;
+    addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",(unsigned long)totlen));
+}
+
+static void substrCommand(redisClient *c) {
+    robj *o;
+    long start = atoi(c->argv[2]->ptr);
+    long end = atoi(c->argv[3]->ptr);
+
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.nullbulk);
+    } else {
+        if (o->type != REDIS_STRING) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            size_t rangelen, strlen;
+            sds range;
+
+            o = getDecodedObject(o);
+            strlen = sdslen(o->ptr);
+
+            /* convert negative indexes */
+            if (start < 0) start = strlen+start;
+            if (end < 0) end = strlen+end;
+            if (start < 0) start = 0;
+            if (end < 0) end = 0;
+
+            /* indexes sanity checks */
+            if (start > end || (size_t)start >= strlen) {
+                /* Out of range start or start > end result in null reply */
+                addReply(c,shared.nullbulk);
+                decrRefCount(o);
+                return;
+            }
+            if ((size_t)end >= strlen) end = strlen-1;
+            rangelen = (end-start)+1;
+
+            /* Return the result */
+            addReplySds(c,sdscatprintf(sdsempty(),"$%zu\r\n",rangelen));
+            range = sdsnewlen((char*)o->ptr+start,rangelen);
+            addReplySds(c,range);
+            addReply(c,shared.crlf);
+            decrRefCount(o);
+        }
+    }
+}
+

 /* ========================= Type agnostic commands ========================= */
 
 static void delCommand(redisClient *c) {
 /* ========================= Type agnostic commands ========================= */
 
 static void delCommand(redisClient *c) {


@@ -3678,7 +4004,7 @@ static void keysCommand(redisClient *c) {
     dictEntry *de;
     sds pattern = c->argv[1]->ptr;
     int plen = sdslen(pattern);
     dictEntry *de;
     sds pattern = c->argv[1]->ptr;
     int plen = sdslen(pattern);

-    unsigned long numkeys = 0, keyslen = 0;
+    unsigned long numkeys = 0;

     robj *lenobj = createObject(REDIS_STRING,NULL);
 
     di = dictGetIterator(c->db->dict);
     robj *lenobj = createObject(REDIS_STRING,NULL);
 
     di = dictGetIterator(c->db->dict);


@@ -3691,17 +4017,15 @@ static void keysCommand(redisClient *c) {
         if ((pattern[0] == '*' && pattern[1] == '\0') ||
             stringmatchlen(pattern,plen,key,sdslen(key),0)) {
             if (expireIfNeeded(c->db,keyobj) == 0) {
         if ((pattern[0] == '*' && pattern[1] == '\0') ||
             stringmatchlen(pattern,plen,key,sdslen(key),0)) {
             if (expireIfNeeded(c->db,keyobj) == 0) {

-                if (numkeys != 0)
-                    addReply(c,shared.space);
+                addReplyBulkLen(c,keyobj);

                 addReply(c,keyobj);
                 addReply(c,keyobj);

+                addReply(c,shared.crlf);

                 numkeys++;
                 numkeys++;

-                keyslen += sdslen(key);

             }
         }
     }
     dictReleaseIterator(di);
             }
         }
     }
     dictReleaseIterator(di);

-    lenobj->ptr = sdscatprintf(sdsempty(),"$%lu\r\n",keyslen+(numkeys ? (numkeys-1) : 0));
-    addReply(c,shared.crlf);
+    lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",numkeys);

 }
 
 static void dbsizeCommand(redisClient *c) {
 }
 
 static void dbsizeCommand(redisClient *c) {


@@ -3727,7 +4051,8 @@ static void typeCommand(redisClient *c) {
         case REDIS_LIST: type = "+list"; break;
         case REDIS_SET: type = "+set"; break;
         case REDIS_ZSET: type = "+zset"; break;
         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));
         }
     }
     addReplySds(c,sdsnew(type));


@@ -3885,7 +4210,7 @@ static void pushGenericCommand(redisClient *c, int where) {
     lobj = lookupKeyWrite(c->db,c->argv[1]);
     if (lobj == NULL) {
         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
     lobj = lookupKeyWrite(c->db,c->argv[1]);
     if (lobj == NULL) {
         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {

-            addReply(c,shared.ok);
+            addReply(c,shared.cone);

             return;
         }
         lobj = createListObject();
             return;
         }
         lobj = createListObject();


@@ -3904,7 +4229,7 @@ static void pushGenericCommand(redisClient *c, int where) {
             return;
         }
         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
             return;
         }
         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {

-            addReply(c,shared.ok);
+            addReply(c,shared.cone);

             return;
         }
         list = lobj->ptr;
             return;
         }
         list = lobj->ptr;


@@ -3916,7 +4241,7 @@ static void pushGenericCommand(redisClient *c, int where) {
         incrRefCount(c->argv[2]);
     }
     server.dirty++;
         incrRefCount(c->argv[2]);
     }
     server.dirty++;

-    addReply(c,shared.ok);
+    addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",listLength(list)));

 }
 
 static void lpushCommand(redisClient *c) {
 }
 
 static void lpushCommand(redisClient *c) {


@@ -4523,6 +4848,7 @@ static void sinterstoreCommand(redisClient *c) {
 
 #define REDIS_OP_UNION 0
 #define REDIS_OP_DIFF 1
 
 #define REDIS_OP_UNION 0
 #define REDIS_OP_DIFF 1

+#define REDIS_OP_INTER 2

 
 static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey, int op) {
     dict **dv = zmalloc(sizeof(dict*)*setsnum);
 
 static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey, int op) {
     dict **dv = zmalloc(sizeof(dict*)*setsnum);


@@ -4654,6 +4980,8 @@ static zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
     zskiplistNode *zn = zmalloc(sizeof(*zn));
 
     zn->forward = zmalloc(sizeof(zskiplistNode*) * level);
     zskiplistNode *zn = zmalloc(sizeof(*zn));
 
     zn->forward = zmalloc(sizeof(zskiplistNode*) * level);

+    if (level > 0)
+        zn->span = zmalloc(sizeof(unsigned int) * (level - 1));

     zn->score = score;
     zn->obj = obj;
     return zn;
     zn->score = score;
     zn->obj = obj;
     return zn;


@@ -4667,8 +4995,13 @@ static zskiplist *zslCreate(void) {
     zsl->level = 1;
     zsl->length = 0;
     zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
     zsl->level = 1;
     zsl->length = 0;
     zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);

-    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++)
+    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {

         zsl->header->forward[j] = NULL;
         zsl->header->forward[j] = NULL;

+
+        /* span has space for ZSKIPLIST_MAXLEVEL-1 elements */
+        if (j < ZSKIPLIST_MAXLEVEL-1)
+            zsl->header->span[j] = 0;
+    }

     zsl->header->backward = NULL;
     zsl->tail = NULL;
     return zsl;
     zsl->header->backward = NULL;
     zsl->tail = NULL;
     return zsl;


@@ -4677,6 +5010,7 @@ static zskiplist *zslCreate(void) {
 static void zslFreeNode(zskiplistNode *node) {
     decrRefCount(node->obj);
     zfree(node->forward);
 static void zslFreeNode(zskiplistNode *node) {
     decrRefCount(node->obj);
     zfree(node->forward);

+    zfree(node->span);

     zfree(node);
 }
 
     zfree(node);
 }
 


@@ -4684,6 +5018,7 @@ static void zslFree(zskiplist *zsl) {
     zskiplistNode *node = zsl->header->forward[0], *next;
 
     zfree(zsl->header->forward);
     zskiplistNode *node = zsl->header->forward[0], *next;
 
     zfree(zsl->header->forward);

+    zfree(zsl->header->span);

     zfree(zsl->header);
     while(node) {
         next = node->forward[0];
     zfree(zsl->header);
     while(node) {
         next = node->forward[0];


@@ -4702,15 +5037,21 @@ static int zslRandomLevel(void) {
 
 static void zslInsert(zskiplist *zsl, double score, robj *obj) {
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
 
 static void zslInsert(zskiplist *zsl, double score, robj *obj) {
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;

+    unsigned int rank[ZSKIPLIST_MAXLEVEL];

     int i, level;
 
     x = zsl->header;
     for (i = zsl->level-1; i >= 0; i--) {
     int i, level;
 
     x = zsl->header;
     for (i = zsl->level-1; i >= 0; i--) {

+        /* store rank that is crossed to reach the insert position */
+        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
+

         while (x->forward[i] &&
             (x->forward[i]->score < score ||
                 (x->forward[i]->score == score &&
         while (x->forward[i] &&
             (x->forward[i]->score < score ||
                 (x->forward[i]->score == score &&

-                compareStringObjects(x->forward[i]->obj,obj) < 0)))
+                compareStringObjects(x->forward[i]->obj,obj) < 0))) {
+            rank[i] += i > 0 ? x->span[i-1] : 1;

             x = x->forward[i];
             x = x->forward[i];

+        }

         update[i] = x;
     }
     /* we assume the key is not already inside, since we allow duplicated
         update[i] = x;
     }
     /* we assume the key is not already inside, since we allow duplicated


@@ -4719,15 +5060,30 @@ static void zslInsert(zskiplist *zsl, double score, robj *obj) {
      * if the element is already inside or not. */
     level = zslRandomLevel();
     if (level > zsl->level) {
      * if the element is already inside or not. */
     level = zslRandomLevel();
     if (level > zsl->level) {

-        for (i = zsl->level; i < level; i++)
+        for (i = zsl->level; i < level; i++) {
+            rank[i] = 0;

             update[i] = zsl->header;
             update[i] = zsl->header;

+            update[i]->span[i-1] = zsl->length;
+        }

         zsl->level = level;
     }
     x = zslCreateNode(level,score,obj);
     for (i = 0; i < level; i++) {
         x->forward[i] = update[i]->forward[i];
         update[i]->forward[i] = x;
         zsl->level = level;
     }
     x = zslCreateNode(level,score,obj);
     for (i = 0; i < level; i++) {
         x->forward[i] = update[i]->forward[i];
         update[i]->forward[i] = x;

+
+        /* update span covered by update[i] as x is inserted here */
+        if (i > 0) {
+            x->span[i-1] = update[i]->span[i-1] - (rank[0] - rank[i]);
+            update[i]->span[i-1] = (rank[0] - rank[i]) + 1;
+        }
+    }
+
+    /* increment span for untouched levels */
+    for (i = level; i < zsl->level; i++) {
+        update[i]->span[i-1]++;

     }
     }

+

     x->backward = (update[0] == zsl->header) ? NULL : update[0];
     if (x->forward[0])
         x->forward[0]->backward = x;
     x->backward = (update[0] == zsl->header) ? NULL : update[0];
     if (x->forward[0])
         x->forward[0]->backward = x;


@@ -4736,6 +5092,31 @@ static void zslInsert(zskiplist *zsl, double score, robj *obj) {
     zsl->length++;
 }
 
     zsl->length++;
 }
 

+/* Internal function used by zslDelete, zslDeleteByScore and zslDeleteByRank */
+void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
+    int i;
+    for (i = 0; i < zsl->level; i++) {
+        if (update[i]->forward[i] == x) {
+            if (i > 0) {
+                update[i]->span[i-1] += x->span[i-1] - 1;
+            }
+            update[i]->forward[i] = x->forward[i];
+        } else {
+            /* invariant: i > 0, because update[0]->forward[0]
+             * is always equal to x */
+            update[i]->span[i-1] -= 1;
+        }
+    }
+    if (x->forward[0]) {
+        x->forward[0]->backward = x->backward;
+    } else {
+        zsl->tail = x->backward;
+    }
+    while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
+        zsl->level--;
+    zsl->length--;
+}
+

 /* Delete an element with matching score/object from the skiplist. */
 static int zslDelete(zskiplist *zsl, double score, robj *obj) {
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
 /* Delete an element with matching score/object from the skiplist. */
 static int zslDelete(zskiplist *zsl, double score, robj *obj) {
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;


@@ -4754,20 +5135,8 @@ static int zslDelete(zskiplist *zsl, double score, robj *obj) {
      * is to find the element with both the right score and object. */
     x = x->forward[0];
     if (x && score == x->score && compareStringObjects(x->obj,obj) == 0) {
      * is to find the element with both the right score and object. */
     x = x->forward[0];
     if (x && score == x->score && compareStringObjects(x->obj,obj) == 0) {

-        for (i = 0; i < zsl->level; i++) {
-            if (update[i]->forward[i] != x) break;
-            update[i]->forward[i] = x->forward[i];
-        }
-        if (x->forward[0]) {
-            x->forward[0]->backward = (x->backward == zsl->header) ?
-                                        NULL : x->backward;
-        } else {
-            zsl->tail = x->backward;
-        }
+        zslDeleteNode(zsl, x, update);

         zslFreeNode(x);
         zslFreeNode(x);

-        while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
-            zsl->level--;
-        zsl->length--;

         return 1;
     } else {
         return 0; /* not found */
         return 1;
     } else {
         return 0; /* not found */


@@ -4779,7 +5148,7 @@ static int zslDelete(zskiplist *zsl, double score, robj *obj) {
  * Min and mx are inclusive, so a score >= min || score <= max is deleted.
  * Note that this function takes the reference to the hash table view of the
  * sorted set, in order to remove the elements from the hash table too. */
  * Min and mx are inclusive, so a score >= min || score <= max is deleted.
  * Note that this function takes the reference to the hash table view of the
  * sorted set, in order to remove the elements from the hash table too. */

-static unsigned long zslDeleteRange(zskiplist *zsl, double min, double max, dict *dict) {
+static unsigned long zslDeleteRangeByScore(zskiplist *zsl, double min, double max, dict *dict) {

     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
     unsigned long removed = 0;
     int i;
     zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
     unsigned long removed = 0;
     int i;


@@ -4794,28 +5163,44 @@ static unsigned long zslDeleteRange(zskiplist *zsl, double min, double max, dict
      * is to find the element with both the right score and object. */
     x = x->forward[0];
     while (x && x->score <= max) {
      * is to find the element with both the right score and object. */
     x = x->forward[0];
     while (x && x->score <= max) {

-        zskiplistNode *next;
+        zskiplistNode *next = x->forward[0];
+        zslDeleteNode(zsl, x, update);
+        dictDelete(dict,x->obj);
+        zslFreeNode(x);
+        removed++;
+        x = next;
+    }
+    return removed; /* not found */
+}

 
 

-        for (i = 0; i < zsl->level; i++) {
-            if (update[i]->forward[i] != x) break;
-            update[i]->forward[i] = x->forward[i];
-        }
-        if (x->forward[0]) {
-            x->forward[0]->backward = (x->backward == zsl->header) ?
-                                        NULL : x->backward;
-        } else {
-            zsl->tail = x->backward;
+/* Delete all the elements with rank between start and end from the skiplist.
+ * Start and end are inclusive. Note that start and end need to be 1-based */
+static unsigned long zslDeleteRangeByRank(zskiplist *zsl, unsigned int start, unsigned int end, dict *dict) {
+    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
+    unsigned long traversed = 0, removed = 0;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] && (traversed + (i > 0 ? x->span[i-1] : 1)) < start) {
+            traversed += i > 0 ? x->span[i-1] : 1;
+            x = x->forward[i];

         }
         }

-        next = x->forward[0];
+        update[i] = x;
+    }
+
+    traversed++;
+    x = x->forward[0];
+    while (x && traversed <= end) {
+        zskiplistNode *next = x->forward[0];
+        zslDeleteNode(zsl, x, update);

         dictDelete(dict,x->obj);
         zslFreeNode(x);
         dictDelete(dict,x->obj);
         zslFreeNode(x);

-        while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
-            zsl->level--;
-        zsl->length--;

         removed++;
         removed++;

+        traversed++;

         x = next;
     }
         x = next;
     }

-    return removed; /* not found */
+    return removed;

 }
 
 /* Find the first node having a score equal or greater than the specified one.
 }
 
 /* Find the first node having a score equal or greater than the specified one.


@@ -4834,6 +5219,53 @@ static zskiplistNode *zslFirstWithScore(zskiplist *zsl, double score) {
     return x->forward[0];
 }
 
     return x->forward[0];
 }
 

+/* Find the rank for an element by both score and key.
+ * Returns 0 when the element cannot be found, rank otherwise.
+ * Note that the rank is 1-based due to the span of zsl->header to the
+ * first element. */
+static unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
+    zskiplistNode *x;
+    unsigned long rank = 0;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] &&
+            (x->forward[i]->score < score ||
+                (x->forward[i]->score == score &&
+                compareStringObjects(x->forward[i]->obj,o) <= 0))) {
+            rank += i > 0 ? x->span[i-1] : 1;
+            x = x->forward[i];
+        }
+
+        /* x might be equal to zsl->header, so test if obj is non-NULL */
+        if (x->obj && compareStringObjects(x->obj,o) == 0) {
+            return rank;
+        }
+    }
+    return 0;
+}
+
+/* Finds an element by its rank. The rank argument needs to be 1-based. */
+zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
+    zskiplistNode *x;
+    unsigned long traversed = 0;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] && (traversed + (i > 0 ? x->span[i-1] : 1)) <= rank) {
+            traversed += i > 0 ? x->span[i-1] : 1;
+            x = x->forward[i];
+        }
+
+        if (traversed == rank) {
+            return x;
+        }
+    }
+    return NULL;
+}
+

 /* The actual Z-commands implementations */
 
 /* This generic command implements both ZADD and ZINCRBY.
 /* The actual Z-commands implementations */
 
 /* This generic command implements both ZADD and ZINCRBY.


@@ -4983,67 +5415,271 @@ static void zremrangebyscoreCommand(redisClient *c) {
             return;
         }
         zs = zsetobj->ptr;
             return;
         }
         zs = zsetobj->ptr;

-        deleted = zslDeleteRange(zs->zsl,min,max,zs->dict);
+        deleted = zslDeleteRangeByScore(zs->zsl,min,max,zs->dict);

         if (htNeedsResize(zs->dict)) dictResize(zs->dict);
         server.dirty += deleted;
         addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",deleted));
     }
 }
 
         if (htNeedsResize(zs->dict)) dictResize(zs->dict);
         server.dirty += deleted;
         addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",deleted));
     }
 }
 

-static void zrangeGenericCommand(redisClient *c, int reverse) {
-    robj *o;
+static void zremrangebyrankCommand(redisClient *c) {

     int start = atoi(c->argv[2]->ptr);
     int end = atoi(c->argv[3]->ptr);
     int start = atoi(c->argv[2]->ptr);
     int end = atoi(c->argv[3]->ptr);

-    int withscores = 0;
-
-    if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
-        withscores = 1;
-    } else if (c->argc >= 5) {
-        addReply(c,shared.syntaxerr);
-        return;
-    }
+    robj *zsetobj;
+    zset *zs;

 
 

-    o = lookupKeyRead(c->db,c->argv[1]);
-    if (o == NULL) {
-        addReply(c,shared.nullmultibulk);
+    zsetobj = lookupKeyWrite(c->db,c->argv[1]);
+    if (zsetobj == NULL) {
+        addReply(c,shared.czero);

     } else {
     } else {

-        if (o->type != REDIS_ZSET) {
+        if (zsetobj->type != REDIS_ZSET) {

             addReply(c,shared.wrongtypeerr);
             addReply(c,shared.wrongtypeerr);

-        } else {
-            zset *zsetobj = o->ptr;
-            zskiplist *zsl = zsetobj->zsl;
-            zskiplistNode *ln;
+            return;
+        }

 
 

-            int llen = zsl->length;
-            int rangelen, j;
-            robj *ele;
+        zs = zsetobj->ptr;
+        int llen = zs->zsl->length;
+        long deleted;

 
 

-            /* convert negative indexes */
-            if (start < 0) start = llen+start;
-            if (end < 0) end = llen+end;
-            if (start < 0) start = 0;
-            if (end < 0) end = 0;
+        /* convert negative indexes */
+        if (start < 0) start = llen+start;
+        if (end < 0) end = llen+end;
+        if (start < 0) start = 0;
+        if (end < 0) end = 0;

 
 

-            /* indexes sanity checks */
-            if (start > end || start >= llen) {
-                /* Out of range start or start > end result in empty list */
-                addReply(c,shared.emptymultibulk);
-                return;
-            }
-            if (end >= llen) end = llen-1;
-            rangelen = (end-start)+1;
+        /* indexes sanity checks */
+        if (start > end || start >= llen) {
+            addReply(c,shared.czero);
+            return;
+        }
+        if (end >= llen) end = llen-1;

 
 

-            /* Return the result in form of a multi-bulk reply */
+        /* increment start and end because zsl*Rank functions
+         * use 1-based rank */
+        deleted = zslDeleteRangeByRank(zs->zsl,start+1,end+1,zs->dict);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        server.dirty += deleted;
+        addReplyLong(c, deleted);
+    }
+}
+
+typedef struct {
+    dict *dict;
+    double weight;
+} zsetopsrc;
+
+static int qsortCompareZsetopsrcByCardinality(const void *s1, const void *s2) {
+    zsetopsrc *d1 = (void*) s1, *d2 = (void*) s2;
+    unsigned long size1, size2;
+    size1 = d1->dict ? dictSize(d1->dict) : 0;
+    size2 = d2->dict ? dictSize(d2->dict) : 0;
+    return size1 - size2;
+}
+
+static void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
+    int i, j, zsetnum;
+    zsetopsrc *src;
+    robj *dstobj;
+    zset *dstzset;
+    dictIterator *di;
+    dictEntry *de;
+
+    /* expect zsetnum input keys to be given */
+    zsetnum = atoi(c->argv[2]->ptr);
+    if (zsetnum < 1) {
+        addReplySds(c,sdsnew("-ERR at least 1 input key is needed for ZUNION/ZINTER\r\n"));
+        return;
+    }
+
+    /* test if the expected number of keys would overflow */
+    if (3+zsetnum > c->argc) {
+        addReply(c,shared.syntaxerr);
+        return;
+    }
+
+    /* read keys to be used for input */
+    src = zmalloc(sizeof(zsetopsrc) * zsetnum);
+    for (i = 0, j = 3; i < zsetnum; i++, j++) {
+        robj *zsetobj = lookupKeyWrite(c->db,c->argv[j]);
+        if (!zsetobj) {
+            src[i].dict = NULL;
+        } else {
+            if (zsetobj->type != REDIS_ZSET) {
+                zfree(src);
+                addReply(c,shared.wrongtypeerr);
+                return;
+            }
+            src[i].dict = ((zset*)zsetobj->ptr)->dict;
+        }
+
+        /* default all weights to 1 */
+        src[i].weight = 1.0;
+    }
+
+    /* parse optional extra arguments */
+    if (j < c->argc) {
+        int remaining = c->argc-j;
+
+        while (remaining) {
+            if (!strcasecmp(c->argv[j]->ptr,"weights")) {
+                j++; remaining--;
+                if (remaining < zsetnum) {
+                    zfree(src);
+                    addReplySds(c,sdsnew("-ERR not enough weights for ZUNION/ZINTER\r\n"));
+                    return;
+                }
+                for (i = 0; i < zsetnum; i++, j++, remaining--) {
+                    src[i].weight = strtod(c->argv[j]->ptr, NULL);
+                }
+            } else {
+                zfree(src);
+                addReply(c,shared.syntaxerr);
+                return;
+            }
+        }
+    }
+
+    dstobj = createZsetObject();
+    dstzset = dstobj->ptr;
+
+    if (op == REDIS_OP_INTER) {
+        /* sort sets from the smallest to largest, this will improve our
+         * algorithm's performance */
+        qsort(src,zsetnum,sizeof(zsetopsrc), qsortCompareZsetopsrcByCardinality);
+
+        /* skip going over all entries if the smallest zset is NULL or empty */
+        if (src[0].dict && dictSize(src[0].dict) > 0) {
+            /* precondition: as src[0].dict is non-empty and the zsets are ordered
+             * from small to large, all src[i > 0].dict are non-empty too */
+            di = dictGetIterator(src[0].dict);
+            while((de = dictNext(di)) != NULL) {
+                double *score = zmalloc(sizeof(double));
+                *score = 0.0;
+
+                for (j = 0; j < zsetnum; j++) {
+                    dictEntry *other = (j == 0) ? de : dictFind(src[j].dict,dictGetEntryKey(de));
+                    if (other) {
+                        *score = *score + src[j].weight * (*(double*)dictGetEntryVal(other));
+                    } else {
+                        break;
+                    }
+                }
+
+                /* skip entry when not present in every source dict */
+                if (j != zsetnum) {
+                    zfree(score);
+                } else {
+                    robj *o = dictGetEntryKey(de);
+                    dictAdd(dstzset->dict,o,score);
+                    incrRefCount(o); /* added to dictionary */
+                    zslInsert(dstzset->zsl,*score,o);
+                    incrRefCount(o); /* added to skiplist */
+                }
+            }
+            dictReleaseIterator(di);
+        }
+    } else if (op == REDIS_OP_UNION) {
+        for (i = 0; i < zsetnum; i++) {
+            if (!src[i].dict) continue;
+
+            di = dictGetIterator(src[i].dict);
+            while((de = dictNext(di)) != NULL) {
+                /* skip key when already processed */
+                if (dictFind(dstzset->dict,dictGetEntryKey(de)) != NULL) continue;
+
+                double *score = zmalloc(sizeof(double));
+                *score = 0.0;
+                for (j = 0; j < zsetnum; j++) {
+                    if (!src[j].dict) continue;
+
+                    dictEntry *other = (i == j) ? de : dictFind(src[j].dict,dictGetEntryKey(de));
+                    if (other) {
+                        *score = *score + src[j].weight * (*(double*)dictGetEntryVal(other));
+                    }
+                }
+
+                robj *o = dictGetEntryKey(de);
+                dictAdd(dstzset->dict,o,score);
+                incrRefCount(o); /* added to dictionary */
+                zslInsert(dstzset->zsl,*score,o);
+                incrRefCount(o); /* added to skiplist */
+            }
+            dictReleaseIterator(di);
+        }
+    } else {
+        /* unknown operator */
+        redisAssert(op == REDIS_OP_INTER || op == REDIS_OP_UNION);
+    }
+
+    deleteKey(c->db,dstkey);
+    dictAdd(c->db->dict,dstkey,dstobj);
+    incrRefCount(dstkey);
+
+    addReplyLong(c, dstzset->zsl->length);
+    server.dirty++;
+    zfree(src);
+}
+
+static void zunionCommand(redisClient *c) {
+    zunionInterGenericCommand(c,c->argv[1], REDIS_OP_UNION);
+}
+
+static void zinterCommand(redisClient *c) {
+    zunionInterGenericCommand(c,c->argv[1], REDIS_OP_INTER);
+}
+
+static void zrangeGenericCommand(redisClient *c, int reverse) {
+    robj *o;
+    int start = atoi(c->argv[2]->ptr);
+    int end = atoi(c->argv[3]->ptr);
+    int withscores = 0;
+
+    if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
+        withscores = 1;
+    } else if (c->argc >= 5) {
+        addReply(c,shared.syntaxerr);
+        return;
+    }
+
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.nullmultibulk);
+    } else {
+        if (o->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            zset *zsetobj = o->ptr;
+            zskiplist *zsl = zsetobj->zsl;
+            zskiplistNode *ln;
+
+            int llen = zsl->length;
+            int rangelen, j;
+            robj *ele;
+
+            /* convert negative indexes */
+            if (start < 0) start = llen+start;
+            if (end < 0) end = llen+end;
+            if (start < 0) start = 0;
+            if (end < 0) end = 0;
+
+            /* indexes sanity checks */
+            if (start > end || start >= llen) {
+                /* Out of range start or start > end result in empty list */
+                addReply(c,shared.emptymultibulk);
+                return;
+            }
+            if (end >= llen) end = llen-1;
+            rangelen = (end-start)+1;
+
+            /* check if starting point is trivial, before searching
+             * the element in log(N) time */

             if (reverse) {
             if (reverse) {

-                ln = zsl->tail;
-                while (start--)
-                    ln = ln->backward;
+                ln = start == 0 ? zsl->tail : zslGetElementByRank(zsl, llen-start);

             } else {
             } else {

-                ln = zsl->header->forward[0];
-                while (start--)
-                    ln = ln->forward[0];
+                ln = start == 0 ? zsl->header->forward[0] : zslGetElementByRank(zsl, start+1);

             }
 
             }
 

+            /* Return the result in form of a multi-bulk reply */

             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",
                 withscores ? (rangelen*2) : rangelen));
             for (j = 0; j < rangelen; j++) {
             addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",
                 withscores ? (rangelen*2) : rangelen));
             for (j = 0; j < rangelen; j++) {


@@ -5067,28 +5703,64 @@ static void zrevrangeCommand(redisClient *c) {
     zrangeGenericCommand(c,1);
 }
 
     zrangeGenericCommand(c,1);
 }
 

-static void zrangebyscoreCommand(redisClient *c) {
+/* This command implements both ZRANGEBYSCORE and ZCOUNT.
+ * If justcount is non-zero, just the count is returned. */
+static void genericZrangebyscoreCommand(redisClient *c, int justcount) {

     robj *o;
     robj *o;

-    double min = strtod(c->argv[2]->ptr,NULL);
-    double max = strtod(c->argv[3]->ptr,NULL);
+    double min, max;
+    int minex = 0, maxex = 0; /* are min or max exclusive? */

     int offset = 0, limit = -1;
     int offset = 0, limit = -1;

+    int withscores = 0;
+    int badsyntax = 0;
+
+    /* Parse the min-max interval. If one of the values is prefixed
+     * by the "(" character, it's considered "open". For instance
+     * ZRANGEBYSCORE zset (1.5 (2.5 will match min < x < max
+     * ZRANGEBYSCORE zset 1.5 2.5 will instead match min <= x <= max */
+    if (((char*)c->argv[2]->ptr)[0] == '(') {
+        min = strtod((char*)c->argv[2]->ptr+1,NULL);
+        minex = 1;
+    } else {
+        min = strtod(c->argv[2]->ptr,NULL);
+    }
+    if (((char*)c->argv[3]->ptr)[0] == '(') {
+        max = strtod((char*)c->argv[3]->ptr+1,NULL);
+        maxex = 1;
+    } else {
+        max = strtod(c->argv[3]->ptr,NULL);
+    }

 
 

-    if (c->argc != 4 && c->argc != 7) {
+    /* Parse "WITHSCORES": note that if the command was called with
+     * the name ZCOUNT then we are sure that c->argc == 4, so we'll never
+     * enter the following paths to parse WITHSCORES and LIMIT. */
+    if (c->argc == 5 || c->argc == 8) {
+        if (strcasecmp(c->argv[c->argc-1]->ptr,"withscores") == 0)
+            withscores = 1;
+        else
+            badsyntax = 1;
+    }
+    if (c->argc != (4 + withscores) && c->argc != (7 + withscores))
+        badsyntax = 1;
+    if (badsyntax) {

         addReplySds(c,
             sdsnew("-ERR wrong number of arguments for ZRANGEBYSCORE\r\n"));
         return;
         addReplySds(c,
             sdsnew("-ERR wrong number of arguments for ZRANGEBYSCORE\r\n"));
         return;

-    } else if (c->argc == 7 && strcasecmp(c->argv[4]->ptr,"limit")) {
+    }
+
+    /* Parse "LIMIT" */
+    if (c->argc == (7 + withscores) && strcasecmp(c->argv[4]->ptr,"limit")) {

         addReply(c,shared.syntaxerr);
         return;
         addReply(c,shared.syntaxerr);
         return;

-    } else if (c->argc == 7) {
+    } else if (c->argc == (7 + withscores)) {

         offset = atoi(c->argv[5]->ptr);
         limit = atoi(c->argv[6]->ptr);
         if (offset < 0) offset = 0;
     }
 
         offset = atoi(c->argv[5]->ptr);
         limit = atoi(c->argv[6]->ptr);
         if (offset < 0) offset = 0;
     }
 

+    /* Ok, lookup the key and get the range */

     o = lookupKeyRead(c->db,c->argv[1]);
     if (o == NULL) {
     o = lookupKeyRead(c->db,c->argv[1]);
     if (o == NULL) {

-        addReply(c,shared.nullmultibulk);
+        addReply(c,justcount ? shared.czero : shared.nullmultibulk);

     } else {
         if (o->type != REDIS_ZSET) {
             addReply(c,shared.wrongtypeerr);
     } else {
         if (o->type != REDIS_ZSET) {
             addReply(c,shared.wrongtypeerr);


@@ -5096,14 +5768,17 @@ static void zrangebyscoreCommand(redisClient *c) {
             zset *zsetobj = o->ptr;
             zskiplist *zsl = zsetobj->zsl;
             zskiplistNode *ln;
             zset *zsetobj = o->ptr;
             zskiplist *zsl = zsetobj->zsl;
             zskiplistNode *ln;

-            robj *ele, *lenobj;
-            unsigned int rangelen = 0;
+            robj *ele, *lenobj = NULL;
+            unsigned long rangelen = 0;

 
 

-            /* Get the first node with the score >= min */
+            /* Get the first node with the score >= min, or with
+             * score > min if 'minex' is true. */

             ln = zslFirstWithScore(zsl,min);
             ln = zslFirstWithScore(zsl,min);

+            while (minex && ln && ln->score == min) ln = ln->forward[0];
+

             if (ln == NULL) {
                 /* No element matching the speciifed interval */
             if (ln == NULL) {
                 /* No element matching the speciifed interval */

-                addReply(c,shared.emptymultibulk);
+                addReply(c,justcount ? shared.czero : shared.emptymultibulk);

                 return;
             }
 
                 return;
             }
 


@@ -5111,30 +5786,49 @@ static void zrangebyscoreCommand(redisClient *c) {
              * are in the list, so we push this object that will represent
              * the multi-bulk length in the output buffer, and will "fix"
              * it later */
              * are in the list, so we push this object that will represent
              * the multi-bulk length in the output buffer, and will "fix"
              * it later */

-            lenobj = createObject(REDIS_STRING,NULL);
-            addReply(c,lenobj);
-            decrRefCount(lenobj);
+            if (!justcount) {
+                lenobj = createObject(REDIS_STRING,NULL);
+                addReply(c,lenobj);
+                decrRefCount(lenobj);
+            }

 
 

-            while(ln && ln->score <= max) {
+            while(ln && (maxex ? (ln->score < max) : (ln->score <= max))) {

                 if (offset) {
                     offset--;
                     ln = ln->forward[0];
                     continue;
                 }
                 if (limit == 0) break;
                 if (offset) {
                     offset--;
                     ln = ln->forward[0];
                     continue;
                 }
                 if (limit == 0) break;

-                ele = ln->obj;
-                addReplyBulkLen(c,ele);
-                addReply(c,ele);
-                addReply(c,shared.crlf);
+                if (!justcount) {
+                    ele = ln->obj;
+                    addReplyBulkLen(c,ele);
+                    addReply(c,ele);
+                    addReply(c,shared.crlf);
+                    if (withscores)
+                        addReplyDouble(c,ln->score);
+                }

                 ln = ln->forward[0];
                 rangelen++;
                 if (limit > 0) limit--;
             }
                 ln = ln->forward[0];
                 rangelen++;
                 if (limit > 0) limit--;
             }

-            lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",rangelen);
+            if (justcount) {
+                addReplyLong(c,(long)rangelen);
+            } else {
+                lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",
+                     withscores ? (rangelen*2) : rangelen);
+            }

         }
     }
 }
 
         }
     }
 }
 

+static void zrangebyscoreCommand(redisClient *c) {
+    genericZrangebyscoreCommand(c,0);
+}
+
+static void zcountCommand(redisClient *c) {
+    genericZrangebyscoreCommand(c,1);
+}
+

 static void zcardCommand(redisClient *c) {
     robj *o;
     zset *zs;
 static void zcardCommand(redisClient *c) {
     robj *o;
     zset *zs;


@@ -5180,6 +5874,196 @@ static void zscoreCommand(redisClient *c) {
     }
 }
 
     }
 }
 

+static void zrankGenericCommand(redisClient *c, int reverse) {
+    robj *o;
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.nullbulk);
+        return;
+    }
+    if (o->type != REDIS_ZSET) {
+        addReply(c,shared.wrongtypeerr);
+    } else {
+        zset *zs = o->ptr;
+        zskiplist *zsl = zs->zsl;
+        dictEntry *de;
+        unsigned long rank;
+
+        de = dictFind(zs->dict,c->argv[2]);
+        if (!de) {
+            addReply(c,shared.nullbulk);
+            return;
+        }
+
+        double *score = dictGetEntryVal(de);
+        rank = zslGetRank(zsl, *score, c->argv[2]);
+        if (rank) {
+            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;
+        }
+    }
+    /* We want to convert the zipmap into an hash table right now if the
+     * entry to be added is too big. Note that we check if the object
+     * is integer encoded before to try fetching the length in the test below.
+     * This is because integers are small, but currently stringObjectLen()
+     * performs a slow conversion: not worth it. */
+    if (o->encoding == REDIS_ENCODING_ZIPMAP &&
+        ((c->argv[2]->encoding == REDIS_ENCODING_RAW &&
+          sdslen(c->argv[2]->ptr) > server.hash_max_zipmap_value) ||
+         (c->argv[3]->encoding == REDIS_ENCODING_RAW &&
+          sdslen(c->argv[3]->ptr) > server.hash_max_zipmap_value)))
+    {
+        convertToRealHash(o);
+    }
+
+    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;
+
+        /* And here there is the second check for hash conversion...
+         * we want to do it only if the operation was not just an update as
+         * zipmapLen() is O(N). */
+        if (!update && zipmapLen(zm) > server.hash_max_zipmap_entries)
+            convertToRealHash(o);
+    } else {
+        tryObjectEncoding(c->argv[2]);
+        /* note that c->argv[3] is already encoded, as the latest arg
+         * of a bulk command is always integer encoded if possible. */
+        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->type != REDIS_HASH) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+
+        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 hdelCommand(redisClient *c) {
+    robj *o = lookupKeyRead(c->db,c->argv[1]);
+
+    if (o == NULL) {
+        addReply(c,shared.czero);
+        return;
+    } else {
+        int deleted = 0;
+
+        if (o->type != REDIS_HASH) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+
+        if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+            o->ptr = zipmapDel((unsigned char*) o->ptr,
+                (unsigned char*) c->argv[2]->ptr,
+                sdslen(c->argv[2]->ptr), &deleted);
+        } else {
+            deleted = dictDelete((dict*)o->ptr,c->argv[2]) == DICT_OK;
+        }
+        addReply(c,deleted ? shared.cone : shared.czero);
+    }
+}
+
+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) {
 /* ========================= Non type-specific commands  ==================== */
 
 static void flushdbCommand(redisClient *c) {


@@ -5588,6 +6472,9 @@ static sds genRedisInfoString(void) {
     time_t uptime = time(NULL)-server.stat_starttime;
     int j;
     char hmem[64];
     time_t uptime = time(NULL)-server.stat_starttime;
     int j;
     char hmem[64];

+
+    server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;
+    server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE;

   
     bytesToHuman(hmem,zmalloc_used_memory());
     info = sdscatprintf(sdsempty(),
   
     bytesToHuman(hmem,zmalloc_used_memory());
     info = sdscatprintf(sdsempty(),


@@ -5608,6 +6495,8 @@ static sds genRedisInfoString(void) {
         "bgrewriteaof_in_progress:%d\r\n"
         "total_connections_received:%lld\r\n"
         "total_commands_processed:%lld\r\n"
         "bgrewriteaof_in_progress:%d\r\n"
         "total_connections_received:%lld\r\n"
         "total_commands_processed:%lld\r\n"

+        "hash_max_zipmap_entries:%ld\r\n"
+        "hash_max_zipmap_value:%ld\r\n"

         "vm_enabled:%d\r\n"
         "role:%s\r\n"
         ,REDIS_VERSION,
         "vm_enabled:%d\r\n"
         "role:%s\r\n"
         ,REDIS_VERSION,


@@ -5618,7 +6507,7 @@ static sds genRedisInfoString(void) {
         uptime/(3600*24),
         listLength(server.clients)-listLength(server.slaves),
         listLength(server.slaves),
         uptime/(3600*24),
         listLength(server.clients)-listLength(server.slaves),
         listLength(server.slaves),

-        server.blockedclients,
+        server.blpop_blocked_clients,

         zmalloc_used_memory(),
         hmem,
         server.dirty,
         zmalloc_used_memory(),
         hmem,
         server.dirty,


@@ -5627,6 +6516,8 @@ static sds genRedisInfoString(void) {
         server.bgrewritechildpid != -1,
         server.stat_numconnections,
         server.stat_numcommands,
         server.bgrewritechildpid != -1,
         server.stat_numconnections,
         server.stat_numcommands,

+        server.hash_max_zipmap_entries,
+        server.hash_max_zipmap_value,

         server.vm_enabled != 0,
         server.masterhost == NULL ? "master" : "slave"
     );
         server.vm_enabled != 0,
         server.masterhost == NULL ? "master" : "slave"
     );


@@ -5656,8 +6547,8 @@ static sds genRedisInfoString(void) {
             "vm_stats_io_newjobs_len:%lu\r\n"
             "vm_stats_io_processing_len:%lu\r\n"
             "vm_stats_io_processed_len:%lu\r\n"
             "vm_stats_io_newjobs_len:%lu\r\n"
             "vm_stats_io_processing_len:%lu\r\n"
             "vm_stats_io_processed_len:%lu\r\n"

-            "vm_stats_io_waiting_clients:%lu\r\n"

             "vm_stats_io_active_threads:%lu\r\n"
             "vm_stats_io_active_threads:%lu\r\n"

+            "vm_stats_blocked_clients:%lu\r\n"

             ,(unsigned long long) server.vm_max_memory,
             (unsigned long long) server.vm_page_size,
             (unsigned long long) server.vm_pages,
             ,(unsigned long long) server.vm_max_memory,
             (unsigned long long) server.vm_page_size,
             (unsigned long long) server.vm_pages,


@@ -5668,8 +6559,8 @@ static sds genRedisInfoString(void) {
             (unsigned long) listLength(server.io_newjobs),
             (unsigned long) listLength(server.io_processing),
             (unsigned long) listLength(server.io_processed),
             (unsigned long) listLength(server.io_newjobs),
             (unsigned long) listLength(server.io_processing),
             (unsigned long) listLength(server.io_processed),

-            (unsigned long) listLength(server.io_clients),
-            (unsigned long) server.io_active_threads
+            (unsigned long) server.io_active_threads,
+            (unsigned long) server.vm_blocked_clients

         );
         unlockThreadedIO();
     }
         );
         unlockThreadedIO();
     }


@@ -5853,6 +6744,18 @@ static void multiCommand(redisClient *c) {
     addReply(c,shared.ok);
 }
 
     addReply(c,shared.ok);
 }
 

+static void discardCommand(redisClient *c) {
+    if (!(c->flags & REDIS_MULTI)) {
+        addReplySds(c,sdsnew("-ERR DISCARD without MULTI\r\n"));
+        return;
+    }
+
+    freeClientMultiState(c);
+    initClientMultiState(c);
+    c->flags &= (~REDIS_MULTI);
+    addReply(c,shared.ok);
+}
+

 static void execCommand(redisClient *c) {
     int j;
     robj **orig_argv;
 static void execCommand(redisClient *c) {
     int j;
     robj **orig_argv;


@@ -5941,8 +6844,7 @@ static void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeou
     }
     /* Mark the client as a blocked client */
     c->flags |= REDIS_BLOCKED;
     }
     /* Mark the client as a blocked client */
     c->flags |= REDIS_BLOCKED;

-    aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
-    server.blockedclients++;
+    server.blpop_blocked_clients++;

 }
 
 /* Unblock a client that's waiting in a blocking operation such as BLPOP */
 }
 
 /* Unblock a client that's waiting in a blocking operation such as BLPOP */


@@ -5968,15 +6870,8 @@ static void unblockClientWaitingData(redisClient *c) {
     zfree(c->blockingkeys);
     c->blockingkeys = NULL;
     c->flags &= (~REDIS_BLOCKED);
     zfree(c->blockingkeys);
     c->blockingkeys = NULL;
     c->flags &= (~REDIS_BLOCKED);

-    server.blockedclients--;
-    /* Ok now we are ready to get read events from socket, note that we
-     * can't trap errors here as it's possible that unblockClientWaitingDatas() is
-     * called from freeClient() itself, and the only thing we can do
-     * if we failed to register the READABLE event is to kill the client.
-     * Still the following function should never fail in the real world as
-     * we are sure the file descriptor is sane, and we exit on out of mem. */
-    aeCreateFileEvent(server.el, c->fd, AE_READABLE, readQueryFromClient, c);
-    /* As a final step we want to process data if there is some command waiting
+    server.blpop_blocked_clients--;
+    /* We want to process data if there is some command waiting

      * in the input buffer. Note that this is safe even if
      * unblockClientWaitingData() gets called from freeClient() because
      * freeClient() will be smart enough to call this function
      * in the input buffer. Note that this is safe even if
      * unblockClientWaitingData() gets called from freeClient() because
      * freeClient() will be smart enough to call this function


@@ -6316,9 +7211,9 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
 
 static int syncWithMaster(void) {
     char buf[1024], tmpfile[256], authcmd[1024];
 
 static int syncWithMaster(void) {
     char buf[1024], tmpfile[256], authcmd[1024];

-    int dumpsize;
+    long dumpsize;

     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);

-    int dfd;
+    int dfd, maxtries = 5;

 
     if (fd == -1) {
         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
 
     if (fd == -1) {
         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",


@@ -6368,11 +7263,16 @@ 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;
     }
         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 */
     /* 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);
+    while(maxtries--) {
+        snprintf(tmpfile,256,
+            "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
+        dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
+        if (dfd != -1) break;
+        sleep(1);
+    }

     if (dfd == -1) {
         close(fd);
         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
     if (dfd == -1) {
         close(fd);
         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));


@@ -6712,7 +7612,7 @@ fmterr:
 }
 
 /* Write an object into a file in the bulk format $<count>\r\n<payload>\r\n */
 }
 
 /* Write an object into a file in the bulk format $<count>\r\n<payload>\r\n */

-static int fwriteBulk(FILE *fp, robj *obj) {
+static int fwriteBulkObject(FILE *fp, robj *obj) {

     char buf[128];
     int decrrc = 0;
 
     char buf[128];
     int decrrc = 0;
 


@@ -6737,6 +7637,18 @@ err:
     return 0;
 }
 
     return 0;
 }
 

+/* Write binary-safe string into a file in the bulkformat
+ * $<count>\r\n<payload>\r\n */
+static int fwriteBulkString(FILE *fp, char *s, unsigned long len) {
+    char buf[128];
+
+    snprintf(buf,sizeof(buf),"$%ld\r\n",(unsigned long)len);
+    if (fwrite(buf,strlen(buf),1,fp) == 0) return 0;
+    if (len && fwrite(s,len,1,fp) == 0) return 0;
+    if (fwrite("\r\n",2,1,fp) == 0) return 0;
+    return 1;
+}
+

 /* Write a double value in bulk format $<count>\r\n<payload>\r\n */
 static int fwriteBulkDouble(FILE *fp, double d) {
     char buf[128], dbuf[128];
 /* Write a double value in bulk format $<count>\r\n<payload>\r\n */
 static int fwriteBulkDouble(FILE *fp, double d) {
     char buf[128], dbuf[128];


@@ -6819,8 +7731,8 @@ static int rewriteAppendOnlyFile(char *filename) {
                 char cmd[]="*3\r\n$3\r\nSET\r\n";
                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                 /* Key and value */
                 char cmd[]="*3\r\n$3\r\nSET\r\n";
                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                 /* Key and value */

-                if (fwriteBulk(fp,key) == 0) goto werr;
-                if (fwriteBulk(fp,o) == 0) goto werr;
+                if (fwriteBulkObject(fp,key) == 0) goto werr;
+                if (fwriteBulkObject(fp,o) == 0) goto werr;

             } else if (o->type == REDIS_LIST) {
                 /* Emit the RPUSHes needed to rebuild the list */
                 list *list = o->ptr;
             } else if (o->type == REDIS_LIST) {
                 /* Emit the RPUSHes needed to rebuild the list */
                 list *list = o->ptr;


@@ -6833,8 +7745,8 @@ static int rewriteAppendOnlyFile(char *filename) {
                     robj *eleobj = listNodeValue(ln);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                     robj *eleobj = listNodeValue(ln);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;

-                    if (fwriteBulk(fp,key) == 0) goto werr;
-                    if (fwriteBulk(fp,eleobj) == 0) goto werr;
+                    if (fwriteBulkObject(fp,key) == 0) goto werr;
+                    if (fwriteBulkObject(fp,eleobj) == 0) goto werr;

                 }
             } else if (o->type == REDIS_SET) {
                 /* Emit the SADDs needed to rebuild the set */
                 }
             } else if (o->type == REDIS_SET) {
                 /* Emit the SADDs needed to rebuild the set */


@@ -6847,8 +7759,8 @@ static int rewriteAppendOnlyFile(char *filename) {
                     robj *eleobj = dictGetEntryKey(de);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                     robj *eleobj = dictGetEntryKey(de);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;

-                    if (fwriteBulk(fp,key) == 0) goto werr;
-                    if (fwriteBulk(fp,eleobj) == 0) goto werr;
+                    if (fwriteBulkObject(fp,key) == 0) goto werr;
+                    if (fwriteBulkObject(fp,eleobj) == 0) goto werr;

                 }
                 dictReleaseIterator(di);
             } else if (o->type == REDIS_ZSET) {
                 }
                 dictReleaseIterator(di);
             } else if (o->type == REDIS_ZSET) {


@@ -6863,11 +7775,43 @@ static int rewriteAppendOnlyFile(char *filename) {
                     double *score = dictGetEntryVal(de);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                     double *score = dictGetEntryVal(de);
 
                     if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;

-                    if (fwriteBulk(fp,key) == 0) goto werr;
+                    if (fwriteBulkObject(fp,key) == 0) goto werr;

                     if (fwriteBulkDouble(fp,*score) == 0) goto werr;
                     if (fwriteBulkDouble(fp,*score) == 0) goto werr;

-                    if (fwriteBulk(fp,eleobj) == 0) goto werr;
+                    if (fwriteBulkObject(fp,eleobj) == 0) goto werr;

                 }
                 dictReleaseIterator(di);
                 }
                 dictReleaseIterator(di);

+            } else if (o->type == REDIS_HASH) {
+                char cmd[]="*4\r\n$4\r\nHSET\r\n";
+
+                /* Emit the HSETs needed to rebuild the hash */
+                if (o->encoding == REDIS_ENCODING_ZIPMAP) {
+                    unsigned char *p = zipmapRewind(o->ptr);
+                    unsigned char *field, *val;
+                    unsigned int flen, vlen;
+
+                    while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) {
+                        if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
+                        if (fwriteBulkObject(fp,key) == 0) goto werr;
+                        if (fwriteBulkString(fp,(char*)field,flen) == -1)
+                            return -1;
+                        if (fwriteBulkString(fp,(char*)val,vlen) == -1)
+                            return -1;
+                    }
+                } else {
+                    dictIterator *di = dictGetIterator(o->ptr);
+                    dictEntry *de;
+
+                    while((de = dictNext(di)) != NULL) {
+                        robj *field = dictGetEntryKey(de);
+                        robj *val = dictGetEntryVal(de);
+
+                        if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
+                        if (fwriteBulkObject(fp,key) == 0) goto werr;
+                        if (fwriteBulkObject(fp,field) == -1) return -1;
+                        if (fwriteBulkObject(fp,val) == -1) return -1;
+                    }
+                    dictReleaseIterator(di);
+                }

             } else {
                 redisAssert(0 != 0);
             }
             } else {
                 redisAssert(0 != 0);
             }


@@ -6877,7 +7821,7 @@ static int rewriteAppendOnlyFile(char *filename) {
                 /* If this key is already expired skip it */
                 if (expiretime < now) continue;
                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
                 /* If this key is already expired skip it */
                 if (expiretime < now) continue;
                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;

-                if (fwriteBulk(fp,key) == 0) goto werr;
+                if (fwriteBulkObject(fp,key) == 0) goto werr;

                 if (fwriteBulkLong(fp,expiretime) == 0) goto werr;
             }
             if (swapped) decrRefCount(o);
                 if (fwriteBulkLong(fp,expiretime) == 0) goto werr;
             }
             if (swapped) decrRefCount(o);


@@ -6933,9 +7877,9 @@ static int rewriteAppendOnlyFileBackground(void) {
         close(server.fd);
         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
         close(server.fd);
         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {

-            exit(0);
+            _exit(0);

         } else {
         } else {

-            exit(1);
+            _exit(1);

         }
     } else {
         /* Parent */
         }
     } else {
         /* Parent */


@@ -7061,7 +8005,7 @@ static void vmInit(void) {
     server.io_newjobs = listCreate();
     server.io_processing = listCreate();
     server.io_processed = listCreate();
     server.io_newjobs = listCreate();
     server.io_processing = listCreate();
     server.io_processed = listCreate();

-    server.io_clients = listCreate();
+    server.io_ready_clients = listCreate();

     pthread_mutex_init(&server.io_mutex,NULL);
     pthread_mutex_init(&server.obj_freelist_mutex,NULL);
     pthread_mutex_init(&server.io_swapfile_mutex,NULL);
     pthread_mutex_init(&server.io_mutex,NULL);
     pthread_mutex_init(&server.obj_freelist_mutex,NULL);
     pthread_mutex_init(&server.io_swapfile_mutex,NULL);


@@ -7091,8 +8035,6 @@ static void vmMarkPageUsed(off_t page) {
     int bit = page&7;
     redisAssert(vmFreePage(page) == 1);
     server.vm_bitmap[byte] |= 1<<bit;
     int bit = page&7;
     redisAssert(vmFreePage(page) == 1);
     server.vm_bitmap[byte] |= 1<<bit;

-    redisLog(REDIS_DEBUG,"Mark used: %lld (byte:%lld bit:%d)\n",
-        (long long)page, (long long)byte, bit);

 }
 
 /* Mark N contiguous pages as used, with 'page' being the first. */
 }
 
 /* Mark N contiguous pages as used, with 'page' being the first. */


@@ -7102,6 +8044,8 @@ static void vmMarkPagesUsed(off_t page, off_t count) {
     for (j = 0; j < count; j++)
         vmMarkPageUsed(page+j);
     server.vm_stats_used_pages += count;
     for (j = 0; j < count; j++)
         vmMarkPageUsed(page+j);
     server.vm_stats_used_pages += count;

+    redisLog(REDIS_DEBUG,"Mark USED pages: %lld pages at %lld\n",
+        (long long)count, (long long)page);

 }
 
 /* Mark the page as free */
 }
 
 /* Mark the page as free */


@@ -7110,8 +8054,6 @@ static void vmMarkPageFree(off_t page) {
     int bit = page&7;
     redisAssert(vmFreePage(page) == 0);
     server.vm_bitmap[byte] &= ~(1<<bit);
     int bit = page&7;
     redisAssert(vmFreePage(page) == 0);
     server.vm_bitmap[byte] &= ~(1<<bit);

-    redisLog(REDIS_DEBUG,"Mark free: %lld (byte:%lld bit:%d)\n",
-        (long long)page, (long long)byte, bit);

 }
 
 /* Mark N contiguous pages as free, with 'page' being the first. */
 }
 
 /* Mark N contiguous pages as free, with 'page' being the first. */


@@ -7121,9 +8063,8 @@ static void vmMarkPagesFree(off_t page, off_t count) {
     for (j = 0; j < count; j++)
         vmMarkPageFree(page+j);
     server.vm_stats_used_pages -= count;
     for (j = 0; j < count; j++)
         vmMarkPageFree(page+j);
     server.vm_stats_used_pages -= count;

-    if (server.vm_stats_used_pages > 100000000) {
-        *((char*)-1) = 'x';
-    }
+    redisLog(REDIS_DEBUG,"Mark FREE pages: %lld pages at %lld\n",
+        (long long)count, (long long)page);

 }
 
 /* Test if the page is free */
 }
 
 /* Test if the page is free */


@@ -7174,7 +8115,6 @@ static int vmFindContiguousPages(off_t *first, off_t n) {
                 numfree = 0;
             }
         }
                 numfree = 0;
             }
         }

-        redisLog(REDIS_DEBUG, "THIS: %lld (%c)\n", (long long) this, vmFreePage(this) ? 'F' : 'X');

         if (vmFreePage(this)) {
             /* This is a free page */
             numfree++;
         if (vmFreePage(this)) {
             /* This is a free page */
             numfree++;


@@ -7182,6 +8122,7 @@ static int vmFindContiguousPages(off_t *first, off_t n) {
             if (numfree == n) {
                 *first = this-(n-1);
                 server.vm_next_page = this+1;
             if (numfree == n) {
                 *first = this-(n-1);
                 server.vm_next_page = this+1;

+                redisLog(REDIS_DEBUG, "FOUND CONTIGUOUS PAGES: %lld pages at %lld\n", (long long) n, (long long) *first);

                 return REDIS_OK;
             }
         } else {
                 return REDIS_OK;
             }
         } else {


@@ -7212,11 +8153,12 @@ static int vmWriteObjectOnSwap(robj *o, off_t page) {
     if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
         if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
         redisLog(REDIS_WARNING,
     if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
         if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
         redisLog(REDIS_WARNING,

-            "Critical VM problem in vmSwapObjectBlocking(): can't seek: %s",
+            "Critical VM problem in vmWriteObjectOnSwap(): can't seek: %s",

             strerror(errno));
         return REDIS_ERR;
     }
     rdbSaveObject(server.vm_fp,o);
             strerror(errno));
         return REDIS_ERR;
     }
     rdbSaveObject(server.vm_fp,o);

+    fflush(server.vm_fp);

     if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
     return REDIS_OK;
 }
     if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
     return REDIS_OK;
 }


@@ -7244,7 +8186,6 @@ static int vmSwapObjectBlocking(robj *key, robj *val) {
         (unsigned long long) page, (unsigned long long) pages);
     server.vm_stats_swapped_objects++;
     server.vm_stats_swapouts++;
         (unsigned long long) page, (unsigned long long) pages);
     server.vm_stats_swapped_objects++;
     server.vm_stats_swapouts++;

-    fflush(server.vm_fp);

     return REDIS_OK;
 }
 
     return REDIS_OK;
 }
 


@@ -7254,14 +8195,14 @@ static robj *vmReadObjectFromSwap(off_t page, int type) {
     if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex);
     if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
         redisLog(REDIS_WARNING,
     if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex);
     if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
         redisLog(REDIS_WARNING,

-            "Unrecoverable VM problem in vmLoadObject(): can't seek: %s",
+            "Unrecoverable VM problem in vmReadObjectFromSwap(): can't seek: %s",

             strerror(errno));
             strerror(errno));

-        exit(1);
+        _exit(1);

     }
     o = rdbLoadObject(type,server.vm_fp);
     if (o == NULL) {
     }
     o = rdbLoadObject(type,server.vm_fp);
     if (o == NULL) {

-        redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmLoadObject(): can't load object from swap file: %s", strerror(errno));
-        exit(1);
+        redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmReadObjectFromSwap(): can't load object from swap file: %s", strerror(errno));
+        _exit(1);

     }
     if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
     return o;
     }
     if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
     return o;


@@ -7275,7 +8216,7 @@ static robj *vmReadObjectFromSwap(off_t page, int type) {
 static robj *vmGenericLoadObject(robj *key, int preview) {
     robj *val;
 
 static robj *vmGenericLoadObject(robj *key, int preview) {
     robj *val;
 

-    redisAssert(key->storage == REDIS_VM_SWAPPED);
+    redisAssert(key->storage == REDIS_VM_SWAPPED || key->storage == REDIS_VM_LOADING);

     val = vmReadObjectFromSwap(key->vm.page,key->vtype);
     if (!preview) {
         key->storage = REDIS_VM_MEMORY;
     val = vmReadObjectFromSwap(key->vm.page,key->vtype);
     if (!preview) {
         key->storage = REDIS_VM_MEMORY;


@@ -7373,7 +8314,7 @@ static double computeObjectSwappability(robj *o) {
         }
         break;
     }
         }
         break;
     }

-    return (double)asize*log(1+asize);
+    return (double)age*log(1+asize);

 }
 
 /* Try to swap an object that's a good candidate for swapping.
 }
 
 /* Try to swap an object that's a good candidate for swapping.


@@ -7424,10 +8365,7 @@ static int vmSwapOneObject(int usethreads) {
             }
         }
     }
             }
         }
     }

-    if (best == NULL) {
-        redisLog(REDIS_DEBUG,"No swappable key found!");
-        return REDIS_ERR;
-    }
+    if (best == NULL) return REDIS_ERR;

     key = dictGetEntryKey(best);
     val = dictGetEntryVal(best);
 
     key = dictGetEntryKey(best);
     val = dictGetEntryVal(best);
 


@@ -7485,8 +8423,9 @@ static int deleteIfSwapped(redisDb *db, robj *key) {
 /* =================== Virtual Memory - Threaded I/O  ======================= */
 
 static void freeIOJob(iojob *j) {
 /* =================== Virtual Memory - Threaded I/O  ======================= */
 
 static void freeIOJob(iojob *j) {

-    if (j->type == REDIS_IOJOB_PREPARE_SWAP ||
-        j->type == REDIS_IOJOB_DO_SWAP)
+    if ((j->type == REDIS_IOJOB_PREPARE_SWAP ||
+        j->type == REDIS_IOJOB_DO_SWAP ||
+        j->type == REDIS_IOJOB_LOAD) && j->val != NULL)

         decrRefCount(j->val);
     decrRefCount(j->key);
     zfree(j);
         decrRefCount(j->val);
     decrRefCount(j->key);
     zfree(j);


@@ -7537,6 +8476,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
         assert(de != NULL);
         key = dictGetEntryKey(de);
         if (j->type == REDIS_IOJOB_LOAD) {
         assert(de != NULL);
         key = dictGetEntryKey(de);
         if (j->type == REDIS_IOJOB_LOAD) {

+            redisDb *db;
+

             /* Key loaded, bring it at home */
             key->storage = REDIS_VM_MEMORY;
             key->vm.atime = server.unixtime;
             /* Key loaded, bring it at home */
             key->storage = REDIS_VM_MEMORY;
             key->vm.atime = server.unixtime;


@@ -7545,7 +8486,12 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
                 (unsigned char*) key->ptr);
             server.vm_stats_swapped_objects--;
             server.vm_stats_swapins++;
                 (unsigned char*) key->ptr);
             server.vm_stats_swapped_objects--;
             server.vm_stats_swapins++;

+            dictGetEntryVal(de) = j->val;
+            incrRefCount(j->val);
+            db = j->db;

             freeIOJob(j);
             freeIOJob(j);

+            /* Handle clients waiting for this key to be loaded. */
+            handleClientsBlockedOnSwappedKey(db,key);

         } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
             /* Now we know the amount of pages required to swap this object.
              * Let's find some space for it, and queue this task again
         } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
             /* Now we know the amount of pages required to swap this object.
              * Let's find some space for it, and queue this task again


@@ -7671,23 +8617,25 @@ again:
                     listDelNode(lists[i],ln);
                     break;
                 case 1: /* io_processing */
                     listDelNode(lists[i],ln);
                     break;
                 case 1: /* io_processing */

-                    /* Oh Shi- the thread is messing with the Job, and
-                     * probably with the object if this is a
-                     * PREPARE_SWAP or DO_SWAP job. Better to wait for the
-                     * job to move into the next queue... */
-                    if (job->type != REDIS_IOJOB_LOAD) {
-                        /* Yes, we try again and again until the job
-                         * is completed. */
-                        unlockThreadedIO();
-                        /* But let's wait some time for the I/O thread
-                         * to finish with this job. After all this condition
-                         * should be very rare. */
-                        usleep(1);
-                        goto again;
-                    } else {
-                        job->canceled = 1;
-                        break;
-                    }
+                    /* Oh Shi- the thread is messing with the Job:
+                     *
+                     * Probably it's accessing the object if this is a
+                     * PREPARE_SWAP or DO_SWAP job.
+                     * If it's a LOAD job it may be reading from disk and
+                     * if we don't wait for the job to terminate before to
+                     * cancel it, maybe in a few microseconds data can be
+                     * corrupted in this pages. So the short story is:
+                     *
+                     * Better to wait for the job to move into the
+                     * next queue (processed)... */
+
+                    /* We try again and again until the job is completed. */
+                    unlockThreadedIO();
+                    /* But let's wait some time for the I/O thread
+                     * to finish with this job. After all this condition
+                     * should be very rare. */
+                    usleep(1);
+                    goto again;

                 case 2: /* io_processed */
                     /* The job was already processed, that's easy...
                      * just mark it as canceled so that we'll ignore it
                 case 2: /* io_processed */
                     /* The job was already processed, that's easy...
                      * just mark it as canceled so that we'll ignore it


@@ -7721,8 +8669,8 @@ static void *IOThreadEntryPoint(void *arg) {
         lockThreadedIO();
         if (listLength(server.io_newjobs) == 0) {
             /* No new jobs in queue, exit. */
         lockThreadedIO();
         if (listLength(server.io_newjobs) == 0) {
             /* No new jobs in queue, exit. */

-            redisLog(REDIS_DEBUG,"Thread %lld exiting, nothing to do",
-                (long long) pthread_self());
+            redisLog(REDIS_DEBUG,"Thread %ld exiting, nothing to do",
+                (long) pthread_self());

             server.io_active_threads--;
             unlockThreadedIO();
             return NULL;
             server.io_active_threads--;
             unlockThreadedIO();
             return NULL;


@@ -7735,11 +8683,12 @@ static void *IOThreadEntryPoint(void *arg) {
         listAddNodeTail(server.io_processing,j);
         ln = listLast(server.io_processing); /* We use ln later to remove it */
         unlockThreadedIO();
         listAddNodeTail(server.io_processing,j);
         ln = listLast(server.io_processing); /* We use ln later to remove it */
         unlockThreadedIO();

-        redisLog(REDIS_DEBUG,"Thread %lld got a new job (type %d): %p about key '%s'",
-            (long long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr);
+        redisLog(REDIS_DEBUG,"Thread %ld got a new job (type %d): %p about key '%s'",
+            (long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr);

 
         /* Process the Job */
         if (j->type == REDIS_IOJOB_LOAD) {
 
         /* Process the Job */
         if (j->type == REDIS_IOJOB_LOAD) {

+            j->val = vmReadObjectFromSwap(j->page,j->key->vtype);

         } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
             FILE *fp = fopen("/dev/null","w+");
             j->pages = rdbSavedObjectPages(j->val,fp);
         } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
             FILE *fp = fopen("/dev/null","w+");
             j->pages = rdbSavedObjectPages(j->val,fp);


@@ -7750,8 +8699,8 @@ static void *IOThreadEntryPoint(void *arg) {
         }
 
         /* Done: insert the job into the processed queue */
         }
 
         /* Done: insert the job into the processed queue */

-        redisLog(REDIS_DEBUG,"Thread %lld completed the job: %p (key %s)",
-            (long long) pthread_self(), (void*)j, (char*)j->key->ptr);
+        redisLog(REDIS_DEBUG,"Thread %ld completed the job: %p (key %s)",
+            (long) pthread_self(), (void*)j, (char*)j->key->ptr);

         lockThreadedIO();
         listDelNode(server.io_processing,ln);
         listAddNodeTail(server.io_processed,j);
         lockThreadedIO();
         listDelNode(server.io_processing,ln);
         listAddNodeTail(server.io_processed,j);


@@ -7765,8 +8714,15 @@ static void *IOThreadEntryPoint(void *arg) {
 
 static void spawnIOThread(void) {
     pthread_t thread;
 
 static void spawnIOThread(void) {
     pthread_t thread;

+    sigset_t mask, omask;

 
 

+    sigemptyset(&mask);
+    sigaddset(&mask,SIGCHLD);
+    sigaddset(&mask,SIGHUP);
+    sigaddset(&mask,SIGPIPE);
+    pthread_sigmask(SIG_SETMASK, &mask, &omask);

     pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL);
     pthread_create(&thread,&server.io_threads_attr,IOThreadEntryPoint,NULL);

+    pthread_sigmask(SIG_SETMASK, &omask, NULL);

     server.io_active_threads++;
 }
 
     server.io_active_threads++;
 }
 


@@ -7800,12 +8756,13 @@ static void waitEmptyIOJobsQueue(void) {
 }
 
 static void vmReopenSwapFile(void) {
 }
 
 static void vmReopenSwapFile(void) {

-    fclose(server.vm_fp);
+    /* Note: we don't close the old one as we are in the child process
+     * and don't want to mess at all with the original file object. */

     server.vm_fp = fopen(server.vm_swap_file,"r+b");
     if (server.vm_fp == NULL) {
         redisLog(REDIS_WARNING,"Can't re-open the VM swap file: %s. Exiting.",
             server.vm_swap_file);
     server.vm_fp = fopen(server.vm_swap_file,"r+b");
     if (server.vm_fp == NULL) {
         redisLog(REDIS_WARNING,"Can't re-open the VM swap file: %s. Exiting.",
             server.vm_swap_file);

-        exit(1);
+        _exit(1);

     }
     server.vm_fd = fileno(server.vm_fp);
 }
     }
     server.vm_fd = fileno(server.vm_fp);
 }


@@ -7843,16 +8800,158 @@ static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
 
 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
 
 
 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
 

+/* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
+ * If there is not already a job loading the key, it is craeted.
+ * The key is added to the io_keys list in the client structure, and also
+ * in the hash table mapping swapped keys to waiting clients, that is,
+ * server.io_waited_keys. */
+static int waitForSwappedKey(redisClient *c, robj *key) {
+    struct dictEntry *de;
+    robj *o;
+    list *l;
+
+    /* If the key does not exist or is already in RAM we don't need to
+     * block the client at all. */
+    de = dictFind(c->db->dict,key);
+    if (de == NULL) return 0;
+    o = dictGetEntryKey(de);
+    if (o->storage == REDIS_VM_MEMORY) {
+        return 0;
+    } else if (o->storage == REDIS_VM_SWAPPING) {
+        /* We were swapping the key, undo it! */
+        vmCancelThreadedIOJob(o);
+        return 0;
+    }
+    
+    /* OK: the key is either swapped, or being loaded just now. */
+
+    /* Add the key to the list of keys this client is waiting for.
+     * This maps clients to keys they are waiting for. */
+    listAddNodeTail(c->io_keys,key);
+    incrRefCount(key);
+
+    /* Add the client to the swapped keys => clients waiting map. */
+    de = dictFind(c->db->io_keys,key);
+    if (de == NULL) {
+        int retval;
+
+        /* For every key we take a list of clients blocked for it */
+        l = listCreate();
+        retval = dictAdd(c->db->io_keys,key,l);
+        incrRefCount(key);
+        assert(retval == DICT_OK);
+    } else {
+        l = dictGetEntryVal(de);
+    }
+    listAddNodeTail(l,c);
+
+    /* Are we already loading the key from disk? If not create a job */
+    if (o->storage == REDIS_VM_SWAPPED) {
+        iojob *j;
+
+        o->storage = REDIS_VM_LOADING;
+        j = zmalloc(sizeof(*j));
+        j->type = REDIS_IOJOB_LOAD;
+        j->db = c->db;
+        j->key = dupStringObject(key);
+        j->key->vtype = o->vtype;
+        j->page = o->vm.page;
+        j->val = NULL;
+        j->canceled = 0;
+        j->thread = (pthread_t) -1;
+        lockThreadedIO();
+        queueIOJob(j);
+        unlockThreadedIO();
+    }
+    return 1;
+}
+

 /* Is this client attempting to run a command against swapped keys?
 /* Is this client attempting to run a command against swapped keys?

- * If so, block it ASAP, load the keys in background, then resume it.4
+ * If so, block it ASAP, load the keys in background, then resume it.
+ *
+ * The important idea about this function is that it can fail! If keys will
+ * still be swapped when the client is resumed, this key lookups will
+ * just block loading keys from disk. In practical terms this should only
+ * happen with SORT BY command or if there is a bug in this function.

  *
  *

- * The improtat thing about this function is that it can fail! If keys will
- * still be swapped when the client is resumed, a few of key lookups will
- * just block loading keys from disk. */
-#if 0
-static void blockClientOnSwappedKeys(redisClient *c) {
+ * Return 1 if the client is marked as blocked, 0 if the client can
+ * continue as the keys it is going to access appear to be in memory. */
+static int blockClientOnSwappedKeys(struct redisCommand *cmd, redisClient *c) {
+    int j, last;
+
+    if (cmd->vm_firstkey == 0) return 0;
+    last = cmd->vm_lastkey;
+    if (last < 0) last = c->argc+last;
+    for (j = cmd->vm_firstkey; j <= last; j += cmd->vm_keystep)
+        waitForSwappedKey(c,c->argv[j]);
+    /* If the client was blocked for at least one key, mark it as blocked. */
+    if (listLength(c->io_keys)) {
+        c->flags |= REDIS_IO_WAIT;
+        aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
+        server.vm_blocked_clients++;
+        return 1;
+    } else {
+        return 0;
+    }
+}
+
+/* Remove the 'key' from the list of blocked keys for a given client.
+ *
+ * The function returns 1 when there are no longer blocking keys after
+ * the current one was removed (and the client can be unblocked). */
+static int dontWaitForSwappedKey(redisClient *c, robj *key) {
+    list *l;
+    listNode *ln;
+    listIter li;
+    struct dictEntry *de;
+
+    /* Remove the key from the list of keys this client is waiting for. */
+    listRewind(c->io_keys,&li);
+    while ((ln = listNext(&li)) != NULL) {
+        if (compareStringObjects(ln->value,key) == 0) {
+            listDelNode(c->io_keys,ln);
+            break;
+        }
+    }
+    assert(ln != NULL);
+
+    /* Remove the client form the key => waiting clients map. */
+    de = dictFind(c->db->io_keys,key);
+    assert(de != NULL);
+    l = dictGetEntryVal(de);
+    ln = listSearchKey(l,c);
+    assert(ln != NULL);
+    listDelNode(l,ln);
+    if (listLength(l) == 0)
+        dictDelete(c->db->io_keys,key);
+
+    return listLength(c->io_keys) == 0;
+}
+
+static void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key) {
+    struct dictEntry *de;
+    list *l;
+    listNode *ln;
+    int len;
+
+    de = dictFind(db->io_keys,key);
+    if (!de) return;
+
+    l = dictGetEntryVal(de);
+    len = listLength(l);
+    /* Note: we can't use something like while(listLength(l)) as the list
+     * can be freed by the calling function when we remove the last element. */
+    while (len--) {
+        ln = listFirst(l);
+        redisClient *c = ln->value;
+
+        if (dontWaitForSwappedKey(c,key)) {
+            /* Put the client in the list of clients ready to go as we
+             * loaded all the keys about it. */
+            listAddNodeTail(server.io_ready_clients,c);
+        }
+    }

 }
 }

-#endif

 
 /* ================================= Debugging ============================== */
 
 
 /* ================================= Debugging ============================== */
 


@@ -7889,13 +8988,22 @@ static void debugCommand(redisClient *c) {
         }
         key = dictGetEntryKey(de);
         val = dictGetEntryVal(de);
         }
         key = dictGetEntryKey(de);
         val = dictGetEntryVal(de);

-        if (server.vm_enabled && (key->storage == REDIS_VM_MEMORY ||
-                                  key->storage == REDIS_VM_SWAPPING)) {
+        if (!server.vm_enabled || (key->storage == REDIS_VM_MEMORY ||
+                                   key->storage == REDIS_VM_SWAPPING)) {
+            char *strenc;
+            char buf[128];
+
+            if (val->encoding < (sizeof(strencoding)/sizeof(char*))) {
+                strenc = strencoding[val->encoding];
+            } else {
+                snprintf(buf,64,"unknown encoding %d\n", val->encoding);
+                strenc = buf;
+            }

             addReplySds(c,sdscatprintf(sdsempty(),
                 "+Key at:%p refcount:%d, value at:%p refcount:%d "
             addReplySds(c,sdscatprintf(sdsempty(),
                 "+Key at:%p refcount:%d, value at:%p refcount:%d "

-                "encoding:%d serializedlength:%lld\r\n",
+                "encoding:%s serializedlength:%lld\r\n",

                 (void*)key, key->refcount, (void*)val, val->refcount,
                 (void*)key, key->refcount, (void*)val, val->refcount,

-                val->encoding, (long long) rdbSavedObjectLen(val,NULL)));
+                strenc, (long long) rdbSavedObjectLen(val,NULL)));

         } else {
             addReplySds(c,sdscatprintf(sdsempty(),
                 "+Key at:%p refcount:%d, value swapped at: page %llu "
         } else {
             addReplySds(c,sdscatprintf(sdsempty(),
                 "+Key at:%p refcount:%d, value swapped at: page %llu "


@@ -7996,6 +9104,8 @@ static void daemonize(void) {
 }
 
 int main(int argc, char **argv) {
 }
 
 int main(int argc, char **argv) {

+    time_t start;
+

     initServerConfig();
     if (argc == 2) {
         resetServerSaveParams();
     initServerConfig();
     if (argc == 2) {
         resetServerSaveParams();


@@ -8012,14 +9122,16 @@ int main(int argc, char **argv) {
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
 #endif
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
 #endif

+    start = time(NULL);

     if (server.appendonly) {
         if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
     if (server.appendonly) {
         if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)

-            redisLog(REDIS_NOTICE,"DB loaded from append only file");
+            redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);

     } else {
         if (rdbLoad(server.dbfilename) == REDIS_OK)
     } else {
         if (rdbLoad(server.dbfilename) == REDIS_OK)

-            redisLog(REDIS_NOTICE,"DB loaded from disk");
+            redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start);

     }
     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
     }
     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);

+    aeSetBeforeSleepProc(server.el,beforeSleep);

     aeMain(server.el);
     aeDeleteEventLoop(server.el);
     return 0;
     aeMain(server.el);
     aeDeleteEventLoop(server.el);
     return 0;


@@ -8047,7 +9159,7 @@ static void *getMcontextEip(ucontext_t *uc) {
   #else
     return (void*) uc->uc_mcontext->__ss.__eip;
   #endif 
   #else
     return (void*) uc->uc_mcontext->__ss.__eip;
   #endif 

-#elif defined(__i386__) || defined(__X86_64__)  || defined(__x86_64__)
+#elif defined(__i386__) || defined(__X86_64__) || defined(__x86_64__)

     return (void*) uc->uc_mcontext.gregs[REG_EIP]; /* Linux 32/64 bit */
 #elif defined(__ia64__) /* Linux IA64 */
     return (void*) uc->uc_mcontext.sc_ip;
     return (void*) uc->uc_mcontext.gregs[REG_EIP]; /* Linux 32/64 bit */
 #elif defined(__ia64__) /* Linux IA64 */
     return (void*) uc->uc_mcontext.sc_ip;


@@ -8090,7 +9202,7 @@ static void segvHandler(int sig, siginfo_t *info, void *secret) {
         }
     }
     /* free(messages); Don't call free() with possibly corrupted memory. */
         }
     }
     /* free(messages); Don't call free() with possibly corrupted memory. */

-    exit(0);
+    _exit(0);

 }
 
 static void setupSigSegvAction(void) {
 }
 
 static void setupSigSegvAction(void) {