Implement Lazyfree for replicaKeysWithExpire to Avoid Main Thread Blocking in FLUSHALL ASYNC

src/db.c

@@ -682,7 +682,7 @@ long long emptyData(int dbnum, int flags, void(callback)(hashtable *)) {
     /* Empty the database structure. */
     removed = emptyDbStructure(server.db, dbnum, async, callback);
 
-    if (dbnum == -1) flushReplicaKeysWithExpireList();
+    if (dbnum == -1) flushReplicaKeysWithExpireList(async);
 
     if (with_functions) {
         serverAssert(dbnum == -1);
@@ -1796,7 +1796,7 @@ void swapMainDbWithTempDb(serverDb **tempDb) {
     }
 
     trackingInvalidateKeysOnFlush(1);
-    flushReplicaKeysWithExpireList();
+    flushReplicaKeysWithExpireList(1);
 }
 
 /* SWAPDB db1 db2 */

src/expire.c

@@ -644,9 +644,13 @@ size_t getReplicaKeyWithExpireCount(void) {
  * but it is not worth it since anyway race conditions using the same set
  * of key names in a writable replica and in its primary will lead to
  * inconsistencies. This is just a best-effort thing we do. */
-void flushReplicaKeysWithExpireList(void) {
+void flushReplicaKeysWithExpireList(int async) {
     if (replicaKeysWithExpire) {
-        dictRelease(replicaKeysWithExpire);
+        if (async) {
+            freeReplicaKeysWithExpireAsync(replicaKeysWithExpire);
+        } else {
+            dictRelease(replicaKeysWithExpire);
+        }
         replicaKeysWithExpire = NULL;
     }
 }

src/expire.h

@@ -64,8 +64,9 @@ int convertExpireArgumentToUnixTime(client *c, robj *arg, long long basetime, in
 void activeExpireCycle(int type);
 void expireReplicaKeys(void);
 void rememberReplicaKeyWithExpire(serverDb *db, robj *key);
-void flushReplicaKeysWithExpireList(void);
+void flushReplicaKeysWithExpireList(int async);
 size_t getReplicaKeyWithExpireCount(void);
 bool timestampIsExpired(mstime_t when);
+void freeReplicaKeysWithExpireAsync(dict *replica_keys_with_expire);
 
 #endif

src/lazyfree.c

@@ -85,6 +85,12 @@ void lazyFreeReplicationBacklogRefMem(void *args[]) {
     atomic_fetch_add_explicit(&lazyfreed_objects, len, memory_order_relaxed);
 }
 
+/* Release the replicaKeysWithExpire dict. */
+void lazyFreeReplicaKeysWithExpire(void *args[]) {
+    dict *replica_keys_with_expire = args[0];
+    dictRelease(replica_keys_with_expire);
+}
+
 /* Return the number of currently pending objects to free. */
 size_t lazyfreeGetPendingObjectsCount(void) {
     size_t aux = atomic_load_explicit(&lazyfree_objects, memory_order_relaxed);
@@ -260,3 +266,13 @@ void freeReplicationBacklogRefMemAsync(list *blocks, rax *index) {
         raxFree(index);
     }
 }
+
+
+/* Free replicaKeysWithExpire dict, if the dict is huge enough, free it in async way. */
+void freeReplicaKeysWithExpireAsync(dict *replica_keys_with_expire) {
+    if (dictSize(replica_keys_with_expire) > LAZYFREE_THRESHOLD) {
+        bioCreateLazyFreeJob(lazyFreeReplicaKeysWithExpire,1,replica_keys_with_expire);
+    } else {
+        dictRelease(replica_keys_with_expire);
+    }
+}