disk store logged messages improved

[redis.git] / src / dscache.c

diff --git a/src/dscache.c b/src/dscache.c
index e2fd7e30224dfc08a89bf10ce81186085b924058..7273f7ffd6decf7fa58cc39daf966e24beb4b4e0 100644 (file)
--- a/src/dscache.c
+++ b/src/dscache.c
@@ -117,7 +117,7 @@ void dsInit(void) {
 
     zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
 
 
     zmalloc_enable_thread_safeness(); /* we need thread safe zmalloc() */
 

-    redisLog(REDIS_NOTICE,"Initializing Disk Store at %s", server.ds_path);
+    redisLog(REDIS_NOTICE,"Opening Disk Store: %s", server.ds_path);

     /* Open Disk Store */
     if (dsOpen() != REDIS_OK) {
         redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");
     /* Open Disk Store */
     if (dsOpen() != REDIS_OK) {
         redisLog(REDIS_WARNING,"Fatal error opening disk store. Exiting.");


@@ -227,6 +227,7 @@ int cacheFreeOneEntry(void) {
         dbDelete(best_db,kobj);
         decrRefCount(kobj);
     }
         dbDelete(best_db,kobj);
         decrRefCount(kobj);
     }

+    return REDIS_OK;

 }
 
 /* Return true if it's safe to swap out objects in a given moment.
 }
 
 /* Return true if it's safe to swap out objects in a given moment.


@@ -240,6 +241,8 @@ int dsCanTouchDiskStore(void) {
 
 void freeIOJob(iojob *j) {
     decrRefCount(j->key);
 
 void freeIOJob(iojob *j) {
     decrRefCount(j->key);

+    /* j->val can be NULL if the job is about deleting the key from disk. */
+    if (j->val) decrRefCount(j->val);

     zfree(j);
 }
 
     zfree(j);
 }
 


@@ -278,72 +281,21 @@ void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata,
 
         /* Post process it in the main thread, as there are things we
          * can do just here to avoid race conditions and/or invasive locks */
 
         /* Post process it in the main thread, as there are things we
          * can do just here to avoid race conditions and/or invasive locks */

