[redis 源码走读] 主从数据复制 ②

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#define CONFIG_RUN_ID_SIZE 40

struct redisServer {
    ...
    list *slaves, *monitors;    /* List of slaves and MONITORs */
    ...
    /* Replication (master) */
    char replid[CONFIG_RUN_ID_SIZE+1];  /* My current replication ID. */
    char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
    long long master_repl_offset;   /* My current replication offset */
    long long master_repl_meaningful_offset; /* Offset minus latest PINGs. */
    long long second_replid_offset; /* Accept offsets up to this for replid2. */
    char *repl_backlog;             /* Replication backlog for partial syncs */
    long long repl_backlog_size;    /* Backlog circular buffer size */
    long long repl_backlog_histlen; /* Backlog actual data length */
    long long repl_backlog_idx;     /* Backlog circular buffer current offset,
                                       that is the next byte will'll write to.*/
    long long repl_backlog_off;     /* Replication "master offset" of first
    ...
    /* Replication (slave) */
    char *masterhost;               /* Hostname of master */
    int masterport;                 /* Port of master */
    client *master;     /* Client that is master for this slave */
    client *cached_master; /* Cached master to be reused for PSYNC. */
    int repl_state;          /* Replication status if the instance is a slave */
    ...
    char master_replid[CONFIG_RUN_ID_SIZE+1];  /* Master PSYNC runid. */
    long long master_initial_offset;           /* Master PSYNC offset. */
}

1
2
3
4
5
6
7
typedef struct client {
    ...
    long long read_reploff; /* Read replication offset if this is a master. */
    long long reploff;      /* Applied replication offset if this is a master. */
    char replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
void syncWithMaster(connection *conn) {
    ...
    if (server.repl_state == REPL_STATE_SEND_PSYNC) {
        // slave 发送 PSYNC 命令给 master。
        if (slaveTryPartialResynchronization(conn,0) == PSYNC_WRITE_ERROR) {
            err = sdsnew("Write error sending the PSYNC command.");
            goto write_error;
        }
        server.repl_state = REPL_STATE_RECEIVE_PSYNC;
        return;
    }
    ...
    /* slave 处理 PSYNC 命令的回复数据包。*/
    psync_result = slaveTryPartialResynchronization(conn,1);
    ...
    /* 增量复制。
     * slave 通过 readQueryFromClient 异步接收 master 增量数据。
     * 复制双方链接成功，slave 通过 replicationResurrectCachedMaster
     * 绑定 readQueryFromClient 异步接收复制数据。*/
    if (psync_result == PSYNC_CONTINUE) {
        ...
        return;
    }
    /* 全量复制。
     * slave 通过 readSyncBulkPayload() 异步接收 master 发送的 rdb 文件数据。*/
    if (connSetReadHandler(conn, readSyncBulkPayload) == C_ERR) {
        ...
        goto error;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
int slaveTryPartialResynchronization(connection *conn, int read_reply) {
    ...
    if (!read_reply) {
        /* 发送 */
        ...
        /* 复制双方有可能是断线重连，断线后，原来的链接 server.master 失效，被回收，
         * 为了重复利用原有数据，slave 会缓存 server.master 链接到 server.cached_master。*/
        if (server.cached_master) {
            psync_replid = server.cached_master->replid;
            // slave 发送当前的数据偏移量。
            snprintf(psync_offset, sizeof(psync_offset), "%lld", 
                     server.cached_master->reploff+1);
        } else {
            // slave 第一次链接 master，还没有 master 对应数据，所以用特殊符号标识。
            psync_replid = "?";
            memcpy(psync_offset,"-1",3);
        }

        // slave 发送 PSYNC 命令到 master。
        reply = sendSynchronousCommand(SYNC_CMD_WRITE, conn, "PSYNC",
                psync_replid, psync_offset, NULL);
        ...
    }

    /* 接收 */
    ...
    /* 全量复制。
     * slave 接收到 master 的回复：+FULLRESYNC <replid> <offset>
     * slave 先更新对应数据，后续 readSyncBulkPayload() 异步接收 master 发送的 rdb 文件数据，
     * 当 rdb 文件数据接收完毕，slave 重新加载对应的 rdb 文件，这样实现全量复制。*/
    if (!strncmp(reply,"+FULLRESYNC",11)) {
        char *replid = NULL, *offset = NULL;

        /* FULL RESYNC, parse the reply in order to extract the run id
         * and the replication offset. */
        replid = strchr(reply,' ');
        if (replid) {
            replid++;
            offset = strchr(replid,' ');
            if (offset) offset++;
        }
        if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) {
            // 全量复制出现错误。
            memset(server.master_replid,0,CONFIG_RUN_ID_SIZE+1);
        } else {
            // slave 更新 master 对应的 replid 和 offset。
            memcpy(server.master_replid, replid, offset-replid-1);
            server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
            server.master_initial_offset = strtoll(offset,NULL,10);
        }

        // 链接成功，清除旧的缓存链接数据，建立新的通信链接。
        replicationDiscardCachedMaster();
        sdsfree(reply);
        return PSYNC_FULLRESYNC;
    }

    /* 增量复制
     * slave 接收到 master 的回复：+CONTINUE <new repl ID> */
    if (!strncmp(reply,"+CONTINUE",9)) {
        /* 检查 master 是否有新的 <new repl ID>，
         * 有可能 redis 集群故障转移后，集群产生新的 master。*/
        char *start = reply+10;
        char *end = reply+9;
        while(end[0] != '\r' && end[0] != '\n' && end[0] != '\0') end++;
        if (end-start == CONFIG_RUN_ID_SIZE) {
            char new[CONFIG_RUN_ID_SIZE+1];
            memcpy(new,start,CONFIG_RUN_ID_SIZE);
            new[CONFIG_RUN_ID_SIZE] = '\0';

            if (strcmp(new, server.cached_master->replid)) {
                /* 缓存旧的 replid2 和 second_replid_offset。
                 * 因为当前 slave 可能有子服务 sub-slave，需要方便它们进行数据复制。*/
                memcpy(server.replid2,server.cached_master->replid,
                    sizeof(server.replid2));
                server.second_replid_offset = server.master_repl_offset+1;

                /* 更新 slave 当前的 replid。一般只有 master 才有自己独立的 replid。
                 * 所以 slave 的 replid 保存 master 的 replid。*/
                memcpy(server.replid,new,sizeof(server.replid));
                /* 更新 master client 对应的 replid。
                 * 因为增量复制是之前曾经链接成功的，后来断开链接了，
                 * 需要缓存断开的链接方，便后续重连操作。
                 * 所以会将原来 server.master，缓存到 server.cached_master。
                 * 当重连成功后 server.cached_master 会被清空。
                 * 详看 replicationResurrectCachedMaster()。*/
                memcpy(server.cached_master->replid,new,sizeof(server.replid));

                /* 如果当前 slave 有子服务 sub-slave，
                 * 那么断开子服务链接，让它们重新走 PSYNC 复制流程。*/
                disconnectSlaves();
            }
        }

        sdsfree(reply);
        // 链接成功，清除旧的缓存链接数据，建立新的通信链接。
        replicationResurrectCachedMaster(conn);

        // 如果积压缓冲区被清空，那么重新创建，如果有 sub-slave 方便数据复制。
        if (server.repl_backlog == NULL) createReplicationBacklog();
        return PSYNC_CONTINUE;
    }
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
void syncCommand(client *c) {
    ...
    if (!strcasecmp(c->argv[0]->ptr,"psync")) {
        /* 检查复制方式。
         * 增量复制，master 从缓冲区给 slave 发送增量数据。
         * 全量复制，master 创建 rdb 快照，异步发送 slave */
        if (masterTryPartialResynchronization(c) == C_OK) {
            server.stat_sync_partial_ok++;
            return;
        }
        ...
    }
    ...
    /* 全量复制。*/
    server.stat_sync_full++;

    /* 更新链接同步状态，建立 slave 数据复制链接。*/
    c->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
    if (server.repl_disable_tcp_nodelay)
        connDisableTcpNoDelay(c->conn); /* Non critical if it fails. */
    c->repldbfd = -1;
    c->flags |= CLIENT_SLAVE;
    listAddNodeTail(server.slaves,c);

    /* 创建复制的积压缓冲区对应数据。*/
    if (listLength(server.slaves) == 1 && server.repl_backlog == NULL) {
        /* When we create the backlog from scratch, we always use a new
         * replication ID and clear the ID2, since there is no valid
         * past history. */
        changeReplicationId();
        clearReplicationId2();
        createReplicationBacklog();
    }
    ...
    /* 如果当前没有子进程正在建立 rdb 文件快照。开始创建 rdb 文件快照 流程。
     * 否则放在时钟里进行定期检查处理，延时该流程。*/
    if (!hasActiveChildProcess()) {
        startBgsaveForReplication(c->slave_capa);
    }
    ...
    return;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
int masterTryPartialResynchronization(client *c) {
    ...
    // 检查 replid 是否一致，检查 slave 的数据偏移量是否超出缓存数据偏移量的范围。
    if (strcasecmp(master_replid, server.replid) &&
        (strcasecmp(master_replid, server.replid2) ||
         psync_offset > server.second_replid_offset)) {
        ...
        // 全量复制
        goto need_full_resync;
    }

    // 检查 slave 的数据偏移量，是否在主服务的数据缓冲区范围内。
    if (!server.repl_backlog ||
        psync_offset < server.repl_backlog_off ||
        psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen)) {
        ...
        // 全量复制
        goto need_full_resync;
    }
    ...

    // 增量复制
    if (c->slave_capa & SLAVE_CAPA_PSYNC2) {
        buflen = snprintf(buf,sizeof(buf),"+CONTINUE %s\r\n", server.replid);
    } else {
        buflen = snprintf(buf,sizeof(buf),"+CONTINUE\r\n");
    }

    // 发送 +CONTINUE 增量复制回包。注意这里是同步发送的，避免异步导致新的数据到来破坏当前同步场景。
    if (connWrite(c->conn,buf,buflen) != buflen) {
        freeClientAsync(c);
        return C_OK;
    }

    // 发送增量数据。
    psync_len = addReplyReplicationBacklog(c,psync_offset);
    ...
    return C_OK; /* The caller can return, no full resync needed. */

need_full_resync:
    return C_ERR;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// master 接收 slave 的 PSYNC 命令，检查 replid 是否一致。
int masterTryPartialResynchronization(client *c) {
    ...
    // 检查 replid 是否一致。
    if (strcasecmp(master_replid, server.replid) &&
        (strcasecmp(master_replid, server.replid2) ||
         psync_offset > server.second_replid_offset)) {
        ...
        goto need_full_resync;
    }
    ...
    // 增量复制
    return C_OK; /* The caller can return, no full resync needed. */

// 全量复制
need_full_resync:
    return C_ERR;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
struct redisServer {
    ...
    long long master_repl_offset;   /* My current replication offset */
    ...
}

// master 需要复制给 slave 的数据都会调用 feedReplicationBacklog，写入缓冲区并更新复制偏移量。
void feedReplicationBacklog(void *ptr, size_t len) {
    ...
    // master 复制偏移量
    server.master_repl_offset += len;
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
typedef struct client {
    ...
    long long reploff;      /* Applied replication offset if this is a master. */
    ...
}

/* 增量复制和正常链接下的数据复制。
 * slave 接收到 master 发送的数据，处理命令后，偏移量增加已处理数据数量
 * （因为 TCP 有可能因为粘包问题，接收数据不是完整的，所以不能全部处理完）。*/
int processCommandAndResetClient(client *c) {
    int deadclient = 0;
    server.current_client = c;
    if (processCommand(c) == C_OK) {
        if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
            // 接收数据后，追加已处理的数据总量到复制偏移量。
            c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
        }
        ...
    }
    ...
}

// 断线重连，slave 向 master 发送 PSYNC 命令，确认是增量复制，还是全量复制。
int slaveTryPartialResynchronization(connection *conn, int read_reply) {
    ...
    if (!read_reply) {
        ...
        if (server.cached_master) {
            psync_replid = server.cached_master->replid;
            // 断线重连 slave 发送保存的数据偏移量。
            snprintf(psync_offset,sizeof(psync_offset),"%lld",
                    server.cached_master->reploff+1);
        } else {
            // slave 第一次链接 master，还没有偏移量，所以用 -1 填充。
            psync_replid = "?";
            memcpy(psync_offset,"-1",3);
        }

        // slave 发送 PSYNC 命令到 master。
        reply = sendSynchronousCommand(SYNC_CMD_WRITE,conn,"PSYNC",
                psync_replid,psync_offset,NULL);
        ...
    }
    ...
    // slave 接收到 master 的回复，进行全量复制。
    if (!strncmp(reply,"+FULLRESYNC",11)) {
        ...
        memcpy(server.master_replid, replid, offset-replid-1);
        server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
        // 更新最新的数据偏移量。
        server.master_initial_offset = strtoll(offset,NULL,10);
        ...
        return PSYNC_FULLRESYNC;
    }
    ...
}

// 全量复制，slave 接收 master 发送的 rdb 文件数据，加载数据并初始化数据偏移量。
void readSyncBulkPayload(connection *conn) {
    ...
    replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db);
    ...
    // slave 更新这个值，因为 slave 有可能要复制数据到 sub-slave。
    server.master_repl_offset = server.master->reploff;
    ...
}

void replicationCreateMasterClient(connection *conn, int dbid) {
    ...
    // slave 更新数据偏移量到 reploff。
    server.master->reploff = server.master_initial_offset;
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
    ...
    /* Handle saving options that generate aux fields. */
    if (rsi) {
        // 当前 master 正在操作的 db。
        if (rdbSaveAuxFieldStrInt(rdb,"repl-stream-db",rsi->repl_stream_db) == -1)
            return -1;
        // master 的 replid。
        if (rdbSaveAuxFieldStrStr(rdb,"repl-id",server.replid) == -1)
            return -1;
        // master 的数据偏移量。
        if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",server.master_repl_offset) == -1)
            return -1;
    }
    ...
    return 1;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
void feedReplicationBacklog(void *ptr, size_t len) {
    unsigned char *p = ptr;

    // 客户端写数据，主服务，当前数据偏移量，实时增加对应的数据量。
    server.master_repl_offset += len;

    // 这是一个环形数据空间，repl_backlog_idx 是当前写数据位置。
    while(len) {
        // 往缓冲区填充数据，不能超过缓冲区范围。当填满缓冲区后，再缓冲区起始位置开始填充数据。
        size_t thislen = server.repl_backlog_size - server.repl_backlog_idx;
        if (thislen > len) thislen = len;
        memcpy(server.repl_backlog+server.repl_backlog_idx,p,thislen);
        server.repl_backlog_idx += thislen;
        if (server.repl_backlog_idx == server.repl_backlog_size)
            server.repl_backlog_idx = 0;
        len -= thislen;
        p += thislen;
        // 缓冲区实际填充数据长度。
        server.repl_backlog_histlen += thislen;
    }

    // 缓冲区实际填充数据长度，不能超过缓冲区大小。
    if (server.repl_backlog_histlen > server.repl_backlog_size)
        server.repl_backlog_histlen = server.repl_backlog_size;

    // 数据缓冲区起始位置内存数据，在 redis 整个内存的数据偏移量。
    server.repl_backlog_off = server.master_repl_offset -
                              server.repl_backlog_histlen + 1;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
// 根据 slave 的数据偏移量，master 回复数据增量。
long long addReplyReplicationBacklog(client *c, long long offset) {
    long long j, skip, len;

    // 如果数据缓冲区还没有数据，不需要回复。
    if (server.repl_backlog_histlen == 0) {
        serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero");
        return 0;
    }

    // 计算 slave 的偏移量在缓冲区的哪个位置上。
    skip = offset - server.repl_backlog_off;

    // j 是缓冲区 buffer 数据起始偏移位置，因为是环装结构，所以需要取模 repl_backlog_size。
    j = (server.repl_backlog_idx +
        (server.repl_backlog_size-server.repl_backlog_histlen)) %
        server.repl_backlog_size;

    // 数据增量在缓冲区起始偏移位置。
    j = (j + skip) % server.repl_backlog_size;

    // 增量数据长度。
    len = server.repl_backlog_histlen - skip;

    // 发送缓冲区增量数据。
    while(len) {
        long long thislen =
            ((server.repl_backlog_size - j) < len) ?
            (server.repl_backlog_size - j) : len;

        addReplySds(c,sdsnewlen(server.repl_backlog + j, thislen));
        len -= thislen;
        j = 0;
    }
    return server.repl_backlog_histlen - skip;
}