/* Server benchmark utility.
 *
 * Copyright (c) 2009-2012, Redis Ltd.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   * Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *   * Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *   * Neither the name of Redis nor the names of its contributors may be used
 *     to endorse or promote products derived from this software without
 *     specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "fmacros.h"

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <time.h>
#include <sys/time.h>
#include <signal.h>
#include <assert.h>
#include <math.h>
#include <pthread.h>
#include <stdatomic.h>

#include "sds.h"
#include "ae.h"
#include <valkey/valkey.h>
#ifdef USE_OPENSSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#include <valkey/tls.h>
#endif
#ifdef USE_RDMA
#include <valkey/rdma.h>
#endif
#include "adlist.h"
#include "dict.h"
#include "zmalloc.h"
#include "crc16_slottable.h"
#include "hdr_histogram.h"
#include "cli_common.h"
#include "mt19937-64.h"
#include "valkey-benchmark-dataset.h"

#define UNUSED(V) ((void)V)
#define RANDPTR_INITIAL_SIZE 8
#define DEFAULT_LATENCY_PRECISION 3
#define MAX_LATENCY_PRECISION 4
#define MAX_THREADS 500
#define CLUSTER_SLOTS 16384
#define CONFIG_LATENCY_HISTOGRAM_MIN_VALUE 10L               /* >= 10 usecs */
#define CONFIG_LATENCY_HISTOGRAM_MAX_VALUE 60000000L         /* <= 60 secs(us precision) */
#define CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE 60000000L /* <= 60 secs(us precision) */
#define SHOW_THROUGHPUT_INTERVAL 250                         /* 250ms */

#define CLIENT_GET_EVENTLOOP(c) (c->thread_id >= 0 ? config.threads[c->thread_id]->el : config.el)

#define PLACEHOLDER_COUNT 10
static const size_t PLACEHOLDER_LEN = 12; // length of BENCHMARK_PLACEHOLDERS strings
/* Cap on -r. Keys are written into PLACEHOLDER_LEN-wide fixed slots, so
 * bumping this requires widening the placeholder handling too. */
#define MAX_KEYSPACELEN 999999999999LL
static const char *PLACEHOLDERS[PLACEHOLDER_COUNT] = {
    "__rand_int__", "__rand_1st__", "__rand_2nd__", "__rand_3rd__", "__rand_4th__",
    "__rand_5th__", "__rand_6th__", "__rand_7th__", "__rand_8th__", "__rand_9th__"};

struct benchmarkThread;
struct clusterNode;
struct serverConfig;

/* Read from replica options */
typedef enum readFromReplica {
    FROM_PRIMARY_ONLY = 0, /* default option */
    FROM_REPLICA_ONLY,
    FROM_ALL
} readFromReplica;

/* Fuzz mode flags */
#define FUZZ_MODE_MALFORMED_COMMANDS (1 << 0)
#define FUZZ_MODE_CONFIG_COMMANDS (1 << 1)

static struct config {
    aeEventLoop *el;
    enum valkeyConnectionType ct;
    cliConnInfo conn_info;
    int tls;
    int mptcp;
    struct cliSSLconfig sslconfig;
    int numclients;
    _Atomic int liveclients;
    int requests;
    int duration;
    int warmup_duration;
    _Atomic int current_warmup_duration;
    _Atomic int requests_issued;
    _Atomic int requests_finished;
    _Atomic int previous_requests_finished;
    int last_printed_bytes;
    long long previous_tick;
    int keysize;
    int datasize;
    int replace_placeholders;
    long long keyspacelen;
    int sequential_replacement;
    int keepalive;
    int pipeline;
    long long start;
    long long totlatency;
    const char *title;
    list *clients;
    list *paused_clients;
    int quiet;
    int csv;
    int loop;
    int idlemode;
    sds input_dbnumstr;
    char *tests;
    int stdinarg; /* get last arg from stdin. (-x option) */
    int precision;
    int num_threads;
    struct benchmarkThread **threads;
    int cluster_mode;
    int fuzz_mode; /* Boolean flag to enable fuzzing */
    const char *fuzz_log_level;
    int fuzz_flags; /* Bit flags for fuzzing modes */
    readFromReplica read_from_replica;
    int cluster_node_count;
    struct clusterNode **cluster_nodes;
    struct serverConfig *server_config;
    struct hdr_histogram *latency_histogram;
    struct hdr_histogram *current_sec_latency_histogram;
    struct hdr_histogram *rps_histogram;
    _Atomic int is_fetching_slots;
    _Atomic int is_updating_slots;
    _Atomic int slots_last_update;
    int enable_tracking;
    int num_functions;
    int num_keys_in_fcall;
    pthread_mutex_t liveclients_mutex;
    pthread_mutex_t is_updating_slots_mutex;
    int resp3; /* use RESP3 */
    int rps;
    atomic_uint_fast64_t last_time_ns;
    uint64_t time_per_token;
    uint64_t time_per_burst;
    /* Dataset support */
    sds dataset_file;
    int max_documents;        /* Maximum documents to load from dataset */
    dataset *current_dataset; /* Current loaded dataset */
    /* Command template for dataset mode */
    int template_argc;
    sds *template_argv;
    int has_field_placeholders;
} config;

/* Locations of the placeholders __rand_int__, __rand_1st__,
 * __rand_2nd, etc. within the RESP encoded command buffer. */
static struct placeholders {
    size_t cmd_len;                     /* length of the command */
    size_t count[PLACEHOLDER_COUNT];    /* number of each placeholder in the command */
    size_t *indices[PLACEHOLDER_COUNT]; /* pointer to indices for each placeholder */
    size_t *index_data;                 /* allocation holding all index data */
} placeholders;

/* Sequence keys for dataset command generation */
static _Atomic uint64_t dataset_seq_key[PLACEHOLDER_COUNT] = {0};

typedef struct _client {
    valkeyContext *context;
    sds obuf;
    char **stagptr;     /* Pointers to slot hashtags (cluster mode only) */
    size_t staglen;     /* Number of pointers in client->stagptr */
    size_t stagfree;    /* Number of unused pointers in client->stagptr */
    size_t written;     /* Bytes of 'obuf' already written */
    long long start;    /* Start time of a request */
    long long latency;  /* Request latency */
    int seqlen;         /* Number of commands in the command sequence */
    int pending;        /* Number of pending requests (replies to consume) */
    int prefix_pending; /* If non-zero, number of pending prefix commands. Commands
                           such as auth and select are prefixed to the pipeline of
                           benchmark commands and discarded after the first send. */
    int prefixlen;      /* Size in bytes of the pending prefix commands */
    int thread_id;
    struct clusterNode *cluster_node;
    int slots_last_update;
    uint64_t paused : 1;
    uint64_t reuse : 1;
    uint64_t request_started : 1; /* Request initialized, even if no bytes were written. */
} *client;

/* Threads. */

typedef struct benchmarkThread {
    int index;
    pthread_t thread;
    aeEventLoop *el;
    list *paused_clients;
    /* Per-thread latency histograms: recording into one shared histogram
     * with hdr_record_value_atomic makes every command from every thread
     * contend on the same counter cache lines. Each thread records into
     * its own histograms lock-free; thread 0 folds them for live display
     * and the main thread folds them after join for the final report. */
    struct hdr_histogram *latency_histogram;
    struct hdr_histogram *current_sec_latency_histogram;
} benchmarkThread;

/* Cluster. */
typedef struct clusterNode {
    char *ip;
    int port;
    sds name;
    int flags;
    sds replicate; /* Primary ID if node is a replica */
    int *slots;
    int slots_count;
    int *updated_slots;      /* Used by updateClusterSlotsConfiguration */
    int updated_slots_count; /* Used by updateClusterSlotsConfiguration */
    int replicas_count;
    struct serverConfig *server_config;
} clusterNode;

typedef struct serverConfig {
    sds save;
    sds appendonly;
} serverConfig;

/* Register a file event. For RDMA with AE_WRITABLE, invoke the handler once after
 * registering: this fd is not POLLOUT-driven like TCP, so the loop may never deliver
 * AE_WRITABLE without an explicit kick.
 *
 * Non-RDMA: same as aeCreateFileEvent. Read-side wakeups when libvalkey drains the CQ on
 * the write path require libvalkey to notify the outer poll (valkey-io/libvalkey#301). */
static inline void createFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData) {
    aeCreateFileEvent(eventLoop, fd, mask, proc, clientData);
    if (config.ct == VALKEY_CONN_RDMA && (mask & AE_WRITABLE)) {
        proc(eventLoop, fd, clientData, mask);
    }
}

/* Prototypes */
static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask);
static void createMissingClients(client c);
static benchmarkThread *createBenchmarkThread(int index);
static void freeBenchmarkThread(benchmarkThread *thread);
static void freeBenchmarkThreads(void);
static void *execBenchmarkThread(void *ptr);
static clusterNode *createClusterNode(char *ip, int port);
static serverConfig *getServerConfig(enum valkeyConnectionType ct, const char *ip_or_path, int port);
static valkeyContext *getValkeyContext(enum valkeyConnectionType ct, const char *ip_or_path, int port);
static void freeServerConfig(serverConfig *cfg);
static int fetchClusterSlotsConfiguration(client c);
static void updateClusterSlotsConfiguration(void);
static long long showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData);
int runFuzzerClients(const char *host, int port, int max_commands, int parallel_clients, int cluster_mode, long long num_keys, cliSSLconfig *ssl_config, const char *log_level, int fuzz_flags);
static int parseCommandTemplate(int argc, char **argv);

/* Dict callbacks */
static uint64_t dictSdsHash(const void *key);
static int dictSdsKeyCompare(const void *key1, const void *key2);

/* Implementation */
static long long ustime(void) {
    struct timeval tv;
    long long ust;

    gettimeofday(&tv, NULL);
    ust = ((long long)tv.tv_sec) * 1000000;
    ust += tv.tv_usec;
    return ust;
}

static long long mstime(void) {
    return ustime() / 1000;
}

static long long nstime(void) {
    struct timespec ts;
    clock_gettime(CLOCK_MONOTONIC, &ts);
    return (long long)ts.tv_sec * 1000000000LL + ts.tv_nsec;
}

static bool isBenchmarkFinished(int request_count) {
    /* don't end in warmup period */
    int warmup_duration = atomic_load_explicit(&config.current_warmup_duration, memory_order_relaxed);
    if (warmup_duration > 0) return false;

    if (config.duration > 0) {
        /* end after the specified duration */
        if ((mstime() - config.start) >= (config.duration * 1000LL)) return true;
    } else {
        /* end after the specified number of requests */
        if (request_count >= config.requests) return true;
    }
    return false;
}

static uint64_t dictSdsHash(const void *key) {
    return dictGenHashFunction(key, sdslen(key));
}

static int dictSdsKeyCompare(const void *key1, const void *key2) {
    int l1, l2;
    l1 = sdslen((sds)key1);
    l2 = sdslen((sds)key2);
    if (l1 != l2) return 0;
    return memcmp(key1, key2, l1) == 0;
}

static dictType dtype = {
    .entryGetKey = dictEntryGetKey,
    .hashFunction = dictSdsHash,
    .keyCompare = dictSdsKeyCompare,
    .entryDestructor = zfree,
};

static valkeyContext *getValkeyContext(enum valkeyConnectionType ct, const char *ip_or_path, int port) {
    valkeyContext *ctx = NULL;
    valkeyReply *reply = NULL;
    struct timeval tv = {0};
    ctx = valkeyConnectWrapper(ct, ip_or_path, port, tv, 0, config.mptcp);
    if (ctx == NULL || ctx->err) {
        fprintf(stderr, "Could not connect to server at ");
        char *err = (ctx != NULL ? ctx->errstr : "");
        if (ct != VALKEY_CONN_UNIX)
            fprintf(stderr, "%s:%d: %s\n", ip_or_path, port, err);
        else
            fprintf(stderr, "%s: %s\n", ip_or_path, err);
        goto cleanup;
    }
    if (config.tls == 1) {
        const char *err = NULL;
        if (cliSecureConnection(ctx, config.sslconfig, &err) == VALKEY_ERR && err) {
            fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
            goto cleanup;
        }
    }
    if (config.conn_info.auth == NULL) return ctx;
    if (config.conn_info.user == NULL)
        reply = valkeyCommand(ctx, "AUTH %s", config.conn_info.auth);
    else
        reply = valkeyCommand(ctx, "AUTH %s %s", config.conn_info.user, config.conn_info.auth);
    if (reply != NULL) {
        if (reply->type == VALKEY_REPLY_ERROR) {
            if (ct != VALKEY_CONN_UNIX)
                fprintf(stderr, "Node %s:%d replied with error:\n%s\n", ip_or_path, port, reply->str);
            else
                fprintf(stderr, "Node %s replied with error:\n%s\n", ip_or_path, reply->str);
            freeReplyObject(reply);
            valkeyFree(ctx);
            exit(1);
        }
        freeReplyObject(reply);
        return ctx;
    }
    fprintf(stderr, "ERROR: failed to fetch reply from ");
    if (ct != VALKEY_CONN_UNIX)
        fprintf(stderr, "%s:%d\n", ip_or_path, port);
    else
        fprintf(stderr, "%s\n", ip_or_path);
cleanup:
    freeReplyObject(reply);
    valkeyFree(ctx);
    return NULL;
}


