Redis Sentinel原理与实现 (中)

int main(int argc, char **argv) {
 ...
    server.sentinel_mode = checkForSentinelMode(argc,argv);
 ...
 if (server.sentinel_mode) {
        initSentinelConfig();
        initSentinel();
    }
 ...
 if (!server.sentinel_mode) {
        ...
    } else {
        sentinelIsRunning();
    }
 ...
    aeMain(server.el);
    aeDeleteEventLoop(server.el);
 return 0;
}

void initSentinel(void) {
 unsigned int j;
 /* Remove usual Redis commands from the command table, then just add
     * the SENTINEL command. */
    dictEmpty(server.commands,NULL);
 for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) {
 int retval;
 struct redisCommand *cmd = sentinelcmds+j;
        retval = dictAdd(server.commands, sdsnew(cmd->name), cmd);
        serverAssert(retval == DICT_OK);
    }
 /* Initialize various data structures. */
 sentinel.current_epoch = 0;
 sentinel.masters = dictCreate(&instancesDictType,NULL);
 sentinel.tilt = 0;
 sentinel.tilt_start_time = 0;
 sentinel.previous_time = mstime();
 sentinel.running_scripts = 0;
 sentinel.scripts_queue = listCreate();
 sentinel.announce_ip = NULL;
 sentinel.announce_port = 0;
 sentinel.simfailure_flags = SENTINEL_SIMFAILURE_NONE;
    memset(sentinel.myid,0,sizeof(sentinel.myid));
}

int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 ...
    run_with_period(100) {
 if (server.sentinel_mode) sentinelTimer();
    }
 ...
}

void sentinelTimer(void) {
    sentinelCheckTiltCondition();
    sentinelHandleDictOfRedisInstances(sentinel.masters);
    sentinelRunPendingScripts();
    sentinelCollectTerminatedScripts();
    sentinelKillTimedoutScripts();
 /* We continuously change the frequency of the Redis "timer interrupt"
     * in order to desynchronize every Sentinel from every other.
     * This non-determinism avoids that Sentinels started at the same time
     * exactly continue to stay synchronized asking to be voted at the
     * same time again and again (resulting in nobody likely winning the
     * election because of split brain voting). */
 server.hz = CONFIG_DEFAULT_HZ + rand() % CONFIG_DEFAULT_HZ;
}

void sentinelCheckTiltCondition(void) {
 mstime_t now = mstime();
 mstime_t delta = now - sentinel.previous_time;
 if (delta < 0 || delta > SENTINEL_TILT_TRIGGER) { // 2000
 sentinel.tilt = 1;
 sentinel.tilt_start_time = mstime();
        sentinelEvent(LL_WARNING,"+tilt",NULL,"#tilt mode entered");
    }
 sentinel.previous_time = mstime();
}

void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
 /* ========== MONITORING HALF ============ */
 /* Every kind of instance */
    sentinelReconnectInstance(ri);  // 建立与其他服务器的连接
    sentinelSendPeriodicCommands(ri); // 发送info/ping/hello消息
 /* ============== ACTING HALF ============= */
 /* We don't proceed with the acting half if we are in TILT mode.
     * TILT happens when we find something odd with the time, like a
     * sudden change in the clock. */
 if (sentinel.tilt) {
 if (mstime()-sentinel.tilt_start_time < SENTINEL_TILT_PERIOD) return;
 sentinel.tilt = 0;
        sentinelEvent(LL_WARNING,"-tilt",NULL,"#tilt mode exited");
    }
 /* Every kind of instance */
    sentinelCheckSubjectivelyDown(ri);
 /* Masters and slaves */
 if (ri->flags & (SRI_MASTER|SRI_SLAVE)) {
 /* Nothing so far. */
    }
 /* Only masters */
 if (ri->flags & SRI_MASTER) { // 如果是master服务器
        sentinelCheckObjectivelyDown(ri);
 if (sentinelStartFailoverIfNeeded(ri))
            sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED);
        sentinelFailoverStateMachine(ri);
        sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS);
    }
}

