Add support for asynchronous release to replicaKeysWithExpire on writable replica (#2849)

## Problem
When executing `FLUSHALL ASYNC` on a **writable replica** that has
a large number of expired keys directly written to it, the main thread
gets blocked for an extended period while synchronously releasing the
`replicaKeysWithExpire` dictionary. 

## Root Cause
`FLUSHALL ASYNC` is designed for asynchronous lazy freeing of core data
structures, but the release of `replicaKeysWithExpire` (a dictionary tracking
expired keys on replicas) still happens synchronously in the main thread.
This synchronous operation becomes a bottleneck when dealing with massive
key volumes, as it cannot be offloaded to the lazyfree background thread.

This PR addresses the issue by moving the release of `replicaKeysWithExpire`
to the lazyfree background thread, aligning it with the asynchronous design
of `FLUSHALL ASYNC` and eliminating main thread blocking.

## User scenarios
In some operations, people often need to do primary-replica switches.
One goal is to avoid noticeable impact on the business—like key loss
or reduced availability (e.g., write failures).

Here is the process: First, temporarily switch traffic to writable replicas.
Then we wait for the primary pending replication data to be fully synced
(so primry and replicas are in sync), before finishing the switch. We don't
usually need to do the flush in this case, but it's an optimization that can
be done.

Signed-off-by: Scut-Corgis <[email protected]>
Signed-off-by: jiegang0219 <[email protected]>
Co-authored-by: Binbin <[email protected]>

Author: Scut-Corgis

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

@@ -51,6 +51,7 @@ enum activeExpiryType {
 typedef struct client client;
 typedef struct serverObject robj;
 typedef struct serverDb serverDb;
+typedef struct dict dict;
 
 /* return the relevant expiration policy based on the current server state and the provided flags.
  * FLAGS can indicate either:
@@ -64,8 +65,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,15 @@ 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];
+    size_t len = dictSize(replica_keys_with_expire);
+    dictRelease(replica_keys_with_expire);
+    atomic_fetch_sub_explicit(&lazyfree_objects, len, memory_order_relaxed);
+    atomic_fetch_add_explicit(&lazyfreed_objects, len, memory_order_relaxed);
+}
+
 /* 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 +269,14 @@ 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) {
+        atomic_fetch_add_explicit(&lazyfree_objects, dictSize(replica_keys_with_expire), memory_order_relaxed);
+        bioCreateLazyFreeJob(lazyFreeReplicaKeysWithExpire, 1, replica_keys_with_expire);
+    } else {
+        dictRelease(replica_keys_with_expire);
+    }
+}