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

Code before:
         * 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
 * and ensures replication reads are not overly aggressive.
 *
 * @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;
    }

    bool is_last_iteration = iteration >= server.repl_cur_reads_per_io_event;

    if (is_last_iteration) {
        /* If the last read filled the buffer AND enough time has passed since the last increase:
         * - Increase the read rate, up to a max limit.
         * - This ensures a gradual ramp-up instead of an overly aggressive approach. */
        if (c->qb_full_read && server.mstime - server.repl_last_rate_update > 100) {
            server.repl_cur_reads_per_io_event = MIN(server.repl_max_reads_per_io_event,
                                                     server.repl_cur_reads_per_io_event + 1);
            server.repl_last_rate_update = server.mstime; // Update the last increase timestamp.
        }
    } else {
        /* If the last read completely filled the buffer, continue reading. */
        if (c->qb_full_read) {
            return 1;
        }

        /* If the buffer was NOT fully filled, it indicates less replication pressure.
         * Reduce the read rate to avoid excessive polling and free up resources for other clients. */
        server.repl_cur_reads_per_io_event = MAX(1, server.repl_cur_reads_per_io_event - 1);
    }

    /* Stop reading for now (if we reached this point, conditions to continue were not met). */
    return 0;
}


void readQueryFromClient(connection *conn) {

Review comment (suggestion):
I suppose I don't really understand why this needs to be adaptive. As long as we got a full read, why can't we repeat up until the max io event size? There is a comment about "The read limit ramps up gradually if full reads continue, avoiding sudden spikes.", but we are just deferring the spikes until later. Most users don't like replication lag, they would rather have the most up to date data if possible.

Code after:
            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
     * robj representing the argument. */

    if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1 && c->bulklen >= PROTO_MBULK_BIG_ARG) {
        ssize_t remaining = (size_t)(c->bulklen + 2) - (qblen - c->qb_pos);
        big_arg = 1;

        /* Note that the 'remaining' variable may be zero in some edge case,
         * for example once we resume a blocked client after CLIENT PAUSE. */
        if (remaining > 0) readlen = remaining;