static serverConfig *getServerConfig(enum valkeyConnectionType ct, const char *ip_or_path, int port) {
    serverConfig *cfg = zcalloc(sizeof(*cfg));
    if (!cfg) return NULL;
    valkeyContext *c = NULL;
    valkeyReply *reply = NULL, *sub_reply = NULL;
    c = getValkeyContext(ct, ip_or_path, port);
    if (c == NULL) {
        freeServerConfig(cfg);
        exit(1);
    }
    valkeyAppendCommand(c, "CONFIG GET %s", "save");
    valkeyAppendCommand(c, "CONFIG GET %s", "appendonly");
    int abort_test = 0;
    int i = 0;
    void *r = NULL;
    for (; i < 2; i++) {
        int res = valkeyGetReply(c, &r);
        if (reply) freeReplyObject(reply);
        reply = res == VALKEY_OK ? ((valkeyReply *)r) : NULL;
        if (res != VALKEY_OK || !r) goto fail;
        if (reply->type == VALKEY_REPLY_ERROR) {
            goto fail;
        }
        if (reply->type != VALKEY_REPLY_ARRAY || reply->elements < 2) goto fail;
        sub_reply = reply->element[1];
        char *value = sub_reply->str;
        if (!value) value = "";
        switch (i) {
        case 0: cfg->save = sdsnew(value); break;
        case 1: cfg->appendonly = sdsnew(value); break;
        }
    }
    freeReplyObject(reply);
    valkeyFree(c);
    return cfg;
fail:
    if (reply && reply->type == VALKEY_REPLY_ERROR && !strncmp(reply->str, "NOAUTH", 6)) {
        if (ct != VALKEY_CONN_UNIX)
            fprintf(stderr, "Node %s:%d replied with error:\n%s\n", ip_or_path, port, reply->str);
        else
            fprintf(stderr, "Node %s replied with error:\n%s\n", ip_or_path, reply->str);
        abort_test = 1;
    }
    freeReplyObject(reply);
    valkeyFree(c);
    freeServerConfig(cfg);
    if (abort_test) exit(1);
    return NULL;
}
static void freeServerConfig(serverConfig *cfg) {
    if (cfg->save) sdsfree(cfg->save);
    if (cfg->appendonly) sdsfree(cfg->appendonly);
    zfree(cfg);
}

void resetPlaceholders(void) {
    if (placeholders.index_data)
        zfree(placeholders.index_data); /* indices are a single contiguous allocation */
    memset(&placeholders, 0, sizeof(placeholders));
}

void initPlaceholders(const char *cmd, size_t cmd_len) {
    resetPlaceholders();
    placeholders.cmd_len = cmd_len;

    /* store placeholder locations in temp arrays */
    size_t total_count = 0;
    size_t *temp_indices[PLACEHOLDER_COUNT];
    for (size_t placeholder = 0; placeholder < PLACEHOLDER_COUNT; placeholder++) {
        size_t *count = &placeholders.count[placeholder];
        *count = 0;

        size_t temp_size = RANDPTR_INITIAL_SIZE;
        temp_indices[placeholder] = zmalloc(sizeof(size_t) * temp_size);
        const char *p = cmd;
        const char *end = cmd + cmd_len;
        while ((p = strstr(p, PLACEHOLDERS[placeholder])) != NULL && p < end) {
            if (*count == temp_size) {
                temp_size *= 2;
                temp_indices[placeholder] = zrealloc(temp_indices[placeholder], sizeof(size_t) * temp_size);
            }
            size_t index = p - cmd;
            temp_indices[placeholder][*count] = index;
            (*count)++;
            total_count++;
            p += PLACEHOLDER_LEN; // Move past the placeholder
        }
    }

    /* consolidate temp data into contiguous allocation */
    placeholders.index_data = zmalloc(sizeof(size_t) * total_count);
    size_t overall_index = 0;
    for (size_t placeholder = 0; placeholder < PLACEHOLDER_COUNT; placeholder++) {
        placeholders.indices[placeholder] = placeholders.index_data + overall_index;

        const size_t count = placeholders.count[placeholder];
        memcpy(placeholders.indices[placeholder], temp_indices[placeholder],
               sizeof(size_t) * count);
        overall_index += count;

        zfree(temp_indices[placeholder]);
    }
    return;
}

/* Batched per-thread accounting for config.requests_finished. A relaxed
 * fetch_add per completed command from every thread turns the counter's
 * cache line into a process-wide contention point at high rates (the same
 * failure mode as the shared latency histogram). Each thread accumulates
 * locally and publishes one fetch_add per REQUESTS_FINISHED_FLUSH_BATCH
 * completions, so the line is written at ~rps/batch instead of ~rps; the
 * per-command read below then almost always hits a Shared cached copy.
 *
 * The residue is flushed from each thread's showThroughput timer (so
 * count-mode termination, detected from the global value, cannot stall on
 * unpublished counts) and after aeMain() returns (so the final report is
 * exact). Consequences of the batching: termination detection and the
 * "stop recording latency at the end" gate can lag by up to
 * batch * num_threads commands, and a warmup-boundary reset can carry over
 * a residue of the same magnitude -- both are the same benign-race class
 * as the surrounding relaxed counter resets. The warmup carry-over is an
 * intentional, documented trade-off: with --threads and --warmup, completion
 * accounting around the warmup boundary is exact only to within
 * batch * num_threads (an exact boundary would reintroduce cross-thread
 * coordination on the completion path -- the contention this design
 * removes). Runs whose -n is small enough for this to matter are far below
 * any meaningful measurement size; the limitation is stated in the --warmup
 * help text. Single-threaded mode keeps the exact per-command path. */
#define REQUESTS_FINISHED_FLUSH_BATCH 256
static _Thread_local int pending_requests_finished = 0;

static void flushRequestsFinished(void) {
    if (pending_requests_finished > 0) {
        atomic_fetch_add_explicit(&config.requests_finished, pending_requests_finished, memory_order_relaxed);
        pending_requests_finished = 0;
    }
}

static int addRequestFinished(void) {
    if (config.num_threads == 0) {
        return atomic_fetch_add_explicit(&config.requests_finished, 1, memory_order_relaxed);
    }
    pending_requests_finished++;
    if (pending_requests_finished >= REQUESTS_FINISHED_FLUSH_BATCH) flushRequestsFinished();
    return atomic_load_explicit(&config.requests_finished, memory_order_relaxed) + pending_requests_finished;
}

static void replacePlaceholder(const size_t *indices, const size_t count, char *cmd, _Atomic uint64_t *key_counter) {
    if (count == 0) return;

    uint64_t key = 0;
    if (config.keyspacelen != 0) {
        if (config.sequential_replacement) {
            key = atomic_fetch_add_explicit(key_counter, 1, memory_order_relaxed);
        } else {
            key = rand62();
        }
        key %= config.keyspacelen;
    }

    /* convert key to string at first location */
    char *p = cmd + indices[0] + PLACEHOLDER_LEN - 1;
    for (size_t j = 0; j < PLACEHOLDER_LEN; j++) {
        *p = '0' + key % 10;
        key /= 10;
        p--;
    }

    /* copy the first instance to the other locations */
    for (size_t i = 1; i < count; i++) {
        char *placeholder = cmd + indices[i];
        memcpy(placeholder, cmd + indices[0], PLACEHOLDER_LEN);
    }
}

static void replacePlaceholders(char *cmd_data, int cmd_count) {
    static _Atomic uint64_t seq_key[PLACEHOLDER_COUNT] = {0};

    for (int cmd_index = 0; cmd_index < cmd_count; cmd_index++) {
        char *cmd = cmd_data + cmd_index * placeholders.cmd_len;

        /* for __rand_int__, multiple instances will have different values */
        size_t *indices = placeholders.indices[0];
        _Atomic uint64_t *key_counter = &seq_key[0];
        for (size_t i = 0; i < placeholders.count[0]; i++) {
            replacePlaceholder(indices + i, 1, cmd, key_counter);
        }

        /* For other placeholders, multiple occurrences within the command will
         * have the same value */
        for (size_t placeholder = 1; placeholder < PLACEHOLDER_COUNT; placeholder++) {
            size_t *indices = placeholders.indices[placeholder];
            size_t count = placeholders.count[placeholder];
            _Atomic uint64_t *key_counter = &seq_key[placeholder];
            replacePlaceholder(indices, count, cmd, key_counter);
        }
    }
}

static void releasePausedClient(client c) {
    if (c->thread_id >= 0) {
        benchmarkThread *thread = config.threads[c->thread_id];
        listNode *ln = listSearchKey(thread->paused_clients, c);
        if (ln != NULL) {
            listDelNode(thread->paused_clients, ln);
        }
    } else {
        listNode *ln = listSearchKey(config.paused_clients, c);
        if (ln != NULL) {
            listDelNode(config.paused_clients, ln);
        }
    }
}

static void freeClient(client c) {
    aeEventLoop *el = CLIENT_GET_EVENTLOOP(c);
    listNode *ln;
    aeDeleteFileEvent(el, c->context->fd, AE_WRITABLE);
    aeDeleteFileEvent(el, c->context->fd, AE_READABLE);
    if (c->thread_id >= 0) {
        int requests_finished = atomic_load_explicit(&config.requests_finished, memory_order_relaxed);
        if (isBenchmarkFinished(requests_finished)) {
            aeStop(el);
        }
    }
    valkeyFree(c->context);
    if (c->paused) releasePausedClient(c);
    sdsfree(c->obuf);
    zfree(c->stagptr);
    zfree(c);
    if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));
    config.liveclients--;
    ln = listSearchKey(config.clients, c);
    assert(ln != NULL);
    listDelNode(config.clients, ln);
    if (config.num_threads) pthread_mutex_unlock(&(config.liveclients_mutex));
}

static void freeAllClients(void) {
    listNode *ln = config.clients->head, *next;

    while (ln) {
        next = ln->next;
        freeClient(ln->value);
        ln = next;
    }
}

static void resetClient(client c) {
    aeEventLoop *el = CLIENT_GET_EVENTLOOP(c);
    aeDeleteFileEvent(el, c->context->fd, AE_WRITABLE);
    aeDeleteFileEvent(el, c->context->fd, AE_READABLE);
    c->written = 0;
    c->request_started = 0;
    c->pending = config.pipeline * c->seqlen;
    /* RDMA registration can invoke writeHandler immediately. */
    createFileEvent(el, c->context->fd, AE_WRITABLE, writeHandler, c);
}

/* Scan buffer for {tag} placeholders and store positions */
static void scanClusterTags(client c, char *buffer_start) {
    if (!config.cluster_mode) return;

    /* Preserve the total capacity across scans so we don't accidentally
     * shrink the allocation when staglen is reset to zero. */
    size_t total_cap = c->staglen + c->stagfree;
    c->staglen = 0;
    c->stagfree = total_cap;
    char *p = buffer_start;
    while ((p = strstr(p, "{tag}")) != NULL) {
        if (c->stagfree == 0) {
            size_t new_size = total_cap ? total_cap * 2 : RANDPTR_INITIAL_SIZE;
            c->stagptr = zrealloc(c->stagptr, new_size * sizeof(char *));
            total_cap = new_size;
            c->stagfree = new_size - c->staglen;
        }
        c->stagptr[c->staglen++] = p;
        c->stagfree--;
        p += 5;
    }
}

static void setClusterKeyHashTag(client c) {
    assert(c->thread_id >= 0);
    clusterNode *node = c->cluster_node;
    assert(node);
    int is_updating_slots = atomic_load_explicit(&config.is_updating_slots, memory_order_relaxed);
    /* If updateClusterSlotsConfiguration is updating the slots array,
     * call updateClusterSlotsConfiguration is order to block the thread
     * since the mutex is locked. When the slots will be updated by the
     * thread that's actually performing the update, the execution of
     * updateClusterSlotsConfiguration won't actually do anything, since
     * the updated_slots_count array will be already NULL. */
    if (is_updating_slots) updateClusterSlotsConfiguration();
    int slot = node->slots[rand() % node->slots_count];
    const char *tag = crc16_slot_table[slot];
    int taglen = strlen(tag);
    size_t i;
    for (i = 0; i < c->staglen; i++) {
        char *p = c->stagptr[i] + 1;
        p[0] = tag[0];
        p[1] = (taglen >= 2 ? tag[1] : '}');
        p[2] = (taglen == 3 ? tag[2] : '}');
    }
}

/* 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,
                memory_order_release,
                memory_order_relaxed)) {
            break;
        }
    }

    return delay_time / 1000000;
}

static void clientDone(client c) {
    int requests_finished = atomic_load_explicit(&config.requests_finished, memory_order_relaxed);
    if (isBenchmarkFinished(requests_finished)) {
        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 (valkeyBufferRead(c->context) != VALKEY_OK) {
        fprintf(stderr, "Error: %s\n", c->context->errstr);
        exit(1);
    } else {
        while (c->pending) {
            if (valkeyGetReply(c->context, &reply) != VALKEY_OK) {
                fprintf(stderr, "Error: %s\n", c->context->errstr);
                exit(1);
            }
            if (reply != NULL) {
                if (reply == (void *)VALKEY_REPLY_ERROR) {
                    fprintf(stderr, "Unexpected error reply, exiting...\n");
                    exit(1);
                }
                valkeyReply *r = reply;
                if (r->type == VALKEY_REPLY_ERROR) {
                    /* Try to update slots configuration if reply error is
                     * MOVED/ASK/CLUSTERDOWN and the key(s) used by the command
                     * contain(s) the slot hash tag.
                     * If the error is not topology-update related then we
                     * immediately exit to avoid false results. */
                    if (c->cluster_node && c->staglen) {
                        int fetch_slots = 0, do_wait = 0;
                        if (!strncmp(r->str, "MOVED", 5) || !strncmp(r->str, "ASK", 3))
                            fetch_slots = 1;
                        else if (!strncmp(r->str, "CLUSTERDOWN", 11)) {
                            /* Usually the cluster is able to recover itself after
                             * a CLUSTERDOWN error, so try to sleep one second
                             * before requesting the new configuration. */
                            fetch_slots = 1;
                            do_wait = 1;
                            fprintf(stderr, "Error from server %s:%d: %s.\n", c->cluster_node->ip,
                                    c->cluster_node->port, r->str);
                        }
                        if (do_wait) sleep(1);
                        if (fetch_slots && !fetchClusterSlotsConfiguration(c)) exit(1);
                    } else {
                        if (c->cluster_node) {
                            fprintf(stderr, "Error from server %s:%d: %s\n", c->cluster_node->ip, c->cluster_node->port,
                                    r->str);
                        } else
                            fprintf(stderr, "Error from server: %s\n", r->str);
                        exit(1);
                    }
                }

                freeReplyObject(reply);
                /* This is an OK for prefix commands such as auth and select.*/
                if (c->prefix_pending > 0) {
                    c->prefix_pending--;
                    c->pending--;
                    /* Discard prefix commands on first response.*/
                    if (c->prefixlen > 0) {
                        size_t j;
                        sdsrange(c->obuf, c->prefixlen, -1);
                        /* Fix the pointers to the slot hash tags */
                        for (j = 0; j < c->staglen; j++) c->stagptr[j] -= c->prefixlen;
                        c->prefixlen = 0;
                    }
                    continue;
                }
                int requests_finished = addRequestFinished();
                if (!isBenchmarkFinished(requests_finished)) {
                    if (config.num_threads == 0) {
                        hdr_record_value(config.latency_histogram, // Histogram to record to
                                         (long)c->latency <= CONFIG_LATENCY_HISTOGRAM_MAX_VALUE
                                             ? (long)c->latency
                                             : CONFIG_LATENCY_HISTOGRAM_MAX_VALUE); // Value to record
                        hdr_record_value(config.current_sec_latency_histogram,      // Histogram to record to
                                         (long)c->latency <= CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE
                                             ? (long)c->latency
                                             : CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE); // Value to record
                    } else {
                        benchmarkThread *thread = config.threads[c->thread_id];
                        hdr_record_value(thread->latency_histogram, // Histogram to record to
                                         (long)c->latency <= CONFIG_LATENCY_HISTOGRAM_MAX_VALUE
                                             ? (long)c->latency
                                             : CONFIG_LATENCY_HISTOGRAM_MAX_VALUE); // Value to record
                        hdr_record_value(thread->current_sec_latency_histogram,     // Histogram to record to
                                         (long)c->latency <= CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE
                                             ? (long)c->latency
                                             : CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE); // Value to record
                    }
                }
                c->pending--;
                if (c->pending == 0) {
                    clientDone(c);
                    break;
                }
            } else {
                break;
            }
        }
    }
}

