From a190bdeea5cd5e54fdabd95651ff5b7b402663a1 Mon Sep 17 00:00:00 2001 From: 1iaan Date: Wed, 4 Mar 2026 07:20:09 +0000 Subject: [PATCH] chainbuffer fixed --- Makefile | 9 +- chainbuffer.plan.md | 189 +++++++++++++ config/config.xml | 2 +- network/chainbuffer.c | 618 +++++++++++++++++++++++++++++++++++++----- network/chainbuffer.h | 23 ++ reactor.c | 18 +- test-redis/README.md | 107 ++++++++ test-redis/bench | Bin 0 -> 32328 bytes test-redis/bench.c | 447 ++++++++++++++++++++++++++++++ 9 files changed, 1335 insertions(+), 78 deletions(-) create mode 100644 chainbuffer.plan.md create mode 100644 test-redis/README.md create mode 100755 test-redis/bench create mode 100644 test-redis/bench.c diff --git a/Makefile b/Makefile index a2aa845..192026a 100644 --- a/Makefile +++ b/Makefile @@ -21,12 +21,14 @@ SUBDIR = ./NtyCo/ TEST_REDIS = ./test-redis/testcase TEST_REDIS_SRCS = ./test-redis/test.c +TEST_REDIS_BENCH = ./test-redis/bench +TEST_REDIS_BENCH_SRCS = ./test-redis/bench.c TEST_REDIS_LDFLAGS = -lhiredis OBJS = $(SRCS:.c=.o) -all: $(SUBDIR) $(TARGET) $(TEST_REDIS) +all: $(SUBDIR) $(TARGET) $(TEST_REDIS) $(TEST_REDIS_BENCH) $(SUBDIR): ECHO make -C $@ @@ -40,11 +42,14 @@ $(TARGET): $(OBJS) $(TEST_REDIS): $(TEST_REDIS_SRCS) $(CC) -g -o $@ $^ $(TEST_REDIS_LDFLAGS) +$(TEST_REDIS_BENCH): $(TEST_REDIS_BENCH_SRCS) + $(CC) -g -o $@ $^ $(TEST_REDIS_LDFLAGS) + %.o: %.c $(CC) $(CFLAGS) $(INC) -c $^ -g -o $@ clean: clmem cldata - rm -rf $(OBJS) $(TARGET) $(TEST_REDIS) + rm -rf $(OBJS) $(TARGET) $(TEST_REDIS) $(TEST_REDIS_BENCH) clmem: rm -rf mem_leak/* diff --git a/chainbuffer.plan.md b/chainbuffer.plan.md new file mode 100644 index 0000000..92d7348 --- /dev/null +++ b/chainbuffer.plan.md @@ -0,0 +1,189 @@ +# ChainBuffer + 全量 WAL（单一方案，slot 驱动） + +## 1. 目标与约束 + +目标： + +- 所有命令都落盘（不再区分 update/get）。 +- 主线程 A 尽量不做内存分配，不构造 uring task。 +- 主线程 A 只做：收包、解析命令边界、摘取 payload、写入预分配 slot。 +- WAL 线程 B 负责：从 slot 取 payload、构造 task、提交 io_uring、归还内存块。 + +约束： + +- 只保留一个实现方案，不引入 v1/v2 双路径。 +- 正确性优先：半包、多包、pipeline、断连、慢盘都可控。 + +--- + +## 2. ChainBuffer 设计（重点） + +## 2.1 结构定义 + +采用“回环 chunk 链表”： + +```c +typedef struct cb_chunk { + struct cb_chunk *next; + uint32_t cap; // 固定大小，默认 4096 + uint32_t rpos; // 读指针 [0, cap) + uint32_t wpos; // 写指针 [0, cap) + uint32_t used; // 已用字节数 [0, cap] + uint32_t refcnt; // 被 WAL payload 持有的引用计数 + uint8_t data[]; +} cb_chunk_t; + +typedef struct chain_buffer { + cb_chunk_t *head; + cb_chunk_t *tail; + size_t total_len; + uint32_t chunk_size; + + // 节点池：避免频繁 malloc/free + cb_chunk_t *free_list; + uint32_t free_count; + uint32_t free_limit; +} chain_buffer_t; +``` + +说明： + +- chunk 内部可回环读写（避免 memmove）。 +- 多 chunk 串联用于扩容。 +- 空 chunk 不释放到系统，先回收到 `free_list`。 + +## 2.2 接收路径（readv 直写） + +主线程不再 `recv -> tmp -> memcpy`，改为： + +1. `chain_buffer_prepare_recv_iov(buf, iov, max_iov)` + - 收集可写空闲段（优先尾 chunk，不够就从 free_list 取或新建 chunk）。 +2. `readv(fd, iov, iovcnt)` +3. `chain_buffer_commit_recv(buf, nread)` + - 推进 `wpos/used/total_len`。 + +这样可去掉接收中转拷贝。 + +## 2.3 消费与摘取 + +提供三个核心能力： + +- `chain_buffer_iter_*`：按字节流遍历，供 RESP 解析命令边界。 +- `chain_buffer_detach_prefix(buf, len, cb_payload_t *out)`：把前缀字节摘成 payload（尽量零拷贝，边界切分允许一次小拷贝）。 +- `chain_buffer_release_payload(buf, payload)`：WAL 线程用完后归还 chunk（refcnt--，归零后回 free_list）。 + +--- + +## 3. slot 队列设计（主线程零分配） + +## 3.1 预分配 ring + +定义单生产者单消费者 ring（A->B）： + +```c +#define WAL_SLOT_CAP 65536 + +typedef struct wal_slot { + uint64_t seq; + uint32_t cmd_len; + cb_payload_t payload; // 命令字节 + uint8_t in_use; +} wal_slot_t; + +typedef struct wal_ring { + wal_slot_t slots[WAL_SLOT_CAP]; + _Atomic uint32_t head; // producer write + _Atomic uint32_t tail; // consumer read + _Atomic uint32_t size; +} wal_ring_t; +``` + +特点： + +- slot 全预分配。 +- 主线程写 slot 不 malloc。 +- WAL 线程消费后清空 slot 并前移 tail。 + +## 3.2 主线程流程（所有命令都落盘） + +对每条完整命令： + +1. 从 chainbuffer 解析得到 `cmd_len`。 +2. 申请一个空 slot（ring 未满）。 +3. `detach_prefix(cmd_len)` 得到 `payload`。 +4. 写入 slot：`seq/cmd_len/payload`。 +5. 发布 head。 + +注意： + +- 不做命令类型判断。 +- 不构造 uring task。 +- 不进行额外 payload malloc。 + +## 3.3 WAL 线程流程 + +循环： + +1. 从 ring 取 slot。 +2. 生成长度头（4B）+ payload iov。 +3. 调 `submit_write()`（当前打包拷贝发生在此线程）。 +4. `chain_buffer_release_payload()` 归还 chunk。 +5. 清 slot，推进 tail。 + +--- + +## 4. 一致性与回压 + +## 4.1 日志顺序 + +- `g_log_off` 只在 WAL 线程维护。 +- 单消费者天然保证写入顺序与入队顺序一致。 + +## 4.2 回压（必须） + +当 ring 达到高水位（例如 80%）： + +1. 暂停该连接读事件（优先）。 +2. 若持续超时（如 200ms）仍高水位，关闭连接保护系统。 + +本方案不再做“回退旧路径”。 + +## 4.3 异常处理 + +- `submit_write` 失败：记录错误并触发保护动作（可选停机/拒绝新连接）。 +- 连接断开：已入 slot 的 payload 继续由 WAL 线程完成归还。 +- 进程退出：先停读，再 drain wal_ring，再 shutdown uring。 + +--- + +## 5. 实施步骤（实际落地） + +1. 重构 `network/chainbuffer.*` + - 回环 chunk + free_list + - `prepare_recv_iov/commit_recv` + - `iter/detach_prefix/release_payload` +2. 修改 `reactor.c` + - `recv_cb` 使用 `readv` + commit +3. 修改 `kvstore.c` + - 按命令边界循环 + - 每条命令统一入 wal_slot（不分类） +4. 新增 `dump/wal_slot_queue.*`（或放 `dump/kvs_oplog.c`） + - SPSC ring + WAL worker +5. 调整 `dump/kvs_oplog.c` + - 改为 WAL 线程消费 slot 后调用 `submit_write` +6. 收敛退出与错误路径 + - close_conn / shutdown / submit 失败全覆盖 + +--- + +## 6. 验收标准 + +- 功能： + - 半包/多包/pipeline 正确； + - 重启回放后数据一致。 +- 性能： + - 主线程不再出现 oplog task 构造热点； + - 接收路径 memcpy 次数下降（去掉 tmp 中转）。 +- 稳定性： + - 慢盘压测下无泄漏、无 double free、无 UAF； + - 高水位触发时系统行为可预期。 diff --git a/config/config.xml b/config/config.xml index 47db930..4df1944 100644 --- a/config/config.xml +++ b/config/config.xml @@ -18,7 +18,7 @@ - incremental + none data kvs_oplog.db diff --git a/network/chainbuffer.c b/network/chainbuffer.c index 079dfa9..cadc6a1 100644 --- a/network/chainbuffer.c +++ b/network/chainbuffer.c @@ -1,48 +1,245 @@ #include "network/chainbuffer.h" +#include #include #include #include #include -#include #define CHAINBUFFER_DEFAULT_CHUNK 4096 #define CHAINBUFFER_MAX_IOV 16 +#define CHAINBUFFER_DEFAULT_FREE_LIMIT 256 struct chain_buffer_node { struct chain_buffer_node *next; - size_t start; - size_t end; size_t cap; + size_t rpos; + size_t wpos; + size_t used; + unsigned refcnt; uint8_t data[]; }; static chain_buffer_node_t *alloc_node(size_t cap) { - chain_buffer_node_t *node = (chain_buffer_node_t *)malloc(sizeof(*node) + cap); + chain_buffer_node_t *node; + + node = (chain_buffer_node_t *)malloc(sizeof(*node) + cap); if (!node) { return NULL; } + node->next = NULL; - node->start = 0; - node->end = 0; node->cap = cap; + node->rpos = 0; + node->wpos = 0; + node->used = 0; + node->refcnt = 0; return node; } +static void reset_node_state(chain_buffer_node_t *node) { + if (!node) { + return; + } + node->next = NULL; + node->rpos = 0; + node->wpos = 0; + node->used = 0; + node->refcnt = 0; +} + +static size_t min_size(size_t a, size_t b) { + return a < b ? a : b; +} + +static size_t node_read_seg1_len(const chain_buffer_node_t *node) { + size_t tail_len; + + if (!node || node->used == 0) { + return 0; + } + + tail_len = node->cap - node->rpos; + return min_size(node->used, tail_len); +} + +static size_t node_write_seg1_len(const chain_buffer_node_t *node) { + size_t free_total; + size_t tail_len; + + if (!node || node->used >= node->cap) { + return 0; + } + + free_total = node->cap - node->used; + tail_len = node->cap - node->wpos; + return min_size(free_total, tail_len); +} + +static void node_read_segments(const chain_buffer_node_t *node, + uint8_t **p1, size_t *l1, + uint8_t **p2, size_t *l2) { + size_t seg1; + size_t seg2; + + seg1 = node_read_seg1_len(node); + seg2 = (node && node->used > seg1) ? (node->used - seg1) : 0; + + if (p1) { + *p1 = (seg1 > 0) ? (uint8_t *)(node->data + node->rpos) : NULL; + } + if (l1) { + *l1 = seg1; + } + if (p2) { + *p2 = (seg2 > 0) ? (uint8_t *)node->data : NULL; + } + if (l2) { + *l2 = seg2; + } +} + +static void node_write_segments(const chain_buffer_node_t *node, + uint8_t **p1, size_t *l1, + uint8_t **p2, size_t *l2) { + size_t free_total; + size_t seg1; + size_t seg2; + + if (!node || node->used >= node->cap) { + if (p1) *p1 = NULL; + if (l1) *l1 = 0; + if (p2) *p2 = NULL; + if (l2) *l2 = 0; + return; + } + + free_total = node->cap - node->used; + seg1 = node_write_seg1_len(node); + seg2 = free_total - seg1; + + if (p1) { + *p1 = (seg1 > 0) ? (uint8_t *)(node->data + node->wpos) : NULL; + } + if (l1) { + *l1 = seg1; + } + if (p2) { + *p2 = (seg2 > 0) ? (uint8_t *)node->data : NULL; + } + if (l2) { + *l2 = seg2; + } +} + +static void node_advance_read(chain_buffer_node_t *node, size_t n) { + if (!node || n == 0) { + return; + } + node->rpos = (node->rpos + n) % node->cap; + node->used -= n; +} + +static void node_advance_write(chain_buffer_node_t *node, size_t n) { + if (!node || n == 0) { + return; + } + node->wpos = (node->wpos + n) % node->cap; + node->used += n; +} + +static chain_buffer_node_t *acquire_node(chain_buffer_t *buf, size_t cap) { + chain_buffer_node_t *node = NULL; + + if (!buf || cap == 0) { + return NULL; + } + + if (cap == buf->chunk_size && buf->free_list) { + node = buf->free_list; + buf->free_list = node->next; + buf->free_count--; + reset_node_state(node); + return node; + } + + node = alloc_node(cap); + return node; +} + +static void recycle_node(chain_buffer_t *buf, chain_buffer_node_t *node) { + if (!node) { + return; + } + + if (!buf || node->cap != buf->chunk_size || buf->free_count >= buf->free_limit) { + free(node); + return; + } + + reset_node_state(node); + node->next = buf->free_list; + buf->free_list = node; + buf->free_count++; +} + +static int append_new_tail(chain_buffer_t *buf, size_t cap) { + chain_buffer_node_t *node; + + node = acquire_node(buf, cap); + if (!node) { + return -1; + } + + if (!buf->tail) { + buf->head = node; + buf->tail = node; + } else { + buf->tail->next = node; + buf->tail = node; + } + return 0; +} + +static void list_append_node(chain_buffer_list_t *list, chain_buffer_node_t *node) { + if (!list || !node) { + return; + } + node->next = NULL; + if (!list->tail) { + list->head = node; + list->tail = node; + } else { + list->tail->next = node; + list->tail = node; + } +} + void chain_buffer_init(chain_buffer_t *buf, size_t chunk_size) { if (!buf) { return; } + memset(buf, 0, sizeof(*buf)); buf->chunk_size = chunk_size ? chunk_size : CHAINBUFFER_DEFAULT_CHUNK; + buf->free_limit = CHAINBUFFER_DEFAULT_FREE_LIMIT; } void chain_buffer_reset(chain_buffer_t *buf) { + chain_buffer_node_t *node; + if (!buf) { return; } - chain_buffer_node_t *node = buf->head; + node = buf->head; + while (node) { + chain_buffer_node_t *next = node->next; + free(node); + node = next; + } + + node = buf->free_list; while (node) { chain_buffer_node_t *next = node->next; free(node); @@ -50,11 +247,7 @@ void chain_buffer_reset(chain_buffer_t *buf) { } free(buf->linear_cache); - buf->linear_cache = NULL; - buf->linear_cap = 0; - buf->head = NULL; - buf->tail = NULL; - buf->total_len = 0; + memset(buf, 0, sizeof(*buf)); } size_t chain_buffer_len(const chain_buffer_t *buf) { @@ -62,8 +255,10 @@ size_t chain_buffer_len(const chain_buffer_t *buf) { } int chain_buffer_append(chain_buffer_t *buf, const void *data, size_t len) { - const uint8_t *src = (const uint8_t *)data; - if (!buf || (!src && len > 0)) { + const uint8_t *src; + size_t remain; + + if (!buf || (!data && len > 0)) { errno = EINVAL; return -1; } @@ -75,78 +270,97 @@ int chain_buffer_append(chain_buffer_t *buf, const void *data, size_t len) { return -1; } - size_t remain = len; + src = (const uint8_t *)data; + remain = len; + while (remain > 0) { - chain_buffer_node_t *tail = buf->tail; - size_t writable = 0; + chain_buffer_node_t *tail; + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + size_t writable; + size_t n; + size_t c1; - if (tail && tail->end < tail->cap) { - writable = tail->cap - tail->end; - } - - if (writable == 0) { + if (!buf->tail || buf->tail->used == buf->tail->cap) { size_t cap = remain > buf->chunk_size ? remain : buf->chunk_size; - chain_buffer_node_t *node = alloc_node(cap); - if (!node) { + if (append_new_tail(buf, cap) < 0) { errno = ENOMEM; return -1; } - - if (buf->tail) { - buf->tail->next = node; - buf->tail = node; - } else { - buf->head = node; - buf->tail = node; - } - tail = node; - writable = tail->cap; } - size_t n = remain < writable ? remain : writable; - memcpy(tail->data + tail->end, src, n); - tail->end += n; + tail = buf->tail; + node_write_segments(tail, &p1, &l1, &p2, &l2); + writable = l1 + l2; + if (writable == 0) { + continue; + } + + n = min_size(remain, writable); + c1 = min_size(n, l1); + if (c1 > 0) { + memcpy(p1, src, c1); + } + if (n > c1) { + memcpy(p2, src + c1, n - c1); + } + + node_advance_write(tail, n); + buf->total_len += n; src += n; remain -= n; - buf->total_len += n; } return 0; } size_t chain_buffer_drain(chain_buffer_t *buf, size_t len) { + size_t remain; + size_t drained; + if (!buf || len == 0 || buf->total_len == 0) { return 0; } - size_t remain = len; - size_t drained = 0; + remain = len; + drained = 0; while (remain > 0 && buf->head) { chain_buffer_node_t *node = buf->head; - size_t avail = node->end - node->start; + size_t n = min_size(remain, node->used); - if (remain < avail) { - node->start += remain; - buf->total_len -= remain; - drained += remain; - break; + if (n == 0) { + buf->head = node->next; + if (!buf->head) { + buf->tail = NULL; + } + recycle_node(buf, node); + continue; } - remain -= avail; - drained += avail; - buf->total_len -= avail; - buf->head = node->next; - if (!buf->head) { - buf->tail = NULL; + node_advance_read(node, n); + buf->total_len -= n; + drained += n; + remain -= n; + + if (node->used == 0) { + assert(node->refcnt == 0); + buf->head = node->next; + if (!buf->head) { + buf->tail = NULL; + } + recycle_node(buf, node); } - free(node); } return drained; } const uint8_t *chain_buffer_linearize(chain_buffer_t *buf, size_t *out_len) { + size_t offset; + if (!buf) { return NULL; } @@ -160,7 +374,13 @@ const uint8_t *chain_buffer_linearize(chain_buffer_t *buf, size_t *out_len) { } if (buf->head == buf->tail && buf->head) { - return buf->head->data + buf->head->start; + chain_buffer_node_t *node = buf->head; + if (node->used == 0) { + return NULL; + } + if (node_read_seg1_len(node) == node->used) { + return node->data + node->rpos; + } } if (buf->linear_cap < buf->total_len) { @@ -172,20 +392,33 @@ const uint8_t *chain_buffer_linearize(chain_buffer_t *buf, size_t *out_len) { buf->linear_cap = buf->total_len; } - size_t offset = 0; + offset = 0; for (chain_buffer_node_t *node = buf->head; node; node = node->next) { - size_t avail = node->end - node->start; - if (avail == 0) { - continue; + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + + node_read_segments(node, &p1, &l1, &p2, &l2); + if (l1 > 0) { + memcpy(buf->linear_cache + offset, p1, l1); + offset += l1; + } + if (l2 > 0) { + memcpy(buf->linear_cache + offset, p2, l2); + offset += l2; } - memcpy(buf->linear_cache + offset, node->data + node->start, avail); - offset += avail; } return buf->linear_cache; } ssize_t chain_buffer_send_fd(chain_buffer_t *buf, int fd, int flags) { + struct iovec iov[CHAINBUFFER_MAX_IOV]; + struct msghdr msg; + size_t iovcnt; + ssize_t n; + if (!buf) { errno = EINVAL; return -1; @@ -194,34 +427,277 @@ ssize_t chain_buffer_send_fd(chain_buffer_t *buf, int fd, int flags) { return 0; } - struct iovec iov[CHAINBUFFER_MAX_IOV]; - size_t iovcnt = 0; - + iovcnt = 0; for (chain_buffer_node_t *node = buf->head; node && iovcnt < CHAINBUFFER_MAX_IOV; node = node->next) { - size_t avail = node->end - node->start; - if (avail == 0) { - continue; + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + + node_read_segments(node, &p1, &l1, &p2, &l2); + if (l1 > 0) { + iov[iovcnt].iov_base = p1; + iov[iovcnt].iov_len = l1; + iovcnt++; + if (iovcnt >= CHAINBUFFER_MAX_IOV) { + break; + } + } + if (l2 > 0) { + iov[iovcnt].iov_base = p2; + iov[iovcnt].iov_len = l2; + iovcnt++; } - iov[iovcnt].iov_base = (void *)(node->data + node->start); - iov[iovcnt].iov_len = avail; - iovcnt++; } if (iovcnt == 0) { return 0; } - struct msghdr msg; memset(&msg, 0, sizeof(msg)); msg.msg_iov = iov; msg.msg_iovlen = iovcnt; - ssize_t n = sendmsg(fd, &msg, flags); + n = sendmsg(fd, &msg, flags); if (n > 0) { chain_buffer_drain(buf, (size_t)n); } return n; } + +int chain_buffer_prepare_recv_iov(chain_buffer_t *buf, struct iovec *iov, int max_iov) { + chain_buffer_node_t *tail; + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + int iovcnt; + + if (!buf || !iov || max_iov <= 0) { + errno = EINVAL; + return -1; + } + + if (!buf->tail || buf->tail->used == buf->tail->cap) { + if (append_new_tail(buf, buf->chunk_size) < 0) { + errno = ENOMEM; + return -1; + } + } + + tail = buf->tail; + node_write_segments(tail, &p1, &l1, &p2, &l2); + + iovcnt = 0; + if (l1 > 0) { + iov[iovcnt].iov_base = p1; + iov[iovcnt].iov_len = l1; + iovcnt++; + } + if (l2 > 0 && iovcnt < max_iov) { + iov[iovcnt].iov_base = p2; + iov[iovcnt].iov_len = l2; + iovcnt++; + } + + return iovcnt; +} + +size_t chain_buffer_commit_recv(chain_buffer_t *buf, size_t len) { + chain_buffer_node_t *tail; + size_t free_total; + + if (!buf || len == 0) { + return 0; + } + + if (!buf->tail || buf->tail->used == buf->tail->cap) { + return 0; + } + + tail = buf->tail; + free_total = tail->cap - tail->used; + if (len > free_total) { + len = free_total; + } + + node_advance_write(tail, len); + buf->total_len += len; + return len; +} + +void chain_buffer_list_init(chain_buffer_list_t *list) { + if (!list) { + return; + } + memset(list, 0, sizeof(*list)); +} + +size_t chain_buffer_list_len(const chain_buffer_list_t *list) { + return list ? list->total_len : 0; +} + +int chain_buffer_list_iov(const chain_buffer_list_t *list, struct iovec *iov, int max_iov) { + int iovcnt = 0; + + if (!list || !iov || max_iov <= 0) { + errno = EINVAL; + return -1; + } + + for (chain_buffer_node_t *node = list->head; node && iovcnt < max_iov; node = node->next) { + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + + node_read_segments(node, &p1, &l1, &p2, &l2); + if (l1 > 0) { + iov[iovcnt].iov_base = p1; + iov[iovcnt].iov_len = l1; + iovcnt++; + if (iovcnt >= max_iov) { + break; + } + } + if (l2 > 0) { + iov[iovcnt].iov_base = p2; + iov[iovcnt].iov_len = l2; + iovcnt++; + } + } + + return iovcnt; +} + +static int copy_prefix_from_node(chain_buffer_node_t *node, size_t len, uint8_t *dst) { + uint8_t *p1; + uint8_t *p2; + size_t l1; + size_t l2; + size_t c1; + + if (!node || !dst || len == 0 || len > node->used) { + return -1; + } + + node_read_segments(node, &p1, &l1, &p2, &l2); + c1 = min_size(len, l1); + if (c1 > 0) { + memcpy(dst, p1, c1); + } + if (len > c1) { + memcpy(dst + c1, p2, len - c1); + } + return 0; +} + +int chain_buffer_detach_prefix(chain_buffer_t *buf, size_t len, chain_buffer_list_t *out) { + size_t remain; + + if (!buf || !out) { + errno = EINVAL; + return -1; + } + + chain_buffer_list_init(out); + + if (len == 0) { + return 0; + } + if (len > buf->total_len) { + errno = EINVAL; + return -1; + } + + remain = len; + while (remain > 0 && buf->head) { + chain_buffer_node_t *node = buf->head; + + if (remain >= node->used) { + size_t take = node->used; + + buf->head = node->next; + if (!buf->head) { + buf->tail = NULL; + } + node->next = NULL; + node->refcnt = 1; + list_append_node(out, node); + out->total_len += take; + buf->total_len -= take; + remain -= take; + continue; + } + + { + chain_buffer_node_t *part = acquire_node(buf, remain); + if (!part) { + chain_buffer_list_release(buf, out); + errno = ENOMEM; + return -1; + } + + if (copy_prefix_from_node(node, remain, part->data) < 0) { + recycle_node(buf, part); + chain_buffer_list_release(buf, out); + errno = EINVAL; + return -1; + } + + part->used = remain; + part->wpos = remain % part->cap; + part->rpos = 0; + part->refcnt = 1; + part->next = NULL; + list_append_node(out, part); + out->total_len += remain; + + node_advance_read(node, remain); + buf->total_len -= remain; + remain = 0; + + if (node->used == 0) { + buf->head = node->next; + if (!buf->head) { + buf->tail = NULL; + } + recycle_node(buf, node); + } + } + } + + return 0; +} + +void chain_buffer_list_release(chain_buffer_t *owner, chain_buffer_list_t *list) { + chain_buffer_node_t *node; + + if (!list) { + return; + } + + node = list->head; + while (node) { + chain_buffer_node_t *next = node->next; + + if (node->refcnt > 0) { + node->refcnt--; + } + + if (node->refcnt == 0) { + if (owner) { + recycle_node(owner, node); + } else { + free(node); + } + } + + node = next; + } + + chain_buffer_list_init(list); +} diff --git a/network/chainbuffer.h b/network/chainbuffer.h index 78e3c3f..1a3c4db 100644 --- a/network/chainbuffer.h +++ b/network/chainbuffer.h @@ -4,6 +4,7 @@ #include #include #include +#include typedef struct chain_buffer_node chain_buffer_node_t; @@ -12,10 +13,21 @@ typedef struct chain_buffer_s { chain_buffer_node_t *tail; size_t total_len; size_t chunk_size; + + chain_buffer_node_t *free_list; + size_t free_count; + size_t free_limit; + uint8_t *linear_cache; size_t linear_cap; } chain_buffer_t; +typedef struct chain_buffer_list_s { + chain_buffer_node_t *head; + chain_buffer_node_t *tail; + size_t total_len; +} chain_buffer_list_t; + void chain_buffer_init(chain_buffer_t *buf, size_t chunk_size); void chain_buffer_reset(chain_buffer_t *buf); @@ -26,4 +38,15 @@ size_t chain_buffer_drain(chain_buffer_t *buf, size_t len); const uint8_t *chain_buffer_linearize(chain_buffer_t *buf, size_t *out_len); ssize_t chain_buffer_send_fd(chain_buffer_t *buf, int fd, int flags); +/* readv zero-copy recv helpers */ +int chain_buffer_prepare_recv_iov(chain_buffer_t *buf, struct iovec *iov, int max_iov); +size_t chain_buffer_commit_recv(chain_buffer_t *buf, size_t len); + +/* ownership transfer helpers */ +void chain_buffer_list_init(chain_buffer_list_t *list); +size_t chain_buffer_list_len(const chain_buffer_list_t *list); +int chain_buffer_list_iov(const chain_buffer_list_t *list, struct iovec *iov, int max_iov); +int chain_buffer_detach_prefix(chain_buffer_t *buf, size_t len, chain_buffer_list_t *out); +void chain_buffer_list_release(chain_buffer_t *owner, chain_buffer_list_t *list); + #endif diff --git a/reactor.c b/reactor.c index e6732f2..440566a 100644 --- a/reactor.c +++ b/reactor.c @@ -12,6 +12,7 @@ #include #include #include +#include #include #include "diskuring/diskuring.h" @@ -200,8 +201,17 @@ int recv_cb(int fd) { c = &conn_list[fd]; while (1) { - uint8_t tmp[RECV_BATCH_BYTES]; - ssize_t n = recv(fd, tmp, sizeof(tmp), 0); + struct iovec iov[4]; + int iovcnt = chain_buffer_prepare_recv_iov(&c->rbuf, iov, (int)(sizeof(iov) / sizeof(iov[0]))); + ssize_t n; + + if (iovcnt < 0) { + printf("fd=%d prepare recv iov failed: %s\n", fd, strerror(errno)); + close_conn(fd); + return 0; + } + + n = readv(fd, iov, iovcnt); if (n > 0) { size_t cur_len = chain_buffer_len(&c->rbuf); @@ -210,8 +220,8 @@ int recv_cb(int fd) { close_conn(fd); return 0; } - if (chain_buffer_append(&c->rbuf, tmp, (size_t)n) < 0) { - printf("fd=%d append read buffer failed: %s\n", fd, strerror(errno)); + if (chain_buffer_commit_recv(&c->rbuf, (size_t)n) != (size_t)n) { + printf("fd=%d commit recv buffer failed\n", fd); close_conn(fd); return 0; } diff --git a/test-redis/README.md b/test-redis/README.md new file mode 100644 index 0000000..57d6e2c --- /dev/null +++ b/test-redis/README.md @@ -0,0 +1,107 @@ +# test-redis 压测记录与优化对比（复测） + +## 先回答你的两个问题 + +### 1) 为什么之前看起来比 Redis 快很多？ + +结论：之前是**单次样本**，抖动很大，不足以说明稳定结论。 +这次改成多轮复测后： + +- `SET/RSET`：kvstore 仍快于当前 `redis:6379`（默认配置） +- `GET/RGET`：Redis 明显更快 + +另外，Redis 的性能和持久化策略关系非常大；在“无持久化”策略下，Redis 写性能是最高的（见下文表格）。 + +### 2) 为什么 GET 的 keyspace 比 SET 小很多？ + +这是有意的： + +- `SET/RSET` 压测为了避免键冲突（`RSET` 冲突会报错），使用超大 keyspace（`1e9`）。 +- `GET/RGET` 必须先 prefill 全部 keyspace，若也设为 `1e9`，预填充成本过高，不适合日常回归测试。 + +--- + +## 测试口径 + +- 时间：2026-03-04 +- 工具：`./test-redis/bench` +- kvstore 测试命令：`RSET/RGET` +- Redis 测试命令：`SET/GET` +- 通用参数： + - 写：`requests=10000 pipeline=128 keyspace=1000000000 value-size=32` + - 读：`requests=300000 pipeline=128 keyspace=100000 value-size=32` +- kvstore 复测时临时使用 `persistence=none`（避免历史 oplog 回放影响）。 + +--- + +## 优化项 #1：ChainBuffer 接收链路改造 + +改造点：`readv` 直写 + 回环 chunk + 节点池（减少接收路径中转拷贝）。 + +### A. 改造前后（kvstore） + +| 指标 | 改造前（旧记录，单次） | 改造后（本次，3轮均值） | 变化 | +|---|---:|---:|---:| +| RSET QPS | 260604 | 331063 | **+27.04%** | +| RGET QPS | 294951 | 288107 | **-2.32%** | + +> 说明：旧值来自此前同文档记录；新值是本次重跑 3 轮的均值，可信度更高。 + +### B. 本次 3 轮明细（kvstore） + +| 场景 | Round1 | Round2 | Round3 | 平均 | +|---|---:|---:|---:|---:| +| RSET QPS | 323041 | 352476 | 317673 | **331063** | +| RGET QPS | 271069 | 313658 | 279593 | **288107** | + +--- + +## Redis 对照（同口径复测） + +### A. 默认 Redis 实例（127.0.0.1:6379，3轮均值） + +| 场景 | Round1 | Round2 | Round3 | 平均 | +|---|---:|---:|---:|---:| +| SET QPS | 299221 | 130792 | 312117 | **247377** | +| GET QPS | 349242 | 343573 | 353091 | **348635** | + +### B. 与 kvstore 对比（本次均值） + +| 指标 | kvstore | redis:6379 | 相对变化（kvstore 对 redis） | +|---|---:|---:|---:| +| 写 QPS | 331063 | 247377 | **+33.83%** | +| 读 QPS | 288107 | 348635 | **-17.36%** | + +> 解释：这说明“kvstore 在当前写路径上有优势，但读路径仍落后 Redis”。 + +--- + +## Redis 持久化策略对比（写压测，SET） + +| 策略 | QPS | avg(us/op) | 备注 | +|---|---:|---:|---| +| `none`（无持久化） | **492561** | 2.03 | 最高吞吐（但不持久） | +| `rdb_default` | **285885** | 3.50 | 本次“持久化策略中最快” | +| `aof_no` | 281870 | 3.55 | AOF，`appendfsync no` | +| `aof_everysec` | 266878 | 3.75 | AOF，`appendfsync everysec` | +| `aof_always` | 110793 | 9.03 | 最慢，但最强一致性 | + +结论： + +- 如果包含“无持久化”，最快是 `none`。 +- 如果限定“必须持久化”，本次最快是 `rdb_default`（略快于 `aof_no`）。 + +--- + +## 复现命令（关键） + +```bash +# kvstore 写（RSET） +./test-redis/bench --host 127.0.0.1 --port 8888 --mode set --set-cmd RSET --get-cmd RGET --requests 10000 --pipeline 128 --keyspace 1000000000 --value-size 32 --key-prefix bench::set: + +# kvstore 读（RGET） +./test-redis/bench --host 127.0.0.1 --port 8888 --mode get --set-cmd RSET --get-cmd RGET --requests 300000 --pipeline 128 --keyspace 100000 --value-size 32 --verify-get --key-prefix bench::get: + +# Redis 策略对比（示例：6381 配置成 rdb_default） +./test-redis/bench --host 127.0.0.1 --port 6381 --mode set --requests 10000 --pipeline 128 --keyspace 1000000000 --value-size 32 --key-prefix bench::redis: +``` diff --git a/test-redis/bench b/test-redis/bench new file mode 100755 index 0000000000000000000000000000000000000000..3959e4ed7d8423e8c9ed59c648991f2fbb2ecb99 GIT binary patch literal 32328 zcmeHweSB2ang6{rImt{ilS~pK2?Q8k0s+knuZA~GfZ&7|i3l209EN1_0?AC8nL%g` zP-0xBV{GEDwA3Gq(eAdi3oYGUTMAl(T7&Jji`}-1t5w>SHW6(xR!d)O_WPWd$(>80 z{q(c{{C*!VWS(=L^SqtsJolV)&pr3veUz?91ZTSO)R%C$1h ziQfWkhL((c632=1asf~(jtiDVXp?v=Ao1l=VicG!(3*nELW0CM*5OGLB?@XTktaSz zNtrjpOHz@7sywc*RLIdxu+%Q_3L5#yH?HNP+_gRtYO>zt5-L7&7xB3zpIh=Ns7Ze+ zsQgJjp}#UI&kWrtqr=UAHOVyz07v1Zhw@+Y|EOj0P)waJqQLOm2m$hxHefYv7`f5#FOPwz zI=S&@KBkT#{|6u(P5$XI@U>&$y<^D#@fh^$$Dn_64EoG5=vR+{FBt<*-Gdu{=3~wn z^jknUnt!g2f$tpy|EV$X6UNYcGw9v;GarutF`7N^0H3K%(A?F6L-ijwLjE+f5ad&V zC;D;|y@vCuv8l~p*A$|YskC_R0?k{sX}!13-{x;<3Wfb`o7S&v4hH<2eA}D-n%CRV z5)61lVP9L=>lLJ<5^|BN+x&iwr&b070e@}y&ZcnVrluBuFcKE{iq=+tpl)TbrNtMh zt7vQ3CD6C{!yEmr&3m+ZNZ_dr+joGU%&jFu7r_!3QM=Py+qlzP?`vw-LgBV%e?V*T zw}kv*6#LuS0zq$cu+|rD3UYjHODm}ANO_B|870J1>kIj10XPT~4n~@rwfg(on!0*^Px`qS$o6{4Jn5VkUsFH}1tbI2L%U^N z)rytgdBqFjskt}ClkSz_l2I%qr0+d_sM)@LK8+xi*^8cI@EZm;;1)U;B)18r|{b(-e!s;mA|k-@KN7L z{i!-G8t`&Tz>5Y9cx5{kTruFsDyE0c0kkd`hW zv{r7w3lDR&<}u*am_jAh20Wk72y(5)fH#hl+YETpt&TuTC29GITiCDFdFyFm?19 z@Hq+zc+P;o&VWB}z~>t9A16Qdz{ei=*aQEsdEniQ$$#{8yk+ym*vps`dOG{Umf`cB zj^}O9@v00jv;aLcTwn#rn4%GVBb5vd{$hA|xSQu`KpPr-DW0c+Y-sRoJWm7I(BL=Y zc^asP1|N^-X&@UK?1|@T02>bo( z0T>#b8_&}~H8eOio~Hq7XfQLLr-5l`P><&`c>bNgsQS{x0rijPX<$PA<9Qm8Q2%(I z1|rm7<>Afto5-81H11F2PvCOjnEyvp{_josL#BL}Dc@upJU9=eY-09(!HMOFFhS^T&do)u8X~oDVe9sxsQY8 z85r<~bB2~_+V8qOUCZ;p>*-=y1QuTn=OFZdgfa?!`|!n#Df`h{+H(?uevk^g7XFgJ z?7w)TS3GBbbF=5{RjWsT!Sm8z!s8&|AEf}>@Wp!Gj*6eokNwMkj$$ox{bo6P`V$~o){$~ z_*;*^%~z|Uf2e|mT??>L1)B|tB7Ei45s;4XG z{waiVQLew-b{Uf$k#x}6&1w3}UF9n>Bs#ZTqWjD9dmq##*6ogCXUvmH$ezPxR~JfE zQ$)5@X2*$a*)w3!2EIlnMW5wm(du%7{pB7iN~}>SymG-4jhwIZ#F7dMQSOOWUY*8~6h>Va?md3d;=QfSTf1CVp!G<6i7=#Qx8*<4NKm4VNIET(n#MGzR31D?f|7s6?j z%J`P{KwmiJ6xw3=GP&hE{Cx!;;w~ot*LmhUxZ*Gy2t!JO!`(1%0Ono1*1RjiCI};i z(Vvky)1d`6qyu&3G^z9~KEF#U&m<)G$i*?VzyBwb_VQ@^tK=I`bn`ir0sWP4QU5*o zAbQ@Gz;VZ& zi&onPyE$U48h=HcKm)1Vs*KJP@JZaVjE9dR`(b89E2%l@O!P}ZXaC5}<4)->=NC(Z zPM9FcR!Fi(6Ulhn{0bwcB%311LWyK_QVu=?GEvv>BQ<)PP-FfASEkgMB|=6`yjpwL z^A_u>b{VSXFhs*SiUtv(pGJg}w>d|w+=c!#qq8(_f#*Xl*dZ-A@dqyCju8viN(*v7 z+Jd9Pf(+t_mFJTFf5B8kc;FmTV|(B|T+5ROipJER|Izv7{fpxIJs+xnztn%? z>R9!6f3*Ii;(U?yca#3_AP!{xks7=HPE~)Q|IQKruao|t@zMGx3;jHxW92hCuGVEd zxHN9SEgx#YVH%pz0LOpN9WZ~yfI@>}Q%5BGlj_33pW?I_yLOdQ_=dklc%N@Y zeL1E3a#HtYy<669Y2~XK4+U^?o`MnR`Vt4%@sN`^xSGZ^PgmtdPxOUnNYB8FzvFY7 zi=L)F?fZ`|#i%{NhqXkR&wQEb5(jEWh8wuQ>H<_`Kp{V zZ8greS~2*a!uWd<SCyJM9gsqBM3!2W^BWO@{JiB*CP?H{Y` z1tilB1D*%$2E-_{{vrlK;yd+SbUfVK-{kd<(6nTMnwI?h-Go|pyj2a$FhocCU>Pi2 zfAU(3P{cI>_76M+0mzd@Ag{atNEQ!B3&^5#-^H2FEhrhqg1gNYxM^Z!*4ke=+N5Y0 zyclxqP;v#s4FZW~MwjoTd`#=O$|8ll2!jF5<|oE8T?-qqxJULM8REg3&5Mw~A0mYm zKC2>1h9n7A}`H zaM3h0G9_N0*k(8$O)jRc9#fr;AusB*K~?h*p{P!mml2Abl@xKS()&=#^+|{33evbo z?oGhe*!s^Mb9v^Ht=yRH{kr5M@D4G@dSpMKXF~ z)n%Tp&MUxHMR5|mge#L3xK{rzt5kkGc^CDS*s-kLqbox!5_a~T@KSEGX+f_dE<+Oi*i0s$J3lWBG0kkqHi2{9>WUC zFF@E}3lIkW&SOBv4$<$3NiM`38BQ@b*?^AXnv@sJem080Bc}e&F3@nyyvsezE|k^{VzRd-?Vt5=P{O9XbkKc*U7aYNtoGZJ<;#S z1<#<^csApazKn16MY%<>LsUHwkW;}QJ9H9>gP4}d=EM0E>l&a!(`j`XOK6FwRp@Ox zt)2pH$J6x7Pb*4@(~3@Rt_8lk@Rsb)220@7j;Bef7LLR?w>aIpfs&`&ex7{#Ac6Sl zMxi=#x^2J82LYpwa9qa^L_b8a#xH%Vb(q3)QP*YPC`wBc)lh5y-)j|Q#}(`_Dk!&M zPB)^V+Sb8C2q!POdMS^pj069I#fkpTe+7n`Yw$3BVuvWJyriEO%M<9bJE^W@#9
hN$#5w!a0b|Xy++Vb?{zF3(r?ort7){`>3r|yOdv6?eYZ})WPk8 z;bfOG{0?+lFydIjia#|5lLmBtd8AOdUl#z)+f5f(bcyo-->=hJ@4C1$YKK&N0 z4n?2Nho$JuTmZ&?YXA#KT{Ow+ANVd>5RXCfd zLR5SL**|a-wLbJ{gXeHvEQqwQi71bVXzn!lB@lV~JGVgqpUw4mHX=#3Eue#=BNZ(TY~M_8PROqc7=lil>WJV?83a9sT)qSCK^f2K@7?;= z^Yq_?j*oYBR_f~XKI-XYhldRBg{^Y7ysEWQUg5O_p>#k1p})R%(;6{2=E&$M^;MN4VlJq{9`Dkb3Tl z&l4glUGXVVcjb!Xze;dLJ4%HsXl9Cjqw-*V#&2J&&nP*&YdP9q7}>d-Y(PJ$TpLz$h`yqebv3a^JYB6hKDuy-h93#NtFl6_Q1ckj9Sj6#FD8mL zHdjU8tJqW#{qtr|SJCHyzJ1;74=At(g8vOYXFsrpr_6mt-hWva9aUI|uH3mar_p+weo7~2u&R_3~G>4bC=gzwkdpNO+ zlM}ZF+dxdeM)ElEqWL#2=5#H=I=_1vc1_-oeUA6HH0}1+fgw)E5uC8i|A~k{6b`wU z1q>47gmX*zA#^vj`mt3Taz`ayw189X^zR9^`fA6b5&}8pE?;xRUlhV_OQXcN>iP3H z73xsbM*BV8%Swxj!OqQ$Q!OlY7xLP;d6k2K<~_6He6=lg?q&EHsac%w_R3A14DKoV z5VBh&8ERG3+UBos+70h>NHXKnw)+FMjY~kZMC$fKvC&J^v~g;cwB;8#M3{`K zcEh4(cj#v6*X!b z)@lWzg64=847O;93)+W`W?xcJ=cawr{yOSrQss_l89v)1_4WQXS&r$Fb~lgX=H{(# zW$*HWh&vK$3N&!;NML6mcwa!1WRmj+H?~wSC+T-{XKi(3-#DrO6sadKA?a3*Sq>}R z$~7e4sEa$tOC11}z&l=egt#G3CR4F_9Gk}7Elr^oU%0kWv=9$>QB#9GZT=8?%&l&J zv#%BB*7Abl`Sl@pI2iUdb2FfTLvAh1LPdgYA&&Y)D`0VHz1z2|0f~9_kx)snmAoVl z{||pyL8<)@6j$gItxJz1O?zCc|7&>oB4F<;!^1XAe@_Cs0k6C|JWT7AFaLaaxD5G8 zuMH2^0R9$mH{ips4-X#(bX*jZ=gXMyT|oY3Os1~_#sG71Wl@0Zj55Hl0M-C*$2Hn+ zz%VY|4g=mLuV-grmG}bk-MH}K54d2~Pq#nYIRTab8~%du z>0=4ypO2Tn1^jPNzBi#f5-)eao^@!0*Rfk7F5mjIWFjE>H=sNXQ|wI%{A=U<>rj3( z%GV{7uZ)*BqkJdIA5AFliI+cw@=GW`kWjuSUVa?qw;&eiZerX%%Rk4<{|)88M0plq zyxeNniGb{X3;3TPc7Bym{_}V_?JRl}G4#VEBmeVB#82&h1Ill~O3OB8DED+&ts5A( z0a%UNZWHJZ;wFoq5UbN&pS4t_=RRz$Om}xBRi+m{#8#!3KA7xDFY8FTCA~b5URIG_ zT9ID3BHg_rJ$FUAYel+^Kf`bYa!x`Hsr&f&*aIJX;QxOQsOON>b4S!w(P4(Pw}K9b z0PzQ*)Uz%ZZV(07iNX){yc4Z+(m`Vi9oUD%5B1y=^^#5*nXzA(Uj(+#3V|x;%bdl>9Va(xILmQqM*;V!6nO-z^PR zcJtf!AdqC5(mz?S#p{O~pS*rh$62wI&+}zc5Z0|!Ip@40^M8>B;noE&SM`wM|JOkL zxoeZeQaPz8m#|vGZ4$OhxL?9<340|xDPf<4=Ow%-;S~w-5G6lc61pWUm9SjGY6-VV zXfpRBv+8-eTUM@I;x630JrW2<+>47B6qgpwjqu#uz4LITzo2;DY=PG(!bP}*WQeGpVCr5#_ zmQ%^h@kNkjv9LcvQc@Db75S7fiZYh=6SO|KoPs3D8(Zm%%S>K}K!+E3QK zgVN7Hr*-1Ec1j;aI_ahHq~F#8nhR{Mts6Ky&CJ>Slv)&M_KncP7TIZ=vdvDrKG}^t zQ?(wDyia`u-!-I ztVxSP-ERAODp{Yk0Ga*vwN#&5Gpdj|!21QZ!S*ENbl5i#&utmsL*}5}Pnj*wLm=z6 ze}gQzd;EEn9JXgu$vt_mAk%CABg)xUZq70LdsLr$?cV|JqTm2Bg- z_1QyY&UV``AmDk;`U!|jPRX-FMJn}k?E3UH+Q6A?`v(vt+o-@n4NP`9S0F)HYKSzN z3MbosKuH%#vHc85PWuiLoHkoaCs(CJ(?0-6=8c_xKloBU!!hYIAuc5*(rZCto8YJg z%CQ_K%)?)XmNJW5^AjSx5y|v?BAYmYN)guq09z|iF1yXvj*NXXY_X?09tVSFC)e83 zGEX3F(`@-DccrG<{!!7TIwakG(51R^XjX5dQ|x7`Hbvl5Amvrjwe(&{PJ1l)$G)}I`7UPkJ-50&U?72E*Bo5*PL5<^>TR_Iq&5qZhIM= z<@Dx}utI2~M;wxEGw^FAJ<1{*;jBf0OUtu4A7sPCKpmCy?*xNpI|j?oJ0q@#Q)FW~ zcNrSXd7o^o`Z~?In-@G_ryj_;hi4Aj|3sNj@@jV5pM-7BPi2v;!*;rGcD7GG44u99 zFH_0hoX?@;m@8b{syQD>p-P>!&!&=tDO8tp@Ix+OvTZZ0vXW`e&sfRm3BnV4T6#Yu z;Et>U50D6Y?%P7vP)-&x7V)*C2(X%qCm(sX9^lWan%{~(-eOwkv zbJ+g~OZA-ecgRup`>Eu*^h>0{UPYPlJd zBF~iCuMk;2&y?A3BC<(bPr3aLDw)j9@o3g_WWp5df54s;GCj|&PqqJ)>adcSrg0|C zo`Z_((|N{bXT&o@nq35Y^#TX6z>lc8W=pxnzoz;Ty_cQT>V;drb z4*keqF!berjRZF6=|p1fd1!C`r$~2$K9NeTJ^S}gS_1~XQO_b1f!H_c$E2Z2ANMLp zT--akHoRNYcj(zknDbkEj_sX9EwAs?$MGiZIlp%jErjUJIu99Z&#AqWQpsU0IzKzC zJty~0`g>vt=-fBfo+Epwe5;U~sf7g4UW7f9n>77lJ(GG2YftZ9&HeTqQCb9(=$MQZ zN=-kmyT~Wjp1!@4rU0v-P>6F_@k+M!1D9-j6n>vNlPWhs=flICX(U3^_^J<}e&k79 zv`nidf?`p(gYDCTl&z!^IHNtE*KDGG1I3%Q=j#3Wyk-;iCBn!n_fNxmE5bo9J;PZB z?w@oQRdtfi8xhWEpWGH`Kv*cm$@a-u{Y7Nxsb@LM#r8>GCl-j~6sY+TB|v%>qdS z)9N&Ro_-yjC)S?x?UViwYl;zIi*A4S6gJ+qcm^tMm>5t+77A&$4BJN{Y$DbzGR zd;pT#%}VV*w7b6}%X>H#%G==dJCulH5J3(02--B+Mp4JRbMgi6zK4N*85GmUk!|nu zwp`^V9!=vbe1Q6q*KvV717hEwCYth8`wTSY`-??Oo@_7RE&0AjG~}`N0^X4C-ywso zeI{)Fy==1!?K623U5&Sp`(Zb!!-P~u;wbLS zQ~2Sv8^{lqo|=Lh!oRkZDsJhiF6bjKzP5(yZRzn8ETBH*wY#W!Ej_DdwrTq7nIyx~ zvkiWK2&L1XqUd~y>$^{0R?{4g4*keOQr}BTABp&JT*UMzN!<&a`MpVk>3{%WPQwGx z)Mzgh$VS^YgYE|DFDwwvbaLMe-VpuMWwY%oc$q<7e+ik7K4RZ2-k+VzBuSQ@@>z6S zK!4FOk!Pzl2k}vcN0P4b4(a-l{Pc37KgG?vTbjpLt$E#k075eFlr-;{H1Cu&uUDG) z9ckWSY2J6FdENVF;O35|pO)nZ_Z3Vg^H0n2{nGq1nWPE*5DZ}l@ zM2G5qc_d%>3zGMZOgeuvr2XrVrb9n6Cj<9R{)1D@q%@*oCc5&IJj^XUjc8zEoL)%* z^;It9j9eGm#6^%09~ftb(Z_u)ho=5y)$~)uu!DQ;Rk_f^?1fDH0Yp}=*--)jOfrm?xQnrCm$FrJ@-s`>^hQEAGhmXX&1i-Lg*uvSQeOpdibUZwwf9Az9ijOLItSS%w3wk zUAl&1pY%%OMHed-Ei=`E`a)xFH4D=r9_XAxVc0)yL=|z0ELc@4~yh`D7Z71 zu8Zj8nzNG5yj9M(_!`OL)Jp|ac zp_N2JZ6!^C+U7`|zXY#B^e$Q;-;3A~h?Hz^LO;Yw@H(=RhT7U|Nfs;azd}?=cuy-P zZ$_O7nG*T_INWfi-6h2s32;xBc#0e85)E%$!y;f&1V4c|2HRVZ#yO6Q#Yk<~ySi#! zB@h@wM^eNX1QOvH+7qJpxi#a3bV>wgqs=iGI|Qc`MYvF_zg8eACuFqweRW8oJ5evZ z;~9+=Xm*S0GOV+7#@=PRHP13vXN5OW#%bH~pC?%6i8Ciy_6e4Cg1JsyV%e8i-X)fC zlsT8N><*@%U<+=tEn{hy7<+)F)i8U5?F6%h*+eXs>TB3I0j2|3>Y1BBn7P)n)LqPR zf??NJlD;MR4YnY87lfo=V)hPZvlX)$f^CWbuE&^T7fY>Yj)oDIo52FiuNVun+$}J6 z8FSd&Y_4Frfj~XW*~0SIu!%J+2SroPuyLTuMzTy$%@VDeu<^Ytr-rQxvMKi_$~?fN%)A~pZ4H}R!}8u>nGwh=W_fGb zs;{!#d)RoH`3lRuH=eQXInHve_8K;&H!gCSu;NAn^(+%+9x#~sGqzlmlU@1&mfQ2U z^1_21|69D|D>OW}X*8Q}7UfF`nDxr~@YWfPSIS-@c@yf(1+HO>sH~nMV#*pexrXIZ z!@n5!=wC;bSci#QWK&K|l$97n*S1cgPbaQ5+EF3Az^z40@}PnAacg%mm-XIoIS!T& z8jJOAH&*4iOC8oFrI}P!mM-+)0eX%+g&XE44qdg2bg26@fWAs zT2?mCG9!yQf6cNwnCmoiu4S$c=0v)KIdCT1I#`+QZrgD)g)A$v-8IBA<7cR`@@trD z3!Tw8AFn40h}2*o44a^KKh~JA1CpSEo>@@ zt`{B40ai|;>-qT%4GSrUlM7WFcZS*F%QGzXK4P?^BRE5x$@N4NW-R%>1Ot4@yDZmd zQ5Tf<4VH!e33I&>r)X2yC3Q?KE>(^UjU~WU%l&qQ? zEp?pgH5K(b&a%7N_{Yh#{4H#zZ40~ZG|L3YdYrjX*R?F~3>(iYJ>@Ajo2obSG|R1L z*^jWiAvPY3QA)Koy93=`?j`1WoaOYeam1CgmgQe!<9gYoqb%olz|Aq@j!TbiLx{j; z?y(udV=&8YD`C^B2#>mC612?9WG+fHvT;2u|2Rbt^hviLRYKCa+s9eg!L!HNRO^lB z*!AcZ!t6%d5LJ}njJY)8?5PMuAgy+c17|Q6z02%9=)I>hTdzuuLybni&-IoBGD@xV9xg$IuMMTPRq1x41q&z*;?l8292J) z2EWn_q>_?tFA_i#2*92|Gw)!mjLo~1t>c3Yk0ndD5T5)H=&Z>ig00DxMH6^5S?NSs zvX9`7dT7uQRYhaFncFP-WjYAm1{+F zJ1nd9OU6L3q+t%~4K-q4E8Vte^|gik-pKrUc=hiFdb#dSZ&<@VNN*%W6Fs`;A$L@I z@mgI@1U4^S;qj@fc&jg_g8l|SNOxnNhIjQs6o1<;r^Bna-7DwbDDG;zi(0~wfPZ;| zKY)$TwaEI~Y8#h}*MGZ<8r((ozR(^_JKKVRU?fx&3EbBds4EK7%ZjnHxt_lx7@M(! zVJHf=6=9AT#FkZ6P}}SaHL7ClvPVH}y&BYTRyTqG)Lol?y+1G%IV#g`pWv+Q? zbE&^ABq}Fgatt|}F#Q$NVXxR|3NPWk$w6-*1poM3ur)iRiQT&NZegt{1b5L(i)p(0 z$zZ^*t>3V!(o4IHfN-m8urrmR}C+AZgBITNFdbI5Wr4!x?zviB(K=Gt_7*p>8S%v z?0{7}h*3!|P4bOXbJ5&*DRZRNOFPz8RW%=f%QLSc&E(OPZ5rRt3y)$4av-RJ!&kS- z7uu=GX}t!^chN1e@GdwG;Y7QC;mKM|<-`8l!*~~S1J_AYd!Z2=^j2qFaS&64$Do%k zJr*qhBfmiL;(>|~W%w3h+9|HI!>i&o)2kzaTK>=lO2iWnS{sifVYQcRk~^Kbt#rA8 zSfEYcUT>XmPjgd4qX=seAG8ge_9&BM`ST7VR=472*+~3zuZ=VcwT0RbEo)sRPaSHa zraI2bgMf3YS8a#=yKz zNn^0ZkC*WK0wtkFUz@)~ZuG~t@!eGJ^|t!z7HW||({67Fy#jx6w|r-J$nUS?=S&c{ zCA_=%x#sb=3A?3cnC;>Tk(?bkzL?iBKcH=KGw=9YG|!0 z*xEutL_vW*Mx8kPq~#}$hyrRY)SjPb>#J%iSBcWSwnYM7A)B6lK-T~xBJ>;&yh9;a26YsB8t*qLJ}ErWHzA0F&2`Ep z6b*C{tfG?FfT#BU3{3;7?Rforfch)pg1ByqN&aqgifL+5?m>Dc0xwoaudCsW z?DXn$(Ywhma~`j_w<*vBJ8C0s-r8Vuq$NOo3VLLrlb@k@UHUHWRBt2z+w11>D$x+a z1N?JXF`(n868@Kjf0j_k0s=7TeX@|@ktup zTQwh6tZej-icgO3A2jkX)Eoj z#A2)Yu;QyT?osjST5G16n%1>4wEd&vUGeYR7`d!krl$67CgQWODZz*lX{(m4sqdE< zORadyVE-Z`M$zZQ-+ON?wQARCYX4^SNtQQD;ic#_j=lD~)3 zU)#QifgjEOUMWYd1Fk}zjxXbnbZ+z9UKB1EG4E?~LY4{3tew;;ifNJnEI0$ zpdU@|O~7O8ljbtff0^vDT-&c!jv;5u82HE-_^vVVUm64d?J@AQe{-~Yy*UP+ZY5BC z&Hix$KboAWW8i7O>1gdz34A8L@}%}7;?`sXq9K05~f-_`pZfTnlWW#}@`DZHd(~N{Ny&U*aziU+677m5+6l-ygOYd+SeWVC; zZ*N_sr3F{zxNM5&_=GosR9#_->jHiy$|uDm*Y7fg6&8`=SMO@=z%WlYTvHpa^sc?y zQJ9g119jy(+VzUpyLw~A`bux*t*hwsPKg!3<>sm_w^ppLTA5gaYY9B{?Db;h##`x; z5Ro`85d4z{Sx>37Ka8?9@o<0&&5%^4c$5sjond z(2^wap{hjOHD8cQECvfcRh5W?k^F;IiC9{O2>ZmRsJJ`LZ1PK1X0F7qdwJ_ZL2o0z z%te=VwA@4sE3~Ro$1k=Mnfc-a|KgXSfq8I?k8LGVd%fF3SoGqN!!IYzU;5(WDz5oF znD{K0Sr)!grb3p2VC*lF-J(7$miPs)M0uztE=&_~T5)Jk3zjYb;cyNcRSK8H{O*$AP?+ff1~WC8nK!Y z&j}f#ztT8|N<)>er3u0WJDN@_rk6R;5=n8Le|w|>l)gX%I#!u&3e*Qx+P8gsn}3(e z(OW@O8Xic_t91qi<#dI1 zL-50FzYkb!{E63JJ(sAnj+PXBSmot%E37F8R&&OSIO71o>3)tMw8EuSi0r-(3GMNd6U) zPOYmbc#)ijgZRm3X8y;4BN8PalQ(flNbzkv%!g+_k2m)jE=b*p8cUnEm%X z$*XFuPz9s6xDu8=nqZ&YF$dfrgaEVtOF+}wi|lsd z>s;!)yJmWoSMVM1VpnCHU#*AbE(a2aq7%81yu#B4T#8+#U#**|?a3uu;WrkyuOVh;!|kKo2tl*aKn=y7vwtmbmS`Vr{YJAe^jhuhcn0@Pf#RoGoA2e{nTBK#{b2T O;8zb=NU_(%%KrdTh*41h literal 0 HcmV?d00001 diff --git a/test-redis/bench.c b/test-redis/bench.c new file mode 100644 index 0000000..247134d --- /dev/null +++ b/test-redis/bench.c @@ -0,0 +1,447 @@ +#include +#include +#include +#include +#include +#include +#include + +#include + +typedef enum { + MODE_SET = 0, + MODE_GET = 1, + MODE_MIXED = 2, +} bench_mode_t; + +typedef struct { + const char *host; + int port; + bench_mode_t mode; + uint64_t requests; + uint32_t pipeline; + uint32_t keyspace; + uint32_t value_size; + uint32_t set_ratio; + const char *set_cmd; + const char *get_cmd; + const char *key_prefix; + uint64_t seed; + int verify_get; +} bench_opts_t; + +typedef struct { + uint64_t set_ops; + uint64_t get_ops; + uint64_t errors; + double elapsed_sec; +} bench_result_t; + +static void usage(const char *prog) { + fprintf(stderr, + "Usage: %s [options]\n" + " --host default: 127.0.0.1\n" + " --port default: 6379\n" + " --mode default: mixed\n" + " --requests default: 1000000\n" + " --pipeline default: 64\n" + " --keyspace default: 100000\n" + " --value-size default: 32\n" + " --set-ratio <0..100> default: 50 (mixed mode only)\n" + " --set-cmd default: SET\n" + " --get-cmd default: GET\n" + " --key-prefix default: bench:key:\n" + " --seed default: time-based\n" + " --verify-get verify GET value content\n" + "\nExamples:\n" + " # Benchmark Redis\n" + " %s --host 127.0.0.1 --port 6379 --mode mixed --requests 2000000\n" + "\n" + " # Benchmark kvstore with Redis-compatible commands\n" + " %s --host 127.0.0.1 --port 8888 --mode mixed --requests 2000000\n" + "\n" + " # Benchmark kvstore RBTree path\n" + " %s --host 127.0.0.1 --port 8888 --mode mixed --set-cmd RSET --get-cmd RGET\n", + prog, prog, prog, prog); +} + +static void opts_init(bench_opts_t *o) { + memset(o, 0, sizeof(*o)); + o->host = "127.0.0.1"; + o->port = 6379; + o->mode = MODE_MIXED; + o->requests = 1000000; + o->pipeline = 64; + o->keyspace = 100000; + o->value_size = 32; + o->set_ratio = 50; + o->set_cmd = "SET"; + o->get_cmd = "GET"; + o->key_prefix = "bench:key:"; + o->seed = (uint64_t)time(NULL); + o->verify_get = 0; +} + +static int parse_u64(const char *s, uint64_t *out) { + char *end = NULL; + unsigned long long v; + errno = 0; + v = strtoull(s, &end, 10); + if (errno != 0 || end == s || *end != 0) { + return -1; + } + *out = (uint64_t)v; + return 0; +} + +static int parse_u32(const char *s, uint32_t *out) { + uint64_t v = 0; + if (parse_u64(s, &v) != 0 || v > UINT32_MAX) { + return -1; + } + *out = (uint32_t)v; + return 0; +} + +static int parse_args(int argc, char **argv, bench_opts_t *o) { + int i; + for (i = 1; i < argc; i++) { + if (strcmp(argv[i], "--host") == 0 && i + 1 < argc) { + o->host = argv[++i]; + } else if (strcmp(argv[i], "--port") == 0 && i + 1 < argc) { + uint32_t p = 0; + if (parse_u32(argv[++i], &p) != 0 || p == 0 || p > 65535) { + return -1; + } + o->port = (int)p; + } else if (strcmp(argv[i], "--mode") == 0 && i + 1 < argc) { + const char *m = argv[++i]; + if (strcmp(m, "set") == 0) { + o->mode = MODE_SET; + } else if (strcmp(m, "get") == 0) { + o->mode = MODE_GET; + } else if (strcmp(m, "mixed") == 0) { + o->mode = MODE_MIXED; + } else { + return -1; + } + } else if (strcmp(argv[i], "--requests") == 0 && i + 1 < argc) { + if (parse_u64(argv[++i], &o->requests) != 0 || o->requests == 0) { + return -1; + } + } else if (strcmp(argv[i], "--pipeline") == 0 && i + 1 < argc) { + if (parse_u32(argv[++i], &o->pipeline) != 0 || o->pipeline == 0) { + return -1; + } + } else if (strcmp(argv[i], "--keyspace") == 0 && i + 1 < argc) { + if (parse_u32(argv[++i], &o->keyspace) != 0 || o->keyspace == 0) { + return -1; + } + } else if (strcmp(argv[i], "--value-size") == 0 && i + 1 < argc) { + if (parse_u32(argv[++i], &o->value_size) != 0 || o->value_size == 0) { + return -1; + } + } else if (strcmp(argv[i], "--set-ratio") == 0 && i + 1 < argc) { + if (parse_u32(argv[++i], &o->set_ratio) != 0 || o->set_ratio > 100) { + return -1; + } + } else if (strcmp(argv[i], "--set-cmd") == 0 && i + 1 < argc) { + o->set_cmd = argv[++i]; + } else if (strcmp(argv[i], "--get-cmd") == 0 && i + 1 < argc) { + o->get_cmd = argv[++i]; + } else if (strcmp(argv[i], "--key-prefix") == 0 && i + 1 < argc) { + o->key_prefix = argv[++i]; + } else if (strcmp(argv[i], "--seed") == 0 && i + 1 < argc) { + if (parse_u64(argv[++i], &o->seed) != 0) { + return -1; + } + } else if (strcmp(argv[i], "--verify-get") == 0) { + o->verify_get = 1; + } else if (strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "-h") == 0) { + return 1; + } else { + return -1; + } + } + return 0; +} + +static uint64_t mono_ns(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec; +} + +static uint64_t xorshift64(uint64_t *state) { + uint64_t x = *state; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + *state = x; + return x; +} + +static int append_set(redisContext *c, const char *cmd, + const char *key, size_t key_len, + const char *val, size_t val_len) { + const char *argv[3]; + size_t argvlen[3]; + argv[0] = cmd; + argvlen[0] = strlen(cmd); + argv[1] = key; + argvlen[1] = key_len; + argv[2] = val; + argvlen[2] = val_len; + return redisAppendCommandArgv(c, 3, argv, argvlen); +} + +static int append_get(redisContext *c, const char *cmd, + const char *key, size_t key_len) { + const char *argv[2]; + size_t argvlen[2]; + argv[0] = cmd; + argvlen[0] = strlen(cmd); + argv[1] = key; + argvlen[1] = key_len; + return redisAppendCommandArgv(c, 2, argv, argvlen); +} + +static int consume_set_reply(redisContext *c) { + redisReply *r = NULL; + if (redisGetReply(c, (void **)&r) != REDIS_OK || r == NULL) { + return -1; + } + if (r->type != REDIS_REPLY_STATUS || r->str == NULL || strcasecmp(r->str, "OK") != 0) { + freeReplyObject(r); + return -1; + } + freeReplyObject(r); + return 0; +} + +static int consume_get_reply(redisContext *c, const char *expect, size_t expect_len, int verify) { + redisReply *r = NULL; + if (redisGetReply(c, (void **)&r) != REDIS_OK || r == NULL) { + return -1; + } + + if (r->type == REDIS_REPLY_STRING) { + if (verify && ((size_t)r->len != expect_len || memcmp(r->str, expect, expect_len) != 0)) { + freeReplyObject(r); + return -1; + } + freeReplyObject(r); + return 0; + } + + if (r->type == REDIS_REPLY_NIL) { + freeReplyObject(r); + return verify ? -1 : 0; + } + + freeReplyObject(r); + return -1; +} + +static int prefill(redisContext *c, const bench_opts_t *o, const char *value, size_t value_len) { + uint64_t i = 0; + char key[256]; + + while (i < o->keyspace) { + uint32_t batch = o->pipeline; + if ((uint64_t)batch > (uint64_t)o->keyspace - i) { + batch = (uint32_t)((uint64_t)o->keyspace - i); + } + + for (uint32_t j = 0; j < batch; j++) { + int klen = snprintf(key, sizeof(key), "%s%" PRIu64, o->key_prefix, i + j); + if (klen <= 0 || (size_t)klen >= sizeof(key)) { + return -1; + } + if (append_set(c, o->set_cmd, key, (size_t)klen, value, value_len) != REDIS_OK) { + return -1; + } + } + + for (uint32_t j = 0; j < batch; j++) { + if (consume_set_reply(c) != 0) { + return -1; + } + } + + i += batch; + } + + return 0; +} + +/* + * Keep append/reply operation choices consistent in each batch by building an op mask. + * This wrapper keeps implementation simple and avoids per-op heap allocation. + */ +static int run_bench_with_mask(redisContext *c, const bench_opts_t *o, + const char *value, size_t value_len, + bench_result_t *res) { + uint64_t done = 0; + uint64_t rng = o->seed ? o->seed : 1; + char key[256]; + uint8_t *opmask = NULL; + uint64_t begin_ns; + + memset(res, 0, sizeof(*res)); + opmask = (uint8_t *)malloc(o->pipeline); + if (!opmask) { + return -1; + } + + begin_ns = mono_ns(); + + while (done < o->requests) { + uint32_t batch = o->pipeline; + if ((uint64_t)batch > o->requests - done) { + batch = (uint32_t)(o->requests - done); + } + + for (uint32_t i = 0; i < batch; i++) { + uint64_t rnd = xorshift64(&rng); + uint64_t key_id = rnd % o->keyspace; + int is_set = 0; + int klen; + + if (o->mode == MODE_SET) { + is_set = 1; + } else if (o->mode == MODE_GET) { + is_set = 0; + } else { + is_set = (rnd % 100) < o->set_ratio; + } + opmask[i] = (uint8_t)is_set; + + klen = snprintf(key, sizeof(key), "%s%" PRIu64, o->key_prefix, key_id); + if (klen <= 0 || (size_t)klen >= sizeof(key)) { + free(opmask); + return -1; + } + + if (is_set) { + if (append_set(c, o->set_cmd, key, (size_t)klen, value, value_len) != REDIS_OK) { + free(opmask); + return -1; + } + res->set_ops++; + } else { + if (append_get(c, o->get_cmd, key, (size_t)klen) != REDIS_OK) { + free(opmask); + return -1; + } + res->get_ops++; + } + } + + for (uint32_t i = 0; i < batch; i++) { + int rc = opmask[i] ? consume_set_reply(c) + : consume_get_reply(c, value, value_len, o->verify_get); + if (rc != 0) { + res->errors++; + free(opmask); + return -1; + } + } + + done += batch; + } + + res->elapsed_sec = (double)(mono_ns() - begin_ns) / 1e9; + free(opmask); + return 0; +} + +int main(int argc, char **argv) { + bench_opts_t opts; + bench_result_t result; + redisContext *ctx; + struct timeval timeout; + char *value; + size_t value_len; + int parse_rc; + + opts_init(&opts); + parse_rc = parse_args(argc, argv, &opts); + if (parse_rc == 1) { + usage(argv[0]); + return 0; + } + if (parse_rc != 0) { + usage(argv[0]); + return 2; + } + + timeout.tv_sec = 3; + timeout.tv_usec = 0; + ctx = redisConnectWithTimeout(opts.host, opts.port, timeout); + if (!ctx || ctx->err) { + fprintf(stderr, "connect %s:%d failed: %s\n", opts.host, opts.port, + ctx ? ctx->errstr : "oom"); + if (ctx) { + redisFree(ctx); + } + return 1; + } + + value_len = opts.value_size; + value = (char *)malloc(value_len); + if (!value) { + fprintf(stderr, "malloc value buffer failed\n"); + redisFree(ctx); + return 1; + } + for (size_t i = 0; i < value_len; i++) { + value[i] = (char)('a' + (int)(i % 26)); + } + + if (opts.mode != MODE_SET) { + fprintf(stdout, "[prefill] keyspace=%u using %s\n", opts.keyspace, opts.set_cmd); + if (prefill(ctx, &opts, value, value_len) != 0) { + fprintf(stderr, "prefill failed, err=%s\n", ctx->err ? ctx->errstr : "unknown"); + free(value); + redisFree(ctx); + return 1; + } + } + + fprintf(stdout, + "[bench] target=%s:%d mode=%s requests=%" PRIu64 + " pipeline=%u keyspace=%u value_size=%u set_cmd=%s get_cmd=%s\n", + opts.host, opts.port, + opts.mode == MODE_SET ? "set" : (opts.mode == MODE_GET ? "get" : "mixed"), + opts.requests, opts.pipeline, opts.keyspace, opts.value_size, + opts.set_cmd, opts.get_cmd); + + if (run_bench_with_mask(ctx, &opts, value, value_len, &result) != 0) { + fprintf(stderr, "benchmark failed, err=%s\n", ctx->err ? ctx->errstr : "reply mismatch"); + free(value); + redisFree(ctx); + return 1; + } + + { + double qps = result.elapsed_sec > 0 ? (double)(result.set_ops + result.get_ops) / result.elapsed_sec : 0.0; + double avg_us = (result.set_ops + result.get_ops) > 0 + ? (result.elapsed_sec * 1e6) / (double)(result.set_ops + result.get_ops) + : 0.0; + fprintf(stdout, + "[result] elapsed=%.3fs total=%" PRIu64 " set=%" PRIu64 " get=%" PRIu64 + " errors=%" PRIu64 " qps=%.0f avg=%.2fus/op\n", + result.elapsed_sec, + result.set_ops + result.get_ops, + result.set_ops, + result.get_ops, + result.errors, + qps, + avg_us); + } + + free(value); + redisFree(ctx); + return 0; +}