Fix RDMA re-entrancy assertion and lost wakeup deadlocks with I/O threads

src/memory_prefetch.c

@@ -9,8 +9,10 @@
  */
 
 #include "memory_prefetch.h"
+#include "adlist.h"
 #include "server.h"
 #include "io_threads.h"
+#include <stdlib.h>
 
 typedef enum {
     PREFETCH_ENTRY, /* Initial state, prefetch entries associated with the given key's hash */
@@ -214,7 +216,7 @@ static void prefetchCommands(void) {
 }
 
 /* Processes all the prefetched commands in the current batch. */
-void processClientsCommandsBatch(void) {
+void processClientsCommandsBatch(list *handled_clients) {
     if (!batch || batch->client_count == 0) return;
 
     /* If executed_commands is not 0,
@@ -231,7 +233,10 @@ void processClientsCommandsBatch(void) {
         /* Set the client to null immediately to avoid accessing it again recursively when ProcessingEventsWhileBlocked */
         batch->clients[i] = NULL;
         batch->executed_commands++;
-        if (processPendingCommandAndInputBuffer(c) != C_ERR) beforeNextClient(c);
+        if (processPendingCommandAndInputBuffer(c) != C_ERR) {
+            beforeNextClient(c);
+            if (handled_clients) listAddNodeTail(handled_clients, c);
+        }
     }
 
     resetCommandsBatch();
@@ -260,7 +265,7 @@ static void addCommandToBatch(struct serverCommand *cmd, robj **argv, int argc,
  * if it becomes full.
  *
  * Returns C_OK if the command was added successfully, C_ERR otherwise. */
-int addCommandToBatchAndProcessIfFull(client *c) {
+int addCommandToBatchAndProcessIfFull(client *c, list *handled_clients) {
     if (!batch) return C_ERR;
 
     batch->clients[batch->client_count++] = c;
@@ -283,7 +288,7 @@ int addCommandToBatchAndProcessIfFull(client *c) {
      * We also check the client count to handle cases where
      * no keys exist for the clients' commands. */
     if (batch->client_count == batch->max_prefetch_size || batch->key_count == batch->max_prefetch_size) {
-        processClientsCommandsBatch();
+        processClientsCommandsBatch(handled_clients);
     }
 
     return C_OK;

src/memory_prefetch.h

@@ -1,11 +1,13 @@
 #ifndef MEMORY_PREFETCH_H
 #define MEMORY_PREFETCH_H
 
+#include "adlist.h"
+
 struct client;
 
 void prefetchCommandsBatchInit(void);
-void processClientsCommandsBatch(void);
-int addCommandToBatchAndProcessIfFull(struct client *c);
+void processClientsCommandsBatch(list *handled_clients);
+int addCommandToBatchAndProcessIfFull(struct client *c, list *handled_clients);
 void removeClientFromPendingCommandsBatch(struct client *c);
 int onMaxBatchSizeChange(const char **err);
 

src/networking.c

@@ -27,6 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include "adlist.h"
 #include "server.h"
 #include "cluster.h"
 #include "cluster_slot_stats.h"
@@ -40,6 +41,7 @@
 #include "module.h"
 #include "connection.h"
 #include "zmalloc.h"
+#include <stdint.h>
 #include <strings.h>
 #include <sys/socket.h>
 #include <sys/uio.h>
@@ -6372,11 +6374,15 @@ int processIOThreadsReadDone(void) {
     if (ProcessingEventsWhileBlocked) {
         /* When ProcessingEventsWhileBlocked we may call processIOThreadsReadDone recursively.
          * In this case, there may be some clients left in the batch waiting to be processed. */
-        processClientsCommandsBatch();
+        processClientsCommandsBatch(NULL);
     }
 
     if (listLength(server.clients_pending_io_read) == 0) return 0;
     int processed = 0;
+    /* Collect clients handled in this iteration to continue parsing any
+     * residual buffered data synchronously after batch execution (important
+     * for edge-triggered transports like RDMA). */
+    list *handled_clients = listCreate();
     listNode *ln;
 
     listNode *next = listFirst(server.clients_pending_io_read);
@@ -6412,7 +6418,6 @@ int processIOThreadsReadDone(void) {
         /* Save the current conn state, as connUpdateState may modify it */
         int in_accept_state = (connGetState(c->conn) == CONN_STATE_ACCEPTING);
         connSetPostponeUpdateState(c->conn, 0);
-        connUpdateState(c->conn);
 
         /* In accept state, no client's data was read - stop here. */
         if (in_accept_state) continue;
@@ -6439,18 +6444,30 @@ int processIOThreadsReadDone(void) {
 
         size_t list_length_before_command_execute = listLength(server.clients_pending_io_read);
         /* try to add the command to the batch */
-        int ret = addCommandToBatchAndProcessIfFull(c);
+        int ret = addCommandToBatchAndProcessIfFull(c, handled_clients);
         /* If the command was not added to the commands batch, process it immediately */
         if (ret == C_ERR) {
-            if (processPendingCommandAndInputBuffer(c) == C_OK) beforeNextClient(c);
+            if (processPendingCommandAndInputBuffer(c) == C_OK) {
+                beforeNextClient(c);
+                listAddNodeTail(handled_clients, c);
+            }
         }
         if (list_length_before_command_execute != listLength(server.clients_pending_io_read)) {
             /* A client was unlink from the list possibly making the next node invalid */
             next = listFirst(server.clients_pending_io_read);
         }
     }
 
-    processClientsCommandsBatch();
+    processClientsCommandsBatch(handled_clients);
+
+    listIter handled_li;
+    listNode *handled_ln;
+    listRewind(handled_clients, &handled_li);
+    while ((handled_ln = listNext(&handled_li))) {
+        client *c = listNodeValue(handled_ln);
+        connUpdateState(c->conn);
+    }
+    listRelease(handled_clients);
 
     return processed;
 }