typedef struct sentinelRedisInstance {
 ...
 /* Master specific. */
    dict *sentinels; /* Other sentinels monitoring the same master. */
 dict *slaves;       /* Slaves for this master instance. */
 ...
 struct sentinelRedisInstance *master; /* Master instance if it's slave. */
 ...
} sentinelRedisInstance;

void sentinelReconnectInstance(sentinelRedisInstance *ri) {
 ...
 if (link->cc == NULL) {
        link->cc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
 if (link->cc->err) {
            ...
        } else {
            ...
            redisAsyncSetConnectCallback(link->cc,
                    sentinelLinkEstablishedCallback);
            redisAsyncSetDisconnectCallback(link->cc,
                    sentinelDisconnectCallback);
            ...
            sentinelSendPing(ri);
        }
    }
 /* Pub / Sub */
 if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && link->pc == NULL) {
 link->pc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
 if (link->pc->err) {
 ...
        } else {
 ...
            redisAsyncSetConnectCallback(link->pc,
                    sentinelLinkEstablishedCallback);
            redisAsyncSetDisconnectCallback(link->pc,
 sentinelDisconnectCallback);
 ...
            retval = redisAsyncCommand(link->pc,
                sentinelReceiveHelloMessages, ri, "SUBSCRIBE %s",
                    SENTINEL_HELLO_CHANNEL);
 ...
        }
    }
 ...
}

void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) {
 mstime_t elapsed = 0;
 if (ri->link->act_ping_time)
        elapsed = mstime() - ri->link->act_ping_time;
 else if (ri->link->disconnected)
        elapsed = mstime() - ri->link->last_avail_time;
 ...
 if (elapsed > ri->down_after_period ||
        (ri->flags & SRI_MASTER &&
         ri->role_reported == SRI_SLAVE &&
         mstime() - ri->role_reported_time >
          (ri->down_after_period+SENTINEL_INFO_PERIOD*2)))
    {
 /* Is subjectively down */
 if ((ri->flags & SRI_S_DOWN) == 0) {
            sentinelEvent(LL_WARNING,"+sdown",ri,"%@");
            ri->s_down_since_time = mstime();
            ri->flags |= SRI_S_DOWN;
        }
    } else {
 /* Is subjectively up */
 if (ri->flags & SRI_S_DOWN) {
            sentinelEvent(LL_WARNING,"-sdown",ri,"%@");
            ri->flags &= ~(SRI_S_DOWN|SRI_SCRIPT_KILL_SENT);
        }
    }
}

void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) {
 dictIterator *di;
 dictEntry *de;
 unsigned int quorum = 0, odown = 0;
 if (master->flags & SRI_S_DOWN) {
 /* Is down for enough sentinels? */
        quorum = 1; /* the current sentinel. */
 /* Count all the other sentinels. */
        di = dictGetIterator(master->sentinels);
 while((de = dictNext(di)) != NULL) {
 sentinelRedisInstance *ri = dictGetVal(de);
 if (ri->flags & SRI_MASTER_DOWN) quorum++;
        }
        dictReleaseIterator(di);
 if (quorum >= master->quorum) odown = 1;
    }
 /* Set the flag accordingly to the outcome. */
 if (odown) {
 if ((master->flags & SRI_O_DOWN) == 0) {
            sentinelEvent(LL_WARNING,"+odown",master,"%@ #quorum %d/%d",
                quorum, master->quorum);
            master->flags |= SRI_O_DOWN;
            master->o_down_since_time = mstime();
        }
    } else {
 if (master->flags & SRI_O_DOWN) {
            sentinelEvent(LL_WARNING,"-odown",master,"%@");
            master->flags &= ~SRI_O_DOWN;
        }
    }
}

void sentinelStartFailover(sentinelRedisInstance *master) {
    master->failover_state = SENTINEL_FAILOVER_STATE_WAIT_START;
    master->flags |= SRI_FAILOVER_IN_PROGRESS;
    master->failover_epoch = ++sentinel.current_epoch;
    sentinelEvent(LL_WARNING,"+new-epoch",master,"%llu",
        (unsigned long long) sentinel.current_epoch);
    sentinelEvent(LL_WARNING,"+try-failover",master,"%@");
    master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC;
    master->failover_state_change_time = mstime();
}