X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/f870935d3e26fe79766f15bbaa5358c5379e390e..59146ef3c107d59b6522b8a2f5539a55703adee1:/redis.c

diff --git a/redis.c b/redis.c
index 084077bb..5da103d8 100644
--- a/redis.c
+++ b/redis.c
@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
-#define REDIS_VERSION "1.3.2"
+#define REDIS_VERSION "1.3.3"
 
 #include "fmacros.h"
 #include "config.h"
@@ -59,6 +59,7 @@
 #include <sys/uio.h>
 #include <limits.h>
 #include <math.h>
+#include <pthread.h>
 
 #if defined(__sun)
 #include "solarisfixes.h"
@@ -162,14 +163,21 @@
  * Check vmFindContiguousPages() to know more about this magic numbers. */
 #define REDIS_VM_MAX_NEAR_PAGES 65536
 #define REDIS_VM_MAX_RANDOM_JUMP 4096
+#define REDIS_VM_MAX_THREADS 32
+#define REDIS_THREAD_STACK_SIZE (1024*1024*4)
+/* The following is the *percentage* of completed I/O jobs to process when the
+ * handelr is called. While Virtual Memory I/O operations are performed by
+ * threads, this operations must be processed by the main thread when completed
+ * in order to take effect. */
+#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_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 */
@@ -213,8 +221,8 @@
 #define APPENDFSYNC_EVERYSEC 2
 
 /* We can print the stacktrace, so our assert is defined this way: */
-#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e),exit(1)))
-static void _redisAssert(char *estr);
+#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
+static void _redisAssert(char *estr, char *file, int line);
 
 /*================================= Data types ============================== */
 
@@ -260,6 +268,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 *io_keys;              /* Keys with clients waiting for VM I/O */
     int id;
 } redisDb;
 
