if (masters_count < 3) {
    # 打印error并且退出
}

# 设置master
for (i = 0; i < masters_count; i++) {
    master->dirty = 1;  // 此时是dirty状态
}
# 分配slave
for (i = 0; i < masters_count; i++) {
    if (slave != NULL) {
        assigned_replicas++;
        available_count--;
        if (slave->replicate) sdsfree(slave->replicate);
        slave->replicate = sdsnew(master->name);
        slave->dirty = 1;  // dirty状态
    }
}

if (confirmWithYes("Can I set the above configuration?", ignore_force)) {
    # 输入了yes
    while ((ln = listNext(&li)) != NULL) {
        clusterManagerNode *node = ln->value;
        char *err = NULL;
        // 这里发送指令
        int flushed = clusterManagerFlushNodeConfig(node, &err);
    }
}

static int clusterManagerFlushNodeConfig(clusterManagerNode *node, char **err) {
    // 如果dirty不是1, 那么退出
    if (!node->dirty) return 0;
    if (node->replicate != NULL) {
        // 指令1 <==============================
        reply = CLUSTER_MANAGER_COMMAND(node, "CLUSTER REPLICATE %s",
                                        node->replicate);
    } else {
        // 指令2 <================================
        int added = clusterManagerAddSlots(node, err);
        if (!added || *err != NULL) success = 0;
    }
    // 把dirty设置为0 <===========================
    node->dirty = 0;
cleanup:
    if (reply != NULL) freeReplyObject(reply);
    return success;
}

int config_epoch = 1;  // 在while中每次都++
listRewind(cluster_manager.nodes, &li);
while ((ln = listNext(&li)) != NULL) {
    clusterManagerNode *node = ln->value;
    redisReply *reply = NULL;
    reply = CLUSTER_MANAGER_COMMAND(node,
                                    "cluster set-config-epoch %d",
                                    config_epoch++);
    if (reply != NULL) freeReplyObject(reply);
}

while ((ln = listNext(&li)) != NULL) {
    clusterManagerNode *node = ln->value;
    if (first == NULL) {  // 保存first节点
        first = node;
        continue;
    }
    redisReply *reply = NULL;
    // 每个节点都向first节点发送meet指令
    reply = CLUSTER_MANAGER_COMMAND(node, "cluster meet %s %d",
                                    first->ip, first->port);

}

// 等待一秒
sleep(1);
// 等待所有节点join完毕
// 这里会发送cluster nodes给每一个节点
// 检查每个节点的返回
clusterManagerWaitForClusterJoin();

static int clusterManagerCommandAddNode(int argc, char **argv) {

clusterManagerNode *first = listFirst(cluster_manager.nodes)->value;
listAddNodeTail(cluster_manager.nodes, new_node);
added = 1;

// Send CLUSTER MEET command to the new node
clusterManagerLogInfo(">>> Send CLUSTER MEET to node %s:%d to make it "
                      "join the cluster.\n", ip, port);
reply = CLUSTER_MANAGER_COMMAND(new_node, "CLUSTER MEET %s %d",
                                first->ip, first->port);

}

#define CLUSTER_MAX_REJOIN_DELAY 5000  // rejoin的最小和最大等待时间  <=====================
#define CLUSTER_MIN_REJOIN_DELAY 500
#define CLUSTER_WRITABLE_DELAY 2000  // 2s的写延迟  <=======================

void clusterUpdateState(void) {

    /* If this is a master node, wait some time before turning the state
     * into OK, since it is not a good idea to rejoin the cluster as a writable
     * master, after a reboot, without giving the cluster a chance to
     * reconfigure this node. Note that the delay is calculated starting from
     * the first call to this function and not since the server start, in order
     * to don't count the DB loading time. */
    if (first_call_time == 0) first_call_time = mstime();
    // 这里是阻止一个处于分区状态的master重启的时候, 读取旧的配置发现自己是fail并且是master
    // 但是我们不知道我们已经处于分区状态多久了, 此时走到下面的
    // rejoin逻辑的话, 会发现其实我们是在线状态, 就直接允许写入, 这个也是不允许的  <===========================
    // 所以这里first_call_time经过第一次修改之后就不会变了
    if (nodeIsMaster(myself) &&
        server.cluster->state == CLUSTER_FAIL &&
        mstime() - first_call_time < CLUSTER_WRITABLE_DELAY) return;


    // 记录下我们处于分区状态的时间
    // 重启就丢失, 所以需要上面的判断 <==================
    /* If we are in a minority partition, change the cluster state
     * to FAIL. */
    {
        int needed_quorum = (server.cluster->size / 2) + 1;

        if (reachable_masters < needed_quorum) {
            new_state = CLUSTER_FAIL;
            among_minority_time = mstime();
        }
    }


    // 需要rejoin
    if (new_state != server.cluster->state) {
        mstime_t rejoin_delay = server.cluster_node_timeout;

        // 至少等待这些多的时间
        /* If the instance is a master and was partitioned away with the
         * minority, don't let it accept queries for some time after the
         * partition heals, to make sure there is enough time to receive
         * a configuration update. */
        if (rejoin_delay > CLUSTER_MAX_REJOIN_DELAY)
            rejoin_delay = CLUSTER_MAX_REJOIN_DELAY;
        if (rejoin_delay < CLUSTER_MIN_REJOIN_DELAY)
            rejoin_delay = CLUSTER_MIN_REJOIN_DELAY;

        // 如果我们当前是ok, 但是等待时间没有达到rejoin_delay
        // 那么继续是fail状态, 等待同步configuration <============================
        if (new_state == CLUSTER_OK &&
            nodeIsMaster(myself) &&
            mstime() - among_minority_time < rejoin_delay)
        {
            return;
        }

}

/* This is executed 10 times every second */
// 每秒执行10次
void clusterCron(void) {
    iteration++; /* Number of times this function was called so far. */
    // 每秒执行10次, 每10次执行一次ping, 所以每秒ping一个instance
    if (!(iteration % 10)) {
        int j;
        /* Check a few random nodes and ping the one with the oldest
         * pong_received time. */
        // 采样5次  <======================================
        for (j = 0; j < 5; j++) {
            de = dictGetRandomKey(server.cluster->nodes);
            clusterNode *this = dictGetVal(de);
            /* Don't ping nodes disconnected or with a ping currently active. */
            if (this->link == NULL || this->ping_sent != 0) continue;
            if (this->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
                continue;
            // 比较我们收到目标节点的时间, 取最小的  <==========================
            if (min_pong_node == NULL || min_pong > this->pong_received) {
                min_pong_node = this;
                min_pong = this->pong_received;
            }
        }
        // ping目标节点   <===============================
        if (min_pong_node) {
            serverLog(LL_DEBUG,"Pinging node %.40s", min_pong_node->name);
            // 发送ping  <========================
            clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);
        }
    }

    // 为了保证在cluster_node_timeout / 2时间内ping完所有的节点  <====================================
    di = dictGetSafeIterator(server.cluster->nodes);
    while((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        now = mstime(); /* Use an updated time at every iteration. */

        /* If we have currently no active ping in this instance, and the
         * received PONG is older than half the cluster timeout, send
         * a new ping now, to ensure all the nodes are pinged without
         * a too big delay. */
        if (node->link &&
            node->ping_sent == 0 &&
            (now - node->pong_received) > server.cluster_node_timeout/2)
        {
            clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);
            continue;
        }
    }

}

// 处理gossip消息
void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
    uint16_t count = ntohs(hdr->count);
    clusterMsgDataGossip *g = (clusterMsgDataGossip*) hdr->data.ping.gossip;
    // 这个是sender  <===========================
    clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender);
    while(count--) {
        // 拿到目标instance  <==========================
        uint16_t flags = ntohs(g->flags);
        clusterNode *node;
        sds ci;
        //
        node = clusterLookupNode(g->nodename);
        if (node) {

            // 这里就是改进方案  <===========================
            /* If from our POV the node is up (no failure flags are set),
             * we have no pending ping for the node, nor we have failure
             * reports for this node, update the last pong time with the
             * one we see from the other nodes. */
            // flags是从g中拿到的   <========================
            // 如果flags不是PFAIL或者FAIL, 那么显然sender说node是在线状态, 我们选择相信  <=================
            if (!(flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) &&
                node->ping_sent == 0 &&
                clusterNodeFailureReportsCount(node) == 0)
            {
                mstime_t pongtime = ntohl(g->pong_received);
                pongtime *= 1000; /* Convert back to milliseconds. */

                /* Replace the pong time with the received one only if
                 * it's greater than our view but is not in the future
                 * (with 500 milliseconds tolerance) from the POV of our
                 * clock. */
                // 选个最大时间, 如果sender收到node的pong时间小于当前instance的时间+500ms, 并且
                // 大于我们收到的node的pong时间  <==========================
                if (pongtime <= (server.mstime+500) &&
                    pongtime > node->pong_received)
                {
                    node->pong_received = pongtime;
                }
            }
        }
    }
}