-        redisLog(REDIS_DEBUG,"COMPLETED Job type: %d, ID %p, key: %s", j->type, (void*)j->id, (unsigned char*)j->key->ptr);
+        redisLog(REDIS_DEBUG,"COMPLETED Job type %s, key: %s",
+            (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
+            (unsigned char*)j->key->ptr);

         de = dictFind(j->db->dict,j->key->ptr);
         redisAssert(de != NULL);
         if (j->type == REDIS_IOJOB_LOAD) {
         de = dictFind(j->db->dict,j->key->ptr);
         redisAssert(de != NULL);
         if (j->type == REDIS_IOJOB_LOAD) {

-            redisDb *db;
-            vmpointer *vp = dictGetEntryVal(de);
-
-            /* Key loaded, bring it at home */
-            vmMarkPagesFree(vp->page,vp->usedpages);
-            redisLog(REDIS_DEBUG, "VM: object %s loaded from disk (threaded)",
-                (unsigned char*) j->key->ptr);
-            server.vm_stats_swapped_objects--;
-            server.vm_stats_swapins++;
-            dictGetEntryVal(de) = j->val;
-            incrRefCount(j->val);
-            db = j->db;
+            /* Create the key-value pair in the in-memory database */
+            dbAdd(j->db,j->key,j->val);

             /* Handle clients waiting for this key to be loaded. */
             /* Handle clients waiting for this key to be loaded. */

-            handleClientsBlockedOnSwappedKey(db,j->key);
+            handleClientsBlockedOnSwappedKey(j->db,j->key);

             freeIOJob(j);
             freeIOJob(j);

-            zfree(vp);
-        } else if (j->type == REDIS_IOJOB_DO_SWAP) {
-            vmpointer *vp;
-
-            /* Key swapped. We can finally free some memory. */
-            if (j->val->storage != REDIS_VM_SWAPPING) {
-                vmpointer *vp = (vmpointer*) j->id;
-                printf("storage: %d\n",vp->storage);
-                printf("key->name: %s\n",(char*)j->key->ptr);
-                printf("val: %p\n",(void*)j->val);
-                printf("val->type: %d\n",j->val->type);
-                printf("val->ptr: %s\n",(char*)j->val->ptr);
-            }
-            redisAssert(j->val->storage == REDIS_VM_SWAPPING);
-            vp = createVmPointer(j->val->type);
-            vp->page = j->page;
-            vp->usedpages = j->pages;
-            dictGetEntryVal(de) = vp;
-            /* Fix the storage otherwise decrRefCount will attempt to
-             * remove the associated I/O job */
-            j->val->storage = REDIS_VM_MEMORY;
-            decrRefCount(j->val);
-            redisLog(REDIS_DEBUG,
-                "VM: object %s swapped out at %lld (%lld pages) (threaded)",
-                (unsigned char*) j->key->ptr,
-                (unsigned long long) j->page, (unsigned long long) j->pages);
-            server.vm_stats_swapped_objects++;
-            server.vm_stats_swapouts++;
+        } else if (j->type == REDIS_IOJOB_SAVE) {
+            redisAssert(j->val->storage == REDIS_DS_SAVING);
+            j->val->storage = REDIS_DS_MEMORY;

             freeIOJob(j);
             freeIOJob(j);

-            /* Put a few more swap requests in queue if we are still
-             * out of memory */
-            if (trytoswap && vmCanSwapOut() &&
-                zmalloc_used_memory() > server.vm_max_memory)
-            {
-                int more = 1;
-                while(more) {
-                    lockThreadedIO();
-                    more = listLength(server.io_newjobs) <
-                            (unsigned) server.vm_max_threads;
-                    unlockThreadedIO();
-                    /* Don't waste CPU time if swappable objects are rare. */
-                    if (vmSwapOneObjectThreaded() == REDIS_ERR) {
-                        trytoswap = 0;
-                        break;
-                    }
-                }
-            }

         }
         processed++;
         if (processed == toprocess) return;
         }
         processed++;
         if (processed == toprocess) return;


@@ -384,22 +336,24 @@ void *IOThreadEntryPoint(void *arg) {
         j = ln->value;
         listDelNode(server.io_newjobs,ln);
         /* Add the job in the processing queue */
         j = ln->value;
         listDelNode(server.io_newjobs,ln);
         /* Add the job in the processing queue */

-        j->thread = pthread_self();

         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 %ld got a new job (type %d): %p about key '%s'",
-            (long) pthread_self(), j->type, (void*)j, (char*)j->key->ptr);
+        redisLog(REDIS_DEBUG,"Thread %ld: new job type %s: %p about key '%s'",
+            (long) pthread_self(),
+            (j->type == REDIS_IOJOB_LOAD) ? "load" : "save",
+            (void*)j, (char*)j->key->ptr);

 
         /* Process the Job */
         if (j->type == REDIS_IOJOB_LOAD) {
 
         /* Process the Job */
         if (j->type == REDIS_IOJOB_LOAD) {

-            vmpointer *vp = (vmpointer*)j->id;
-            j->val = vmReadObjectFromSwap(j->page,vp->vtype);
-        } else if (j->type == REDIS_IOJOB_PREPARE_SWAP) {
-            j->pages = rdbSavedObjectPages(j->val);
-        } else if (j->type == REDIS_IOJOB_DO_SWAP) {
-            if (vmWriteObjectOnSwap(j->val,j->page) == REDIS_ERR)
-                j->canceled = 1;
+            j->val = dsGet(j->db,j->key);
+            redisAssert(j->val != NULL);
+        } else if (j->type == REDIS_IOJOB_SAVE) {
+            redisAssert(j->val->storage == REDIS_DS_SAVING);
+            if (j->val)
+                dsSet(j->db,j->key,j->val);
+            else
+                dsDel(j->db,j->key);

         }
 
         /* Done: insert the job into the processed queue */
         }
 
         /* Done: insert the job into the processed queue */


@@ -474,52 +428,114 @@ void queueIOJob(iojob *j) {
         spawnIOThread();
 }
 
         spawnIOThread();
 }
 

