strictly honour memory limit

diff --git a/src/diskstore.c b/src/diskstore.c
index b1dc3ad4c1574f46ee1c155129a3b5c780957e28..3e0ad406ffaffcff22ad194be50d82a86aa14be3 100644 (file)
--- a/src/diskstore.c
+++ b/src/diskstore.c
@@ -190,7 +190,7 @@ int dsSet(redisDb *db, robj *key, robj *val) {
 
     len = dsKeyToPath(db,buf,key);
     memcpy(buf2,buf,len);
 
     len = dsKeyToPath(db,buf,key);
     memcpy(buf2,buf,len);

-    snprintf(buf2+len,sizeof(buf2)-len,"%ld.%ld",(long)time(NULL),(long)val);
+    snprintf(buf2+len,sizeof(buf2)-len,"_%ld_%ld",(long)time(NULL),(long)val);

     fp = fopen(buf2,"w");
     if ((retval = rdbSaveKeyValuePair(fp,db,key,val,time(NULL))) == -1)
         return REDIS_ERR;
     fp = fopen(buf2,"w");
     if ((retval = rdbSaveKeyValuePair(fp,db,key,val,time(NULL))) == -1)
         return REDIS_ERR;

diff --git a/src/dscache.c b/src/dscache.c
index 92f529418359fe767eb9639872c30d4b155c24d6..901c70f07e4f44502f00b6426844c7b9afbcd1e9 100644 (file)
--- a/src/dscache.c
+++ b/src/dscache.c
@@ -107,6 +107,7 @@
  */
 
 void spawnIOThread(void);
  */
 
 void spawnIOThread(void);

+int cacheScheduleIOPushJobs(int onlyloads);

 
 /* =================== Virtual Memory - Blocking Side  ====================== */
 
 
 /* =================== Virtual Memory - Blocking Side  ====================== */
 


@@ -210,10 +211,17 @@ int cacheFreeOneEntry(void) {
         }
     }
     if (best == NULL) {
         }
     }
     if (best == NULL) {

-        /* FIXME: If there are objects that are in the write queue
-         * so we can't delete them we should block here, at the cost of
-         * slowness as the object cache memory limit is considered 
-         * n hard limit. */
+        /* Was not able to fix 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 */
+        if (listLength(server.cache_io_queue) > 0) {
+            cacheScheduleIOPushJobs(0);
+            waitEmptyIOJobsQueue();
+            processAllPendingIOJobs();
+            return REDIS_OK;
+        }
+        /* Nothing to free at all... */

         return REDIS_ERR;
     }
     key = dictGetEntryKey(best);
         return REDIS_ERR;
     }
     key = dictGetEntryKey(best);


@@ -597,8 +605,6 @@ void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {
 #define REDIS_IO_LOADINPROG 4
 #define REDIS_IO_SAVEINPROG 8
 
 #define REDIS_IO_LOADINPROG 4
 #define REDIS_IO_SAVEINPROG 8
 

-void cacheScheduleIOPushJobs(int onlyloads);
-

 void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) {
     struct dictEntry *de = dictFind(db->io_queued,key);
 
 void cacheScheduleIOAddFlag(redisDb *db, robj *key, long flag) {
     struct dictEntry *de = dictFind(db->io_queued,key);
 


@@ -675,10 +681,10 @@ void cacheScheduleIO(redisDb *db, robj *key, int type) {
  * (not protected by lookupKey() that will block on keys in IO_SAVEINPROG
  * state. */
 #define MAX_IO_JOBS_QUEUE 100
  * (not protected by lookupKey() that will block on keys in IO_SAVEINPROG
  * state. */
 #define MAX_IO_JOBS_QUEUE 100

-void cacheScheduleIOPushJobs(int onlyloads) {
+int cacheScheduleIOPushJobs(int onlyloads) {

     time_t now = time(NULL);
     listNode *ln;
     time_t now = time(NULL);
     listNode *ln;

-    int jobs, topush = 0;
+    int jobs, topush = 0, pushed = 0;

 
     /* Sync stuff on disk, but only if we have less
      * than MAX_IO_JOBS_QUEUE IO jobs. */
 
     /* Sync stuff on disk, but only if we have less
      * than MAX_IO_JOBS_QUEUE IO jobs. */


@@ -749,10 +755,12 @@ void cacheScheduleIOPushJobs(int onlyloads) {
             listDelNode(server.cache_io_queue,ln);
             decrRefCount(op->key);
             zfree(op);
             listDelNode(server.cache_io_queue,ln);
             decrRefCount(op->key);
             zfree(op);

+            pushed++;

         } else {
             break; /* too early */
         }
     }
         } else {
             break; /* too early */
         }
     }

+    return pushed;

 }
 
 void cacheCron(void) {
 }
 
 void cacheCron(void) {