redis server replication repl_functions

Replication Functions

Redis Server Course	Redis Technical Support	Redis Enterprise Server

Replication Functions

이 문서는 버전 7.0.8을 기준으로 작성했습니다.

• Full Sync : Master Functions
• Full Sync : Replica Functions
• replication.c Functions
• rdb.c Functions

Full Sync : Master Functions

마스터 서버에서 전체 동기화(full sync) 용도로 사용되는 함수들을 정리했습니다.

replconfCommand()

Redis version 3.0에 만들어졌고, replication.c에 있습니다.
안에 있는 버전(ver)은 각 기능이 해당 버전에서 처음 만들졌다는 표시입니다. "listening-port" c->slave_listening_port = listening-port ver 3.0.0 "ip-address" c->slave_addr = ip-address ver 3.2.2 "capa", "eof" c->slave_capa |= SLAVE_CAPA_EOF; ver 3.2.0 "capa", "psync2" c->slave_capa |= SLAVE_CAPA_PSYNC2; ver 4.0.0 "ack", offset c->repl_ack_off = offset; ver 3.0.0 c->repl_ack_time = server.unixtime; “getack" send "ack", offset ver 3.0.0 "rdb-only" c->flags |= CLIENT_REPL_RDBONLY ver 6.2.0 "rdb-filter-only", "functions“ -> function만 받겠다. (데이터는 받지 않겠다) ver 7.0.0 c->slave_req &= ~SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS;

syncCommand()

Redis version 1.0에 만들어졌고, replication.c에 있습니다. "Replica ip:port asks for synchronization“ if (psync) { if (masterTryPartialResynchronization()) == C_OK) return; // PSYNC 성공 } // FULL SYNC “Replication backlog created, my new replication IDs are “ if (bgsaving and disk) { "Waiting for end of BGSAVE for SYNC“ } else if (bgsaving and socket) { "Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC“ } else { if (server.repl_diskless_sync and server.repl_diskless_sync_delay) "Delay next BGSAVE for diskless SYNC“ else startBgsaveForReplication(); }

masterTryPartialResynchronization()

replication.c에 있습니다. if (server.replid != replid) { if (replid[0] != ‘?’) "Partial resynchronization not accepted” else "Full resync requested by replica” goto need_full_resync; } c->replstate = SLAVE_STATE_ONLINE; listAddNodeTail(server.slaves,c); connWrite(c->conn,”+CONTINUE”); "Partial resynchronization request from ip:port accepted.” refreshGoodSlavesCount(); return C_OK; need_full_resync: return C_ERR;

startBgsaveForReplication()

replication.c에 있습니다. if (socket) rdbSaveToSlavesSockets() "Starting BGSAVE for SYNC with target: replicas sockets” else rdbSaveBackground() "Starting BGSAVE for SYNC with target: disk” if (disk) replicationSetupSlaveForFullResync() "+FULLRESYNC replid offset"

rdbSaveToSlavesSockets()

