Spawn an RDB child that writes the RDB to the sockets of the slaves * that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */
| 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. */ |
| 3776 | int 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) { |
no test coverage detected