Redis replication 中的探活 redis

redis 在运行过程中需要一些探活机制来保证对另一端的感知能力。
slave 重建 replication 阶段 【Redis replication 中的探活】当由于网络或其他原因导致主从 link 断开后，slave 会尝试重建 replication 。在这个过程中，slave 的复制状态机 repl_state 变量会经过一系列流传，最终为 REPL_STATE_CONNECTED 状态。
repl_state 在很多状态的停留时间都有超时设定，以便出错后尽早是否资源。server.repl_transfer_lastio 变量起到了计时器的作用，它记录了slave 上一次从 master 进行 io 交互（即读写事件）的时间。
REPL_STATE_CONNECTING 超时
在 REPL_STATE_CONNECTING 阶段，slave 会主从 connet master，该过程使用了非阻塞 IO，在replicationCron 函数里周期性检查是否超时。

/* Non blocking connection timeout? */ if (server.masterhost && (server.repl_state == REPL_STATE_CONNECTING || slaveIsInHandshakeState()) && (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) { serverLog(LL_WARNING,"Timeout connecting to the MASTER..."); cancelReplicationHandshake(); }

REPL_STATE_TRANSFER 超时
在 REPL_STATE_TRANSFER 阶段，slave 会从 master 接收 rdb 文件，通常不会在一次 read 里完成，所以需要在 replicationCron 函数里周期性检查该过程是否超时。

/* Bulk transfer I/O timeout? */ if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER && (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) { cancelReplicationHandshake(); }

如果 rdb 数据量过大，可能需要做一些额外处理，下面进行说明。
1）如果 master dump rdb 时间过长，在 slave 侧，master client 迟迟没有发来数据，回调函数 readSyncBulkPayload 不会触发，那么 repl_transfer_lastio 变量始终得不到刷新，会在超时检查中 cancelReplicationHandshake，导致此次主从同步的失败。因此，master 对于正在等待接收 rdb 的 slave，会周期性地发送 \n 做探活。

// master 处理逻辑 listRewind(server.slaves,&li); while((ln = listNext(&li))) { client *slave = ln->value; int is_presync = (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START || (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && server.rdb_child_type != RDB_CHILD_TYPE_SOCKET)); if (is_presync) { if (write(slave->fd, "\n", 1) == -1) { /* Don't worry about socket errors, it's just a ping. */ } } }

2）在 slave 接收到 rdb 文件后，如果 rdb 过大，加载过程会 hang 住，repl_ack_time 变量得不到刷新，让 master 以为 slave 挂掉了，因此，在rdbLoadRio 时定期发送 \n 做探活。

// slave 处理逻辑 if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER) replicationSendNewlineToMaster(); ...void replicationSendNewlineToMaster(void) { static time_t newline_sent; if (time(NULL) != newline_sent) { newline_sent = time(NULL); if (write(server.repl_transfer_s,"\n",1) == -1) { /* Pinging back in this stage is best-effort. */ } } }

master 收到 \n 后在 processInlineBuffer 函数中刷新 client->repl_ack_time 时间，防止 client 检测超时。

// master 处理逻辑 if (querylen == 0 && getClientType(c) == CLIENT_TYPE_SLAVE) c->repl_ack_time = server.unixtime;

master-slave 正常 replication 阶段当主从正常进行 replication 时，master 会向 slave 持续发送 commands stream，以维持主从 dataset 状态的一致。
master 与 slave 在此过程中会发送一些探活包，以感知主从复制的状态。
slave 探活
slave 的探活依赖于 client.lastinteraction 变量，它记录了本实例上一次从该 client fd 读到数据的时间，在 read 事件回调函数 readQueryFromClient 中更新。
我们知道，在 master 与 slave 看来，对方都是一种携带特殊 flag 的 client。
在 replicationCron 函数里会检查 master 是否正常。若异常，则释放 master client。

// slave 处理逻辑 if (server.masterhost && server.repl_state == REPL_STATE_CONNECTED && (time(NULL)-server.master->lastinteraction) > server.repl_timeout) { serverLog(LL_WARNING,"MASTER timeout: no data nor PING received..."); freeClient(server.master); }

client.lastinteraction 变量的更新需要依赖于 master 的处理逻辑。
对于自己所认识的 slave 节点，master 会周期性地发送 ping 命令，这个周期由配置参数 repl-ping-replica-period 决定，单位为 s。

// master 处理逻辑 if ((replication_cron_loops % server.repl_ping_slave_period) == 0 && listLength(server.slaves)) { int manual_failover_in_progress = server.cluster_enabled && server.cluster->mf_end && clientsArePaused(); // 跳过处于 mf 过程中的 slave if (!manual_failover_in_progress) { ping_argv[0] = createStringObject("PING",4); replicationFeedSlaves(server.slaves, server.slaveseldb, ping_argv, 1); decrRefCount(ping_argv[0]); } }

slave 收到 pong 回复后刷新 lastinteraction 值，并每秒进行超时检查。
master 探活
master 的探活依赖于 client.repl_ack_time 变量，它记录了上一次收到 slave 的 REPLCONF 命令的时间。
在 replicationCron 函数里会检查 slave 是否正常。若异常，则释放 slave client。

// master 处理逻辑 /* Disconnect timedout slaves. */ if (listLength(server.slaves)) { listIter li; listNode *ln; listRewind(server.slaves,&li); while((ln = listNext(&li))) { client *slave = ln->value; // 注意：这里是 SLAVE_STATE_ONLINE 状态的 slave， // 必然将 slave fd 挂上了写事件回调。 if (slave->replstate != SLAVE_STATE_ONLINE) continue; if (slave->flags & CLIENT_PRE_PSYNC) continue; if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) { serverLog(LL_WARNING, "Disconnecting timedout replica: %s", replicationGetSlaveName(slave)); freeClient(slave); } } }

为什么 master 的探活不像 slave 那样通过 ping 完成呢？
这是因为 master 需要知道它所认识的每个 slave 的 repl offset，因此，slave 每秒发送 REPLCONF ACK （offset 值取自 client->reploff 变量，在 slave processCommand 后更新），这同样达到了 ping 的目的。