int clusterProcessPacket(clusterLink *link) {

    if (sender) sender->data_received = now; // 不管怎么样, 更新data_received  <=====================

    /* PING, PONG, MEET: process config information. */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
        type == CLUSTERMSG_TYPE_MEET)
    {
        if (link->node && type == CLUSTERMSG_TYPE_PONG) {
            link->node->pong_received = now;
            link->node->ping_sent = 0;
        }


        // 最后会处理gossip消息  <=======================
        if (sender) clusterProcessGossipSection(hdr,link);
    }

}

void clusterSendPing(clusterLink *link, int type) {

    if (link->node && type == CLUSTERMSG_TYPE_PING)
        link->node->ping_sent = mstime();

}

/* This is executed 10 times every second */
void clusterCron(void) {

    di = dictGetSafeIterator(server.cluster->nodes);
    while((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        /* If we are not receiving any data for more than half the cluster
         * timeout, reconnect the link: maybe there is a connection
         * issue even if the node is alive. */
        // 计算ping_delay和data_delay  <======================
        mstime_t ping_delay = now - node->ping_sent;
        mstime_t data_delay = now - node->data_received;
        // 判断是否需要重连   <===========================
        if (node->link && /* is connected */
            now - node->link->ctime >
            server.cluster_node_timeout && /* was not already reconnected */
            node->ping_sent && /* we already sent a ping */
            node->pong_received < node->ping_sent && /* still waiting pong */
            /* and we are waiting for the pong more than timeout/2 */
            ping_delay > server.cluster_node_timeout/2 &&
            /* and in such interval we are not seeing any traffic at all. */
            data_delay > server.cluster_node_timeout/2)
        {
            /* Disconnect the link, it will be reconnected automatically. */
            freeClusterLink(node->link);
        }

        // 如果没有发送ping, 那就不管它  <=============================
        /* Check only if we have an active ping for this instance. */
        if (node->ping_sent == 0) continue;

        /* Check if this node looks unreachable.
         * Note that if we already received the PONG, then node->ping_sent
         * is zero, so can't reach this code at all, so we don't risk of
         * checking for a PONG delay if we didn't sent the PING.
         *
         * We also consider every incoming data as proof of liveness, since
         * our cluster bus link is also used for data: under heavy data
         * load pong delays are possible. */
        // 1. 如果没有收到pong, 那么data_received为0, 那么node_delay取ping_delay
        // 2. 如果收到了pong, 那么ping_sent为0, 取data_received
        // 这里是取最小值  <======================================
        mstime_t node_delay = (ping_delay < data_delay) ? ping_delay :
                                                          data_delay;

        // 超时时间大于cluster_node_timeout, 设置instance为PFAIL状态  <===========================
        if (node_delay > server.cluster_node_timeout) {
            /* Timeout reached. Set the node as possibly failing if it is
             * not already in this state. */
            if (!(node->flags & (CLUSTER_NODE_PFAIL|CLUSTER_NODE_FAIL))) {
                serverLog(LL_DEBUG,"*** NODE %.40s possibly failing",
                    node->name);
                node->flags |= CLUSTER_NODE_PFAIL;
                update_state = 1;
            }
        }

    }

}

/* This is executed 10 times every second */
void clusterCron(void) {

    di = dictGetSafeIterator(server.cluster->nodes);
    server.cluster->stats_pfail_nodes = 0;  // 标志为先置为0  <==============================

    while((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);

        /* Not interested in reconnecting the link with myself or nodes
         * for which we have no address. */
        if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR)) continue;

        if (node->flags & CLUSTER_NODE_PFAIL)
            server.cluster->stats_pfail_nodes++;  // 如果某个instance是PFAIL状态, 增加计数  <==============
    }


}

