Function flushReplBacklogToClients

src/replication.cpp:5611–5688

void _clientAsyncReplyBufferReserve(client *c, size_t len);

void flushReplBacklogToClients()
{
    serverAssert(GlobalLocksAcquired());
    /* If we have the repl backlog lock, we will deadlock */
    serverAssert(!g_pserver->repl_backlog_lock.fOwnLock());
    if (g_pserver->repl_batch_offStart < 0)
        return;
    
    if (g_pserver->repl_batch_offStart != g_pserver->master_repl_offset) {
        bool fAsyncWrite = false;
        long long min_offset = LLONG_MAX;
        // Ensure no overflow
        serverAssert(g_pserver->repl_batch_offStart < g_pserver->master_repl_offset);
        if (g_pserver->master_repl_offset - g_pserver->repl_batch_offStart > g_pserver->repl_backlog_size) {
            // We overflowed
            listIter li;
            listNode *ln;
            listRewind(g_pserver->slaves, &li);
            while ((ln = listNext(&li))) {
                client *c = (client*)listNodeValue(ln);
                sds sdsClient = catClientInfoString(sdsempty(),c);
                freeClientAsync(c);
                serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", sdsClient);
                sdsfree(sdsClient);
            }
            goto LDone;
        }

        // Ensure no overflow if we get here
        serverAssert(g_pserver->master_repl_offset - g_pserver->repl_batch_offStart <= g_pserver->repl_backlog_size);
        serverAssert(g_pserver->repl_batch_idxStart != g_pserver->repl_backlog_idx);

        // Repl backlog writes must become visible to all threads at this point
        std::atomic_thread_fence(std::memory_order_release);

        listIter li;
        listNode *ln;
        listRewind(g_pserver->slaves, &li);
        /* We don't actually write any data in this function since we send data 
         * directly from the replication backlog to replicas in writeToClient.
         * 
         * What we do however, is set the end offset of each replica here. This way, 
         * future calls to writeToClient will know up to where in the replication
         * backlog is valid for writing. */  
        while ((ln = listNext(&li))) {
            client *replica = (client*)listNodeValue(ln);

            if (!canFeedReplicaReplBuffer(replica)) continue;
            if (replica->flags & CLIENT_CLOSE_ASAP) continue;

            std::unique_lock<fastlock> ul(replica->lock);
            if (!FCorrectThread(replica))
                fAsyncWrite = true;

            /* We should have set the repl_curr_off when synchronizing, so it shouldn't be -1 here */
            serverAssert(replica->repl_curr_off != -1);

            min_offset = std::min(min_offset, replica->repl_curr_off);