-int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db) {
+void dsCreateIOJob(int type, redisDb *db, robj *key, robj *val) {

     iojob *j;
 
     j = zmalloc(sizeof(*j));
     iojob *j;
 
     j = zmalloc(sizeof(*j));

-    j->type = REDIS_IOJOB_PREPARE_SWAP;
+    j->type = type;

     j->db = db;
     j->key = key;
     incrRefCount(key);
     j->db = db;
     j->key = key;
     incrRefCount(key);

-    j->id = j->val = val;
+    j->val = val;

     incrRefCount(val);
     incrRefCount(val);

-    j->canceled = 0;
-    j->thread = (pthread_t) -1;
-    val->storage = REDIS_VM_SWAPPING;

 
     lockThreadedIO();
     queueIOJob(j);
     unlockThreadedIO();
 
     lockThreadedIO();
     queueIOJob(j);
     unlockThreadedIO();

-    return REDIS_OK;
+}
+
+void cacheScheduleForFlush(redisDb *db, robj *key) {
+    dirtykey *dk;
+    dictEntry *de;
+    
+    de = dictFind(db->dict,key->ptr);
+    if (de) {
+        robj *val = dictGetEntryVal(de);
+        if (val->storage == REDIS_DS_DIRTY)
+            return;
+        else
+            val->storage = REDIS_DS_DIRTY;
+    }
+
+    redisLog(REDIS_DEBUG,"Scheduling key %s for saving",key->ptr);
+    dk = zmalloc(sizeof(*dk));
+    dk->db = db;
+    dk->key = key;
+    incrRefCount(key);
+    dk->ctime = time(NULL);
+    listAddNodeTail(server.cache_flush_queue, key);
+}
+
+void cacheCron(void) {
+    time_t now = time(NULL);
+    listNode *ln;
+
+    /* Sync stuff on disk */
+    while((ln = listFirst(server.cache_flush_queue)) != NULL) {
+        dirtykey *dk = ln->value;
+
+        if ((now - dk->ctime) >= server.cache_flush_delay) {
+            struct dictEntry *de;
+            robj *val;
+
+            redisLog(REDIS_DEBUG,"Creating IO Job to save key %s",dk->key->ptr);
+
+            /* Lookup the key. We need to check if it's still here and
+             * possibly access to the value. */
+            de = dictFind(dk->db->dict,dk->key->ptr);
+            if (de) {
+                val = dictGetEntryVal(de);
+                redisAssert(val->storage == REDIS_DS_DIRTY);
+                val->storage = REDIS_DS_SAVING;
+            } else {
+                /* Setting the value to NULL tells the IO thread to delete
+                 * the key on disk. */
+                val = NULL;
+            }
+            dsCreateIOJob(REDIS_IOJOB_SAVE,dk->db,dk->key,val);
+            listDelNode(server.cache_flush_queue,ln);
+        } else {
+            break; /* too early */
+        }
+    }
+
+    /* Reclaim memory from the object cache */
+    while (server.ds_enabled && zmalloc_used_memory() >
+            server.cache_max_memory)
+    {
+        if (cacheFreeOneEntry() == REDIS_ERR) break;
+    }

 }
 
 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
 
 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.
 }
 
 /* ============ Virtual Memory - Blocking clients on missing keys =========== */
 
 /* This function makes the clinet 'c' waiting for the key 'key' to be loaded.

- * If there is not already a job loading the key, it is craeted.
- * The key is added to the io_keys list in the client structure, and also
+ * If the key is already in memory we don't need to block, regardless
+ * of the storage of the value object for this key:
+ *
+ * - If it's REDIS_DS_MEMORY we have the key in memory.
+ * - If it's REDIS_DS_DIRTY they key was modified, but still in memory.
+ * - if it's REDIS_DS_SAVING the key is being saved by an IO Job. When
+ *   the client will lookup the key it will block if the key is still
+ *   in this stage but it's more or less the best we can do.
+ *
+ *   FIXME: we should try if it's actually better to suspend the client
+ *   accessing an object that is being saved, and awake it only when
+ *   the saving was completed.
+ *
+ * Otherwise if the key is not in memory, we block the client and start
+ * an IO Job to load it:
+ *
+ * the key is added to the io_keys list in the client structure, and also

  * in the hash table mapping swapped keys to waiting clients, that is,
  * server.io_waited_keys. */
 int waitForSwappedKey(redisClient *c, robj *key) {
     struct dictEntry *de;
  * in the hash table mapping swapped keys to waiting clients, that is,
  * server.io_waited_keys. */
 int waitForSwappedKey(redisClient *c, robj *key) {
     struct dictEntry *de;

-    robj *o;

     list *l;
 
     list *l;
 

-    /* If the key does not exist or is already in RAM we don't need to
-     * block the client at all. */
+    /* Return ASAP if the key is in memory */

     de = dictFind(c->db->dict,key->ptr);
     de = dictFind(c->db->dict,key->ptr);

-    if (de == NULL) return 0;
-    o = dictGetEntryVal(de);
-    if (o->storage == REDIS_VM_MEMORY) {
-        return 0;
-    } else if (o->storage == REDIS_VM_SWAPPING) {
-        /* We were swapping the key, undo it! */
-        vmCancelThreadedIOJob(o);
-        return 0;
-    }
-
-    /* OK: the key is either swapped, or being loaded just now. */
+    if (de != NULL) return 0;

 
     /* Add the key to the list of keys this client is waiting for.
      * This maps clients to keys they are waiting for. */
 
     /* Add the key to the list of keys this client is waiting for.
      * This maps clients to keys they are waiting for. */


@@ -542,25 +558,8 @@ int waitForSwappedKey(redisClient *c, robj *key) {
     listAddNodeTail(l,c);
 
     /* Are we already loading the key from disk? If not create a job */
     listAddNodeTail(l,c);
 
     /* Are we already loading the key from disk? If not create a job */

-    if (o->storage == REDIS_VM_SWAPPED) {
-        iojob *j;
-        vmpointer *vp = (vmpointer*)o;
-
-        o->storage = REDIS_VM_LOADING;
-        j = zmalloc(sizeof(*j));
-        j->type = REDIS_IOJOB_LOAD;
-        j->db = c->db;
-        j->id = (robj*)vp;
-        j->key = key;
-        incrRefCount(key);
-        j->page = vp->page;
-        j->val = NULL;
-        j->canceled = 0;
-        j->thread = (pthread_t) -1;
-        lockThreadedIO();
-        queueIOJob(j);
-        unlockThreadedIO();
-    }
+    if (de == NULL)
+        dsCreateIOJob(REDIS_IOJOB_LOAD,c->db,key,NULL);

     return 1;
 }
 
     return 1;
 }
 


@@ -638,7 +637,7 @@ int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd) {
     if (listLength(c->io_keys)) {
         c->flags |= REDIS_IO_WAIT;
         aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
     if (listLength(c->io_keys)) {
         c->flags |= REDIS_IO_WAIT;
         aeDeleteFileEvent(server.el,c->fd,AE_READABLE);

-        server.vm_blocked_clients++;
+        server.cache_blocked_clients++;

         return 1;
     } else {
         return 0;
         return 1;
     } else {
         return 0;