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没有实际意义, // 因为无法存储一份完整数据。 voidfeedReplicationBacklog(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; }
6. addReplyReplicationBacklog-读repl_backlog
当主节点判断可部分复制时,会记录如下日志:
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"); return0; } 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; }