/* Send a PING or PONG packet to the specified node, making sure to add enough
 * gossip information. */
void clusterSendPing(clusterLink *link, int type) {

    int pfail_wanted = server.cluster->stats_pfail_nodes;  // 拿到计数  <=======================

    while(freshnodes > 0 && gossipcount < wanted && maxiterations--) {

        // 如果有instance为PFAIL状态, 那么我们要带上  <===================
        /* If there are PFAIL nodes, add them at the end. */
        if (pfail_wanted) {
            dictIterator *di;
            dictEntry *de;

            di = dictGetSafeIterator(server.cluster->nodes);
            while((de = dictNext(di)) != NULL && pfail_wanted > 0) {
                clusterNode *node = dictGetVal(de);
                if (node->flags & CLUSTER_NODE_HANDSHAKE) continue;
                if (node->flags & CLUSTER_NODE_NOADDR) continue;
                if (!(node->flags & CLUSTER_NODE_PFAIL)) continue;
                clusterSetGossipEntry(hdr,gossipcount,node);  // 这里设置hdr中的flag为node中的flag, 所以把PFAIL状态给带上了  <===========================
                freshnodes--;
                gossipcount++;
                /* We take the count of the slots we allocated, since the
                 * PFAIL stats may not match perfectly with the current number
                 * of PFAIL nodes. */
                pfail_wanted--;
            }
            dictReleaseIterator(di);
        }

    }
}

void clusterSetGossipEntry(clusterMsg *hdr, int i, clusterNode *n) {
    clusterMsgDataGossip *gossip;
    gossip = &(hdr->data.ping.gossip[i]);
    memcpy(gossip->nodename,n->name,CLUSTER_NAMELEN);
    gossip->ping_sent = htonl(n->ping_sent/1000);
    gossip->pong_received = htonl(n->pong_received/1000);
    memcpy(gossip->ip,n->ip,sizeof(n->ip));
    gossip->port = htons(n->port);
    gossip->cport = htons(n->cport);
    gossip->flags = htons(n->flags);
    gossip->notused1 = 0;
}

void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {

    uint16_t count = ntohs(hdr->count);  // gossip列表的长度  <====================
    clusterMsgDataGossip *g = (clusterMsgDataGossip*) hdr->data.ping.gossip;
    clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender);

    while(count--) {
        uint16_t flags = ntohs(g->flags);

        /* Update our state accordingly to the gossip sections */
        node = clusterLookupNode(g->nodename); // 拿到instance  <===================
        if (node) {
            /* We already know this node.
               Handle failure reports, only when the sender is a master. */
            // 如果sender是master, 我们采取行动!!! <====================================
            if (sender && nodeIsMaster(sender) && node != myself) {
                // A说C是PFAIL状态  <=========================
                if (flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) {
                    if (clusterNodeAddFailureReport(node,sender)) {
                        serverLog(LL_VERBOSE,
                            "Node %.40s reported node %.40s as not reachable.",
                            sender->name, node->name);
                    }
                    //  更新一下自己对C的状态
                    markNodeAsFailingIfNeeded(node);
                } else {
                    if (clusterNodeDelFailureReport(node,sender)) {
                        serverLog(LL_VERBOSE,
                            "Node %.40s reported node %.40s is back online.",
                            sender->name, node->name);
                    }
                }
            }
        }
    }

}

* 1) Either we reach the majority and eventually the FAIL state will propagate
*    to all the cluster.
* 2) Or there is no majority so no slave promotion will be authorized and the
*    FAIL flag will be cleared after some time.
void markNodeAsFailingIfNeeded(clusterNode *node) {
    int failures;
    int needed_quorum = (server.cluster->size / 2) + 1;

    if (!nodeTimedOut(node)) return; /* We can reach it. */ // 这里查询node->, 在clusterCron函数中更新!!!!  <=====================
    if (nodeFailed(node)) return; /* Already FAILing. */

    // 走到这里表示该节点是我们第一次不能访问, 但是还要根据大多数节点确认 <=======================
    failures = clusterNodeFailureReportsCount(node);  // 这里查询本地保存的其他node的gossip
    /* Also count myself as a voter if I'm a master. */
    if (nodeIsMaster(myself)) failures++;
    // 如果没有大多数节点说对方掉线了, 那么它还是在线的, 即使我们连接不到它  <======================
    if (failures < needed_quorum) return; /* No weak agreement from masters. */

    serverLog(LL_NOTICE,
        "Marking node %.40s as failing (quorum reached).", node->name);

    // 走到这里表示我们既不能连接它同时大多数节点说也连接不到它, 那么它就是掉线了
    // 开始把对方标识为FAIL <=====================
    /* Mark the node as failing. */
    node->flags &= ~CLUSTER_NODE_PFAIL;
    node->flags |= CLUSTER_NODE_FAIL;
    node->fail_time = mstime();

    /* Broadcast the failing node name to everybody, forcing all the other
     * reachable nodes to flag the node as FAIL.
     * We do that even if this node is a replica and not a master: anyway
     * the failing state is triggered collecting failure reports from masters,
     * so here the replica is only helping propagating this status. */
    clusterSendFail(node->name);  // 广播, 不走ping  <===============================

}

/* This function is called only if a node is marked as FAIL, but we are able
 * to reach it again. It checks if there are the conditions to undo the FAIL
 * state. */
void clearNodeFailureIfNeeded(clusterNode *node) {
    mstime_t now = mstime();

    serverAssert(nodeFailed(node));

    /* For slaves we always clear the FAIL flag if we can contact the
     * node again. */
    // 如果该node是slave, 直接清除
    // 如果该node是master, 同时其不负责任何的slot, 那么也直接清除  <============================
    if (nodeIsSlave(node) || node->numslots == 0) {
        serverLog(LL_NOTICE,
            "Clear FAIL state for node %.40s: %s is reachable again.",
                node->name,
                nodeIsSlave(node) ? "replica" : "master without slots");
        node->flags &= ~CLUSTER_NODE_FAIL;  // 清除该node的FAIL状态位 <================================
        clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
    }

    /* If it is a master and...
     * 1) The FAIL state is old enough.
     * 2) It is yet serving slots from our point of view (not failed over).
     * Apparently no one is going to fix these slots, clear the FAIL flag. */
    // 这里是第3个情况  <=========================
    if (nodeIsMaster(node) && node->numslots > 0 &&
        (now - node->fail_time) >
        (server.cluster_node_timeout * CLUSTER_FAIL_UNDO_TIME_MULT))
    {
        serverLog(LL_NOTICE,
            "Clear FAIL state for node %.40s: is reachable again and nobody is serving its slots after some time.",
                node->name);
        node->flags &= ~CLUSTER_NODE_FAIL;
        clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
    }
}