/*
 * When a client is paused, the function is called by the event loop to
 * awaken the client.
 *
 * Return the number of milliseconds to wait before calling the function again.
 *
 * If the function returns AE_NOMORE, the event is removed.
 */
static long long awakenPausedClient(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    UNUSED(id);
    benchmarkThread *thread = (benchmarkThread *)clientData;

    list *paused_clients = NULL;
    if (thread == NULL) {
        paused_clients = config.paused_clients;
    } else {
        paused_clients = thread->paused_clients;
    }

    listIter li;
    listNode *ln;
    long long delay = 0;
    listRewind(paused_clients, &li);
    while ((ln = listNext(&li)) != NULL) {
        client c = ln->value;
        delay = acquireTokenOrWait(config.pipeline);
        if (delay) {
            break;
        }
        // When client acquires a token, try to write with `reuse`.
        c->paused = 0;
        c->reuse = 1;
        /* writeHandler may make no progress (EAGAIN) or only partially write.
         * The pause path removed AE_WRITABLE, so invoke it through the normal
         * registration helper: TCP retains a future writable wakeup and RDMA
         * keeps its required register-then-kick behavior. */
        createFileEvent(eventLoop, c->context->fd, AE_WRITABLE, writeHandler, c);
        listDelNode(paused_clients, ln);
    }

    // If there are no more paused clients, remove the event.
    if (delay == 0) {
        return AE_NOMORE;
    }
    return delay;
}

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

    /* Acquire a token only for a new request not already resumed by the timer. */
    if (config.rps > 0 && c->reuse == 0 && !c->request_started) {
        /* Acquire a token from the token bucket. */
        long long delay = acquireTokenOrWait(config.pipeline);

        if (delay) {
            int thread_id = c->thread_id;
            int paused_clients_count = 0;

            c->paused = 1;
            aeDeleteFileEvent(el, c->context->fd, AE_WRITABLE);

            benchmarkThread *thread = NULL;
            if (thread_id < 0) {
                paused_clients_count = listLength(config.paused_clients);
                listAddNodeTail(config.paused_clients, c);
            } else {
                thread = config.threads[thread_id];
                paused_clients_count = listLength(thread->paused_clients);
                listAddNodeTail(thread->paused_clients, c);
            }
            if (paused_clients_count == 0) {
                /* Create a time event to awaken the client. */
                aeCreateTimeEvent(el, delay, awakenPausedClient, (void *)thread, NULL);
            }
            return;
        }
    }
    c->reuse = 0;

    /* Initialize each request only once. A would-block write can leave written
     * at zero across callbacks, but must not consume another request or change
     * the buffer (TLS retries also require the same write contents). */
    if (!c->request_started) {
        /* Enforce upper bound to number of requests. */
        int requests_issued = atomic_fetch_add_explicit(&config.requests_issued,
                                                        config.pipeline * c->seqlen,
                                                        memory_order_relaxed);
        if (isBenchmarkFinished(requests_issued)) {
            return;
        }

        /* Dataset field access mode - completely independent command generation */
        if (config.has_field_placeholders && config.current_dataset && config.current_dataset->record_count > 0) {
            static _Atomic uint64_t record_counter = 0;

            /* Generate complete pipeline commands for dataset placeholders */
            sdssetlen(c->obuf, c->prefixlen);
            for (int p = 0; p < config.pipeline; p++) {
                uint64_t record_index = atomic_fetch_add_explicit(&record_counter, 1, memory_order_relaxed) % config.current_dataset->record_count;
                sds complete_cmd = datasetGenerateCommand(config.current_dataset, record_index,
                                                          config.template_argv, config.template_argc,
                                                          dataset_seq_key, config.replace_placeholders,
                                                          config.keyspacelen, config.sequential_replacement);
                c->obuf = sdscatlen(c->obuf, complete_cmd, sdslen(complete_cmd));
                sdsfree(complete_cmd);
            }

            /* Scan generated commands for {tag} in cluster mode */
            if (config.cluster_mode && c->stagptr) {
                scanClusterTags(c, c->obuf + c->prefixlen);
            }
        } else {
            /* Standard mode */
            if (config.replace_placeholders) {
                replacePlaceholders(c->obuf + c->prefixlen, config.pipeline);
            }
        }

        if (config.cluster_mode && c->staglen > 0) setClusterKeyHashTag(c);
        c->slots_last_update = atomic_load_explicit(&config.slots_last_update, memory_order_relaxed);
        c->start = ustime();
        c->latency = -1;
        c->request_started = 1;
    }
    const ssize_t buflen = sdslen(c->obuf);
    const ssize_t writeLen = buflen - c->written;
    if (writeLen > 0) {
        void *ptr = c->obuf + c->written;
        while (1) {
            /* Optimistically try to write before checking if the file descriptor
             * is actually writable. At worst we get EAGAIN. */
            const ssize_t nwritten = cliWriteConn(c->context, ptr, writeLen);
            if (nwritten == writeLen) {
                aeDeleteFileEvent(el, c->context->fd, AE_WRITABLE);
                createFileEvent(el, c->context->fd, AE_READABLE, readHandler, c);
                return;
            }

            if (nwritten > 0) {
                c->written += nwritten;
                return;
            }

            if (nwritten == -1 && errno != EAGAIN) {
                if (errno != EPIPE) fprintf(stderr, "Error writing to the server: %s\n", strerror(errno));
                freeClient(c);
                return;
            }

            /* cliWriteConn reports a nonblocking would-block as zero. Leave
             * the writable event armed: retrying here would spin this worker
             * and prevent it from servicing replies and benchmark timers. */
            return;
        }
    }
}

/* Create a benchmark client, configured to send the command passed as 'cmd' of
 * 'len' bytes.
 *
 * The command is copied N times in the client output buffer (that is reused
 * again and again to send the request to the server) accordingly to the configured
 * pipeline size.
 *
 * Also an initial SELECT command is prepended in order to make sure the right
 * database is selected, if needed. The initial SELECT will be discarded as soon
 * as the first reply is received.
 *
 * To create a client from scratch, the 'from' pointer is set to NULL. If instead
 * we want to create a client using another client as reference, the 'from' pointer
 * points to the client to use as reference. In such a case the following
 * information is take from the 'from' client:
 *
 * 1) The command line to use.
 * 2) The offsets of the __rand_int__ elements inside the command line, used
 *    for arguments randomization.
 *
 * Even when cloning another client, prefix commands are applied if needed.*/
static client createClient(char *cmd, int len, int seqlen, client from, int thread_id) {
    int is_cluster_client = (config.cluster_mode && thread_id >= 0);
    client c = zmalloc(sizeof(struct _client));

    const char *ip = NULL;
    int port = 0;
    struct timeval tv = {0};
    c->cluster_node = NULL;

    if (!is_cluster_client) {
        ip = config.conn_info.hostip;
        port = config.conn_info.hostport;
    } else {
        int node_idx = 0;
        if (config.num_threads < config.cluster_node_count)
            node_idx = config.liveclients % config.cluster_node_count;
        else
            node_idx = thread_id % config.cluster_node_count;
        clusterNode *node = config.cluster_nodes[node_idx];
        assert(node != NULL);
        ip = (const char *)node->ip;
        port = node->port;
        c->cluster_node = node;
    }

    c->context = valkeyConnectWrapper(config.ct, ip, port, tv, 1, config.mptcp);
    if (c->context->err) {
        fprintf(stderr, "Could not connect to server at ");
        if (config.ct != VALKEY_CONN_UNIX || is_cluster_client)
            fprintf(stderr, "%s:%d: %s\n", ip, port, c->context->errstr);
        else
            fprintf(stderr, "%s: %s\n", ip, c->context->errstr);
        exit(1);
    }
    if (config.tls == 1) {
        const char *err = NULL;
        if (cliSecureConnection(c->context, config.sslconfig, &err) == VALKEY_ERR && err) {
            fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
            exit(1);
        }
    }
    c->paused = 0;
    c->reuse = 0;
    c->request_started = 0;
    c->thread_id = thread_id;
    /* Suppress libvalkey cleanup of unused buffers for max speed. */
    c->context->reader->maxbuf = 0;

    /* Build the request buffer:
     * Queue N requests accordingly to the pipeline size, or simply clone
     * the example client buffer. */
    c->obuf = sdsempty();
    /* Prefix the request buffer with AUTH and/or SELECT commands, if applicable.
     * These commands are discarded after the first response, so if the client is
     * reused the commands will not be used again. */
    c->prefix_pending = 0;
    if (config.conn_info.auth) {
        char *buf = NULL;
        int len;
        if (config.conn_info.user == NULL)
            len = valkeyFormatCommand(&buf, "AUTH %s", config.conn_info.auth);
        else
            len = valkeyFormatCommand(&buf, "AUTH %s %s", config.conn_info.user, config.conn_info.auth);
        c->obuf = sdscatlen(c->obuf, buf, len);
        free(buf);
        c->prefix_pending++;
    }

    if (config.enable_tracking) {
        char *buf = NULL;
        int len = valkeyFormatCommand(&buf, "CLIENT TRACKING on");
        c->obuf = sdscatlen(c->obuf, buf, len);
        free(buf);
        c->prefix_pending++;
    }

    /* If a DB number different than zero is selected, prefix our request
     * buffer with the SELECT command, that will be discarded the first
     * time the replies are received, so if the client is reused the
     * SELECT command will not be used again. */
    if (config.conn_info.input_dbnum) {
        c->obuf = sdscatprintf(c->obuf, "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", (int)sdslen(config.input_dbnumstr),
                               config.input_dbnumstr);
        c->prefix_pending++;
    }

    if (config.resp3) {
        char *buf = NULL;
        int len = valkeyFormatCommand(&buf, "HELLO 3");
        c->obuf = sdscatlen(c->obuf, buf, len);
        free(buf);
        c->prefix_pending++;
    }

    if (config.cluster_mode && (config.read_from_replica == FROM_REPLICA_ONLY || config.read_from_replica == FROM_ALL)) {
        char *buf = NULL;
        int len;
        len = valkeyFormatCommand(&buf, "READONLY");
        c->obuf = sdscatlen(c->obuf, buf, len);
        free(buf);
        c->prefix_pending++;
    }

    c->prefixlen = sdslen(c->obuf);
    /* Append the request itself. */
    if (from) {
        c->obuf = sdscatlen(c->obuf, from->obuf + from->prefixlen, sdslen(from->obuf) - from->prefixlen);
        seqlen = from->seqlen;
    } else {
        for (int j = 0; j < config.pipeline; j++) c->obuf = sdscatlen(c->obuf, cmd, len);
    }

    c->written = 0;
    c->seqlen = seqlen;
    c->pending = config.pipeline * seqlen + c->prefix_pending;
    c->stagptr = NULL;
    c->staglen = 0;

    /* If cluster mode is enabled, set slot hashtags pointers. */
    if (config.cluster_mode) {
        if (from) {
            c->staglen = from->staglen;
            c->stagfree = 0;
            c->stagptr = zmalloc(sizeof(char *) * c->staglen);
            /* copy the offsets. */
            for (size_t j = 0; j < c->staglen; j++) {
                c->stagptr[j] = c->obuf + (from->stagptr[j] - from->obuf);
                /* Adjust for the different select prefix length. */
                c->stagptr[j] += c->prefixlen - from->prefixlen;
            }
        } else {
            c->stagfree = RANDPTR_INITIAL_SIZE;
            c->stagptr = zmalloc(sizeof(char *) * c->stagfree);
            scanClusterTags(c, c->obuf);
        }
    }
    aeEventLoop *el = NULL;
    if (thread_id < 0)
        el = config.el;
    else {
        benchmarkThread *thread = config.threads[thread_id];
        el = thread->el;
    }
    if (config.idlemode == 0) {
        createFileEvent(el, c->context->fd, AE_WRITABLE, writeHandler, c);
    } else
        /* In idle mode, clients still need to register readHandler for catching errors */
        aeCreateFileEvent(el, c->context->fd, AE_READABLE, readHandler, c);

    listAddNodeTail(config.clients, c);
    atomic_fetch_add_explicit(&config.liveclients, 1, memory_order_relaxed);

    c->slots_last_update = atomic_load_explicit(&config.slots_last_update, memory_order_relaxed);
    return c;
}

static void createMissingClients(client c) {
    int n = 0;
    while (config.liveclients < config.numclients) {
        int thread_id = -1;
        if (config.num_threads) thread_id = config.liveclients % config.num_threads;
        createClient(NULL, 0, 0, c, thread_id);

        /* Listen backlog is quite limited on most systems */
        if (++n > 64) {
            usleep(50000);
            n = 0;
        }
    }
}

