slave now detect lost connection during SYNC, fixing Issue 173

[redis.git] / src / networking.c

diff --git a/src/networking.c b/src/networking.c
index 31844a09f359bc56193d7a14798405848f105be6..a39be7c4ea8b81ead00a145cdc7a2dcec516d9c7 100644 (file)
--- a/src/networking.c
+++ b/src/networking.c
@@ -235,22 +235,28 @@ void freeClient(redisClient *c) {
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);
     ln = listSearchKey(server.clients,c);
     redisAssert(ln != NULL);
     listDelNode(server.clients,ln);

-    /* Remove from the list of clients that are now ready to be restarted
-     * after waiting for swapped keys */
-    if (c->flags & REDIS_IO_WAIT && listLength(c->io_keys) == 0) {
-        ln = listSearchKey(server.io_ready_clients,c);
-        if (ln) {
+    /* Remove from the list of clients waiting for swapped keys, or ready
+     * to be restarted, but not yet woken up again. */
+    if (c->flags & REDIS_IO_WAIT) {
+        redisAssert(server.vm_enabled);
+        if (listLength(c->io_keys) == 0) {
+            ln = listSearchKey(server.io_ready_clients,c);
+
+            /* When this client is waiting to be woken up (REDIS_IO_WAIT),
+             * it should be present in the list io_ready_clients */
+            redisAssert(ln != NULL);

             listDelNode(server.io_ready_clients,ln);
             listDelNode(server.io_ready_clients,ln);

-            server.vm_blocked_clients--;
+        } else {
+            while (listLength(c->io_keys)) {
+                ln = listFirst(c->io_keys);
+                dontWaitForSwappedKey(c,ln->value);
+            }

         }
         }

-    }
-    /* Remove from the list of clients waiting for swapped keys */
-    while (server.vm_enabled && listLength(c->io_keys)) {
-        ln = listFirst(c->io_keys);
-        dontWaitForSwappedKey(c,ln->value);
+        server.vm_blocked_clients--;

     }
     listRelease(c->io_keys);
     }
     listRelease(c->io_keys);

-    /* Master/slave cleanup */
+    /* Master/slave cleanup.
+     * Case 1: we lost the connection with a slave. */

     if (c->flags & REDIS_SLAVE) {
         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
             close(c->repldbfd);
     if (c->flags & REDIS_SLAVE) {
         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
             close(c->repldbfd);


@@ -259,9 +265,20 @@ void freeClient(redisClient *c) {
         redisAssert(ln != NULL);
         listDelNode(l,ln);
     }
         redisAssert(ln != NULL);
         listDelNode(l,ln);
     }

+
+    /* Case 2: we lost the connection with the master. */

     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
         server.replstate = REDIS_REPL_CONNECT;
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
         server.replstate = REDIS_REPL_CONNECT;

+        /* Since we lost the connection with the master, we should also
+         * close the connection with all our slaves if we have any, so
+         * when we'll resync with the master the other slaves will sync again
+         * with us as well. Note that also when the slave is not connected
+         * to the master it will keep refusing connections by other slaves. */
+        while (listLength(server.slaves)) {
+            ln = listFirst(server.slaves);
+            freeClient((redisClient*)ln->value);
+        }

     }
     /* Release memory */
     zfree(c->argv);
     }
     /* Release memory */
     zfree(c->argv);


@@ -461,6 +478,7 @@ void closeTimedoutClients(void) {
         if (server.maxidletime &&
             !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
             !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */
         if (server.maxidletime &&
             !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
             !(c->flags & REDIS_MASTER) &&   /* no timeout for masters */

+            !(c->flags & REDIS_BLOCKED) &&  /* no timeout for BLPOP */

             dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
             listLength(c->pubsub_patterns) == 0 &&
             (now - c->lastinteraction > server.maxidletime))
             dictSize(c->pubsub_channels) == 0 && /* no timeout for pubsub */
             listLength(c->pubsub_patterns) == 0 &&
             (now - c->lastinteraction > server.maxidletime))