Fixed issue #435 and at the same time introduced explicit ping in the...

Fixed issue #435 and at the same time introduced explicit ping in the master-slave channel that will detect a blocked master or a broken even if apparently connected TCP link.

src/networking.c

@@ -497,7 +497,6 @@ void freeClient(redisClient *c) {
     /* Case 2: we lost the connection with the master. */
     if (c->flags & REDIS_MASTER) {
         server.master = NULL;
-        /* FIXME */
         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

src/redis.c

@@ -516,7 +516,7 @@ void updateLRUClock(void) {
 }
 
 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
-    int j, loops = server.cronloops++;
+    int j, loops = server.cronloops;
     REDIS_NOTUSED(eventLoop);
     REDIS_NOTUSED(id);
     REDIS_NOTUSED(clientData);
@@ -646,6 +646,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
      * to detect transfer failures. */
     if (!(loops % 10)) replicationCron();
 
+    server.cronloops++;
     return 100;
 }
 

src/replication.c

@@ -328,6 +328,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
                 buf+1);
             replicationAbortSyncTransfer();
             return;
+        } else if (buf[0] == '\0') {
+            /* At this stage just a newline works as a PING in order to take
+             * the connection live. So we refresh our last interaction
+             * timestamp. */
+            server.repl_transfer_lastio = time(NULL);
+            return;
         } else if (buf[0] != '$') {
             redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
             replicationAbortSyncTransfer();
@@ -488,17 +494,26 @@ void slaveofCommand(redisClient *c) {
 
 /* --------------------------- REPLICATION CRON  ---------------------------- */
 
-#define REDIS_REPL_TRANSFER_TIMEOUT 60
+#define REDIS_REPL_TIMEOUT 60
+#define REDIS_REPL_PING_SLAVE_PERIOD 10
 
 void replicationCron(void) {
     /* Bulk transfer I/O timeout? */
     if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
-        (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TRANSFER_TIMEOUT)
+        (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TIMEOUT)
     {
         redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
         replicationAbortSyncTransfer();
     }
 
+    /* Timed out master when we are an already connected slave? */
+    if (server.masterhost && server.replstate == REDIS_REPL_CONNECTED &&
+        (time(NULL)-server.master->lastinteraction) > REDIS_REPL_TIMEOUT)
+    {
+        redisLog(REDIS_WARNING,"MASTER time out: no data nor PING received...");
+        freeClient(server.master);
+    }
+
     /* Check if we should connect to a MASTER */
     if (server.replstate == REDIS_REPL_CONNECT) {
         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
@@ -507,4 +522,33 @@ void replicationCron(void) {
             if (server.appendonly) rewriteAppendOnlyFileBackground();
         }
     }
+    
+    /* If we have attached slaves, PING them from time to time.
+     * So slaves can implement an explicit timeout to masters, and will
+     * be able to detect a link disconnection even if the TCP connection
+     * will not actually go down. */
+    if (!(server.cronloops % (REDIS_REPL_PING_SLAVE_PERIOD*10))) {
+        listIter li;
+        listNode *ln;
+
+        listRewind(server.slaves,&li);
+        while((ln = listNext(&li))) {
+            redisClient *slave = ln->value;
+
+            /* Don't ping slaves that are in the middle of a bulk transfer
+             * with the master for first synchronization. */
+            if (slave->replstate == REDIS_REPL_SEND_BULK) continue;
+            if (slave->replstate == REDIS_REPL_ONLINE) {
+                /* If the slave is online send a normal ping */
+                addReplySds(slave,sdsnew("PING\r\n"));
+            } else {
+                /* Otherwise we are in the pre-synchronization stage.
+                 * Just a newline will do the work of refreshing the
+                 * connection last interaction time, and at the same time
+                 * we'll be sure that being a single char there are no
+                 * short-write problems. */
+                write(slave->fd, "\n", 1);
+            }
+        }
+    }
 }