static void showRPSReport(void) {
    if (config.rps_histogram && config.rps_histogram->total_count > 0) {
        const float target_rps = (float)config.rps;

        const float avg_rps = hdr_mean(config.rps_histogram);
        const float p0 = (float)hdr_min(config.rps_histogram);
        const float p50 = (float)hdr_value_at_percentile(config.rps_histogram, 50.0);
        const float p95 = (float)hdr_value_at_percentile(config.rps_histogram, 95.0);
        const float p99 = (float)hdr_value_at_percentile(config.rps_histogram, 99.0);
        const float p100 = (float)hdr_max(config.rps_histogram);

        printf("\n");
        printf("RPS Summary:\n");
        printf("  target RPS: %.2f\n", target_rps);
        printf("  RPS distribution (reqs/sec):\n");
        printf("    %9s %9s %9s %9s %9s %9s\n", "avg", "min", "p50", "p95", "p99", "max");
        printf("    %9.3f %9.3f %9.3f %9.3f %9.3f %9.3f\n", avg_rps, p0, p50, p95, p99, p100);
    }
}

static void showReport(void) {
    const float reqpersec = (float)config.requests_finished / ((float)config.totlatency / 1000.0f);
    const float p0 = ((float)hdr_min(config.latency_histogram)) / 1000.0f;
    const float p50 = hdr_value_at_percentile(config.latency_histogram, 50.0) / 1000.0f;
    const float p95 = hdr_value_at_percentile(config.latency_histogram, 95.0) / 1000.0f;
    const float p99 = hdr_value_at_percentile(config.latency_histogram, 99.0) / 1000.0f;
    const float p100 = ((float)hdr_max(config.latency_histogram)) / 1000.0f;
    const float avg = hdr_mean(config.latency_histogram) / 1000.0f;

    if (!config.quiet && !config.csv) {
        printf("%*s\r", config.last_printed_bytes, " "); // ensure there is a clean line
        printf("====== %s ======\n", config.title);
        printf("  %d requests completed in %.2f seconds\n", config.requests_finished, (float)config.totlatency / 1000);
        printf("  %d parallel clients\n", config.numclients);
        printf("  %d bytes payload\n", config.datasize);
        printf("  keep alive: %d\n", config.keepalive);
        if (config.cluster_mode) {
            const char *node_roles = NULL;
            if (config.read_from_replica == FROM_ALL) {
                node_roles = "cluster";
            } else if (config.read_from_replica == FROM_REPLICA_ONLY) {
                node_roles = "replica";
            } else {
                node_roles = "primary";
            }
            printf("  cluster mode: yes (%d %s)\n", config.cluster_node_count, node_roles);
            int m;
            for (m = 0; m < config.cluster_node_count; m++) {
                clusterNode *node = config.cluster_nodes[m];
                serverConfig *cfg = node->server_config;
                if (cfg == NULL) continue;
                printf("  node [%d] configuration:\n", m);
                printf("    save: %s\n", sdslen(cfg->save) ? cfg->save : "NONE");
                printf("    appendonly: %s\n", cfg->appendonly);
            }
        } else {
            if (config.server_config) {
                printf("  host configuration \"save\": %s\n", config.server_config->save);
                printf("  host configuration \"appendonly\": %s\n", config.server_config->appendonly);
            }
        }
        printf("  multi-thread: %s\n", (config.num_threads ? "yes" : "no"));
        if (config.num_threads) printf("  threads: %d\n", config.num_threads);
        /* Show the RPS Report */
        showRPSReport();
        printf("\n");
        printf("Latency by percentile distribution:\n");
        struct hdr_iter iter;
        long long previous_cumulative_count = -1;
        const long long total_count = config.latency_histogram->total_count;
        hdr_iter_percentile_init(&iter, config.latency_histogram, 1);
        struct hdr_iter_percentiles *percentiles = &iter.specifics.percentiles;
        while (hdr_iter_next(&iter)) {
            const double value = iter.highest_equivalent_value / 1000.0f;
            const double percentile = percentiles->percentile;
            const long long cumulative_count = iter.cumulative_count;
            if (previous_cumulative_count != cumulative_count || cumulative_count == total_count) {
                printf("%3.3f%% <= %.3f milliseconds (cumulative count %lld)\n", percentile, value, cumulative_count);
            }
            previous_cumulative_count = cumulative_count;
        }
        printf("\n");
        printf("Cumulative distribution of latencies:\n");
        previous_cumulative_count = -1;
        hdr_iter_linear_init(&iter, config.latency_histogram, 100);
        while (hdr_iter_next(&iter)) {
            const double value = iter.highest_equivalent_value / 1000.0f;
            const long long cumulative_count = iter.cumulative_count;
            const double percentile = ((double)cumulative_count / (double)total_count) * 100.0;
            if (previous_cumulative_count != cumulative_count || cumulative_count == total_count) {
                printf("%3.3f%% <= %.3f milliseconds (cumulative count %lld)\n", percentile, value, cumulative_count);
            }
            /* After the 2 milliseconds latency to have percentages split
             * by decimals will just add a lot of noise to the output. */
            if (iter.highest_equivalent_value > 2000) {
                hdr_iter_linear_set_value_units_per_bucket(&iter, 1000);
            }
            previous_cumulative_count = cumulative_count;
        }
        printf("\n");
        printf("Summary:\n");
        printf("  throughput summary: %.2f requests per second\n", reqpersec);
        printf("  latency summary (msec):\n");
        printf("    %9s %9s %9s %9s %9s %9s\n", "avg", "min", "p50", "p95", "p99", "max");
        printf("    %9.3f %9.3f %9.3f %9.3f %9.3f %9.3f\n", avg, p0, p50, p95, p99, p100);
    } else if (config.csv) {
        printf("\"%s\",\"%.2f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\"\n", config.title, reqpersec, avg,
               p0, p50, p95, p99, p100);
    } else {
        printf("%*s\r", config.last_printed_bytes, " "); // ensure there is a clean line
        printf("%s: %.2f requests per second, p50=%.3f msec\n", config.title, reqpersec, p50);
    }
}

static void initBenchmarkThreads(void) {
    int i;
    if (config.threads) freeBenchmarkThreads();
    config.threads = zmalloc(config.num_threads * sizeof(benchmarkThread *));
    for (i = 0; i < config.num_threads; i++) {
        benchmarkThread *thread = createBenchmarkThread(i);
        config.threads[i] = thread;
    }
}

static void startBenchmarkThreads(void) {
    int i;
    for (i = 0; i < config.num_threads; i++) {
        benchmarkThread *t = config.threads[i];
        if (pthread_create(&(t->thread), NULL, execBenchmarkThread, t)) {
            fprintf(stderr, "FATAL: Failed to start thread %d.\n", i);
            exit(1);
        }
    }
    for (i = 0; i < config.num_threads; i++) pthread_join(config.threads[i]->thread, NULL);
    /* All threads have stopped: fold the per-thread latency histograms into
     * the global histogram for exact final reporting. Reset it first --
     * showThroughput may have populated it with approximate live merges. */
    hdr_reset(config.latency_histogram);
    for (i = 0; i < config.num_threads; i++) {
        hdr_add(config.latency_histogram, config.threads[i]->latency_histogram);
    }
}

/* Benchmark a sequence of commands. The cmd is RESP encoded of length len and
 * seqlen is the number of commands included in cmd. */
static void benchmarkSequence(const char *title, char *cmd, int len, int seqlen) {
    client c;

    config.title = title;
    config.requests_issued = 0;
    config.requests_finished = 0;
    config.previous_requests_finished = 0;
    config.last_printed_bytes = 0;
    config.current_warmup_duration = config.warmup_duration;
    hdr_init(CONFIG_LATENCY_HISTOGRAM_MIN_VALUE,         // Minimum value
             CONFIG_LATENCY_HISTOGRAM_MAX_VALUE,         // Maximum value
             config.precision,                           // Number of significant figures
             &config.latency_histogram);                 // Pointer to initialise
    hdr_init(CONFIG_LATENCY_HISTOGRAM_MIN_VALUE,         // Minimum value
             CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE, // Maximum value
             config.precision,                           // Number of significant figures
             &config.current_sec_latency_histogram);     // Pointer to initialise

    if (config.rps > 0) {
        hdr_init(1,
                 config.rps * 2,
                 config.precision,
                 &config.rps_histogram);
    }

    initPlaceholders(cmd, len);
    if (config.num_threads) initBenchmarkThreads();

    if (config.rps > 0) {
        config.time_per_token = 1000000000 / config.rps;
        config.time_per_burst = config.time_per_token * config.rps;
        config.last_time_ns = 0;
    }

    int thread_id = config.num_threads > 0 ? 0 : -1;
    c = createClient(cmd, len, seqlen, NULL, thread_id);
    createMissingClients(c);

    config.start = mstime();
    if (!config.num_threads) {
        aeMain(config.el);
    } else
        startBenchmarkThreads();
    config.totlatency = mstime() - config.start;
    showReport();
    freeAllClients();
    if (config.threads) freeBenchmarkThreads();
    if (config.current_sec_latency_histogram) hdr_close(config.current_sec_latency_histogram);
    if (config.latency_histogram) hdr_close(config.latency_histogram);
    if (config.rps_histogram) hdr_close(config.rps_histogram);
}

/* Benchmark a single RESP-encoded command of length len. */
static void benchmark(const char *title, char *cmd, int len) {
    benchmarkSequence(title, cmd, len, 1);
}

/* Thread functions. */

static benchmarkThread *createBenchmarkThread(int index) {
    benchmarkThread *thread = zmalloc(sizeof(*thread));
    if (thread == NULL) return NULL;
    thread->index = index;
    thread->el = aeCreateEventLoop(1024 * 10);
    thread->paused_clients = listCreate();
    hdr_init(CONFIG_LATENCY_HISTOGRAM_MIN_VALUE,         // Minimum value
             CONFIG_LATENCY_HISTOGRAM_MAX_VALUE,         // Maximum value
             config.precision,                           // Number of significant figures
             &thread->latency_histogram);                // Pointer to initialise
    hdr_init(CONFIG_LATENCY_HISTOGRAM_MIN_VALUE,         // Minimum value
             CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE, // Maximum value
             config.precision,                           // Number of significant figures
             &thread->current_sec_latency_histogram);    // Pointer to initialise
    aeCreateTimeEvent(thread->el, 1, showThroughput, (void *)thread, NULL);
    return thread;
}

static void freeBenchmarkThread(benchmarkThread *thread) {
    if (thread->el) aeDeleteEventLoop(thread->el);
    if (thread->latency_histogram) hdr_close(thread->latency_histogram);
    if (thread->current_sec_latency_histogram) hdr_close(thread->current_sec_latency_histogram);
    listRelease(thread->paused_clients);
    zfree(thread);
}

static void freeBenchmarkThreads(void) {
    int i = 0;
    for (; i < config.num_threads; i++) {
        benchmarkThread *thread = config.threads[i];
        if (thread) freeBenchmarkThread(thread);
    }
    zfree(config.threads);
    config.threads = NULL;
}

static void *execBenchmarkThread(void *ptr) {
    benchmarkThread *thread = (benchmarkThread *)ptr;
    aeMain(thread->el);
    /* Publish any batched completion residue: the final report reads
     * config.requests_finished after all threads are joined. */
    flushRequestsFinished();
    return NULL;
}

/* Cluster helper functions. */

static clusterNode *createClusterNode(char *ip, int port) {
    clusterNode *node = zmalloc(sizeof(*node));
    if (!node) return NULL;
    node->ip = ip;
    node->port = port;
    node->name = NULL;
    node->flags = 0;
    node->replicate = NULL;
    node->replicas_count = 0;
    node->slots = zmalloc(CLUSTER_SLOTS * sizeof(int));
    node->slots_count = 0;
    node->updated_slots = NULL;
    node->updated_slots_count = 0;
    node->server_config = NULL;
    return node;
}

static void freeClusterNode(clusterNode *node) {
    if (node->name) sdsfree(node->name);
    if (node->replicate) sdsfree(node->replicate);
    if (node->ip) sdsfree(node->ip);
    if (node->server_config != NULL) freeServerConfig(node->server_config);
    zfree(node->slots);
    zfree(node);
}

static void freeClusterNodes(void) {
    int i = 0;
    for (; i < config.cluster_node_count; i++) {
        clusterNode *n = config.cluster_nodes[i];
        if (n) freeClusterNode(n);
    }
    zfree(config.cluster_nodes);
    config.cluster_nodes = NULL;
}

static clusterNode **addClusterNode(clusterNode *node) {
    int count = config.cluster_node_count + 1;
    config.cluster_nodes = zrealloc(config.cluster_nodes, count * sizeof(*node));
    if (!config.cluster_nodes) return NULL;
    config.cluster_nodes[config.cluster_node_count++] = node;
    return config.cluster_nodes;
}

