Reuse c->argv after command execution to reduce memory allocation overhead (#13521)

inspred by https://github.com/redis/redis/pull/12730

Before this PR, we allocate new memory to store the user command
arguments, however, if the size of the current `c->argv` is larger than
the current command, we can reuse the previously allocated argv to avoid
allocating new memory for the current command.
And we will free `c->argv` in client cron when the client is idle for 2
seconds.

---------

Co-authored-by: Ozan Tezcan <ozantezcan@gmail.com>

src/networking.c

@@ -1441,16 +1441,22 @@ void freeClientOriginalArgv(client *c) {
     c->original_argc = 0;
 }
 
-void freeClientArgv(client *c) {
+static inline void freeClientArgvInternal(client *c, int free_argv) {
     int j;
     for (j = 0; j < c->argc; j++)
         decrRefCount(c->argv[j]);
     c->argc = 0;
     c->cmd = NULL;
     c->argv_len_sum = 0;
-    c->argv_len = 0;
-    zfree(c->argv);
-    c->argv = NULL;
+    if (free_argv) {
+        c->argv_len = 0;
+        zfree(c->argv);
+        c->argv = NULL;
+    }
+}
+
+void freeClientArgv(client *c) {
+    freeClientArgvInternal(c, 1);
 }
 
 /* Close all the slaves connections. This is useful in chained replication
@@ -2152,11 +2158,10 @@ int handleClientsWithPendingWrites(void) {
     return processed;
 }
 
-/* resetClient prepare the client to process the next command */
-void resetClient(client *c) {
+static inline void resetClientInternal(client *c, int free_argv) {
     redisCommandProc *prevcmd = c->cmd ? c->cmd->proc : NULL;
 
-    freeClientArgv(c);
+    freeClientArgvInternal(c, free_argv);
     c->cur_script = NULL;
     c->reqtype = 0;
     c->multibulklen = 0;
@@ -2195,6 +2200,11 @@ void resetClient(client *c) {
     }
 }
 
+/* resetClient prepare the client to process the next command */
+void resetClient(client *c) {
+    resetClientInternal(c, 1);
+}
+
 /* This function is used when we want to re-enter the event loop but there
  * is the risk that the client we are dealing with will be freed in some
  * way. This happens for instance in:
@@ -2292,9 +2302,12 @@ int processInlineBuffer(client *c) {
 
     /* Setup argv array on client structure */
     if (argc) {
-        if (c->argv) zfree(c->argv);
-        c->argv_len = argc;
-        c->argv = zmalloc(sizeof(robj*)*c->argv_len);
+        /* Create new argv if space is insufficient. */
+        if (unlikely(argc > c->argv_len)) {
+            zfree(c->argv);
+            c->argv = zmalloc(sizeof(robj*)*argc);
+            c->argv_len = argc;
+        }
         c->argv_len_sum = 0;
     }
 
@@ -2395,10 +2408,13 @@ int processMultibulkBuffer(client *c) {
 
         c->multibulklen = ll;
 
-        /* Setup argv array on client structure */
-        if (c->argv) zfree(c->argv);
-        c->argv_len = min(c->multibulklen, 1024);
-        c->argv = zmalloc(sizeof(robj*)*c->argv_len);
+        /* Setup argv array on client structure.
+         * Create new argv if space is insufficient or if we need to allocate it gradually. */
+        if (unlikely(c->multibulklen > c->argv_len || c->multibulklen > 1024)) {
+            zfree(c->argv);
+            c->argv_len = min(c->multibulklen, 1024);
+            c->argv = zmalloc(sizeof(robj*)*c->argv_len);
+        }
         c->argv_len_sum = 0;
     }
 
@@ -2530,7 +2546,7 @@ void commandProcessed(client *c) {
     if (c->flags & CLIENT_BLOCKED) return;
 
     reqresAppendResponse(c);
-    resetClient(c);
+    resetClientInternal(c, 0);
 
     long long prev_offset = c->reploff;
     if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
@@ -2661,7 +2677,7 @@ int processInputBuffer(client *c) {
 
         /* Multibulk processing could see a <= 0 length. */
         if (c->argc == 0) {
-            resetClient(c);
+            resetClientInternal(c, 0);
         } else {
             /* If we are in the context of an I/O thread, we can't really
              * execute the command here. All we can do is to flag the client

src/server.c

@@ -784,6 +784,19 @@ int clientsCronResizeQueryBuffer(client *c) {
     return 0;
 }
 
+/* If the client has been idle for too long, free the client's arguments. */
+int clientsCronFreeArgvIfIdle(client *c) {
+    /* If the arguments have already been freed or are still in use, exit ASAP. */
+    if (!c->argv || c->argc) return 0;
+    time_t idletime = server.unixtime - c->lastinteraction;
+    if (idletime > 2) {
+        c->argv_len = 0;
+        zfree(c->argv);
+        c->argv = NULL;
+    }
+    return 0;
+}
+
 /* The client output buffer can be adjusted to better fit the memory requirements.
  *
  * the logic is:
@@ -1050,6 +1063,7 @@ void clientsCron(void) {
          * terminated. */
         if (clientsCronHandleTimeout(c,now)) continue;
         if (clientsCronResizeQueryBuffer(c)) continue;
+        if (clientsCronFreeArgvIfIdle(c)) continue;
         if (clientsCronResizeOutputBuffer(c,now)) continue;
 
         if (clientsCronTrackExpansiveClients(c, curr_peak_mem_usage_slot)) continue;