static int clusterManagerCommandCreate(int argc, char **argv) {
        int config_epoch = 1;
        listRewind(cluster_manager.nodes, &li);
        // 在while中, 每次config_epoch都++
        while ((ln = listNext(&li)) != NULL) {
            clusterManagerNode *node = ln->value;
            redisReply *reply = NULL;
            reply = CLUSTER_MANAGER_COMMAND(node,
                                            "cluster set-config-epoch %d",
                                            config_epoch++);
            if (reply != NULL) freeReplyObject(reply);
        }

}

void clusterCommand(client *c) {
    if (server.cluster_enabled == 0) {
        addReplyError(c,"This instance has cluster support disabled");
        return;
    }
    if(){
      // 其他命令
    } else if (!strcasecmp(c->argv[1]->ptr,"set-config-epoch") && c->argc == 3)
    {
        // 这里说明只能向一个fresh的instance发起set-config-epoch命令  <=======================
        // 一个fresh的instance是config_epoch为0, 不知道任何其他节点的(也就是没加入任何集群) <<=====================
        /* CLUSTER SET-CONFIG-EPOCH <epoch>
         *
         * The user is allowed to set the config epoch only when a node is
         * totally fresh: no config epoch, no other known node, and so forth.
         * This happens at cluster creation time to start with a cluster where
         * every node has a different node ID, without to rely on the conflicts
         * resolution system which is too slow when a big cluster is created. */
        long long epoch;

        if(){
          // 其他条件
        } else {
            // 设置自己的configEpoch  <==========================
            myself->configEpoch = epoch;
            serverLog(LL_WARNING,
                "configEpoch set to %llu via CLUSTER SET-CONFIG-EPOCH",
                (unsigned long long) myself->configEpoch);

            // 这里server.cluster->currentEpoch和myself->configEpoch相等
            // 都是设置的值  <=================================
            if (server.cluster->currentEpoch < (uint64_t)epoch)
                server.cluster->currentEpoch = epoch;
            /* No need to fsync the config here since in the unlucky event
             * of a failure to persist the config, the conflict resolution code
             * will assign a unique config to this node. */
            clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|
                                 CLUSTER_TODO_SAVE_CONFIG);
            addReply(c,shared.ok);
        }
    }
}

void clusterCommand(client *c) {
    if(){
      // 其他命令
    } else if ((!strcasecmp(c->argv[1]->ptr,"addslots") ||
               !strcasecmp(c->argv[1]->ptr,"delslots")) && c->argc >= 3)
    {
        // 并不会提升epoch
        /* CLUSTER ADDSLOTS <slot> [slot] ... */
        /* CLUSTER DELSLOTS <slot> [slot] ... */
        // 很多代码被省略了
        zfree(slots);
        clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
        addReply(c,shared.ok);

    }else if (!strcasecmp(c->argv[1]->ptr,"setslot") && c->argc >= 4) {
        if(){
           // 其他条件, 比如migrating和importing
        } else if (!strcasecmp(c->argv[3]->ptr,"node") && c->argc == 5) {
            // slot已经迁移完成了!!!!!!!! <<====================
            /* CLUSTER SETSLOT <SLOT> NODE <NODE ID> */

            // 下面两句分别是操作server.cluster->slots
            // server.cluster->slots保存的是哪些slot由哪些master负责
            // server.cluster->slots = [master1, master1, master2, master2, ....]
            // 删除就是server.cluster->slots[i] = NULL
            // 添加就是server.cluster->slots[j] = master  <========================
            clusterDelSlot(slot);
            clusterAddSlot(n,slot);

            /* If this node was importing this slot, assigning the slot to
             * itself also clears the importing status. */
            if (n == myself &&
                server.cluster->importing_slots_from[slot])
            {
                /* This slot was manually migrated, set this node configEpoch
                 * to a new epoch so that the new version can be propagated
                 * by the cluster.
                 *
                 * Note that if this ever results in a collision with another
                 * node getting the same configEpoch, for example because a
                 * failover happens at the same time we close the slot, the
                 * configEpoch collision resolution will fix it assigning
                 * a different epoch to each node. */
                // 这里提升自己的epoch
                // 如果提升?  <=================================
                if (clusterBumpConfigEpochWithoutConsensus() == C_OK) {
                    serverLog(LL_WARNING,
                        "configEpoch updated after importing slot %d", slot);
                }
                server.cluster->importing_slots_from[slot] = NULL;
                /* After importing this slot, let the other nodes know as
                 * soon as possible. */
                clusterBroadcastPong(CLUSTER_BROADCAST_ALL);
            }
        }
    }
}

myepoch = (nodeIsSlave(myself) && myself->slaveof) ?
          myself->slaveof->configEpoch : myself->configEpoch; // 拿到的是master的configEpoch <========
sds info = sdscatprintf(sdsempty(),
    "cluster_state:%s\r\n"
    "cluster_slots_assigned:%d\r\n"
    "cluster_slots_ok:%d\r\n"
    "cluster_slots_pfail:%d\r\n"
    "cluster_slots_fail:%d\r\n"
    "cluster_known_nodes:%lu\r\n"
    "cluster_size:%d\r\n"
    "cluster_current_epoch:%llu\r\n"
    "cluster_my_epoch:%llu\r\n"  // 显示的是master的configEpoch <================
    , statestr[server.cluster->state],
    slots_assigned,
    slots_ok,
    slots_pfail,
    slots_fail,
    dictSize(server.cluster->nodes),
    server.cluster->size,
    (unsigned long long) server.cluster->currentEpoch,
    (unsigned long long) myepoch
);

{n1: {"configEpoch": 1}, n2: {"configEpoch": 2}, n3: {"configEpoch": 3}}

