REdis主从复制之repl_backlog

struct redisServer {
/* My current replication offset */
long long master_repl_offset;
/* Accept offsets up to this for replid2. */
long long second_replid_offset;
/* Replication backlog for partial syncs */
char *repl_backlog; // 环形缓冲复制队列
/* Backlog circular buffer size */
long long repl_backlog_size; // 环形缓冲复制队列容量
/* Backlog actual data length */
long long repl_backlog_histlen; // 环形缓冲复制队列已用大小（影响是否能部分复制）
/* Backlog circular buffer current offset,
that is the next byte will'll write to.*/
// 实际上谈不上空闲，因为总是环绕覆盖写，
// 理解为最新数据的截止位置更为合适，更新的数据总是从这里开始写入到repl_backlog中。
long long repl_backlog_idx; // 环形缓冲复制队列空闲起始位置（写从这里开始）
/* Replication "master offset" of first
byte in the replication backlog buffer.*/
long long repl_backlog_off; // 数据在环形缓冲复制队列的起始位置（读从这里开始）
/* Time without slaves after the backlog
gets released. */
time_t repl_backlog_time_limit; // 环形缓冲复制队列生存时长
/* We have no slaves since that time.
Only valid if server.slaves len is 0. */
time_t repl_no_slaves_since; // 无可用从节点的发生时间
/* Min number of slaves to write. */
int repl_min_slaves_to_write; // 最小需写的从节点数
/* Max lag of  slaves to write. */
int repl_min_slaves_max_lag;
/* Number of slaves with lag <= max_lag. */
int repl_good_slaves_count;
/* Send RDB to slaves sockets directly. */
int repl_diskless_sync; // 不落磁盘（无盘）往从节点发送RDB（全量复制）
/* Delay to start a diskless repl BGSAVE. */
// 无盘复制时，延迟指定的时长，以等待更多的从节点
int repl_diskless_sync_delay;
};

/* Add data to the replication backlog.
* This function also increments the global replication offset stored at
* server.master_repl_offset, because there is no case where we want to feed
* the backlog without incrementing the offset. */
// 主要被replicationFeedSlaves调用
// 写len长的数据ptr到repl_backlog中
// repl_backlog是一个环形buffer，不存在溢出的问题，策略是最新数据覆盖最老数据。
// 如果参数len大于repl_backlog_size，
// 则repl_backlog没有实际意义，
// 因为无法存储一份完整数据。
void feedReplicationBacklog(void *ptr, size_t len) {
unsigned char *p = ptr;
server.master_repl_offset += len;
/* This is a circular buffer, so write as much data we can at every
* iteration and rewind the "idx" index if we reach the limit. */
// 因为repl_backlog是环形buffer，
// 剩余的空间可能容纳不了len长的数据，
// 当不够时，就需要环绕从头开始写，
// 因此这里需while循环。
while(len) {
// repl_backlog_size为repl_backlog的容量大小，
// 由配置项决定repl_backlog_size值决定，
// repl_backlog_idx是repl_backlog空闲区域的起始位置，
// 这两个值相减得到repl_backlog可用大小。
size_t thislen = server.repl_backlog_size - server.repl_backlog_idx;
// 如果thislen大于len，则表示足够容纳
if (thislen > len) thislen = len;
memcpy(server.repl_backlog+server.repl_backlog_idx,p,thislen);
// 空闲位置往后挪动
server.repl_backlog_idx += thislen;
// 如果空闲位置达到容量大小，则环绕回去从0开始
if (server.repl_backlog_idx == server.repl_backlog_size)
server.repl_backlog_idx = 0;
len -= thislen;
p += thislen;
// repl_backlog_histlen记录了repl_backlog中的数据大小
server.repl_backlog_histlen += thislen;
}
// 修正存储在repl_backlog中的数据大小，
// 它不可能超过repl_backlog_size值。
if (server.repl_backlog_histlen > server.repl_backlog_size)
server.repl_backlog_histlen = server.repl_backlog_size;
/* Set the offset of the first byte we have in the backlog. */
server.repl_backlog_off = server.master_repl_offset -
server.repl_backlog_histlen + 1;
}

Partial resynchronization request from %s accepted. Sending %lld bytes of backlog starting from offset %lld.

/* Feed the slave 'c' with the replication backlog starting from the
* specified 'offset' up to the end of the backlog. */
// 被masterTryPartialResynchronization调用
// 而masterTryPartialResynchronization被syncCommand调用（对应命令PSYNC）。
// 从repl_backlog取数据发给slave，
// 数据的开始位置由offset指定。
long long addReplyReplicationBacklog(client *c, long long offset) {
long long j, skip, len;
serverLog(LL_DEBUG, "[PSYNC] Replica request offset: %lld", offset);
// repl_backlog_histlen为0，
// 表示repl_backlog中无数据。
if (server.repl_backlog_histlen == 0) {
serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero");
return 0;
}
serverLog(LL_DEBUG, "[PSYNC] Backlog size: %lld",
server.repl_backlog_size);
serverLog(LL_DEBUG, "[PSYNC] First byte: %lld",
server.repl_backlog_off);
serverLog(LL_DEBUG, "[PSYNC] History len: %lld",
server.repl_backlog_histlen);
serverLog(LL_DEBUG, "[PSYNC] Current index: %lld",
server.repl_backlog_idx);
/* Compute the amount of bytes we need to discard. */
skip = offset - server.repl_backlog_off;
serverLog(LL_DEBUG, "[PSYNC] Skipping: %lld", skip);
/* Point j to the oldest byte, that is actually our
* server.repl_backlog_off byte. */
j = (server.repl_backlog_idx +
(server.repl_backlog_size-server.repl_backlog_histlen)) %
server.repl_backlog_size;
serverLog(LL_DEBUG, "[PSYNC] Index of first byte: %lld", j);
/* Discard the amount of data to seek to the specified 'offset'. */
j = (j + skip) % server.repl_backlog_size;
/* Feed slave with data. Since it is a circular buffer we have to
* split the reply in two parts if we are cross-boundary. */
len = server.repl_backlog_histlen - skip;
serverLog(LL_DEBUG, "[PSYNC] Reply total length: %lld", len);
while(len) {
long long thislen =
((server.repl_backlog_size - j) < len) ?
(server.repl_backlog_size - j) : len;
serverLog(LL_DEBUG, "[PSYNC] addReply() length: %lld", thislen);
addReplySds(c,sdsnewlen(server.repl_backlog + j, thislen));
len -= thislen;
j = 0;
}
return server.repl_backlog_histlen - skip;
}