REDIS_MAX_COMPLETED_JOBS_PROCESSED is now in percentage, not number of jobs. Moved...

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index f489072e26c9872238cb3846f21a9ff5fea9f11e..76784debc0ab52a81e82986ebe900e4be946a6f1 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -164,6 +164,12 @@
 #define REDIS_VM_MAX_NEAR_PAGES 65536
 #define REDIS_VM_MAX_RANDOM_JUMP 4096
 #define REDIS_VM_MAX_THREADS 32
 #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 */
 
 /* Client flags */
 #define REDIS_CLOSE 1       /* This client connection should be closed ASAP */


@@ -216,8 +222,8 @@
 #define APPENDFSYNC_EVERYSEC 2
 
 /* We can print the stacktrace, so our assert is defined this way: */
 #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 ============================== */
 
 
 /*================================= Data types ============================== */
 


@@ -376,6 +382,7 @@ struct redisServer {
     int sort_bypattern;
     /* Virtual memory configuration */
     int vm_enabled;
     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;
     off_t vm_page_size;
     off_t vm_pages;
     unsigned long long vm_max_memory;


@@ -395,6 +402,9 @@ struct redisServer {
     list *io_processed; /* List of VM I/O jobs already processed */
     list *io_clients; /* All the clients waiting for SWAP I/O operations */
     pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
     list *io_processed; /* List of VM I/O jobs already processed */
     list *io_clients; /* All the clients waiting for SWAP I/O operations */
     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
     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


@@ -542,7 +552,7 @@ static robj *vmPreviewObject(robj *key);
 static int vmSwapOneObjectBlocking(void);
 static int vmSwapOneObjectThreaded(void);
 static int vmCanSwapOut(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 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);


@@ -551,6 +561,11 @@ 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 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 void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);


@@ -856,7 +871,7 @@ static void redisLog(int level, const char *fmt, ...) {
 
         now = time(NULL);
         strftime(buf,64,"%d %b %H:%M:%S",localtime(&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);
         vfprintf(fp, fmt, ap);
         fprintf(fp,"\n");
         fflush(fp);


@@ -939,6 +954,7 @@ static unsigned int dictEncObjHash(const void *key) {
     return hash;
 }
 
     return hash;
 }
 

+/* Sets type and expires */

 static dictType setDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */
 static dictType setDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */


@@ -948,6 +964,7 @@ static dictType setDictType = {
     NULL                       /* val destructor */
 };
 
     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 */
 static dictType zsetDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */


@@ -957,6 +974,7 @@ static dictType zsetDictType = {
     dictVanillaFree            /* val destructor of malloc(sizeof(double)) */
 };
 
     dictVanillaFree            /* val destructor of malloc(sizeof(double)) */
 };
 

+/* Db->dict */

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


@@ -966,6 +984,16 @@ static dictType hashDictType = {
     dictRedisObjectDestructor   /* val destructor */
 };
 
     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) */
 static dictType keylistDictType = {
 /* Keylist hash table type has unencoded redis objects as keys and
  * lists as values. It's used for blocking operations (BLPOP) */
 static dictType keylistDictType = {


@@ -995,9 +1023,10 @@ static void closeTimedoutClients(void) {
     redisClient *c;
     listNode *ln;
     time_t now = time(NULL);
     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 */
         c = listNodeValue(ln);
         if (server.maxidletime &&
             !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */


@@ -1239,21 +1268,23 @@ static int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientD
         while (server.vm_enabled && zmalloc_used_memory() >
                 server.vm_max_memory)
         {
         while (server.vm_enabled && zmalloc_used_memory() >
                 server.vm_max_memory)
         {

-            if (listLength(server.objfreelist)) {
-                freeOneObjectFromFreelist();
-            } else {
-                if (vmSwapOneObjectThreaded() == 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!");
-                    }
-                }
-                /* Note that we freed just one object, because anyway when
-                 * the I/O thread in charge to swap this object out will
-                 * do its work, the handler of completed jobs will try to swap
-                 * more objects if we are out of memory. */
-                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;

         }
     }
 
         }
     }
 


@@ -1343,6 +1374,7 @@ static void initServerConfig() {
     server.blockedclients = 0;
     server.maxmemory = 0;
     server.vm_enabled = 0;
     server.blockedclients = 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_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 */


@@ -1395,7 +1427,7 @@ static void initServer() {
     }
     for (j = 0; j < server.dbnum; j++) {
         server.db[j].dict = dictCreate(&hashDictType,NULL);
     }
     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);
         server.db[j].id = j;
     }
         server.db[j].blockingkeys = dictCreate(&keylistDictType,NULL);
         server.db[j].id = j;
     }


