]> git.saurik.com Git - redis.git/blobdiff - redis.c
VM now swaps objects out while loading datasets not fitting into vm-max-memory bytes...
[redis.git] / redis.c
diff --git a/redis.c b/redis.c
index a97427cc8f61d05b753ccffc0479e9f7988c41de..8cf8d1746c14b0f2474ac2025f658cd398f2c654 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
-#define REDIS_VERSION "1.3.0"
+#define REDIS_VERSION "1.3.2"
 
 #include "fmacros.h"
 #include "config.h"
 #define REDIS_RDB_ENC_INT32 2       /* 32 bit signed integer */
 #define REDIS_RDB_ENC_LZF 3         /* string compressed with FASTLZ */
 
+/* Virtual memory object->where field. */
+#define REDIS_VM_MEMORY 0       /* The object is on memory */
+#define REDIS_VM_SWAPPED 1      /* The object is on disk */
+#define REDIS_VM_SWAPPING 2     /* Redis is swapping this object on disk */
+#define REDIS_VM_LOADING 3      /* Redis is loading this object from disk */
+
+/* Virtual memory static configuration stuff.
+ * Check vmFindContiguousPages() to know more about this magic numbers. */
+#define REDIS_VM_MAX_NEAR_PAGES 65536
+#define REDIS_VM_MAX_RANDOM_JUMP 4096
+
 /* 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 */
 
 /* Slave replication state - slave side */
 #define REDIS_REPL_NONE 0   /* No active replication */
 
 /* Log levels */
 #define REDIS_DEBUG 0
-#define REDIS_NOTICE 1
-#define REDIS_WARNING 2
+#define REDIS_VERBOSE 1
+#define REDIS_NOTICE 2
+#define REDIS_WARNING 3
 
 /* Anti-warning macro... */
 #define REDIS_NOTUSED(V) ((void) V)
@@ -206,12 +219,29 @@ static void _redisAssert(char *estr);
 /*================================= Data types ============================== */
 
 /* A redis object, that is a type able to hold a string / list / set */
+
+/* The VM object structure */
+struct redisObjectVM {
+    off_t page;         /* the page at witch the object is stored on disk */
+    off_t usedpages;    /* number of pages used on disk */
+    time_t atime;       /* Last access time */
+} vm;
+
+/* The actual Redis Object */
 typedef struct redisObject {
     void *ptr;
     unsigned char type;
     unsigned char encoding;
-    unsigned char notused[2];
+    unsigned char storage;  /* If this object is a key, where is the value?
+                             * REDIS_VM_MEMORY, REDIS_VM_SWAPPED, ... */
+    unsigned char vtype; /* If this object is a key, and value is swapped out,
+                          * this is the type of the swapped out object. */
     int refcount;
+    /* VM fields, this are only allocated if VM is active, otherwise the
+     * object allocation function will just allocate
+     * sizeof(redisObjct) minus sizeof(redisObjectVM), so using
+     * Redis without VM active will not have any overhead. */
+    struct redisObjectVM vm;
 } robj;
 
 /* Macro used to initalize a Redis object allocated on the stack.
@@ -223,11 +253,13 @@ typedef struct redisObject {
     _var.type = REDIS_STRING; \
     _var.encoding = REDIS_ENCODING_RAW; \
     _var.ptr = _ptr; \
+    if (server.vm_enabled) _var.storage = REDIS_VM_MEMORY; \
 } while(0);
 
 typedef struct redisDb {
-    dict *dict;
-    dict *expires;
+    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) */
     int id;
 } redisDb;
 
