Function handleClientsWithPendingWrites

src/networking.cpp:2067–2130

This function is called just before entering the event loop, in the hope * we can just write the replies to the client output buffer without any * need to use a syscall in order to install the writable event handler, * get it called, and so forth. */

 * need to use a syscall in order to install the writable event handler,
 * get it called, and so forth. */
int handleClientsWithPendingWrites(int iel, int aof_state) {
    int processed = 0;
    serverAssert(iel == (serverTL - g_pserver->rgthreadvar));

    if (listLength(serverTL->clients_pending_asyncwrite))
    {
        AeLocker locker;
        locker.arm(nullptr);
        ProcessPendingAsyncWrites();
    }

    int ae_flags = AE_WRITABLE|AE_WRITE_THREADSAFE;
    /* For the fsync=always policy, we want that a given FD is never
        * served for reading and writing in the same event loop iteration,
        * so that in the middle of receiving the query, and serving it
        * to the client, we'll call beforeSleep() that will do the
        * actual fsync of AOF to disk. AE_BARRIER ensures that. */
    if (aof_state == AOF_ON &&
        g_pserver->aof_fsync == AOF_FSYNC_ALWAYS)
    {
        ae_flags |= AE_BARRIER;
    }

    std::unique_lock<fastlock> lockf(g_pserver->rgthreadvar[iel].lockPendingWrite);
    auto vec = std::move(g_pserver->rgthreadvar[iel].clients_pending_write);
    lockf.unlock();
    processed += (int)vec.size();

    for (client *c : vec) {
        serverAssertDebug(FCorrectThread(c));

        uint64_t flags = c->flags.fetch_and(~CLIENT_PENDING_WRITE, std::memory_order_relaxed);

        /* If a client is protected, don't do anything,
        * that may trigger write error or recreate handler. */
        if ((flags & CLIENT_PROTECTED) && !(flags & CLIENT_SLAVE)) continue;

        /* Don't write to clients that are going to be closed anyway. */
        if (c->flags & CLIENT_CLOSE_ASAP) continue;

        /* Try to write buffers to the client socket, unless its a replica in multithread mode */
        if (writeToClient(c,0) == C_ERR) 
        {
            if (c->flags & CLIENT_CLOSE_ASAP)
            {
                AeLocker ae;
                ae.arm(nullptr);
                freeClient(c); // writeToClient will only async close, but there's no need to wait
            }
            continue;
        }

        /* If after the synchronous writes above we still have data to
        * output to the client, we need to install the writable handler. */
        std::unique_lock<decltype(c->lock)> lock(c->lock);
        if (clientHasPendingReplies(c)) {
            if (connSetWriteHandlerWithBarrier(c->conn, sendReplyToClient, ae_flags, true) == C_ERR) {
                freeClientAsync(c);