void clusterCommand(client *c) {
    if(){
        // 其他命令
    } else if (!strcasecmp(c->argv[3]->ptr,"node") && c->argc == 5) {
        /* CLUSTER SETSLOT <SLOT> NODE <NODE ID> */

        clusterDelSlot(slot);
        clusterAddSlot(n,slot);

        /* If this node was importing this slot, assigning the slot to
         * itself also clears the importing status. */
        if (n == myself &&
            server.cluster->importing_slots_from[slot])
        {
            /* This slot was manually migrated, set this node configEpoch
             * to a new epoch so that the new version can be propagated
             * by the cluster.
             *
             * Note that if this ever results in a collision with another
             * node getting the same configEpoch, for example because a
             * failover happens at the same time we close the slot, the
             * configEpoch collision resolution will fix it assigning
             * a different epoch to each node. */
            // 这里提升server.cluster->currentEpoch和myelf->configEpoch !!!!!!!!!!!!!!!! <=======================
            if (clusterBumpConfigEpochWithoutConsensus() == C_OK) {
                serverLog(LL_WARNING,
                    "configEpoch updated after importing slot %d", slot);
            }
            server.cluster->importing_slots_from[slot] = NULL;
            /* After importing this slot, let the other nodes know as
             * soon as possible. */
            clusterBroadcastPong(CLUSTER_BROADCAST_ALL); // 广播配置改变的  <==========================
        }
    }

}

int clusterProcessPacket(clusterLink *link) {

    /* PING, PONG, MEET: process config information. */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
        type == CLUSTERMSG_TYPE_MEET)
    {
        /* If our config epoch collides with the sender's try to fix
         * the problem. */
        // 如果发现configEpoch冲突!!!!! <<====================
        if (sender &&
            nodeIsMaster(myself) && nodeIsMaster(sender) &&
            senderConfigEpoch == myself->configEpoch)
        {
            clusterHandleConfigEpochCollision(sender);
        }

    }

}

 /*
 * When this function gets called, what happens is that if this node
 * has the lexicographically smaller Node ID compared to the other node
 * with the conflicting epoch (the 'sender' node), it will assign itself
 * the greatest configuration epoch currently detected among nodes plus 1.
 *
 * This means that even if there are multiple nodes colliding, the node
 * with the greatest Node ID never moves forward, so eventually all the nodes
 * end with a different configuration epoch.
 */
void clusterHandleConfigEpochCollision(clusterNode *sender) {
    /* Prerequisites: nodes have the same configEpoch and are both masters. */
    if (sender->configEpoch != myself->configEpoch ||
        !nodeIsMaster(sender) || !nodeIsMaster(myself)) return;
    /* Don't act if the colliding node has a smaller Node ID. */
    // sender的名字的字典序小于我们, 那么sender可以继续, 我们就不继续了!!!!!!!  <<=========================
    if (memcmp(sender->name,myself->name,CLUSTER_NAMELEN) <= 0) return;
    // 走到这里说明我们的名字的字典序比较小, 自增一下!!!!  <<===========================
    /* Get the next ID available at the best of this node knowledge. */
    server.cluster->currentEpoch++;
    myself->configEpoch = server.cluster->currentEpoch;
    clusterSaveConfigOrDie(1);
    serverLog(LL_VERBOSE,
        "WARNING: configEpoch collision with node %.40s."
        " configEpoch set to %llu",
        sender->name,
        (unsigned long long) myself->configEpoch);
}

int clusterGetSlaveRank(void) {
    long long myoffset;
    int j, rank = 0;  // rank默认为0  <======================
    clusterNode *master;

    serverAssert(nodeIsSlave(myself));
    master = myself->slaveof;
    if (master == NULL) return 0; /* Never called by slaves without master. */

    myoffset = replicationGetSlaveOffset();
    // 遍历所有的slaves, 找出有多少个slave的offset比我们的大  <================
    for (j = 0; j < master->numslaves; j++)
        if (master->slaves[j] != myself &&
            !nodeCantFailover(master->slaves[j]) &&
            master->slaves[j]->repl_offset > myoffset) rank++;
    return rank;  // 没有人比我们的offset更大, 那么我们就是rank0
}

void clusterHandleSlaveFailover(void) {

    // 这几个条件就是上面提到的  <====================================
    /* Pre conditions to run the function, that must be met both in case
     * of an automatic or manual failover:
     * 1) We are a slave.
     * 2) Our master is flagged as FAIL, or this is a manual failover.
     * 3) We don't have the no failover configuration set, and this is
     *    not a manual failover.
     * 4) It is serving slots. */
    if (nodeIsMaster(myself) ||
        myself->slaveof == NULL ||
        (!nodeFailed(myself->slaveof) && !manual_failover) ||
        (server.cluster_slave_no_failover && !manual_failover) ||
        myself->slaveof->numslots == 0)
    {
        /* There are no reasons to failover, so we set the reason why we
         * are returning without failing over to NONE. */
        server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE;
        return;
    }

}

/* This is executed 10 times every second */
void clusterCron(void) {

    if (nodeIsSlave(myself)) {
        clusterHandleManualFailover();
        // 注意这里的判断条件
        // 只有设置了CLUSTER_MODULE_FLAG_NO_FAILOVER才不会去执行选举
        // 但是这个标志只有手动设置, 代码中不会设置
        // 所以正常情况下这个条件总是!0  <=============================================
        if (!(server.cluster_module_flags & CLUSTER_MODULE_FLAG_NO_FAILOVER))
            clusterHandleSlaveFailover();
    }

}

