Avoid allocating and releasing list node in reply copy avoidance (#14739)

Optimizes handling of clients with referenced replies by embedding the
`pending_ref_reply_node` list node in `client` and avoiding
per-operation node alloc/free.

there is an improvement: ~2% on 4 and 16 io-threads. ~1% on 8 io-threads

src/networking.c

@@ -239,7 +239,7 @@ client *createClient(connection *conn) {
     c->auth_callback_privdata = NULL;
     c->auth_module = NULL;
     listInitNode(&c->clients_pending_write_node, c);
-    c->pending_ref_reply_node = NULL;
+    listInitNode(&c->pending_ref_reply_node, c);
     c->mem_usage_bucket = NULL;
     c->mem_usage_bucket_node = NULL;
     c->net_input_bytes_curr_cmd = 0;
@@ -519,6 +519,15 @@ void _addReplyToBufferOrList(client *c, const char *s, size_t len) {
         _addReplyPayloadToList(c, c->reply, s + reply_len, len - reply_len, PLAIN_REPLY);
 }
 
+/* Check if the client's pending_ref_reply_node is currently linked in the list.
+ * A node is considered linked if it has neighbors (prev/next), or if it's the
+ * only node in the list (head points to it). */
+static inline int clientIsInPendingRefReplyList(client *c) {
+    return listNextNode(&c->pending_ref_reply_node) != NULL ||
+           listPrevNode(&c->pending_ref_reply_node) != NULL ||
+           listFirst(server.clients_with_pending_ref_reply) == &c->pending_ref_reply_node;
+}
+
 /* Increment reference to object and add pointer to object and
  * pointer to string itself to current reply buffer */
 static void _addBulkStrRefToBufferOrList(client *c, robj *obj, size_t len) {
@@ -549,9 +558,8 @@ static void _addBulkStrRefToBufferOrList(client *c, robj *obj, size_t len) {
     }
 
     /* Track clients with pending referenced reply objects for async flushdb protection. */
-    if (c->pending_ref_reply_node == NULL) {
-        listAddNodeTail(server.clients_with_pending_ref_reply, c);
-        c->pending_ref_reply_node = listLast(server.clients_with_pending_ref_reply);
+    if (!clientIsInPendingRefReplyList(c)) {
+        listLinkNodeTail(server.clients_with_pending_ref_reply, &c->pending_ref_reply_node);
     }
 }
 
@@ -1903,9 +1911,8 @@ void unlinkClient(client *c) {
  * This should only be used when we are certain that the replies no longer
  * contain any referenced robj. */
 void tryUnlinkClientFromPendingRefReply(client *c, int force) {
-    if (c->pending_ref_reply_node && (force || !clientHasPendingReplies(c))) {
-        listDelNode(server.clients_with_pending_ref_reply, c->pending_ref_reply_node);
-        c->pending_ref_reply_node = NULL;
+    if (clientIsInPendingRefReplyList(c) && (force || !clientHasPendingReplies(c))) {
+        listUnlinkNode(server.clients_with_pending_ref_reply, &c->pending_ref_reply_node);
     }
 }
 

src/server.h

@@ -1577,7 +1577,7 @@ typedef struct client {
     /* list node in clients_pending_write list */
     listNode clients_pending_write_node;
     /* list node in clients_with_pending_ref_reply list */
-    listNode *pending_ref_reply_node;
+    listNode pending_ref_reply_node;
     /* Statistics and metrics */
     size_t net_input_bytes_curr_cmd; /* Total network input bytes read for the
                                       * execution of this client's current command. */