rdb.c에 있습니다. anetPipe(); listRewind(server.slaves,&li); while((ln = listNext(&li))) if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) replicationSetupSlaveForFullResync(); -> Send “FULLRESYNC replid offset” if (redisFork() == 0) { // Child Process rdbSaveRioWithEOFMark(); sendChildCowInfo(); exitFromChild(); } else { // Parent Process "Background RDB transfer started by pid“ aeCreateFileEvent(rdbPipeReadHandler); }

rdbSaveBackground()

rdb.c에 있습니다. if (redisFork() == 0) { // Child Process rdbSave(); sendChildCowInfo(); exitFromChild(); } else { // Parent Process "Background saving started by pid” }

replicationSetupSlaveForFullResync()

replication.c에 있습니다. slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END; connWrite(“+FULLRESYNC replid offset”);

rdbSaveRioWithEOFMark()

rdb.c에 있습니다. startSaving(); rioWrite(“$EOF:eofmark”); rdbSaveRio(); rioWrite(“$EOF:eofmark”); stopSaving();

rdbSave()

rdb.c에 있습니다. fopen(tmpfile); startSaving(); if (server.rdb_save_incremental_fsync) rioSetAutoSync(); rdbSaveRio(); fflush(); fsync(); fclose(); rename(); fsyncFileDir(); "DB saved on disk“ server.lastsave = time(NULL); stopSaving();

rdbSaveRio()

rdb.c에 있습니다. rdbWriteRaw(rdb,magic); rdbSaveInfoAuxFields(rdb,rdbflags); rdbSaveDb(rdb,dbnum); // dbnum별로 저장 rdbSaveType(rdb,RDB_OPCODE_EOF); rioWrite(rdb,&chsum);

checkChildrenDone()

server.c에 있습니다. pid = waitpid(); if (pid == server.child_pid) { if (server.child_type == CHILD_TYPE_RDB) backgroundSaveDoneHandler(); else if (server.child_type == CHILD_TYPE_AOF) backgroundRewriteDoneHandler(); else if (server.child_type == CHILD_TYPE_MODULE) ModuleForkDoneHandler(); } replicationStartPendingFork();

backgroundSaveDoneHandler()

server.c if (disk) backgroundSaveDoneHandlerDisk() else backgroundSaveDoneHandlerSocket() updateSlavesWaitingBgsave();

backgroundSaveDoneHandlerDisk()

rdb.c "Background saving terminated with success“ server.lastsave = time(NULL); server.lastbgsave_status = C_OK;

backgroundSaveDoneHandlerSocket()

rdb.c "Background RDB transfer terminated with success“ aeDeleteFileEvent();

updateSlavesWaitingBgsave()

replication.c if (socket) "Streamed RDB transfer with replica %s succeeded (socket)" replicaPutOnline() else open(filename) connSetWriteHandler(sendBulkToSlave)

replicaPutOnline()

replication.c slave->replstate = SLAVE_STATE_ONLINE; slave->repl_ack_time = server.unixtime; refreshGoodSlavesCount(); "Synchronization with replica ip:port succeeded"

sendBulkToSlave()

replication.c connWrite(length); lseek(slave->repldbfd); read(slave->repldbfd,buf,16kb); connWrite(buf); if (slave->repldboff == slave->repldbsize) { close(slave->repldbfd); connSetWriteHandler(slave->conn,NULL); replicaPutOnline(slave); replicaStartCommandStream(slave); }

Full Sync : Replica Functions

복제 서버에서 전체 동기화(full sync) 용도로 사용되는 함수들을 정리했습니다.

replicaofCommand()

replication.c if “replicaof no one” replicationUnsetMaster() else replicationSetMaster() "REPLICAOF ip:port enabled (user request from '%s'")

replicationSetMaster()

replication.c server.masterhost = sdsnew(ip); server.masterport = port; if (was_master) { // 기존에 마스터가 있었으면 replicationDiscardCachedMaster(); replicationCacheMasterUsingMyself(); } server.repl_state = REPL_STATE_CONNECT; "Connecting to MASTER IP:Port“ connectWithMaster()

connectWithMaster()

replication.c server.repl_transfer_s = server.tls_replication ? connCreateTLS() : connCreateSocket(); connConnect(server.repl_transfer_s, ip, port, bind_source_addr, syncWithMaster) server.repl_state = REPL_STATE_CONNECTING; "MASTER <-> REPLICA sync started"

syncWithMaster()

replication.c if (server.repl_state == REPL_STATE_CONNECTING) { "Non blocking connect for SYNC fired the event." connSetReadHandler(conn, syncWithMaster); connSetWriteHandler(conn, NULL); server.repl_state = REPL_STATE_RECEIVE_PING_REPLY; sendCommand(conn,"PING",NULL); } if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) { receiveSynchronousResponse(conn); if (“PONG” or “NOAUTH”) "Master replied to PING, replication can continue..." else "Error reply to PING from master: err” server.repl_state = REPL_STATE_SEND_HANDSHAKE; } if (server.repl_state == REPL_STATE_SEND_HANDSHAKE) { if (server.masterauth) sendCommandArgv(“AUTH”,”USERNAME”,”PASSWORD”); sendCommand(conn,"REPLCONF“,"listening-port",port); if (server.slave_announce_ip) sendCommand(conn,"REPLCONF","ip-address",IP); sendCommand(conn,"REPLCONF","capa","eof","capa","psync2“); server.repl_state = REPL_STATE_RECEIVE_AUTH_REPLY; } if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) { receiveSynchronousResponse(conn); if (err) "Unable to AUTH to MASTER: err”; goto error; server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY; } if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) { receiveSynchronousResponse(conn); if (err) "(Non critical) Master does not understand REPLCONF listening-port" server.repl_state = REPL_STATE_RECEIVE_IP_REPLY; } if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) { receiveSynchronousResponse(conn); if (err) "(Non critical) Master does not understand REPLCONF ip-address" server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY; } if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) { receiveSynchronousResponse(conn); if (err) "(Non critical) Master does not understand REPLCONF capa“ server.repl_state = REPL_STATE_SEND_PSYNC; } if (server.repl_state == REPL_STATE_SEND_PSYNC) { slaveTryPartialResynchronization(conn,0); // send MSG server.repl_state = REPL_STATE_RECEIVE_PSYNC_REPLY; } if (server.repl_state != REPL_STATE_RECEIVE_PSYNC_REPLY) "syncWithMaster(): state machine error" slaveTryPartialResynchronization(conn,1); // read MSG if (disk-load) open(tmpfile); connSetReadHandler(conn, readSyncBulkPayload); server.repl_state = REPL_STATE_TRANSFER; server.repl_transfer_size = -1; // 나중에 파일 사이즈를 넣는다. server.repl_transfer_read = 0; // 파일에서 읽을 때마다 증가 += nread

slaveTryPartialResynchronization()

replication.c if (send) { if (server.cached_master) "Trying a partial resynchronization (request replid:offset)." else "Partial resynchronization not possible (no cached master)“ sendCommand(conn,"PSYNC",replid,offset); return PSYNC_WAIT_REPLY; } // read reply = receiveSynchronousResponse(conn); connSetReadHandler(conn, NULL); if (reply == “+FULLRESYNC”) { memcpy(server.master_replid,replid); “Full resync from master: replid:offset” return PSYNC_FULLRESYNC; } if (reply == "+CONTINUE“) { "Successful partial resynchronization with master.“ memcpy(server.replid2,server.cached_master->replid); memcpy(server.replid,new); return PSYNC_CONTINUE; } return PSYNC_NOT_SUPPORTED;

readSyncBulkPayload()

Master로 부터 데이터를 받는다. replication.c if (socket) // diskless load 이면 to parser, disk load이면 to disk "MASTER <-> REPLICA sync: receiving streamed RDB from master with EOF to parser or to disk " else "MASTER <-> REPLICA sync: receiving x bytes from master to parser or to disk" if (file 저장) { nread = connRead(conn,buf,readlen); // socket에서 읽어서 write(server.repl_transfer_fd,buf,nread)) // file에 쓴다. if (stream 받기) eofmark 비교 -> stream 끝 판단. else (file 받기) file length 비교 -> file 끝 판단. } if (diskless_load) disklessLoadInitTempDb(); else { replicationAttachToNewMaster(); "MASTER <-> REPLICA sync: Flushing old data“; emptyData(-1); } "MASTER <-> REPLICA sync: Loading DB in memory” if (diskless_load) { rioInitWithConn(); // rio.c connBlock(conn); // connection.c connRecvTimeout(); // connection.c startLoading(); rdbLoadRioWithLoadingCtx(); if (swapdb) { replicationAttachToNewMaster(); "MASTER <-> REPLICA sync: Swapping active DB with loaded DB“ swapMainDbWithTempDb(diskless_load_tempDb); db.c disklessLoadDiscardTempDb(diskless_load_tempDb); replication.c "MASTER <-> REPLICA sync: Discarding old DB in background“ } stopLoading() } else { // disk load fsync(server.repl_transfer_fd) open(server.rdb_filename,...) rename(tmpfile, filename) fsyncFileDir(filename) rdbLoad(filename) } "MASTER <-> REPLICA sync: Finished with success“ if (usemark) replicationSendAck();

rdbLoad()

rdb.c fopen() startLoadingFile() rioInitWithFile() rdbLoadRio() rdbLoadRioWithLoadingCtx() fclose() stopLoading()

rdbLoadRioWithLoadingCtx()

rdb.c "Loading RDB produced by version 7.0.8" "RDB age 5 seconds" "RDB memory usage when created 963.87 Mb“ while (1) { key = rdbGenericLoadStringObject() val = rdbLoadObject() dbAddRDBLoad(db,key,val) if (RDB_OPCODE_EOF) break; } "Done loading RDB, keys loaded: 9949010, ..."

replication.c Functions

소스 파일 replication.c에 있는 함수들입니다. 버전 7.0.8 기준으로 작성했습니다.

1. ----- Utility functions -----
2. replicationGetSlaveName()
3. bg_unlink()
4. ----- MASTER -----
5. createReplicationBacklog()
6. resizeReplicationBacklog()
7. freeReplicationBacklog()
8. createReplicationBacklogIndex ()
9. rebaseReplicationBuffer ()
10. resetReplicationBuffer ()
11. prepareReplicasToWrite()
12. feedReplicationBufferWithObject()
13. incrementalTrimReplicationBacklog()
14. freeReplicaReferencedReplBuffer()
15. feedReplicationBuffer()
16. replicationFeedSlaves()
17. showLatestBacklog()
18. replicationFeedStreamFromMasterStream()
19. addReplyReplicationBacklog()
20. getPsyncInitialOffset()
21. replicationSetupSlaveForFullResync()
22. masterTryPartialResynchronization()
23. startBgsaveForReplication()
24. syncCommand()
25. replconfCommand()
26. replicaPutOnline()
27. replicaStartCommandStream()
28. removeRDBUsedToSyncReplicas()
29. sendBulkToSlave()
30. rdbPipeWriteHandlerConnRemoved()
31. rdbPipeWriteHandler()
32. rdbPipeReadHandler()
33. updateSlavesWaitingBgsave()
34. changeReplicationId()
35. clearReplicationId2()
36. shiftReplicationId()
37. ----------- SLAVE(REPLICA) -------------
38. slaveIsInHandshakeState()
39. replicationSendNewlineToMaster()
40. replicationEmptyDbCallback()
41. replicationCreateMasterClient()
42. restartAOFAfterSYNC()
43. useDisklessLoad()
44. disklessLoadInitTempDb()
45. disklessLoadDiscardTempDb()
46. replicationAttachToNewMaster()
47. readSyncBulkPayload()
48. receiveSynchronousResponse()
49. sendCommandRaw()
50. sendCommand()
51. sendCommandArgv()
52. slaveTryPartialResynchronization()
53. syncWithMaster()
54. connectWithMaster()
55. undoConnectWithMaster()
56. replicationAbortSyncTransfer()
57. cancelReplicationHandshake()
58. replicationSetMaster()
59. replicationUnsetMaster()
60. replicationHandleMasterDisconnection()
61. replicaofCommand()
62. roleCommand()
63. replicationSendAck()
64. ------ MASTER CACHING FOR PSYNC ----
65. replicationCacheMaster()
66. replicationCacheMasterUsingMyself()
67. replicationDiscardCachedMaster()
68. replicationResurrectCachedMaster()
69. ----- MIN-SLAVES-TO-WRITE ----
70. refreshGoodSlavesCount()
71. checkGoodReplicasStatus()
72. ----- SYNCHRONOUS REPLICATION ----
73. replicationRequestAckFromSlaves()
74. replicationCountAcksByOffset()
75. waitCommand()
76. unblockClientWaitingReplicas()
77. processClientsWaitingReplicas()
78. replicationGetSlaveOffset()
79. ----- REPLICATION CRON ----
80. replicationCron()
81. shouldStartChildReplication()
82. replicationStartPendingFork()
83. findReplica()
84. getFailoverStateString()
85. clearFailoverState()
86. abortFailover()
87. failoverCommand()
88. updateFailoverStatus()

rdb.c Functions

소스 파일 rdb.c에 있는 함수들입니다. 버전 7.0.8 기준으로 작성했습니다.

1. rdbReportError()
2. rdbWriteRaw()
3. rdbSaveType()
4. rdbLoadType()
5. rdbLoadTime()
6. rdbSaveMillisecondTime()
7. rdbLoadMillisecondTime()
8. rdbSaveLen()
9. rdbLoadLenByRef()
10. rdbLoadLen()
11. rdbEncodeInteger()
12. rdbLoadIntegerObject()
13. rdbTryIntegerEncoding()
14. rdbSaveLzfBlob()
15. rdbSaveLzfStringObject()
16. rdbLoadLzfStringObject()
17. rdbSaveRawString()
18. rdbSaveLongLongAsStringObject()
19. rdbSaveStringObject()
20. rdbGenericLoadStringObject()
21. rdbLoadStringObject()
22. rdbLoadEncodedStringObject()
23. rdbSaveDoubleValue()
24. rdbLoadDoubleValue()
25. rdbSaveBinaryDoubleValue()
26. rdbLoadBinaryDoubleValue()
27. rdbSaveBinaryFloatValue()
28. rdbLoadBinaryFloatValue()
29. rdbSaveObjectType()
30. rdbLoadObjectType()
31. rdbSaveStreamPEL()
32. rdbSaveStreamConsumers()
33. rdbSaveObject()
34. rdbSavedObjectLen()
35. rdbSaveKeyValuePair()
36. rdbSaveAuxField()
37. rdbSaveAuxFieldStrStr()
38. rdbSaveAuxFieldStrInt()
39. rdbSaveInfoAuxFields()
40. rdbSaveSingleModuleAux()
41. rdbSaveFunctions()
42. rdbSaveDb()
43. rdbSaveRio()
44. rdbSaveRioWithEOFMark()
45. rdbSave()
46. rdbSaveBackground()
47. rdbRemoveTempFile()
48. rdbLoadCheckModuleValue()
49. _ziplistPairsEntryConvertAndValidate()
50. ziplistPairsConvertAndValidateIntegrity()
51. _ziplistEntryConvertAndValidate()
52. _listZiplistEntryConvertAndValidate()
53. _lpPairsEntryValidation()
54. lpPairsValidateIntegrityAndDups()
55. rdbLoadObject()
56. startLoading()
57. startLoadingFile()
58. loadingAbsProgress()
59. loadingIncrProgress()
60. updateLoadingFileName()
61. stopLoading()
62. startSaving()
63. stopSaving()
64. rdbLoadProgressCallback()
65. rdbFunctionLoad()
66. rdbLoadRio()
67. rdbLoadRioWithLoadingCtx()
68. rdbLoad()
69. backgroundSaveDoneHandlerDisk()
70. backgroundSaveDoneHandlerSocket()
71. backgroundSaveDoneHandler()
72. killRDBChild()
73. rdbSaveToSlavesSockets()
74. saveCommand()
75. bgsaveCommand()
76. rdbPopulateSaveInfo()