static int fetchClusterConfiguration(void) {
    int success = 1;
    valkeyContext *ctx = NULL;
    valkeyReply *reply = NULL;
    dict *nodes = NULL;
    const char *errmsg = "Failed to fetch cluster configuration";
    size_t i, j;
    ctx = getValkeyContext(config.ct, config.conn_info.hostip, config.conn_info.hostport);
    if (ctx == NULL) {
        exit(1);
    }

    reply = valkeyCommand(ctx, "CLUSTER SLOTS");
    if (reply == NULL || reply->type == VALKEY_REPLY_ERROR) {
        success = 0;
        if (reply) fprintf(stderr, "%s\nCLUSTER SLOTS ERROR: %s\n", errmsg, reply->str);
        goto cleanup;
    }
    assert(reply->type == VALKEY_REPLY_ARRAY);
    nodes = dictCreate(&dtype);
    for (i = 0; i < reply->elements; i++) {
        valkeyReply *r = reply->element[i];
        assert(r->type == VALKEY_REPLY_ARRAY);
        assert(r->elements >= 3);
        int from = r->element[0]->integer;
        int to = r->element[1]->integer;
        sds primary = NULL;
        for (j = 2; j < r->elements; j++) {
            valkeyReply *nr = r->element[j];
            assert(nr->type == VALKEY_REPLY_ARRAY && nr->elements >= 3);
            assert(nr->element[0]->str != NULL);
            assert(nr->element[2]->str != NULL);

            int is_primary = (j == 2);
            if (is_primary) primary = sdsnew(nr->element[2]->str);
            int is_cluster_option_only = (config.read_from_replica == FROM_PRIMARY_ONLY);
            if ((config.read_from_replica == FROM_REPLICA_ONLY && is_primary) || (is_cluster_option_only && !is_primary)) continue;

            sds ip = sdsnew(nr->element[0]->str);
            sds name = sdsnew(nr->element[2]->str);
            int port = nr->element[1]->integer;
            int slot_start = from;
            int slot_end = to;

            clusterNode *node = NULL;
            dictEntry *entry = dictFind(nodes, name);
            if (entry == NULL) {
                node = createClusterNode(ip, port);
                if (node == NULL) {
                    success = 0;
                    sdsfree(ip);
                    sdsfree(name);
                    goto cleanup;
                } else {
                    node->name = name;
                    if (!is_primary) node->replicate = sdsdup(primary);
                }
            } else {
                node = dictGetVal(entry);
                sdsfree(ip);
                sdsfree(name);
            }
            if (slot_start == slot_end) {
                node->slots[node->slots_count++] = slot_start;
            } else {
                while (slot_start <= slot_end) {
                    int slot = slot_start++;
                    node->slots[node->slots_count++] = slot;
                }
            }
            if (node->slots_count == 0) {
                fprintf(stderr, "WARNING: Node %s:%d has no slots, skipping...\n", node->ip, node->port);
                continue;
            }
            if (entry == NULL) {
                dictReplace(nodes, node->name, node);
                if (!addClusterNode(node)) {
                    success = 0;
                    goto cleanup;
                }
            }
        }
        sdsfree(primary);
    }
cleanup:
    if (ctx) valkeyFree(ctx);
    if (!success) {
        if (config.cluster_nodes) freeClusterNodes();
    }
    if (reply) freeReplyObject(reply);
    if (nodes) dictRelease(nodes);
    return success;
}

/* Request the current cluster slots configuration by calling CLUSTER SLOTS
 * and atomically update the slots after a successful reply. */
static int fetchClusterSlotsConfiguration(client c) {
    UNUSED(c);
    int success = 1, is_fetching_slots = 0, last_update = 0;
    size_t i, j;

    last_update = atomic_load_explicit(&config.slots_last_update, memory_order_relaxed);
    if (c->slots_last_update < last_update) {
        c->slots_last_update = last_update;
        return -1;
    }
    valkeyReply *reply = NULL;

    is_fetching_slots = atomic_fetch_add_explicit(&config.is_fetching_slots, 1, memory_order_relaxed);
    if (is_fetching_slots) return -1; // TODO: use other codes || errno ?
    atomic_store_explicit(&config.is_fetching_slots, 1, memory_order_relaxed);
    fprintf(stderr, "WARNING: Cluster slots configuration changed, fetching new one...\n");
    const char *errmsg = "Failed to update cluster slots configuration";

    /* printf("[%d] fetchClusterSlotsConfiguration\n", c->thread_id); */
    dict *nodes = dictCreate(&dtype);
    valkeyContext *ctx = NULL;
    for (i = 0; i < (size_t)config.cluster_node_count; i++) {
        clusterNode *node = config.cluster_nodes[i];
        assert(node->ip != NULL);
        assert(node->name != NULL);
        assert(node->port);
        /* Use first node as entry point to connect to. */
        if (ctx == NULL) {
            ctx = getValkeyContext(config.ct, node->ip, node->port);
            if (!ctx) {
                success = 0;
                goto cleanup;
            }
        }
        if (node->updated_slots != NULL) zfree(node->updated_slots);
        node->updated_slots = NULL;
        node->updated_slots_count = 0;
        dictReplace(nodes, node->name, node);
    }
    reply = valkeyCommand(ctx, "CLUSTER SLOTS");
    if (reply == NULL || reply->type == VALKEY_REPLY_ERROR) {
        success = 0;
        if (reply) fprintf(stderr, "%s\nCLUSTER SLOTS ERROR: %s\n", errmsg, reply->str);
        goto cleanup;
    }
    assert(reply->type == VALKEY_REPLY_ARRAY);
    for (i = 0; i < reply->elements; i++) {
        valkeyReply *r = reply->element[i];
        assert(r->type == VALKEY_REPLY_ARRAY);
        assert(r->elements >= 3);
        int from, to, slot;
        from = r->element[0]->integer;
        to = r->element[1]->integer;
        size_t start, end;
        if (config.read_from_replica == FROM_ALL) {
            start = 2;
            end = r->elements;
        } else if (config.read_from_replica == FROM_REPLICA_ONLY) {
            start = 3;
            end = r->elements;
        } else {
            start = 2;
            end = 3;
        }

        for (j = start; j < end; j++) {
            valkeyReply *nr = r->element[j];
            assert(nr->type == VALKEY_REPLY_ARRAY && nr->elements >= 3);
            assert(nr->element[2]->str != NULL);
            sds name = sdsnew(nr->element[2]->str);
            dictEntry *entry = dictFind(nodes, name);
            if (entry == NULL) {
                success = 0;
                fprintf(stderr,
                        "%s: could not find node with ID %s in current "
                        "configuration.\n",
                        errmsg, name);
                if (name) sdsfree(name);
                goto cleanup;
            }
            sdsfree(name);
            clusterNode *node = dictGetVal(entry);
            if (node->updated_slots == NULL) node->updated_slots = zcalloc(CLUSTER_SLOTS * sizeof(int));
            for (slot = from; slot <= to; slot++) node->updated_slots[node->updated_slots_count++] = slot;
        }
    }
    updateClusterSlotsConfiguration();
cleanup:
    freeReplyObject(reply);
    valkeyFree(ctx);
    dictRelease(nodes);
    atomic_store_explicit(&config.is_fetching_slots, 0, memory_order_relaxed);
    return success;
}

/* Atomically update the new slots configuration. */
static void updateClusterSlotsConfiguration(void) {
    pthread_mutex_lock(&config.is_updating_slots_mutex);
    atomic_store_explicit(&config.is_updating_slots, 1, memory_order_relaxed);

    int i;
    for (i = 0; i < config.cluster_node_count; i++) {
        clusterNode *node = config.cluster_nodes[i];
        if (node->updated_slots != NULL) {
            int *oldslots = node->slots;
            node->slots = node->updated_slots;
            node->slots_count = node->updated_slots_count;
            node->updated_slots = NULL;
            node->updated_slots_count = 0;
            zfree(oldslots);
        }
    }
    atomic_store_explicit(&config.is_updating_slots, 0, memory_order_relaxed);
    atomic_fetch_add_explicit(&config.slots_last_update, 1, memory_order_relaxed);
    pthread_mutex_unlock(&config.is_updating_slots_mutex);
}

/* Free dataset memory */
static void cleanupDataset(void) {
    if (config.current_dataset) {
        datasetFree(config.current_dataset);
        config.current_dataset = NULL;
    }
}

/* Add RESP command to sequence with repeat count */
static void addRespCommandToSequence(sds *sds_args, size_t *argvlen, int start, int end, int repeat, sds *cmd_seq, int *seq_len) {
    char *cmd = NULL;
    int len = valkeyFormatCommandArgv(&cmd, end - start, (const char **)sds_args + start, argvlen + start);
    for (int j = 0; j < repeat; j++) {
        *cmd_seq = sdscatlen(*cmd_seq, cmd, len);
    }
    *seq_len += repeat;
    free(cmd);
}

/* Parse and setup command template for dataset field validation */
static int parseCommandTemplate(int argc, char **argv) {
    sds *sds_args = getSdsArrayFromArgv(argc, argv, 0);
    if (!sds_args) {
        fprintf(stderr, "Invalid quoted string\n");
        return 0;
    }

    /* Detect field placeholders */
    config.has_field_placeholders = 0;
    for (int i = 0; i < argc; i++) {
        if (strstr(sds_args[i], FIELD_PREFIX)) {
            config.has_field_placeholders = 1;
            break;
        }
    }

    if (config.has_field_placeholders) {
        config.template_argc = argc;
        config.template_argv = zmalloc(argc * sizeof(sds));
        for (int i = 0; i < argc; i++) {
            config.template_argv[i] = sdsdup(sds_args[i]);
        }
    }

    sdsfreesplitres(sds_args, argc);
    return 1;
}


/* Generate random data for the benchmark. See #7196. */
static void genBenchmarkRandomData(char *data, int count) {
    static uint32_t state = 1234;
    int i = 0;

    while (count--) {
        state = (state * 1103515245 + 12345);
        data[i++] = '0' + ((state >> 16) & 63);
    }
}