@@ -1586,15 +1618,18 @@ static void loadServerConfig(char *filename) {
                 goto loaderr;
             }
         } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
                 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) {
         } 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) {
         } 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-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) {
         } 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) {


@@ -1686,10 +1721,11 @@ static void glueReplyBuffersIfNeeded(redisClient *c) {
     int copylen = 0;
     char buf[GLUEREPLY_UP_TO];
     listNode *ln;
     int copylen = 0;
     char buf[GLUEREPLY_UP_TO];
     listNode *ln;

+    listIter li;

     robj *o;
 
     robj *o;
 

-    listRewind(c->reply);
-    while((ln = listYield(c->reply))) {
+    listRewind(c->reply,&li);
+    while((ln = listNext(&li))) {

         int objlen;
 
         o = ln->value;
         int objlen;
 
         o = ln->value;


@@ -2051,6 +2087,7 @@ static int processCommand(redisClient *c) {
 
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
     listNode *ln;
 
 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 */
     int outc = 0, j;
     robj **outv;
     /* (args*2)+1 is enough room for args, spaces, newlines */


@@ -2081,8 +2118,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]);
      * 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 */
         redisClient *slave = ln->value;
 
         /* Don't feed slaves that are still waiting for BGSAVE to start */


@@ -2369,12 +2406,15 @@ static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 static robj *createObject(int type, void *ptr) {
     robj *o;
 
 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 (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) {
     } else {
         if (server.vm_enabled) {

+            pthread_mutex_unlock(&server.obj_freelist_mutex);

             o = zmalloc(sizeof(*o));
         } else {
             o = zmalloc(sizeof(*o)-sizeof(struct redisObjectVM));
             o = zmalloc(sizeof(*o));
         } else {
             o = zmalloc(sizeof(*o)-sizeof(struct redisObjectVM));


@@ -2385,6 +2425,10 @@ static robj *createObject(int type, void *ptr) {
     o->ptr = ptr;
     o->refcount = 1;
     if (server.vm_enabled) {
     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;
     }
         o->vm.atime = server.unixtime;
         o->storage = REDIS_VM_MEMORY;
     }


@@ -2454,7 +2498,8 @@ static void incrRefCount(robj *o) {
 static void decrRefCount(void *obj) {
     robj *o = obj;
 
 static void decrRefCount(void *obj) {
     robj *o = obj;
 

-    /* Object is swapped out, or in the process of being loaded. */
+    /* 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 (server.vm_enabled &&
         (o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING))
     {


@@ -2465,9 +2510,11 @@ static void decrRefCount(void *obj) {
         redisAssert(o->type == REDIS_STRING);
         freeStringObject(o);
         vmMarkPagesFree(o->vm.page,o->vm.usedpages);
         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);
         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;
     }
         server.vm_stats_swapped_objects--;
         return;
     }


@@ -2483,9 +2530,11 @@ static void decrRefCount(void *obj) {
         case REDIS_HASH: freeHashObject(o); break;
         default: redisAssert(0 != 0); break;
         }
         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 (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
             !listAddNodeHead(server.objfreelist,o))
             zfree(o);

+        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);

     }
 }
 
     }
 }
 


@@ -2838,21 +2887,16 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
 static int rdbSaveStringObject(FILE *fp, robj *obj) {
     int retval;
 
 static int rdbSaveStringObject(FILE *fp, robj *obj) {
     int retval;
 

-    if (obj->storage == REDIS_VM_MEMORY &&
-       obj->encoding != REDIS_ENCODING_RAW)
-    {
+    /* 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 {
         obj = getDecodedObject(obj);
         retval = rdbSaveStringObjectRaw(fp,obj);
         decrRefCount(obj);
     } else {

-        /* This is a fast path when we are sure the object is not encoded.
-         * Note that's any *faster* actually as we needed to add the conditional
-         * but because this may happen in a background process we don't want
-         * to touch the object fields with incr/decrRefCount in order to
-         * preveny copy on write of pages.
-         *
-         * Also incrRefCount() will have a failing assert() if we try to call
-         * it against an object with storage != REDIS_VM_MEMORY. */

         retval = rdbSaveStringObjectRaw(fp,obj);
     }
     return retval;
         retval = rdbSaveStringObjectRaw(fp,obj);
     }
     return retval;


@@ -2893,11 +2937,12 @@ static int rdbSaveObject(FILE *fp, robj *o) {
     } else if (o->type == REDIS_LIST) {
         /* Save a list value */
         list *list = o->ptr;
     } else if (o->type == REDIS_LIST) {
         /* Save a list value */
         list *list = o->ptr;

+        listIter li;

         listNode *ln;
 
         listNode *ln;
 

-        listRewind(list);

         if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
         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;
             robj *eleobj = listNodeValue(ln);
 
             if (rdbSaveStringObject(fp,eleobj) == -1) return -1;


@@ -2963,6 +3008,12 @@ static int rdbSave(char *filename) {
     int j;
     time_t now = time(NULL);
 
     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) {
     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
     fp = fopen(tmpfile,"w");
     if (!fp) {


@@ -3052,8 +3103,10 @@ static int rdbSaveBackground(char *filename) {
     pid_t childpid;
 
     if (server.bgsavechildpid != -1) return REDIS_ERR;
     pid_t childpid;
 
     if (server.bgsavechildpid != -1) return REDIS_ERR;

+    if (server.vm_enabled) waitEmptyIOJobsQueue();

     if ((childpid = fork()) == 0) {
         /* Child */
     if ((childpid = fork()) == 0) {
         /* Child */

+        if (server.vm_enabled) vmReopenSwapFile();

         close(server.fd);
         if (rdbSave(filename) == REDIS_OK) {
             exit(0);
         close(server.fd);
         if (rdbSave(filename) == REDIS_OK) {
             exit(0);


@@ -3724,6 +3777,7 @@ static void shutdownCommand(redisClient *c) {
     if (server.appendonly) {
         /* Append only file: fsync() the AOF and exit */
         fsync(server.appendfd);
     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 */
         exit(0);
     } else {
         /* Snapshotting. Perform a SYNC SAVE and exit */


@@ -3732,6 +3786,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...");
                 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.
             exit(0);
         } else {
             /* Ooops.. error saving! The best we can do is to continue operating.


@@ -5333,9 +5388,10 @@ static void sortCommand(redisClient *c) {
     if (sortval->type == REDIS_LIST) {
         list *list = sortval->ptr;
         listNode *ln;
     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;
             robj *ele = ln->value;
             vector[j].obj = ele;
             vector[j].u.score = 0;


@@ -5431,13 +5487,15 @@ static void sortCommand(redisClient *c) {
         addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
         for (j = start; j <= end; j++) {
             listNode *ln;
         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);
             }
             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);
                 redisSortOperation *sop = ln->value;
                 robj *val = lookupKeyByPattern(c->db,sop->pattern,
                     vector[j].obj);


@@ -5462,12 +5520,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;
         /* 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);
             }
             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);
                 redisSortOperation *sop = ln->value;
                 robj *val = lookupKeyByPattern(c->db,sop->pattern,
                     vector[j].obj);


@@ -5589,6 +5649,7 @@ static sds genRedisInfoString(void) {
         );
     }
     if (server.vm_enabled) {
         );
     }
     if (server.vm_enabled) {

+        lockThreadedIO();

         info = sdscatprintf(info,
             "vm_conf_max_memory:%llu\r\n"
             "vm_conf_page_size:%llu\r\n"
         info = sdscatprintf(info,
             "vm_conf_max_memory:%llu\r\n"
             "vm_conf_page_size:%llu\r\n"


@@ -5601,6 +5662,7 @@ static sds genRedisInfoString(void) {
             "vm_stats_io_processing_len:%lu\r\n"
             "vm_stats_io_processed_len:%lu\r\n"
             "vm_stats_io_waiting_clients:%lu\r\n"
             "vm_stats_io_processing_len:%lu\r\n"
             "vm_stats_io_processed_len:%lu\r\n"
             "vm_stats_io_waiting_clients:%lu\r\n"

+            "vm_stats_io_active_threads:%lu\r\n"

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


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

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

         );
         );

+        unlockThreadedIO();

     }
     for (j = 0; j < server.dbnum; j++) {
         long long keys, vkeys;
     }
     for (j = 0; j < server.dbnum; j++) {
         long long keys, vkeys;


@@ -6105,9 +6169,10 @@ static void syncCommand(redisClient *c) {
          * registering differences since the server forked to save */
         redisClient *slave;
         listNode *ln;
          * 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;
         }
             slave = ln->value;
             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
         }


@@ -6204,9 +6269,10 @@ static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
 static void updateSlavesWaitingBgsave(int bgsaveerr) {
     listNode *ln;
     int startbgsave = 0;
 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) {
         redisClient *slave = ln->value;
 
         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {


@@ -6238,9 +6304,11 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
     }
     if (startbgsave) {
         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
     }
     if (startbgsave) {
         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {

-            listRewind(server.slaves);
+            listIter li;
+
+            listRewind(server.slaves,&li);

             redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
             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)
                 redisClient *slave = ln->value;
 
                 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)


@@ -6378,16 +6446,25 @@ static void slaveofCommand(redisClient *c) {
 
 /* ============================ Maxmemory directive  ======================== */
 
 
 /* ============================ 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;
 
     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
 }
 
 /* This function gets called when 'maxmemory' is set on the config file to limit


@@ -6403,35 +6480,32 @@ static void freeOneObjectFromFreelist(void) {
  */
 static void freeMemoryIfNeeded(void) {
     while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
  */
 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... */

     }
 }
 
     }
 }
 


@@ -6646,7 +6720,12 @@ static int fwriteBulk(FILE *fp, robj *obj) {
     char buf[128];
     int decrrc = 0;
 
     char buf[128];
     int decrrc = 0;
 

-    if (obj->storage == REDIS_VM_MEMORY && obj->encoding != REDIS_ENCODING_RAW){
+    /* 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;
     }
         obj = getDecodedObject(obj);
         decrrc = 1;
     }


@@ -6750,9 +6829,10 @@ static int rewriteAppendOnlyFile(char *filename) {
                 /* Emit the RPUSHes needed to rebuild the list */
                 list *list = o->ptr;
                 listNode *ln;
                 /* 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);
 
                     char cmd[]="*3\r\n$5\r\nRPUSH\r\n";
                     robj *eleobj = listNodeValue(ln);
 


@@ -6848,11 +6928,13 @@ static int rewriteAppendOnlyFileBackground(void) {
     pid_t childpid;
 
     if (server.bgrewritechildpid != -1) return REDIS_ERR;
     pid_t childpid;
 
     if (server.bgrewritechildpid != -1) return REDIS_ERR;

+    if (server.vm_enabled) waitEmptyIOJobsQueue();

     if ((childpid = fork()) == 0) {
         /* Child */
         char tmpfile[256];
     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);
         snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
             exit(0);


@@ -6922,13 +7004,40 @@ static void aofRemoveTempFile(pid_t childpid) {
  */
 
 /* =================== Virtual Memory - Blocking Side  ====================== */
  */
 
 /* =================== 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];
 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) {
     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);
         exit(1);
     }
     server.vm_fd = fileno(server.vm_fp);


@@ -6951,9 +7060,6 @@ 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);
     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();
 
     /* Initialize threaded I/O (used by Virtual Memory) */
     server.io_newjobs = listCreate();


@@ -6961,6 +7067,8 @@ static void vmInit(void) {
     server.io_processed = listCreate();
     server.io_clients = listCreate();
     pthread_mutex_init(&server.io_mutex,NULL);
     server.io_processed = listCreate();
     server.io_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."
     server.io_active_threads = 0;
     if (pipe(pipefds) == -1) {
         redisLog(REDIS_WARNING,"Unable to intialized VM: pipe(2): %s. Exiting."


@@ -6970,6 +7078,11 @@ static void vmInit(void) {
     server.io_ready_pipe_read = pipefds[0];
     server.io_ready_pipe_write = pipefds[1];
     redisAssert(anetNonBlock(NULL,server.io_ready_pipe_read) != ANET_ERR);
     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)
     /* Listen for events in the threaded I/O pipe */
     if (aeCreateFileEvent(server.el, server.io_ready_pipe_read, AE_READABLE,
         vmThreadedIOCompletedJob, NULL) == AE_ERR)


@@ -6980,6 +7093,7 @@ static void vmInit(void) {
 static void vmMarkPageUsed(off_t page) {
     off_t byte = page/8;
     int bit = page&7;
 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);
     server.vm_bitmap[byte] |= 1<<bit;
     redisLog(REDIS_DEBUG,"Mark used: %lld (byte:%lld bit:%d)\n",
         (long long)page, (long long)byte, bit);


@@ -6998,7 +7112,10 @@ static void vmMarkPagesUsed(off_t page, off_t count) {
 static void vmMarkPageFree(off_t page) {
     off_t byte = page/8;
     int bit = page&7;
 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);
     server.vm_bitmap[byte] &= ~(1<<bit);

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

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


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

+    if (server.vm_stats_used_pages > 100000000) {
+        *((char*)-1) = 'x';
+    }

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


@@ -7036,7 +7156,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!"
  */
  * 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) {
     off_t base, offset = 0, since_jump = 0, numfree = 0;
 
     if (server.vm_near_pages == REDIS_VM_MAX_NEAR_PAGES) {


@@ -7049,7 +7169,6 @@ static int vmFindContiguousPages(off_t *first, int n) {
     while(offset < server.vm_pages) {
         off_t this = base+offset;
 
     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 we overflow, restart from page zero */
         if (this >= server.vm_pages) {
             this -= server.vm_pages;


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

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

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


@@ -7090,6 +7210,21 @@ static int vmFindContiguousPages(off_t *first, int n) {
     return REDIS_ERR;
 }
 
     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
 /* 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


@@ -7101,13 +7236,7 @@ static int vmSwapObjectBlocking(robj *key, robj *val) {
     assert(key->storage == REDIS_VM_MEMORY);
     assert(key->refcount == 1);
     if (vmFindContiguousPages(&page,pages) == REDIS_ERR) return REDIS_ERR;
     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 vmSwapObjectBlocking(): 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;
     key->vm.page = page;
     key->vm.usedpages = pages;
     key->storage = REDIS_VM_SWAPPED;


@@ -7123,26 +7252,35 @@ static int vmSwapObjectBlocking(robj *key, robj *val) {
     return REDIS_OK;
 }
 
     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;
+static robj *vmReadObjectFromSwap(off_t page, int type) {
+    robj *o;

 
 

-    redisAssert(key->storage == REDIS_VM_SWAPPED);
-    if (fseeko(server.vm_fp,key->vm.page*server.vm_page_size,SEEK_SET) == -1) {
+    if (server.vm_enabled) pthread_mutex_lock(&server.io_swapfile_mutex);
+    if (fseeko(server.vm_fp,page*server.vm_page_size,SEEK_SET) == -1) {

         redisLog(REDIS_WARNING,
             "Unrecoverable VM problem in vmLoadObject(): can't seek: %s",
             strerror(errno));
         exit(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) {
+    o = rdbLoadObject(type,server.vm_fp);
+    if (o == NULL) {

         redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmLoadObject(): can't load object from swap file: %s", strerror(errno));
         exit(1);
     }
         redisLog(REDIS_WARNING, "Unrecoverable VM problem in vmLoadObject(): 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.
+ *
+ * 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);
+    val = vmReadObjectFromSwap(key->vm.page,key->vtype);

     if (!preview) {
         key->storage = REDIS_VM_MEMORY;
         key->vm.atime = server.unixtime;
     if (!preview) {
         key->storage = REDIS_VM_MEMORY;
         key->vm.atime = server.unixtime;


@@ -7268,7 +7406,14 @@ static int vmSwapOneObject(int usethreads) {
             de = dictGetRandomKey(db->dict);
             key = dictGetEntryKey(de);
             val = dictGetEntryVal(de);
             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;
             }
                 if (maxtries) i--; /* don't count this try */
                 continue;
             }


@@ -7356,6 +7501,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
 {
     char buf[1];
     int retval;
 {
     char buf[1];
     int retval;

+    int processed = 0;
+    int toprocess = -1;

     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
     REDIS_NOTUSED(privdata);
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
     REDIS_NOTUSED(privdata);


@@ -7369,10 +7516,14 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
         struct dictEntry *de;
 
         redisLog(REDIS_DEBUG,"Processing I/O completed job");
         struct dictEntry *de;
 
         redisLog(REDIS_DEBUG,"Processing I/O completed job");

-        assert(listLength(server.io_processed) != 0);

 
         /* Get the processed element (the oldest one) */
         lockThreadedIO();
 
         /* 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);
         ln = listFirst(server.io_processed);
         j = ln->value;
         listDelNode(server.io_processed,ln);


@@ -7384,11 +7535,7 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
         }
         /* Post process it in the main thread, as there are things we
          * can do just here to avoid race conditions and/or invasive locks */
         }
         /* 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 type: %d, key at %p (%s) refcount: %d\n", j->type, (void*)j->key, (char*)j->key->ptr, j->key->refcount);
-        if (j->key->refcount <= 0) {
-            printf("Ooops ref count is <= 0!\n");
-            exit(1);
-        }
+        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);
         de = dictFind(j->db->dict,j->key);
         assert(de != NULL);
         key = dictGetEntryKey(de);


@@ -7406,10 +7553,18 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             /* 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. */
             /* 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 (vmFindContiguousPages(&j->page,j->pages) == REDIS_ERR) {
-                /* Ooops... no space! */
+            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);
                 freeIOJob(j);

+                key->storage = REDIS_VM_MEMORY; /* undo operation */

             } else {
             } 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);
                 j->type = REDIS_IOJOB_DO_SWAP;
                 lockThreadedIO();
                 queueIOJob(j);


@@ -7419,6 +7574,15 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             robj *val;
 
             /* Key swapped. We can finally free some memory. */
             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;
             val = dictGetEntryVal(de);
             key->vm.page = j->page;
             key->vm.usedpages = j->pages;


@@ -7426,7 +7590,6 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             key->vtype = j->val->type;
             decrRefCount(val); /* Deallocate the object from memory. */
             dictGetEntryVal(de) = NULL;
             key->vtype = j->val->type;
             decrRefCount(val); /* Deallocate the object from memory. */
             dictGetEntryVal(de) = NULL;

-            vmMarkPagesUsed(j->page,j->pages);

             redisLog(REDIS_DEBUG,
                 "VM: object %s swapped out at %lld (%lld pages) (threaded)",
                 (unsigned char*) key->ptr,
             redisLog(REDIS_DEBUG,
                 "VM: object %s swapped out at %lld (%lld pages) (threaded)",
                 (unsigned char*) key->ptr,


@@ -7436,7 +7599,7 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             freeIOJob(j);
             /* Put a few more swap requests in queue if we are still
              * out of memory */
             freeIOJob(j);
             /* Put a few more swap requests in queue if we are still
              * out of memory */

-            if (zmalloc_used_memory() > server.vm_max_memory) {
+            if (vmCanSwapOut() && zmalloc_used_memory() > server.vm_max_memory){

                 int more = 1;
                 while(more) {
                     lockThreadedIO();
                 int more = 1;
                 while(more) {
                     lockThreadedIO();


@@ -7448,6 +7611,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
                 }
             }
         }
                 }
             }
         }

+        processed++;
+        if (processed == toprocess) return;

     }
     if (retval < 0 && errno != EAGAIN) {
         redisLog(REDIS_WARNING,
     }
     if (retval < 0 && errno != EAGAIN) {
         redisLog(REDIS_WARNING,


@@ -7468,37 +7633,68 @@ static void unlockThreadedIO(void) {
  * processed, otherwise make sure to flag it as canceled. */
 static void vmCancelThreadedIOJob(robj *o) {
     list *lists[3] = {
  * processed, otherwise make sure to flag it as canceled. */
 static void vmCancelThreadedIOJob(robj *o) {
     list *lists[3] = {

-        server.io_newjobs, server.io_processing, server.io_processed
+        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);
     };
     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;
     lockThreadedIO();
     /* Search for a matching key in one of the queues */
     for (i = 0; i < 3; i++) {
         listNode *ln;

+        listIter li;

 
 

-        listRewind(lists[i]);
-        while ((ln = listYield(lists[i])) != NULL) {
+        listRewind(lists[i],&li);
+        while ((ln = listNext(&li)) != NULL) {

             iojob *job = ln->value;
 
             iojob *job = ln->value;
 

+            if (job->canceled) continue; /* Skip this, already canceled. */

             if (compareStringObjects(job->key,o) == 0) {
             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 */
                 switch(i) {
                 case 0: /* io_newjobs */

-                    /* If the job was not yet processed the best thing to do
+                    /* If the job was yet not processed the best thing to do

                      * is to remove it from the queue at all */
                      * is to remove it from the queue at all */

-                    decrRefCount(job->key);
-                    if (job->type == REDIS_IOJOB_PREPARE_SWAP ||
-                        job->type == REDIS_IOJOB_DO_SWAP)
-                        decrRefCount(job->val);
+                    freeIOJob(job);

                     listDelNode(lists[i],ln);
                     listDelNode(lists[i],ln);

-                    zfree(job);

                     break;
                 case 1: /* io_processing */
                     break;
                 case 1: /* io_processing */

+                    /* Oh Shi- the thread is messing with the Job, and
+                     * probably with the object if this is a
+                     * PREPARE_SWAP or DO_SWAP job. Better to wait for the
+                     * job to move into the next queue... */
+                    if (job->type != REDIS_IOJOB_LOAD) {
+                        /* Yes, we try again and again until the job
+                         * is completed. */
+                        unlockThreadedIO();
+                        /* But let's wait some time for the I/O thread
+                         * to finish with this job. After all this condition
+                         * should be very rare. */
+                        usleep(1);
+                        goto again;
+                    } else {
+                        job->canceled = 1;
+                        break;
+                    }

                 case 2: /* io_processed */
                 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;
                 }
                     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)
                 if (o->storage == REDIS_VM_LOADING)
                     o->storage = REDIS_VM_SWAPPED;
                 else if (o->storage == REDIS_VM_SWAPPING)


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

-            printf("Thread %lld exiting, nothing to do\n",
+            redisLog(REDIS_DEBUG,"Thread %lld exiting, nothing to do",

                 (long long) pthread_self());
             server.io_active_threads--;
             unlockThreadedIO();
                 (long long) pthread_self());
             server.io_active_threads--;
             unlockThreadedIO();


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

-        printf("Thread %lld got a new job: %p about key '%s'\n",
-            (long long) pthread_self(), (void*)j, (char*)j->key->ptr);
+        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) {
 
         /* Process the Job */
         if (j->type == REDIS_IOJOB_LOAD) {


@@ -7547,11 +7743,13 @@ static void *IOThreadEntryPoint(void *arg) {
             j->pages = rdbSavedObjectPages(j->val,fp);
             fclose(fp);
         } else if (j->type == REDIS_IOJOB_DO_SWAP) {
             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 */
         }
 
         /* Done: insert the job into the processed queue */

-        printf("Thread %lld completed the job: %p\n",
-            (long long) pthread_self(), (void*)j);
+        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);
         lockThreadedIO();
         listDelNode(server.io_processing,ln);
         listAddNodeTail(server.io_processed,j);


@@ -7566,12 +7764,54 @@ static void *IOThreadEntryPoint(void *arg) {
 static void spawnIOThread(void) {
     pthread_t thread;
 
 static void spawnIOThread(void) {
     pthread_t thread;
 

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

     server.io_active_threads++;
 }
 
     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) {
+    fclose(server.vm_fp);
+    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) {
 /* 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();
     listAddNodeTail(server.io_newjobs,j);
     if (server.io_active_threads < server.vm_max_threads)
         spawnIOThread();


@@ -7640,7 +7880,7 @@ static void debugCommand(redisClient *c) {
                 "+Key at:%p refcount:%d, value at:%p refcount:%d "
                 "encoding:%d serializedlength:%lld\r\n",
                 (void*)key, key->refcount, (void*)val, val->refcount,
                 "+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,NULL)));
+                val->encoding, (long long) rdbSavedObjectLen(val,NULL)));

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


@@ -7683,9 +7923,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,"=== 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';
 #ifdef HAVE_BACKTRACE
     redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)");
     *((char*)-1) = 'x';


@@ -7721,7 +7961,6 @@ static void daemonize(void) {
     FILE *fp;
 
     if (fork() != 0) exit(0); /* parent exits */
     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
     setsid(); /* create a new session */
 
     /* Every output goes to /dev/null. If Redis is daemonized but