From 9253d8507320dd1d7665a55e11e2cba3ae91c78d Mon Sep 17 00:00:00 2001
From: antirez <antirez@gmail.com>
Date: Wed, 23 Sep 2015 16:46:36 +0200
Subject: [PATCH] Threaded lazyfree WIP #1.

---
 src/bio.c      |  2 ++
 src/bio.h      |  3 ++-
 src/lazyfree.c | 26 +++++++++++++++++++++++---
 src/server.c   |  2 +-
 src/server.h   | 11 ++++++++---
 5 files changed, 36 insertions(+), 8 deletions(-)

diff --git a/src/bio.c b/src/bio.c
index 135b8f3b..d0f74a29 100644
--- a/src/bio.c
+++ b/src/bio.c
@@ -185,6 +185,8 @@ void *bioProcessBackgroundJobs(void *arg) {
             close((long)job->arg1);
         } else if (type == BIO_AOF_FSYNC) {
             aof_fsync((long)job->arg1);
+        } else if (type == BIO_LAZY_FREE) {
+            decrRefCount((robj*)job->arg1);
         } else {
             serverPanic("Wrong job type in bioProcessBackgroundJobs().");
         }
diff --git a/src/bio.h b/src/bio.h
index eb3a2f5d..4b15d1c4 100644
--- a/src/bio.h
+++ b/src/bio.h
@@ -38,4 +38,5 @@ void bioKillThreads(void);
 /* Background job opcodes */
 #define BIO_CLOSE_FILE    0 /* Deferred close(2) syscall. */
 #define BIO_AOF_FSYNC     1 /* Deferred AOF fsync. */
-#define BIO_NUM_OPS       2
+#define BIO_LAZY_FREE     2 /* Deferred objects freeing. */
+#define BIO_NUM_OPS       3
diff --git a/src/lazyfree.c b/src/lazyfree.c
index 84d4b375..27f9a56e 100644
--- a/src/lazyfree.c
+++ b/src/lazyfree.c
@@ -1,4 +1,7 @@
 #include "server.h"
+#include "bio.h"
+
+static int lazyfree_threaded = 1; /* Use a thread to reclaim objects. */
 
 /* Initialization of the lazy free engine. Must be called only once at server
  * startup. */
@@ -97,7 +100,17 @@ size_t lazyfreeFastStep(void) {
 
 /* Handles slow or fast collection steps. */
 size_t lazyfreeStep(int type) {
-    if (type == LAZYFREE_STEP_FAST) return lazyfreeFastStep();
+    /* Threaded implementaiton: only block for STEP_OOM. */
+    if (lazyfree_threaded) {
+        if (type == LAZYFREE_STEP_OOM)
+            return bioWaitStepOfType(BIO_LAZY_FREE);
+        return 0;
+    }
+
+    /* Non threaded implementation: free things incrementally avoiding
+     * to block. */
+    if (type == LAZYFREE_STEP_FAST ||
+        type == LAZYFREE_STEP_OOM) return lazyfreeFastStep();
 
     size_t totalwork = 0;
     mstime_t end = mstime()+2;
@@ -130,8 +143,12 @@ int dbAsyncDelete(redisDb *db, robj *key) {
         /* If releasing the object is too much work, let's put it into the
          * lazy free list. */
         if (free_effort > LAZYFREE_THRESHOLD) {
-            listAddNodeTail(server.lazyfree_obj,val);
-            server.lazyfree_elements += free_effort;
+            if (lazyfree_threaded) {
+                bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
+            } else {
+                listAddNodeTail(server.lazyfree_obj,val);
+                server.lazyfree_elements += free_effort;
+            }
             dictSetVal(db->dict,de,NULL);
         }
     }
@@ -165,6 +182,9 @@ int lazyfreeCron(struct aeEventLoop *eventLoop, long long id, void *clientData)
     UNUSED(id);
     UNUSED(clientData);
 
+    /* Threaded lazy free does not need a timer, unregister the timer event. */
+    if (lazyfree_threaded) return AE_NOMORE;
+
     static size_t prev_mem;
     static int timer_period = 1000; /* Defauls to 1HZ */
     static double mem_trend = 0;
diff --git a/src/server.c b/src/server.c
index 98c915a3..c5ab43b7 100644
--- a/src/server.c
+++ b/src/server.c
@@ -3330,7 +3330,7 @@ int freeMemoryIfNeeded(void) {
         latencyStartMonitor(eviction_latency);
         while (mem_freed < mem_tofree) {
             delta = (long long) zmalloc_used_memory();
-            size_t workdone = lazyfreeStep(LAZYFREE_STEP_FAST);
+            size_t workdone = lazyfreeStep(LAZYFREE_STEP_OOM);
             delta -= (long long) zmalloc_used_memory();
             mem_freed += delta;
             if (!workdone) break; /* Lazy free list is empty. */
diff --git a/src/server.h b/src/server.h
index 3bc3a22e..1d97d936 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1402,9 +1402,14 @@ void slotToKeyAdd(robj *key);
 void slotToKeyDel(robj *key);
 void slotToKeyFlush(void);
 
-/* Lazy free */
-#define LAZYFREE_STEP_SLOW 0
-#define LAZYFREE_STEP_FAST 1
+/* Lazy free. Note that SLOW and FAST are only useful when incremental
+ * lazy free is active. For threaded lazy free the actual freeing of objects
+ * happens in the background. Only STEP_OOM is used since it blocks waiting
+ * for the freeing thread to do some work before returning. */
+#define LAZYFREE_STEP_SLOW 0    /* Take 1-2 milliseconds to reclaim memory. */
+#define LAZYFREE_STEP_FAST 1    /* Free a few elements ASAP and return. */
+#define LAZYFREE_STEP_OOM 2     /* Free a few elements at any cost if there
+                                   is something to free: we are out of memory */
 int dbAsyncDelete(redisDb *db, robj *key);
 void initLazyfreeEngine(void);
 size_t lazyfreeStep(int type);