/* Returns number of consumed options. */
int parseOptions(int argc, char **argv) {
    int i;
    int lastarg;
    int exit_status = 1;
    char *tls_usage;
    char *rdma_usage;

    for (i = 1; i < argc; i++) {
        lastarg = (i == (argc - 1));

        if (!strcmp(argv[i], "-c")) {
            if (lastarg) goto invalid;
            config.numclients = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "-v") || !strcmp(argv[i], "--version")) {
            sds version = cliVersion();
            printf("valkey-benchmark %s\n", version);
            sdsfree(version);
            exit(0);
        } else if (!strcmp(argv[i], "-n")) {
            if (lastarg) goto invalid;
            if (config.duration > 0) {
                fprintf(stderr, "Options -n and --duration are mutually exclusive.\n");
                exit(1);
            }
            config.requests = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "--duration")) {
            if (lastarg) goto invalid;
            if (config.requests > 0) {
                fprintf(stderr, "Options -n and --duration are mutually exclusive.\n");
                exit(1);
            }
            config.duration = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "--warmup")) {
            if (lastarg) goto invalid;
            config.warmup_duration = atoi(argv[++i]);

        } else if (!strcmp(argv[i], "-k")) {
            if (lastarg) goto invalid;
            config.keepalive = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "-h")) {
            if (lastarg) goto invalid;
            sdsfree(config.conn_info.hostip);
            config.conn_info.hostip = sdsnew(argv[++i]);
        } else if (!strcmp(argv[i], "-p")) {
            if (lastarg) goto invalid;
            config.conn_info.hostport = atoi(argv[++i]);
            if (config.conn_info.hostport < 0 || config.conn_info.hostport > 65535) {
                fprintf(stderr, "Invalid server port.\n");
                exit(1);
            }
        } else if (!strcmp(argv[i], "-s")) {
            if (lastarg) goto invalid;
            sdsfree(config.conn_info.hostip);
            config.conn_info.hostip = sdsnew(argv[++i]);
            config.ct = VALKEY_CONN_UNIX;
        } else if (!strcmp(argv[i], "-x")) {
            config.stdinarg = 1;
        } else if (!strcmp(argv[i], "-a")) {
            if (lastarg) goto invalid;
            config.conn_info.auth = sdsnew(argv[++i]);
        } else if (!strcmp(argv[i], "--user")) {
            if (lastarg) goto invalid;
            config.conn_info.user = sdsnew(argv[++i]);
        } else if (!strcmp(argv[i], "--rps")) {
            if (lastarg) goto invalid;
            config.rps = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "-u") && !lastarg) {
            parseUri(argv[++i], "valkey-benchmark", &config.conn_info, &config.tls);
            if (config.conn_info.hostport < 0 || config.conn_info.hostport > 65535) {
                fprintf(stderr, "Invalid server port.\n");
                exit(1);
            }
            config.input_dbnumstr = sdsfromlonglong(config.conn_info.input_dbnum);
        } else if (!strcmp(argv[i], "-3")) {
            config.resp3 = 1;
        } else if (!strcmp(argv[i], "-d")) {
            if (lastarg) goto invalid;
            config.datasize = atoi(argv[++i]);
            if (config.datasize < 1) config.datasize = 1;
            if (config.datasize > 1024 * 1024 * 1024) config.datasize = 1024 * 1024 * 1024;
        } else if (!strcmp(argv[i], "-P")) {
            if (lastarg) goto invalid;
            config.pipeline = atoi(argv[++i]);
            if (config.pipeline <= 0) config.pipeline = 1;
        } else if (!strcmp(argv[i], "-r")) {
            if (lastarg) goto invalid;
            const char *next = argv[++i];
            char *endptr;
            errno = 0;
            long long val = strtoll(next, &endptr, 10);
            if (endptr == next || *endptr != '\0') {
                fprintf(stderr, "Invalid value for -r: '%s' is not a valid number\n", next);
                exit(1);
            }
            if (errno == ERANGE || val < 0 || val > MAX_KEYSPACELEN) {
                fprintf(stderr, "Invalid value for -r: keyspacelen must be between 0 and %lld\n", MAX_KEYSPACELEN);
                exit(1);
            }
            config.replace_placeholders = 1;
            config.keyspacelen = val;
        } else if (!strcmp(argv[i], "--sequential")) {
            config.sequential_replacement = 1;
        } else if (!strcmp(argv[i], "-q")) {
            config.quiet = 1;
        } else if (!strcmp(argv[i], "--csv")) {
            config.csv = 1;
        } else if (!strcmp(argv[i], "-l")) {
            config.loop = 1;
        } else if (!strcmp(argv[i], "-I")) {
            config.idlemode = 1;
        } else if (!strcmp(argv[i], "-e")) {
            fprintf(stderr, "WARNING: -e option has no effect. "
                            "We now immediately exit on error to avoid false results.\n");
        } else if (!strcmp(argv[i], "--seed")) {
            if (lastarg) goto invalid;
            int rand_seed = atoi(argv[++i]);
            srandom(rand_seed);
            init_genrand64(rand_seed);
        } else if (!strcmp(argv[i], "-t")) {
            if (lastarg) goto invalid;
            /* We get the list of tests to run as a string in the form
             * get,set,lrange,...,test_N. Then we add a comma before and
             * after the string in order to make sure that searching
             * for ",testname," will always get a match if the test is
             * enabled. */
            config.tests = sdsnew(",");
            config.tests = sdscat(config.tests, (char *)argv[++i]);
            config.tests = sdscat(config.tests, ",");
            sdstolower(config.tests);
        } else if (!strcmp(argv[i], "--dbnum")) {
            if (lastarg) goto invalid;
            config.conn_info.input_dbnum = atoi(argv[++i]);
            config.input_dbnumstr = sdsfromlonglong(config.conn_info.input_dbnum);
        } else if (!strcmp(argv[i], "--precision")) {
            if (lastarg) goto invalid;
            config.precision = atoi(argv[++i]);
            if (config.precision < 0) config.precision = DEFAULT_LATENCY_PRECISION;
            if (config.precision > MAX_LATENCY_PRECISION) config.precision = MAX_LATENCY_PRECISION;
        } else if (!strcmp(argv[i], "--threads")) {
            if (lastarg) goto invalid;
            config.num_threads = atoi(argv[++i]);
            if (config.num_threads > MAX_THREADS) {
                fprintf(stderr, "WARNING: Too many threads, limiting threads to %d.\n", MAX_THREADS);
                config.num_threads = MAX_THREADS;
            } else if (config.num_threads < 0)
                config.num_threads = 0;
        } else if (!strcmp(argv[i], "--cluster")) {
            config.cluster_mode = 1;
        } else if (!strcmp(argv[i], "--rfr")) {
            if (argv[++i]) {
                if (!strcmp(argv[i], "all")) {
                    config.read_from_replica = FROM_ALL;
                } else if (!strcmp(argv[i], "yes")) {
                    config.read_from_replica = FROM_REPLICA_ONLY;
                } else if (!strcmp(argv[i], "no")) {
                    config.read_from_replica = FROM_PRIMARY_ONLY;
                } else {
                    goto invalid;
                }
            } else
                goto invalid;
        } else if (!strcmp(argv[i], "--enable-tracking")) {
            config.enable_tracking = 1;
        } else if (!strcmp(argv[i], "--num-functions")) {
            config.num_functions = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "--num-keys-in-fcall")) {
            config.num_keys_in_fcall = atoi(argv[++i]);
        } else if (!strcmp(argv[i], "--dataset")) {
            if (lastarg) goto invalid;
            config.dataset_file = sdsnew(argv[++i]);
        } else if (!strcmp(argv[i], "--maxdocs")) {
            if (lastarg) goto invalid;
            config.max_documents = atoi(argv[++i]);
            if (config.max_documents <= 0) config.max_documents = -1;
        } else if (!strcmp(argv[i], "--help")) {
            exit_status = 0;
            goto usage;
#ifdef USE_OPENSSL
        } else if (!strcmp(argv[i], "--tls")) {
            config.tls = 1;
        } else if (!strcmp(argv[i], "--sni")) {
            if (lastarg) goto invalid;
            config.sslconfig.sni = strdup(argv[++i]);
        } else if (!strcmp(argv[i], "--cacertdir")) {
            if (lastarg) goto invalid;
            config.sslconfig.cacertdir = strdup(argv[++i]);
        } else if (!strcmp(argv[i], "--cacert")) {
            if (lastarg) goto invalid;
            config.sslconfig.cacert = strdup(argv[++i]);
        } else if (!strcmp(argv[i], "--insecure")) {
            config.sslconfig.skip_cert_verify = 1;
        } else if (!strcmp(argv[i], "--cert")) {
            if (lastarg) goto invalid;
            config.sslconfig.cert = strdup(argv[++i]);
        } else if (!strcmp(argv[i], "--key")) {
            if (lastarg) goto invalid;
            config.sslconfig.key = strdup(argv[++i]);
        } else if (!strcmp(argv[i], "--tls-ciphers")) {
            if (lastarg) goto invalid;
            config.sslconfig.ciphers = strdup(argv[++i]);
#ifdef TLS1_3_VERSION
        } else if (!strcmp(argv[i], "--tls-ciphersuites")) {
            if (lastarg) goto invalid;
            config.sslconfig.ciphersuites = strdup(argv[++i]);
#endif
#endif
#ifdef USE_RDMA
        } else if (!strcmp(argv[i], "--rdma")) {
            if (valkeyInitiateRdma() != VALKEY_OK) {
                fprintf(stderr, "Failed to initialize RDMA support from libvalkey\n");
                exit(1);
            }
            config.ct = VALKEY_CONN_RDMA;
#endif
        } else if (!strcmp(argv[i], "--fuzz")) {
            config.fuzz_mode = 1;
        } else if (!strcmp(argv[i], "--fuzz-loglevel")) {
            if (lastarg) goto invalid;
            config.fuzz_log_level = argv[++i];
        } else if (!strcmp(argv[i], "--fuzz-mode")) {
            if (lastarg) goto invalid;
            int count = 0;
            const char *modes_arg = argv[++i];
            sds *modes = sdssplitlen(modes_arg, strlen(modes_arg), ",", 1, &count);
            for (int j = 0; j < count; j++) {
                if (!strcmp(modes[j], "malformed-commands"))
                    config.fuzz_flags |= FUZZ_MODE_MALFORMED_COMMANDS;
                else if (!strcmp(modes[j], "config-commands"))
                    config.fuzz_flags |= FUZZ_MODE_CONFIG_COMMANDS;
                else {
                    fprintf(stderr, "Invalid fuzz mode: %s\n", modes[j]);
                    sdsfreesplitres(modes, count);
                    exit(1);
                }
            }
            sdsfreesplitres(modes, count);
        } else if (!strcmp(argv[i], "--mptcp")) {
            config.mptcp = 1;
        } else if (!strcmp(argv[i], "--")) {
            /* End of options. */
            return i + 1;
        } else {
            /* Assume the user meant to provide an option when the arg starts
             * with a dash. We're done otherwise and should use the remainder
             * as the command and arguments for running the benchmark. */
            if (argv[i][0] == '-') goto invalid;
            return i;
        }
    }

    return i;

invalid:
    printf("Invalid option \"%s\" or option argument missing\n\n", argv[i]);

usage:
    tls_usage =
#ifdef USE_OPENSSL
        " --tls              Establish a secure TLS connection.\n"
        " --sni <host>       Server name indication for TLS.\n"
        " --cacert <file>    CA Certificate file to verify with.\n"
        " --cacertdir <dir>  Directory where trusted CA certificates are stored.\n"
        "                    If neither cacert nor cacertdir are specified, the default\n"
        "                    system-wide trusted root certs configuration will apply.\n"
        " --insecure         Allow insecure TLS connection by skipping cert validation.\n"
        " --cert <file>      Client certificate to authenticate with.\n"
        " --key <file>       Private key file to authenticate with.\n"
        " --tls-ciphers <list> Sets the list of preferred ciphers (TLSv1.2 and below)\n"
        "                    in order of preference from highest to lowest separated by\n"
        "                    colon (\":\"). See the ciphers(1ssl) manpage for more\n"
        "                    information about the syntax of this string.\n"
#ifdef TLS1_3_VERSION
        " --tls-ciphersuites <list> Sets the list of preferred ciphersuites (TLSv1.3)\n"
        "                    in order of preference from highest to lowest separated by\n"
        "                    colon (\":\"). See the ciphers(1ssl) manpage for more\n"
        "                    information about the syntax of this string, and\n"
        "                    specifically for TLSv1.3 ciphersuites.\n"
#endif
#endif
        "";

    rdma_usage =
#ifdef USE_RDMA
        " --rdma             Establish a RDMA connection.\n"
#endif
        "";


    printf(
        "%s%s%s%s%s%s", /* Split to avoid strings longer than 4095 (-Woverlength-strings). */
        "Usage: valkey-benchmark [OPTIONS] [--] [COMMAND ARGS...]\n\n"
        "Simulates sending commands using multiple clients. The utility provides a\n"
        "default set of tests. You can run a subset of the tests using the -t option or\n"
        "supply one or more custom commands on the command line.\n\n"
        "To supply multiple commands on the command line, separate them with ';' as in\n"
        "`SET foo bar ';' GET foo`. You can also prefix a command in the sequence with\n"
        "a number N to repeat the command N times. In command arguments, the following\n"
        "placeholders are substituted:\n\n"
        " __rand_int__       Replaced with a zero-padded random integer in the range\n"
        "                    selected using the -r option. Multiple occurrences within the\n"
        "                    command will have different values.\n"
        "__rand_1st__        Like __rand_int__ but multiple occurrences will have the same\n"
        "                    value. __rand_2nd__ through __rand_9th__ are also available.\n"
        " __data__           Replaced with data of the size specified by the -d option.\n"
        " __field:name__     Replaced with data from the specified field/column in the\n"
        "                    dataset. Requires --dataset option.\n"
        " {tag}              Replaced with a tag that routes the command to each node in\n"
        "                    a cluster. Include this in key names when running in cluster\n"
        "                    mode.\n"
        "\n",
        "Options:\n"
        "\n"
        " -h <hostname>      Server hostname (default 127.0.0.1)\n"
        " -p <port>          Server port (default 6379)\n"
        " -s <socket>        Server socket (overrides host and port)\n"
        " -a <password>      Password for Valkey Auth\n"
        " --user <username>  Used to send ACL style 'AUTH username pass'. Needs -a.\n"
        " -u <uri>           Server URI on format valkey://user:password@host:port/dbnum\n"
        "                    User, password and dbnum are optional. For authentication\n"
        "                    without a username, use username 'default'. For TLS, use\n"
        "                    the scheme 'valkeys'.\n"
        " -c <clients>       Number of parallel connections (default 50).\n"
        "                    Note: If --cluster is used then number of clients has to be\n"
        "                    the same or higher than the number of nodes.\n"
        " -n <requests>      Total number of requests (default 100000)\n"
        " --duration <seconds>\n"
        "                    Run benchmark for specified number of seconds\n"
        "                    (mutually exclusive with -n)\n"
        " --warmup <seconds> Run benchmark for specified warmup period before\n"
        "                    recording data. With --threads, completed-request\n"
        "                    accounting is batched per thread, so a small number\n"
        "                    of pre-warmup completions may carry into the counted\n"
        "                    total at the warmup boundary.\n"
        " -d <size>          Data size of SET/GET value in bytes (default 3)\n"
        " --dbnum <db>       SELECT the specified db number (default 0)\n"
        " -3                 Start session in RESP3 protocol mode.\n"
        " --threads <num>    Enable multi-thread mode.\n"
        " --cluster          Enable cluster mode.\n"
        "                    If the command is supplied on the command line in cluster\n"
        "                    mode, the key must contain \"{tag}\". Otherwise, the\n"
        "                    command will not be sent to the right cluster node.\n"
        " --rfr <mode>       Enable read from replicas in cluster mode.\n"
        "                    This command must be used with the --cluster option.\n"
        "                    There are three modes for reading from replicas:\n"
        "                    'no' - sends read requests to primaries only (default) \n"
        "                    'yes' - sends read requests to replicas only.\n"
        "                    'all' - sends read requests to all nodes.\n"
        "                    Since write commands will not be accepted by replicas,\n"
        "                    it is recommended to enable read from replicas only for read\n"
        "                    command tests.\n"
        " --enable-tracking  Send CLIENT TRACKING on before starting benchmark.\n"
        " -k <boolean>       1=keep alive 0=reconnect (default 1)\n"
        " -r <keyspacelen>   Use random keys for SET/GET/INCR, random values for SADD,\n"
        "                    random members and scores for ZADD.\n"
        "                    Using this option the benchmark will replace the string\n"
        "                    __rand_int__ inside an argument with a random 12 digit\n"
        "                    number in the specified range from 0 to keyspacelen-1. The\n"
        "                    substitution changes every time a command is executed.\n"
        "                    Default tests use this to hit random keys in the specified\n"
        "                    range.\n"
        "                    Note: If -r is omitted, all commands in a benchmark will\n"
        "                    use the same key.\n"
        " --sequential       Modifies the -r argument to replace the string __rand_int__\n"
        "                    with 12 digit numbers sequentially instead of randomly.\n"
        "                    __rand_1st__ through __rand_9th__ are available with\n"
        "                    independent counters. Used to create expected number of\n"
        "                    elements with multiple replacements.\n"
        "                    example: ZADD myzset __rand_int__ element:__rand_1st__\n"
        " -P <numreq>        Pipeline <numreq> requests. That is, send multiple requests\n"
        "                    before waiting for the replies. Default 1 (no pipeline).\n"
        "                    When multiple commands are specified on the command line,\n"
        "                    then the full command sequence counts as one and -P controls\n"
        "                    the number of times the command sequence is sent in each\n"
        "                    pipeline.\n",
        " -q                 Quiet. Just show query/sec values\n"
        " --precision        Number of decimal places to display in latency output\n"
        "                    (default 0)\n"
        " --csv              Output in CSV format\n"
        " -l                 Loop. Run the tests forever\n"
        " -t <tests>         Only run the comma separated list of tests. The test\n"
        "                    names are the same as the ones produced as output.\n"
        "                    The -t option is ignored if a specific command is supplied\n"
        "                    on the command line.\n"
        " -I                 Idle mode. Just open N idle connections and wait.\n"
        " -x                 Read last argument from STDIN.\n"
        " --rps <requests>   Limit the total number of requests per second.\n"
        "                    Default 0 (no limit)\n"
        " --seed <num>       Set the seed for random number generator.\n"
        "                    With --threads, per-thread streams are seeded\n"
        "                    deterministically but their order of first use\n"
        "                    depends on thread scheduling.\n"
        "                    Default seed is based on time.\n"
        " --num-functions <num>\n"
        "                    Sets the number of functions present in the Lua lib that is\n"
        "                    loaded when running the 'function_load' test. (default 10).\n"
        " --num-keys-in-fcall <num>\n"
        "                    Sets the number of keys passed to FCALL command when running\n"
        "                    the 'fcall' test. (default 1)\n"
        " --dataset <file>   Path to CSV/TSV dataset file for field placeholder replacement.\n"
        "                    All fields auto-detected with natural content lengths.\n"
        " --maxdocs <num>    Maximum number of documents to load from dataset file.\n"
        "                    Default: unlimited.\n",
        tls_usage,
        rdma_usage,
        " --mptcp            Enable an MPTCP connection.\n"
        " --fuzz             Enable fuzzy mode to generate random commands. WARNING: Recommended for testing only, not for use with production data.\n"
        " --fuzz-mode <modes> Set fuzzing modes (comma-separated): malformed-commands, config-commands.\n"
        "                    malformed-commands: Generates also malformed commands.\n"
        "                    config-commands: Allows CONFIG SET commands.\n"
        "                    Default: valid commands only.\n"
        " --fuzz-loglevel <level>\n"
        "                    Set log level for fuzzer (none, error, info, debug).\n"
        "                    Default is 'info'.\n"
        " --help             Output this help and exit.\n"
        " --version          Output version and exit.\n\n"
        "Examples:\n\n"
        " Run the benchmark with the default configuration against 127.0.0.1:6379:\n"
        "   $ valkey-benchmark\n\n"
        " Use 20 parallel clients, for a total of 100k requests, against 192.168.1.1:\n"
        "   $ valkey-benchmark -h 192.168.1.1 -p 6379 -n 100000 -c 20\n\n"
        " Fill 127.0.0.1:6379 with about 1 million keys only using the SET test:\n"
        "   $ valkey-benchmark -t set -n 1000000 -r 100000000\n\n"
        " Benchmark 127.0.0.1:6379 for a few commands producing CSV output:\n"
        "   $ valkey-benchmark -t ping,set,get -n 100000 --csv\n\n"
        " Benchmark a specific command line:\n"
        "   $ valkey-benchmark -r 10000 -n 10000 eval 'return server.call(\"ping\")' 0\n\n"
        " Fill a list with 10000 random elements:\n"
        "   $ valkey-benchmark -r 10000 -n 10000 lpush mylist __rand_int__\n\n"
        " Benchmark a specific transaction:\n"
        "   $ valkey-benchmark -- multi ';' set key:__rand_int__ __data__ ';' \\\n"
        "                         incr counter ';' exec\n\n");
    exit(exit_status);
}

