Function ProcessPendingAsyncWrites

src/networking.cpp:1983–2061

}

void ProcessPendingAsyncWrites()
{
    if (serverTL == nullptr)
        return; // module fake call

    serverAssert(GlobalLocksAcquired());

    while(listLength(serverTL->clients_pending_asyncwrite)) {
        client *c = (client*)listNodeValue(listFirst(serverTL->clients_pending_asyncwrite));
        listDelNode(serverTL->clients_pending_asyncwrite, listFirst(serverTL->clients_pending_asyncwrite));
        std::lock_guard<decltype(c->lock)> lock(c->lock);

        serverAssert(c->fPendingAsyncWrite);
        if (c->flags & (CLIENT_CLOSE_ASAP | CLIENT_CLOSE_AFTER_REPLY))
        {
            if (c->replyAsync != nullptr){
                zfree(c->replyAsync);
                c->replyAsync = nullptr;
            }
            c->fPendingAsyncWrite = FALSE;
            continue;
        }

        /* since writes from master to replica can come directly from the replication backlog,
         * writes may have been signalled without having been copied to the replyAsync buffer,
         * thus causing the buffer to be NULL */ 
        if (c->replyAsync != nullptr){
            size_t size = c->replyAsync->used;

            if (listLength(c->reply) == 0 && size <= static_cast<size_t>(PROTO_REPLY_CHUNK_BYTES - c->bufpos)) {
                memcpy(c->buf + c->bufpos, c->replyAsync->buf(), size);
                c->bufpos += size;
            } else {
                c->reply_bytes += c->replyAsync->size;
                listAddNodeTail(c->reply, c->replyAsync);
                c->replyAsync = nullptr;
            }

            zfree(c->replyAsync);
            c->replyAsync = nullptr;
        } else {
            /* Only replicas should have empty async reply buffers */
            serverAssert(c->flags & CLIENT_SLAVE);
        }

        c->fPendingAsyncWrite = FALSE;

        if (!((c->replstate == REPL_STATE_NONE || c->replstate == SLAVE_STATE_FASTSYNC_TX ||
         (c->replstate == SLAVE_STATE_ONLINE && !c->repl_put_online_on_ack))))
            continue;

        closeClientOnOutputBufferLimitReached(c, 1);
        if (c->flags & CLIENT_CLOSE_ASAP)
            continue;   // we will never write this so don't post an op

        std::atomic_thread_fence(std::memory_order_seq_cst);

        if (FCorrectThread(c))