void clusterHandleSlaveFailover(void) {

    // 我们之前发送auth request的时间就是failover_auth_time
    // 那么auth_age就是我们等待ack等待了多久  <==============================
    mstime_t auth_age = mstime() - server.cluster->failover_auth_time;
    int needed_quorum = (server.cluster->size / 2) + 1;  // 注意这里是大多数master而不是大多数节点

    // 这里auth_timeout就是等待FAILOVER_AUTH_ACK的超时
    // auth_retry_time就是重新发起选举的间隔  <===========================
    auth_timeout = server.cluster_node_timeout*2;
    if (auth_timeout < 2000) auth_timeout = 2000;
    auth_retry_time = auth_timeout*2;


    if (auth_age > auth_retry_time) {
       // 等了至少4s了, 重新发送auth request吧
       // 再次设置延迟时间!!!!!! <=======================
        server.cluster->failover_auth_time = mstime() +
            500 + /* Fixed delay of 500 milliseconds, let FAIL msg propagate. */
            random() % 500; /* Random delay between 0 and 500 milliseconds. */
        server.cluster->failover_auth_count = 0;
        // 设置failover_auth_sent为0表示需要发送auth request!!! <=======================
        server.cluster->failover_auth_sent = 0;
        server.cluster->failover_auth_rank = clusterGetSlaveRank();
        /* We add another delay that is proportional to the slave rank.
         * Specifically 1 second * rank. This way slaves that have a probably
         * less updated replication offset, are penalized. */
        server.cluster->failover_auth_time +=
            server.cluster->failover_auth_rank * 1000;
    }

    if (auth_age > auth_timeout) {
        // 如果我们等待ack超过了2s了, 那么记录一下, 退出
        // 下次进来的时候, 就走重新发送auth request逻辑  <======================
        clusterLogCantFailover(CLUSTER_CANT_FAILOVER_EXPIRED);
        return;
    }


    // 这里我们没有发送auth request, 并且
    // 不是手动failover的话, 我们需要检查一下我们是否收到
    // 了更大的rank, 也就是别人的rank更大
    // 那么就等待更久, 为什么?
    // 因为比我们大的rank可能也会掉线, 所以我们需要随时准备好了 <=============================
    /* It is possible that we received more updated offsets from other
     * slaves for the same master since we computed our election delay.
     * Update the delay if our rank changed.
     *
     * Not performed if this is a manual failover. */
    if (server.cluster->failover_auth_sent == 0 &&
        server.cluster->mf_end == 0)
    {
        int newrank = clusterGetSlaveRank();
        if (newrank > server.cluster->failover_auth_rank) {
            long long added_delay =
                (newrank - server.cluster->failover_auth_rank) * 1000;
            server.cluster->failover_auth_time += added_delay;
            server.cluster->failover_auth_rank = newrank;
            serverLog(LL_WARNING,
                "Replica rank updated to #%d, added %lld milliseconds of delay.",
                newrank, added_delay);
        }
    }

    // 经过那么判断只会, 那么我们需要发送auth request了 <======================
    /* Ask for votes if needed. */
    if (server.cluster->failover_auth_sent == 0) {
        server.cluster->currentEpoch++;
        server.cluster->failover_auth_epoch = server.cluster->currentEpoch;
        serverLog(LL_WARNING,"Starting a failover election for epoch %llu.",
            (unsigned long long) server.cluster->currentEpoch);
        clusterRequestFailoverAuth();
        // 设置failover_auth_sent标志位为1
        server.cluster->failover_auth_sent = 1;
        clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
                             CLUSTER_TODO_UPDATE_STATE|
                             CLUSTER_TODO_FSYNC_CONFIG);
        return; /* Wait for replies. */
    }

    // 最后最后!!!!! <=======================
    // 这里说明我们收到了auth ack, 所以我们需要判断一下是否达到了大多数master <=========================
    /* Check if we reached the quorum. */
    if (server.cluster->failover_auth_count >= needed_quorum) {
        /* We have the quorum, we can finally failover the master. */

        serverLog(LL_WARNING,
            "Failover election won: I'm the new master.");

        /* Update my configEpoch to the epoch of the election. */
        if (myself->configEpoch < server.cluster->failover_auth_epoch) {
            myself->configEpoch = server.cluster->failover_auth_epoch;
            serverLog(LL_WARNING,
                "configEpoch set to %llu after successful failover",
                (unsigned long long) myself->configEpoch);
        }

        /* Take responsibility for the cluster slots. */
        clusterFailoverReplaceYourMaster(); // 广播我是master这个消息  <========================
    } else {
        clusterLogCantFailover(CLUSTER_CANT_FAILOVER_WAITING_VOTES);
    }
}

void clusterFailoverReplaceYourMaster(void) {
    int j;
    clusterNode *oldmaster = myself->slaveof;

    if (nodeIsMaster(myself) || oldmaster == NULL) return;

    /* 1) Turn this node into a master. */
    clusterSetNodeAsMaster(myself);
    replicationUnsetMaster();

    /* 2) Claim all the slots assigned to our master. */
    // 这里把所有oldmaster负责的slot转移到自己来 <===========================
    for (j = 0; j < CLUSTER_SLOTS; j++) {
        if (clusterNodeGetSlotBit(oldmaster,j)) {
            clusterDelSlot(j);
            clusterAddSlot(myself,j);
        }
    }

    /* 3) Update state and save config. */
    clusterUpdateState();
    clusterSaveConfigOrDie(1);

    /* 4) Pong all the other nodes so that they can update the state
     *    accordingly and detect that we switched to master role. */
    clusterBroadcastPong(CLUSTER_BROADCAST_ALL); // 广播PONG信息给所有的节点 <==========================

    /* 5) If there was a manual failover in progress, clear the state. */
    resetManualFailover();
}

typedef struct clusterState {
    clusterNode *slots[CLUSTER_SLOTS];

}

typedef struct clusterNode {
    unsigned char slots[CLUSTER_SLOTS/8]; /* slots handled by this node */
}

void clusterSendPing(clusterLink *link, int type) {

    clusterBuildMessageHdr(hdr,type);

}

void clusterBuildMessageHdr(clusterMsg *hdr, int type) {

    clusterNode *master;

    // 如果我们不是master, 那么我们发送的其实是master的信息
    master = (nodeIsSlave(myself) && myself->slaveof) ?
              myself->slaveof : myself;


    // 拷贝clusterNode中的slots信息
    memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));

}