long long showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData) {
    UNUSED(eventLoop);
    UNUSED(id);
    benchmarkThread *thread = (benchmarkThread *)clientData;
    /* Publish this thread's batched completion count so global termination
     * detection and the displayed totals stay fresh (see addRequestFinished). */
    flushRequestsFinished();
    int requests_finished = atomic_load_explicit(&config.requests_finished, memory_order_relaxed);
    int previous_requests_finished = atomic_load_explicit(&config.previous_requests_finished, memory_order_relaxed);
    long long current_tick = mstime();

    int liveclients = atomic_load_explicit(&config.liveclients, memory_order_relaxed);
    if (liveclients == 0 && !isBenchmarkFinished(requests_finished)) {
        fprintf(stderr, "All clients disconnected... aborting.\n");
        exit(1);
    }
    int warmup_duration = atomic_load_explicit(&config.current_warmup_duration, memory_order_relaxed);
    if (warmup_duration > 0) {
        if ((current_tick - config.start) >= (warmup_duration * 1000LL)) {
            /* exit the warmup period, clear all stats */
            atomic_store_explicit(&config.current_warmup_duration, 0, memory_order_relaxed);

            config.start = current_tick;
            atomic_store_explicit(&config.requests_finished, 0, memory_order_relaxed);
            atomic_store_explicit(&config.requests_issued, 0, memory_order_relaxed);
            atomic_store_explicit(&config.previous_requests_finished, 0, memory_order_relaxed);
            hdr_reset(config.latency_histogram);
            /* Per-thread histograms carry the recorded values now; clear them
             * too. Concurrent recording may leak a warmup sample into the
             * results -- same benign-race class as the counter resets above. */
            for (int t = 0; t < config.num_threads; t++) {
                hdr_reset(config.threads[t]->latency_histogram);
                hdr_reset(config.threads[t]->current_sec_latency_histogram);
            }
        }
    } else if (isBenchmarkFinished(requests_finished)) {
        aeStop(eventLoop);
        /* In multi-threaded mode, return AE_NOMORE to delete the timer since
         * the thread's event loop will be destroyed. In single-threaded mode,
         * we must keep the timer alive for subsequent benchmark tests */
        if (config.num_threads) {
            return AE_NOMORE;
        }
    }
    if (config.csv) return SHOW_THROUGHPUT_INTERVAL;
    /* only first thread output throughput */
    if (thread != NULL && thread->index != 0) {
        return SHOW_THROUGHPUT_INTERVAL;
    }
    if (config.idlemode == 1) {
        printf("clients: %d\r", config.liveclients);
        fflush(stdout);
        return SHOW_THROUGHPUT_INTERVAL;
    }
    if (config.num_threads) {
        /* Use thread-0's own histograms for the live display line. This is
         * race-free (showThroughput runs on thread-0's event loop, same thread
         * that records into these histograms) and representative under uniform
         * workloads. The final report folds all threads exactly after join. */
        benchmarkThread *t0 = config.threads[0];
        hdr_reset(config.latency_histogram);
        hdr_add(config.latency_histogram, t0->latency_histogram);
        hdr_reset(config.current_sec_latency_histogram);
        hdr_add(config.current_sec_latency_histogram, t0->current_sec_latency_histogram);
        hdr_reset(t0->current_sec_latency_histogram);
    }
    const float dt = (float)(current_tick - config.start) / 1000.0;
    const float rps = (float)requests_finished / dt;
    const float instantaneous_dt = (float)(current_tick - config.previous_tick) / 1000.0;
    const float instantaneous_rps = (float)(requests_finished - previous_requests_finished) / instantaneous_dt;

    if (config.rps_histogram) {
        hdr_record_value(config.rps_histogram, (int64_t)instantaneous_rps);
    }

    config.previous_tick = current_tick;
    atomic_store_explicit(&config.previous_requests_finished, requests_finished, memory_order_relaxed);

    printf("%*s\r", config.last_printed_bytes, " "); /* ensure there is a clean line */
    config.last_printed_bytes = 0;
    if (warmup_duration > 0) {
        config.last_printed_bytes += printf("Warming up ");
    }
    config.last_printed_bytes +=
        printf("%s: rps=%.1f (overall: %.1f) avg_msec=%.3f (overall: %.3f)", config.title, instantaneous_rps, rps,
               hdr_mean(config.current_sec_latency_histogram) / 1000.0f, hdr_mean(config.latency_histogram) / 1000.0f);
    if (warmup_duration > 0 || config.duration > 0) {
        config.last_printed_bytes += printf(" %.1f seconds\r", dt);
    } else {
        config.last_printed_bytes += printf(" %d requests\r", requests_finished);
    }

    hdr_reset(config.current_sec_latency_histogram);
    fflush(stdout);
    return SHOW_THROUGHPUT_INTERVAL;
}

char *generateFunctionScript(uint32_t num_functions, int with_keys) {
    /* 64K buffer to hold script code */
    const size_t buffer_len = 64 * 1024;
    char *buffer = zmalloc(buffer_len);
    memset(buffer, 0, buffer_len);

    int written = snprintf(buffer, buffer_len, "#!lua name=benchlib\n");
    while (num_functions > 0 && (buffer_len - written) > 0) {
        assert(buffer_len - written > 0);
        int n = 0;
        if (with_keys) {
            n = snprintf(buffer + written, buffer_len - written,
                         "local function foo%u(keys, args)\nreturn keys[0]\nend\n",
                         num_functions);
        } else {
            n = snprintf(buffer + written, buffer_len - written,
                         "local function foo%u()\nreturn 0\nend\n",
                         num_functions);
        }

        if (n < 0 || (size_t)n >= buffer_len - written) {
            break;
        }
        written += n;

        n = snprintf(buffer + written, buffer_len - written,
                     "server.register_function('foo%u', foo%u)\n",
                     num_functions,
                     num_functions);
        written += n;

        num_functions--;
    }

    return buffer;
}

/* Return true if the named test was selected using the -t command line
 * switch, or if all the tests are selected (no -t passed by user). */
int test_is_selected(const char *name) {
    char buf[256];
    int l = strlen(name);

    if (config.tests == NULL) return 1;
    buf[0] = ',';
    memcpy(buf + 1, name, l);
    buf[l + 1] = ',';
    buf[l + 2] = '\0';
    return strstr(config.tests, buf) != NULL;
}

