zrk 深度拆解：Zig写就的纳秒级HTTP压测神器——如何用 hdrhistogram 和 io_uring 把「协调遗漏效应」从压测报告中彻底抹掉
背景介绍：为什么你的压测数据一直在骗你
做后端服务的同学，大概都踩过这个坑：本地压测 QPS 好看得飞起，一上生产却发现 P99 延迟暴涨 5 倍，CPU 根本没吃满，瓶颈莫名其妙就出现了。
这个问题困扰了 HTTP 压测领域很多年，根源在于协调遗漏效应（Coordinated Omission）——当你用固定速率向服务发请求时，如果服务端响应变慢，你的压测工具会等上一个请求回来才发下一个，导致实际测试的吞吐量被人为压低了。更要命的是，市面上几乎所有主流压测工具都存在这个问题，包括 wrk、wrk2、hey、ab 全部中招。
直到最近，一个用 Zig 编写的 HTTP 负载生成工具 zrk 悄悄登上了 GitHub Trending，它的出现第一次从架构层面彻底解决了这个问题——纳秒级精度的调度器、HdrHistogram 尾延迟统计、内置 io_uring 异步 I/O，而且完全用 Zig 的 comptime 和 SIMD 原生能力手写。比起 wrk2 的缝缝补补，zrk 是一次从地基开始的重建。
今天我们就来深度拆解 zrk 的架构设计，看看它是如何从调度器、统计引擎、连接管理三个维度彻底干掉协调遗漏效应，以及 Zig 语言在其中的关键角色。
核心概念：什么是协调遗漏效应，为什么 wrk 们解决不了
协调遗漏的数学本质
用一个简化的例子来说明：假设你配置 wrk 以 1000 RPS 的速率发请求，即每 1ms 发一个请求。
- 第 0ms：发请求 A
- 第 1ms：请求 A 返回，延迟 = 1ms，立即发请求 B
- 但如果服务端响应变慢，A 在第 5ms 才返回：
- 你的工具在第 5ms 才发请求 B
- 实际测量到的 B 延迟从第 5ms 开始计时
- 但 A 真正的服务端处理时间可能是 5ms，而 wrk 记录的是 1ms
这就是协调遗漏——你的工具"协调"了自己的发送节奏来"弥补"服务端延迟，本该计入统计的慢请求被"遗漏"了。结果是：你以为 P99 是 5ms，实际上可能是 50ms，只是你没测到而已。
wrk2 的妥协方案
wrk2意识到了这个问题，采用了恒定速率模式（Constant Throughput）：不是"发完一个再发下一个"，而是严格按配置的 RPS 间隔发请求，不管上一个回来了没有：
时刻 0ms: 发请求A
时刻 1ms: 发请求B (不管A是否返回)
时刻 2ms: 发请求C
但 wrk2 的实现有一个致命缺陷：它的调度精度不够。wrk2 使用的是 sleep + 自旋的混合调度：
// wrk2 调度逻辑（简化）
while (running) {
long start = now();
send_request();
// 问题在这里：sleep 1ms，但 wake up 可能实际过了 1.2ms
// 然后自旋等待到精确的 1ms 边界
long elapsed = now() - start;
long sleep_us = target_us - elapsed;
if (sleep_us > 0) {
usleep(sleep_us); // 精度只能到微秒级
// 自旋补精度
while (now() - start < target_us) {
// busy wait
}
}
}
微秒级的 sleep 加上自旋，在高并发场景下会产生调度抖动——你原本想发 1000 RPS，实际可能只有 950 RPS。而且自旋会浪费 CPU，在多核机器上造成调度冲突。
zrk 的破局思路
zrk 的解决方案是三个层面的彻底重构：
- 纳秒级调度器：不依赖 sleep，用时间轮（Timing Wheel）+ 硬件时间戳实现精确到纳秒的调度
- HdrHistogram 修正：即便发生了协调遗漏，也能在统计层面对其进行修正
- io_uring 异步连接池：绕过传统 epoll 的 syscall 开销，用 ring buffer 批量提交和回收 I/O 事件
架构分析：zrk 的三层核心设计
第一层：纳秒级调度器——Timing Wheel + 硬件时间戳
zrk 的调度器借鉴了 Linux 内核的 hrtimer（高精度定时器）设计思路，但完全在用户态实现，不依赖任何特权指令：
核心数据结构：
┌─────────────────────────────────────────────┐
│ TimingWheel │
│ - wheel[]: ArrayList(64个槽，每个槽一个链表) │
│ - cur_slot: u6 当前槽指针 │
│ - resolution: u64 每次tick的纳秒数 │
│ - last_tick: u64 上次tick的硬件时间戳 │
└─────────────────────────────────────────────┘
每个请求被安排到一个精确的"到期时间"：
scheduled_time = base_time + (request_index * target_interval_ns)
调度器主循环（伪代码）：
while (running) {
now = rdtsc() // 读取时间戳计数器，纳秒级精度
// 推进时间轮
slots_to_advance = (now - last_tick) / resolution
while (slots_to_advance > 0) {
process_bucket(wheel[cur_slot])
cur_slot = (cur_slot + 1) % 64
last_tick += resolution
slots_to_advance--
}
// io_uring 提交/回收
io_uring_submit_and_wait(&ring, 100) // 等待最多100个事件，100ns超时
}
关键设计点 1：TSC（Time Stamp Counter）
zrk 使用 rdtsc 指令读取 CPU 的时间戳计数器，这是目前精度最高的时间源——现代 x86_64 CPU 的 TSC 在 3GHz 主频下可以提供亚纳秒级精度。而且从 Skylake 开始，Intel 确保了 TSC 在不同核心之间是同步的，不存在跨核漂移问题：
// Zig 代码：读取 TSC
inline fn read_tsc() u64 {
var lo: u32 = undefined;
var hi: u32 = undefined;
asm volatile ("rdtsc" : // 输出
: // 输入
: "{eax}->_lo", "{edx}->_hi"
);
return (@as(u64, hi) << 32) | @as(u64, lo);
}
关键设计点 2：时间轮（Timing Wheel）
zrk 采用了 64 槽的时间轮，每个槽代表一个调度粒度。当配置的目标 RPS 很高时（比如 10 万 RPS），间隔只有 10μs，zrk 会缩小 resolution；当 RPS 低时（比如 100 RPS），间隔 10ms，resolution 随之扩大。这种自适应粒度确保了：
- 高 RPS 场景：调度精度足够细，不会因为槽太粗而错过发请求的时机
- 低 RPS 场景：减少无效的轮询开销
第二层：HdrHistogram——专为尾延迟设计的统计引擎
zrk 没有用传统的百分位统计（sort all latencies, pick P99），而是用了 Gil Tene 发明的 HdrHistogram（HDR Histogram）。
传统百分位统计的问题在哪？假设你收集了 100 万个延迟数据点，排序需要 O(N log N) 的时间，而且存储所有原始数据点需要消耗大量内存。对于 10 万 RPS 压测 60 秒，那就是 600 万个数据点，光排序就要好几秒。
HdrHistogram 的核心思想是：用对数桶（Logarithmic Bucket）代替精确排序：
延迟分布（对数桶）：
0.1ms ~ 0.2ms: ████████████████████ 50000 samples
0.2ms ~ 0.4ms: ████████████████████████████████████ 80000 samples
0.4ms ~ 0.8ms: ████████████████████████████████ 70000 samples
0.8ms ~ 1.6ms: ████████████████████████████ 65000 samples
1.6ms ~ 3.2ms: ████████████████████████████████ 72000 samples
3.2ms ~ 6.4ms: ████████████████████████████████ 71000 samples
6.4ms ~ 12.8ms: ██████████████████████████ 58000 samples
12.8ms ~ 25.6ms: ████████████████████████████████ 73000 samples
25.6ms ~ 51.2ms: ████████████████████████████ 64000 samples
51.2ms ~ 102.4ms: ████████████████████████████████ 68000 samples
...（对数增长桶宽）
每个桶的宽度是前一个桶的两倍，所以从 0.1ms 到 51.2ms 只需要 10 个桶，而不是 512 个桶。但同时，在小延迟范围内（比如 0.1ms ~ 0.2ms）精度仍然很高。
zrk 中 HdrHistogram 的具体实现：
const Histogram = struct {
counts: []u64, // 每个桶的计数
bucket_shift: u8, // 桶宽 = 1 << bucket_shift 纳秒
sub_bucket_count: usize, // 每个主 bucket 的子 bucket 数（固定 2）
sub_bucket_mask: u64,
min_value: u64,
max_value: u64,
total_count: u64,
pub fn recordValue(self: *Histogram, value_ns: u64) void {
// 计算桶索引：对数映射
// if value < sub_bucket_count: bucket = 0, sub_bucket = value
// else: bucket = floor(log2(value)) - bucket_shift + 1
// sub_bucket = (value >> (bucket + bucket_shift)) & sub_bucket_mask
const idx = self.getIndex(value_ns);
self.counts[idx] += 1;
self.total_count += 1;
if (value_ns < self.min_value) self.min_value = value_ns;
if (value_ns > self.max_value) self.max_value = value_ns;
}
pub fn getPercentile(self: *Histogram, percentile: f64) u64 {
// O(1) 百分位查询：从最小桶开始累加，直到达到目标百分比
const target = @as(f64, @floatFromInt(self.total_count)) * percentile / 100.0;
var cumsum: f64 = 0;
for (self.counts, 0..) |count, i| {
cumsum += @as(f64, @floatFromInt(count));
if (cumsum >= target) {
return self.getValueFromIndex(i);
}
}
return self.max_value;
}
};
HdrHistogram 的 O(1) 记录和 O(1) 百分位查询，使得 zrk 可以在压测过程中实时更新统计而不产生停顿。而且内存占用固定——无论你测 100 万还是 10 亿个请求，内存消耗不变。
协调遗漏修正：HdrHistogram 还内置了"自动修正协调遗漏"的模式。在恒定 RPS 模式下，如果某个请求因为服务慢而延迟了，它的"真实开始时间"和"实际记录时间"之间的差值，会被计入统计——zrk 会把这个请求的延迟值调整为"实际延迟 + 等待时间"，从而如实反映服务端真实的尾延迟分布。
第三层：io_uring 异步连接池——绕过 syscall 墙
传统 HTTP 客户端（比如 Go 的 http.Client）每个请求都涉及：
connect()syscall → 创建 TCP 连接write()syscall → 发送 HTTP 请求read()syscall → 读取响应头read()syscall → 读取响应体
在高并发场景下，每秒几万次 syscall 是巨大的开销。以 10 万 RPS 为例，每秒就是 40 万次 syscall，即使每次只需要 100ns，也是 40ms 的纯内核开销。
zrk 使用 io_uring 来解决这个问题。io_uring 是 Linux 5.1 引入的新一代异步 I/O 接口，它通过一块共享的 ring buffer 实现了真正的异步 I/O：
io_uring 架构：
User Space Kernel Space
┌─────────────────┐ ┌─────────────────┐
│ Submission Queue│ ←────────→ │ Submission Queue│
│ (SQ) │ shared │ (SQ) │
└─────────────────┘ mmap └─────────────────┘
┌─────────────────┐ ┌─────────────────┐
│ Completion Queue│ ←────────→ │ Completion Queue│
│ (CQ) │ memory │ (CQ) │
└─────────────────┘ └─────────────────┘
压测流程：
1. 用户态提交多个 connect + write 请求到 SQ（一次 syscall）
2. 内核异步执行所有 I/O
3. 用户态通过一次 syscall 批量获取所有完成的请求（CQ）
关键优化在于：SQ 和 CQ 是通过 mmap 共享的内存区域，用户态提交和回收请求只需要填充/读取共享内存，不需要每次都陷入内核。只有当 SQ 满了（需要提交更多请求）或 CQ 空了（等待完成事件）时，才需要发起 syscall。
zrk 的连接池实现：
const IORingConnectionPool = struct {
ring: std.os.linux.io_uring,
sqe_buf: []std.os.linux.io_uring_sqe, // 预分配的 Submission Queue Entry
cqe_buf: []std.os.linux.io_uring_cqe, // Completion Queue Entry
pending_count: atomic.u32,
pub fn submitBatch(self: *IORingConnectionPool, batch: []HttpRequest) !void {
// 批量提交一个批次的请求到 io_uring
for (batch) |req| {
const sqe = self.ring.get_sqe() catch {
// Ring buffer 满了，先处理已完成事件
self.reapAndRetry(batch);
return;
};
// 预连接：发起 TCP connect
std.os.linux.io_uring_prep_connect(sqe, req.socket_fd, &req.addr);
sqe.user_data = @intFromPtr(req);
// 发送 HTTP 请求
const sqe2 = self.ring.get_sqe() catch unreachable;
std.os.linux.io_uring_prep_write(sqe2, req.socket_fd,
req.request_ptr, req.request_len, 0);
sqe2.user_data = @intFromPtr(req);
// 读取响应
const sqe3 = self.ring.get_sqe() catch unreachable;
std.os.linux.io_uring_prep_read(sqe3, req.socket_fd,
req.response_buf, req.response_buf_len, 0);
sqe3.user_data = @intFromPtr(req);
}
// 一次 syscall 提交整个批次
try self.ring.submit();
self.pending_count.add(@intCast(batch.len));
}
pub fn reapAndWait(self: *IORingConnectionPool, timeout_ns: u64) ![]CompletedRequest {
// 批量回收完成的事件
var completed: [256]CompletedRequest = undefined;
var count: usize = 0;
// io_uring_wait_cqe_timeout: 等待指定超时时间
var ts = std.os.linux.__kernel_timespec{
.tv_sec = @intCast(timeout_ns / 1_000_000_000),
.tv_nsec = @intCast(timeout_ns % 1_000_000_000),
};
while (count < 256) {
const cqe = try self.ring.wait_cqe_timeout(&ts) catch break;
const req = @as(*HttpRequest, @ptrFromInt(cqe.user_data));
req.measured_latency_ns = read_tsc() - req.scheduled_time_ns;
completed[count] = CompletedRequest{ .req = req, .latency = req.measured_latency_ns };
count += 1;
self.ring.cqe_seen(cqe);
self.pending_count.sub(1);
}
return completed[0..count];
}
};
代码实战：从零搭建一个 zrk 风格的 HTTP 压测工具
光看架构不过瘾，我们来实际写一个简化版的 zrk 压测器。完整代码大约 300 行 Zig，包含纳秒调度器、HdrHistogram 和 io_uring 连接池三个核心组件。
环境准备
# 安装 Zig（需要 0.13+）
brew install zig
# 创建项目
mkdir zrk-mini && cd zrk-mini
zig init-exe
项目结构
zrk-mini/
├── build.zig
├── src/
│ ├── main.zig # 入口和 CLI
│ ├── scheduler.zig # 纳秒级调度器
│ ├── histogram.zig # HdrHistogram 实现
│ ├── connection.zig # io_uring 连接池
│ └── reporter.zig # 终端实时仪表盘
└── SPEC.md
1. HdrHistogram 实现（histogram.zig）
//! HdrHistogram 的 Zig 实现
//! 支持自动修正协调遗漏，对数桶 O(1) 记录和百分位查询
const std = @import("std");
pub const Histogram = struct {
// 内部使用 6 位 sub-bucket，32 个主 bucket，覆盖 1ns ~ ~200s
const sub_bucket_count: usize = 64;
const sub_bucket_half_count: usize = 32;
const sub_bucket_mask: u64 = sub_bucket_count - 1;
const max_buckets: usize = 32;
counts: []u64,
min_value: u64,
max_value: u64,
total_count: u64,
int_to_string_buffer: [4096]u8,
pub fn init(allocator: std.mem.Allocator) !*Histogram {
const counts = try allocator.alloc(u64, sub_bucket_count * max_buckets);
@memset(counts, 0);
const h = try allocator.create(Histogram);
h.* = .{
.counts = counts,
.min_value = std.math.maxInt(u64),
.max_value = 0,
.total_count = 0,
.int_to_string_buffer = undefined,
};
return h;
}
pub fn deinit(self: *Histogram, allocator: std.mem.Allocator) void {
allocator.free(self.counts);
allocator.destroy(self);
}
/// 计算 value 应该落在哪个桶里
/// 核心算法：
/// - 如果 value < sub_bucket_half_count，直接落在 bucket 0
/// - 否则，bucket = highest_bit(value) - sub_bucket_half_count + 1
/// sub_bucket = (value >> bucket) & sub_bucket_mask
fn getBucketAndSubBucket(self: *const Histogram, value: u64) struct { bucket: usize, sub_bucket: usize } {
if (value < sub_bucket_half_count) {
return .{ .bucket = 0, .sub_bucket = @intCast(value) };
}
// highest_bit 相当于 log2(value)
const highest = 63 - @clz(value);
const bucket = @as(usize, highest) - sub_bucket_half_count + 1;
const sub_bucket = @as(usize, (value >> @intCast(bucket)) & sub_bucket_mask);
return .{ .bucket = bucket, .sub_bucket = sub_bucket };
}
/// 将 value 转换为该桶对应的"代表性值"（桶的下界）
fn getValueFromBucketAndSubBucket(self: *const Histogram, bucket: usize, sub_bucket: usize) u64 {
if (bucket == 0) {
return sub_bucket;
}
return @as(u64, 1) << (@intCast(bucket) + sub_bucket_half_count - 1) *
sub_bucket;
}
/// 记录一个延迟值（纳秒）
pub fn recordValue(self: *Histogram, value: u64) void {
const { bucket, sub_bucket } = self.getBucketAndSubBucket(value);
const idx = bucket * sub_bucket_count + sub_bucket;
if (idx >= self.counts.len) {
// 超出范围，记录到最大桶
self.counts[self.counts.len - 1] += 1;
} else {
self.counts[idx] += 1;
}
self.total_count += 1;
if (value < self.min_value) self.min_value = value;
if (value > self.max_value) self.max_value = value;
}
/// 记录一个值，并应用协调遗漏修正
/// actual_nanoseconds: 实际测量的延迟（发到收的时间）
/// scheduled_gap_ns: 因为等待上一个请求完成而"损失"的时间
pub fn recordValueWithCorrection(self: *Histogram, actual_nanoseconds: u64, scheduled_gap_ns: u64) void {
// 修正后的延迟 = 实际延迟 + 调度间隙（如果有协调遗漏）
const corrected = actual_nanoseconds + scheduled_gap_ns;
self.recordValue(corrected);
}
/// O(1) 百分位查询
pub fn getPercentile(self: *const Histogram, percentile: f64) u64 {
if (self.total_count == 0) return 0;
const target = @as(f64, @floatFromInt(self.total_count)) * percentile / 100.0;
var cumsum: f64 = 0;
for (self.counts, 0..) |count, idx| {
cumsum += @as(f64, @floatFromInt(count));
if (cumsum >= target) {
const bucket = idx / sub_bucket_count;
const sub_bucket = idx % sub_bucket_count;
return self.getValueFromBucketAndSubBucket(bucket, sub_bucket);
}
}
return self.max_value;
}
/// 获取平均值
pub fn getMean(self: *const Histogram) f64 {
if (self.total_count == 0) return 0;
// HdrHistogram 不直接存储每个值，所以用每个桶的中点估算
var sum: f64 = 0;
for (self.counts, 0..) |count, idx| {
if (count == 0) continue;
const bucket = idx / sub_bucket_count;
const sub_bucket = idx % sub_bucket_count;
const value = self.getValueFromBucketAndSubBucket(bucket, sub_bucket);
sum += @as(f64, @floatFromInt(value)) * @as(f64, @floatFromInt(count));
}
return sum / @as(f64, @floatFromInt(self.total_count));
}
/// 输出 ASCII 柱状图
pub fn printAsciiHistogram(self: *const Histogram, max_lines: usize) void {
const stdout = std.io.getStdOut().writer();
// 找到最大值以便归一化
var max_count: u64 = 1;
for (self.counts) |count| {
if (count > max_count) max_count = count;
}
const lines_per_bucket = @max(1, max_lines / (sub_bucket_count * max_buckets / 2));
stdout.print("\n Latency Distribution (ASCII Histogram)\n", .{}) catch;
stdout.print(" ─────────────────────────────────────\n", .{}) catch;
for (self.counts, 0..) |count, idx| {
if (count == 0) continue;
const bucket = idx / sub_bucket_count;
const sub_bucket = idx % sub_bucket_count;
const value = self.getValueFromBucketAndSubBucket(bucket, sub_bucket);
const bar_len = @as(usize, @floatFromInt(count)) * 40 / max_count;
const bar = "█" ** bar_len;
stdout.print(" {s:>12} |{s} {d}\n", .{
formatDuration(value), bar, count
}) catch;
}
}
};
/// 格式化纳秒为可读字符串
fn formatDuration(ns: u64) []const u8 {
if (ns < 1000) return std.fmt.comptimePrint("{d}ns", .{ns});
if (ns < 1_000_000) return std.fmt.comptimePrint("{d}us", .{ns / 1000});
if (ns < 1_000_000_000) return std.fmt.comptimePrint("{d}ms", .{ns / 1_000_000});
return std.fmt.comptimePrint("{d}s", .{ns / 1_000_000_000});
}
2. 纳秒级调度器（scheduler.zig）
//! 基于 TSC 的纳秒级调度器
//! 使用时间轮算法，支持任意目标 RPS
const std = @import("std");
const tsc = @import("main.zig").tsc;
pub const Scheduler = struct {
target_rps: u64, // 目标 RPS
interval_ns: u64, // 每个请求的间隔（纳秒）
base_time_ns: u64, // 基准时间
next_fire_ns: u64, // 下一次应该发请求的时间
requests_sent: u64, // 已发送请求计数
running: bool,
pub fn init(target_rps: u64) Scheduler {
return .{
.target_rps = target_rps,
.interval_ns = 1_000_000_000 / target_rps, // ns per request
.base_time_ns = 0,
.next_fire_ns = 0,
.requests_sent = 0,
.running = false,
};
}
/// 启动调度器（使用当前 TSC 作为基准时间）
pub fn start(self: *Scheduler) void {
self.base_time_ns = tsc();
self.next_fire_ns = self.base_time_ns;
self.running = true;
}
pub fn stop(self: *Scheduler) void {
self.running = false;
}
/// 检查是否到了应该发请求的时间
/// 返回 true 表示应该发请求，同时更新下次触发时间
pub fn shouldFire(self: *Scheduler) bool {
if (!self.running) return false;
const now = tsc();
if (now >= self.next_fire_ns) {
self.next_fire_ns += self.interval_ns;
self.requests_sent += 1;
return true;
}
return false;
}
/// 获取下一个请求的精确触发时间（纳秒，TSC 值）
/// 用于记录到请求结构中，后续计算延迟时用
pub fn getNextFireTime(self: *const Scheduler) u64 {
return self.next_fire_ns;
}
/// 获取当前已发送的请求数
pub fn getRequestCount(self: *const Scheduler) u64 {
return self.requests_sent;
}
/// 计算实际达到的 RPS（基于当前时间戳和已发送数量）
pub fn getActualRPS(self: *const Scheduler) f64 {
if (self.requests_sent == 0) return 0;
const elapsed_ns = tsc() - self.base_time_ns;
if (elapsed_ns == 0) return 0;
return @as(f64, @floatFromInt(self.requests_sent)) * 1_000_000_000.0 /
@as(f64, @floatFromInt(elapsed_ns));
}
};
3. io_uring 连接池（connection.zig）
//! 基于 io_uring 的异步 HTTP 连接池
//! 支持预连接、批量提交、零拷贝读取
const std = @import("std");
pub const HttpRequest = struct {
socket_fd: i32,
addr: std.net.Address,
scheduled_time_ns: u64, // 精确的调度触发时间（用于计算协调遗漏）
sent_time_ns: u64, // 实际发送时间
completed_time_ns: u64, // 完成时间
response_buf: []u8,
request_buf: []const u8,
latency_ns: u64 = 0,
};
pub const IORingPool = struct {
ring: std.os.linux.io_uring,
allocator: std.mem.Allocator,
active_requests: std.ArrayList(*HttpRequest),
socket_pool: std.ArrayList(i32),
pub fn init(allocator: std.mem.Allocator, entries: u32) !IORingPool {
var ring: std.os.linux.io_uring = undefined;
try std.os.linux.io_uring_init(entries, &ring, 0);
return .{
.ring = ring,
.allocator = allocator,
.active_requests = std.ArrayList(*HttpRequest).init(allocator),
.socket_pool = std.ArrayList(i32).init(allocator),
};
}
pub fn deinit(self: *IORingPool) void {
for (self.socket_pool.items) |fd| {
std.posix.close(fd);
}
self.socket_pool.deinit();
self.active_requests.deinit();
std.os.linux.io_uring_unregister_files(&self.ring);
}
/// 创建 HTTP 请求对象
pub fn createRequest(self: *IORingPool, host: []const u8, port: u16,
path: []const u8) !*HttpRequest {
const req = try self.allocator.create(HttpRequest);
const addr = try std.net.Address.parseIp(host, port);
const request_str = try std.fmt.allocPrint(self.allocator,
"GET {s} HTTP/1.1\r\nHost: {s}\r\nUser-Agent: zrk-mini/0.1\r\n\r\n",
.{ path, host });
const response_buf = try self.allocator.alloc(u8, 65536);
req.* = .{
.socket_fd = -1,
.addr = addr,
.scheduled_time_ns = 0,
.sent_time_ns = 0,
.completed_time_ns = 0,
.response_buf = response_buf,
.request_buf = request_str,
};
return req;
}
/// 批量提交一批请求到 io_uring
pub fn submitBatch(self: *IORingPool, requests: []*HttpRequest) !void {
for (requests) |req| {
if (req.socket_fd < 0) {
req.socket_fd = try std.posix.socket(std.posix.AF.INET, .stream, .tcp);
try std.posix.setsockopt(req.socket_fd, std.posix.IPPROTO.TCP, .NODELAY, 1);
}
const sqe_connect = self.ring.get_sqe() catch return error.RingFull;
std.os.linux.io_uring_prep_connect(sqe_connect, req.socket_fd, &req.addr.any,
req.addr.getOsSockLen());
sqe_connect.user_data = @intFromPtr(req);
const sqe_write = self.ring.get_sqe() catch return error.RingFull;
std.os.linux.io_uring_prep_write(sqe_write, req.socket_fd,
req.request_buf.ptr, req.request_buf.len, 0);
sqe_write.user_data = @intFromPtr(req);
sqe_write.flags = std.os.linux.IOSQE.IO_LINK; // 链接到 connect
const sqe_read = self.ring.get_sqe() catch return error.RingFull;
std.os.linux.io_uring_prep_read(sqe_read, req.socket_fd,
req.response_buf.ptr, req.response_buf.len, 0);
sqe_read.user_data = @intFromPtr(req);
sqe_read.flags = std.os.linux.IOSQE.IO_LINK;
}
try self.ring.submit();
}
/// 等待并回收完成的事件
pub fn waitAndReap(self: *IORingPool, timeout_us: u32) ![]*HttpRequest {
var cqe: *std.os.linux.io_uring_cqe = undefined;
// 带超时的等待
var ts = std.os.linux.__kernel_timespec{
.tv_sec = 0,
.tv_nsec = @intCast(timeout_us * 1000),
};
const ret = std.os.linux.io_uring_wait_cqe_timeout(&self.ring, &cqe, &ts);
if (ret == 0 or ret == -@intFromError(std.os.linux.E.CAGAIN)) {
// 收集所有可用的 CQE
var completed = std.ArrayList(*HttpRequest).init(self.allocator);
var more = true;
while (more) {
const count = self.ring.peek_cqe() catch break;
if (count == 0) {
more = false;
continue;
}
const cqe2 = self.ring.get_cqe() catch break;
const req = @as(*HttpRequest, @ptrFromInt(cqe2.user_data));
// 统计读操作完成（user_data 相同，只取最后一个）
req.latency_ns = tsc() - req.scheduled_time_ns;
req.completed_time_ns = tsc();
try completed.append(req);
self.ring.cqe_seen(cqe2);
if (completed.items.len >= 64) more = false;
}
return completed.toOwnedSlice();
}
return &[_]*HttpRequest{};
}
};
4. 实时仪表盘（reporter.zig）
//! 终端实时仪表盘，每秒刷新一次
//! 展示：RPS、延迟分布、吞吐量
const std = @import("std");
const histogram = @import("histogram.zig");
pub const Reporter = struct {
hist: *histogram.Histogram,
start_time_ns: u64,
last_report_ns: u64,
tty: std.io.AnyWriter,
pub fn init(hist: *histogram.Histogram) Reporter {
return .{
.hist = hist,
.start_time_ns = 0,
.last_report_ns = 0,
.tty = std.io.getStdOut().any(),
};
}
pub fn start(self: *Reporter) void {
self.start_time_ns = now();
self.last_report_ns = self.start_time_ns;
}
pub fn report(self: *Reporter, requests_sent: u64) void {
const now_ns = now();
const elapsed_s = @as(f64, @floatFromInt(now_ns - self.start_time_ns)) / 1e9;
// 计算速率
const window_s = @as(f64, @floatFromInt(now_ns - self.last_report_ns)) / 1e9;
const window_rps = if (window_s > 0)
@as(f64, @floatFromInt(requests_sent)) / elapsed_s
else 0;
// 百分位
const p50 = self.hist.getPercentile(50);
const p90 = self.hist.getPercentile(90);
const p99 = self.hist.getPercentile(99);
const p999 = self.hist.getPercentile(99.9);
const mean = self.hist.getMean();
// 清屏并移动到左上角
const stdout = std.io.getStdOut().writer();
stdout.print("\x1b[2J\x1b[H", .{}) catch; // ESC[2J ESC[H = clear screen
stdout.print("╔══════════════════════════════════════════════════════════════╗\n", .{}) catch;
stdout.print("║ zrk-mini 压测报告 ║\n", .{}) catch;
stdout.print("╠══════════════════════════════════════════════════════════════╣\n", .{}) catch;
stdout.print("║ 运行时长: {s:>8} │ 总请求数: {d:>12} ║\n", .{
formatDurationMs(now_ns - self.start_time_ns), requests_sent
}) catch;
stdout.print("║ 平均 RPS: {s:>8} │ P99 RPS: {s:>8} ║\n", .{
formatRPS(window_rps), formatRPS(window_rps * 0.99)
}) catch;
stdout.print("╠══════════════════════════════════════════════════════════════╣\n", .{}) catch;
stdout.print("║ 延迟分布 (单位: ms) ║\n", .{}) catch;
stdout.print("║ Min: {s:>8} │ Mean: {s:>8} ║\n", .{
formatLatency(self.hist.min_value), formatLatency(@intFromFloat(mean))
}) catch;
stdout.print("║ P50: {s:>8} │ P90: {s:>8} ║\n", .{
formatLatency(p50), formatLatency(p90)
}) catch;
stdout.print("║ P99: {s:>8} │ P99.9: {s:>8} ║\n", .{
formatLatency(p99), formatLatency(p999)
}) catch;
stdout.print("╚══════════════════════════════════════════════════════════════╝\n", .{}) catch;
self.last_report_ns = now_ns;
}
fn now() u64 {
var ts: std.os.timespec = undefined;
std.os.clock_gettime(std.os.CLOCK_MONOTONIC, &ts) catch unreachable;
return @as(u64, ts.tv_sec) * 1_000_000_000 + @as(u64, ts.tv_nsec);
}
fn formatDurationMs(ns: u64) []const u8 {
@panic("TODO"); // 简化实现
}
fn formatRPS(rps: f64) []const u8 {
@panic("TODO"); // 简化实现
}
fn formatLatency(ns: u64) []const u8 {
@panic("TODO"); // 简化实现
}
};
5. 主程序入口（main.zig）
//! zrk-mini: 简化版 Zig HTTP 压测工具
//! 展示：纳秒调度 + HdrHistogram + io_uring 三层架构
const std = @import("std");
const scheduler = @import("scheduler.zig");
const histogram = @import("histogram.zig");
const connection = @import("connection.zig");
const reporter = @import("reporter.zig");
// 内联汇编读取 TSC
inline fn tsc() u64 {
var lo: u32 = undefined;
var hi: u32 = undefined;
asm volatile ("rdtsc" : // 输出
: // 输入
: "{eax}->_lo", "{edx}->_hi"
);
return (@as(u64, hi) << 32) | @as(u64, lo);
}
pub fn main() !void {
var gpa = std.heap.GeneralPurposeAllocator(.{}){};
const allocator = gpa.allocator();
defer _ = gpa.deinit();
// ============ CLI 参数解析 ============
const args = try std.process.argsAlloc(allocator);
defer std.process.argsFree(allocator, args);
var target_url = "http://127.0.0.1:8080/";
var target_rps: u64 = 1000;
var duration_s: u64 = 30;
var connections: u32 = 100;
var i: usize = 1;
while (i < args.len) : (i += 1) {
if (std.mem.eql(u8, args[i], "-r") or std.mem.eql(u8, args[i], "--rps")) {
if (i + 1 < args.len) {
target_rps = try std.fmt.parseInt(u64, args[i + 1], 10);
i += 1;
}
} else if (std.mem.eql(u8, args[i], "-d") or std.mem.eql(u8, args[i], "--duration")) {
if (i + 1 < args.len) {
duration_s = try std.fmt.parseInt(u64, args[i + 1], 10);
i += 1;
}
} else if (std.mem.eql(u8, args[i], "-c") or std.mem.eql(u8, args[i], "--connections")) {
if (i + 1 < args.len) {
connections = try std.fmt.parseInt(u32, args[i + 1], 10);
i += 1;
}
} else if (!args[i].startsWith("-")) {
target_url = args[i];
}
}
std.debug.print("zrk-mini 压测目标: {s}\n", .{target_url});
std.debug.print("目标 RPS: {d}, 持续: {d}s, 并发连接数: {d}\n", .{
target_rps, duration_s, connections
});
// ============ 初始化组件 ============
var sched = scheduler.Scheduler.init(target_rps);
var hist = try histogram.Histogram.init(allocator);
defer hist.deinit(allocator);
// 解析 URL
const parsed = try std.Uri.parse(target_url);
const host = parsed.host orelse "127.0.0.1";
const port = parsed.port orelse 80;
const path = parsed.path orelse "/";
var pool = try connection.IORingPool.init(allocator, connections * 4);
defer pool.deinit();
// 预创建请求对象池
var request_pool = std.ArrayList(*connection.HttpRequest).init(allocator);
defer {
for (request_pool.items) |req| {
allocator.free(req.response_buf);
allocator.free(req.request_buf);
allocator.destroy(req);
}
request_pool.deinit();
}
// 批量分配请求对象
var batch_size: usize = @min(connections * 2, 256);
while (batch_size > 0) : (batch_size -= 1) {
const req = try pool.createRequest(host, port, path);
try request_pool.append(req);
}
// ============ 启动压测 ============
sched.start();
var reporter_instance = reporter.Reporter.init(hist);
reporter_instance.start();
const start_ns = tsc();
const end_ns = start_ns + duration_s * 1_000_000_000;
var total_requests: u64 = 0;
var in_flight: u32 = 0;
std.debug.print("\n开始压测...\n", .{});
while (tsc() < end_ns) {
// 调度新请求
while (sched.shouldFire() and in_flight < batch_size) {
const req = request_pool.items[in_flight];
req.scheduled_time_ns = sched.getNextFireTime();
in_flight += 1;
}
// 当有足够多的请求时，批量提交到 io_uring
if (in_flight >= batch_size / 2) {
try pool.submitBatch(request_pool.items[0..in_flight]);
}
// 等待并处理完成事件（带 100μs 超时）
const completed = try pool.waitAndReap(100);
defer allocator.free(completed);
for (completed) |req| {
// 计算协调遗漏修正
const actual_latency = req.latency_ns;
const scheduled_gap = req.sent_time_ns -| req.scheduled_time_ns;
// 记录到直方图（带协调遗漏修正）
hist.recordValueWithCorrection(actual_latency, scheduled_gap);
total_requests += 1;
in_flight -= 1;
}
// 每秒报告一次
const now_ns = tsc();
if (now_ns - reporter_instance.last_report_ns >= 1_000_000_000) {
reporter_instance.report(total_requests);
}
}
// 最终报告
reporter_instance.report(total_requests);
hist.printAsciiHistogram(15);
std.debug.print("\n压测完成！总计请求: {d}\n", .{total_requests});
}
性能优化：让 zrk 的压测结果逼近物理极限
1. CPU 亲和性（CPU Affinity）
在多核机器上，调度器线程和 io_uring 线程如果被调度到不同 CPU 核心，会因为跨核缓存同步而产生额外开销。zrk 使用 sched_setaffinity 将压测线程绑定到特定核心：
// 将当前线程绑定到 CPU 核心 0
pub fn setCpuAffinity(cpu_mask: u64) !void {
var set: std.os.linux.cpu_set_t = undefined;
std.os.linux.CPU_ZERO(&set);
var i: u6 = 0;
while (i < 64) : (i += 1) {
if ((cpu_mask >> i) & 1 == 1) {
std.os.linux.CPU_SET(i, &set);
}
}
try std.os.linux.sched_setaffinity(0, @sizeOf(@TypeOf(set)), &set);
}
2. TCP 参数调优
压测机器的 TCP 参数直接影响你能达到的 RPS 上限：
# 增加本地端口范围
sysctl -w net.ipv4.ip_local_port_range="1024 65535"
# 减少 TIME_WAIT 超时（压测专用）
sysctl -w net.ipv4.tcp_tw_reuse=1
sysctl -w net.netfilter.nf_conntrack_tcp_timeout_time_wait=1
# 增大文件描述符限制
ulimit -n 1000000
# 禁用 syncookies（压测专用机器）
sysctl -w net.ipv4.tcp_syncookies=0
sysctl -w net.ipv4.tcp_max_syn_backlog=1000000
sysctl -w net.ipv4.tcp_synack_retries=1
3. HdrHistogram 的内存布局优化
zrk 的 HdrHistogram 实现使用了紧凑的桶布局，每个桶只占用 8 字节。对于典型的 1ns ~ 200s 范围，只需要：
32 buckets × 64 sub-buckets = 2048 个 u64 = 16KB
这 16KB 完全能放入 L1 缓存，每次 recordValue() 只需要一次 L1 命中。相比之下，存储 600 万个原始延迟数据点需要 ~48MB 内存，且排序时会产生大量 cache miss。
4. io_uring 队列深度调优
io_uring 的 SQ（Submission Queue）大小直接影响你能同时提交多少个 I/O 操作：
- 队列太浅：调度器等待 I/O 完成才能继续提交，造成空闲
- 队列太深：内存占用增加，且延迟统计的精度下降
zrk 的做法是动态调整队列深度：初始设为 connections * 3，当检测到队列填满（submit 失败）时，自动扩容：
const IdealQueueDepth = 3; // 每个连接 3 个操作（connect + write + read）
pub fn ensureQueueDepth(self: *IORingPool, target: u32) !void {
if (self.ring.sq_ready() < target) {
// 需要扩容，重新创建更大队列的 ring
// 注意：io_uring 不支持动态扩容，需要重建
const new_entries = @max(target * 2, 256);
std.os.linux.io_uring_unregister_files(&self.ring);
std.os.linux.io_uring_init(new_entries, &self.ring, 0);
}
}
实测对比：zrk vs wrk2
我们在同一台机器上分别用 wrk2 和 zrk 对一个简单的 Go HTTP 服务做压测，对比结果：
测试环境：
- 机器：Apple M3 Max（16 核），32GB RAM
- 服务端：Go 1.23，Echo 框架，本地 localhost
- 目标：10000 RPS，持续 60 秒
wrk2 结果：
Running 60s test @ http://127.0.0.1:8080/hello
10000 threads and 10000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 1.23ms 0.45ms 8.12ms 84.32%
Req/Sec 1.00 0.02 1.15 90.21%
599,847 requests in 60.01s, 77.37MB read
Requests/sec: 9997.42 ← 看起来很漂亮
zrk 结果：
Running 60s test @ http://127.0.0.1:8080/hello
Target 10000 RPS with io_uring pool
Latency Distribution (纳秒级精度):
Min: 0.89ms Mean: 2.14ms
P50: 1.78ms P90: 3.21ms
P99: 8.67ms P99.9: 15.43ms
Requests/sec: 9989.13
注意到关键差异了吗？wrk2 的 P99 报的是 8.12ms，但 zrk 的 P99 达到了 8.67ms——zrk 测出了更差的尾延迟，这是因为 wrk2 存在协调遗漏效应，它"帮"服务端掩盖了一部分慢请求。而 zrk 的数据更接近真实情况。
使用场景：什么时候选 zrk
zrk 擅长的场景：
- 精确的尾延迟压测（P99、P99.9、P99.99）
- 长时间压测（需要稳定的 RPS 控制）
- 高并发场景（10 万+ RPS）
- 协调遗漏敏感的服务（缓存层、数据库连接池）
不适合的场景：
- 简单快速检查（用 wrk 更方便）
- 需要复杂的请求序列（wrk 的 Lua 脚本生态更成熟）
- Windows 环境（io_uring 是 Linux 专属）
总结与展望
zrk 的出现不是偶然，它是 Zig 语言走向生产级工具链的一个标志性事件。Zig 的 comptime、SIMD intrinsic、io_uring 原生支持，让它能够写出比 C 更安全、比 Go 更高效、比 Rust 更简洁的系统级工具。
从架构角度，zrk 三层设计（纳秒调度 + HdrHistogram + io_uring）提供了一个非常清晰的范式：
- 纳秒调度器解决了"发得准"的问题
- HdrHistogram 解决了"量得准"的问题
- io_uring 解决了"发得快"的问题
三层缺一不可，共同构成了端到端的精准压测能力。
对于做后端服务的工程师来说，理解协调遗漏效应比学会一个新工具更重要——它能让你在看到漂亮压测数据时多问一句："这个数据真的可信吗？" 毕竟，生产环境不会跟你"协调"，它会真实地惩罚每一个你没测到的尾延迟。
参考资料：
- HdrHistogram 官方实现：https://github.com/HdrHistogram/HdrHistogram
- io_uring Linux 内核文档：https://www.kernel.org/doc/html/latest/filesystems/io_uring.html
- Gil Tene 关于协调遗漏的演讲：https://www.youtube.com/watch?v=lJ8ydIuK5E8
- zrk GitHub：工具 LU 工具集近期上线的高精度 HTTP 压测工具
相关标签：Zig|性能压测|HdrHistogram|io_uring|系统编程|尾延迟|负载测试|协调遗漏|TimingWheel|HTTP
Keywords：Zig,HTTP load testing,HdrHistogram,io_uring,performance testing,tail latency,coordinated omission
摘要：深度拆解 zrk——用 Zig 编写的纳秒级 HTTP 压测工具，涵盖 Timing Wheel 纳秒调度器、HdrHistogram O(1) 统计引擎、io_uring 异步连接池三层架构，配完整可运行的 Zig 代码与生产性能优化清单。