Make disconnectSlaves() synchronous in the base case.

Otherwise we run into that:

Backtrace:
src/redis-server 127.0.0.1:21322(logStackTrace+0x45)[0x479035]
src/redis-server 127.0.0.1:21322(sigsegvHandler+0xb9)[0x4797f9]
/lib/x86_64-linux-gnu/libpthread.so.0(+0x11390)[0x7fd373c5e390]
src/redis-server 127.0.0.1:21322(_serverAssert+0x6a)[0x47660a]
src/redis-server 127.0.0.1:21322(freeReplicationBacklog+0x42)[0x451282]
src/redis-server 127.0.0.1:21322[0x4552d4]
src/redis-server 127.0.0.1:21322[0x4c5593]
src/redis-server 127.0.0.1:21322(aeProcessEvents+0x2e6)[0x42e786]
src/redis-server 127.0.0.1:21322(aeMain+0x1d)[0x42eb0d]
src/redis-server 127.0.0.1:21322(main+0x4c5)[0x42b145]
/lib/x86_64-linux-gnu/libc.so.6(__libc_start_main+0xf0)[0x7fd3738a3830]
src/redis-server 127.0.0.1:21322(_start+0x29)[0x42b409]

Since we disconnect all the replicas and free the replication backlog in
certain replication paths, and the code that will free the replication
backlog expects that no replica is connected.

However we still need to free the replicas asynchronously in certain
cases, as documented in the top comment of disconnectSlaves().

src/networking.c

@@ -1037,14 +1037,25 @@ static void freeClientArgv(client *c) {
 
 /* Close all the slaves connections. This is useful in chained replication
  * when we resync with our own master and want to force all our slaves to
- * resync with us as well. */
+ * resync with us as well.
-void disconnectSlaves(void) {
+ *
+ * If 'async' is non-zero we free the clients asynchronously. This is needed
+ * when we call this function from a context where in the chain of the
+ * callers somebody is iterating the list of clients. For instance when
+ * CLIENT KILL TYPE master is called, caching the master client may
+ * adjust the meaningful offset of replication, and in turn call
+ * discionectSlaves(). Since CLIENT KILL iterates the clients this is
+ * not safe. */
+void disconnectSlaves(int async) {
     listIter li;
     listNode *ln;
     listRewind(server.slaves,&li);
     while((ln = listNext(&li))) {
         listNode *ln = listFirst(server.slaves);
-        freeClientAsync((client*)ln->value);
+        if (async)
+            freeClientAsync((client*)ln->value);
+        else
+            freeClient((client*)ln->value);
     }
 }
 

src/replication.c

@@ -2076,7 +2076,7 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) {
                 memcpy(server.cached_master->replid,new,sizeof(server.replid));
 
                 /* Disconnect all the sub-slaves: they need to be notified. */
-                disconnectSlaves();
+                disconnectSlaves(0);
             }
         }
 
@@ -2349,7 +2349,7 @@ void syncWithMaster(connection *conn) {
      * as well, if we have any sub-slaves. The master may transfer us an
      * entirely different data set and we have no way to incrementally feed
      * our slaves after that. */
-    disconnectSlaves(); /* Force our slaves to resync with us as well. */
+    disconnectSlaves(0); /* Force our slaves to resync with us as well. */
     freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
 
     /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
@@ -2496,7 +2496,7 @@ void replicationSetMaster(char *ip, int port) {
 
     /* Force our slaves to resync with us as well. They may hopefully be able
      * to partially resync with us, but we can notify the replid change. */
-    disconnectSlaves();
+    disconnectSlaves(0);
     cancelReplicationHandshake();
     /* Before destroying our master state, create a cached master using
      * our own parameters, to later PSYNC with the new master. */
@@ -2543,7 +2543,7 @@ void replicationUnsetMaster(void) {
      * of the replication ID change (see shiftReplicationId() call). However
      * the slaves will be able to partially resync with us, so it will be
      * a very fast reconnection. */
-    disconnectSlaves();
+    disconnectSlaves(0);
     server.repl_state = REPL_STATE_NONE;
 
     /* We need to make sure the new master will start the replication stream
@@ -2778,7 +2778,7 @@ void replicationCacheMaster(client *c) {
      * from the stream and their offset would no longer match: upon
      * disconnection they will also trim the final PINGs and will be able
      * to incrementally sync without issues. */
-    if (offset_adjusted) disconnectSlaves();
+    if (offset_adjusted) disconnectSlaves(1);
 }
 
 /* If the "meaningful" offset, that is the offset without the final PINGs

src/server.h

@@ -1660,7 +1660,7 @@ int getClientType(client *c);
 int getClientTypeByName(char *name);
 char *getClientTypeName(int class);
 void flushSlavesOutputBuffers(void);
-void disconnectSlaves(void);
+void disconnectSlaves(int async);
 int listenToPort(int port, int *fds, int *count);
 void pauseClients(mstime_t duration);
 int clientsArePaused(void);