Merge branch 'unstable' of github.com:/antirez/redis into unstable

2 files changed, 14 insertions, 6 deletions

diff --git a/src/networking.c b/src/networking.c
index c7dd2f23c..0690bbdf6 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -3040,16 +3040,22 @@ int handleClientsWithPendingReadsUsingThreads(void) {
     if (tio_debug) printf("I/O READ All threads finshed\n");
 
     /* Run the list of clients again to process the new buffers. */
-    listRewind(server.clients_pending_read,&li);
-    while((ln = listNext(&li))) {
+    while(listLength(server.clients_pending_read)) {
+        ln = listFirst(server.clients_pending_read);
         client *c = listNodeValue(ln);
         c->flags &= ~CLIENT_PENDING_READ;
+        listDelNode(server.clients_pending_read,ln);
+
         if (c->flags & CLIENT_PENDING_COMMAND) {
-            c->flags &= ~ CLIENT_PENDING_COMMAND;
-            processCommandAndResetClient(c);
+            c->flags &= ~CLIENT_PENDING_COMMAND;
+            if (processCommandAndResetClient(c) == C_ERR) {
+                /* If the client is no longer valid, we avoid
+                 * processing the client later. So we just go
+                 * to the next. */
+                continue;
+            }
         }
         processInputBufferAndReplicate(c);
     }
-    listEmpty(server.clients_pending_read);
     return processed;
 }
diff --git a/src/server.c b/src/server.c
index a6d4b357e..f702da94a 100644
--- a/src/server.c
+++ b/src/server.c
@@ -2088,6 +2088,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 void beforeSleep(struct aeEventLoop *eventLoop) {
     UNUSED(eventLoop);
 
+    /* We should handle pending reads clients ASAP after event loop. */
+    handleClientsWithPendingReadsUsingThreads();
+
     /* Handle TLS pending data. (must be done before flushAppendOnlyFile) */
     tlsProcessPendingData();
     /* If tls still has pending unread data don't sleep at all. */
@@ -2157,7 +2160,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 void afterSleep(struct aeEventLoop *eventLoop) {
     UNUSED(eventLoop);
     if (moduleCount()) moduleAcquireGIL();
-    handleClientsWithPendingReadsUsingThreads();
 }
 
 /* =========================== Server initialization ======================== */