Undo slaves state change on failed rdbSaveToSlavesSockets().

As Oran Agra suggested, in startBgsaveForReplication() when the BGSAVE
attempt returns an error, we scan the list of slaves in order to remove
them since there is no way to serve them currently.

However we check for the replication state BGSAVE_START, which was
modified by rdbSaveToSlaveSockets() before forking(). So when fork fails
the state of slaves remain BGSAVE_END and no cleanup is performed.

This commit fixes the problem by making rdbSaveToSlavesSockets() able to
undo the state change on fork failure.

src/rdb.c

@@ -1498,27 +1498,43 @@ int rdbSaveToSlavesSockets(void) {
         exitFromChild((retval == REDIS_OK) ? 0 : 1);
     } else {
         /* Parent */
-        zfree(clientids); /* Not used by parent. Free ASAP. */
         server.stat_fork_time = ustime()-start;
         server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
         latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
         if (childpid == -1) {
             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
                 strerror(errno));
-            zfree(fds);
+
+            /* Undo the state change. The caller will perform cleanup on
+             * all the slaves in BGSAVE_START state, but an early call to
+             * replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
+            listRewind(server.slaves,&li);
+            while((ln = listNext(&li))) {
+                client *slave = ln->value;
+                int j;
+
+                for (j = 0; j < numfds; j++) {
+                    if (slave->id == clientids[j]) {
+                        slave->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
+                        break;
+                    }
+                }
+            }
             close(pipefds[0]);
             close(pipefds[1]);
-            return REDIS_ERR;
+        } else {
+            redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",
+                childpid);
+            server.rdb_save_time_start = time(NULL);
+            server.rdb_child_pid = childpid;
+            server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
+            updateDictResizePolicy();
         }
-        redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",childpid);
-        server.rdb_save_time_start = time(NULL);
-        server.rdb_child_pid = childpid;
-        server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
-        updateDictResizePolicy();
+        zfree(clientids);
         zfree(fds);
-        return REDIS_OK;
+        return (childpid == -1) ? REDIS_ERR : REDIS_OK;
     }
-    return REDIS_OK; /* unreached */
+    return REDIS_OK; /* Unreached. */
 }
 
 void saveCommand(redisClient *c) {