Let IO threads free argv and rewrite objects (#13968)

Some objects that are allocated in the IO thread, we should let IO
thread free them, so we can avoid memory arena contention and also
reduce the load of the main thread.

These objects include:
- client argv objects
- the rewrite objects that are only `OBJ_ENCODING_RAW` encoding strings,
since only the type object is usually allocated by IO threads.

For the implementation, if the client is assigned to IO threads, we will
create a `deferred_objects` array of size 32. We will put objects into
the `deferred_objects` when main thread wants to free above objects,
and finally they are be freed by IO threads.

src/db.c

@@ -496,7 +496,14 @@ static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link
         }
     }
 
-    if (server.lazyfree_lazy_server_del) {
+    if (server.io_threads_num > 1 && old->encoding == OBJ_ENCODING_RAW) {
+        /* In multi-threaded mode, the OBJ_ENCODING_RAW string object usually is
+         * allocated in the IO thread, so we defer the free to the IO thread.
+         * Besides, we never free a string object in BIO threads, so, even with
+         * lazyfree-lazy-server-del enabled, a fallback to main thread freeing
+         * due to defer free failure doesn't go against the config intention. */
+        tryDeferFreeClientObject(server.current_client, old);
+    } else if (server.lazyfree_lazy_server_del) {
         freeObjAsync(key, old, db->id);
     } else {
         decrRefCount(old);

src/iothread.c

@@ -93,6 +93,7 @@ void keepClientInMainThread(client *c) {
     c->io_flags |= CLIENT_IO_READ_ENABLED | CLIENT_IO_WRITE_ENABLED;
     c->running_tid = IOTHREAD_MAIN_THREAD_ID;
     c->tid = IOTHREAD_MAIN_THREAD_ID;
+    freeClientDeferredObjects(c, 1); /* Free deferred objects. */
     /* Main thread starts to manage it. */
     server.io_threads_clients_num[c->tid]++;
 }
@@ -130,6 +131,7 @@ void fetchClientFromIOThread(client *c) {
     /* Now main thread can process it. */
     c->running_tid = IOTHREAD_MAIN_THREAD_ID;
     resumeIOThread(c->tid);
+    freeClientDeferredObjects(c, 1); /* Free deferred objects. */
 }
 
 /* For some clients, we must handle them in the main thread, since there is
@@ -170,6 +172,9 @@ void assignClientToIOThread(client *c) {
     c->running_tid = min_id;
     server.io_threads_clients_num[min_id]++;
 
+    /* The client running in IO thread needs to have deferred objects array. */
+    c->deferred_objects = zmalloc(sizeof(robj*) * CLIENT_MAX_DEFERRED_OBJECTS);
+
     /* Unbind connection of client from main thread event loop, disable read and
      * write, and then put it in the list, main thread will send these clients
      * to IO thread in beforeSleep. */
@@ -542,6 +547,9 @@ int processClientsFromMainThread(IOThread *t) {
         listAddNodeTail(t->clients, c);
         c->io_thread_client_list_node = listLast(t->clients);
 
+        /* The client now is in the IO thread, let's free deferred objects. */
+        freeClientDeferredObjects(c, 0);
+
         /* The client is asked to close, we just let main thread free it. */
         if (c->io_flags & CLIENT_IO_CLOSE_ASAP) {
             enqueuePendingClientsToMainThread(c, 1);

src/networking.c

@@ -164,6 +164,8 @@ client *createClient(connection *conn) {
     c->argv_len_sum = 0;
     c->original_argc = 0;
     c->original_argv = NULL;
+    c->deferred_objects = NULL;
+    c->deferred_objects_num = 0;
     c->cmd = c->lastcmd = c->realcmd = c->iolookedcmd = NULL;
     c->cur_script = NULL;
     c->multibulklen = 0;
@@ -1465,6 +1467,37 @@ void acceptCommonHandler(connection *conn, int flags, char *ip) {
     }
 }
 
+/* Attempt to defer freeing the object to the IO thread. We usually call this since
+ * we know the object is allocated in the IO thread, to avoid memory arena contention,
+ * and also reducing the load of the main thread. */
+void tryDeferFreeClientObject(client *c, robj *o) {
+    if (!c || c->tid == IOTHREAD_MAIN_THREAD_ID || o->refcount > 1) {
+        decrRefCount(o);
+        return;
+    }
+
+    /* Put the object in the deferred objects array. */
+    if (c->deferred_objects && c->deferred_objects_num < CLIENT_MAX_DEFERRED_OBJECTS) {
+        c->deferred_objects[c->deferred_objects_num++] = o;
+    } else {
+        decrRefCount(o);
+    }
+}
+
+/* Free the objects in the deferred_objects array. If free_array is true
+ * then free the array itself as well. */
+void freeClientDeferredObjects(client *c, int free_array) {
+    for (int j = 0; j < c->deferred_objects_num; j++) {
+        decrRefCount(c->deferred_objects[j]);
+    }
+    c->deferred_objects_num = 0;
+
+    if (free_array) {
+        zfree(c->deferred_objects);
+        c->deferred_objects = NULL;
+    }
+}
+
 void freeClientOriginalArgv(client *c) {
     /* We didn't rewrite this client */
     if (!c->original_argv) return;
@@ -1478,8 +1511,13 @@ void freeClientOriginalArgv(client *c) {
 
 static inline void freeClientArgvInternal(client *c, int free_argv) {
     int j;
-    for (j = 0; j < c->argc; j++)
-        decrRefCount(c->argv[j]);
+    if (c->tid == IOTHREAD_MAIN_THREAD_ID) {
+        for (j = 0; j < c->argc; j++)
+            decrRefCount(c->argv[j]);
+    } else {
+        for (j = 0; j < c->argc; j++)
+            tryDeferFreeClientObject(c, c->argv[j]);
+    }
     c->argc = 0;
     c->cmd = NULL;
     c->iolookedcmd = NULL;
@@ -1777,6 +1815,7 @@ void freeClient(client *c) {
     freeReplicaReferencedReplBuffer(c);
     freeClientArgv(c);
     freeClientOriginalArgv(c);
+    freeClientDeferredObjects(c, 1);
     if (c->deferred_reply_errors)
         listRelease(c->deferred_reply_errors);
 #ifdef LOG_REQ_RES

src/server.h

@@ -431,6 +431,9 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
 /* Any flag that does not let optimize FLUSH SYNC to run it in bg as blocking client ASYNC */
 #define CLIENT_AVOID_BLOCKING_ASYNC_FLUSH (CLIENT_DENY_BLOCKING|CLIENT_MULTI|CLIENT_LUA_DEBUG|CLIENT_LUA_DEBUG_SYNC|CLIENT_MODULE)
 
+/* Max deferred objects to be freed by IO thread for each client. */
+#define CLIENT_MAX_DEFERRED_OBJECTS 32
+
 /* Client flags for client IO */
 #define CLIENT_IO_READ_ENABLED (1ULL<<0) /* Client can read from socket. */
 #define CLIENT_IO_WRITE_ENABLED (1ULL<<1) /* Client can write to socket. */
@@ -1335,6 +1338,8 @@ typedef struct client {
     int original_argc;      /* Num of arguments of original command if arguments were rewritten. */
     robj **original_argv;   /* Arguments of original command if arguments were rewritten. */
     size_t argv_len_sum;    /* Sum of lengths of objects in argv list. */
+    robj **deferred_objects;    /* Array of deferred objects to free. */
+    int deferred_objects_num;   /* Number of deferred objects to free. */
     struct redisCommand *cmd, *lastcmd;  /* Last command executed. */
     struct redisCommand *iolookedcmd;    /* Command looked up in IO threads. */
     struct redisCommand *realcmd; /* The original command that was executed by the client,
@@ -2795,6 +2800,8 @@ void clearClientConnectionState(client *c);
 void resetClient(client *c);
 void freeClientOriginalArgv(client *c);
 void freeClientArgv(client *c);
+void tryDeferFreeClientObject(client *c, robj *o);
+void freeClientDeferredObjects(client *c, int free_array);
 void sendReplyToClient(connection *conn);
 void *addReplyDeferredLen(client *c);
 void setDeferredArrayLen(client *c, void *node, long length);