MCPcopy Create free account
hub / github.com/Snapchat/KeyDB / rdbSaveToSlavesSockets

Function rdbSaveToSlavesSockets

src/rdb.cpp:3776–3957  ·  view source on GitHub ↗

Spawn an RDB child that writes the RDB to the sockets of the slaves * that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */

Source from the content-addressed store, hash-verified

3774/* Spawn an RDB child that writes the RDB to the sockets of the slaves
3775 * that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */
3776int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
3777 serverAssert(GlobalLocksAcquired());
3778 listNode *ln;
3779 listIter li;
3780 pthread_t child;
3781 int pipefds[2];
3782 rdbSaveSocketThreadArgs *args = nullptr;
3783
3784 if (hasActiveChildProcessOrBGSave()) return C_ERR;
3785
3786 /* Even if the previous fork child exited, don't start a new one until we
3787 * drained the pipe. */
3788 if (g_pserver->rdb_pipe_conns) return C_ERR;
3789
3790 /* Before to fork, create a pipe that is used to transfer the rdb bytes to
3791 * the parent, we can't let it write directly to the sockets, since in case
3792 * of TLS we must let the parent handle a continuous TLS state when the
3793 * child terminates and parent takes over. */
3794 if (pipe(pipefds) == -1) return C_ERR;
3795
3796 args = (rdbSaveSocketThreadArgs*)zmalloc(sizeof(rdbSaveSocketThreadArgs) + sizeof(redisDbPersistentDataSnapshot*)*(cserver.dbnum-1), MALLOC_LOCAL);
3797 g_pserver->rdb_pipe_read = pipefds[0]; /* read end */
3798 args->rdb_pipe_write = pipefds[1]; /* write end */
3799 anetNonBlock(NULL, g_pserver->rdb_pipe_read);
3800
3801 args->rsi = *(new (&args->rsi) rdbSaveInfo(*rsi));
3802 memcpy(&args->rsi.repl_id, g_pserver->replid, sizeof(g_pserver->replid));
3803 args->rsi.master_repl_offset = g_pserver->master_repl_offset;
3804
3805 /* create another pipe that is used by the parent to signal to the child
3806 * that it can exit. */
3807 if (pipe(pipefds) == -1) {
3808 close(args->rdb_pipe_write);
3809 close(g_pserver->rdb_pipe_read);
3810 zfree(args);
3811 return C_ERR;
3812 }
3813 args->safe_to_exit_pipe = pipefds[0]; /* read end */
3814 g_pserver->rdb_child_exit_pipe = pipefds[1]; /* write end */
3815
3816 /* Collect the connections of the replicas we want to transfer
3817 * the RDB to, which are i WAIT_BGSAVE_START state. */
3818 g_pserver->rdb_pipe_conns = (connection**)zmalloc(sizeof(connection *)*listLength(g_pserver->slaves));
3819 g_pserver->rdb_pipe_numconns = 0;
3820 g_pserver->rdb_pipe_numconns_writing = 0;
3821 listRewind(g_pserver->slaves,&li);
3822 while((ln = listNext(&li))) {
3823 client *slave = (client*)ln->value;
3824 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
3825 g_pserver->rdb_pipe_conns[g_pserver->rdb_pipe_numconns++] = slave->conn;
3826 replicationSetupSlaveForFullResync(slave,getPsyncInitialOffset());
3827 }
3828 }
3829
3830 /* Create the child process. */
3831 if (cserver.fForkBgSave) {
3832 pid_t childpid;
3833 if ((childpid = redisFork(CHILD_TYPE_RDB)) == 0) {

Callers 1

Calls 15

GlobalLocksAcquiredFunction · 0.85
zmallocFunction · 0.85
anetNonBlockFunction · 0.85
zfreeFunction · 0.85
listRewindFunction · 0.85
listNextFunction · 0.85
getPsyncInitialOffsetFunction · 0.85
redisForkFunction · 0.85
rioInitWithFdFunction · 0.85
redisSetProcTitleFunction · 0.85

Tested by

no test coverage detected