Repository: valkey-io/valkey
File: src/networking.c (C)

Code before:
    c->is_qb_full_read = (size_t)c->nread == readlen ? 1 : 0;

    sdsIncrLen(c->querybuf, c->nread);
    qblen = sdslen(c->querybuf);
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    if (!is_primary) {
        /* The commands cached in the MULTI/EXEC queue have not been executed yet,
         * so they are also considered a part of the query buffer in a broader sense.
         *
         * For unauthenticated clients, the query buffer cannot exceed 1MB at most. */
        size_t qb_memory = sdslen(c->querybuf) + (c->mstate ? c->mstate->argv_len_sums : 0);
        if (qb_memory > server.client_max_querybuf_len ||
            (qb_memory > 1024 * 1024 && (c->read_flags & READ_FLAGS_AUTH_REQUIRED))) {
            c->read_flags |= READ_FLAGS_QB_LIMIT_REACHED;
        }
    }
}

/**
 * This function is designed to prioritize replication flow.
 * Determines whether the replica should continue reading from the primary.
 * It dynamically adjusts the read rate based on buffer utilization
 *
 * @return          1 if another read should be attempted, 0 otherwise.
 */
int shouldRepeatRead(client *c, int iteration) {
    // If the client is not a primary replica, is closing, or flow control is disabled, no more reads.
    if (!(c->flag.primary) || c->flag.close_asap || !server.repl_flow_control_enabled) {
        return 0;
    }

    if (iteration < server.repl_max_reads_per_io_event &&
        c->is_qb_full_read) {
        return 1;
    }

    return 0;
}


void readQueryFromClient(connection *conn) {
    client *c = connGetPrivateData(conn);
    /* Check if we can send the client to be handled by the IO-thread */
    if (postponeClientRead(c)) return;

    if (c->io_write_state != CLIENT_IDLE || c->io_read_state != CLIENT_IDLE) return;

    bool shouldRepeat = false;
    int iter = 0;
    do {
        readToQueryBuf(c);

Review comment (suggestion):
```suggestion
static bool shouldRepeatReadFromPrimary(client *c, int iteration) {
```
Kind of a nit, I think generic function names are often confusing when it does something very specific.

Code after:

        parseCommand(c);

        if (handleParseResults(c) != PARSE_OK) {
            break;
        }

        if (c->argc == 0) {
            /* No command to process - continue parsing the query buf. */
            continue;
        }

        if (c->querybuf == thread_shared_qb) {
            /* Before processing the command, reset the shared query buffer to its default state.
             * This avoids unintentionally modifying the shared qb during processCommand as we may use
             * the shared qb for other clients during processEventsWhileBlocked */
            resetSharedQueryBuf(c);
        }

        /* We are finally ready to execute the command. */
        if (processCommandAndResetClient(c) == C_ERR) {
            /* If the client is no longer valid, we avoid exiting this
             * loop and trimming the client buffer later. So we return
             * ASAP in that case. */
            return C_ERR;
        }
    }

    return C_OK;
}

/* This function can be called from the main-thread or from the IO-thread.
 * The function allocates query-buf for the client if required and reads to it from the network.
 * It will set c->nread to the bytes read from the network.
 * Returns true if the buffer was filled (more data may be available). */
static bool readToQueryBuf(client *c) {
    int big_arg = 0;
    size_t qblen, readlen;

    /* If the replica RDB client is marked as closed ASAP, do not try to read from it */
    if (c->flag.close_asap) return false;

    int is_primary = c->read_flags & READ_FLAGS_PRIMARY;

    readlen = PROTO_IOBUF_LEN;
    qblen = c->querybuf ? sdslen(c->querybuf) : 0;
    /* If this is a multi bulk request, and we are processing a bulk reply
     * that is large enough, try to maximize the probability that the query
     * buffer contains exactly the SDS string representing the object, even
     * at the risk of requiring more read(2) calls. This way the function
     * processMultiBulkBuffer() can avoid copying buffers to create the