diskstore cache bug fixing

diff --git a/src/db.c b/src/db.c
index cf99bbb299780f3ac81e8e53b908d6d0afc28235..f360607765cbec7fe21371eea58272ad0d6b6d2e 100644 (file)
--- a/src/db.c
+++ b/src/db.c
@@ -21,6 +21,7 @@ robj *lookupKey(redisDb *db, robj *key) {
             /* FIXME: change this code to just wait for our object to
              * get out of the IO Job. */
             waitEmptyIOJobsQueue();
             /* FIXME: change this code to just wait for our object to
              * get out of the IO Job. */
             waitEmptyIOJobsQueue();

+            processAllPendingIOJobs();

             redisAssert(val->storage != REDIS_DS_SAVING);
         }
         server.stat_keyspace_hits++;
             redisAssert(val->storage != REDIS_DS_SAVING);
         }
         server.stat_keyspace_hits++;

diff --git a/src/dscache.c b/src/dscache.c
index fe3f1c1dd20fbd25f5a777cb11a8ff6205ecf9f1..05112cbb7303e2b7510f3bfc8e1661f8638dc2cb 100644 (file)
--- a/src/dscache.c
+++ b/src/dscache.c
@@ -404,16 +404,14 @@ void spawnIOThread(void) {
     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. */
+/* Wait that all the pending IO Jobs are processed */

 void waitEmptyIOJobsQueue(void) {
     while(1) {
         int io_processed_len;
 
         lockThreadedIO();
         if (listLength(server.io_newjobs) == 0 &&
 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)
+            listLength(server.io_processing) == 0)

         {
             unlockThreadedIO();
             return;
         {
             unlockThreadedIO();
             return;


@@ -434,6 +432,21 @@ void waitEmptyIOJobsQueue(void) {
     }
 }
 
     }
 }
 

+/* Process all the IO Jobs already completed by threads but still waiting
+ * processing from the main thread. */
+void processAllPendingIOJobs(void) {
+    while(1) {
+        int io_processed_len;
+
+        lockThreadedIO();
+        io_processed_len = listLength(server.io_processed);
+        unlockThreadedIO();
+        if (io_processed_len == 0) return;
+        vmThreadedIOCompletedJob(NULL,server.io_ready_pipe_read,
+                                                    (void*)0xdeadbeef,0);
+    }
+}
+

 /* This function must be called while with threaded IO locked */
 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 */
 void queueIOJob(iojob *j) {
     redisLog(REDIS_DEBUG,"Queued IO Job %p type %d about key '%s'\n",

diff --git a/src/redis.h b/src/redis.h
index 3a5a274b02558820486149d9e52eaba2f4e9fdf7..bb9aee7a991f3d4946746c51225902f62e15390f 100644 (file)
--- a/src/redis.h
+++ b/src/redis.h
@@ -799,6 +799,7 @@ void unlockThreadedIO(void);
 void freeIOJob(iojob *j);
 void queueIOJob(iojob *j);
 void waitEmptyIOJobsQueue(void);
 void freeIOJob(iojob *j);
 void queueIOJob(iojob *j);
 void waitEmptyIOJobsQueue(void);

+void processAllPendingIOJobs(void);

 void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd);
 void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd);