redis源码系列--（三）--主从同步流程
·
主从同步：主从结构的redis中当master阻塞或者崩溃时，slave不会自动切换成master
配置：
master不用管
slave要配置：1：修改slave.conf的port，因为我们是在同一台机器上同时启动；2：修改dir
port 6379
dir ./master #. 表示redis-server可执行文件所在的目录，编译后的redis-server可执行文件是放在src目录下，所以需要在src目录下新建master和slave文件夹
pidfile /var/run/redis_7379.pid
#repl-diskless-load disabled #三选一，不同的选择会走不同的流程
#repl-diskless-load on-empty-db
#repl-diskless-load swapdb
断点：根据日志，然后去搜索源码，然后打断点；百度：slave是发送replicaof命令给maste，所以根据源码命名习惯尝试搜索replicaofCommand函数，还真有，于是又一个断点打好了
复制缓冲区与复制积压缓冲区区别
复制缓冲区是指主节点将待复制的数据先写入缓冲区，等待从节点连接后再将数据发送给从节点。
复制积压缓冲区是指从节点在与主节点断开连接时，为了避免数据丢失，将待复制的数据先存储在缓冲区中，等待连接恢复后再将数据发送给主节点。
流程重点：同步流程
1-1：slave-fullsync流程
同步分为fullsync和continuesync，fullsync表示全量复制，而continuesync则是相当于增量复制。本部分主要讲slave的fullsync流程，continuesync略（因为今天虚拟机又崩了，又出现了齐齐挂怪的bug，一直没解决，导致心态崩了，所以就没分析continuesync了 20241030 17:20）。slave同步流程就两大块：1：协商；2：同步。协商对应的read_event_handler为syncWithMaster，就是协商各种参数，比如eof、是否支持psync2、ip-addr等，协商完这些参数后就是协商是fullsync还是continuesync了,协商的过程就是一个状态机，如果当前状态是a则实行操作1，然后进入下一个状态，如果是状态b则实行操作2，然后进入下一个状态，如此直到协商完成，协商完成后就会重新设置read_event_handler，这样下次触发读事件就是进入数据同步阶段了；continuesync同步阶段对应的handler为readQueryFromClient；fullsync同步阶段对应的handler为readSyncBulkPayload，fullsync就是master发数据给slave，slave收数据，这个fullsync同步又分为两种：diskLess和！diskless，前者就是不刷盘，直接从网络读到内存，然后更新内存数据库，然后再异步aofrewrite保存到磁盘，而后者则是创建临时文件，然后从网络读取master发来的rdb文件，然后刷盘，然后再加载到内存并更新内存数据库，然后再rewrite,就是说diskless和!diskless区别在于diskless省去了一次刷盘和读盘操作。
server.main
server.el=ae.aeCreateEventLoop
ae.aeMain(server.el)
while(!el.stop):
ae.aeProcessEvents
socket.connSocketEventHandler
connhelpers.callHandler
1：协商阶段
replication.syncWithMaster #同步事件协商阶段对应的handler
......
if server.repl_state = REPL_STATE_CONNECTED: #协商阶段最开始的状态为REPL_STATE_CONNECTED
...... #这个状态表示与master已建立连接
if server.repl_state == 状态x : #如果是状态x就执行a操作
replication.receiveSynchronousResponse #从io缓冲区读取master对slave在状态x-1中发过去的请求的响应
......
replication.sendCommand #向master发送下一个协商命令，协商下一项
server.repl_state = 状态x+1 #执行完操作后进入下一个状态x+1
return #本次调用结束，等待下一次调用，下一次进来的时候就是状态x+1了
if server.repl_state == 状态x+1:
......
if server.repl_state==REPL_STATE_SEND_PSYNC:
replication.slaveTryPartialResynchronization(read_replay=0) #！！！向master发送psync命令
replication.sendCommand(conn,"PSYNC",psync_replid,psync_offset,NULL); #！！！发送psync命令后面会详解
#！！！slave会根据master的响应来决定是部分还是全量同步
...中间还有很多状态，略...
psync_result=replication.slaveTryPartialResynchronization (read_only=1) #读取master对slave发过去的psync命令的响应并
#据此判断是启用fullsync还是continuesync
#备注：这个函数会根据read_only来判断是发送psync命令给master
#还是read response
reply=replication.receiveSynchronousResponse #此处read_only=1，前面已经发送过psync了，所以这里就是读取响应
#这个函数下面会详解
if (!strncmp(reply,"+FULLRESYNC",11)): #！！！表示master要求进行FULLSYNC
return PSYNC_FULLSYNC
if (!strncmp(reply,"+CONTINUE",9)): #！！！表示master此次要求进行CONTINUESYNC
return PSYNC_CONTINUE
if psync_result == PSYNC_CONTINUE: #如果启用的是continuesync直接返回了。
return #需补充continuesync过程 2024/10/30
#！！！如果是部分同步，那么同步流程到此就直接结束了
#！！！即下面的fullsync是FULLSYNC流程，不是用于处理部分同步的
#！！！部分同步流程在
#！！！slaveTryPartialResynchronization函数中就处理完了
#后文详解
if psync_result= PSYNC_FULLSYNC: #如果启用的是fullsync
if !replication.useDiskLessLoad(): #并且如果选择的同步方式是刷盘：
#先创建临时文件接收完整rdb文件，然后刷盘，然后再从磁盘加载
server.repl_transfer_fd=filefcntl.open #打开一个临时文件接收数据
connection.connSetReadHandler(conn, readSyncBulkPayload) #fullsync需要进一步处理，下一步就是fullsync同步过程
#所以重置read_event对应的handler
server.repl_state = REPL_STATE_TRANSFER; #标记下一个状态为数据传输阶段即fullsync同步阶段
2：fullsync同步阶段（continuesync不是这个同步流程） #！！！此处是fullsync
replication.readSyncBulkPayload #fullsync同步阶段，也类似一个状态机
connection.connSyncReadLine #首先从网络io读取请求开头字段。
#同步时需要确定要传输多少字节，有两种方式， $<count>和$EOF
#前者指定了rdb文件大小，后者则通过EOF标志来表明rdb文件已传输结束
#所以这里就读取请求开头几个字段来判断使用的是哪一种
if !use_diskless_load #如果不是使用无盘方式即选择的是刷盘方式，
#那么需要把读取的数据先保存到磁盘
readlen=xxx #获取要读取的字节数（8.0默认是EOF）
connection.connRead(conn,buf,readlen); #从网络io读取readLen个字节，如果是$<count>方式，
#则readLen为rdb文件大小反之则是当前所用buf的大小
unistd.write(server.repl_transfer_fd,buf,nread) #把读取的数据写入临时文件
if (server.repl_transfer_read >=
server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC) #如果未刷盘的数据超过了阈值，就强制刷盘一下
#read表示已经读取的字节数,off表示已经刷盘的字节数，
#MAX表示最多允许这么多字节未刷盘
rdb_fsync_range #强制刷盘，以免突然崩溃导致数据丢失太多
if !eof_reached: #如果rdb文件还没有传输结束，本次就返回，等待下次
return #所以说同步阶段也类似于一个状态机，如果还没有收到整个文件就先返回
#直到读完整个文件，流程才会继续往下
#读了$<count>个字节或者读到EOF了就表明文件读完了
#注意：此处是选择刷盘方式才会读完整个文件才往下走 ，
#无盘方式则不读直接往下走
aof.stopAppendOnly #走到这里表明选择的是无盘方式或者选择的是刷盘方式但已经收到并刷盘了整个rdb文件
#关闭aof，以免aof操作与同步过程冲突
aof.flushAppendOnlyFile #先刷盘再关闭
aof.killAppendOnlyChild #如果有子进程就杀掉，没有就什么也不做
#？？？？？疑问：貌似是杀掉rewrite进程，待完善
rdb.killRDBChild #也要杀掉background rdb进程以免冲突
if (use_diskless_load && server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB): #如果使用无盘方式并且以替换的方式更新db
#REPL_DISKLESS_LOAD_SWAPDB表示允许在同步期间继续处理读
#只不过读到的是旧数据
#！！！redis中slave只能处理client读请求不能处理写请求
replication.disklessLoadInitTempDb #在内存里创建一个临时的db，master同步的数据会先写到这个里面
#注意：此时无盘方式还没有开始读取数据，别搞混了
else
replication.replicationAttachToNewMaster #fullsync表示slave从master收到了一个全新的数据集，
#此时slave是无法和slave的slave进行增量同步的，
#所以需要通知slave的slave重新和slave进行fsync
#fullsync过程中slave会用master的repliid来更换自己的replid
#这样当slave的slave重新和slave进行sync的时候slave会发现replid不匹配
#所以slave就会和slave的slave进行fullsync
#注意，这个else是刷盘方式时要调用的replicationAttachToNewMaster
#无盘方式还要进行一些处理才会调用replicationAttachToNewMaster
networking.disconnectSlaves #断开slave与所有slave的slave之间的连接
#连接断开后，slave的slave会自动重连slave
#重新连接时首先肯定会尝试与slave进行同步
#此时就会触发fullsync，因为slave数据集变了
connection.connSetReadHandler(conn, NULL) #先关闭slave-master连接上的read事件
serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");
if use_diskless_load: #开始处理无盘方式：先阻塞读取master发来的整个rdb到内存然后替换
if server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB{： #如果开启swap则新数据存放到新的db和function_ctx中
/* Async loading means we continue serving read commands during full resync, and
* "swap" the new db with the old db only when loading is done.
* It is enabled only on SWAPDB diskless replication when master replication ID hasn't changed,
* because in that state the old content of the db represents a different point in time of the same
* data set we're currently receiving from the master. */
if memcmp(server.replid, server.master_replid, CONFIG_RUN_ID_SIZE) ==0 #说明本次主从同步是基于相同数据集的，可以异步加载
#反之不合适，只能fullsync
#！！！即部分同步-异步加载；fullsync-同步加载
asyncLoading = 1 #！！！asyncLoading表示本次是continue sync
#！！！我们说过：部分同步是把同步任务分散到了每次事件循环
#！！！意思就是处理事件循环中的事件时正处于loading状态
#！！！意思就是redis在processEventsWhileBlock中会嵌套处理事件循环中的事件
#！！！所以redis会用server.loading和server.async_loading来标记当前状态
#！！！如果loading=0表明此时没有进行加载rdb，所以普通client的命令畅通无阻
#！！！如果loading=1&&async_loading=0,表明在loding结束前不允许执行普通命令
#！！！如果loading=1&&async_loading=1表明允许在loding结束前执行部分普通命令
#！！！命令是有属性的，有些命令可以在loading=x&&async_loading=y时执行，有些不可以
#！！！不允许执行命令：就是说收到client发来的命令，
#！！！然后直接发送reject响应，告诉client当前不允许执行普通命令
#！！！也就是说：当rdb加载完成以后会重置loading=0
#！！！也就是说只有loading被设置为0才表示同步真正结束了
#！！！也就是说如果在processEventsWhileBlock中会嵌套处理事件循环中检测到loading=1
#！！！那么如上述所属，有些命令不允许执行
#！！！注意：同步操作是在主线程中执行，
#！！！也就是说从master读取同步rdb数据以及从磁盘加载rdb这两个操作都是在主线程中执行
#！！！因为redis是单线程，所以在这些操作完成之前不会处理其他事件
#！！！但是和lua类似，redis会执行一段代码后终端当前代码去处理事件循环
#REPL_DISKLESS_LOAD_SWAPDDB不会阻塞主线程对外提供读服务
#若不是，则会阻止主线程对外提供读服务
#swap方式就是先把数据写到新的临时的rdb，
#此时旧rdb和主线程可以继续对外服务，
#当新数据万全准备好的时候就直接一个swap交换新旧db指针
#这样一下就实现了替换，在此之后就是对外提供新数据了
dbarray = diskless_load_tempDb; #swap方式，则写到新的临时的rdb和ctx
functions_lib_ctx = temp_functions_lib_ctx;
} else { #如果没有开启swap则直接在原地操作。server.db是当前使用的db
dbarray = server.db;
functions_lib_ctx = functionsLibCtxGetCurrent(); #function_lib_ctx是当前使用的function_lib_ctx
functionsLibCtxClear(functions_lib_ctx);
}
rdb.loadingSetFlags(NULL, server.repl_transfer_size, asyncLoading) #！！！标记server同步状态，即loading和asyncLoading
#！！！表示server进入loading状态
server.loading = 1;
if (async == 1)
server.async_loading = 1;
#----------------举个例子：嵌套处理events时需要检测loading-------------------------
#networking.processEventWhileBlocks
# ......
# server.processCommand
# ......
# if (server.loading && !server.async_loading && is_denyloading_command) {
# rejectCommand(c, shared.loadingerr); #此时不允许执行该命令，直接reject
# return C_OK;
# }
# if (server.async_loading && is_deny_async_loading_command) {
# rejectCommand(c,shared.loadingerr);
# return C_OK;
# }
#------------------------------------------------------------------------------
if server.repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB: #如果选择是非交换方式，则同步期间slave不会继续处理read
#笔记：redis slave只能处理读事件
db.emptyData(callback=replicationEmptyDbCallback); #清空redis所有db的数据
.....
replication.replicationEmptyDbCallback #处理完后会调用回调函数
networking.processEventsWhileBlocked #！！！处理events，注意这个processEvents是嵌套的
rio.rioInitWithConn(rdb,conn) #把conn绑定到rdb，这样这个rio结构就会从conn上读取数据
#redis中rio是对rdb文件操作的一个抽象，这里先跳过，后续补，因为现在我也不懂
socket.connBlock #无盘方式要开始读取数据了，不过为了简化流程，直接使用阻塞读取
#所以这里就是把master-slave之间的连接对象设置成阻塞
socket.connRecvTimeout #设置读取超时时间
rdb.dbLoadRioWithLoadingCtx #执行读取
if usemark&&!rio.rioRead(eof) #如果采用的是eof方式标记结束，但是没有检测到eof就表明加载失败
loadingFailed=true
if loadingFailed:
rdb.stopLoading
server.loading = 0; #重置loading和async_loading，表示结束loading状态
server.async_loading = 0;
.....
replication.replicationEmptyDbCallback #处理完后会调用回调函数
networking.processEventsWhileBlocked #！！！处理events，注意这个processEvents是嵌套的
#！！！和lua脚本执行一样，执行一段时间就中断一下当前事件处理
#！！！去执行嵌套的eventloop，以免事件循环长时间阻塞
#！！！注意：同步过程中是在执行某个操作后才会调用这个来处理其他事件
#！！！但是，因为redis是单线程，当主线程执行这个操作时是不能处理其他事件的
#！！！所以说在执行完操作后中断一下来处理其他事件只能算一个小优化
#！！！因为如果操作本身就是耗时的那就会长时间阻塞主线程
...若加载失败，则执行一些资源清理操作...
if server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB: #swap db方式，即直接交换收到的db和old-db
#注意：如果不是swapdb方式则是直接在原db上操作，只有swap方式才需要swap
replication.replicationAttachToNewMaster #首先断开slave的slave，让他们自动与slave重新进行连接和fullsync
db.swapMainDbWithTempDb #交换收到的db和old-db，交换后新db就立即启用了
#注意：这里的交换是指交换内存中两个指针
#具体过程还没细看，不太懂，总之就是：
#1：清空相关辅助数据结构；2：通知其他人key无效；3：交换底层数据结构
db.touchAllWatchedKeysInDb(old-db,new-db)： #这一步骤会标记一些watchKey，这些key同时存在于old-db和new-db中
#因为这些key会被新rdb文件覆盖，所以要标记，从而让对应事务可以检测到失败
db.scanDatabaseForDeletedKeys #此函数会检查在原有数据库中存在的键在新数据库中是否已经被删除
...交换数据库各种对象，略... #因为无盘方式是纯内存操作
db.scanDatabaseForReadyKeys #在交换数据库后，需要检查是否有客户端在等待列表操作。
#如果新数据库中存在某个键，可能会解除这些客户端的阻塞状态。
tracking.trackingInvalidateKeysOnFlush #通知所有客户端缓存的key都无效了,并且server释放相关内存
#客户端会缓存一些key-value来减少请求次数
#当数据无效的时候redis服务器会通知他们数据已经无效了
tracking.sendTrackingMessage #通知client
expire.flushSlaveKeysWithExpireList #移除slaveKeysWithExpire中的key，以免这些expirekey与新数据冲突(不太懂)
function.functionsLibCtxSwapWithCurrent #也要交换function ctx，不懂，pass
moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, #通知其他模块事件发生
REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED,
NULL);
replication.disklessLoadDiscardTempDb #丢弃掉old-db。注意：交换后old-db就变成了临时db，这里释放掉
stopLoading(1);
server.loading = 0; #重置loading和async_loading，表示结束loading状态
server.async_loading = 0;
rioFreeConn(&rdb, NULL);
connNonBlock(conn);
connRecvTimeout(conn,0);
else: #if代码块处理完了无盘方式，else代码块处理刷盘方式
#刷盘方式在走到这里之前就已经从网络读完了整个文件
unistd.fsync #把rdb_fsync_range操作中没有覆盖到的末尾部分的数据强制刷盘
old_rdb_fd=fcntl.open #打开磁盘上旧的正式的db文件
stdio.rename(server.repl_transfer_tmpfile,server.rdb_filename)#把tempdb重命名为正式rdb文件
#？？？疑问：旧的正式的rdb不是已经存在吗？rename不会冲突吗？
#应该有地方已经提前清掉了旧rdb吧，先下班了，2024/10/29 22:56
#2024/10/30 11:15 原来rename会强制覆盖掉旧的同名的文件
if (old_rdb_fd != -1)
bio.bioCreateCloseJob #如果旧rdb文件已经打开，则关闭该文件
util.fsyncFileDir #刷盘rename操作
rdb.rdbLoadWithEmptyFunc(rdb.rdbLoadEmptyDbFunc) #从磁盘加载刚从master获取的rdb文件
#！！！此处并没有fork子进程，即fullsync+刷盘的组合，是一个同步操作
#！！！也就是说fullsync+刷盘直接在主线程执行，会阻塞主线程
#！！！就是说如果rdb文件特别大，则从磁盘加载rdb会长时间阻塞主线程
#！！！fullsync不管是刷盘还是无盘方式，从master读取数据时都会阻塞主线程
#！！！也就是说会等到读取完所有同步数据后才会继续往下执行
#！！！所以说fullsync会阻塞主线程
stdio.fopen #打开新的rdb文件
loadingSetFlags(filename, sb.st_size, 0); #标记loading=1
rdbLoadEmptyDbFunc #清空内存中old-db
db.emptyData(replicationEmptyDbCallback) #1：清空db；2：调用processEventWhileBlock
......
replication.replicationEmptyDbCallback #处理完后会调用回调函数
networking.processEventsWhileBlocked #！！！处理events，注意这个processEvents是嵌套的
#！！！即此时是在fullsync同步事件处理过程中嵌套处理eventloop
#！！！即此时loading=1，loadingSetFlags里设置了loading=1
#！！！所以redis在事件处理中需要检测server是否为loading状态
rioInitWithFile(&rdb,fp); #初始化rio结构，src为文件
rdbLoadRio(&rdb,rdbflags,rsi); #从新的rdb文件加载db
stopLoading(retval==C_OK); #标记加载完成。
server.loading = 0; #重置loading和async_loading，表示结束loading状态
server.async_loading = 0;
replication.replicationCreateMasterClient #创建一个master对应的client，封装了对应的conn
server.master = createClient(conn);
server.repl_state = REPL_STATE_CONNECTED #重置sync状态
replication.clearReplicationId2 #当slave与master一致时使用master的replicationId和offset
replication.replicationSendAck #通知master slave已经同步完成，可以上线
aof.startAppendOnlyWithRetry #！！！重新开启aof日志，会执行rewrite操作
#！！！无盘有盘方式都会通过rewrite操作持久化
#？？？存疑：有盘方式会保存rdb，又rewrtie会怎么样，不是重复了吗
#！！！答疑：区分rdb文件和rdb备份
#！！！rdb文件类似于数据库，不能保证数据是可靠的，比如写一半崩溃了
#！！！rdb备份相当于rdb文件某个时间点的备份，一定是可靠的
#一堆temp-file-name，正式-filename，没有去仔细理清了
#！！！rewrite不会使用以前的aof文件而是直接dump 内存中最新的数据库。
#！！！rewrite完成后直接丢弃旧aof文件
aof.startAppendOnly
aof.rewriteAppendOnlyFileBackground #fork，然后子进程去处理rewrite
redis.fork
aof.rewriteAppendOnlyFile #这里是子进程
fopen(tmpfile,"w") #创建临时文件
rdb.rdbSaveRio #dump整个内存数据库到磁盘上的临时文件
#rewrite和rdbsave实质上差不多，但会同时存在，
#保存的文件叫做
stdio.fflush #强制刷盘
server.sendChildCowInfo #通知父进程rewrite结束
server.exitFromChild #结束子进程
------father会检测子进程rewrite结果-------->
--->ae.processEvents #下一次事件循环时在定时任务中会检测到
ae.processTimeEvents #注意：在上一次循环调用stopLoading后新数据就可以正常对外提供了
#这里只是进行一些扫尾工作
server.serverCorn #每次事件循环都会处理event和timer，这里是timer
server.checkChildrenDone
aof.backgroundRewriteDoneHandler
if rewrite done:
rename(xx,xx.base.xx) #把临时base aof文件命名为正式的base aof文件
rename(xx, xx.incr.xx) #把临时的incr aof文件命名为正式的incr aof
1-2：slave-continue sync流程
continue sync流程就是slave如果从master收到的对psync命令的回复信息是以"+CONTINUE"开头，说明主节点可以进行部分重同步，这种情况下，设置复制状态为REDIS_REPL_CONNECTED，后续将主节点当成一个客户端，接收该主节点客户端发来的命令请求，像处理普通客户端一样处理即可。因此函数slaveTryPartialResynchronization返回PSYNC_CONTINUE后，slave本次同步流程结束，也就是说部分同步把整个大的同步任务分散到了每个事件循环。
#这是上文slaveTryPartialResynchronization函数的详解，部分同步就是在这个函数里处理完。
replication.slaveTryPartialResynchronization(read_replay) #这个函数尝试与master部分同步，
#部分同步：slave把master当做普通client，
#然后把master发过来同步数据当做普通命令执行就行了
#如果失败，则返回fullsync
#这个函数根据read_replay来决定本次是写还是读连接
#read_replay=0表示本次是slave发送psync命令给master
#read_replay=1表示读取master对slave发过去的psync命令的响应
if !read_replay: #发送psync给master
if server.cached_master { #cached_master表示slave之前保存的master同步信息
#psync：paritial-sync。slave发一个psync给master
#如果master支持，那么master上执行的就是psync
#如果master不支持，那么master上执行的就是syncCommand
#意思就是：master会根据实际情况来决定psync命令的实际执行方式
#发送psync命令给master，psync格式：psync runid offset
#runid则是slave保存的masterid，
#offset表示slave要从offset位置开始复制
psync_replid = server.cached_master->replid; #replid表示缓存的masterid如果id不匹配表明master变了
psync_offset = server.cached_master->reploff+1 #reploff表示复制到了这里，master应该从reploff+1位置复制
} else { #表示slave缓存的master信息失效或者这是slave第一次同步
#发送: psync ? -1，表示slave要求全同步
psync_replid = "?";
memcpy(psync_offset,"-1",3);
}
replication.sendCommand("PSYNC",psync_replid,psync_offset) #发送psync命令给master
return PSYNC_WAIT_REPLY
else: #表示本次是读取master对slave刚发送的psync命令的响应
#就是说第一次事件处理发送psync，然后返回，
#然后master发来响应，连接可读，触发事件再次走到这里
replication.receiveSynchronousResponse #读取master发来的响应
connection.connSetReadHandler(conn, NULL);
if (!strncmp(reply,"+FULLRESYNC",11)): #表示master要求进行fullsync
#意思就是说slave发一个psync过去，master可能支持全同步或者部分同步
#fullsync表示master此次要求是全同步
#因为是全同步，所以直接用master发来的runid和更新slave保存的replid
server.master_replid=xxx #从master发来的响应中提取master的runid并缓存
server.master_initial_offset=xxx #从master发来的响应中提取master的reploff并缓存
return PSYNC_FULLRESYNC; #表示本次协商结果为fullsync全同步
if (!strncmp(reply,"+CONTINUE",9)): #表示本次协商结果为continue sync
if (strcmp(new,server.cached_master->replid)): #如果master发来的响应中的masterid与slave自己保存的不一样，说明masterid变了
memcpy(server.replid2,server.cached_master->replid); #保存旧的cached_master id和offset到server.replid2
server.second_replid_offset = server.master_repl_offset+1; #更新slave的replid_offset
memcpy(server.replid,new); #更新slave的replid
memcpy(server.cached_master->replid,new); #更新缓存的masterid
networking.disconnectSlaves #通知slave的slave断开连接，因为slave的master变了
replication.replicationResurrectCachedMaster #！！！实现部分同步，部分同步的原理如下：
#！！！与master建立连接，把master当做一个普通client
#！！！这样master要与slave同步一条数据，master就发一条命令过来
#！！！因为master发来的命令就是一个普通client发来的命令
#！！！所以slave直接在事件循环里执行这个命令就行了
#！！！即continuesync方式把同步操作分散到了每次事件循环中
#！！！而fullsync方式则是一次就执行完整个复制操作
#！！！所以如果是continuesync方式，直接返回就行了
#！！！slave无需继续往下走
server.repl_state = REPL_STATE_CONNECTED; #重置sync状态,协商阶段就是从这个状态开始的,也就是表明本次部分同步过程已完成
server.master = xxxx #server.master是一个client*，这里就是初始化这个client
#就是把conn包装到这个client中
connection.connSetReadHandler(readQueryFromClient) #向epoll注册master对应的client上的读事件
#此后就像处理普通client上的读事件一样处理master上的同步命令
#也就是说continuesync方式把同步操作分散到了每次事件循环
#因为这里是同一个conn，也就是说重置了read_handler
#即continuesync:readhandler=readQueryFromCLient
#即fullsync:readhandler=readSyncBulkPayload
replication.createReplicationBacklog #创建挤压日志缓冲区，不太懂，暂略
return PSYNC_CONTINUE; #表示本次协商结果为部分同步
2：master fullsync和continuesync
master流程和slave流程是相对的，也差不多。就是slave发命令过来，master执行相应的命令，然后slave再发psync过来，master检查最后决定是fullsync还是continuesync，如果是continuesync就结束了，如果是fullsync又分两种，如果是diskless即无盘方式则直接把内存db发到socket上，然后由slave去读取，如果是刷盘方式则开一个子进程去bgsave rdb文件到磁盘，然后每次事件循环的时候会处理事件和定时器事件，redis就在定时器事件里检查bgsave是否完成，一旦bgsave完成就会把文件发送给slave。注意：和slave实现不同的地方之一在于协商阶段slave是一个syncWithMaster发送各种命令给master，而master则是每个命令都对应不同的函数，而不是一个函数处理所有的，这里我们只介绍最重要的syncCommand函数，因为这个函数是对slave发来的psync命令的处理，相比slave，我们的记录会简洁很多，因为和slave差不多。
备注：continuesync即部分同步，这部分源码流程里的repl_backlog和replication_buffer没有去细究了，暂时还不太命令，可能有些许错误（比如写的是replication_buffer，但我记成了repl_backlog），这些错误不影响整体流程，总之就是执行命令的时候会把命令保存到同步缓冲区，然后beforesleep里把缓冲区里的数据发送给slave，就这样。也懒得去深究了，因为这周双休，而现在 20241101 17:50，下班ing
replication.syncCommand #这个函数处理sync或者psync，如果支持psync就是psync，就是continue sync
#如果不支持psync，那么就是sync，即fullsync
if (!strcasecmp(c->argv[0]->ptr,"psync")): #如果发来的命令是psync。这个if是为了兼容以前的sync，如果不是psync那么就是sync，就是fullsync
replication.masterTryPartialResynchronization #master根据某些条件来判断本次是continuesync还是fullsync
if repid对不上 or 其他:
return C_ERR #C_ERR表示进行fullsync
replication.addReplyReplicationBacklog #！！！实现部分同步，原理很简单：
#！！！master用一个repl_backlog来保存增量同步的数据，
#！！！这个repl_backlog是所有slave共用的
#！！！repl_backlog[x,end]表示缓冲区中从x到end位置的数据都是要发送给slave x的
#！！！repl_backlog[y,end]表示缓冲区中从y到end位置的数据都是要发送给slave y的
#！！！repl_backlog[z,end]表示缓冲区中从z到end位置的数据都是要发送给slave z的
#！！！所以repl_backlog用一个结构体来保存每个slave所需的数据在repl_backlog中的位置信息
#！！！即xx_node表示的是slave所需的数据在master上的repl_backlog缓冲区中的位置
#！！！我们把信息封装到client中，这样这个client就包含了slave->conn,slave->data
#！！！然后把slave对应的client丢到pending_write_queue，
#！！！然后redis会在每次事件循环开始前的beforesleep中
#！！！都会调用handleClientsWithPendingWritesUsingThreads来处理这个list
#！！！因为这个node包含了对应的连接对象和数据，所以直接write连接就行
#！！！redis在执行普通的set命令时会顺便把set命令写到repl_backlog 缓冲区
#！！！也就是说不管有没有slave，master执行写命令时都会写repl_backlog
#！！！就是说这个缓冲区是一直有数据的，因为xx_node相当于一个指针，
#！！！所以只要node在pending_write_queue中，就一直可以获取到最新的数据，
#！！！就会进行部分同步，如果不需要了，则只需要从queue中移除就行了
...... #1:从repl_backlog->blocks_index中获取slave对应的xx_node的信息
raxStart(&ri, server.repl_backlog->blocks_index);#准备搜索repl_backlog->blocks_index
......
node = (listNode *)ri.data #在repl_backlog中找到包含slave所需的增量同步数据信息的node
......
c->ref_repl_buf_node = node; #这里就是把对应的信息填充slave对应的xx_node
c->ref_block_pos = offset - o->repl_offset; #也就是说我们这一步把相关信息保存到了slave对应的client内部
networking.prepareClientToWrite #2:把保存了足够多信息的client丢到pending_write_queue
networking.putClientInPendingWriteQueue
listLinkNodeHead(server.clients_pending_write, &c->clients_pending_write_node); #丢到pending队列
----->这里来到事件循环
--->server.main
ae.aeMain(server.el)
ae.aeProcessEvents #开始处理新一轮事件循环
ae.aeEventLoop.beforesleep #！！！在正式处理eventloop之前执行beforesleep，
#！！！完成上一轮迭代中未完成的工作，比如网络io
networking.handleClientsWithPendingWritesUsingThreads
for server.clients_pending_write: #！！！这里遍历pending_write链表
networking.writeToClient
networking._writeToClient
replBufBlock *o = listNodeValue(c->ref_repl_buf_node); #提取我们前面保存的要发送的数据的信息
if xx: #redis会根据情况来选择是writev还是普通write
networking._writevToClient #发送给slave
else:
connection.connWrite #发送给slave,至此，增量同步源码流程结束
return C_OK #C_OK表示进行continuesync即psync
if C_OK: #continuesync直接返回就行了，无需继续往后走,下面是fullsync流程
return
replication.startBgsaveForReplication #开启一个bgsave过程。1：会fork；2：会根据参数决定dst是本地磁盘还是socket
#备注：可能此时正有一个bgsave在运行中，如果是disk，
#那么他会把slave添加到target列表，bgsave结束后会自动发送
#如果是socket，那么就会推迟稍后再发送
rdb.rdbPopulateSaveInfo
if target==socket: #1：即bgsave的目的地为网络，即无盘同步方式
#无盘方式流程：1：fork，然后创建pipe，然后父进程注册管道读事件
#2：子进程写，父进程读，然后父进程把读到的数据发送到socket上
rdb.rdbSaveToSlavesSockets
anet.anetPipe #创建父子进程通信的pipe
server.rdb_pipe_read = pipefds[0]; #子进程写rdb数据，父进程读rdb数据
rdb_pipe_write = pipefds[1]; #所以子进程会关闭rdb_pipe_read
#而父进程则会关闭rdb_pipe_write
for all need sync slaves: #！！！master上同步操作是批量的即一次性发送给多个slave
server.rdb_pipe_conns[server.rdb_pipe_numconns++] = slave->conn; #保存slave对应的连接
#rdbPipeReadHandler函数中会取出来
server.redisFork #fork
if child: #这里是子进程
rio.rioInitWithFd(rdb_pipe_write) #初始化rio结构，dst=pipe[1]即子进程中rio负责把数据写到管道
unistd.close(server.rdb_pipe_read) #子进程关闭管道读取端，因为子进程只写
rdb.rdbSaveRioWithEOFMark #发送数据到管道
rio.rioWrite(rdb,"$EOF:",5) #发送$EOF表示本次传输以EOF字段结束
rdb.rdbSaveRio #执行传输。rio是对redis对io的抽象
#主要用于rdb，这里dst=rdb_pipe_write
#就是说这里会dump内存所有db到管道
#再次强调：这里只适用于无盘方式的fullsync
rio.rioFlush
server.sendChildCowInfo #通知父进程子进程ok
else parent: #这里是父进程
close(rdb_pipe_write); #父进程只会读管道，所以这里关闭管道写端
ae.aeCreateFileEvent(server.el, server.rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL)
#！！！rdb_pipe_read是管道的读取端读
#！！！这里把rdb_pipe_read的读事件注册到epoll中
#！！！这样当子进程通过rio写管道的时候就会触发父进程读事件
#！！！然后父进程调用rdbPipeReadHandler
#！！！这个函数从管道读取数据，然后发送给slave
--->这里来到redis事件处理:
--->rdb.rdbPipeReadHandler #子进程往管道写数据事父进程会处理这个事件
buflen=unistd.read(rdb_pipe_read) #从管道读取数据
conn = server.rdb_pipe_conns[i] #取出我们前面保存的slave对应的连接
nwrite=connection.connWrite(conn) #把数据发往slave
if nwritten != server.rdb_pipe_bufflen: #表明数据一次写不完，所以通过事件循环来发送剩余数据
connection.connSetWriteHandler(conn, rdbPipeWriteHandler);
#rdbPipeWriteHandler就直接发送数据给slave
#没有管道读操作，因为数据已经读到了buf
if buflen==0: #buflen=0表示管道已结束即子进程数据全部发送完毕
ae.aeDeleteFileEvent(sserver.rdb_pipe_read, AE_READABLE); #epollfd中移除管道读事件
rdb.killRDBChild #kill bgsave进程
else: #bgsave目的地为本地磁盘
rdb.rdbSaveBackground
server.redisFork #fork
if child:
rdb.rdbSave #dump内存到磁盘，后续会在定时任务中发送给slave
rdb.startSaving(rdbflags); #通知其他模块 bgsave事件
rdb.rdbSaveInternal #保存到磁盘上的临时文件
stdio.rename(tmpfile,filename) #把临时文件重命名为正式文件
stdio.fsyncFileDir #强制刷盘
rdb.stopSaving(success) #通知其他模块bgsave成功
server.sendChildCowInfo #通知父进程子进程ok
----->这里来到事件循环定时器--启动时创建定时器任务
--->server.main
server.initServer
ae.aeCreateTimeEvent(server.el, 1, serverCron) #向ae中注册serverCron定时器事件，每隔1s处理一次
----->这里来到每次事件循环中的定时器任务处理
--->server.main
ae.aeMain
ae.aeProcessEvents #事件包括fd上的事件也包括定时器事件
ae.aeProcessTimeEvents
server.serverCron #定时器事件
server.checkChildrenDone #检查子进程是否结束
rdb.backgroundSaveDoneHandler #1：做一些清理工作；2：如果有需要则传送给slave
case RDB_CHILD_TYPE_DISK:
rdb.backgroundSaveDoneHandlerDisk #做一些清理工作
case RDB_CHILD_TYPE_SOCKET:
rdb.backgroundSaveDoneHandlerSocket #做一些清理工作
replication.updateSlavesWaitingBgsave
for all wait_bgsave_end slaves: #发给所有需要等待bgsave事件的slave
fcntl.open(server.rdb_filename) #打开刚才dump的rdb文件，“dump.db”
connection.connSetWriteHandler(slave->conn,sendBulkToSlave) #注册write事件到eventLoop
#也就是说redis收发数据都是通过eventloop主线程来处理的
----->这里来到eventloop conn可写事件
--->replication.sendBulkToSlave #发送数据给slave
unistd.read #读取rdb文件
connection.connWrite #发送给slave
未完待续：
缓冲区问题以及一些细节。可以很方便的通过linux系统提供的fork功能来快速获得一份内存rdb的备份，如果写比较少，那么可以省部分，fork只能获取一份快照，但是快照点之后的更新怎么处理的，网上文章说是通过双写缓冲区来实现的，repl_backlog和replication buffer两个buffer的作用，暂时还没有细看（找到了repl_backlog，但是源码里并没有找到replication buffer，所以这两个东西还不太清楚），笔记也可能有些许谬误，但大体是没错的，暂时先放下了
一些文章：
https://www.cnblogs.com/jing99/p/16142254.html
https://www.cnblogs.com/feiyu2/p/16996233.html
https://www.cnblogs.com/lsx1993/p/4633032.html
---------------杂记--------------------------
我猜应该是bgsave完成后有地方异步检测吧
还真有，replicationcron()
anyOtherSlaveWaitRdb()
updateSlavesWaitingBgsave()
removeRDBUsedToSyncReplicas()
slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END
repl_backlog是什么replicationSendAck()
master replid=server.replid
master replid2="26d5d3c287176c5ad84f5c7ff2f8355448e5de2e"
slave replid2"5222388599547b4956b1540c4170411ae2362fbb"
slave server.rdb_name=dump.db
无盘方式：tmpfile="temp-1730251273.4143.rdb"
刷盘方式中rewrite也会执行吗？会。区分dump.db和aof.base
更多推荐
22
0- 0
已为社区贡献3条内容
所有评论(0)