/* Vote for the node asking for our vote if there are the conditions. */
// 这里node是gossip的sender, 也就是slave
void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) {
    clusterNode *master = node->slaveof;  // 拿到master  <========================
    uint64_t requestCurrentEpoch = ntohu64(request->currentEpoch);
    uint64_t requestConfigEpoch = ntohu64(request->configEpoch);
    unsigned char *claimed_slots = request->myslots;
    int force_ack = request->mflags[0] & CLUSTERMSG_FLAG0_FORCEACK;
    int j;

    /* IF we are not a master serving at least 1 slot, we don't have the
     * right to vote, as the cluster size in Redis Cluster is the number
     * of masters serving at least one slot, and quorum is the cluster
     * size + 1 */
    // 如果我不是master或者我不负责任何slot, 不管  <===========================
    if (nodeIsSlave(myself) || myself->numslots == 0) return;

    /* Request epoch must be >= our currentEpoch.
     * Note that it is impossible for it to actually be greater since
     * our currentEpoch was updated as a side effect of receiving this
     * request, if the request epoch was greater. */
    // 判断request的epoch和自己的epoch
    // 如果前者小于后者, 拒绝  <===========================
    if (requestCurrentEpoch < server.cluster->currentEpoch) {
        serverLog(LL_WARNING,
            "Failover auth denied to %.40s: reqEpoch (%llu) < curEpoch(%llu)",
            node->name,
            (unsigned long long) requestCurrentEpoch,
            (unsigned long long) server.cluster->currentEpoch);
        return;
    }

    // lastVoteEpoch表示我们最近投票的epoch
    // 如果lastVoteEpoch和我们当前看到的currentEpoch相等, 相当于我们已经投过票了并且配置已经改变过了
    // 不能再投票了, 这里可能是某个slave的auth request因为网络问题(比如网络分区)很久才到达
    // 到达的时候另外一个slave已经发起并完成了另一轮auth request  <======================
    /* I already voted for this epoch? Return ASAP. */
    if (server.cluster->lastVoteEpoch == server.cluster->currentEpoch) {
        serverLog(LL_WARNING,
                "Failover auth denied to %.40s: already voted for epoch %llu",
                node->name,
                (unsigned long long) server.cluster->currentEpoch);
        return;
    }

    // 判断该slave的master是否是FAIL状态 <===================
    /* Node must be a slave and its master down.
     * The master can be non failing if the request is flagged
     * with CLUSTERMSG_FLAG0_FORCEACK (manual failover). */
    if (nodeIsMaster(node) || master == NULL ||
        (!nodeFailed(master) && !force_ack))
    {
        if (nodeIsMaster(node)) {
            serverLog(LL_WARNING,
                    "Failover auth denied to %.40s: it is a master node",
                    node->name);
        } else if (master == NULL) {
            serverLog(LL_WARNING,
                    "Failover auth denied to %.40s: I don't know its master",
                    node->name);
        } else if (!nodeFailed(master)) {
            serverLog(LL_WARNING,
                    "Failover auth denied to %.40s: its master is up",
                    node->name);
        }
        return;
    }

    // 如果我们之前已经为这个master的选举通过票了
    // 即使这个request的epoch更大
    // 我们也不能在cluster_node_timeout * 2的时间内对同一个master的选举投两次票  <=========================
    /* We did not voted for a slave about this master for two
     * times the node timeout. This is not strictly needed for correctness
     * of the algorithm but makes the base case more linear. */
    if (mstime() - node->slaveof->voted_time < server.cluster_node_timeout * 2)
    {
        serverLog(LL_WARNING,
                "Failover auth denied to %.40s: "
                "can't vote about this master before %lld milliseconds",
                node->name,
                (long long) ((server.cluster_node_timeout*2)-
                             (mstime() - node->slaveof->voted_time)));
        return;
    }

    // slave发送的configEpoch小于master的configEpoch, 也不会发送auth ok <======================
    /* The slave requesting the vote must have a configEpoch for the claimed
     * slots that is >= the one of the masters currently serving the same
     * slots in the current configuration. */
    for (j = 0; j < CLUSTER_SLOTS; j++) {
        if (bitmapTestBit(claimed_slots, j) == 0) continue;
        if (server.cluster->slots[j] == NULL ||
            server.cluster->slots[j]->configEpoch <= requestConfigEpoch)
        {
            continue;
        }
        /* If we reached this point we found a slot that in our current slots
         * is served by a master with a greater configEpoch than the one claimed
         * by the slave requesting our vote. Refuse to vote for this slave. */
        serverLog(LL_WARNING,
                "Failover auth denied to %.40s: "
                "slot %d epoch (%llu) > reqEpoch (%llu)",
                node->name, j,
                (unsigned long long) server.cluster->slots[j]->configEpoch,
                (unsigned long long) requestConfigEpoch);
        return;
    }

    // 否则发送auth ok, 然后记录下时间
    /* We can vote for this slave. */
    server.cluster->lastVoteEpoch = server.cluster->currentEpoch;
    node->slaveof->voted_time = mstime();
    clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_FSYNC_CONFIG);
    clusterSendFailoverAuth(node);
    serverLog(LL_WARNING, "Failover auth granted to %.40s for epoch %llu",
        node->name, (unsigned long long) server.cluster->currentEpoch);
}

int clusterProcessPacket(clusterLink *link) {

    /* PING, PONG, MEET: process config information. */
    if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
        type == CLUSTERMSG_TYPE_MEET)
    {

        /* Check for role switch: slave -> master or master -> slave. */
        if (sender) {
            // 如果该instance的slaveof为空, 那么这个instance就是master
            if (!memcmp(hdr->slaveof,CLUSTER_NODE_NULL_NAME,
                sizeof(hdr->slaveof)))
            {
                /* Node is a master. */
                // 我们得知该instance是master
                // 更新本地信息中这个node的flag  <==================
                clusterSetNodeAsMaster(sender);
            } else {
                /* Node is a slave. */
            }
        }

        // sender是master, 同时dirty_slots为1表示slots的归属发生了变化 <=======================
        clusterNode *sender_master = NULL; /* Sender or its master if slave. */
        int dirty_slots = 0; /* Sender claimed slots don't match my view? */

        if (sender) {
            sender_master = nodeIsMaster(sender) ? sender : sender->slaveof;
            // 比较slots这个bitmap, 如果不一样, dirty_slots就为1 <=====================
            if (sender_master) {
                dirty_slots = memcmp(sender_master->slots,
                        hdr->myslots,sizeof(hdr->myslots)) != 0;
            }
        }

        /* 1) If the sender of the message is a master, and we detected that
         *    the set of slots it claims changed, scan the slots to see if we
         *    need to update our configuration. */
        // slots发生了变化, 更新slots信息!!!!!!!!!!!!!! <============================
        if (sender && nodeIsMaster(sender) && dirty_slots)
            clusterUpdateSlotsConfigWith(sender,senderConfigEpoch,hdr->myslots);

    }

}


/* Reconfigure the specified node 'n' as a master. This function is called when
 * a node that we believed to be a slave is now acting as master in order to
 * update the state of the node. */
