Function writeToClient

src/networking.cpp:1821–1965

Write data in output buffers to client. Return C_OK if the client * is still valid after the call, C_ERR if it was freed because of some * error. If handler_installed is set, it will attempt to clear the * write event. * * This function is called by threads, but always with handler_installed * set to 0. So when handler_installed is set to 0 the function must be * thread safe. */

 * set to 0. So when handler_installed is set to 0 the function must be
 * thread safe. */
int writeToClient(client *c, int handler_installed) {
    /* Update total number of writes on server */
    g_pserver->stat_total_writes_processed.fetch_add(1, std::memory_order_relaxed);

    ssize_t nwritten = 0, totwritten = 0;
    clientReplyBlock *o;
    serverAssertDebug(FCorrectThread(c));

    std::unique_lock<decltype(c->lock)> lock(c->lock);

    /* We can only directly read from the replication backlog if the client 
       is a replica, so only attempt to do so if that's the case. */
    if (c->flags & CLIENT_SLAVE && !(c->flags & CLIENT_MONITOR) && c->replstate == SLAVE_STATE_ONLINE) {
        std::unique_lock<fastlock> repl_backlog_lock (g_pserver->repl_backlog_lock);
        // Ensure all writes to the repl backlog are visible
        std::atomic_thread_fence(std::memory_order_acquire);

        while (clientHasPendingReplies(c)) {
            long long repl_end_idx = getReplIndexFromOffset(c->repl_end_off);
            serverAssert(c->repl_curr_off != -1);

            if (c->repl_curr_off != c->repl_end_off){
                long long repl_curr_idx = getReplIndexFromOffset(c->repl_curr_off); 
                long long nwritten2ndStage = 0; /* How much was written from the start of the replication backlog
                                                * in the event of a wrap around write */
                /* normal case with no wrap around */
                if (repl_end_idx >= repl_curr_idx){
                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, repl_end_idx - repl_curr_idx);
                /* wrap around case */
                } else {
                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, g_pserver->repl_backlog_size - repl_curr_idx);
                    /* only attempt wrapping if we write the correct number of bytes */
                    if (nwritten == g_pserver->repl_backlog_size - repl_curr_idx){
                        nwritten2ndStage = connWrite(c->conn, g_pserver->repl_backlog, repl_end_idx);
                        if (nwritten2ndStage != -1)
                            nwritten += nwritten2ndStage;
                    }                
                }

                /* only increment bytes if an error didn't occur */
                if (nwritten > 0){
                    totwritten += nwritten;
                    c->repl_curr_off += nwritten;
                    serverAssert(c->repl_curr_off <= c->repl_end_off);
                }

                /* If the second part of a write didn't go through, we still need to register that */
                if (nwritten2ndStage == -1) nwritten = -1;
                if (nwritten == -1)
                    break;
            } else {
                break;
            }
        }
    } else {
        while(clientHasPendingReplies(c)) {
            if (c->bufpos > 0) {
                auto bufpos = c->bufpos;