int main(int argc, char **argv) {
    int i;
    char *data, *cmd, *tag;
    int len = 0;

    client c;
    sds title = NULL, cmd_seq = NULL;
    sds *sds_args = NULL;
    size_t *argvlen = NULL;
    int seq_len = 0; /* Total number of commands in the sequence. */

    srandom(time(NULL) ^ getpid());
    init_genrand64(ustime() ^ getpid());
    signal(SIGHUP, SIG_IGN);
    signal(SIGPIPE, SIG_IGN);

    memset(&config.sslconfig, 0, sizeof(config.sslconfig));
    config.ct = VALKEY_CONN_TCP;
    config.numclients = 50;
    config.requests = -1;
    config.duration = -1;
    config.warmup_duration = -1;
    config.current_warmup_duration = -1;
    config.liveclients = 0;
    config.el = aeCreateEventLoop(1024 * 10);
    aeCreateTimeEvent(config.el, 1, showThroughput, NULL, NULL);
    config.keepalive = 1;
    config.datasize = 3;
    config.pipeline = 1;
    config.replace_placeholders = 0;
    config.keyspacelen = 0;
    config.sequential_replacement = 0;
    config.quiet = 0;
    config.csv = 0;
    config.loop = 0;
    config.idlemode = 0;
    config.clients = listCreate();
    config.paused_clients = listCreate();
    config.conn_info.hostip = sdsnew("127.0.0.1");
    config.conn_info.hostport = 6379;
    config.tests = NULL;
    config.conn_info.input_dbnum = 0;
    config.stdinarg = 0;
    config.conn_info.auth = NULL;
    config.precision = DEFAULT_LATENCY_PRECISION;
    config.num_threads = 0;
    config.threads = NULL;
    config.cluster_mode = 0;
    config.fuzz_mode = 0;
    config.fuzz_log_level = "info";
    config.fuzz_flags = 0;
    config.rps = 0;
    config.read_from_replica = FROM_PRIMARY_ONLY;
    config.cluster_node_count = 0;
    config.cluster_nodes = NULL;
    config.server_config = NULL;
    config.is_fetching_slots = 0;
    config.is_updating_slots = 0;
    config.slots_last_update = 0;
    config.enable_tracking = 0;
    config.num_functions = 10;
    config.num_keys_in_fcall = 1;
    config.resp3 = 0;
    config.dataset_file = NULL;
    config.max_documents = -1; /* -1 = unlimited */
    config.current_dataset = NULL;
    config.template_argc = 0;
    config.template_argv = NULL;
    config.has_field_placeholders = 0;
    resetPlaceholders();

    i = parseOptions(argc, argv);
    argc -= i;
    argv += i;

    /* Setup dataset if specified */
    if (config.dataset_file) {
        if (argc == 0) {
            fprintf(stderr, "Error: Dataset mode requires a command with field placeholders\n");
            fprintf(stderr, "Example: SET doc:__rand_int__ \"__field:content__\"\n");
            exit(1);
        }

        /* Parse command template and setup field placeholder detection */
        if (!parseCommandTemplate(argc, argv)) {
            exit(1);
        }

        /* Dataset mode requires at least one field placeholder in the command
         * template. Without it, the dataset would be initialized and reported
         * but never used for command generation. */
        if (!config.has_field_placeholders) {
            fprintf(stderr,
                    "Error: Dataset mode requires at least one field placeholder\n");
            fprintf(stderr,
                    "Example: SET doc:__rand_int__ \"__field:content__\"\n");
            exit(1);
        }

        /* Initialize dataset - single call does everything atomically */
        int verbose = !config.csv && !config.quiet;
        config.current_dataset = datasetInit(config.dataset_file,
                                             config.max_documents,
                                             config.has_field_placeholders,
                                             config.template_argv, config.template_argc,
                                             verbose);
        if (!config.current_dataset) {
            fprintf(stderr, "Failed to initialize dataset\n");
            exit(1);
        }
    }
    /* Set default for requests if not specified */
    if (config.requests < 0) config.requests = 100000;

    tag = "";

#ifdef USE_OPENSSL
    if (config.tls) {
        cliSecureInit();
    }
#endif

    if (config.mptcp && (config.ct != VALKEY_CONN_TCP)) {
        fprintf(stderr, "Options --mptcp is only supported by TCP\n");
        exit(1);
    }

    /* Parse and validate the command sequence given in the remaining arguments
     * before connecting, so a malformed invocation is reported up front without
     * needing a reachable server. */
    if (argc && !config.fuzz_mode && !config.idlemode) {
        title = sdsnew(argv[0]);
        for (i = 1; i < argc; i++) {
            title = sdscatlen(title, " ", 1);
            title = sdscatlen(title, (char *)argv[i], strlen(argv[i]));
        }
        sds_args = getSdsArrayFromArgv(argc, argv, 0);
        if (!sds_args) {
            fprintf(stderr, "Invalid quoted string\n");
            return 1;
        }
        if (config.stdinarg) {
            sds_args = sds_realloc(sds_args, (argc + 1) * sizeof(sds));
            sds_args[argc] = readArgFromStdin();
            argc++;
        }
        /* Set up argument length */
        argvlen = zmalloc(argc * sizeof(size_t));
        for (i = 0; i < argc; i++) argvlen[i] = sdslen(sds_args[i]);
        /* RESP-encode the command(s) given on the syntax
         *
         *     [N] command args [ ";" [N] command args [...] ]
         */
        int start = 0;             /* Argument index where the current command starts. */
        int repeat = 1;            /* Number of times to repeat the current command. */
        int has_repeat = 0;        /* Current segment is prefixed by a repeat count. */
        int malformed_segment = 0; /* A segment was left with no command to run. */
        cmd_seq = sdsempty();
        for (i = 0; i <= argc; i++) {
            if (i < argc && i == start && sds_args[i][0] >= '1' && sds_args[i][0] <= '9') {
                /* Command prefixed by number means repeat command N times. */
                repeat = atoi(sds_args[i]);
                has_repeat = 1;
                start++;
            } else if (i == argc || strcmp(";", sds_args[i]) == 0) {
                cmd = NULL;
                if (i == start) {
                    /* Empty segment. Only a trailing separator (i == argc) is
                     * harmless; otherwise the caller lost a command or a count. */
                    if (has_repeat || i < argc) malformed_segment = 1;
                    start = i + 1;
                    repeat = 1;
                    has_repeat = 0;
                    continue;
                }

                addRespCommandToSequence(sds_args, argvlen, start, i, repeat, &cmd_seq, &seq_len);
                start = i + 1;
                repeat = 1;
                has_repeat = 0;
            } else if (strstr(sds_args[i], "__data__")) {
                if (config.current_dataset) {
                    fprintf(stderr, "Error: __data__ placeholders cannot be used with --dataset option\n");
                    exit(1);
                }
                /* Replace data placeholders with data of length given by -d. */
                int num_parts;
                sds *parts = sdssplitlen(sds_args[i], sdslen(sds_args[i]),
                                         "__data__", strlen("__data__"),
                                         &num_parts);
                sds newarg = parts[0];
                parts[0] = NULL; /* prevent it from being freed below */
                for (int j = 1; j < num_parts; j++) {
                    char data[config.datasize];
                    genBenchmarkRandomData(data, config.datasize);
                    newarg = sdscatlen(newarg, data, config.datasize);
                    newarg = sdscatlen(newarg, parts[j], sdslen(parts[j]));
                }
                sdsfreesplitres(parts, num_parts);
                sdsfree(sds_args[i]);
                sds_args[i] = newarg;
                argvlen[i] = sdslen(sds_args[i]);
            }
            /* NOTE: Field placeholder processing is handled above in the command-level loop to ensure row consistency */
        }
        if (seq_len <= 0 || malformed_segment) {
            fprintf(stderr, "Invalid command sequence: a command is missing.\n"
                            "Use -n <requests> to set the total number of requests.\n");
            return 1;
        }
        len = sdslen(cmd_seq);
        /* adjust the datasize to the parsed command */
        config.datasize = len;
    }

    if (config.cluster_mode && !config.fuzz_mode) {
        // We only include the slot placeholder {tag} if cluster mode is enabled
        tag = ":{tag}";

        /* Fetch cluster configuration. */
        if (!fetchClusterConfiguration() || !config.cluster_nodes) {
            if (config.ct != VALKEY_CONN_UNIX) {
                fprintf(stderr,
                        "Failed to fetch cluster configuration from "
                        "%s:%d\n",
                        config.conn_info.hostip, config.conn_info.hostport);
            } else {
                fprintf(stderr,
                        "Failed to fetch cluster configuration from "
                        "%s\n",
                        config.conn_info.hostip);
            }
            exit(1);
        }
        if (config.cluster_node_count == 0) {
            fprintf(stderr, "Invalid cluster: %d node(s).\n", config.cluster_node_count);
            exit(1);
        }
        const char *node_roles = NULL;
        if (config.read_from_replica == FROM_ALL) {
            node_roles = "cluster";
        } else if (config.read_from_replica == FROM_REPLICA_ONLY) {
            node_roles = "replica";
        } else {
            node_roles = "primary";
        }
        printf("Cluster has %d %s nodes:\n\n", config.cluster_node_count, node_roles);
        int i = 0;
        for (; i < config.cluster_node_count; i++) {
            clusterNode *node = config.cluster_nodes[i];
            if (!node) {
                fprintf(stderr, "Invalid cluster node #%d\n", i);
                exit(1);
            }
            const char *node_type = (node->replicate == NULL ? "Primary" : "Replica");
            printf("Node %d(%s): ", i, node_type);
            if (node->name) printf("%s ", node->name);
            printf("%s:%d\n", node->ip, node->port);
            node->server_config = getServerConfig(config.ct, node->ip, node->port);
            if (node->server_config == NULL) {
                fprintf(stderr, "WARNING: Could not fetch node CONFIG %s:%d\n", node->ip, node->port);
            }
        }
        printf("\n");
        /* Automatically set thread number to node count if not specified
         * by the user. */
        if (config.num_threads == 0) config.num_threads = config.cluster_node_count;
    } else {
        config.server_config = getServerConfig(config.ct, config.conn_info.hostip, config.conn_info.hostport);
        if (config.server_config == NULL) {
            fprintf(stderr, "WARNING: Could not fetch server CONFIG\n");
        }
    }
    if (config.num_threads > 0) {
        pthread_mutex_init(&(config.liveclients_mutex), NULL);
        pthread_mutex_init(&(config.is_updating_slots_mutex), NULL);
    }

    if (config.keepalive == 0) {
        fprintf(stderr, "WARNING: Keepalive disabled. You probably need "
                        "'echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse' for Linux and "
                        "'sudo sysctl -w net.inet.tcp.msl=1000' for Mac OS X in order "
                        "to use a lot of clients/requests\n");
    }
    if (argc > 0 && config.tests != NULL) {
        fprintf(stderr, "WARNING: Option -t is ignored.\n");
    }

    if (config.idlemode) {
        printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients);
        int thread_id = -1, use_threads = (config.num_threads > 0);
        if (use_threads) {
            thread_id = 0;
            initBenchmarkThreads();
        }
        c = createClient("", 0, 1, NULL, thread_id); /* will never receive a reply */
        createMissingClients(c);
        if (use_threads)
            startBenchmarkThreads();
        else
            aeMain(config.el);
        /* and will wait for every */
    }
    if (config.csv) {
        printf("\"test\",\"rps\",\"avg_latency_ms\",\"min_latency_ms\",\"p50_latency_ms\",\"p95_latency_ms\",\"p99_"
               "latency_ms\",\"max_latency_ms\"\n");
    }

    if (config.fuzz_mode) {
        return runFuzzerClients(
            config.conn_info.hostip,
            config.conn_info.hostport,
            config.requests,
            config.numclients,
            config.cluster_mode,
            config.keyspacelen,
            config.tls ? &config.sslconfig : NULL,
            config.fuzz_log_level,
            config.fuzz_flags);
    }

    /* Run benchmark with the command sequence already parsed above. */
    if (argc && cmd_seq) {
        do {
            benchmarkSequence(title, cmd_seq, len, seq_len);
        } while (config.loop);
        sdsfree(cmd_seq);
        sdsfreesplitres(sds_args, argc);

        sdsfree(title);
        if (config.server_config != NULL) freeServerConfig(config.server_config);
        zfree(argvlen);
        return 0;
    }

    /* Run default benchmark suite. */
    data = zmalloc(config.datasize + 1);
    do {
        genBenchmarkRandomData(data, config.datasize);
        data[config.datasize] = '\0';

        if (test_is_selected("ping_inline") || test_is_selected("ping")) benchmark("PING_INLINE", "PING\r\n", 6);

        if (test_is_selected("ping_mbulk") || test_is_selected("ping")) {
            len = valkeyFormatCommand(&cmd, "PING");
            benchmark("PING_MBULK", cmd, len);
            free(cmd);
        }

        if (test_is_selected("set")) {
            len = valkeyFormatCommand(&cmd, "SET key%s:__rand_int__ %s", tag, data);
            benchmark("SET", cmd, len);
            free(cmd);
        }

        if (test_is_selected("get")) {
            len = valkeyFormatCommand(&cmd, "GET key%s:__rand_int__", tag);
            benchmark("GET", cmd, len);
            free(cmd);
        }

        if (test_is_selected("incr")) {
            len = valkeyFormatCommand(&cmd, "INCR counter%s:__rand_int__", tag);
            benchmark("INCR", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lpush")) {
            len = valkeyFormatCommand(&cmd, "LPUSH mylist%s %s", tag, data);
            benchmark("LPUSH", cmd, len);
            free(cmd);
        }

        if (test_is_selected("rpush")) {
            len = valkeyFormatCommand(&cmd, "RPUSH mylist%s %s", tag, data);
            benchmark("RPUSH", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lpop")) {
            len = valkeyFormatCommand(&cmd, "LPOP mylist%s", tag);
            benchmark("LPOP", cmd, len);
            free(cmd);
        }

        if (test_is_selected("rpop")) {
            len = valkeyFormatCommand(&cmd, "RPOP mylist%s", tag);
            benchmark("RPOP", cmd, len);
            free(cmd);
        }

        if (test_is_selected("sadd")) {
            len = valkeyFormatCommand(&cmd, "SADD myset%s element:__rand_int__", tag);
            benchmark("SADD", cmd, len);
            free(cmd);
        }

        if (test_is_selected("hset")) {
            len = valkeyFormatCommand(&cmd, "HSET myhash%s element:__rand_int__ %s", tag, data);
            benchmark("HSET", cmd, len);
            free(cmd);
        }

        if (test_is_selected("spop")) {
            len = valkeyFormatCommand(&cmd, "SPOP myset%s", tag);
            benchmark("SPOP", cmd, len);
            free(cmd);
        }

        if (test_is_selected("zadd")) {
            char *score = "0";
            if (config.replace_placeholders) score = "__rand_int__";
            len = valkeyFormatCommand(&cmd, "ZADD myzset%s %s element:__rand_1st__", tag, score);
            benchmark("ZADD", cmd, len);
            free(cmd);
        }

        if (test_is_selected("zpopmin")) {
            len = valkeyFormatCommand(&cmd, "ZPOPMIN myzset%s", tag);
            benchmark("ZPOPMIN", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lrange") || test_is_selected("lrange_100") || test_is_selected("lrange_300") ||
            test_is_selected("lrange_500") || test_is_selected("lrange_600")) {
            len = valkeyFormatCommand(&cmd, "LPUSH mylist%s %s", tag, data);
            benchmark("LPUSH (needed to benchmark LRANGE)", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lrange") || test_is_selected("lrange_100")) {
            len = valkeyFormatCommand(&cmd, "LRANGE mylist%s 0 99", tag);
            benchmark("LRANGE_100 (first 100 elements)", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lrange") || test_is_selected("lrange_300")) {
            len = valkeyFormatCommand(&cmd, "LRANGE mylist%s 0 299", tag);
            benchmark("LRANGE_300 (first 300 elements)", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lrange") || test_is_selected("lrange_500")) {
            len = valkeyFormatCommand(&cmd, "LRANGE mylist%s 0 499", tag);
            benchmark("LRANGE_500 (first 500 elements)", cmd, len);
            free(cmd);
        }

        if (test_is_selected("lrange") || test_is_selected("lrange_600")) {
            len = valkeyFormatCommand(&cmd, "LRANGE mylist%s 0 599", tag);
            benchmark("LRANGE_600 (first 600 elements)", cmd, len);
            free(cmd);
        }

        if (test_is_selected("mset")) {
            const char *cmd_argv[21];
            cmd_argv[0] = "MSET";
            sds key_placeholder = sdscatprintf(sdsnew(""), "key%s:__rand_int__", tag);
            for (i = 1; i < 21; i += 2) {
                cmd_argv[i] = key_placeholder;
                cmd_argv[i + 1] = data;
            }
            len = valkeyFormatCommandArgv(&cmd, 21, cmd_argv, NULL);
            benchmark("MSET (10 keys)", cmd, len);
            free(cmd);
            sdsfree(key_placeholder);
        }

        if (test_is_selected("mget")) {
            const char *cmd_argv[11];
            cmd_argv[0] = "MGET";
            sds key_placeholder = sdscatprintf(sdsnew(""), "key%s:__rand_int__", tag);
            for (i = 1; i < 11; i++) {
                cmd_argv[i] = key_placeholder;
            }
            len = valkeyFormatCommandArgv(&cmd, 11, cmd_argv, NULL);
            benchmark("MGET (10 keys)", cmd, len);
            free(cmd);
            sdsfree(key_placeholder);
        }

        if (test_is_selected("xadd")) {
            len = valkeyFormatCommand(&cmd, "XADD mystream%s * myfield %s", tag, data);
            benchmark("XADD", cmd, len);
            free(cmd);
        }

        if (test_is_selected("function_load")) {
            char *script = generateFunctionScript(config.num_functions, 0);
            len = valkeyFormatCommand(&cmd, "function load replace %s", script);
            benchmark("FUNCTION LOAD", cmd, len);
            zfree(script);
            free(cmd);
        }

        if (test_is_selected("fcall")) {
            char *script = generateFunctionScript(1, config.num_keys_in_fcall > 0);

            valkeyContext *ctx = getValkeyContext(config.ct, config.conn_info.hostip, config.conn_info.hostport);
            if (ctx == NULL) {
                exit(1);
            }

            assert(ctx != NULL && ctx->err == 0);
            void *reply = valkeyCommand(ctx, "FUNCTION LOAD REPLACE %s", script);

            assert(reply != NULL);
            freeReplyObject(reply);
            valkeyFree(ctx);
            zfree(script);

            char **cmd_argv = zmalloc(sizeof(char *) * (config.num_keys_in_fcall + 3));
            int ret = asprintf(&(cmd_argv[0]), "fcall");
            UNUSED(ret);
            ret = asprintf(&(cmd_argv[1]), "foo1");
            UNUSED(ret);
            ret = asprintf(&(cmd_argv[2]), "%d", config.num_keys_in_fcall);
            UNUSED(ret);
            for (int i = 0; i < config.num_keys_in_fcall; i++) {
                ret = asprintf(&(cmd_argv[3 + i]), "key%d", i + 1);
                UNUSED(ret);
            }
            len = valkeyFormatCommandArgv(&cmd, config.num_keys_in_fcall + 3, (const char **)cmd_argv, NULL);
            for (int i = 0; i < config.num_keys_in_fcall + 3; i++) {
                free(cmd_argv[i]);
            }
            zfree(cmd_argv);

            benchmark("FCALL", cmd, len);
            free(cmd);
        }

        if (!config.csv) printf("\n");
    } while (config.loop);

    zfree(data);
    freeCliConnInfo(config.conn_info);
    if (config.server_config != NULL) freeServerConfig(config.server_config);
    resetPlaceholders();
    cleanupDataset();

    /* Clean up command template */
    if (config.template_argv) {
        for (int i = 0; i < config.template_argc; i++) {
            sdsfree(config.template_argv[i]);
        }
        zfree(config.template_argv);
    }

    return 0;
}