// 这个函数只是单纯的把sender在本地的角色设置为master
// 并没有判断sender是否是myself的slave
// 降级处理是在下面  <========================
void clusterSetNodeAsMaster(clusterNode *n) {
    // 如果这个node本来就是master, 那么退出 <===================
    if (nodeIsMaster(n)) return;

    if (n->slaveof) {
        clusterNodeRemoveSlave(n->slaveof,n);
        if (n != myself) n->flags |= CLUSTER_NODE_MIGRATE_TO;
    }
    // 更新各种flag <===================
    n->flags &= ~CLUSTER_NODE_SLAVE;
    n->flags |= CLUSTER_NODE_MASTER;
    n->slaveof = NULL;

    // 保存配置
    /* Update config and state. */
    clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
                         CLUSTER_TODO_UPDATE_STATE);
}

// 更新slots信息
// 这里有包含了2中情况
void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoch, unsigned char *slots) {
    int j;
    clusterNode *curmaster, *newmaster = NULL;
    /* The dirty slots list is a list of slots for which we lose the ownership
     * while having still keys inside. This usually happens after a failover
     * or after a manual cluster reconfiguration operated by the admin.
     *
     * If the update message is not able to demote a master to slave (in this
     * case we'll resync with the master updating the whole key space), we
     * need to delete all the keys in the slots we lost ownership. */
    uint16_t dirty_slots[CLUSTER_SLOTS];
    int dirty_slots_count = 0;

    /* Here we set curmaster to this node or the node this node
     * replicates to if it's a slave. In the for loop we are
     * interested to check if slots are taken away from curmaster. */
    // 如果我们是master, 那么这个curmaster就是我们自己  <=======================
    // 否则这个curmaster就是我们的master
    // 而newmaster则会在下面赋值为sender的!!!! <===========================
    curmaster = nodeIsMaster(myself) ? myself : myself->slaveof;

    // 如果sender是自己, 那么就不需要更新了!!!!!!!!!!!!!
    if (sender == myself) {
        serverLog(LL_WARNING,"Discarding UPDATE message about myself.");
        return;
    }

    for (j = 0; j < CLUSTER_SLOTS; j++) {
        if (bitmapTestBit(slots,j)) {
            /* The slot is already bound to the sender of this message. */
            if (server.cluster->slots[j] == sender) continue;

            /* The slot is in importing state, it should be modified only
             * manually via redis-trib (example: a resharding is in progress
             * and the migrating side slot was already closed and is advertising
             * a new config. We still want the slot to be closed manually). */
            // 一个状态为importing的slot会在最后更新的, 所以这里没必要更新 <======================
            if (server.cluster->importing_slots_from[j]) continue;

            /* We rebind the slot to the new node claiming it if:
             * 1) The slot was unassigned or the new node claims it with a
             *    greater configEpoch.
             * 2) We are not currently importing the slot. */
            // 如果这个slot的configEpoch小于sender的configEpoch
            // 说明slave提升了epoch, 那么可以修改配置
            // 如果该slot没有人负责, 那么直接可以赋值  <===================================
            if (server.cluster->slots[j] == NULL ||
                server.cluster->slots[j]->configEpoch < senderConfigEpoch)
            {
                /* Was this slot mine, and still contains keys? Mark it as
                 * a dirty slot. */
                // 本来这个slot是我们(我们是master)负责, 同时sender不是我们自己
                // sender要抢走这个slot!!!!!!!!!!!!!! <=======================
                // 设置dirty_slot_count和dirty_slots数组
                // 如果我们是slave, 那么不会设置dirty_slots_count和dirty_slots的
                // 所以意味着是把我们这个master负责的slot移动另外一个
                // dirty_slots_count和dirty_slots是保存那些是本来我们负责的slot同时这些slot
                // 还有key, 我们需要删除这些key!!!!!!!!!!!!!! <==========================
                if (server.cluster->slots[j] == myself &&
                    countKeysInSlot(j) &&
                    sender != myself)
                {
                    dirty_slots[dirty_slots_count] = j;
                    dirty_slots_count++;
                }

                if (server.cluster->slots[j] == curmaster)
                    // 把newmaster赋值为sender
                    newmaster = sender;
                // 修改两个slots数组 <===============================
                // 同时减少curmaster->numslots的计数和增加sender->numslots的计数!!!!!!!!!!!!!!!!!!!!!!
                //
                clusterDelSlot(j);
                clusterAddSlot(sender,j);
                clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
                                     CLUSTER_TODO_UPDATE_STATE|
                                     CLUSTER_TODO_FSYNC_CONFIG);
            }
        }
    }

    /* If at least one slot was reassigned from a node to another node
     * with a greater configEpoch, it is possible that:
     * 1) We are a master left without slots. This means that we were
     *    failed over and we should turn into a replica of the new
     *    master.
     * 2) We are a slave and our master is left without slots. We need
     *    to replicate to the new slots owner. */
    // 如果curmaster->numslots为0, 表示没有需要清除的slot, 那么设置设置master
    // 因为我们在调用clusterDelSlot和和clusterAddSlot的时候已经操作过
    // 注意如果是一个master变为slave, 即使我们的dirty_slots_count不为0
    // 但是其实curmaster->numslots为0， 我们也不需要删除
    // 因为我们会变为slave, 和新的master同步slot!!!
    if (newmaster && curmaster->numslots == 0) {
        serverLog(LL_WARNING,
            "Configuration change detected. Reconfiguring myself "
            "as a replica of %.40s", sender->name);
        clusterSetMaster(sender);
        clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|
                             CLUSTER_TODO_UPDATE_STATE|
                             CLUSTER_TODO_FSYNC_CONFIG);
    } else if (dirty_slots_count) {
        // 我们是master, 但是我们只需要移动一部分slot到其他的master中
        // 所以我们不需要降级, 只是需要
        // 清除一下dirty_slos中的key!!!!!!! <==========================
        /* If we are here, we received an update message which removed
         * ownership for certain slots we still have keys about, but still
         * we are serving some slots, so this master node was not demoted to
         * a slave.
         *
         * In order to maintain a consistent state between keys and slots
         * we need to remove all the keys from the slots we lost. */
        for (j = 0; j < dirty_slots_count; j++)
            delKeysInSlot(dirty_slots[j]);
    }
}