less delays in diskstore when there are many writes. more work to do.

diff --git a/src/dscache.c b/src/dscache.c
index aeeae40a107ca4a3a7858d6860eead989c3c1afc..15134cc710fb6e85bc0e086fc69424a3c5363f23 100644 (file)
--- a/src/dscache.c
+++ b/src/dscache.c
@@ -323,7 +323,10 @@ 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
 
 /* 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. */

 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             int mask)
 {
 void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             int mask)
 {


@@ -331,7 +334,6 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
     int retval, processed = 0, toprocess = -1;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
     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. */
 
     /* For every byte we read in the read side of the pipe, there is one
      * I/O job completed to process. */


@@ -390,6 +392,7 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
             freeIOJob(j);
         }
         processed++;
             freeIOJob(j);
         }
         processed++;

+        if (privdata != NULL) cacheScheduleIOPushJobs(0);

         if (processed == toprocess) return;
     }
     if (retval < 0 && errno != EAGAIN) {
         if (processed == toprocess) return;
     }
     if (retval < 0 && errno != EAGAIN) {


@@ -411,6 +414,7 @@ void *IOThreadEntryPoint(void *arg) {
     iojob *j;
     listNode *ln;
     REDIS_NOTUSED(arg);
     iojob *j;
     listNode *ln;
     REDIS_NOTUSED(arg);

+    long long start;

 
     pthread_detach(pthread_self());
     lockThreadedIO();
 
     pthread_detach(pthread_self());
     lockThreadedIO();


@@ -423,6 +427,7 @@ void *IOThreadEntryPoint(void *arg) {
             redisLog(REDIS_DEBUG,"[T] signal received");
             continue;
         }
             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);
         redisLog(REDIS_DEBUG,"[T] %ld IO jobs to process",
             listLength(server.io_newjobs));
         ln = listFirst(server.io_newjobs);


@@ -464,6 +469,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);
 
         /* Signal the main thread there is new stuff to process */
         redisAssert(write(server.io_ready_pipe_write,"x",1) == 1);

+        redisLog(REDIS_WARNING,"TIME (%c): %lld\n", j->type == REDIS_IOJOB_LOAD ? 'L' : 'S', ustime()-start);

     }
     /* never reached, but that's the full pattern... */
     unlockThreadedIO();
     }
     /* never reached, but that's the full pattern... */
     unlockThreadedIO();


@@ -739,7 +745,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.
  */
  * 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;
 int cacheScheduleIOPushJobs(int flags) {
     time_t now = time(NULL);
     listNode *ln;

diff --git a/src/redis.h b/src/redis.h
index 6d49243dbba0568765591a6e58eaa78557c1948e..800f03f847efcb2acebe0da41f60cad89e76f3d8 100644 (file)
--- a/src/redis.h
+++ b/src/redis.h
@@ -637,6 +637,9 @@ dictType hashDictType;
  * Functions prototypes
  *----------------------------------------------------------------------------*/
 
  * Functions prototypes
  *----------------------------------------------------------------------------*/
 

+/* Utils */
+long long ustime(void);
+

 /* networking.c -- Networking and Client related operations */
 redisClient *createClient(int fd);
 void closeTimedoutClients(void);
 /* networking.c -- Networking and Client related operations */
 redisClient *createClient(int fd);
 void closeTimedoutClients(void);