fixed diskstore race condition

[redis.git] / src / dscache.c

diff --git a/src/dscache.c b/src/dscache.c
index aeeae40a107ca4a3a7858d6860eead989c3c1afc..cbe9bb016cac41e8637d9c49ef512ad45aafe70d 100644 (file)
--- a/src/dscache.c
+++ b/src/dscache.c
@@ -212,7 +212,7 @@ int cacheFreeOneEntry(void) {
         }
     }
     if (best == NULL) {
-        /* Was not able to fix a single object... we should check if our
+        /* Not able to free a single object? we should check if our
          * IO queues have stuff in queue, and try to consume the queue
          * otherwise we'll use an infinite amount of memory if changes to
          * the dataset are faster than I/O */
@@ -240,13 +240,6 @@ int cacheFreeOneEntry(void) {
     return REDIS_OK;
 }
 
-/* Return true if it's safe to swap out objects in a given moment.
- * Basically we don't want to swap objects out while there is a BGSAVE
- * or a BGAEOREWRITE running in backgroud. */
-int dsCanTouchDiskStore(void) {
-    return (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1);
-}
-
 /* ==================== Disk store negative caching  ========================
  *
  * When disk store is enabled, we need negative caching, that is, to remember
@@ -323,7 +316,14 @@ void freeIOJob(iojob *j) {
 
 /* Every time a thread finished a Job, it writes a byte into the write side
  * of an unix pipe in order to "awake" the main thread, and this function
- * is called. */
+ * is called.
+ *
+ * If privdata == NULL the function will try to put more jobs in the queue
+ * of IO jobs to process as more room is made. privdata is equal to NULL
+ * when the function is called from the event loop, so we want to push
+ * more IO jobs in the queue. Instead when the function is called by
+ * other functions that want to create a write-barrier to avoid race 
+ * conditions we don't push new jobs in the queue. */
 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             int mask)
 {
@@ -331,7 +331,6 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
     int retval, processed = 0, toprocess = -1;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
-    REDIS_NOTUSED(privdata);
 
     /* For every byte we read in the read side of the pipe, there is one
      * I/O job completed to process. */
@@ -384,12 +383,12 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             }
             cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_LOADINPROG);
             handleClientsBlockedOnSwappedKey(j->db,j->key);
-            freeIOJob(j);
         } else if (j->type == REDIS_IOJOB_SAVE) {
             cacheScheduleIODelFlag(j->db,j->key,REDIS_IO_SAVEINPROG);
-            freeIOJob(j);
         }
+        freeIOJob(j);
         processed++;
+        if (privdata == NULL) cacheScheduleIOPushJobs(0);
         if (processed == toprocess) return;
     }
     if (retval < 0 && errno != EAGAIN) {
@@ -411,6 +410,7 @@ void *IOThreadEntryPoint(void *arg) {
     iojob *j;
     listNode *ln;
     REDIS_NOTUSED(arg);
+    long long start;
 
     pthread_detach(pthread_self());
     lockThreadedIO();
@@ -423,6 +423,7 @@ void *IOThreadEntryPoint(void *arg) {
             redisLog(REDIS_DEBUG,"[T] signal received");
             continue;
         }
+        start = ustime();
         redisLog(REDIS_DEBUG,"[T] %ld IO jobs to process",
             listLength(server.io_newjobs));
         ln = listFirst(server.io_newjobs);
@@ -464,6 +465,7 @@ void *IOThreadEntryPoint(void *arg) {
 
         /* Signal the main thread there is new stuff to process */
         redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);
+        redisLog(REDIS_DEBUG,"TIME (%c): %lld\n", j->type == REDIS_IOJOB_LOAD ? 'L' : 'S', ustime()-start);
     }
     /* never reached, but that's the full pattern... */
     unlockThreadedIO();
@@ -585,8 +587,6 @@ 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();
 }
 
 /* Consume all the IO scheduled operations, and all the thread IO jobs
@@ -739,7 +739,7 @@ void cacheScheduleIO(redisDb *db, robj *key, int type) {
  * scheduled completion time, but just do the operation ASAP. This is useful
  * when we need to reclaim memory from the IO queue.
  */
-#define MAX_IO_JOBS_QUEUE 100
+#define MAX_IO_JOBS_QUEUE 10
 int cacheScheduleIOPushJobs(int flags) {
     time_t now = time(NULL);
     listNode *ln;
@@ -890,8 +890,16 @@ int waitForSwappedKey(redisClient *c, robj *key) {
     listAddNodeTail(l,c);
 
     /* Are we already loading the key from disk? If not create a job */
-    if (de == NULL)
-        cacheScheduleIO(c->db,key,REDIS_IO_LOAD);
+    if (de == NULL) {
+        int flags = cacheScheduleIOGetFlags(c->db,key);
+
+        /* It is possible that even if there are no clients waiting for
+         * a load operation, still we have a load operation in progress.
+         * For instance think to a client performing a GET and then
+         * closing the connection */
+        if ((flags & (REDIS_IO_LOAD|REDIS_IO_LOADINPROG)) == 0)
+            cacheScheduleIO(c->db,key,REDIS_IO_LOAD);
+    }
     return 1;
 }