useless debugging messages removed

[redis.git] / redis.c

diff --git a/redis.c b/redis.c
index c4c1cfcb9aaf5a5d1d2133d21cc57b2c91d9fc11..4ed55d71a4c92a69e988c744ff653baf561d95e9 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -75,6 +75,11 @@
 #include "lzf.h"    /* LZF compression library */
 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
 
 #include "lzf.h"    /* LZF compression library */
 #include "pqsort.h" /* Partial qsort for SORT+LIMIT */
 

+/* #define REDIS_HELGRIND_FRIENDLY */
+#if defined(__GNUC__) && defined(REDIS_HELGRIND_FRIENDLY)
+#warning "Remember to undef REDIS_HELGRIND_FRIENDLY before to commit"
+#endif
+

 /* Error codes */
 #define REDIS_OK                0
 #define REDIS_ERR               -1
 /* Error codes */
 #define REDIS_OK                0
 #define REDIS_ERR               -1


@@ -164,6 +169,13 @@
 #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 number of completed I/O jobs to process when the
+ * handelr is called. 1 is the minimum, and also the default, as it allows
+ * to block as little as possible other accessing clients. While Virtual
+ * Memory I/O operations are performed by threads, this operations must
+ * be processed by the main thread when completed 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 */


@@ -397,6 +409,7 @@ struct redisServer {
     pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
     pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
     pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
     pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
     pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
     pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */

+    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


@@ -555,6 +568,7 @@ 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 queueIOJob(iojob *j);
 static int vmWriteObjectOnSwap(robj *o, off_t page);
 static robj *vmReadObjectFromSwap(off_t page, int type);

+static void waitZeroActiveThreads(void);

 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);


@@ -943,6 +957,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 */


@@ -952,6 +967,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 */


@@ -961,6 +977,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 */


@@ -970,6 +987,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 = {


@@ -999,9 +1026,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 */


@@ -1243,18 +1271,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)
         {

+            int retval;
+

             if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
             if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;

-            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!");
-                }
+            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 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;
+            /* 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;

         }
     }
 
         }
     }
 


@@ -1396,7 +1429,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;
     }


@@ -1687,10 +1720,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;


@@ -2052,6 +2086,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 */


@@ -2082,8 +2117,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 */


@@ -2389,6 +2424,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;
     }


@@ -2846,21 +2885,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;


@@ -2901,11 +2935,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;


@@ -3060,6 +3095,7 @@ 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) waitZeroActiveThreads();

     if ((childpid = fork()) == 0) {
         /* Child */
         close(server.fd);
     if ((childpid = fork()) == 0) {
         /* Child */
         close(server.fd);


@@ -5341,9 +5377,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;


@@ -5439,13 +5476,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);


@@ -5470,12 +5509,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);


@@ -5597,6 +5638,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"


@@ -5609,6 +5651,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,


@@ -5619,8 +5662,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;


@@ -6113,9 +6158,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;
         }


@@ -6212,9 +6258,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) {


@@ -6246,9 +6293,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)


@@ -6660,7 +6709,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;
     }


@@ -6764,9 +6818,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);
 


@@ -6862,6 +6917,7 @@ 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) waitZeroActiveThreads();

     if ((childpid = fork()) == 0) {
         /* Child */
         char tmpfile[256];
     if ((childpid = fork()) == 0) {
         /* Child */
         char tmpfile[256];


@@ -6939,6 +6995,10 @@ static void aofRemoveTempFile(pid_t childpid) {
 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");
     if (server.vm_fp == NULL) {
 
     server.vm_fp = fopen("/tmp/redisvm","w+b");
     if (server.vm_fp == NULL) {


@@ -6986,6 +7046,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)


@@ -7052,7 +7117,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) {


@@ -7302,7 +7367,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;
             }


@@ -7390,6 +7462,7 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
 {
     char buf[1];
     int retval;
 {
     char buf[1];
     int retval;

+    int processed = 0;

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


@@ -7403,10 +7476,10 @@ 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);

         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);


@@ -7440,6 +7513,10 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
                 /* Ooops... no space! */
                 freeIOJob(j);
             } else {
                 /* Ooops... no space! */
                 freeIOJob(j);
             } 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);


@@ -7465,7 +7542,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,


@@ -7487,7 +7563,8 @@ static void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
                 }
             }
         }
                 }
             }
         }

-        return; /* XXX REMOVE ME */
+        processed++;
+        if (processed == REDIS_MAX_COMPLETED_JOBS_PROCESSED) return;

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


@@ -7515,19 +7592,27 @@ static void vmCancelThreadedIOJob(robj *o) {
     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;
 
             if (job->canceled) continue; /* Skip this, already canceled. */
             if (compareStringObjects(job->key,o) == 0) {
             iojob *job = ln->value;
 
             if (job->canceled) continue; /* Skip this, already canceled. */
             if (compareStringObjects(job->key,o) == 0) {

-                redisLog(REDIS_DEBUG,"*** CANCELED %p (%s)\n",
-                    (void*)job, (char*)o->ptr);
+                redisLog(REDIS_DEBUG,"*** CANCELED %p (%s) (LIST ID %d)\n",
+                    (void*)job, (char*)o->ptr, i);
+                /* Mark the pages as free since the swap didn't happened
+                 * or happened but is now discarded. */
+                if (job->type == REDIS_IOJOB_DO_SWAP)
+                    vmMarkPagesFree(job->page,job->pages);
+                /* Cancel the job. It depends on the list the job is
+                 * living in. */

                 switch(i) {
                 case 0: /* io_newjobs */
                     /* If the job was yet not processed the best thing to do
                 switch(i) {
                 case 0: /* io_newjobs */
                     /* If the job was yet not processed the best thing to do


@@ -7536,10 +7621,32 @@ static void vmCancelThreadedIOJob(robj *o) {
                     listDelNode(lists[i],ln);
                     break;
                 case 1: /* io_processing */
                     listDelNode(lists[i],ln);
                     break;
                 case 1: /* io_processing */

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

                 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)


@@ -7563,8 +7670,17 @@ static void *IOThreadEntryPoint(void *arg) {
         /* Get a new job to process */
         lockThreadedIO();
         if (listLength(server.io_newjobs) == 0) {
         /* Get a new job to process */
         lockThreadedIO();
         if (listLength(server.io_newjobs) == 0) {

+#ifdef REDIS_HELGRIND_FRIENDLY
+            /* No new jobs? Wait and retry, because to be Helgrind
+             * (valgrind --tool=helgrind) what's needed is to take
+             * the same threads running instead to create/destroy threads
+             * as needed (otherwise valgrind will fail) */
+            unlockThreadedIO();
+            usleep(1); /* Give some time for the I/O thread to work. */
+            continue;
+#endif

             /* No new jobs in queue, exit. */
             /* No new jobs in queue, exit. */

-            redisLog(REDIS_DEBUG,"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();


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

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


@@ -7593,7 +7709,7 @@ static void *IOThreadEntryPoint(void *arg) {
         }
 
         /* Done: insert the job into the processed queue */
         }
 
         /* Done: insert the job into the processed queue */

-        redisLog(REDIS_DEBUG,"Thread %lld completed the job: %p (key %s)\n",
+        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);
             (long long) pthread_self(), (void*)j, (char*)j->key->ptr);
         lockThreadedIO();
         listDelNode(server.io_processing,ln);


@@ -7609,10 +7725,24 @@ 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 waitZeroActiveThreads(void) {
+    while(1) {
+        lockThreadedIO();
+        if (server.io_active_threads == 0) {
+            unlockThreadedIO();
+            return;
+        }
+        unlockThreadedIO();
+        usleep(10000); /* 10 milliseconds */
+    }
+}
+

 /* 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",
 /* 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",