Function rdbPipeReadHandler

src/replication.cpp:1981–2083

Called in diskless master, when there's data to read from the child's rdb pipe */

/* Called in diskless master, when there's data to read from the child's rdb pipe */
void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask) {
    UNUSED(mask);
    UNUSED(clientData);

    int i;
    if (!g_pserver->rdb_pipe_buff)
        g_pserver->rdb_pipe_buff = (char*)zmalloc(PROTO_IOBUF_LEN);
    serverAssert(g_pserver->rdb_pipe_numconns_writing==0);
    serverAssert(eventLoop == g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el);

    while (1) {
        g_pserver->rdb_pipe_bufflen = read(fd, g_pserver->rdb_pipe_buff, PROTO_IOBUF_LEN);
        if (g_pserver->rdb_pipe_bufflen < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                return;
            serverLog(LL_WARNING,"Diskless rdb transfer, read error sending DB to replicas: %s", strerror(errno));
            for (i=0; i < g_pserver->rdb_pipe_numconns; i++) {
                connection *conn = g_pserver->rdb_pipe_conns[i];
                if (!conn)
                    continue;
                client *slave = (client*)connGetPrivateData(conn);
                freeClientAsync(slave);
                g_pserver->rdb_pipe_conns[i] = NULL;
            }
            killRDBChild();
            return;
        }

        if (g_pserver->rdb_pipe_bufflen == 0 && !g_pserver->rdbThreadVars.fRdbThreadCancel) {
            /* EOF - write end was closed. */
            int stillUp = 0;
            aeDeleteFileEvent(eventLoop, g_pserver->rdb_pipe_read, AE_READABLE);
            for (i=0; i < g_pserver->rdb_pipe_numconns; i++)
            {
                connection *conn = g_pserver->rdb_pipe_conns[i];
                if (!conn)
                    continue;
                stillUp++;
            }
            serverLog(LL_WARNING,"Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp);
            /* Now that the replicas have finished reading, notify the child that it's safe to exit. 
             * When the server detectes the child has exited, it can mark the replica as online, and
             * start streaming the replication buffers. */
            close(g_pserver->rdb_child_exit_pipe);
            g_pserver->rdb_child_exit_pipe = -1;
            return;
        }

        int stillAlive = 0;
        for (i=0; i < g_pserver->rdb_pipe_numconns; i++)
        {
            int nwritten;
            connection *conn = g_pserver->rdb_pipe_conns[i];
            if (!conn)
                continue;

            client *slave = (client*)connGetPrivateData(conn);
            std::unique_lock<fastlock> ul(slave->lock);