@@ -266,6 +298,11 @@ 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
+                             * 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. */
 } redisClient;
 
 struct saveparam {
@@ -278,7 +315,7 @@ struct redisServer {
     int port;
     int fd;
     redisDb *db;
-    dict *sharingpool;
+    dict *sharingpool;          /* Poll used for object sharing */
     unsigned int sharingpoolsize;
     long long dirty;            /* changes to DB from the last save */
     list *clients;
@@ -325,12 +362,30 @@ struct redisServer {
     redisClient *master;    /* client that is master for this slave */
     int replstate;
     unsigned int maxclients;
-    unsigned long maxmemory;
+    unsigned long long maxmemory;
+    unsigned int blockedclients;
     /* 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;
     int sort_alpha;
     int sort_bypattern;
+    /* Virtual memory configuration */
+    int vm_enabled;
+    off_t vm_page_size;
+    off_t vm_pages;
+    unsigned long long vm_max_memory;
+    /* Virtual memory state */
+    FILE *vm_fp;
+    int vm_fd;
+    off_t vm_next_page; /* Next probably empty page */
+    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 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;
 };
 
 typedef void redisCommandProc(redisClient *c);
@@ -410,6 +465,7 @@ static void addReplySds(redisClient *c, sds s);
 static void incrRefCount(robj *o);
 static int rdbSaveBackground(char *filename);
 static robj *createStringObject(char *ptr, size_t len);
+static robj *dupStringObject(robj *o);
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);
 static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
 static int syncWithMaster(void);
@@ -419,6 +475,7 @@ static robj *getDecodedObject(robj *o);
 static int removeExpire(redisDb *db, robj *key);
 static int expireIfNeeded(redisDb *db, robj *key);
 static int deleteIfVolatile(redisDb *db, robj *key);
+static int deleteIfSwapped(redisDb *db, robj *key);
 static int deleteKey(redisDb *db, robj *key);
 static time_t getExpire(redisDb *db, robj *key);
 static int setExpire(redisDb *db, robj *key, time_t when);
@@ -437,6 +494,15 @@ 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 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 vmCanSwapOut(void);
+static void freeOneObjectFromFreelist(void);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -513,6 +579,8 @@ static void zscoreCommand(redisClient *c);
 static void zremrangebyscoreCommand(redisClient *c);
 static void multiCommand(redisClient *c);
 static void execCommand(redisClient *c);
+static void blpopCommand(redisClient *c);
+static void brpopCommand(redisClient *c);
 
 /*================================= Globals ================================= */
 
@@ -531,6 +599,8 @@ static struct redisCommand cmdTable[] = {
     {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"rpop",rpopCommand,2,REDIS_CMD_INLINE},
     {"lpop",lpopCommand,2,REDIS_CMD_INLINE},
+    {"brpop",brpopCommand,-3,REDIS_CMD_INLINE},
+    {"blpop",blpopCommand,-3,REDIS_CMD_INLINE},
     {"llen",llenCommand,2,REDIS_CMD_INLINE},
     {"lindex",lindexCommand,3,REDIS_CMD_INLINE},
     {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
@@ -760,6 +830,12 @@ static void dictVanillaFree(void *privdata, void *val)
     zfree(val);
 }
 
+static void dictListDestructor(void *privdata, void *val)
+{
+    DICT_NOTUSED(privdata);
+    listRelease((list*)val);
+}
+
 static int sdsDictKeyCompare(void *privdata, const void *key1,
         const void *key2)
 {
@@ -776,6 +852,7 @@ static void dictRedisObjectDestructor(void *privdata, void *val)
 {
     DICT_NOTUSED(privdata);
 
+    if (val == NULL) return; /* Values of swapped out keys as set to NULL */
     decrRefCount(val);
 }
 
@@ -841,6 +918,17 @@ static dictType hashDictType = {
     dictRedisObjectDestructor   /* val destructor */
 };
 
+/* Keylist hash table type has unencoded redis objects as keys and
+ * lists as values. It's used for blocking operations (BLPOP) */
+static dictType keylistDictType = {
+    dictObjHash,                /* hash function */
+    NULL,                       /* key dup */
+    NULL,                       /* val dup */
+    dictObjKeyCompare,          /* key compare */
+    dictRedisObjectDestructor,  /* key destructor */
+    dictListDestructor          /* val destructor */
+};
+
 /* ========================= Random utility functions ======================= */
 
 /* Redis generally does not try to recover from out of memory conditions
@@ -863,11 +951,18 @@ static void closeTimedoutClients(void) {
     listRewind(server.clients);
     while ((ln = listYield(server.clients)) != NULL) {
         c = listNodeValue(ln);
-        if (!(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
+        if (server.maxidletime &&
+            !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
             !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
-             (now - c->lastinteraction > server.maxidletime)) {
-            redisLog(REDIS_DEBUG,"Closing idle client");
+             (now - c->lastinteraction > server.maxidletime))
+        {
+            redisLog(REDIS_VERBOSE,"Closing idle client");
             freeClient(c);
+        } else if (c->flags & REDIS_BLOCKED) {
+            if (c->blockingto != 0 && c->blockingto < now) {
+                addReply(c,shared.nullmultibulk);
+                unblockClient(c);
+            }
         }
     }
 }
@@ -888,9 +983,9 @@ static void tryResizeHashTables(void) {
 
     for (j = 0; j < server.dbnum; j++) {
         if (htNeedsResize(server.db[j].dict)) {
-            redisLog(REDIS_DEBUG,"The hash table %d is too sparse, resize it...",j);
+            redisLog(REDIS_VERBOSE,"The hash table %d is too sparse, resize it...",j);
             dictResize(server.db[j].dict);
-            redisLog(REDIS_DEBUG,"Hash table %d resized.",j);
+            redisLog(REDIS_VERBOSE,"Hash table %d resized.",j);
         }
         if (htNeedsResize(server.db[j].expires))
             dictResize(server.db[j].expires);
@@ -987,6 +1082,12 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
     REDIS_NOTUSED(id);
     REDIS_NOTUSED(clientData);
 
+    /* We take a cached value of the unix time in the global state because
+     * with virtual memory and aging there is to store the current time
+     * in objects at every object access, and accuracy is not needed.
+     * 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();
 
@@ -998,7 +1099,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         used = dictSize(server.db[j].dict);
         vkeys = dictSize(server.db[j].expires);
         if (!(loops % 5) && (used || vkeys)) {
-            redisLog(REDIS_DEBUG,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
+            redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
             /* dictPrintStats(server.dict); */
         }
     }
@@ -1013,7 +1114,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
 
     /* Show information about connected clients */
     if (!(loops % 5)) {
-        redisLog(REDIS_DEBUG,"%d clients connected (%d slaves), %zu bytes in use, %d shared objects",
+        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,
@@ -1021,7 +1122,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
     }
 
     /* Close connections of timedout clients */
-    if (server.maxidletime && !(loops % 10))
+    if ((server.maxidletime && !(loops % 10)) || server.blockedclients)
         closeTimedoutClients();
 
     /* Check if a background saving or AOF rewrite in progress terminated */
@@ -1064,7 +1165,7 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         /* Continue to expire if at the end of the cycle more than 25%
          * of the keys were expired. */
         do {
-            int num = dictSize(db->expires);
+            long num = dictSize(db->expires);
             time_t now = time(NULL);
 
             expired = 0;
@@ -1084,6 +1185,24 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         } while (expired > REDIS_EXPIRELOOKUPS_PER_CRON/4);
     }
 
+    /* Swap a few keys on disk if we are over the memory limit and VM
+     * is enbled. Try to free objects from the free list first. */
+    if (vmCanSwapOut()) {
+        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;
+            }
+        }
+    }
+
     /* Check if we should connect to a MASTER */
     if (server.replstate == REDIS_REPL_CONNECT) {
         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
@@ -1147,7 +1266,7 @@ static void resetServerSaveParams() {
 static void initServerConfig() {
     server.dbnum = REDIS_DEFAULT_DBNUM;
     server.port = REDIS_SERVERPORT;
-    server.verbosity = REDIS_DEBUG;
+    server.verbosity = REDIS_VERBOSE;
     server.maxidletime = REDIS_MAXIDLETIME;
     server.saveparams = NULL;
     server.logfile = NULL; /* NULL = log on standard output */
@@ -1167,7 +1286,13 @@ static void initServerConfig() {
     server.rdbcompression = 1;
     server.sharingpoolsize = 1024;
     server.maxclients = 0;
+    server.blockedclients = 0;
     server.maxmemory = 0;
+    server.vm_enabled = 0;
+    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 */
+
     resetServerSaveParams();
 
     appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
@@ -1211,6 +1336,7 @@ 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].blockingkeys = dictCreate(&keylistDictType,NULL);
         server.db[j].id = j;
     }
     server.cronloops = 0;
@@ -1223,6 +1349,7 @@ static void initServer() {
     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 (server.appendonly) {
@@ -1233,6 +1360,8 @@ static void initServer() {
             exit(1);
         }
     }
+
+    if (server.vm_enabled) vmInit();
 }
 
 /* Empty the whole database */
@@ -1317,6 +1446,7 @@ static void loadServerConfig(char *filename) {
             }
         } else if (!strcasecmp(argv[0],"loglevel") && argc == 2) {
             if (!strcasecmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
+            else if (!strcasecmp(argv[1],"verbose")) server.verbosity = REDIS_VERBOSE;
             else if (!strcasecmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
             else if (!strcasecmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
             else {
@@ -1399,6 +1529,16 @@ static void loadServerConfig(char *filename) {
           server.pidfile = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
           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-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 {
             err = "Bad directive or wrong number of arguments"; goto loaderr;
         }
@@ -1432,9 +1572,18 @@ static void freeClientArgv(redisClient *c) {
 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. */
+    sdsfree(c->querybuf);
+    c->querybuf = NULL;
+    if (c->flags & REDIS_BLOCKED)
+        unblockClient(c);
+
     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
-    sdsfree(c->querybuf);
     listRelease(c->reply);
     freeClientArgv(c);
     close(c->fd);
@@ -1539,7 +1688,7 @@ static void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask)
         if (errno == EAGAIN) {
             nwritten = 0;
         } else {
-            redisLog(REDIS_DEBUG,
+            redisLog(REDIS_VERBOSE,
                 "Error writing to client: %s", strerror(errno));
             freeClient(c);
             return;
@@ -1592,7 +1741,7 @@ static void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int
         /* write all collected blocks at once */
         if((nwritten = writev(fd, iov, ion)) < 0) {
             if (errno != EAGAIN) {
-                redisLog(REDIS_DEBUG,
+                redisLog(REDIS_VERBOSE,
                          "Error writing to client: %s", strerror(errno));
                 freeClient(c);
                 return;
@@ -1898,6 +2047,13 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
 
 static void processInputBuffer(redisClient *c) {
 again:
+    /* Before to process the input buffer, make sure the client is not
+     * waitig for a blocking operation such as BLPOP. Note that the first
+     * iteration the client is never blocked, otherwise the processInputBuffer
+     * 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->bulklen == -1) {
         /* Read the first line of the query */
         char *p = strchr(c->querybuf,'\n');
@@ -1946,7 +2102,7 @@ again:
             }
             return;
         } else if (sdslen(c->querybuf) >= REDIS_REQUEST_MAX_SIZE) {
-            redisLog(REDIS_DEBUG, "Client protocol error");
+            redisLog(REDIS_VERBOSE, "Client protocol error");
             freeClient(c);
             return;
         }
@@ -1983,12 +2139,12 @@ static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mas
         if (errno == EAGAIN) {
             nread = 0;
         } else {
-            redisLog(REDIS_DEBUG, "Reading from client: %s",strerror(errno));
+            redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
             freeClient(c);
             return;
         }
     } else if (nread == 0) {
-        redisLog(REDIS_DEBUG, "Client closed connection");
+        redisLog(REDIS_VERBOSE, "Client closed connection");
         freeClient(c);
         return;
     }
@@ -2034,6 +2190,8 @@ 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);
     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
@@ -2052,6 +2210,11 @@ static void addReply(redisClient *c, robj *obj) {
          c->replstate == REDIS_REPL_ONLINE) &&
         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
         sendReplyToClient, c) == AE_ERR) return;
+
+    if (server.vm_enabled && obj->storage != REDIS_VM_MEMORY) {
+        obj = dupStringObject(obj);
+        obj->refcount = 0; /* getDecodedObject() will increment the refcount */
+    }
     listAddNodeTail(c->reply,getDecodedObject(obj));
 }
 
@@ -2100,10 +2263,10 @@ static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 
     cfd = anetAccept(server.neterr, fd, cip, &cport);
     if (cfd == AE_ERR) {
-        redisLog(REDIS_DEBUG,"Accepting client connection: %s", server.neterr);
+        redisLog(REDIS_VERBOSE,"Accepting client connection: %s", server.neterr);
         return;
     }
-    redisLog(REDIS_DEBUG,"Accepted %s:%d", cip, cport);
+    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
     if ((c = createClient(cfd)) == NULL) {
         redisLog(REDIS_WARNING,"Error allocating resoures for the client");
         close(cfd); /* May be already closed, just ingore errors */
@@ -2136,12 +2299,20 @@ static robj *createObject(int type, void *ptr) {
         o = listNodeValue(head);
         listDelNode(server.objfreelist,head);
     } else {
-        o = zmalloc(sizeof(*o));
+        if (server.vm_enabled) {
+            o = zmalloc(sizeof(*o));
+        } else {
+            o = zmalloc(sizeof(*o)-sizeof(struct redisObjectVM));
+        }
     }
     o->type = type;
     o->encoding = REDIS_ENCODING_RAW;
     o->ptr = ptr;
     o->refcount = 1;
+    if (server.vm_enabled) {
+        o->vm.atime = server.unixtime;
+        o->storage = REDIS_VM_MEMORY;
+    }
     return o;
 }
 
@@ -2149,6 +2320,10 @@ static robj *createStringObject(char *ptr, size_t len) {
     return createObject(REDIS_STRING,sdsnewlen(ptr,len));
 }
 
+static robj *dupStringObject(robj *o) {
+    return createStringObject(o->ptr,sdslen(o->ptr));
+}
+
 static robj *createListObject(void) {
     list *l = listCreate();
 
@@ -2196,20 +2371,26 @@ static void freeHashObject(robj *o) {
 }
 
 static void incrRefCount(robj *o) {
+    redisAssert(!server.vm_enabled || o->storage == REDIS_VM_MEMORY);
     o->refcount++;
-#ifdef DEBUG_REFCOUNT
-    if (o->type == REDIS_STRING)
-        printf("Increment '%s'(%p), now is: %d\n",o->ptr,o,o->refcount);
-#endif
 }
 
 static void decrRefCount(void *obj) {
     robj *o = obj;
 
-#ifdef DEBUG_REFCOUNT
-    if (o->type == REDIS_STRING)
-        printf("Decrement '%s'(%p), now is: %d\n",o->ptr,o,o->refcount-1);
-#endif
+    /* REDIS_VM_SWAPPED */
+    if (server.vm_enabled && o->storage == REDIS_VM_SWAPPED) {
+        redisAssert(o->refcount == 1);
+        redisAssert(o->type == REDIS_STRING);
+        freeStringObject(o);
+        vmMarkPagesFree(o->vm.page,o->vm.usedpages);
+        if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
+            !listAddNodeHead(server.objfreelist,o))
+            zfree(o);
+        server.vm_stats_swapped_objects--;
+        return;
+    }
+    /* REDIS_VM_MEMORY */
     if (--(o->refcount) == 0) {
         switch(o->type) {
         case REDIS_STRING: freeStringObject(o); break;
@@ -2227,7 +2408,25 @@ static void decrRefCount(void *obj) {
 
 static robj *lookupKey(redisDb *db, robj *key) {
     dictEntry *de = dictFind(db->dict,key);
-    return de ? dictGetEntryVal(de) : NULL;
+    if (de) {
+        robj *key = dictGetEntryKey(de);
+        robj *val = dictGetEntryVal(de);
+
+        if (server.vm_enabled) {
+            if (key->storage == REDIS_VM_MEMORY) {
+                /* Update the access time of the key for the aging algorithm. */
+                key->vm.atime = server.unixtime;
+            } else {
+                /* Our value was swapped on disk. Bring it at home. */
+                redisAssert(val == NULL);
+                val = vmLoadObject(key);
+                dictGetEntryVal(de) = val;
+            }
+        }
+        return val;
+    } else {
+        return NULL;
+    }
 }
 
 static robj *lookupKeyRead(redisDb *db, robj *key) {
@@ -2410,7 +2609,7 @@ static size_t stringObjectLen(robj *o) {
     }
 }
 
-/*============================ DB saving/loading ============================ */
+/*============================ RDB saving/loading =========================== */
 
 static int rdbSaveType(FILE *fp, unsigned char type) {
     if (fwrite(&type,1,1,fp) == 0) return -1;
@@ -2583,6 +2782,79 @@ static int rdbSaveDoubleValue(FILE *fp, double val) {
     return 0;
 }
 
+/* Save a Redis object. */
+static int rdbSaveObject(FILE *fp, robj *o) {
+    if (o->type == REDIS_STRING) {
+        /* Save a string value */
+        if (rdbSaveStringObject(fp,o) == -1) return -1;
+    } else if (o->type == REDIS_LIST) {
+        /* Save a list value */
+        list *list = o->ptr;
+        listNode *ln;
+
+        listRewind(list);
+        if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
+        while((ln = listYield(list))) {
+            robj *eleobj = listNodeValue(ln);
+
+            if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
+        }
+    } else if (o->type == REDIS_SET) {
+        /* Save a set value */
+        dict *set = o->ptr;
+        dictIterator *di = dictGetIterator(set);
+        dictEntry *de;
+
+        if (rdbSaveLen(fp,dictSize(set)) == -1) return -1;
+        while((de = dictNext(di)) != NULL) {
+            robj *eleobj = dictGetEntryKey(de);
+
+            if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
+        }
+        dictReleaseIterator(di);
+    } else if (o->type == REDIS_ZSET) {
+        /* Save a set value */
+        zset *zs = o->ptr;
+        dictIterator *di = dictGetIterator(zs->dict);
+        dictEntry *de;
+
+        if (rdbSaveLen(fp,dictSize(zs->dict)) == -1) return -1;
+        while((de = dictNext(di)) != NULL) {
+            robj *eleobj = dictGetEntryKey(de);
+            double *score = dictGetEntryVal(de);
+
+            if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
+            if (rdbSaveDoubleValue(fp,*score) == -1) return -1;
+        }
+        dictReleaseIterator(di);
+    } else {
+        redisAssert(0 != 0);
+    }
+    return 0;
+}
+
+/* Return the length the object will have on disk if saved with
+ * 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);
+
+    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);
+    
+    return (bytes+(server.vm_page_size-1))/server.vm_page_size;
+}
+
 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
 static int rdbSave(char *filename) {
     dictIterator *di = NULL;
@@ -2626,54 +2898,27 @@ static int rdbSave(char *filename) {
                 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
                 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
             }
-            /* Save the key and associated value */
-            if (rdbSaveType(fp,o->type) == -1) goto werr;
-            if (rdbSaveStringObject(fp,key) == -1) goto werr;
-            if (o->type == REDIS_STRING) {
-                /* Save a string value */
-                if (rdbSaveStringObject(fp,o) == -1) goto werr;
-            } else if (o->type == REDIS_LIST) {
-                /* Save a list value */
-                list *list = o->ptr;
-                listNode *ln;
-
-                listRewind(list);
-                if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
-                while((ln = listYield(list))) {
-                    robj *eleobj = listNodeValue(ln);
-
-                    if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
-                }
-            } else if (o->type == REDIS_SET) {
-                /* Save a set value */
-                dict *set = o->ptr;
-                dictIterator *di = dictGetIterator(set);
-                dictEntry *de;
-
-                if (rdbSaveLen(fp,dictSize(set)) == -1) goto werr;
-                while((de = dictNext(di)) != NULL) {
-                    robj *eleobj = dictGetEntryKey(de);
-
-                    if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
-                }
-                dictReleaseIterator(di);
-            } else if (o->type == REDIS_ZSET) {
-                /* Save a set value */
-                zset *zs = o->ptr;
-                dictIterator *di = dictGetIterator(zs->dict);
-                dictEntry *de;
-
-                if (rdbSaveLen(fp,dictSize(zs->dict)) == -1) goto werr;
-                while((de = dictNext(di)) != NULL) {
-                    robj *eleobj = dictGetEntryKey(de);
-                    double *score = dictGetEntryVal(de);
-
-                    if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
-                    if (rdbSaveDoubleValue(fp,*score) == -1) goto werr;
-                }
-                dictReleaseIterator(di);
+            /* 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) {
+                /* 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 {
-                redisAssert(0 != 0);
+                robj *po, *newkey;
+                /* 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 (rdbSaveObject(fp,po) == -1) goto werr;
+                /* Remove the loaded object from memory */
+                decrRefCount(po);
+                decrRefCount(newkey);
             }
         }
         dictReleaseIterator(di);
@@ -2756,35 +3001,29 @@ static time_t rdbLoadTime(FILE *fp) {
  *
  * isencoded is set to 1 if the readed length is not actually a length but
  * an "encoding type", check the above comments for more info */
-static uint32_t rdbLoadLen(FILE *fp, int rdbver, int *isencoded) {
+static uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
     unsigned char buf[2];
     uint32_t len;
+    int type;
 
     if (isencoded) *isencoded = 0;
-    if (rdbver == 0) {
+    if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
+    type = (buf[0]&0xC0)>>6;
+    if (type == REDIS_RDB_6BITLEN) {
+        /* Read a 6 bit len */
+        return buf[0]&0x3F;
+    } else if (type == REDIS_RDB_ENCVAL) {
+        /* Read a 6 bit len encoding type */
+        if (isencoded) *isencoded = 1;
+        return buf[0]&0x3F;
+    } else if (type == REDIS_RDB_14BITLEN) {
+        /* Read a 14 bit len */
+        if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
+        return ((buf[0]&0x3F)<<8)|buf[1];
+    } else {
+        /* Read a 32 bit len */
         if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
         return ntohl(len);
-    } else {
-        int type;
-
-        if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
-        type = (buf[0]&0xC0)>>6;
-        if (type == REDIS_RDB_6BITLEN) {
-            /* Read a 6 bit len */
-            return buf[0]&0x3F;
-        } else if (type == REDIS_RDB_ENCVAL) {
-            /* Read a 6 bit len encoding type */
-            if (isencoded) *isencoded = 1;
-            return buf[0]&0x3F;
-        } else if (type == REDIS_RDB_14BITLEN) {
-            /* Read a 14 bit len */
-            if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
-            return ((buf[0]&0x3F)<<8)|buf[1];
-        } else {
-            /* Read a 32 bit len */
-            if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
-            return ntohl(len);
-        }
     }
 }
 
@@ -2812,13 +3051,13 @@ static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
     return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
 }
 
-static robj *rdbLoadLzfStringObject(FILE*fp, int rdbver) {
+static robj *rdbLoadLzfStringObject(FILE*fp) {
     unsigned int len, clen;
     unsigned char *c = NULL;
     sds val = NULL;
 
-    if ((clen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
-    if ((len = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR) return NULL;
+    if ((clen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+    if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
     if ((c = zmalloc(clen)) == NULL) goto err;
     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
     if (fread(c,clen,1,fp) == 0) goto err;
@@ -2831,12 +3070,12 @@ err:
     return NULL;
 }
 
-static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
+static robj *rdbLoadStringObject(FILE*fp) {
     int isencoded;
     uint32_t len;
     sds val;
 
-    len = rdbLoadLen(fp,rdbver,&isencoded);
+    len = rdbLoadLen(fp,&isencoded);
     if (isencoded) {
         switch(len) {
         case REDIS_RDB_ENC_INT8:
@@ -2844,7 +3083,7 @@ static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
         case REDIS_RDB_ENC_INT32:
             return tryObjectSharing(rdbLoadIntegerObject(fp,len));
         case REDIS_RDB_ENC_LZF:
-            return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
+            return tryObjectSharing(rdbLoadLzfStringObject(fp));
         default:
             redisAssert(0!=0);
         }
@@ -2877,6 +3116,59 @@ static int rdbLoadDoubleValue(FILE *fp, double *val) {
     }
 }
 
+/* Load a Redis object of the specified type from the specified file.
+ * On success a newly allocated object is returned, otherwise NULL. */
+static robj *rdbLoadObject(int type, FILE *fp) {
+    robj *o;
+
+    if (type == REDIS_STRING) {
+        /* Read string value */
+        if ((o = rdbLoadStringObject(fp)) == NULL) return NULL;
+        tryObjectEncoding(o);
+    } else if (type == REDIS_LIST || type == REDIS_SET) {
+        /* Read list/set value */
+        uint32_t listlen;
+
+        if ((listlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        o = (type == REDIS_LIST) ? createListObject() : createSetObject();
+        /* Load every single element of the list/set */
+        while(listlen--) {
+            robj *ele;
+
+            if ((ele = rdbLoadStringObject(fp)) == NULL) return NULL;
+            tryObjectEncoding(ele);
+            if (type == REDIS_LIST) {
+                listAddNodeTail((list*)o->ptr,ele);
+            } else {
+                dictAdd((dict*)o->ptr,ele,NULL);
+            }
+        }
+    } else if (type == REDIS_ZSET) {
+        /* Read list/set value */
+        uint32_t zsetlen;
+        zset *zs;
+
+        if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        o = createZsetObject();
+        zs = o->ptr;
+        /* Load every single element of the list/set */
+        while(zsetlen--) {
+            robj *ele;
+            double *score = zmalloc(sizeof(double));
+
+            if ((ele = rdbLoadStringObject(fp)) == NULL) return NULL;
+            tryObjectEncoding(ele);
+            if (rdbLoadDoubleValue(fp,score) == -1) return NULL;
+            dictAdd(zs->dict,ele,score);
+            zslInsert(zs->zsl,*score,ele);
+            incrRefCount(ele); /* added to skiplist */
+        }
+    } else {
+        redisAssert(0 != 0);
+    }
+    return o;
+}
+
 static int rdbLoad(char *filename) {
     FILE *fp;
     robj *keyobj = NULL;
@@ -2886,6 +3178,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;
@@ -2897,7 +3190,7 @@ static int rdbLoad(char *filename) {
         return REDIS_ERR;
     }
     rdbver = atoi(buf+5);
-    if (rdbver > 1) {
+    if (rdbver != 1) {
         fclose(fp);
         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
         return REDIS_ERR;
@@ -2915,7 +3208,7 @@ static int rdbLoad(char *filename) {
         if (type == REDIS_EOF) break;
         /* Handle SELECT DB opcode as a special case */
         if (type == REDIS_SELECTDB) {
-            if ((dbid = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
+            if ((dbid = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR)
                 goto eoferr;
             if (dbid >= (unsigned)server.dbnum) {
                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
@@ -2926,55 +3219,9 @@ static int rdbLoad(char *filename) {
             continue;
         }
         /* Read key */
-        if ((keyobj = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
-
-        if (type == REDIS_STRING) {
-            /* Read string value */
-            if ((o = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
-            tryObjectEncoding(o);
-        } else if (type == REDIS_LIST || type == REDIS_SET) {
-            /* Read list/set value */
-            uint32_t listlen;
-
-            if ((listlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
-                goto eoferr;
-            o = (type == REDIS_LIST) ? createListObject() : createSetObject();
-            /* Load every single element of the list/set */
-            while(listlen--) {
-                robj *ele;
-
-                if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
-                tryObjectEncoding(ele);
-                if (type == REDIS_LIST) {
-                    listAddNodeTail((list*)o->ptr,ele);
-                } else {
-                    dictAdd((dict*)o->ptr,ele,NULL);
-                }
-            }
-        } else if (type == REDIS_ZSET) {
-            /* Read list/set value */
-            uint32_t zsetlen;
-            zset *zs;
-
-            if ((zsetlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
-                goto eoferr;
-            o = createZsetObject();
-            zs = o->ptr;
-            /* Load every single element of the list/set */
-            while(zsetlen--) {
-                robj *ele;
-                double *score = zmalloc(sizeof(double));
-
-                if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
-                tryObjectEncoding(ele);
-                if (rdbLoadDoubleValue(fp,score) == -1) goto eoferr;
-                dictAdd(zs->dict,ele,score);
-                zslInsert(zs->zsl,*score,ele);
-                incrRefCount(ele); /* added to skiplist */
-            }
-        } else {
-            redisAssert(0 != 0);
-        }
+        if ((keyobj = rdbLoadStringObject(fp)) == NULL) goto eoferr;
+        /* Read value */
+        if ((o = rdbLoadObject(type,fp)) == NULL) goto eoferr;
         /* Add the new object in the hash table */
         retval = dictAdd(d,keyobj,o);
         if (retval == DICT_ERR) {
@@ -2989,6 +3236,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 (vmSwapOneObject() == REDIS_ERR) break;
+            }
+        }
     }
     fclose(fp);
     return REDIS_OK;
@@ -3031,6 +3285,12 @@ static void setGenericCommand(redisClient *c, int nx) {
     retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
     if (retval == DICT_ERR) {
         if (!nx) {
+            /* If the key is about a swapped value, we want a new key object
+             * to overwrite the old. So we delete the old key in the database.
+             * This will also make sure that swap pages about the old object
+             * will be marked as free. */
+            if (deleteIfSwapped(c->db,c->argv[1]))
+                incrRefCount(c->argv[1]);
             dictReplace(c->db->dict,c->argv[1],c->argv[2]);
             incrRefCount(c->argv[2]);
         } else {
@@ -3477,6 +3737,10 @@ static void pushGenericCommand(redisClient *c, int where) {
 
     lobj = lookupKeyWrite(c->db,c->argv[1]);
     if (lobj == NULL) {
+        if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
+            addReply(c,shared.ok);
+            return;
+        }
         lobj = createListObject();
         list = lobj->ptr;
         if (where == REDIS_HEAD) {
@@ -3492,6 +3756,10 @@ static void pushGenericCommand(redisClient *c, int where) {
             addReply(c,shared.wrongtypeerr);
             return;
         }
+        if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
+            addReply(c,shared.ok);
+            return;
+        }
         list = lobj->ptr;
         if (where == REDIS_HEAD) {
             listAddNodeHead(list,c->argv[2]);
@@ -3796,20 +4064,24 @@ static void rpoplpushcommand(redisClient *c) {
                 robj *ele = listNodeValue(ln);
                 list *dstlist;
 
-                if (dobj == NULL) {
-
-                    /* Create the list if the key does not exist */
-                    dobj = createListObject();
-                    dictAdd(c->db->dict,c->argv[2],dobj);
-                    incrRefCount(c->argv[2]);
-                } else if (dobj->type != REDIS_LIST) {
+                if (dobj && dobj->type != REDIS_LIST) {
                     addReply(c,shared.wrongtypeerr);
                     return;
                 }
-                /* Add the element to the target list */
-                dstlist = dobj->ptr;
-                listAddNodeHead(dstlist,ele);
-                incrRefCount(ele);
+
+                /* Add the element to the target list (unless it's directly
+                 * passed to some BLPOP-ing client */
+                if (!handleClientsWaitingListPush(c,c->argv[2],ele)) {
+                    if (dobj == NULL) {
+                        /* Create the list if the key does not exist */
+                        dobj = createListObject();
+                        dictAdd(c->db->dict,c->argv[2],dobj);
+                        incrRefCount(c->argv[2]);
+                    }
+                    dstlist = dobj->ptr;
+                    listAddNodeHead(dstlist,ele);
+                    incrRefCount(ele);
+                }
 
                 /* Send the element to the client as reply as well */
                 addReplyBulkLen(c,ele);
@@ -5147,10 +5419,12 @@ static sds genRedisInfoString(void) {
         "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"
         "changes_since_last_save:%lld\r\n"
         "bgsave_in_progress:%d\r\n"
@@ -5158,14 +5432,17 @@ static sds genRedisInfoString(void) {
         "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.dirty,
         server.bgsavechildpid != -1,
@@ -5173,6 +5450,7 @@ static sds genRedisInfoString(void) {
         server.bgrewritechildpid != -1,
         server.stat_numconnections,
         server.stat_numcommands,
+        server.vm_enabled != 0,
         server.masterhost == NULL ? "master" : "slave"
     );
     if (server.masterhost) {
@@ -5188,6 +5466,24 @@ static sds genRedisInfoString(void) {
             server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1
         );
     }
+    if (server.vm_enabled) {
+        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"
+            ,(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
+        );
+    }
     for (j = 0; j < server.dbnum; j++) {
         long long keys, vkeys;
 
@@ -5393,6 +5689,206 @@ static void execCommand(redisClient *c) {
     c->flags &= (~REDIS_MULTI);
 }
 
+/* =========================== Blocking Operations  ========================= */
+
+/* Currently Redis blocking operations support is limited to list POP ops,
+ * so the current implementation is not fully generic, but it is also not
+ * completely specific so it will not require a rewrite to support new
+ * kind of blocking operations in the future.
+ *
+ * Still it's important to note that list blocking operations can be already
+ * used as a notification mechanism in order to implement other blocking
+ * operations at application level, so there must be a very strong evidence
+ * of usefulness and generality before new blocking operations are implemented.
+ *
+ * This is how the current blocking POP works, we use BLPOP as example:
+ * - If the user calls BLPOP and the key exists and contains a non empty list
+ *   then LPOP is called instead. So BLPOP is semantically the same as LPOP
+ *   if there is not to block.
+ * - If instead BLPOP is called and the key does not exists or the list is
+ *   empty we need to block. In order to do so we remove the notification for
+ *   new data to read in the client socket (so that we'll not serve new
+ *   requests if the blocking request is not served). Also we put the client
+ *   in a dictionary (db->blockingkeys) mapping keys to a list of clients
+ *   blocking for this keys.
+ * - If a PUSH operation against a key with blocked clients waiting is
+ *   performed, we serve the first in the list: basically instead to push
+ *   the new element inside the list we return it to the (first / oldest)
+ *   blocking client, unblock the client, and remove it form the list.
+ *
+ * The above comment and the source code should be enough in order to understand
+ * the implementation and modify / fix it later.
+ */
+
+/* Set a client in blocking mode for the specified key, with the specified
+ * timeout */
+static void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout) {
+    dictEntry *de;
+    list *l;
+    int j;
+
+    c->blockingkeys = zmalloc(sizeof(robj*)*numkeys);
+    c->blockingkeysnum = numkeys;
+    c->blockingto = timeout;
+    for (j = 0; j < numkeys; j++) {
+        /* Add the key in the client structure, to map clients -> keys */
+        c->blockingkeys[j] = keys[j];
+        incrRefCount(keys[j]);
+
+        /* And in the other "side", to map keys -> clients */
+        de = dictFind(c->db->blockingkeys,keys[j]);
+        if (de == NULL) {
+            int retval;
+
+            /* For every key we take a list of clients blocked for it */
+            l = listCreate();
+            retval = dictAdd(c->db->blockingkeys,keys[j],l);
+            incrRefCount(keys[j]);
+            assert(retval == DICT_OK);
+        } else {
+            l = dictGetEntryVal(de);
+        }
+        listAddNodeTail(l,c);
+    }
+    /* Mark the client as a blocked client */
+    c->flags |= REDIS_BLOCKED;
+    aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
+    server.blockedclients++;
+}
+
+/* Unblock a client that's waiting in a blocking operation such as BLPOP */
+static void unblockClient(redisClient *c) {
+    dictEntry *de;
+    list *l;
+    int j;
+
+    assert(c->blockingkeys != NULL);
+    /* The client may wait for multiple keys, so unblock it for every key. */
+    for (j = 0; j < c->blockingkeysnum; j++) {
+        /* Remove this client from the list of clients waiting for this key. */
+        de = dictFind(c->db->blockingkeys,c->blockingkeys[j]);
+        assert(de != NULL);
+        l = dictGetEntryVal(de);
+        listDelNode(l,listSearchKey(l,c));
+        /* If the list is empty we need to remove it to avoid wasting memory */
+        if (listLength(l) == 0)
+            dictDelete(c->db->blockingkeys,c->blockingkeys[j]);
+        decrRefCount(c->blockingkeys[j]);
+    }
+    /* Cleanup the client structure */
+    zfree(c->blockingkeys);
+    c->blockingkeys = NULL;
+    c->flags &= (~REDIS_BLOCKED);
+    server.blockedclients--;
+    /* Ok now we are ready to get read events from socket, note that we
+     * can't trap errors here as it's possible that unblockClients() 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. */
+    if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
+}
+
+/* This should be called from any function PUSHing into lists.
+ * 'c' is the "pushing client", 'key' is the key it is pushing data against,
+ * 'ele' is the element pushed.
+ *
+ * If the function returns 0 there was no client waiting for a list push
+ * against this key.
+ *
+ * If the function returns 1 there was a client waiting for a list push
+ * against this key, the element was passed to this client thus it's not
+ * needed to actually add it to the list and the caller should return asap. */
+static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
+    struct dictEntry *de;
+    redisClient *receiver;
+    list *l;
+    listNode *ln;
+
+    de = dictFind(c->db->blockingkeys,key);
+    if (de == NULL) return 0;
+    l = dictGetEntryVal(de);
+    ln = listFirst(l);
+    assert(ln != NULL);
+    receiver = ln->value;
+
+    addReplySds(receiver,sdsnew("*2\r\n"));
+    addReplyBulkLen(receiver,key);
+    addReply(receiver,key);
+    addReply(receiver,shared.crlf);
+    addReplyBulkLen(receiver,ele);
+    addReply(receiver,ele);
+    addReply(receiver,shared.crlf);
+    unblockClient(receiver);
+    return 1;
+}
+
+/* Blocking RPOP/LPOP */
+static void blockingPopGenericCommand(redisClient *c, int where) {
+    robj *o;
+    time_t timeout;
+    int j;
+
+    for (j = 1; j < c->argc-1; j++) {
+        o = lookupKeyWrite(c->db,c->argv[j]);
+        if (o != NULL) {
+            if (o->type != REDIS_LIST) {
+                addReply(c,shared.wrongtypeerr);
+                return;
+            } else {
+                list *list = o->ptr;
+                if (listLength(list) != 0) {
+                    /* If the list contains elements fall back to the usual
+                     * non-blocking POP operation */
+                    robj *argv[2], **orig_argv;
+                    int orig_argc;
+                   
+                    /* We need to alter the command arguments before to call
+                     * popGenericCommand() as the command takes a single key. */
+                    orig_argv = c->argv;
+                    orig_argc = c->argc;
+                    argv[1] = c->argv[j];
+                    c->argv = argv;
+                    c->argc = 2;
+
+                    /* Also the return value is different, we need to output
+                     * the multi bulk reply header and the key name. The
+                     * "real" command will add the last element (the value)
+                     * for us. If this souds like an hack to you it's just
+                     * because it is... */
+                    addReplySds(c,sdsnew("*2\r\n"));
+                    addReplyBulkLen(c,argv[1]);
+                    addReply(c,argv[1]);
+                    addReply(c,shared.crlf);
+                    popGenericCommand(c,where);
+
+                    /* Fix the client structure with the original stuff */
+                    c->argv = orig_argv;
+                    c->argc = orig_argc;
+                    return;
+                }
+            }
+        }
+    }
+    /* If the list is empty or the key does not exists we must block */
+    timeout = strtol(c->argv[c->argc-1]->ptr,NULL,10);
+    if (timeout > 0) timeout += time(NULL);
+    blockForKeys(c,c->argv+1,c->argc-2,timeout);
+}
+
+static void blpopCommand(redisClient *c) {
+    blockingPopGenericCommand(c,REDIS_HEAD);
+}
+
+static void brpopCommand(redisClient *c) {
+    blockingPopGenericCommand(c,REDIS_TAIL);
+}
+
 /* =============================== Replication  ============================= */
 
 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
@@ -5548,7 +6044,7 @@ static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     if ((nwritten = write(fd,buf,buflen)) == -1) {
-        redisLog(REDIS_DEBUG,"Write error sending DB to slave: %s",
+        redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
             strerror(errno));
         freeClient(slave);
         return;
@@ -5752,6 +6248,18 @@ 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) {
+    robj *o;
+
+    listNode *head = listFirst(server.objfreelist);
+    o = listNodeValue(head);
+    listDelNode(server.objfreelist,head);
+    zfree(o);
+}
+
 /* This function gets called when 'maxmemory' is set on the config file to limit
  * the max memory used by the server, and we are out of memory.
  * This function will try to, in order:
@@ -5766,12 +6274,7 @@ static void slaveofCommand(redisClient *c) {
 static void freeMemoryIfNeeded(void) {
     while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
         if (listLength(server.objfreelist)) {
-            robj *o;
-
-            listNode *head = listFirst(server.objfreelist);
-            o = listNodeValue(head);
-            listDelNode(server.objfreelist,head);
-            zfree(o);
+            freeOneObjectFromFreelist();
         } else {
             int j, k, freed = 0;
 
@@ -5922,6 +6425,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;
@@ -5983,6 +6487,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 (vmSwapOneObject() == REDIS_ERR) break;
+            }
+        }
     }
     fclose(fp);
     freeFakeClient(fakeClient);
@@ -6073,9 +6584,20 @@ static int rewriteAppendOnlyFile(char *filename) {
 
         /* Iterate this DB writing every entry */
         while((de = dictNext(di)) != NULL) {
-            robj *key = dictGetEntryKey(de);
-            robj *o = dictGetEntryVal(de);
-            time_t expiretime = getExpire(db,key);
+            robj *key, *o;
+            time_t expiretime;
+            int swapped;
+
+            key = dictGetEntryKey(de);
+            if (!server.vm_enabled || key->storage == REDIS_VM_MEMORY) {
+                o = dictGetEntryVal(de);
+                swapped = 0;
+            } else {
+                o = vmPreviewObject(key);
+                key = dupStringObject(key);
+                swapped = 1;
+            }
+            expiretime = getExpire(db,key);
 
             /* Save the key and associated value */
             if (o->type == REDIS_STRING) {
@@ -6143,6 +6665,12 @@ 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);
+            }
         }
         dictReleaseIterator(di);
     }
@@ -6238,6 +6766,379 @@ static void aofRemoveTempFile(pid_t childpid) {
     unlink(tmpfile);
 }
 
+/* =============================== Virtual Memory =========================== */
+static void vmInit(void) {
+    off_t totsize;
+
+    server.vm_fp = fopen("/tmp/redisvm","w+b");
+    if (server.vm_fp == NULL) {
+        redisLog(REDIS_WARNING,"Impossible to open the swap file. Exiting.");
+        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) {
+        redisLog(REDIS_WARNING,"Can't ftruncate swap file: %s. Exiting.",
+            strerror(errno));
+        exit(1);
+    } else {
+        redisLog(REDIS_NOTICE,"Swap file allocated with success");
+    }
+    server.vm_bitmap = zmalloc((server.vm_pages+7)/8);
+    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");
+}
+
+/* Mark the page as used */
+static void vmMarkPageUsed(off_t page) {
+    off_t byte = page/8;
+    int bit = page&7;
+    server.vm_bitmap[byte] |= 1<<bit;
+    redisLog(REDIS_DEBUG,"Mark used: %lld (byte:%lld bit:%d)\n",
+        (long long)page, (long long)byte, bit);
+}
+
+/* Mark N contiguous pages as used, with 'page' being the first. */
+static void vmMarkPagesUsed(off_t page, off_t count) {
+    off_t j;
+
+    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;
+    server.vm_bitmap[byte] &= ~(1<<bit);
+}
+
+/* Mark N contiguous pages as free, with 'page' being the first. */
+static void vmMarkPagesFree(off_t page, off_t count) {
+    off_t j;
+
+    for (j = 0; j < count; j++)
+        vmMarkPageFree(page+j);
+    server.vm_stats_used_pages -= count;
+}
+
+/* Test if the page is free */
+static int vmFreePage(off_t page) {
+    off_t byte = page/8;
+    int bit = page&7;
+    return (server.vm_bitmap[byte] & (1<<bit)) == 0;
+}
+
+/* Find N contiguous free pages storing the first page of the cluster in *first.
+ * Returns REDIS_OK if it was able to find N contiguous pages, otherwise 
+ * REDIS_ERR is returned.
+ *
+ * This function uses a simple algorithm: we try to allocate
+ * REDIS_VM_MAX_NEAR_PAGES sequentially, when we reach this limit we start
+ * again from the start of the swap file searching for free spaces.
+ *
+ * If it looks pretty clear that there are no free pages near our offset
+ * we try to find less populated places doing a forward jump of
+ * REDIS_VM_MAX_RANDOM_JUMP, then we start scanning again a few pages
+ * without hurry, and then we jump again and so forth...
+ * 
+ * This function can be improved using a free list to avoid to guess
+ * too much, since we could collect data about freed pages.
+ *
+ * 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) {
+    off_t base, offset = 0, since_jump = 0, numfree = 0;
+
+    if (server.vm_near_pages == REDIS_VM_MAX_NEAR_PAGES) {
+        server.vm_near_pages = 0;
+        server.vm_next_page = 0;
+    }
+    server.vm_near_pages++; /* Yet another try for pages near to the old ones */
+    base = server.vm_next_page;
+
+    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;
+            if (this == 0) {
+                /* Just overflowed, what we found on tail is no longer
+                 * interesting, as it's no longer contiguous. */
+                numfree = 0;
+            }
+        }
+        if (vmFreePage(this)) {
+            /* This is a free page */
+            numfree++;
+            /* Already got N free pages? Return to the caller, with success */
+            if (numfree == n) {
+                *first = this-(n-1);
+                server.vm_next_page = this+1;
+                return REDIS_OK;
+            }
+        } else {
+            /* The current one is not a free page */
+            numfree = 0;
+        }
+
+        /* Fast-forward if the current page is not free and we already
+         * searched enough near this place. */
+        since_jump++;
+        if (!numfree && since_jump >= REDIS_VM_MAX_RANDOM_JUMP/4) {
+            offset += random() % REDIS_VM_MAX_RANDOM_JUMP;
+            since_jump = 0;
+            /* Note that even if we rewind after the jump, we are don't need
+             * to make sure numfree is set to zero as we only jump *if* it
+             * is set to zero. */
+        } else {
+            /* Otherwise just check the next page */
+            offset++;
+        }
+    }
+    return REDIS_ERR;
+}
+
+/* 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);
+    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);
+    key->vm.page = page;
+    key->vm.usedpages = pages;
+    key->storage = REDIS_VM_SWAPPED;
+    key->vtype = val->type;
+    decrRefCount(val); /* Deallocate the object from memory. */
+    vmMarkPagesUsed(page,pages);
+    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++;
+    return REDIS_OK;
+}
+
+/* Load the value object relative to the 'key' object from swap to memory.
+ * The newly allocated object is returned.
+ *
+ * If preview is true the unserialized object is returned to the caller but
+ * no changes are made to the key object, nor the pages are marked as freed */
+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);
+    }
+    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) {
+    return vmGenericLoadObject(key,0);
+}
+
+/* Just load the value on disk, without to modify the key.
+ * This is useful when we want to perform some operation on the value
+ * without to really bring it from swap to memory, like while saving the
+ * dataset or rewriting the append only log. */
+static robj *vmPreviewObject(robj *key) {
+    return vmGenericLoadObject(key,1);
+}
+
+/* How a good candidate is this object for swapping?
+ * The better candidate it is, the greater the returned value.
+ *
+ * Currently we try to perform a fast estimation of the object size in
+ * memory, and combine it with aging informations.
+ *
+ * Basically swappability = idle-time * log(estimated size)
+ *
+ * Bigger objects are preferred over smaller objects, but not
+ * proportionally, this is why we use the logarithm. This algorithm is
+ * just a first try and will probably be tuned later. */
+static double computeObjectSwappability(robj *o) {
+    time_t age = server.unixtime - o->vm.atime;
+    long asize = 0;
+    list *l;
+    dict *d;
+    struct dictEntry *de;
+    int z;
+
+    if (age <= 0) return 0;
+    switch(o->type) {
+    case REDIS_STRING:
+        if (o->encoding != REDIS_ENCODING_RAW) {
+            asize = sizeof(*o);
+        } else {
+            asize = sdslen(o->ptr)+sizeof(*o)+sizeof(long)*2;
+        }
+        break;
+    case REDIS_LIST:
+        l = o->ptr;
+        listNode *ln = listFirst(l);
+
+        asize = sizeof(list);
+        if (ln) {
+            robj *ele = ln->value;
+            long elesize;
+
+            elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
+                            (sizeof(*o)+sdslen(ele->ptr)) :
+                            sizeof(*o);
+            asize += (sizeof(listNode)+elesize)*listLength(l);
+        }
+        break;
+    case REDIS_SET:
+    case REDIS_ZSET:
+        z = (o->type == REDIS_ZSET);
+        d = z ? ((zset*)o->ptr)->dict : o->ptr;
+
+        asize = sizeof(dict)+(sizeof(struct dictEntry*)*dictSlots(d));
+        if (z) asize += sizeof(zset)-sizeof(dict);
+        if (dictSize(d)) {
+            long elesize;
+            robj *ele;
+
+            de = dictGetRandomKey(d);
+            ele = dictGetEntryKey(de);
+            elesize = (ele->encoding == REDIS_ENCODING_RAW) ?
+                            (sizeof(*o)+sdslen(ele->ptr)) :
+                            sizeof(*o);
+            asize += (sizeof(struct dictEntry)+elesize)*dictSize(d);
+            if (z) asize += sizeof(zskiplistNode)*dictSize(d);
+        }
+        break;
+    }
+    return (double)asize*log(1+asize);
+}
+
+/* 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) {
+    int j, i;
+    struct dictEntry *best = NULL;
+    double best_swappability = 0;
+    robj *key, *val;
+
+    for (j = 0; j < server.dbnum; j++) {
+        redisDb *db = server.db+j;
+        int maxtries = 1000;
+
+        if (dictSize(db->dict) == 0) continue;
+        for (i = 0; i < 5; i++) {
+            dictEntry *de;
+            double swappability;
+
+            if (maxtries) maxtries--;
+            de = dictGetRandomKey(db->dict);
+            key = dictGetEntryKey(de);
+            val = dictGetEntryVal(de);
+            if (key->storage != REDIS_VM_MEMORY) {
+                if (maxtries) i--; /* don't count this try */
+                continue;
+            }
+            swappability = computeObjectSwappability(val);
+            if (!best || swappability > best_swappability) {
+                best = de;
+                best_swappability = swappability;
+            }
+        }
+    }
+    if (best == NULL) {
+        redisLog(REDIS_DEBUG,"No swappable key found!");
+        return REDIS_ERR;
+    }
+    key = dictGetEntryKey(best);
+    val = dictGetEntryVal(best);
+
+    redisLog(REDIS_DEBUG,"Key with best swappability: %s, %f",
+        key->ptr, best_swappability);
+
+    /* Unshare the key if needed */
+    if (key->refcount > 1) {
+        robj *newkey = dupStringObject(key);
+        decrRefCount(key);
+        key = dictGetEntryKey(best) = newkey;
+    }
+    /* Swap it */
+    if (vmSwapObject(key,val) == REDIS_OK) {
+        dictGetEntryVal(best) = NULL;
+        return REDIS_OK;
+    } else {
+        return REDIS_ERR;
+    }
+}
+
+/* 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. */
+static int vmCanSwapOut(void) {
+    return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
+}
+
+/* Delete a key if swapped. Returns 1 if the key was found, was swapped
+ * and was deleted. Otherwise 0 is returned. */
+static int deleteIfSwapped(redisDb *db, robj *key) {
+    dictEntry *de;
+    robj *foundkey;
+
+    if ((de = dictFind(db->dict,key)) == NULL) return 0;
+    foundkey = dictGetEntryKey(de);
+    if (foundkey->storage == REDIS_VM_MEMORY) return 0;
+    deleteKey(db,key);
+    return 1;
+}
+
 /* ================================= Debugging ============================== */
 
 static void debugCommand(redisClient *c) {
@@ -6274,12 +7175,41 @@ 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\r\n",
+            "+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));
+                val->encoding, rdbSavedObjectLen(val)));
+    } else if (!strcasecmp(c->argv[1]->ptr,"swapout") && c->argc == 3) {
+        dictEntry *de = dictFind(c->db->dict,c->argv[2]);
+        robj *key, *val;
+
+        if (!server.vm_enabled) {
+            addReplySds(c,sdsnew("-ERR Virtual Memory is disabled\r\n"));
+            return;
+        }
+        if (!de) {
+            addReply(c,shared.nokeyerr);
+            return;
+        }
+        key = dictGetEntryKey(de);
+        val = dictGetEntryVal(de);
+        /* If the key is shared we want to create a copy */
+        if (key->refcount > 1) {
+            robj *newkey = dupStringObject(key);
+            decrRefCount(key);
+            key = dictGetEntryKey(de) = newkey;
+        }
+        /* 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) {
+            dictGetEntryVal(de) = NULL;
+            addReply(c,shared.ok);
+        } else {
+            addReply(c,shared.err);
+        }
     } else {
         addReplySds(c,sdsnew(
-            "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>|RELOAD]\r\n"));
+            "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>|SWAPOUT <key>|RELOAD]\r\n"));
     }
 }
 
@@ -6395,8 +7325,8 @@ static void *getMcontextEip(ucontext_t *uc) {
   #else
     return (void*) uc->uc_mcontext->__ss.__eip;
   #endif 
-#elif defined(__i386__) || defined(__X86_64__) /* Linux x86 */
-    return (void*) uc->uc_mcontext.gregs[REG_EIP];
+#elif defined(__i386__) || defined(__X86_64__)  || defined(__x86_64__)
+    return (void*) uc->uc_mcontext.gregs[REG_EIP]; /* Linux 32/64 bit */
 #elif defined(__ia64__) /* Linux IA64 */
     return (void*) uc->uc_mcontext.sc_ip;
 #else
@@ -6437,7 +7367,7 @@ static void segvHandler(int sig, siginfo_t *info, void *secret) {
             redisLog(REDIS_WARNING,"%d redis-server %p %s + %d", i, trace[i], fn, (unsigned int)offset);
         }
     }
-    // free(messages); Don't call free() with possibly corrupted memory.
+    /* free(messages); Don't call free() with possibly corrupted memory. */
     exit(0);
 }