Threaded lazyfree WIP #1.

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().");
         }

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

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) {
+            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;

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. */

src/server.h

@@ -1402,9 +1402,14 @@ void slotToKeyAdd(robj *key);
 void slotToKeyDel(robj *key);
 void slotToKeyFlush(void);
 
-/* Lazy free */
+/* Lazy free. Note that SLOW and FAST are only useful when incremental
-#define LAZYFREE_STEP_SLOW 0
+ * lazy free is active. For threaded lazy free the actual freeing of objects
-#define LAZYFREE_STEP_FAST 1
+ * 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);