@@ -289,8 +298,7 @@ typedef struct redisClient {
     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 */
@@ -298,11 +306,13 @@ typedef struct redisClient {
     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
                              * is >= blockingto then the operation timed out. */
+    list *io_keys;          /* Keys this client is waiting to be loaded from the
+                             * swap file in order to continue. */
 } redisClient;
 
 struct saveparam {
@@ -325,7 +335,6 @@ struct redisServer {
     int cronloops;              /* number of times the cron function run */
     list *objfreelist;          /* A list of freed objects to avoid malloc() */
     time_t lastsave;            /* Unix time of last save succeeede */
-    size_t usedmemory;             /* Used memory in megabytes */
     /* Fields used only for stats */
     time_t stat_starttime;         /* server start time */
     long long stat_numcommands;    /* number of processed commands */
@@ -363,7 +372,8 @@ struct redisServer {
     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;
@@ -371,6 +381,7 @@ struct redisServer {
     int sort_bypattern;
     /* Virtual memory configuration */
     int vm_enabled;
+    char *vm_swap_file;
     off_t vm_page_size;
     off_t vm_pages;
     unsigned long long vm_max_memory;
@@ -381,6 +392,32 @@ struct redisServer {
     off_t vm_near_pages; /* Number of pages allocated sequentially */
     unsigned char *vm_bitmap; /* Bitmap of free/used pages */
     time_t unixtime;    /* Unix time sampled every second. */
+    /* Virtual memory I/O threads stuff */
+    /* An I/O thread process an element taken from the io_jobs queue and
+     * put the result of the operation in the io_done list. While the
+     * job is being processed, it's put on io_processing queue. */
+    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_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_attr_t io_threads_attr; /* attributes for threads creation */
+    int io_active_threads; /* Number of running I/O threads */
+    int vm_max_threads; /* Max number of I/O threads running at the same time */
+    /* Our main thread is blocked on the event loop, locking for sockets ready
+     * to be read or written, so when a threaded I/O operation is ready to be
+     * processed by the main thread, the I/O thread will use a unix pipe to
+     * awake the main thread. The followings are the two pipe FDs. */
+    int io_ready_pipe_read;
+    int io_ready_pipe_write;
+    /* Virtual memory stats */
+    unsigned long long vm_stats_used_pages;
+    unsigned long long vm_stats_swapped_objects;
+    unsigned long long vm_stats_swapouts;
+    unsigned long long vm_stats_swapins;
+    FILE *devnull;
 };
 
 typedef void redisCommandProc(redisClient *c);
@@ -446,6 +483,22 @@ struct sharedObjectsStruct {
 
 static double R_Zero, R_PosInf, R_NegInf, R_Nan;
 
+/* VM threaded I/O request message */
+#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 iojob {
+    int type;   /* Request type, REDIS_IOJOB_* */
+    redisDb *db;/* Redis database */
+    robj *key;  /* This I/O request is about swapping this key */
+    robj *val;  /* the value to swap for REDIS_IOREQ_*_SWAP, otherwise this
+                 * field is populated by the I/O thread for REDIS_IOREQ_LOAD. */
+    off_t page; /* Swap page where to read/write the object */
+    off_t pages; /* Swap pages needed to safe object. PREPARE_SWAP return val */
+    int canceled; /* True if this command was canceled by blocking side of VM */
+    pthread_t thread; /* ID of the thread processing this entry */
+} iojob;
+
 /*================================ Prototypes =============================== */
 
 static void freeStringObject(robj *o);
@@ -489,15 +542,36 @@ static void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int
 static void initClientMultiState(redisClient *c);
 static void freeClientMultiState(redisClient *c);
 static void queueMultiCommand(redisClient *c, struct redisCommand *cmd);
-static void unblockClient(redisClient *c);
+static void unblockClientWaitingData(redisClient *c);
 static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele);
 static void vmInit(void);
 static void vmMarkPagesFree(off_t page, off_t count);
 static robj *vmLoadObject(robj *key);
 static robj *vmPreviewObject(robj *key);
-static int vmSwapOneObject(void);
+static int vmSwapOneObjectBlocking(void);
+static int vmSwapOneObjectThreaded(void);
 static int vmCanSwapOut(void);
-static void freeOneObjectFromFreelist(void);
+static int tryFreeOneObjectFromFreelist(void);
+static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
+static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, int mask);
+static void vmCancelThreadedIOJob(robj *o);
+static void lockThreadedIO(void);
+static void unlockThreadedIO(void);
+static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db);
+static void freeIOJob(iojob *j);
+static void queueIOJob(iojob *j);
+static int vmWriteObjectOnSwap(robj *o, off_t page);
+static robj *vmReadObjectFromSwap(off_t page, int type);
+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 authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -576,6 +650,7 @@ static void multiCommand(redisClient *c);
 static void execCommand(redisClient *c);
 static void blpopCommand(redisClient *c);
 static void brpopCommand(redisClient *c);
+static void appendCommand(redisClient *c);
 
 /*================================= Globals ================================= */
 
@@ -585,6 +660,7 @@ 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},
+    {"append",appendCommand,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},
@@ -602,7 +678,7 @@ static struct redisCommand cmdTable[] = {
     {"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
     {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
     {"lrem",lremCommand,4,REDIS_CMD_BULK},
-    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
+    {"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},
@@ -797,13 +873,13 @@ static void redisLog(int level, const char *fmt, ...) {
 
     va_start(ap, fmt);
     if (level >= server.verbosity) {
-        char *c = ".-*";
+        char *c = ".-*#";
         char buf[64];
         time_t now;
 
         now = time(NULL);
         strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
-        fprintf(fp,"%s %c ",buf,c[level]);
+        fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
         vfprintf(fp, fmt, ap);
         fprintf(fp,"\n");
         fflush(fp);
@@ -880,12 +956,27 @@ static int dictEncObjKeyCompare(void *privdata, const void *key1,
 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 */
 static dictType setDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */
@@ -895,6 +986,7 @@ static dictType setDictType = {
     NULL                       /* val destructor */
 };
 
+/* Sorted sets hash (note: a skiplist is used in addition to the hash table) */
 static dictType zsetDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */
@@ -904,6 +996,7 @@ static dictType zsetDictType = {
     dictVanillaFree            /* val destructor of malloc(sizeof(double)) */
 };
 
+/* Db->dict */
 static dictType hashDictType = {
     dictObjHash,                /* hash function */
     NULL,                       /* key dup */
@@ -913,8 +1006,19 @@ static dictType hashDictType = {
     dictRedisObjectDestructor   /* val destructor */
 };
 
+/* Db->expires */
+static dictType keyptrDictType = {
+    dictObjHash,               /* hash function */
+    NULL,                      /* key dup */
+    NULL,                      /* val dup */
+    dictObjKeyCompare,         /* key compare */
+    dictRedisObjectDestructor, /* key destructor */
+    NULL                       /* val destructor */
+};
+
 /* 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 */
@@ -942,9 +1046,10 @@ static void closeTimedoutClients(void) {
     redisClient *c;
     listNode *ln;
     time_t now = time(NULL);
+    listIter li;
 
-    listRewind(server.clients);
-    while ((ln = listYield(server.clients)) != NULL) {
+    listRewind(server.clients,&li);
+    while ((ln = listNext(&li)) != NULL) {
         c = listNodeValue(ln);
         if (server.maxidletime &&
             !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
@@ -956,7 +1061,7 @@ static void closeTimedoutClients(void) {
         } else if (c->flags & REDIS_BLOCKED) {
             if (c->blockingto != 0 && c->blockingto < now) {
                 addReply(c,shared.nullmultibulk);
-                unblockClient(c);
+                unblockClientWaitingData(c);
             }
         }
     }
@@ -1083,9 +1188,6 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
      * To access a global var is faster than calling time(NULL) */
     server.unixtime = time(NULL);
 
-    /* Update the global state with the amount of used memory */
-    server.usedmemory = zmalloc_used_memory();
-
     /* Show some info about non-empty databases */
     for (j = 0; j < server.dbnum; j++) {
         long long size, used, vkeys;
@@ -1112,12 +1214,12 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         redisLog(REDIS_VERBOSE,"%d clients connected (%d slaves), %zu bytes in use, %d shared objects",
             listLength(server.clients)-listLength(server.slaves),
             listLength(server.slaves),
-            server.usedmemory,
+            zmalloc_used_memory(),
             dictSize(server.sharingpool));
     }
 
     /* 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 */
@@ -1186,15 +1288,23 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         while (server.vm_enabled && zmalloc_used_memory() >
                 server.vm_max_memory)
         {
-            if (listLength(server.objfreelist)) {
-                freeOneObjectFromFreelist();
-            } else if (vmSwapOneObject() == REDIS_ERR) {
-                if ((loops % 30) == 0 && zmalloc_used_memory() >
-                    (server.vm_max_memory+server.vm_max_memory/10)) {
-                    redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!");
-                }
-                break;
+            int retval;
+
+            if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
+            retval = (server.vm_max_threads == 0) ?
+                        vmSwapOneObjectBlocking() :
+                        vmSwapOneObjectThreaded();
+            if (retval == REDIS_ERR && (loops % 30) == 0 &&
+                zmalloc_used_memory() >
+                (server.vm_max_memory+server.vm_max_memory/10))
+            {
+                redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!");
             }
+            /* Note that when using threade I/O we free just one object,
+             * because anyway when the I/O thread in charge to swap this
+             * object out will finish, the handler of completed jobs
+             * will try to swap more objects if we are still out of memory. */
+            if (retval == REDIS_ERR || server.vm_max_threads > 0) break;
         }
     }
 
@@ -1208,6 +1318,38 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
     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"));
@@ -1281,12 +1423,15 @@ static void initServerConfig() {
     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.vm_page_size = 256;          /* 256 bytes per page */
     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;
 
     resetServerSaveParams();
 
@@ -1315,6 +1460,11 @@ static void initServer() {
     signal(SIGPIPE, SIG_IGN);
     setupSigSegvAction();
 
+    server.devnull = fopen("/dev/null","w");
+    if (server.devnull == NULL) {
+        redisLog(REDIS_WARNING, "Can't open /dev/null: %s", server.neterr);
+        exit(1);
+    }
     server.clients = listCreate();
     server.slaves = listCreate();
     server.monitors = listCreate();
@@ -1330,8 +1480,10 @@ static void initServer() {
     }
     for (j = 0; j < server.dbnum; j++) {
         server.db[j].dict = dictCreate(&hashDictType,NULL);
-        server.db[j].expires = dictCreate(&setDictType,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;
@@ -1340,12 +1492,13 @@ static void initServer() {
     server.bgrewritebuf = sdsempty();
     server.lastsave = time(NULL);
     server.dirty = 0;
-    server.usedmemory = 0;
     server.stat_numcommands = 0;
     server.stat_numconnections = 0;
     server.stat_starttime = time(NULL);
     server.unixtime = time(NULL);
     aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
+    if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
+        acceptHandler, NULL) == AE_ERR) oom("creating file event");
 
     if (server.appendonly) {
         server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
@@ -1519,21 +1672,26 @@ static void loadServerConfig(char *filename) {
                 goto loaderr;
             }
         } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
-          server.requirepass = zstrdup(argv[1]);
+            server.requirepass = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
-          server.pidfile = zstrdup(argv[1]);
+            server.pidfile = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
-          server.dbfilename = zstrdup(argv[1]);
+            server.dbfilename = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"vm-enabled") && argc == 2) {
             if ((server.vm_enabled = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"vm-swap-file") && argc == 2) {
+            zfree(server.vm_swap_file);
+            server.vm_swap_file = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"vm-max-memory") && argc == 2) {
             server.vm_max_memory = strtoll(argv[1], NULL, 10);
         } else if (!strcasecmp(argv[0],"vm-page-size") && argc == 2) {
             server.vm_page_size = strtoll(argv[1], NULL, 10);
         } else if (!strcasecmp(argv[0],"vm-pages") && argc == 2) {
             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 {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
@@ -1568,23 +1726,38 @@ static void freeClient(redisClient *c) {
     listNode *ln;
 
     /* Note that if the client we are freeing is blocked into a blocking
-     * call, we have to set querybuf to NULL *before* to call unblockClient()
-     * to avoid processInputBuffer() will get called. Also it is important
-     * to remove the file events after this, because this call adds
-     * the READABLE event. */
+     * call, we have to set querybuf to NULL *before* to call
+     * unblockClientWaitingData() to avoid processInputBuffer() will get
+     * called. Also it is important to remove the file events after
+     * this, because this call adds the READABLE event. */
     sdsfree(c->querybuf);
     c->querybuf = NULL;
     if (c->flags & REDIS_BLOCKED)
-        unblockClient(c);
+        unblockClientWaitingData(c);
 
     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
     listRelease(c->reply);
     freeClientArgv(c);
     close(c->fd);
+    /* Remove from the list of clients */
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);
+    /* 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 */
     if (c->flags & REDIS_SLAVE) {
         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
             close(c->repldbfd);
@@ -1608,10 +1781,11 @@ static void glueReplyBuffersIfNeeded(redisClient *c) {
     int copylen = 0;
     char buf[GLUEREPLY_UP_TO];
     listNode *ln;
+    listIter li;
     robj *o;
 
-    listRewind(c->reply);
-    while((ln = listYield(c->reply))) {
+    listRewind(c->reply,&li);
+    while((ln = listNext(&li))) {
         int objlen;
 
         o = ln->value;
@@ -1893,6 +2067,9 @@ static int processCommand(redisClient *c) {
         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,
@@ -1913,6 +2090,7 @@ static int processCommand(redisClient *c) {
         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]);
@@ -1934,6 +2112,8 @@ static int processCommand(redisClient *c) {
             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;
         }
     }
@@ -1959,20 +2139,19 @@ static int processCommand(redisClient *c) {
         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 */
-    if (c->flags & REDIS_CLOSE) {
-        freeClient(c);
-        return 0;
-    }
     resetClient(c);
     return 1;
 }
 
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     listNode *ln;
+    listIter li;
     int outc = 0, j;
     robj **outv;
     /* (args*2)+1 is enough room for args, spaces, newlines */
@@ -2003,8 +2182,8 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
      * be sure to free objects if there is no slave in a replication state
      * able to be feed with commands */
     for (j = 0; j < outc; j++) incrRefCount(outv[j]);
-    listRewind(slaves);
-    while((ln = listYield(slaves))) {
+    listRewind(slaves,&li);
+    while((ln = listNext(&li))) {
         redisClient *slave = ln->value;
 
         /* Don't feed slaves that are still waiting for BGSAVE to start */
@@ -2048,7 +2227,7 @@ again:
      * would not be called at all, but after the execution of the first commands
      * in the input buffer the client may be blocked, and the "goto again"
      * will try to reiterate. The following line will make it return asap. */
-    if (c->flags & REDIS_BLOCKED) return;
+    if (c->flags & REDIS_BLOCKED || c->flags & REDIS_IO_WAIT) return;
     if (c->bulklen == -1) {
         /* Read the first line of the query */
         char *p = strchr(c->querybuf,'\n');
@@ -2185,10 +2364,12 @@ static redisClient *createClient(int fd) {
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->reply = listCreate();
-    c->blockingkeys = NULL;
-    c->blockingkeysnum = 0;
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
+    c->blockingkeys = NULL;
+    c->blockingkeysnum = 0;
+    c->io_keys = listCreate();
+    listSetFreeMethod(c->io_keys,decrRefCount);
     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
         readQueryFromClient, c) == AE_ERR) {
         freeClient(c);
@@ -2289,12 +2470,15 @@ static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 static robj *createObject(int type, void *ptr) {
     robj *o;
 
+    if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
     if (listLength(server.objfreelist)) {
         listNode *head = listFirst(server.objfreelist);
         o = listNodeValue(head);
         listDelNode(server.objfreelist,head);
+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
     } else {
         if (server.vm_enabled) {
+            pthread_mutex_unlock(&server.obj_freelist_mutex);
             o = zmalloc(sizeof(*o));
         } else {
             o = zmalloc(sizeof(*o)-sizeof(struct redisObjectVM));
@@ -2305,6 +2489,10 @@ static robj *createObject(int type, void *ptr) {
     o->ptr = ptr;
     o->refcount = 1;
     if (server.vm_enabled) {
+        /* Note that this code may run in the context of an I/O thread
+         * and accessing to server.unixtime in theory is an error
+         * (no locks). But in practice this is safe, and even if we read
+         * garbage Redis will not fail, as it's just a statistical info */
         o->vm.atime = server.unixtime;
         o->storage = REDIS_VM_MEMORY;
     }
@@ -2316,6 +2504,7 @@ static robj *createStringObject(char *ptr, size_t len) {
 }
 
 static robj *dupStringObject(robj *o) {
+    assert(o->encoding == REDIS_ENCODING_RAW);
     return createStringObject(o->ptr,sdslen(o->ptr));
 }
 
@@ -2373,19 +2562,30 @@ static void incrRefCount(robj *o) {
 static void decrRefCount(void *obj) {
     robj *o = obj;
 
-    /* REDIS_VM_SWAPPED */
-    if (server.vm_enabled && o->storage == REDIS_VM_SWAPPED) {
-        redisAssert(o->refcount == 1);
+    /* Object is a key of a swapped out value, or in the process of being
+     * loaded. */
+    if (server.vm_enabled &&
+        (o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING))
+    {
+        if (o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING) {
+            redisAssert(o->refcount == 1);
+        }
+        if (o->storage == REDIS_VM_LOADING) vmCancelThreadedIOJob(obj);
         redisAssert(o->type == REDIS_STRING);
         freeStringObject(o);
         vmMarkPagesFree(o->vm.page,o->vm.usedpages);
+        pthread_mutex_lock(&server.obj_freelist_mutex);
         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
             !listAddNodeHead(server.objfreelist,o))
             zfree(o);
+        pthread_mutex_unlock(&server.obj_freelist_mutex);
+        server.vm_stats_swapped_objects--;
         return;
     }
-    /* REDIS_VM_MEMORY */
+    /* Object is in memory, or in the process of being swapped out. */
     if (--(o->refcount) == 0) {
+        if (server.vm_enabled && o->storage == REDIS_VM_SWAPPING)
+            vmCancelThreadedIOJob(obj);
         switch(o->type) {
         case REDIS_STRING: freeStringObject(o); break;
         case REDIS_LIST: freeListObject(o); break;
@@ -2394,9 +2594,11 @@ static void decrRefCount(void *obj) {
         case REDIS_HASH: freeHashObject(o); break;
         default: redisAssert(0 != 0); break;
         }
+        if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
             !listAddNodeHead(server.objfreelist,o))
             zfree(o);
+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
     }
 }
 
@@ -2407,14 +2609,26 @@ static robj *lookupKey(redisDb *db, robj *key) {
         robj *val = dictGetEntryVal(de);
 
         if (server.vm_enabled) {
-            if (key->storage == REDIS_VM_MEMORY) {
+            if (key->storage == REDIS_VM_MEMORY ||
+                key->storage == REDIS_VM_SWAPPING)
+            {
+                /* If we were swapping the object out, stop it, this key
+                 * was requested. */
+                if (key->storage == REDIS_VM_SWAPPING)
+                    vmCancelThreadedIOJob(key);
                 /* 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;
+
+                /* Clients blocked by the VM subsystem may be waiting for
+                 * this key... */
+                if (notify) handleClientsBlockedOnSwappedKey(db,key);
             }
         }
         return val;
@@ -2743,9 +2957,18 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
 static int rdbSaveStringObject(FILE *fp, robj *obj) {
     int retval;
 
-    obj = getDecodedObject(obj);
-    retval = rdbSaveStringObjectRaw(fp,obj);
-    decrRefCount(obj);
+    /* Avoid incr/decr ref count business when possible.
+     * This plays well with copy-on-write given that we are probably
+     * in a child process (BGSAVE). Also this makes sure key objects
+     * of swapped objects are not incRefCount-ed (an assert does not allow
+     * this in order to avoid bugs) */
+    if (obj->encoding != REDIS_ENCODING_RAW) {
+        obj = getDecodedObject(obj);
+        retval = rdbSaveStringObjectRaw(fp,obj);
+        decrRefCount(obj);
+    } else {
+        retval = rdbSaveStringObjectRaw(fp,obj);
+    }
     return retval;
 }
 
@@ -2784,11 +3007,12 @@ static int rdbSaveObject(FILE *fp, robj *o) {
     } else if (o->type == REDIS_LIST) {
         /* Save a list value */
         list *list = o->ptr;
+        listIter li;
         listNode *ln;
 
-        listRewind(list);
         if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
-        while((ln = listYield(list))) {
+        listRewind(list,&li);
+        while((ln = listNext(&li))) {
             robj *eleobj = listNodeValue(ln);
 
             if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
@@ -2831,20 +3055,16 @@ static int rdbSaveObject(FILE *fp, robj *o) {
  * the rdbSaveObject() function. Currently we use a trick to get
  * this length with very little changes to the code. In the future
  * we could switch to a faster solution. */
-static off_t rdbSavedObjectLen(robj *o) {
-    static FILE *fp = NULL;
-
-    if (fp == NULL) fp = fopen("/dev/null","w");
-    assert(fp != NULL);
-
+static off_t rdbSavedObjectLen(robj *o, FILE *fp) {
+    if (fp == NULL) fp = server.devnull;
     rewind(fp);
     assert(rdbSaveObject(fp,o) != 1);
     return ftello(fp);
 }
 
 /* Return the number of pages required to save this object in the swap file */
-static off_t rdbSavedObjectPages(robj *o) {
-    off_t bytes = rdbSavedObjectLen(o);
+static off_t rdbSavedObjectPages(robj *o, FILE *fp) {
+    off_t bytes = rdbSavedObjectLen(o,fp);
     
     return (bytes+(server.vm_page_size-1))/server.vm_page_size;
 }
@@ -2858,6 +3078,12 @@ static int rdbSave(char *filename) {
     int j;
     time_t now = time(NULL);
 
+    /* Wait for I/O therads to terminate, just in case this is a
+     * foreground-saving, to avoid seeking the swap file descriptor at the
+     * same time. */
+    if (server.vm_enabled)
+        waitEmptyIOJobsQueue();
+
     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
     fp = fopen(tmpfile,"w");
     if (!fp) {
@@ -2894,25 +3120,23 @@ static int rdbSave(char *filename) {
             }
             /* Save the key and associated value. This requires special
              * handling if the value is swapped out. */
-            if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY) {
+            if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY ||
+                                      key->storage == REDIS_VM_SWAPPING) {
                 /* Save type, key, value */
                 if (rdbSaveType(fp,o->type) == -1) goto werr;
                 if (rdbSaveStringObject(fp,key) == -1) goto werr;
                 if (rdbSaveObject(fp,o) == -1) goto werr;
             } else {
-                robj *po, *newkey;
+                /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
+                robj *po;
                 /* Get a preview of the object in memory */
                 po = vmPreviewObject(key);
-                /* Also duplicate the key object, to pass around a standard
-                 * string object. */
-                newkey = dupStringObject(key);
                 /* Save type, key, value */
                 if (rdbSaveType(fp,key->vtype) == -1) goto werr;
-                if (rdbSaveStringObject(fp,newkey) == -1) goto werr;
+                if (rdbSaveStringObject(fp,key) == -1) goto werr;
                 if (rdbSaveObject(fp,po) == -1) goto werr;
                 /* Remove the loaded object from memory */
                 decrRefCount(po);
-                decrRefCount(newkey);
             }
         }
         dictReleaseIterator(di);
@@ -2949,13 +3173,15 @@ static int rdbSaveBackground(char *filename) {
     pid_t childpid;
 
     if (server.bgsavechildpid != -1) return REDIS_ERR;
+    if (server.vm_enabled) waitEmptyIOJobsQueue();
     if ((childpid = fork()) == 0) {
         /* Child */
+        if (server.vm_enabled) vmReopenSwapFile();
         close(server.fd);
         if (rdbSave(filename) == REDIS_OK) {
-            exit(0);
+            _exit(0);
         } else {
-            exit(1);
+            _exit(1);
         }
     } else {
         /* Parent */
@@ -3125,6 +3351,10 @@ static robj *rdbLoadObject(int type, FILE *fp) {
 
         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;
@@ -3172,6 +3402,7 @@ static int rdbLoad(char *filename) {
     redisDb *db = server.db+0;
     char buf[1024];
     time_t expiretime = -1, now = time(NULL);
+    long long loadedkeys = 0;
 
     fp = fopen(filename,"r");
     if (!fp) return REDIS_ERR;
@@ -3229,6 +3460,13 @@ static int rdbLoad(char *filename) {
             expiretime = -1;
         }
         keyobj = o = NULL;
+        /* Handle swapping while loading big datasets when VM is on */
+        loadedkeys++;
+        if (server.vm_enabled && (loadedkeys % 5000) == 0) {
+            while (zmalloc_used_memory() > server.vm_max_memory) {
+                if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
+            }
+        }
     }
     fclose(fp);
     return REDIS_OK;
@@ -3459,6 +3697,52 @@ static void decrbyCommand(redisClient *c) {
     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));
+}
+
 /* ========================= Type agnostic commands ========================= */
 
 static void delCommand(redisClient *c) {
@@ -3613,6 +3897,7 @@ static void shutdownCommand(redisClient *c) {
     if (server.appendonly) {
         /* Append only file: fsync() the AOF and exit */
         fsync(server.appendfd);
+        if (server.vm_enabled) unlink(server.vm_swap_file);
         exit(0);
     } else {
         /* Snapshotting. Perform a SYNC SAVE and exit */
@@ -3621,6 +3906,7 @@ static void shutdownCommand(redisClient *c) {
                 unlink(server.pidfile);
             redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
             redisLog(REDIS_WARNING,"Server exit now, bye bye...");
+            if (server.vm_enabled) unlink(server.vm_swap_file);
             exit(0);
         } else {
             /* Ooops.. error saving! The best we can do is to continue operating.
@@ -5222,9 +5508,10 @@ static void sortCommand(redisClient *c) {
     if (sortval->type == REDIS_LIST) {
         list *list = sortval->ptr;
         listNode *ln;
+        listIter li;
 
-        listRewind(list);
-        while((ln = listYield(list))) {
+        listRewind(list,&li);
+        while((ln = listNext(&li))) {
             robj *ele = ln->value;
             vector[j].obj = ele;
             vector[j].u.score = 0;
@@ -5320,13 +5607,15 @@ static void sortCommand(redisClient *c) {
         addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
         for (j = start; j <= end; j++) {
             listNode *ln;
+            listIter li;
+
             if (!getop) {
                 addReplyBulkLen(c,vector[j].obj);
                 addReply(c,vector[j].obj);
                 addReply(c,shared.crlf);
             }
-            listRewind(operations);
-            while((ln = listYield(operations))) {
+            listRewind(operations,&li);
+            while((ln = listNext(&li))) {
                 redisSortOperation *sop = ln->value;
                 robj *val = lookupKeyByPattern(c->db,sop->pattern,
                     vector[j].obj);
@@ -5351,12 +5640,14 @@ static void sortCommand(redisClient *c) {
         /* STORE option specified, set the sorting result as a List object */
         for (j = start; j <= end; j++) {
             listNode *ln;
+            listIter li;
+
             if (!getop) {
                 listAddNodeTail(listPtr,vector[j].obj);
                 incrRefCount(vector[j].obj);
             }
-            listRewind(operations);
-            while((ln = listYield(operations))) {
+            listRewind(operations,&li);
+            while((ln = listNext(&li))) {
                 redisSortOperation *sop = ln->value;
                 robj *val = lookupKeyByPattern(c->db,sop->pattern,
                     vector[j].obj);
@@ -5393,6 +5684,27 @@ static void sortCommand(redisClient *c) {
     zfree(vector);
 }
 
+/* Convert an amount of bytes into a human readable string in the form
+ * of 100B, 2G, 100M, 4K, and so forth. */
+static void bytesToHuman(char *s, unsigned long long n) {
+    double d;
+
+    if (n < 1024) {
+        /* Bytes */
+        sprintf(s,"%lluB",n);
+        return;
+    } else if (n < (1024*1024)) {
+        d = (double)n/(1024);
+        sprintf(s,"%.2fK",d);
+    } else if (n < (1024LL*1024*1024)) {
+        d = (double)n/(1024*1024);
+        sprintf(s,"%.2fM",d);
+    } else if (n < (1024LL*1024*1024*1024)) {
+        d = (double)n/(1024LL*1024*1024);
+        sprintf(s,"%.2fG",d);
+    }
+}
+
 /* Create the string returned by the INFO command. This is decoupled
  * by the INFO command itself as we need to report the same information
  * on memory corruption problems. */
@@ -5400,39 +5712,47 @@ static sds genRedisInfoString(void) {
     sds info;
     time_t uptime = time(NULL)-server.stat_starttime;
     int j;
-    
+    char hmem[64];
+  
+    bytesToHuman(hmem,zmalloc_used_memory());
     info = sdscatprintf(sdsempty(),
         "redis_version:%s\r\n"
         "arch_bits:%s\r\n"
         "multiplexing_api:%s\r\n"
+        "process_id:%ld\r\n"
         "uptime_in_seconds:%ld\r\n"
         "uptime_in_days:%ld\r\n"
         "connected_clients:%d\r\n"
         "connected_slaves:%d\r\n"
         "blocked_clients:%d\r\n"
         "used_memory:%zu\r\n"
+        "used_memory_human:%s\r\n"
         "changes_since_last_save:%lld\r\n"
         "bgsave_in_progress:%d\r\n"
         "last_save_time:%ld\r\n"
         "bgrewriteaof_in_progress:%d\r\n"
         "total_connections_received:%lld\r\n"
         "total_commands_processed:%lld\r\n"
+        "vm_enabled:%d\r\n"
         "role:%s\r\n"
         ,REDIS_VERSION,
         (sizeof(long) == 8) ? "64" : "32",
         aeGetApiName(),
+        (long) getpid(),
         uptime,
         uptime/(3600*24),
         listLength(server.clients)-listLength(server.slaves),
         listLength(server.slaves),
-        server.blockedclients,
-        server.usedmemory,
+        server.blpop_blocked_clients,
+        zmalloc_used_memory(),
+        hmem,
         server.dirty,
         server.bgsavechildpid != -1,
         server.lastsave,
         server.bgrewritechildpid != -1,
         server.stat_numconnections,
         server.stat_numcommands,
+        server.vm_enabled != 0,
         server.masterhost == NULL ? "master" : "slave"
     );
     if (server.masterhost) {
@@ -5448,6 +5768,36 @@ static sds genRedisInfoString(void) {
             server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1
         );
     }
+    if (server.vm_enabled) {
+        lockThreadedIO();
+        info = sdscatprintf(info,
+            "vm_conf_max_memory:%llu\r\n"
+            "vm_conf_page_size:%llu\r\n"
+            "vm_conf_pages:%llu\r\n"
+            "vm_stats_used_pages:%llu\r\n"
+            "vm_stats_swapped_objects:%llu\r\n"
+            "vm_stats_swappin_count:%llu\r\n"
+            "vm_stats_swappout_count:%llu\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_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_stats_used_pages,
+            (unsigned long long) server.vm_stats_swapped_objects,
+            (unsigned long long) server.vm_stats_swapins,
+            (unsigned long long) server.vm_stats_swapouts,
+            (unsigned long) listLength(server.io_newjobs),
+            (unsigned long) listLength(server.io_processing),
+            (unsigned long) listLength(server.io_processed),
+            (unsigned long) server.io_active_threads,
+            (unsigned long) server.vm_blocked_clients
+        );
+        unlockThreadedIO();
+    }
     for (j = 0; j < server.dbnum; j++) {
         long long keys, vkeys;
 
@@ -5717,11 +6067,11 @@ static void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeou
     /* 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 */
-static void unblockClient(redisClient *c) {
+static void unblockClientWaitingData(redisClient *c) {
     dictEntry *de;
     list *l;
     int j;
@@ -5743,18 +6093,19 @@ static void unblockClient(redisClient *c) {
     zfree(c->blockingkeys);
     c->blockingkeys = NULL;
     c->flags &= (~REDIS_BLOCKED);
-    server.blockedclients--;
+    server.blpop_blocked_clients--;
     /* Ok now we are ready to get read events from socket, note that we
-     * can't trap errors here as it's possible that unblockClients() is
+     * 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
-     * in the input buffer. Note that this is safe even if unblockClient()
-     * gets called from freeClient() because freeClient() will be smart
-     * enough to call this function *after* c->querybuf was set to NULL. */
+     * 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
+     * *after* c->querybuf was set to NULL. */
     if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
 }
 
@@ -5788,7 +6139,7 @@ static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
     addReplyBulkLen(receiver,ele);
     addReply(receiver,ele);
     addReply(receiver,shared.crlf);
-    unblockClient(receiver);
+    unblockClientWaitingData(receiver);
     return 1;
 }
 
@@ -5939,9 +6290,10 @@ static void syncCommand(redisClient *c) {
          * registering differences since the server forked to save */
         redisClient *slave;
         listNode *ln;
+        listIter li;
 
-        listRewind(server.slaves);
-        while((ln = listYield(server.slaves))) {
+        listRewind(server.slaves,&li);
+        while((ln = listNext(&li))) {
             slave = ln->value;
             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
         }
@@ -6038,9 +6390,10 @@ static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
 static void updateSlavesWaitingBgsave(int bgsaveerr) {
     listNode *ln;
     int startbgsave = 0;
+    listIter li;
 
-    listRewind(server.slaves);
-    while((ln = listYield(server.slaves))) {
+    listRewind(server.slaves,&li);
+    while((ln = listNext(&li))) {
         redisClient *slave = ln->value;
 
         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
@@ -6072,9 +6425,11 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
     }
     if (startbgsave) {
         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
-            listRewind(server.slaves);
+            listIter li;
+
+            listRewind(server.slaves,&li);
             redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
-            while((ln = listYield(server.slaves))) {
+            while((ln = listNext(&li))) {
                 redisClient *slave = ln->value;
 
                 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
@@ -6212,16 +6567,25 @@ static void slaveofCommand(redisClient *c) {
 
 /* ============================ Maxmemory directive  ======================== */
 
-/* Free one object form the pre-allocated objects free list. This is useful
- * under low mem conditions as by default we take 1 million free objects
- * allocated. */
-static void freeOneObjectFromFreelist(void) {
+/* Try to free one object form the pre-allocated objects free list.
+ * This is useful under low mem conditions as by default we take 1 million
+ * free objects allocated. On success REDIS_OK is returned, otherwise
+ * REDIS_ERR. */
+static int tryFreeOneObjectFromFreelist(void) {
     robj *o;
 
-    listNode *head = listFirst(server.objfreelist);
-    o = listNodeValue(head);
-    listDelNode(server.objfreelist,head);
-    zfree(o);
+    if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
+    if (listLength(server.objfreelist)) {
+        listNode *head = listFirst(server.objfreelist);
+        o = listNodeValue(head);
+        listDelNode(server.objfreelist,head);
+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
+        zfree(o);
+        return REDIS_OK;
+    } else {
+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
+        return REDIS_ERR;
+    }
 }
 
 /* This function gets called when 'maxmemory' is set on the config file to limit
@@ -6237,35 +6601,32 @@ static void freeOneObjectFromFreelist(void) {
  */
 static void freeMemoryIfNeeded(void) {
     while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
-        if (listLength(server.objfreelist)) {
-            freeOneObjectFromFreelist();
-        } else {
-            int j, k, freed = 0;
-
-            for (j = 0; j < server.dbnum; j++) {
-                int minttl = -1;
-                robj *minkey = NULL;
-                struct dictEntry *de;
-
-                if (dictSize(server.db[j].expires)) {
-                    freed = 1;
-                    /* From a sample of three keys drop the one nearest to
-                     * the natural expire */
-                    for (k = 0; k < 3; k++) {
-                        time_t t;
-
-                        de = dictGetRandomKey(server.db[j].expires);
-                        t = (time_t) dictGetEntryVal(de);
-                        if (minttl == -1 || t < minttl) {
-                            minkey = dictGetEntryKey(de);
-                            minttl = t;
-                        }
+        int j, k, freed = 0;
+
+        if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
+        for (j = 0; j < server.dbnum; j++) {
+            int minttl = -1;
+            robj *minkey = NULL;
+            struct dictEntry *de;
+
+            if (dictSize(server.db[j].expires)) {
+                freed = 1;
+                /* From a sample of three keys drop the one nearest to
+                 * the natural expire */
+                for (k = 0; k < 3; k++) {
+                    time_t t;
+
+                    de = dictGetRandomKey(server.db[j].expires);
+                    t = (time_t) dictGetEntryVal(de);
+                    if (minttl == -1 || t < minttl) {
+                        minkey = dictGetEntryKey(de);
+                        minttl = t;
                     }
-                    deleteKey(server.db+j,minkey);
                 }
+                deleteKey(server.db+j,minkey);
             }
-            if (!freed) return; /* nothing to free... */
         }
+        if (!freed) return; /* nothing to free... */
     }
 }
 
@@ -6389,6 +6750,7 @@ int loadAppendOnlyFile(char *filename) {
     struct redisClient *fakeClient;
     FILE *fp = fopen(filename,"r");
     struct redis_stat sb;
+    unsigned long long loadedkeys = 0;
 
     if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
         return REDIS_ERR;
@@ -6450,6 +6812,13 @@ int loadAppendOnlyFile(char *filename) {
         /* Clean up, ready for the next command */
         for (j = 0; j < argc; j++) decrRefCount(argv[j]);
         zfree(argv);
+        /* Handle swapping while loading big datasets when VM is on */
+        loadedkeys++;
+        if (server.vm_enabled && (loadedkeys % 5000) == 0) {
+            while (zmalloc_used_memory() > server.vm_max_memory) {
+                if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
+            }
+        }
     }
     fclose(fp);
     freeFakeClient(fakeClient);
@@ -6470,16 +6839,26 @@ fmterr:
 /* Write an object into a file in the bulk format $<count>\r\n<payload>\r\n */
 static int fwriteBulk(FILE *fp, robj *obj) {
     char buf[128];
-    obj = getDecodedObject(obj);
+    int decrrc = 0;
+
+    /* Avoid the incr/decr ref count business if possible to help
+     * copy-on-write (we are often in a child process when this function
+     * is called).
+     * Also makes sure that key objects don't get incrRefCount-ed when VM
+     * is enabled */
+    if (obj->encoding != REDIS_ENCODING_RAW) {
+        obj = getDecodedObject(obj);
+        decrrc = 1;
+    }
     snprintf(buf,sizeof(buf),"$%ld\r\n",(long)sdslen(obj->ptr));
     if (fwrite(buf,strlen(buf),1,fp) == 0) goto err;
     if (sdslen(obj->ptr) && fwrite(obj->ptr,sdslen(obj->ptr),1,fp) == 0)
         goto err;
     if (fwrite("\r\n",2,1,fp) == 0) goto err;
-    decrRefCount(obj);
+    if (decrrc) decrRefCount(obj);
     return 1;
 err:
-    decrRefCount(obj);
+    if (decrrc) decrRefCount(obj);
     return 0;
 }
 
@@ -6545,12 +6924,16 @@ static int rewriteAppendOnlyFile(char *filename) {
             int swapped;
 
             key = dictGetEntryKey(de);
-            if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY) {
+            /* If the value for this key is swapped, load a preview in memory.
+             * We use a "swapped" flag to remember if we need to free the
+             * value object instead to just increment the ref count anyway
+             * in order to avoid copy-on-write of pages if we are forked() */
+            if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY ||
+                key->storage == REDIS_VM_SWAPPING) {
                 o = dictGetEntryVal(de);
                 swapped = 0;
             } else {
                 o = vmPreviewObject(key);
-                key = dupStringObject(key);
                 swapped = 1;
             }
             expiretime = getExpire(db,key);
@@ -6567,9 +6950,10 @@ static int rewriteAppendOnlyFile(char *filename) {
                 /* Emit the RPUSHes needed to rebuild the list */
                 list *list = o->ptr;
                 listNode *ln;
+                listIter li;
 
-                listRewind(list);
-                while((ln = listYield(list))) {
+                listRewind(list,&li);
+                while((ln = listNext(&li))) {
                     char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
                     robj *eleobj = listNodeValue(ln);
 
@@ -6621,12 +7005,7 @@ static int rewriteAppendOnlyFile(char *filename) {
                 if (fwriteBulk(fp,key) == 0) goto werr;
                 if (fwriteBulkLong(fp,expiretime) == 0) goto werr;
             }
-            /* We created a few temp objects if the key->value pair
-             * was about a swapped out object. Free both. */
-            if (swapped) {
-                decrRefCount(key);
-                decrRefCount(o);
-            }
+            if (swapped) decrRefCount(o);
         }
         dictReleaseIterator(di);
     }
@@ -6670,16 +7049,18 @@ static int rewriteAppendOnlyFileBackground(void) {
     pid_t childpid;
 
     if (server.bgrewritechildpid != -1) return REDIS_ERR;
+    if (server.vm_enabled) waitEmptyIOJobsQueue();
     if ((childpid = fork()) == 0) {
         /* Child */
         char tmpfile[256];
-        close(server.fd);
 
+        if (server.vm_enabled) vmReopenSwapFile();
+        close(server.fd);
         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
-            exit(0);
+            _exit(0);
         } else {
-            exit(1);
+            _exit(1);
         }
     } else {
         /* Parent */
@@ -6722,18 +7103,71 @@ static void aofRemoveTempFile(pid_t childpid) {
     unlink(tmpfile);
 }
 
-/* =============================== Virtual Memory =========================== */
+/* Virtual Memory is composed mainly of two subsystems:
+ * - Blocking Virutal Memory
+ * - Threaded Virtual Memory I/O
+ * The two parts are not fully decoupled, but functions are split among two
+ * different sections of the source code (delimited by comments) in order to
+ * make more clear what functionality is about the blocking VM and what about
+ * the threaded (not blocking) VM.
+ *
+ * Redis VM design:
+ *
+ * Redis VM is a blocking VM (one that blocks reading swapped values from
+ * disk into memory when a value swapped out is needed in memory) that is made
+ * unblocking by trying to examine the command argument vector in order to
+ * load in background values that will likely be needed in order to exec
+ * the command. The command is executed only once all the relevant keys
+ * are loaded into memory.
+ *
+ * This basically is almost as simple of a blocking VM, but almost as parallel
+ * as a fully non-blocking VM.
+ */
+
+/* =================== Virtual Memory - Blocking Side  ====================== */
+
+/* substitute the first occurrence of '%p' with the process pid in the
+ * swap file name. */
+static void expandVmSwapFilename(void) {
+    char *p = strstr(server.vm_swap_file,"%p");
+    sds new;
+    
+    if (!p) return;
+    new = sdsempty();
+    *p = '\0';
+    new = sdscat(new,server.vm_swap_file);
+    new = sdscatprintf(new,"%ld",(long) getpid());
+    new = sdscat(new,p+2);
+    zfree(server.vm_swap_file);
+    server.vm_swap_file = new;
+}
+
 static void vmInit(void) {
     off_t totsize;
+    int pipefds[2];
+    size_t stacksize;
+
+    if (server.vm_max_threads != 0)
+        zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
 
-    server.vm_fp = fopen("/tmp/redisvm","w+b");
+    expandVmSwapFilename();
+    redisLog(REDIS_NOTICE,"Using '%s' as swap file",server.vm_swap_file);
+    if ((server.vm_fp = fopen(server.vm_swap_file,"r+b")) == NULL) {
+        server.vm_fp = fopen(server.vm_swap_file,"w+b");
+    }
     if (server.vm_fp == NULL) {
-        redisLog(REDIS_WARNING,"Impossible to open the swap file. Exiting.");
+        redisLog(REDIS_WARNING,
+            "Impossible to open the swap file: %s. Exiting.",
+            strerror(errno));
         exit(1);
     }
     server.vm_fd = fileno(server.vm_fp);
     server.vm_next_page = 0;
     server.vm_near_pages = 0;
+    server.vm_stats_used_pages = 0;
+    server.vm_stats_swapped_objects = 0;
+    server.vm_stats_swapouts = 0;
+    server.vm_stats_swapins = 0;
     totsize = server.vm_pages*server.vm_page_size;
     redisLog(REDIS_NOTICE,"Allocating %lld bytes of swap file",totsize);
     if (ftruncate(server.vm_fd,totsize) == -1) {
@@ -6747,15 +7181,40 @@ static void vmInit(void) {
     redisLog(REDIS_VERBOSE,"Allocated %lld bytes page table for %lld pages",
         (long long) (server.vm_pages+7)/8, server.vm_pages);
     memset(server.vm_bitmap,0,(server.vm_pages+7)/8);
-    /* Try to remove the swap file, so the OS will really delete it from the
-     * file system when Redis exists. */
-    unlink("/tmp/redisvm");
+
+    /* Initialize threaded I/O (used by Virtual Memory) */
+    server.io_newjobs = listCreate();
+    server.io_processing = listCreate();
+    server.io_processed = 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);
+    server.io_active_threads = 0;
+    if (pipe(pipefds) == -1) {
+        redisLog(REDIS_WARNING,"Unable to intialized VM: pipe(2): %s. Exiting."
+            ,strerror(errno));
+        exit(1);
+    }
+    server.io_ready_pipe_read = pipefds[0];
+    server.io_ready_pipe_write = pipefds[1];
+    redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
+    /* LZF requires a lot of stack */
+    pthread_attr_init(&server.io_threads_attr);
+    pthread_attr_getstacksize(&server.io_threads_attr, &stacksize);
+    while (stacksize < REDIS_THREAD_STACK_SIZE) stacksize *= 2;
+    pthread_attr_setstacksize(&server.io_threads_attr, stacksize);
+    /* Listen for events in the threaded I/O pipe */
+    if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
+        vmThreadedIOCompletedJob, NULL) == AE_ERR)
+        oom("creating file event");
 }
 
 /* Mark the page as used */
 static void vmMarkPageUsed(off_t page) {
     off_t byte = page/8;
     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);
@@ -6767,13 +7226,17 @@ static void vmMarkPagesUsed(off_t page, off_t count) {
 
     for (j = 0; j < count; j++)
         vmMarkPageUsed(page+j);
+    server.vm_stats_used_pages += count;
 }
 
 /* Mark the page as free */
 static void vmMarkPageFree(off_t page) {
     off_t byte = page/8;
     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. */
@@ -6782,6 +7245,10 @@ static void vmMarkPagesFree(off_t page, off_t 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';
+    }
 }
 
 /* Test if the page is free */
@@ -6810,7 +7277,7 @@ static int vmFreePage(off_t page) {
  * note: I implemented this function just after watching an episode of
  * Battlestar Galactica, where the hybrid was continuing to say "JUMP!"
  */
-static int vmFindContiguousPages(off_t *first, int n) {
+static int vmFindContiguousPages(off_t *first, off_t n) {
     off_t base, offset = 0, since_jump = 0, numfree = 0;
 
     if (server.vm_near_pages == REDIS_VM_MAX_NEAR_PAGES) {
@@ -6823,7 +7290,6 @@ static int vmFindContiguousPages(off_t *first, int n) {
     while(offset < server.vm_pages) {
         off_t this = base+offset;
 
-        redisLog(REDIS_DEBUG, "THIS: %lld (%c)\n", (long long) this, vmFreePage(this) ? 'F' : 'X');
         /* If we overflow, restart from page zero */
         if (this >= server.vm_pages) {
             this -= server.vm_pages;
@@ -6833,6 +7299,7 @@ static int vmFindContiguousPages(off_t *first, int n) {
                 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++;
@@ -6864,24 +7331,33 @@ static int vmFindContiguousPages(off_t *first, int n) {
     return REDIS_ERR;
 }
 
+/* Write the specified object at the specified page of the swap file */
+static int vmWriteObjectOnSwap(robj *o, off_t page) {
+    if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex);
+    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",
+            strerror(errno));
+        return REDIS_ERR;
+    }
+    rdbSaveObject(server.vm_fp,o);
+    if (server.vm_enabled) pthread_mutex_unlock(&server.io_swapfile_mutex);
+    return REDIS_OK;
+}
+
 /* Swap the 'val' object relative to 'key' into disk. Store all the information
  * needed to later retrieve the object into the key object.
  * If we can't find enough contiguous empty pages to swap the object on disk
  * REDIS_ERR is returned. */
-static int vmSwapObject(robj *key, robj *val) {
-    off_t pages = rdbSavedObjectPages(val);
+static int vmSwapObjectBlocking(robj *key, robj *val) {
+    off_t pages = rdbSavedObjectPages(val,NULL);
     off_t page;
 
     assert(key->storage == REDIS_VM_MEMORY);
     assert(key->refcount == 1);
     if (vmFindContiguousPages(&page,pages) == REDIS_ERR) return REDIS_ERR;
-    if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {
-        redisLog(REDIS_WARNING,
-            "Critical VM problem in vmSwapObject(): can't seek: %s",
-            strerror(errno));
-        return REDIS_ERR;
-    }
-    rdbSaveObject(server.vm_fp,val);
+    if (vmWriteObjectOnSwap(val,page) == REDIS_ERR) return REDIS_ERR;
     key->vm.page = page;
     key->vm.usedpages = pages;
     key->storage = REDIS_VM_SWAPPED;
@@ -6891,9 +7367,31 @@ static int vmSwapObject(robj *key, robj *val) {
     redisLog(REDIS_DEBUG,"VM: object %s swapped out at %lld (%lld pages)",
         (unsigned char*) key->ptr,
         (unsigned long long) page, (unsigned long long) pages);
+    server.vm_stats_swapped_objects++;
+    server.vm_stats_swapouts++;
+    fflush(server.vm_fp);
     return REDIS_OK;
 }
 
+static robj *vmReadObjectFromSwap(off_t page, int type) {
+    robj *o;
+
+    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 vmReadObjectFromSwap(): can't seek: %s",
+            strerror(errno));
+        _exit(1);
+    }
+    o = rdbLoadObject(type,server.vm_fp);
+    if (o == NULL) {
+        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;
+}
+
 /* Load the value object relative to the 'key' object from swap to memory.
  * The newly allocated object is returned.
  *
@@ -6902,30 +7400,29 @@ static int vmSwapObject(robj *key, robj *val) {
 static robj *vmGenericLoadObject(robj *key, int preview) {
     robj *val;
 
-    redisAssert(key->storage == REDIS_VM_SWAPPED);
-    if (fseeko(server.vm_fp,key->vm.page*server.vm_page_size,SEEK_SET) == -1) {
-        redisLog(REDIS_WARNING,
-            "Unrecoverable VM problem in vmLoadObject(): can't seek: %s",
-            strerror(errno));
-        exit(1);
-    }
-    val = rdbLoadObject(key->vtype,server.vm_fp);
-    if (val == NULL) {
-        redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmLoadObject(): can't load object from swap file: %s", strerror(errno));
-        exit(1);
-    }
+    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;
         key->vm.atime = server.unixtime;
         vmMarkPagesFree(key->vm.page,key->vm.usedpages);
         redisLog(REDIS_DEBUG, "VM: object %s loaded from disk",
             (unsigned char*) key->ptr);
+        server.vm_stats_swapped_objects--;
+    } else {
+        redisLog(REDIS_DEBUG, "VM: object %s previewed from disk",
+            (unsigned char*) key->ptr);
     }
+    server.vm_stats_swapins++;
     return val;
 }
 
 /* Plain object loading, from swap to memory */
 static robj *vmLoadObject(robj *key) {
+    /* If we are loading the object in background, stop it, we
+     * need to load this object synchronously ASAP. */
+    if (key->storage == REDIS_VM_LOADING)
+        vmCancelThreadedIOJob(key);
     return vmGenericLoadObject(key,0);
 }
 
@@ -7006,16 +7503,23 @@ static double computeObjectSwappability(robj *o) {
 
 /* Try to swap an object that's a good candidate for swapping.
  * Returns REDIS_OK if the object was swapped, REDIS_ERR if it's not possible
- * to swap any object at all. */
-static int vmSwapOneObject(void) {
+ * to swap any object at all.
+ *
+ * If 'usethreaded' is true, Redis will try to swap the object in background
+ * using I/O threads. */
+static int vmSwapOneObject(int usethreads) {
     int j, i;
     struct dictEntry *best = NULL;
     double best_swappability = 0;
+    redisDb *best_db = NULL;
     robj *key, *val;
 
     for (j = 0; j < server.dbnum; j++) {
         redisDb *db = server.db+j;
-        int maxtries = 1000;
+        /* Why maxtries is set to 100?
+         * Because this way (usually) we'll find 1 object even if just 1% - 2%
+         * are swappable objects */
+        int maxtries = 100;
 
         if (dictSize(db->dict) == 0) continue;
         for (i = 0; i < 5; i++) {
@@ -7026,7 +7530,14 @@ static int vmSwapOneObject(void) {
             de = dictGetRandomKey(db->dict);
             key = dictGetEntryKey(de);
             val = dictGetEntryVal(de);
-            if (key->storage != REDIS_VM_MEMORY) {
+            /* Only swap objects that are currently in memory.
+             *
+             * Also don't swap shared objects if threaded VM is on, as we
+             * try to ensure that the main thread does not touch the
+             * object while the I/O thread is using it, but we can't
+             * control other keys without adding additional mutex. */
+            if (key->storage != REDIS_VM_MEMORY ||
+                (server.vm_max_threads != 0 && val->refcount != 1)) {
                 if (maxtries) i--; /* don't count this try */
                 continue;
             }
@@ -7034,6 +7545,7 @@ static int vmSwapOneObject(void) {
             if (!best || swappability > best_swappability) {
                 best = de;
                 best_swappability = swappability;
+                best_db = db;
             }
         }
     }
@@ -7054,14 +7566,27 @@ static int vmSwapOneObject(void) {
         key = dictGetEntryKey(best) = newkey;
     }
     /* Swap it */
-    if (vmSwapObject(key,val) == REDIS_OK) {
-        dictGetEntryVal(best) = NULL;
+    if (usethreads) {
+        vmSwapObjectThreaded(key,val,best_db);
         return REDIS_OK;
     } else {
-        return REDIS_ERR;
+        if (vmSwapObjectBlocking(key,val) == REDIS_OK) {
+            dictGetEntryVal(best) = NULL;
+            return REDIS_OK;
+        } else {
+            return REDIS_ERR;
+        }
     }
 }
 
+static int vmSwapOneObjectBlocking() {
+    return vmSwapOneObject(0);
+}
+
+static int vmSwapOneObjectThreaded() {
+    return vmSwapOneObject(1);
+}
+
 /* Return true if it's safe to swap out objects in a given moment.
  * Basically we don't want to swap objects out while there is a BGSAVE
  * or a BGAEOREWRITE running in backgroud. */
@@ -7082,6 +7607,535 @@ static int deleteIfSwapped(redisDb *db, robj *key) {
     return 1;
 }
 
+/* =================== Virtual Memory - Threaded I/O  ======================= */
+
+static void freeIOJob(iojob *j) {
+    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);
+}
+
+/* Every time a thread finished a Job, it writes a byte into the write side
+ * of an unix pipe in order to "awake" the main thread, and this function
+ * is called. */
+static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
+            int mask)
+{
+    char buf[1];
+    int retval, processed = 0, toprocess = -1, trytoswap = 1;
+    REDIS_NOTUSED(el);
+    REDIS_NOTUSED(mask);
+    REDIS_NOTUSED(privdata);
+
+    /* For every byte we read in the read side of the pipe, there is one
+     * I/O job completed to process. */
+    while((retval = read(fd,buf,1)) == 1) {
+        iojob *j;
+        listNode *ln;
+        robj *key;
+        struct dictEntry *de;
+
+        redisLog(REDIS_DEBUG,"Processing I/O completed job");
+
+        /* Get the processed element (the oldest one) */
+        lockThreadedIO();
+        assert(listLength(server.io_processed) != 0);
+        if (toprocess == -1) {
+            toprocess = (listLength(server.io_processed)*REDIS_MAX_COMPLETED_JOBS_PROCESSED)/100;
+            if (toprocess <= 0) toprocess = 1;
+        }
+        ln = listFirst(server.io_processed);
+        j = ln->value;
+        listDelNode(server.io_processed,ln);
+        unlockThreadedIO();
+        /* If this job is marked as canceled, just ignore it */
+        if (j->canceled) {
+            freeIOJob(j);
+            continue;
+        }
+        /* Post process it in the main thread, as there are things we
+         * can do just here to avoid race conditions and/or invasive locks */
+        redisLog(REDIS_DEBUG,"Job %p type: %d, key at %p (%s) refcount: %d\n", (void*) j, j->type, (void*)j->key, (char*)j->key->ptr, j->key->refcount);
+        de = dictFind(j->db->dict,j->key);
+        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;
+            vmMarkPagesFree(key->vm.page,key->vm.usedpages);
+            redisLog(REDIS_DEBUG, "VM: object %s loaded from disk (threaded)",
+                (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);
+            /* 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
+             * rebranded as REDIS_IOJOB_DO_SWAP. */
+            if (!vmCanSwapOut() ||
+                vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR)
+            {
+                /* Ooops... no space or we can't swap as there is
+                 * a fork()ed Redis trying to save stuff on disk. */
+                freeIOJob(j);
+                key->storage = REDIS_VM_MEMORY; /* undo operation */
+            } else {
+                /* Note that we need to mark this pages as used now,
+                 * if the job will be canceled, we'll mark them as freed
+                 * again. */
+                vmMarkPagesUsed(j->page,j->pages);
+                j->type = REDIS_IOJOB_DO_SWAP;
+                lockThreadedIO();
+                queueIOJob(j);
+                unlockThreadedIO();
+            }
+        } else if (j->type == REDIS_IOJOB_DO_SWAP) {
+            robj *val;
+
+            /* Key swapped. We can finally free some memory. */
+            if (key->storage != REDIS_VM_SWAPPING) {
+                printf("key->storage: %d\n",key->storage);
+                printf("key->name: %s\n",(char*)key->ptr);
+                printf("key->refcount: %d\n",key->refcount);
+                printf("val: %p\n",(void*)j->val);
+                printf("val->type: %d\n",j->val->type);
+                printf("val->ptr: %s\n",(char*)j->val->ptr);
+            }
+            redisAssert(key->storage == REDIS_VM_SWAPPING);
+            val = dictGetEntryVal(de);
+            key->vm.page = j->page;
+            key->vm.usedpages = j->pages;
+            key->storage = REDIS_VM_SWAPPED;
+            key->vtype = j->val->type;
+            decrRefCount(val); /* Deallocate the object from memory. */
+            dictGetEntryVal(de) = NULL;
+            redisLog(REDIS_DEBUG,
+                "VM: object %s swapped out at %lld (%lld pages) (threaded)",
+                (unsigned char*) key->ptr,
+                (unsigned long long) j->page, (unsigned long long) j->pages);
+            server.vm_stats_swapped_objects++;
+            server.vm_stats_swapouts++;
+            freeIOJob(j);
+            /* Put a few more swap requests in queue if we are still
+             * out of memory */
+            if (trytoswap && vmCanSwapOut() &&
+                zmalloc_used_memory() > server.vm_max_memory)
+            {
+                int more = 1;
+                while(more) {
+                    lockThreadedIO();
+                    more = listLength(server.io_newjobs) <
+                            (unsigned) server.vm_max_threads;
+                    unlockThreadedIO();
+                    /* Don't waste CPU time if swappable objects are rare. */
+                    if (vmSwapOneObjectThreaded() == REDIS_ERR) {
+                        trytoswap = 0;
+                        break;
+                    }
+                }
+            }
+        }
+        processed++;
+        if (processed == toprocess) return;
+    }
+    if (retval < 0 && errno != EAGAIN) {
+        redisLog(REDIS_WARNING,
+            "WARNING: read(2) error in vmThreadedIOCompletedJob() %s",
+            strerror(errno));
+    }
+}
+
+static void lockThreadedIO(void) {
+    pthread_mutex_lock(&server.io_mutex);
+}
+
+static void unlockThreadedIO(void) {
+    pthread_mutex_unlock(&server.io_mutex);
+}
+
+/* Remove the specified object from the threaded I/O queue if still not
+ * processed, otherwise make sure to flag it as canceled. */
+static void vmCancelThreadedIOJob(robj *o) {
+    list *lists[3] = {
+        server.io_newjobs,      /* 0 */
+        server.io_processing,   /* 1 */
+        server.io_processed     /* 2 */
+    };
+    int i;
+
+    assert(o->storage == REDIS_VM_LOADING || o->storage == REDIS_VM_SWAPPING);
+again:
+    lockThreadedIO();
+    /* Search for a matching key in one of the queues */
+    for (i = 0; i < 3; i++) {
+        listNode *ln;
+        listIter li;
+
+        listRewind(lists[i],&li);
+        while ((ln = listNext(&li)) != NULL) {
+            iojob *job = ln->value;
+
+            if (job->canceled) continue; /* Skip this, already canceled. */
+            if (compareStringObjects(job->key,o) == 0) {
+                redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (type %d) (LIST ID %d)\n",
+                    (void*)job, (char*)o->ptr, job->type, i);
+                /* Mark the pages as free since the swap didn't happened
+                 * or happened but is now discarded. */
+                if (i != 1 && job->type == REDIS_IOJOB_DO_SWAP)
+                    vmMarkPagesFree(job->page,job->pages);
+                /* Cancel the job. It depends on the list the job is
+                 * living in. */
+                switch(i) {
+                case 0: /* io_newjobs */
+                    /* If the job was yet not processed the best thing to do
+                     * is to remove it from the queue at all */
+                    freeIOJob(job);
+                    listDelNode(lists[i],ln);
+                    break;
+                case 1: /* io_processing */
+                    /* 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
+                     * when processing completed jobs. */
+                    job->canceled = 1;
+                    break;
+                }
+                /* Finally we have to adjust the storage type of the object
+                 * in order to "UNDO" the operaiton. */
+                if (o->storage == REDIS_VM_LOADING)
+                    o->storage = REDIS_VM_SWAPPED;
+                else if (o->storage == REDIS_VM_SWAPPING)
+                    o->storage = REDIS_VM_MEMORY;
+                unlockThreadedIO();
+                return;
+            }
+        }
+    }
+    unlockThreadedIO();
+    assert(1 != 1); /* We should never reach this */
+}
+
+static void *IOThreadEntryPoint(void *arg) {
+    iojob *j;
+    listNode *ln;
+    REDIS_NOTUSED(arg);
+
+    pthread_detach(pthread_self());
+    while(1) {
+        /* Get a new job to process */
+        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());
+            server.io_active_threads--;
+            unlockThreadedIO();
+            return NULL;
+        }
+        ln = listFirst(server.io_newjobs);
+        j = ln->value;
+        listDelNode(server.io_newjobs,ln);
+        /* Add the job in the processing queue */
+        j->thread = pthread_self();
+        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);
+
+        /* 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);
+            fclose(fp);
+        } else if (j->type == REDIS_IOJOB_DO_SWAP) {
+            if (vmWriteObjectOnSwap(j->val,j->page) == REDIS_ERR)
+                j->canceled = 1;
+        }
+
+        /* 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);
+        lockThreadedIO();
+        listDelNode(server.io_processing,ln);
+        listAddNodeTail(server.io_processed,j);
+        unlockThreadedIO();
+        
+        /* Signal the main thread there is new stuff to process */
+        assert(write(server.io_ready_pipe_write,"x",1) == 1);
+    }
+    return NULL; /* never reached */
+}
+
+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_sigmask(SIG_SETMASK, &omask, NULL);
+    server.io_active_threads++;
+}
+
+/* We need to wait for the last thread to exit before we are able to
+ * fork() in order to BGSAVE or BGREWRITEAOF. */
+static void waitEmptyIOJobsQueue(void) {
+    while(1) {
+        int io_processed_len;
+
+        lockThreadedIO();
+        if (listLength(server.io_newjobs) == 0 &&
+            listLength(server.io_processing) == 0 &&
+            server.io_active_threads == 0)
+        {
+            unlockThreadedIO();
+            return;
+        }
+        /* While waiting for empty jobs queue condition we post-process some
+         * finshed job, as I/O threads may be hanging trying to write against
+         * the io_ready_pipe_write FD but there are so much pending jobs that
+         * it's blocking. */
+        io_processed_len = listLength(server.io_processed);
+        unlockThreadedIO();
+        if (io_processed_len) {
+            vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,NULL,0);
+            usleep(1000); /* 1 millisecond */
+        } else {
+            usleep(10000); /* 10 milliseconds */
+        }
+    }
+}
+
+static void vmReopenSwapFile(void) {
+    /* 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);
+        _exit(1);
+    }
+    server.vm_fd = fileno(server.vm_fp);
+}
+
+/* This function must be called while with threaded IO locked */
+static void queueIOJob(iojob *j) {
+    redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",
+        (void*)j, j->type, (char*)j->key->ptr);
+    listAddNodeTail(server.io_newjobs,j);
+    if (server.io_active_threads < server.vm_max_threads)
+        spawnIOThread();
+}
+
+static int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
+    iojob *j;
+    
+    assert(key->storage == REDIS_VM_MEMORY);
+    assert(key->refcount == 1);
+
+    j = zmalloc(sizeof(*j));
+    j->type = REDIS_IOJOB_PREPARE_SWAP;
+    j->db = db;
+    j->key = dupStringObject(key);
+    j->val = val;
+    incrRefCount(val);
+    j->canceled = 0;
+    j->thread = (pthread_t) -1;
+    key->storage = REDIS_VM_SWAPPING;
+
+    lockThreadedIO();
+    queueIOJob(j);
+    unlockThreadedIO();
+    return REDIS_OK;
+}
+
+/* ============ 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?
+ * 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.
+ *
+ * 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) {
+    if (cmd->proc == getCommand) {
+        waitForSwappedKey(c,c->argv[1]);
+    }
+    /* 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);
+        }
+    }
+}
+
 /* ================================= Debugging ============================== */
 
 static void debugCommand(redisClient *c) {
@@ -7117,10 +8171,20 @@ static void debugCommand(redisClient *c) {
         }
         key = dictGetEntryKey(de);
         val = dictGetEntryVal(de);
-        addReplySds(c,sdscatprintf(sdsempty(),
-            "+Key at:%p refcount:%d, value at:%p refcount:%d encoding:%d serializedlength:%lld\r\n",
+        if (!server.vm_enabled || (key->storage == REDIS_VM_MEMORY ||
+                                   key->storage == REDIS_VM_SWAPPING)) {
+            addReplySds(c,sdscatprintf(sdsempty(),
+                "+Key at:%p refcount:%d, value at:%p refcount:%d "
+                "encoding:%d serializedlength:%lld\r\n",
                 (void*)key, key->refcount, (void*)val, val->refcount,
-                val->encoding, rdbSavedObjectLen(val)));
+                val->encoding, (long long) rdbSavedObjectLen(val,NULL)));
+        } else {
+            addReplySds(c,sdscatprintf(sdsempty(),
+                "+Key at:%p refcount:%d, value swapped at: page %llu "
+                "using %llu pages\r\n",
+                (void*)key, key->refcount, (unsigned long long) key->vm.page,
+                (unsigned long long) key->vm.usedpages));
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) {
         dictEntry *de = dictFind(c->db->dict,c->argv[2]);
         robj *key, *val;
@@ -7144,7 +8208,7 @@ static void debugCommand(redisClient *c) {
         /* Swap it */
         if (key->storage != REDIS_VM_MEMORY) {
             addReplySds(c,sdsnew("-ERR This key is not in memory\r\n"));
-        } else if (vmSwapObject(key,val) == REDIS_OK) {
+        } else if (vmSwapObjectBlocking(key,val) == REDIS_OK) {
             dictGetEntryVal(de) = NULL;
             addReply(c,shared.ok);
         } else {
@@ -7156,9 +8220,9 @@ static void debugCommand(redisClient *c) {
     }
 }
 
-static void _redisAssert(char *estr) {
+static void _redisAssert(char *estr, char *file, int line) {
     redisLog(REDIS_WARNING,"=== ASSERTION FAILED ===");
-    redisLog(REDIS_WARNING,"==> %s\n",estr);
+    redisLog(REDIS_WARNING,"==> %s:%d '%s' is not true\n",file,line,estr);
 #ifdef HAVE_BACKTRACE
     redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)");
     *((char*)-1) = 'x';
@@ -7194,7 +8258,6 @@ static void daemonize(void) {
     FILE *fp;
 
     if (fork() != 0) exit(0); /* parent exits */
-    printf("New pid: %d\n", getpid());
     setsid(); /* create a new session */
 
     /* Every output goes to /dev/null. If Redis is daemonized but
@@ -7215,6 +8278,8 @@ static void daemonize(void) {
 }
 
 int main(int argc, char **argv) {
+    time_t start;
+
     initServerConfig();
     if (argc == 2) {
         resetServerSaveParams();
@@ -7231,16 +8296,16 @@ int main(int argc, char **argv) {
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
 #endif
+    start = time(NULL);
     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)
-            redisLog(REDIS_NOTICE,"DB loaded from disk");
+            redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start);
     }
-    if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
-        acceptHandler, NULL) == AE_ERR) oom("creating file event");
     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;
@@ -7311,7 +8376,7 @@ static void segvHandler(int sig, siginfo_t *info, void *secret) {
         }
     }
     /* free(messages); Don't call free() with possibly corrupted memory. */
-    exit(0);
+    _exit(0);
 }
 
 static void setupSigSegvAction(void) {