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

Code before:
    long long min_time = now_since_epoch - time_per_burst_;
    if (min_time > last_time_) {
        new_time = min_time + (time_per_token_ * tokens);
    } else {
        new_time = last_time_ + (time_per_token_ * tokens);
    }

    long long delay_time = 0;
    if (new_time > now_since_epoch) {
        delay_time = new_time - now_since_epoch;
    } else {
        config.last_time = new_time;
    }
    if (config.num_threads) pthread_mutex_unlock(&(config.token_bucket_mutex));
    return delay_time;
}

static void clientDone(client c) {
    int requests_finished = atomic_load_explicit(&config.requests_finished, memory_order_relaxed);
    if (requests_finished >= config.requests) {
        freeClient(c);
        if (!config.num_threads && config.el) aeStop(config.el);
        return;
    }
    if (config.keepalive) {
        resetClient(c);
    } else {
        if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));
        config.liveclients--;
        createMissingClients(c);
        config.liveclients++;
        if (config.num_threads) pthread_mutex_unlock(&(config.liveclients_mutex));
        freeClient(c);
    }
}

static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    client c = privdata;
    void *reply = NULL;
    UNUSED(el);
    UNUSED(fd);
    UNUSED(mask);

    /* Calculate latency only for the first read event. This means that the
     * server already sent the reply and we need to parse it. Parsing overhead
     * is not part of the latency, so calculate it only once, here. */
    if (c->latency < 0) c->latency = ustime() - (c->start);

    if (redisBufferRead(c->context) != REDIS_OK) {
        fprintf(stderr, "Error: %s\n", c->context->errstr);
        exit(1);

Review comment (suggestion):
looks like the intent is to support bursting for up to 1s. the logic is a bit hard to follow. also I think we can use atomic variable instead. something like this:

```c
long long acquireTokenOrWait(RateLimiterConfig* config, int tokens) {
    long long new_scheduled_time;
    long long current_last_time;

    /* request tokens */
    while (1) {
        current_last_time = atomic_load_explicit(&config->last_time_ns, memory_order_relaxed);
        long long now = nstime();
        long long start_time = max(current_last_time, now - config->time_per_burst_ns);
        new_scheduled_time = start_time + (tokens * config->time_per_token_ns);
        if (atomic_compare_exchange_weak_explicit(
                &config->last_time_ns, // The atomic object to modify
                &current_last_time,    // The expected value we read earlier (address of)
                new_scheduled_time,    // The new value to write if the check passes
                memory_order_release,
                memory_order_relaxed)) {
            break;
        }
    }

    /* compute delays */
    long long now = nstime();
    long long delay_ns = 0;
    if (new_scheduled_time > now) {
        delay_ns = new_scheduled_time - now;
    }

    return delay_ns;
}
```

Code after:
}

/* Acquires the specified number of tokens from the token bucket or calculates the wait time if tokens are not available.
 * This function implements a token bucket rate limiting algorithm to control access to a resource.
 *
 * The tokens parameter is the number of tokens to acquire.
 *
 * Returns the delay time in milliseconds that the caller should wait before proceeding, or 0 if tokens are immediately available.
 *
 * Token Bucket Algorithm Explanation:
 * - The token bucket algorithm allows a certain number of tokens to be accumulated over time, which can then be used to control the rate of requests.
 * - Due to the time event only allowing a delay of 1ms, a request for the next 1ms is issued.
 *
 * The function is thread-safe. */
static long long acquireTokenOrWait(int tokens) {
    uint64_t time_per_token = config.time_per_token;
    uint64_t time_per_burst = config.time_per_burst;
    uint64_t new_time = 0;
    uint64_t now_epoch, next_epoch, min_time, delay_time;
    uint64_t last_time_ns, old_last_time_ns;

    while (1) {
        old_last_time_ns = atomic_load_explicit(&config.last_time_ns, memory_order_relaxed);
        last_time_ns = old_last_time_ns;
        now_epoch = nstime();

        // If the last_time_ns is 0, it means this is the first request, so we set it to now_epoch.
        if (last_time_ns == 0) {
            last_time_ns = now_epoch;
        }

        next_epoch = now_epoch + 1000000;
        min_time = next_epoch - time_per_burst;

        if (min_time > last_time_ns) { // if the last time is too old, reset it
            new_time = min_time + (time_per_token * tokens);
        } else {
            new_time = last_time_ns + (time_per_token * tokens);
        }

        delay_time = 0;
        if (new_time > next_epoch) { // if the new time is in the next epoch, we need to wait
            delay_time = new_time - now_epoch;
        } else {
            last_time_ns = new_time;
        }

        if (atomic_compare_exchange_weak_explicit(
                &config.last_time_ns,
                &old_last_time_ns,
                last_time_ns,