diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bb5f394..15be22f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,7 +16,7 @@ jobs: with: version: 0.16.0 - - name: Build examples and benchmark + - name: Build examples and benchmark tools run: zig build - name: Run tests @@ -24,3 +24,8 @@ jobs: - name: Build optimized artifacts run: zig build -Doptimize=ReleaseFast + + - name: Remote benchmark integration + run: | + sudo prlimit --pid $$ --memlock=8388608:8388608 + python3 bench/remote-smoke.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1fc9f70..962e2b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,9 @@ # Changelog ## Unreleased +- Add configurable standalone `bench-server` build target and trusted-network remote benchmark protocol +- Account unique remote NON receipts/loss, isolate run leases, and load DTLS PSKs from bounded files +- Add per-client multi-thread CON windows; fix plain benchmark MID-wrap/duplicate accounting and startup errors - Breaking: CON response APIs take caller-owned buffers; remove result allocation/deinit and sentinel flags - Bound Block2 reassembly; correlate capacity failures and support local request cancellation - Preserve Block2 tokens/handles, sort continuation options, and handle empty ACK/RST and separate replies diff --git a/README.md b/README.md index 200579d..c91a8f5 100644 --- a/README.md +++ b/README.md @@ -922,8 +922,8 @@ Log messages: ## Benchmarks The included bench suite runs a matrix of scenarios grouped by transport -(Plain/DTLS) × type (NON/CON) × threads × payload size. NON throughput -is measured server-side via shared-memory counters; CON measures echo +(Plain/DTLS) × type (NON/CON) × threads × payload size. Embedded NON throughput +uses shared-memory counters; remote NON uses unique receipt accounting. CON measures echo round-trip latency. Results vary by hardware — run on your own system: ```bash @@ -940,6 +940,19 @@ workers synchronize after handshake/warmup and use one common wall interval for successful CON response throughput. See [bounded response storage and matched benchmarks](docs/CON_BUFFERS.md) and [the Zig 0.16 migration measurements](docs/ZIG_016_PERFORMANCE.md). +For two-machine testing, build the dedicated configurable server: + +```sh +zig build bench-server -Doptimize=ReleaseFast +./zig-out/bin/bench-server --bind --threads 8 --allocator smp +``` + +Connect with `--no-server --host `. `--window` and +`--multi-window` configure per-client CON concurrency. The server supports bounded +NON receipt accounting, DTLS credential files, capacity/rate/affinity controls, +and one measurement lease at a time. Plain controls require a trusted network. +See [remote benchmark setup and accounting](docs/REMOTE_BENCHMARKS.md). + ## Requirements - Linux (io_uring support, kernel 5.13+ for multishot recvmsg) diff --git a/bench/client.zig b/bench/client.zig index ce94bd5..cf209ad 100644 --- a/bench/client.zig +++ b/bench/client.zig @@ -8,6 +8,9 @@ const platform = @import("coap").platform; const posix = platform; const coapz = @import("coapz"); const coap = @import("coap"); +const RemoteSession = @import("remote.zig").Session; +const Credentials = @import("credentials.zig").Credentials; +const protocol = @import("protocol.zig"); const bench_psk: coap.Psk = .{ .identity = "bench", @@ -22,6 +25,10 @@ const SuiteConfig = struct { port: u16 = 5683, warmup_count: u32 = 1_000, window_size: u16 = 256, + multi_window_size: u16 = 64, + remote_prefix: ?[4]u8 = null, + psk_file: ?[]const u8 = null, + credentials: ?*const Credentials = null, allocator_kind: enum { debug, smp } = .debug, embedded_server: bool = true, thread_count: u16 = 0, // 0 = nproc @@ -33,6 +40,10 @@ const SuiteConfig = struct { filter_non: bool = true, filter_single: bool = true, filter_multi: bool = true, + fn psk(self: *const SuiteConfig) coap.Psk { + if (self.credentials) |c| return c.psk(); + return bench_psk; + } }; const Scenario = struct { @@ -50,6 +61,8 @@ const ScenarioResult = struct { p99_us: f64, p999_us: f64, errors: u64, + sent: u64 = 0, + received: u64 = 0, }; const ServerGroup = struct { @@ -87,7 +100,10 @@ const scenario_templates = [_]Scenario{ // ── Main ─────────────────────────────────────────────────────────────── pub fn main(init: std.process.Init.Minimal) !void { - const config = parse_args(init.args); + var config = parse_args(init.args); + var credentials: ?Credentials = if (config.psk_file) |path| try Credentials.load(path) else null; + defer if (credentials) |*c| c.clear(); + if (credentials) |*c| config.credentials = c; var gpa = std.heap.DebugAllocator(.{}){}; defer _ = gpa.deinit(); const allocator = switch (config.allocator_kind) { @@ -97,7 +113,7 @@ pub fn main(init: std.process.Init.Minimal) !void { const cpu_count: u16 = if (config.thread_count > 0) config.thread_count else - @intCast(@min(std.Thread.getCpuCount() catch 1, std.math.maxInt(u16))); + @intCast(@min(std.Thread.getCpuCount() catch 1, 256)); var scenarios: [scenario_templates.len]?Scenario = .{null} ** scenario_templates.len; var total: u16 = 0; @@ -137,8 +153,8 @@ pub fn main(init: std.process.Init.Minimal) !void { std.debug.print( " \x1b[2mhost\x1b[0m {s}:{d}\n" ++ " \x1b[2mthreads\x1b[0m {d} \x1b[2m(single: 1, multi: {d})\x1b[0m\n" ++ - " \x1b[2mserver bufs\x1b[0m 512 × 1280B\n" ++ - " \x1b[2mclient window\x1b[0m {d} \x1b[2m(1T)\x1b[0m, 64 \x1b[2m(multi)\x1b[0m\n" ++ + " \x1b[2mserver mode\x1b[0m {s}\n" ++ + " \x1b[2mclient window\x1b[0m {d} \x1b[2m(1T)\x1b[0m, {d} \x1b[2m(multi)\x1b[0m\n" ++ " \x1b[2mwarmup\x1b[0m {s}\n" ++ " \x1b[2mrequests\x1b[0m {s}\n\n", .{ @@ -146,7 +162,9 @@ pub fn main(init: std.process.Init.Minimal) !void { config.port, cpu_count, cpu_count, + if (config.embedded_server) "embedded (512 x 1280B)" else "remote (configured separately)", config.window_size, + config.multi_window_size, fmtInt(&warmup_buf, config.warmup_count), count_str, }, @@ -177,7 +195,7 @@ pub fn main(init: std.process.Init.Minimal) !void { if (need_restart) { kill_server(&server_pid); - const psk: ?coap.Psk = if (s.use_dtls) bench_psk else null; + const psk: ?coap.Psk = if (s.use_dtls) config.psk() else null; server_pid = try fork_server(port, srv_tc, psk, counters, config.bind_address); platform.sleep(150 * std.time.ns_per_ms); current_group = group; @@ -187,16 +205,42 @@ pub fn main(init: std.process.Init.Minimal) !void { // Reset server counters before each scenario. counters.reset(); - const window: u16 = if (s.multi_thread) 64 else config.window_size; + const window: u16 = if (s.multi_thread) config.multi_window_size else config.window_size; + var remote: ?RemoteSession = if (!config.embedded_server) + try RemoteSession.open(allocator, config.host, port, if (s.use_dtls) config.psk() else null, s.request_count, if (s.use_confirmable) .con else .non) + else + null; + defer if (remote) |*session| session.deinit(); + var scenario_config = config; + if (remote) |*session| scenario_config.remote_prefix = session.prefix(); + if (!config.embedded_server and remote == null) + std.debug.print("remote: generic echo server (CON only; no benchmark lease)\n", .{}); const label = format_label(s.label, client_tc); // Overwrite progress line in-place. std.debug.print("\x1b[2K\r \x1b[33m[{d}/{d}]\x1b[0m {s} \x1b[2m...\x1b[0m", .{ scenario_num, total, &label }); const result = if (s.use_dtls) - try run_scenario_dtls(allocator, config, s, client_tc, port, window) + try run_scenario_dtls(allocator, scenario_config, s, client_tc, port, window) else - try run_scenario_plain(allocator, config, s, client_tc, port, window); + try run_scenario_plain(allocator, scenario_config, s, client_tc, port, window); + + if (remote) |*session| { + if (!s.use_confirmable) { + _ = try session.drain(); + const stats = try session.finish(); + if (stats.tracking_failed or stats.expected != s.request_count or stats.received > result.sent) + return error.InvalidBenchmarkAccounting; + const missing = result.sent - stats.received; + std.debug.print("\nREMOTE_BENCHMARK {{\"version\":1,\"transport\":\"{s}\",\"threads\":{d},\"payload\":{d},\"attempted\":{d},\"sent\":{d},\"received\":{d},\"missing\":{d},\"duplicates\":{d},\"local_errors\":{d},\"server_elapsed_ns\":{d}}}\n", .{ + if (s.use_dtls) "dtls" else "plain", client_tc, s.payload_size, s.request_count, result.sent, stats.received, + missing, stats.duplicates, result.errors, stats.elapsed_ns, + }); + results[i] = .{ .rps = if (stats.elapsed_ns > 0) @as(f64, @floatFromInt(stats.received)) * 1e9 / @as(f64, @floatFromInt(stats.elapsed_ns)) else 0, .p50_us = 0, .p99_us = 0, .p999_us = 0, .errors = result.errors, .sent = result.sent, .received = stats.received }; + continue; + } + _ = try session.finish(); + } // For NON scenarios, use server-side throughput measurement. if (!s.use_confirmable) { @@ -252,11 +296,12 @@ fn run_scenario_plain_non( defer allocator.free(payload); @memset(payload, 'x'); + const measured_token = if (config.remote_prefix) |prefix| protocol.token(prefix, 0) else @as([8]u8, @splat(0)); const template = coapz.Packet{ .kind = .non_confirmable, .code = .get, .msg_id = 0, - .token = &.{ 0x00, 0x00 }, + .token = if (config.remote_prefix != null) &measured_token else &.{ 0x00, 0x00 }, .options = &.{}, .payload = payload, .data_buf = &.{}, @@ -268,10 +313,12 @@ fn run_scenario_plain_non( if (config.warmup_count > 0) { const fd = try make_client_socket(config.host, port); defer posix.close(fd); + var warm: [1280]u8 = undefined; + @memcpy(warm[0..template_wire.len], template_wire); + if (config.remote_prefix) |prefix| warm[4] = ~prefix[0]; for (0..config.warmup_count) |_| { - _ = posix.send(fd, template_wire, 0) catch continue; + _ = posix.send(fd, warm[0..template_wire.len], 0) catch continue; } - posix.close(fd); } const extra: u16 = tc -| 1; @@ -281,11 +328,14 @@ fn run_scenario_plain_non( const remainder = count % tc; var send_errors: [256]u64 = .{0} ** 256; + var spawned: usize = 0; + errdefer for (threads[0..spawned]) |thread| thread.join(); for (0..extra) |j| { threads[j] = try std.Thread.spawn(.{}, plain_non_worker, .{ - config.host, port, template_wire, per_thread, &send_errors[j], + config.host, port, template_wire, per_thread, @as(u32, @intCast(j)) * per_thread, &send_errors[j], }); + spawned += 1; } // Main thread. @@ -294,7 +344,8 @@ fn run_scenario_plain_non( var send_buf: [1280]u8 = undefined; @memcpy(send_buf[0..template_wire.len], template_wire); var msg_id: u16 = 0; - for (0..per_thread + remainder) |_| { + for (0..per_thread + remainder) |i| { + if (config.remote_prefix != null) std.mem.writeInt(u32, send_buf[8..12], extra * per_thread + @as(u32, @intCast(i)), .big); send_buf[2] = @intCast(msg_id >> 8); send_buf[3] = @intCast(msg_id & 0xFF); _ = posix.send(fd, send_buf[0..template_wire.len], 0) catch { @@ -307,6 +358,7 @@ fn run_scenario_plain_non( posix.close(fd); for (threads) |t| t.join(); + spawned = 0; // Small drain delay so server processes remaining packets. platform.sleep(50 * std.time.ns_per_ms); @@ -315,16 +367,20 @@ fn run_scenario_plain_non( for (send_errors[0..extra]) |e| total_errors += e; // Server counters provide the actual throughput (filled by caller). - return .{ .rps = 0, .p50_us = 0, .p99_us = 0, .p999_us = 0, .errors = total_errors }; + return .{ .rps = 0, .p50_us = 0, .p99_us = 0, .p999_us = 0, .errors = total_errors, .sent = count - total_errors }; } -fn plain_non_worker(host: []const u8, port: u16, wire: []const u8, count: u32, errors_out: *u64) void { - const fd = make_client_socket(host, port) catch return; +fn plain_non_worker(host: []const u8, port: u16, wire: []const u8, count: u32, base: u32, errors_out: *u64) void { + const fd = make_client_socket(host, port) catch { + errors_out.* = count; + return; + }; var send_buf: [1280]u8 = undefined; @memcpy(send_buf[0..wire.len], wire); var errors: u64 = 0; var msg_id: u16 = 0; - for (0..count) |_| { + for (0..count) |i| { + if (wire[0] & 0x0f == 8) std.mem.writeInt(u32, send_buf[8..12], base + @as(u32, @intCast(i)), .big); send_buf[2] = @intCast(msg_id >> 8); send_buf[3] = @intCast(msg_id & 0xFF); _ = posix.send(fd, send_buf[0..wire.len], 0) catch { @@ -379,6 +435,7 @@ fn run_scenario_plain_con( const remainder = count % tc; const worker_results = try allocator.alloc(PlainWorkerResult, tc); + @memset(worker_results, .{}); defer { for (worker_results) |*wr| { if (wr.result) |*r| { @@ -389,12 +446,15 @@ fn run_scenario_plain_con( } const start = platform.nanoTimestamp(); + var spawned: usize = 0; + errdefer for (threads[0..spawned]) |thread| thread.join(); for (0..extra) |j| { worker_results[j] = .{}; threads[j] = try std.Thread.spawn(.{}, plain_worker, .{ allocator, config.host, port, template_wire, per_thread, window, &worker_results[j], }); + spawned += 1; } const main_fd = try make_client_socket(config.host, port); @@ -403,6 +463,7 @@ fn run_scenario_plain_con( worker_results[extra] = .{ .result = main_result }; for (threads) |t| t.join(); + spawned = 0; const elapsed_ns = platform.nanoTimestamp() - start; return aggregate_plain(allocator, worker_results, elapsed_ns); @@ -410,6 +471,7 @@ fn run_scenario_plain_con( const PlainWorkerResult = struct { result: ?BenchResult = null, + failed_requests: u64 = 0, }; fn plain_worker( @@ -421,22 +483,26 @@ fn plain_worker( window_size: u16, out: *PlainWorkerResult, ) void { + out.failed_requests = count; const fd = make_client_socket(host, port) catch return; defer posix.close(fd); out.result = run_bench(allocator, fd, template_wire, count, window_size, true) catch return; + out.failed_requests = 0; } fn aggregate_plain(allocator: std.mem.Allocator, worker_results: []PlainWorkerResult, elapsed_ns: i128) !ScenarioResult { var total_sent: u64 = 0; + var completed: u64 = 0; var total_errors: u64 = 0; var total_lat_count: u64 = 0; for (worker_results) |wr| { if (wr.result) |r| { total_sent += r.sent; + completed += r.received; total_errors += r.errors; total_lat_count += r.latency_count; - } + } else total_errors += wr.failed_requests; } const merged = if (total_lat_count > 0) @@ -459,7 +525,9 @@ fn aggregate_plain(allocator: std.mem.Allocator, worker_results: []PlainWorkerRe const elapsed_s: f64 = @as(f64, @floatFromInt(elapsed_ns)) / 1e9; return .{ - .rps = if (elapsed_s > 0) @as(f64, @floatFromInt(total_sent)) / elapsed_s else 0, + .rps = if (elapsed_s > 0) @as(f64, @floatFromInt(completed)) / elapsed_s else 0, + .sent = total_sent, + .received = completed, .p50_us = percentile_us(merged, 0.50), .p99_us = percentile_us(merged, 0.99), .p999_us = percentile_us(merged, 0.999), @@ -527,8 +595,9 @@ fn run_scenario_dtls( } for (0..extra) |j| { threads[j] = try std.Thread.spawn(.{}, dtls_worker, .{ - allocator, config.host, port, window, s.payload_size, - s.use_confirmable, per_thread, config.warmup_count, &worker_results[j], &gate, + allocator, config.host, port, window, s.payload_size, + s.use_confirmable, per_thread, config.warmup_count, &worker_results[j], &gate, + config.psk(), @as(u32, @intCast(j)) * per_thread, config.remote_prefix, }); spawned += 1; } @@ -544,9 +613,13 @@ fn run_scenario_dtls( config.warmup_count, &worker_results[extra], &gate, + config.psk(), + extra * per_thread, + config.remote_prefix, ); for (threads) |t| t.join(); + spawned = 0; return aggregate_dtls(allocator, worker_results, gate.start_ns, s.use_confirmable); } @@ -574,6 +647,9 @@ fn dtls_worker( warmup_count: u32, out: *DtlsWorkerResult, gate: *MeasurementGate, + psk: coap.Psk, + base: u32, + prefix: ?[4]u8, ) void { var entered = false; defer if (!entered) { @@ -584,7 +660,7 @@ fn dtls_worker( var client = coap.Client.init(allocator, .{ .host = host, .port = port, - .psk = bench_psk, + .psk = psk, .max_in_flight = window, }) catch return; defer client.deinit(); @@ -610,7 +686,7 @@ fn dtls_worker( if (use_confirmable) { dtls_worker_con(&client, allocator, window, payload, count, out, gate, &entered); } else { - dtls_worker_non(&client, payload, count, out, gate, &entered); + dtls_worker_non(&client, payload, count, out, gate, &entered, base, prefix); } } @@ -699,6 +775,8 @@ fn dtls_worker_non( out: *DtlsWorkerResult, gate: *MeasurementGate, entered: *bool, + base: u32, + prefix: ?[4]u8, ) void { var sent: u64 = 0; var errors: u64 = 0; @@ -706,8 +784,15 @@ fn dtls_worker_non( if (!gate.wait()) return; const bench_start = gate.start_ns; - for (0..count) |_| { - client.cast(.get, &.{}, payload) catch { + for (0..count) |i| { + if (prefix) |tag| { + const id = base + @as(u32, @intCast(i)); + const token = protocol.token(tag, id); + client.sendRaw(.{ .kind = .non_confirmable, .code = .get, .msg_id = @truncate(id), .token = &token, .options = &.{}, .payload = payload, .data_buf = &.{} }) catch { + errors += 1; + continue; + }; + } else client.cast(.get, &.{}, payload) catch { errors += 1; continue; }; @@ -725,8 +810,10 @@ fn aggregate_dtls(allocator: std.mem.Allocator, worker_results: []DtlsWorkerResu var total_errors: u64 = 0; var total_lat_count: u64 = 0; var max_elapsed_ns: i128 = 0; + var successful_sends: u64 = 0; for (worker_results) |wr| { + successful_sends += wr.sent; total_sent += if (confirmable) wr.received else wr.sent; max_elapsed_ns = @max(max_elapsed_ns, wr.finished_ns - start_ns); total_errors += wr.errors; @@ -758,6 +845,8 @@ fn aggregate_dtls(allocator: std.mem.Allocator, worker_results: []DtlsWorkerResu .p99_us = percentile_us(merged, 0.99), .p999_us = percentile_us(merged, 0.999), .errors = total_errors, + .sent = successful_sends, + .received = if (confirmable) total_sent else 0, }; } @@ -805,7 +894,9 @@ fn run_bench( .latency_count = 0, }; - const timestamps = try allocator.alloc(i128, window); + // Index by MID, not MID % window: arbitrary windows otherwise alias at + // the 16-bit wrap. Never reuse an ID while its request is outstanding. + const timestamps = try allocator.alloc(i128, 65536); defer allocator.free(timestamps); @memset(timestamps, 0); @@ -821,16 +912,16 @@ fn run_bench( while (result.received + result.errors < count) { while (in_flight < window and total_sent < count) { - const id_hi: u8 = @intCast(msg_id >> 8); - const id_lo: u8 = @intCast(msg_id & 0xFF); - send_buf[2] = id_hi; - send_buf[3] = id_lo; - send_buf[4] = id_hi; - send_buf[5] = id_lo; - - timestamps[msg_id % window] = platform.nanoTimestamp(); + while (timestamps[msg_id] != 0) msg_id +%= 1; + // Write the selected free MID after skipping occupied IDs. + send_buf[2] = @intCast(msg_id >> 8); + send_buf[3] = @intCast(msg_id & 0xff); + send_buf[4] = send_buf[2]; + send_buf[5] = send_buf[3]; + timestamps[msg_id] = platform.nanoTimestamp(); _ = posix.send(fd, send_buf[0..template.len], 0) catch { + timestamps[msg_id] = 0; result.errors += 1; total_sent += 1; msg_id +%= 1; @@ -852,6 +943,7 @@ fn run_bench( if (consecutive_timeouts >= 3 and in_flight > 0) { result.errors += @intCast(in_flight); in_flight = 0; + @memset(timestamps, 0); if (total_sent >= count) break; } continue; @@ -861,25 +953,26 @@ fn run_bench( continue; }; + if (n < 6 or recv_buf[0] >> 6 != 1 or recv_buf[0] & 0x0f != 2) continue; + const token: u16 = @as(u16, recv_buf[4]) << 8 | recv_buf[5]; + const ts = timestamps[token]; + if (ts == 0) continue; // duplicate or unrelated response + timestamps[token] = 0; consecutive_timeouts = 0; - in_flight -|= 1; + in_flight -= 1; + if (recv_buf[1] != @intFromEnum(coapz.Code.content)) { + result.errors += 1; + continue; + } result.received += 1; result.bytes_received += n; - - if (n >= 6 and (recv_buf[0] & 0x0F) >= 2) { - const token: u16 = @as(u16, recv_buf[4]) << 8 | recv_buf[5]; - const slot = token % window; - const ts = timestamps[slot]; - if (ts > 0) { - const latency = platform.nanoTimestamp() - ts; - result.latency_sum_ns += latency; - if (latency < result.latency_min_ns) result.latency_min_ns = latency; - if (latency > result.latency_max_ns) result.latency_max_ns = latency; - if (collect_latencies and result.latency_count < count) { - result.latencies[result.latency_count] = @intCast(latency); - result.latency_count += 1; - } - } + const latency = platform.nanoTimestamp() - ts; + result.latency_sum_ns += latency; + if (latency < result.latency_min_ns) result.latency_min_ns = latency; + if (latency > result.latency_max_ns) result.latency_max_ns = latency; + if (collect_latencies and result.latency_count < count) { + result.latencies[result.latency_count] = @intCast(latency); + result.latency_count += 1; } } @@ -1229,7 +1322,9 @@ fn print_usage() void { " --port Server port (default: 5683)\n" ++ " --count Override request count per scenario\n" ++ " --warmup Warmup requests (default: 1000)\n" ++ - " --window Client sliding window size (default: 256)\n" ++ + " --window Single-thread CON window, 1..16384 (default: 256)\n" ++ + " --multi-window Per-client multi CON window, 1..16384 (default: 64)\n" ++ + " --psk-file DTLS JSON credentials; otherwise embedded test PSK\n" ++ " --allocator Client allocator: debug (default) or smp\n" ++ " --threads Thread count, 0 = nproc (default: 0)\n" ++ " --no-server Don't fork embedded echo server\n" ++ @@ -1247,6 +1342,11 @@ fn print_usage() void { " -h, --help Show this help\n", .{}); } +fn badArgument(flag: []const u8) noreturn { + std.debug.print("error: missing or invalid value for {s}\n", .{flag}); + std.process.exit(1); +} + fn parse_args(process_args: std.process.Args) SuiteConfig { var config = SuiteConfig{}; var args = process_args.iterate(); @@ -1257,13 +1357,13 @@ fn parse_args(process_args: std.process.Args) SuiteConfig { print_usage(); std.process.exit(0); } else if (std.mem.eql(u8, arg, "--host")) { - config.host = args.next() orelse "127.0.0.1"; + config.host = args.next() orelse badArgument(arg); } else if (std.mem.eql(u8, arg, "--port")) { - const val = args.next() orelse "5683"; - config.port = std.fmt.parseInt(u16, val, 10) catch 5683; + const val = args.next() orelse badArgument(arg); + config.port = std.fmt.parseInt(u16, val, 10) catch badArgument(arg); } else if (std.mem.eql(u8, arg, "--count")) { - const val = args.next() orelse "100000"; - config.count_override = std.fmt.parseInt(u32, val, 10) catch null; + const val = args.next() orelse badArgument(arg); + config.count_override = std.fmt.parseInt(u32, val, 10) catch badArgument(arg); } else if (std.mem.eql(u8, arg, "--payload-only")) { const val = args.next() orelse { std.debug.print("error: --payload-only requires 0, 100, or 1000\n", .{}); @@ -1276,11 +1376,28 @@ fn parse_args(process_args: std.process.Args) SuiteConfig { } config.payload_filter = size; } else if (std.mem.eql(u8, arg, "--warmup")) { - const val = args.next() orelse "1000"; - config.warmup_count = std.fmt.parseInt(u32, val, 10) catch 1_000; + const val = args.next() orelse badArgument(arg); + config.warmup_count = std.fmt.parseInt(u32, val, 10) catch badArgument(arg); } else if (std.mem.eql(u8, arg, "--window")) { - const val = args.next() orelse "256"; - config.window_size = std.fmt.parseInt(u16, val, 10) catch 256; + const val = args.next() orelse badArgument(arg); + config.window_size = std.fmt.parseInt(u16, val, 10) catch { + std.debug.print("invalid --window\n", .{}); + std.process.exit(1); + }; + } else if (std.mem.eql(u8, arg, "--multi-window")) { + const val = args.next() orelse { + std.debug.print("missing --multi-window value\n", .{}); + std.process.exit(1); + }; + config.multi_window_size = std.fmt.parseInt(u16, val, 10) catch { + std.debug.print("invalid --multi-window\n", .{}); + std.process.exit(1); + }; + } else if (std.mem.eql(u8, arg, "--psk-file")) { + config.psk_file = args.next() orelse { + std.debug.print("missing --psk-file value\n", .{}); + std.process.exit(1); + }; } else if (std.mem.eql(u8, arg, "--allocator")) { const val = args.next() orelse ""; config.allocator_kind = std.meta.stringToEnum(@TypeOf(config.allocator_kind), val) orelse { @@ -1288,8 +1405,8 @@ fn parse_args(process_args: std.process.Args) SuiteConfig { std.process.exit(1); }; } else if (std.mem.eql(u8, arg, "--threads")) { - const val = args.next() orelse "0"; - config.thread_count = std.fmt.parseInt(u16, val, 10) catch 0; + const val = args.next() orelse badArgument(arg); + config.thread_count = std.fmt.parseInt(u16, val, 10) catch badArgument(arg); } else if (std.mem.eql(u8, arg, "--no-server")) { config.embedded_server = false; } else if (std.mem.eql(u8, arg, "--ipv6")) { @@ -1320,5 +1437,9 @@ fn parse_args(process_args: std.process.Args) SuiteConfig { } } + if (config.port == 0 or config.window_size == 0 or config.window_size > 16384 or config.multi_window_size == 0 or config.multi_window_size > 16384 or config.thread_count > 256 or (config.count_override != null and config.count_override.? == 0)) { + std.debug.print("invalid port/window/thread/request count\n", .{}); + std.process.exit(1); + } return config; } diff --git a/bench/credentials.zig b/bench/credentials.zig new file mode 100644 index 0000000..f9c38e6 --- /dev/null +++ b/bench/credentials.zig @@ -0,0 +1,69 @@ +const std = @import("std"); +const coap = @import("coap"); +pub const Credentials = struct { + identity: [128]u8 = @splat(0), + identity_len: usize = 0, + key: [64]u8 = @splat(0), + key_len: usize = 0, + pub fn psk(self: *const Credentials) coap.Psk { + return .{ .identity = self.identity[0..self.identity_len], .key = self.key[0..self.key_len] }; + } + pub fn clear(self: *Credentials) void { + std.crypto.secureZero(u8, &self.key); + } + pub fn parse(bytes: []const u8) !Credentials { + if (bytes.len > 2048) return error.CredentialFileTooLarge; + var scratch: [8192]u8 = undefined; + defer std.crypto.secureZero(u8, &scratch); + var fba = std.heap.FixedBufferAllocator.init(&scratch); + const parsed = try std.json.parseFromSlice(struct { identity: []const u8, key_hex: []const u8 }, fba.allocator(), bytes, .{}); + defer parsed.deinit(); + const value = parsed.value; + if (value.identity.len == 0 or value.identity.len > 128 or value.key_hex.len < 32 or value.key_hex.len > 128 or value.key_hex.len % 2 != 0) return error.InvalidCredentials; + var result: Credentials = .{}; + errdefer result.clear(); + @memcpy(result.identity[0..value.identity.len], value.identity); + result.identity_len = value.identity.len; + const key = try std.fmt.hexToBytes(&result.key, value.key_hex); + result.key_len = key.len; + return result; + } + pub fn load(path: []const u8) !Credentials { + var path_buffer: [std.fs.max_path_bytes + 1]u8 = undefined; + const zpath = try std.fmt.bufPrintZ(&path_buffer, "{s}", .{path}); + const fd = try coap.platform.open(zpath, .{ .ACCMODE = .RDONLY, .CLOEXEC = true }, 0); + defer coap.platform.close(fd); + var buffer: [2049]u8 = undefined; + defer std.crypto.secureZero(u8, &buffer); + var length: usize = 0; + while (length < buffer.len) { + const rc = std.os.linux.read(fd, buffer[length..].ptr, buffer.len - length); + switch (std.os.linux.errno(rc)) { + .SUCCESS => {}, + .INTR => continue, + else => return error.CredentialReadFailed, + } + if (rc == 0) break; + length += rc; + } + if (length == buffer.len) return error.CredentialFileTooLarge; + return parse(buffer[0..length]); + } +}; +test "PSK rejects duplicate fields, invalid hex and oversized input; clears owned key" { + const oversized: [2049]u8 = @splat(' '); + try std.testing.expectError(error.CredentialFileTooLarge, Credentials.parse(&oversized)); + try std.testing.expectError(error.DuplicateField, Credentials.parse("{\"identity\":\"a\",\"identity\":\"b\",\"key_hex\":\"30313233343536373839616263646566\"}")); + try std.testing.expectError(error.InvalidCharacter, Credentials.parse("{\"identity\":\"a\",\"key_hex\":\"zz313233343536373839616263646566\"}")); + var c = try Credentials.parse("{\"identity\":\"a\",\"key_hex\":\"30313233343536373839616263646566\"}"); + c.clear(); + try std.testing.expect(std.mem.allEqual(u8, &c.key, 0)); +} + +test "PSK file parsing is bounded and strict" { + var c = try Credentials.parse("{\"identity\":\"bench\",\"key_hex\":\"30313233343536373839616263646566\"}"); + defer c.clear(); + try std.testing.expectEqualStrings("0123456789abcdef", c.psk().key); + try std.testing.expectError(error.InvalidCredentials, Credentials.parse("{\"identity\":\"\",\"key_hex\":\"00\"}")); + try std.testing.expectError(error.UnknownField, Credentials.parse("{\"identity\":\"a\",\"key_hex\":\"00\",\"extra\":1}")); +} diff --git a/bench/measurement.zig b/bench/measurement.zig new file mode 100644 index 0000000..52363ec --- /dev/null +++ b/bench/measurement.zig @@ -0,0 +1,317 @@ +//! Caller-bounded benchmark state. Only initialization allocates. +//! Per-worker locks avoid a globally contended reader counter on the data path. +const std = @import("std"); +const coap = @import("coap"); +const protocol = @import("protocol.zig"); +const Atomic = std.atomic.Value; +const Row = struct { + mutex: std.Io.Mutex align(std.atomic.cache_line) = .init, + unique: u64 = 0, + duplicates: u64 = 0, + first: u64 = 0, + last: u64 = 0, + // Only this row's worker writes replies. Serialization finishes before its + // next callback; other workers/control operations never overwrite replies. + reply: [protocol.stats_len]u8 = undefined, +}; +var next_context_id: Atomic(u64) = .init(1); +threadlocal var cached_context: u64 = 0; +threadlocal var cached_row: usize = 0; + +pub const Context = struct { + allocator: std.mem.Allocator, + io: std.Io, + rows: []Row, + seen: []Atomic(u64), + capacity: u32, + timeout_ns: u64, + context_id: u64, + used: Atomic(usize) = .init(0), + tracking_failed: Atomic(bool) = .init(false), + control: std.Io.Mutex = .init, + active: bool = false, + nonce: [16]u8 = @splat(0), + expected: u32 = 0, + mode: protocol.Mode = .non, + owner: std.Io.net.IpAddress = coap.platform.ip4(.{ 127, 0, 0, 1 }, 0), + started: u64 = 0, + touched: u64 = 0, + + pub fn init(allocator: std.mem.Allocator, capacity: u32, row_count: usize, timeout_ms: u32, io: std.Io) !Context { + if (capacity == 0 or capacity > protocol.max_requests or row_count == 0 or timeout_ms < 1000) return error.InvalidCapacity; + const rows = try allocator.alloc(Row, row_count); + errdefer allocator.free(rows); + for (rows) |*entry| entry.* = .{}; + const seen = try allocator.alloc(Atomic(u64), (@as(usize, capacity) + 63) / 64); + for (seen) |*word| word.* = .init(0); + return .{ .allocator = allocator, .io = io, .rows = rows, .seen = seen, .capacity = capacity, .timeout_ns = @as(u64, timeout_ms) * std.time.ns_per_ms, .context_id = next_context_id.fetchAdd(1, .monotonic) }; + } + pub fn deinit(self: *Context) void { + std.crypto.secureZero(u8, &self.nonce); + self.allocator.free(self.seen); + self.allocator.free(self.rows); + } + fn row(self: *Context) ?*Row { + if (cached_context != self.context_id) { + cached_row = self.used.fetchAdd(1, .monotonic); + cached_context = self.context_id; + } + if (cached_row >= self.rows.len) { + self.tracking_failed.store(true, .release); + return null; + } + return &self.rows[cached_row]; + } + fn freeze(self: *Context) void { + for (self.rows) |*r| r.mutex.lockUncancelable(self.io); + } + fn thaw(self: *Context) void { + var i = self.rows.len; + while (i > 0) { + i -= 1; + self.rows[i].mutex.unlock(self.io); + } + } + fn snapshot(self: *Context, timestamp: u64) protocol.Stats { + var stats: protocol.Stats = .{ .capacity = self.capacity, .expected = self.expected, .active = self.active, .tracking_failed = self.tracking_failed.load(.acquire) }; + var first: u64 = 0; + var last: u64 = 0; + for (self.rows) |r| { + stats.received += r.unique; + stats.duplicates += r.duplicates; + if (r.first != 0 and (first == 0 or r.first < first)) first = r.first; + last = @max(last, r.last); + } + stats.elapsed_ns = if (first == 0) 0 else last -| first; + stats.idle_ns = timestamp -| @max(last, self.started); + return stats; + } + fn expired(self: *Context, timestamp: u64) bool { + var activity = @max(self.started, self.touched); + for (self.rows) |r| activity = @max(activity, r.last); + return timestamp -| activity >= self.timeout_ns; + } + fn sameHost(a: std.Io.net.IpAddress, b: std.Io.net.IpAddress) bool { + var x = a; + var y = b; + switch (x) { + .ip4 => |*v| v.port = 0, + .ip6 => |*v| v.port = 0, + } + switch (y) { + .ip4 => |*v| v.port = 0, + .ip6 => |*v| v.port = 0, + } + return coap.platform.addressEqual(x, y); + } + fn authenticated(self: *Context, peer: std.Io.net.IpAddress, payload: []const u8) bool { + return payload.len == 16 and std.mem.eql(u8, payload, &self.nonce) and coap.platform.addressEqual(peer, self.owner); + } + fn response(r: *Row, stats: protocol.Stats) coap.Response { + r.reply = stats.encode(); + return coap.Response.ok(&r.reply); + } + pub fn handle(self: *Context, request: coap.Request) ?coap.Response { + return self.handleAt(request, @intCast(coap.platform.nanoTimestamp())); + } + fn handleAt(self: *Context, request: coap.Request, timestamp: u64) ?coap.Response { + const r = self.row() orelse return coap.Response.withCode(.service_unavailable); + var path = request.pathSegments(); + if (path.next()) |part| { + if (std.mem.eql(u8, part.value, "__bench")) { + const v = path.next() orelse return coap.Response.notFound(); + const action = path.next() orelse return coap.Response.notFound(); + if (!std.mem.eql(u8, v.value, "v1") or path.next() != null) return coap.Response.notFound(); + if (request.packet.kind != .confirmable) return coap.Response.badRequest(); + self.control.lockUncancelable(self.io); + defer self.control.unlock(self.io); + self.freeze(); + defer self.thaw(); + if (std.mem.eql(u8, action.value, "info")) { + if (request.method() != .get or request.payload().len != 0) return coap.Response.badRequest(); + return response(r, .{ .capacity = self.capacity, .active = self.active }); + } + if (request.method() != .post) return coap.Response.methodNotAllowed(); + if (std.mem.eql(u8, action.value, "start")) { + const start = protocol.Start.decode(request.payload()) catch return coap.Response.badRequest(); + if (start.expected > self.capacity) return coap.Response.withCode(.request_entity_too_large); + if (self.active and !self.expired(timestamp)) { + if (!coap.platform.addressEqual(request.peer_address, self.owner) or !std.mem.eql(u8, &start.nonce, &self.nonce)) return coap.Response.withCode(.service_unavailable); + if (start.expected != self.expected or start.mode != self.mode) return coap.Response.badRequest(); + self.touched = timestamp; + return response(r, self.snapshot(timestamp)); + } + self.active = true; + self.nonce = start.nonce; + self.expected = start.expected; + self.mode = start.mode; + self.owner = request.peer_address; + self.started = timestamp; + self.touched = timestamp; + self.tracking_failed.store(self.used.load(.acquire) > self.rows.len, .release); + for (self.seen) |*word| word.store(0, .monotonic); + for (self.rows) |*entry| { + entry.unique = 0; + entry.duplicates = 0; + entry.first = 0; + entry.last = 0; + } + return response(r, self.snapshot(timestamp)); + } + if (!self.authenticated(request.peer_address, request.payload())) return coap.Response.withCode(.forbidden); + if (self.active and self.expired(timestamp)) { + self.active = false; + return coap.Response.withCode(.request_entity_incomplete); + } + if (std.mem.eql(u8, action.value, "stats")) { + if (!self.active) return coap.Response.withCode(.request_entity_incomplete); + self.touched = timestamp; + return response(r, self.snapshot(timestamp)); + } + if (std.mem.eql(u8, action.value, "finish")) { + self.active = false; + return response(r, self.snapshot(timestamp)); + } + return coap.Response.notFound(); + } + } + r.mutex.lockUncancelable(self.io); + defer r.mutex.unlock(self.io); + if (!self.active or !sameHost(request.peer_address, self.owner)) return coap.Response.ok(request.payload()); + if (self.mode == .con) { + if (request.packet.kind == .confirmable) r.last = timestamp; + } else if (request.packet.kind == .non_confirmable) { + const token = request.packet.token; + if (token.len != 8 or !std.mem.eql(u8, token[0..4], self.nonce[0..4])) return coap.Response.ok(request.payload()); + const id = std.mem.readInt(u32, token[4..8], .big); + if (id >= self.expected) return coap.Response.ok(request.payload()); + const mask = @as(u64, 1) << @as(u6, @truncate(id)); + if (self.seen[id / 64].fetchOr(mask, .monotonic) & mask == 0) r.unique += 1 else r.duplicates += 1; + if (r.first == 0) r.first = timestamp; + r.last = timestamp; + } + return coap.Response.ok(request.payload()); + } +}; + +const coapz = @import("coapz"); +const test_owner = coap.platform.ip4(.{ 127, 0, 0, 1 }, 34000); +fn controlRequest(ctx: *Context, action: []const u8, body: []const u8, peer: std.Io.net.IpAddress, time: u64) coap.Response { + const options = [_]coapz.Option{ .{ .kind = .uri_path, .value = "__bench" }, .{ .kind = .uri_path, .value = "v1" }, .{ .kind = .uri_path, .value = action } }; + return ctx.handleAt(.{ .packet = .{ .kind = .confirmable, .code = .post, .msg_id = 1, .token = "aa", .options = &options, .payload = body, .data_buf = &.{} }, .peer_address = peer, .arena = std.testing.failing_allocator }, time).?; +} +fn beginTest(ctx: *Context, nonce: [16]u8, count: u32, peer: std.Io.net.IpAddress, time: u64) !protocol.Stats { + const start: protocol.Start = .{ .nonce = nonce, .expected = count, .mode = .non }; + const body = start.encode(); + const response = controlRequest(ctx, "start", &body, peer, time); + try std.testing.expectEqual(coapz.Code.content, response.code); + return protocol.Stats.decode(response.payload); +} +fn dataRequest(ctx: *Context, nonce: [16]u8, id: u32, peer: std.Io.net.IpAddress, time: u64) void { + const token = protocol.token(nonce[0..4].*, id); + _ = ctx.handleAt(.{ .packet = .{ .kind = .non_confirmable, .code = .get, .msg_id = @truncate(id), .token = &token, .options = &.{}, .payload = "data", .data_buf = &.{} }, .peer_address = peer, .arena = std.testing.failing_allocator }, time); +} +fn readTest(ctx: *Context, nonce: [16]u8, time: u64) !protocol.Stats { + const response = controlRequest(ctx, "stats", &nonce, test_owner, time); + try std.testing.expectEqual(coapz.Code.content, response.code); + return protocol.Stats.decode(response.payload); +} +test "measurement isolates owner, phase, duplicates and late runs without request allocation" { + var ctx = try Context.init(std.testing.allocator, 100, 4, 1000, std.testing.io); + defer ctx.deinit(); + const nonce: [16]u8 = @splat(7); + _ = try beginTest(&ctx, nonce, 5, test_owner, 100); + dataRequest(&ctx, nonce, 0, test_owner, 200); + dataRequest(&ctx, nonce, 4, coap.platform.ip4(.{ 127, 0, 0, 1 }, 34001), 300); + dataRequest(&ctx, nonce, 4, test_owner, 400); + dataRequest(&ctx, nonce, 5, test_owner, 500); // out of range + dataRequest(&ctx, @splat(8), 1, test_owner, 600); // different run + dataRequest(&ctx, nonce, 2, coap.platform.ip4(.{ 127, 0, 0, 2 }, 34000), 700); // different host + var stats = try readTest(&ctx, nonce, 800); + try std.testing.expectEqual(@as(u64, 2), stats.received); + try std.testing.expectEqual(@as(u64, 1), stats.duplicates); + try std.testing.expectEqual(@as(u64, 200), stats.elapsed_ns); + // Retrying start is idempotent, not a counter reset. + stats = try beginTest(&ctx, nonce, 5, test_owner, 900); + try std.testing.expectEqual(@as(u64, 2), stats.received); + const start: protocol.Start = .{ .nonce = @splat(9), .expected = 5, .mode = .non }; + const body = start.encode(); + try std.testing.expectEqual(coapz.Code.service_unavailable, controlRequest(&ctx, "start", &body, test_owner, 950).code); + try std.testing.expectEqual(coapz.Code.forbidden, controlRequest(&ctx, "stats", &nonce, coap.platform.ip4(.{ 127, 0, 0, 1 }, 34001), 960).code); + try std.testing.expectEqual(coapz.Code.forbidden, controlRequest(&ctx, "stats", &start.nonce, test_owner, 970).code); + var finish = controlRequest(&ctx, "finish", &nonce, test_owner, 1000); + try std.testing.expect(!(try protocol.Stats.decode(finish.payload)).active); + dataRequest(&ctx, nonce, 1, test_owner, 1100); + finish = controlRequest(&ctx, "finish", &nonce, test_owner, 1200); + try std.testing.expectEqual(@as(u64, 2), (try protocol.Stats.decode(finish.payload)).received); + _ = try beginTest(&ctx, @splat(9), 5, test_owner, 1300); + dataRequest(&ctx, nonce, 0, test_owner, 1400); + stats = try readTest(&ctx, @splat(9), 1500); + try std.testing.expectEqual(@as(u64, 0), stats.received); +} +test "measurement lease recovers after crash and valid data refreshes activity" { + var ctx = try Context.init(std.testing.allocator, 8, 4, 1000, std.testing.io); + defer ctx.deinit(); + const nonce: [16]u8 = @splat(1); + _ = try beginTest(&ctx, nonce, 8, test_owner, 1); + dataRequest(&ctx, nonce, 1, test_owner, 900_000_000); + const start: protocol.Start = .{ .nonce = @splat(2), .expected = 8, .mode = .non }; + const body = start.encode(); + const other = coap.platform.ip4(.{ 127, 0, 0, 2 }, 34000); + try std.testing.expectEqual(coapz.Code.service_unavailable, controlRequest(&ctx, "start", &body, other, 1_400_000_000).code); + _ = try beginTest(&ctx, start.nonce, 8, other, 2_100_000_000); + try std.testing.expectEqual(coapz.Code.forbidden, controlRequest(&ctx, "stats", &nonce, test_owner, 2_100_000_001).code); + try std.testing.expectEqual(coapz.Code.request_entity_incomplete, controlRequest(&ctx, "stats", &start.nonce, other, 3_200_000_000).code); +} +test "measurement capacity and control methods reject without changing session" { + var ctx = try Context.init(std.testing.allocator, 8, 2, 1000, std.testing.io); + defer ctx.deinit(); + const start: protocol.Start = .{ .nonce = @splat(1), .expected = 9, .mode = .non }; + const body = start.encode(); + try std.testing.expectEqual(coapz.Code.request_entity_too_large, controlRequest(&ctx, "start", &body, test_owner, 1).code); + try std.testing.expect(!ctx.active); + try std.testing.expectEqual(coapz.Code.bad_request, controlRequest(&ctx, "start", &.{}, test_owner, 2).code); + const options = [_]coapz.Option{ .{ .kind = .uri_path, .value = "__bench" }, .{ .kind = .uri_path, .value = "v1" }, .{ .kind = .uri_path, .value = "start" } }; + const response = ctx.handleAt(.{ .packet = .{ .kind = .non_confirmable, .code = .post, .msg_id = 1, .token = "aa", .options = &options, .payload = &body, .data_buf = &.{} }, .peer_address = test_owner, .arena = std.testing.failing_allocator }, 3).?; + try std.testing.expectEqual(coapz.Code.bad_request, response.code); + try std.testing.expect(!ctx.active); +} +fn concurrentData(ctx: *Context, index: u32) void { + const nonce: [16]u8 = @splat(3); + for (0..500) |i| dataRequest(ctx, nonce, index * 500 + @as(u32, @intCast(i)), test_owner, 1000 + i); + dataRequest(ctx, nonce, 2000, test_owner, 2000); +} +test "measurement concurrent workers deduplicate globally with stable control snapshots" { + var ctx = try Context.init(std.testing.allocator, 2001, 8, 1000, std.testing.io); + defer ctx.deinit(); + const nonce: [16]u8 = @splat(3); + _ = try beginTest(&ctx, nonce, 2001, test_owner, 1); + var threads: [4]std.Thread = undefined; + var spawned: usize = 0; + defer for (threads[0..spawned]) |thread| thread.join(); + for (&threads, 0..) |*thread, i| { + thread.* = try std.Thread.spawn(.{}, concurrentData, .{ &ctx, @as(u32, @intCast(i)) }); + spawned += 1; + } + for (0..20) |_| { + const snapshot = try readTest(&ctx, nonce, 2100); + try std.testing.expect(snapshot.received <= 2001); + } + for (threads) |thread| thread.join(); + spawned = 0; + const stats = try readTest(&ctx, nonce, 2200); + try std.testing.expectEqual(@as(u64, 2001), stats.received); + try std.testing.expectEqual(@as(u64, 3), stats.duplicates); + try std.testing.expect(!stats.tracking_failed); +} +test "measurement exhausted worker storage is explicit instead of hidden delivery loss" { + var ctx = try Context.init(std.testing.allocator, 2001, 1, 1000, std.testing.io); + defer ctx.deinit(); + const nonce: [16]u8 = @splat(3); + _ = try beginTest(&ctx, nonce, 2001, test_owner, 1); + const thread = try std.Thread.spawn(.{}, concurrentData, .{ &ctx, @as(u32, 0) }); + thread.join(); + const stats = try readTest(&ctx, nonce, 2000); + try std.testing.expect(stats.tracking_failed); + try std.testing.expectEqual(@as(u64, 0), stats.received); +} diff --git a/bench/protocol.zig b/bench/protocol.zig new file mode 100644 index 0000000..aaa78ee --- /dev/null +++ b/bench/protocol.zig @@ -0,0 +1,92 @@ +//! Private, versioned benchmark control protocol. No cross-machine clock comparison. +const std = @import("std"); +pub const version: u8 = 1; +pub const nonce_len = 16; +pub const start_len = 21; +pub const stats_len = 48; +pub const max_requests: u32 = 16_000_000; +pub const Mode = enum(u8) { con = 0, non = 1 }; +pub const Start = struct { + nonce: [nonce_len]u8, + expected: u32, + mode: Mode, + pub fn encode(self: Start) [start_len]u8 { + var bytes: [start_len]u8 = undefined; + @memcpy(bytes[0..16], &self.nonce); + std.mem.writeInt(u32, bytes[16..20], self.expected, .big); + bytes[20] = @intFromEnum(self.mode); + return bytes; + } + pub fn decode(bytes: []const u8) !Start { + if (bytes.len != start_len) return error.InvalidStart; + const count = std.mem.readInt(u32, bytes[16..20], .big); + if (count == 0 or count > max_requests or bytes[20] > 1) return error.InvalidStart; + return .{ .nonce = bytes[0..16].*, .expected = count, .mode = @enumFromInt(bytes[20]) }; + } +}; +pub const Stats = struct { + received: u64 = 0, + duplicates: u64 = 0, + elapsed_ns: u64 = 0, + idle_ns: u64 = 0, + expected: u32 = 0, + capacity: u32 = 0, + active: bool = false, + tracking_failed: bool = false, + pub fn encode(self: Stats) [stats_len]u8 { + var bytes: [stats_len]u8 = @splat(0); + @memcpy(bytes[0..4], "CBEN"); + bytes[4] = version; + bytes[5] = @as(u8, @intFromBool(self.active)) | (@as(u8, @intFromBool(self.tracking_failed)) << 1); + std.mem.writeInt(u64, bytes[8..16], self.received, .big); + std.mem.writeInt(u64, bytes[16..24], self.duplicates, .big); + std.mem.writeInt(u64, bytes[24..32], self.elapsed_ns, .big); + std.mem.writeInt(u64, bytes[32..40], self.idle_ns, .big); + std.mem.writeInt(u32, bytes[40..44], self.expected, .big); + std.mem.writeInt(u32, bytes[44..48], self.capacity, .big); + return bytes; + } + pub fn decode(bytes: []const u8) !Stats { + if (bytes.len != stats_len or !std.mem.eql(u8, bytes[0..4], "CBEN")) return error.NotBenchmarkServer; + if (bytes[4] != version) return error.UnsupportedBenchmarkVersion; + if (bytes[5] & ~@as(u8, 3) != 0 or bytes[6] != 0 or bytes[7] != 0) return error.InvalidStats; + return .{ .received = std.mem.readInt(u64, bytes[8..16], .big), .duplicates = std.mem.readInt(u64, bytes[16..24], .big), .elapsed_ns = std.mem.readInt(u64, bytes[24..32], .big), .idle_ns = std.mem.readInt(u64, bytes[32..40], .big), .expected = std.mem.readInt(u32, bytes[40..44], .big), .capacity = std.mem.readInt(u32, bytes[44..48], .big), .active = bytes[5] & 1 != 0, .tracking_failed = bytes[5] & 2 != 0 }; + } +}; +pub fn token(prefix: [4]u8, id: u32) [8]u8 { + var bytes: [8]u8 = undefined; + @memcpy(bytes[0..4], &prefix); + std.mem.writeInt(u32, bytes[4..8], id, .big); + return bytes; +} +test "protocol rejects malformed bounds, flags and reserved bytes" { + const start: Start = .{ .nonce = @splat(0), .expected = 0, .mode = .non }; + var bytes = start.encode(); + try std.testing.expectError(error.InvalidStart, Start.decode(&bytes)); + std.mem.writeInt(u32, bytes[16..20], max_requests + 1, .big); + try std.testing.expectError(error.InvalidStart, Start.decode(&bytes)); + std.mem.writeInt(u32, bytes[16..20], 1, .big); + bytes[20] = 2; + try std.testing.expectError(error.InvalidStart, Start.decode(&bytes)); + var stats = (Stats{}).encode(); + try std.testing.expectError(error.NotBenchmarkServer, Stats.decode(stats[0..47])); + stats[5] = 4; + try std.testing.expectError(error.InvalidStats, Stats.decode(&stats)); + stats[5] = 0; + stats[6] = 1; + try std.testing.expectError(error.InvalidStats, Stats.decode(&stats)); +} + +test "protocol bounded encoding and validation" { + const start: Start = .{ .nonce = @splat(7), .expected = 100000, .mode = .non }; + const bytes = start.encode(); + try std.testing.expectEqualDeep(start, try Start.decode(&bytes)); + try std.testing.expectError(error.InvalidStart, Start.decode(bytes[0..20])); + const stats: Stats = .{ .received = 99999, .expected = 100000, .capacity = max_requests, .elapsed_ns = 42, .active = true }; + var encoded = stats.encode(); + try std.testing.expectEqualDeep(stats, try Stats.decode(&encoded)); + encoded[4] = 2; + try std.testing.expectError(error.UnsupportedBenchmarkVersion, Stats.decode(&encoded)); + const t = token(.{ 1, 2, 3, 4 }, 65536); + try std.testing.expectEqual(@as(u32, 65536), std.mem.readInt(u32, t[4..8], .big)); +} diff --git a/bench/remote-smoke.py b/bench/remote-smoke.py new file mode 100644 index 0000000..acb6590 --- /dev/null +++ b/bench/remote-smoke.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Local two-process integration checks. Requires built bench and bench-server.""" +import json +from pathlib import Path +import re +import secrets +import signal +import socket +import struct +import subprocess +import tempfile +import threading +import time +import sys + +ROOT = Path(__file__).resolve().parents[1] +BIN = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else ROOT / 'zig-out/bin' +ANSI = re.compile(r'\x1b\[[0-9;]*[A-Za-z]') + + +def free_port(family): + with socket.socket(family, socket.SOCK_DGRAM) as s: + s.bind(('::1' if family == socket.AF_INET6 else '127.0.0.1', 0)) + return s.getsockname()[1] + + +def client(port, flags, ok=True): + p = subprocess.run([str(BIN / 'bench'), '--no-server', '--port', str(port), + '--count', '4099', '--warmup', '37', '--payload-only', '100', + '--allocator', 'smp', *flags], capture_output=True, text=True, timeout=25) + output = ANSI.sub('', p.stdout + p.stderr) + assert (p.returncode == 0) == ok, output + return output + + +RPC_MID = 1000 + +def rpc(s, port, action, body=b''): + # CON with two-byte token; monotonically unique MID across these test calls. + global RPC_MID + RPC_MID += 1 + mid = RPC_MID + token = mid.to_bytes(2, 'big') + parts = (b'__bench', b'v1', action.encode()) + options = bytes([0xB0 | len(parts[0])]) + parts[0] + for part in parts[1:]: + options += bytes([len(part)]) + part + wire = b'\x42' + bytes([1 if action == 'info' else 2]) + token + token + options + if body: + wire += b'\xff' + body + s.sendto(wire, ('127.0.0.1', port)) + data, _ = s.recvfrom(2048) + assert data[4:6] == token + return data[1] + + +for flag, value in (('--window', '0'), ('--multi-window', '16385'), ('--threads', '257'), ('--count', 'bad'), ('--port', '0')): + client(1, [flag, value], ok=False) +client(1, ['--window'], ok=False) +print('PASS invalid client arguments are rejected before startup', flush=True) + +with tempfile.TemporaryDirectory(prefix='coap-remote-') as directory: + secret = Path(directory) / 'psk.json' + secret.write_text(json.dumps({'identity': 'custom-smoke-identity', 'key_hex': secrets.token_hex(32)})) + secret.chmod(0o600) + for ipv6 in (False, True): + family = socket.AF_INET6 if ipv6 else socket.AF_INET + host = '::1' if ipv6 else '127.0.0.1' + for secure in (False, True): + port = free_port(family) + server_args = [str(BIN / 'bench-server'), '--bind', host, '--port', str(port), + '--threads', '2', '--max-requests', '20000', '--session-timeout-ms', '1000' if not ipv6 and not secure else '30000'] + if secure: + server_args += ['--psk-file', str(secret)] + server = subprocess.Popen(server_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + time.sleep(.4) + assert server.poll() is None, server.communicate() + for con in (False, True): + for multi in (False, True): + flags = ['--host', host, '--dtls-only' if secure else '--plain-only', + '--con-only' if con else '--non-only', '--multi-only' if multi else '--single-only', + '--threads', '2', '--window', '17', '--multi-window', '19'] + if secure: + flags += ['--psk-file', str(secret)] + output = client(port, flags) + if not con: + rows = [json.loads(line.removeprefix('REMOTE_BENCHMARK ')) for line in output.splitlines() if line.startswith('REMOTE_BENCHMARK ')] + assert len(rows) == 1, output + r = rows[0] + assert r['sent'] == 4099 and r['local_errors'] == 0, r + assert r['received'] + r['missing'] == r['sent'] and r['duplicates'] == 0, r + assert r['threads'] == (2 if multi else 1) and r['payload'] == 100, r + else: + assert re.search(r'\b[12]T\s+100B\s+[^\n]*\s0\s*$', output, re.MULTILINE), output + print(f'PASS {host} {"DTLS" if secure else "plain"} {"CON" if con else "NON"} {"multi" if multi else "single"}', flush=True) + if not ipv6 and not secure: + # Capacity rejection must not fall back to send-rate reporting. + output = client(port, ['--plain-only', '--non-only', '--single-only', '--count', '20001'], ok=False) + assert 'BenchmarkCapacityExceeded' in output + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as control: + control.settimeout(3) + nonce = secrets.token_bytes(16) + start = nonce + struct.pack('!I', 4096) + b'\x01' + assert rpc(control, port, 'start', start) == 69 + output = client(port, ['--plain-only', '--non-only', '--single-only'], ok=False) + assert 'BenchmarkBusy' in output + assert rpc(control, port, 'finish', nonce) == 69 + assert rpc(control, port, 'start', nonce + struct.pack('!I', 4096) + b'\x01') == 69 + # Simulate a disappeared coordinator: no finish, then wait past lease. + time.sleep(1.1) + client(port, ['--plain-only', '--non-only', '--single-only']) + print('PASS capacity, overlap rejection, abandoned-run lease recovery', flush=True) + finally: + server.send_signal(signal.SIGINT if secure else signal.SIGTERM) + try: + out, err = server.communicate(timeout=5) + except subprocess.TimeoutExpired: + server.kill(); out, err = server.communicate() + raise AssertionError('server did not stop on SIGINT/SIGTERM: ' + err) + assert server.returncode == 0, out + err + +# A generic echo server remains usable for CON. Duplicate replies must not +# manufacture completions; an arbitrary window must survive the 16-bit MID wrap. +with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as legacy: + legacy.bind(('127.0.0.1', 0)); legacy.settimeout(.1) + stop = threading.Event() + data_requests = [0] + def echo(): + while not stop.is_set(): + try: + data, peer = legacy.recvfrom(2048) + except (socket.timeout, ConnectionRefusedError): + continue + if len(data) < 6: + continue + length = data[0] & 15 + offset = 4 + length + marker = data.find(b'\xff', offset) + body = data[marker:] if marker >= 0 else b'' + response = bytes([0x60 | length, 69]) + data[2:offset] + body + if marker >= 0: + data_requests[0] += 1 + try: + legacy.sendto(response, peer) + if marker >= 0: + legacy.sendto(response, peer) + except ConnectionRefusedError: + pass + thread = threading.Thread(target=echo) + thread.start() + try: + output = client(legacy.getsockname()[1], ['--plain-only', '--con-only', '--single-only', '--window', '17', '--warmup', '0', '--count', '70000']) + assert 'generic echo server' in output + assert data_requests[0] == 70000, data_requests + assert re.search(r'\b1T\s+100B\s+[^\n]*\s0\s*$', output, re.MULTILINE), output + output = client(legacy.getsockname()[1], ['--plain-only', '--non-only', '--single-only'], ok=False) + assert 'NotBenchmarkServer' in output and 'REMOTE_BENCHMARK ' not in output + finally: + stop.set(); thread.join(timeout=2) + print('PASS legacy CON, duplicate replies, arbitrary window/MID wrap, unsupported NON rejection', flush=True) +print('Remote benchmark integration checks passed.') diff --git a/bench/remote.zig b/bench/remote.zig new file mode 100644 index 0000000..3d4035b --- /dev/null +++ b/bench/remote.zig @@ -0,0 +1,95 @@ +const std = @import("std"); +const coap = @import("coap"); +const coapz = @import("coapz"); +const protocol = @import("protocol.zig"); +pub const Session = struct { + client: coap.Client, + nonce: [16]u8, + expected: u32, + active: bool = true, + + pub fn open(allocator: std.mem.Allocator, host: []const u8, port: u16, psk: ?coap.Psk, count: u32, mode: protocol.Mode) !?Session { + if (count == 0 or count > protocol.max_requests) return error.BenchmarkCapacityExceeded; + var session: Session = .{ .client = try coap.Client.init(allocator, .{ .host = host, .port = port, .psk = psk, .max_in_flight = 1, .handshake_timeout_ms = 5000 }), .nonce = undefined, .expected = count, .active = false }; + errdefer session.deinit(); + if (psk != null) try session.client.handshake(); + const info = session.request(.get, "/__bench/v1/info", &.{}) catch |err| { + // Preserve CON against generic echo servers. NON must never fall + // back to sent/sec when remote receipts are unavailable. + if (mode == .con and err == error.NotBenchmarkServer) { + session.client.deinit(); + return null; + } + return err; + }; + if (info.capacity == 0 or info.capacity > protocol.max_requests) return error.InvalidBenchmarkStats; + if (count > info.capacity) return error.BenchmarkCapacityExceeded; + coap.platform.randomBytes(&session.nonce); + const start: protocol.Start = .{ .nonce = session.nonce, .expected = count, .mode = mode }; + const bytes = start.encode(); + const initial = try session.request(.post, "/__bench/v1/start", &bytes); + session.active = true; + _ = try session.validate(initial); + if (initial.received != 0 or initial.duplicates != 0) return error.InvalidBenchmarkStats; + return session; + } + pub fn prefix(self: *const Session) [4]u8 { + return self.nonce[0..4].*; + } + pub fn deinit(self: *Session) void { + if (self.active) _ = self.finish() catch |err| { + std.debug.print("warning: remote finish failed ({s}); server lease will expire\n", .{@errorName(err)}); + }; + self.client.deinit(); + std.crypto.secureZero(u8, &self.nonce); + } + fn request(self: *Session, code: coapz.Code, path: []const u8, body: []const u8) !protocol.Stats { + return self.requestUntil(code, path, body, coap.platform.nanoTimestamp() + 5 * std.time.ns_per_s); + } + fn requestUntil(self: *Session, code: coapz.Code, path: []const u8, body: []const u8, deadline: i128) !protocol.Stats { + var storage: coap.Client.ResponseStorage(0) = .{}; + var options: [coap.uri.max_options]coapz.Option = undefined; + const handle = try self.client.submit(storage.buffers(), code, coap.uri.fromUri(path, &options), body); + var completed = false; + defer if (!completed) self.client.cancel(handle); + while (coap.platform.nanoTimestamp() < deadline) { + const completion = (try self.client.poll(50)) orelse continue; + if (completion.handle != handle) return error.UnexpectedControlCompletion; + completed = true; + const result = try completion.result; + switch (result.code) { + .content => return protocol.Stats.decode(result.payload), + .not_found, .method_not_allowed => return error.NotBenchmarkServer, + .service_unavailable => return error.BenchmarkBusy, + .request_entity_too_large => return error.BenchmarkCapacityExceeded, + .request_entity_incomplete => return error.BenchmarkSessionExpired, + .forbidden => return error.BenchmarkOwnerMismatch, + else => return error.BenchmarkControlRejected, + } + } + return error.BenchmarkControlTimeout; + } + pub fn stats(self: *Session) !protocol.Stats { + const value = try self.request(.post, "/__bench/v1/stats", &self.nonce); + return self.validate(value); + } + fn validate(self: *const Session, value: protocol.Stats) !protocol.Stats { + if (!value.active or value.expected != self.expected or value.received > self.expected or value.capacity < self.expected or value.capacity > protocol.max_requests or value.tracking_failed) return error.InvalidBenchmarkStats; + return value; + } + pub fn drain(self: *Session) !protocol.Stats { + const deadline = coap.platform.nanoTimestamp() + 5 * std.time.ns_per_s; + while (true) { + const value = try self.validate(try self.requestUntil(.post, "/__bench/v1/stats", &self.nonce, deadline)); + if (value.received == self.expected or value.idle_ns >= 250 * std.time.ns_per_ms) return value; + if (coap.platform.nanoTimestamp() >= deadline) return error.BenchmarkDrainTimeout; + coap.platform.sleep(25 * std.time.ns_per_ms); + } + } + pub fn finish(self: *Session) !protocol.Stats { + const value = try self.request(.post, "/__bench/v1/finish", &self.nonce); + if (!value.active) self.active = false; + if (value.active or value.expected != self.expected or value.received > self.expected or value.capacity < self.expected or value.capacity > protocol.max_requests or value.tracking_failed) return error.InvalidBenchmarkStats; + return value; + } +}; diff --git a/bench/server.zig b/bench/server.zig new file mode 100644 index 0000000..3930091 --- /dev/null +++ b/bench/server.zig @@ -0,0 +1,198 @@ +//! Dedicated trusted-network benchmark server, not a production application. +const std = @import("std"); +const coap = @import("coap"); +const measurement = @import("measurement.zig"); +const Credentials = @import("credentials.zig").Credentials; +const protocol = @import("protocol.zig"); + +pub const Config = struct { + bind: []const u8 = "127.0.0.1", + port: u16 = 5683, + threads: u16 = 8, + buffer_count: u16 = 512, + buffer_size: u32 = 1280, + exchange_count: u16 = 256, + dtls_sessions: u32 = 4096, + capacity: u32 = protocol.max_requests, + timeout_ms: u32 = 30000, + rate_ips: u16 = 0, + rate: u16 = 100, + burst: u16 = 200, + allocator: enum { smp, debug } = .smp, + psk_file: ?[]const u8 = null, + cpus: [256]u16 = undefined, + cpu_count: usize = 0, + help: bool = false, + verbose: bool = false, +}; +const usage = + "Usage: bench-server [options]\n" ++ + " --bind IP Bind address (default: 127.0.0.1; explicit for remote)\n" ++ + " --port N Port (default: 5683; DTLS defaults/auto-switches to 5684)\n" ++ + " --threads N Server workers, 1..256 (default: 8)\n" ++ + " --allocator smp|debug Startup/server allocator (default: smp)\n" ++ + " --buffer-count N Per worker, power of 2, 4..8192 (default: 512; memlock may clamp)\n" ++ + " --buffer-size N Datagram buffer bytes (default: 1280)\n" ++ + " --exchange-count N CON slots/worker, 1..16384 (default: 256)\n" ++ + " --dtls-sessions N DTLS capacity per worker (default: 4096)\n" ++ + " --cpu-affinity A,B,... Optional numeric CPU IDs, assigned cyclically\n" ++ + " --max-requests N Unique NON IDs per run, 1..16000000 (default: 16000000)\n" ++ + " --session-timeout-ms N Inactive run lease, minimum 1000 (default: 30000)\n" ++ + " --rate-limit-ips N Tracked IPs, 0..16384; 0 disables (default: 0)\n" ++ + " --rate-limit N Per-IP tokens/sec if enabled (default: 100)\n" ++ + " --rate-burst N Per-IP burst if enabled (default: 200)\n" ++ + " --psk-file PATH Enable DTLS; JSON {identity,key_hex}, 16..64 key bytes\n" ++ + " --verbose Include routine transport logs (can perturb benchmarks)\n" ++ + " -h, --help Show help\n" ++ + "One measurement run at a time. Plain control is NOT authenticated: use WireGuard/trusted networks.\n"; + +pub fn parse(args: []const []const u8) !Config { + var result: Config = .{}; + var index: usize = 0; + while (index < args.len) : (index += 1) { + const flag = args[index]; + if (std.mem.eql(u8, flag, "--help") or std.mem.eql(u8, flag, "-h")) { + result.help = true; + continue; + } + if (std.mem.eql(u8, flag, "--verbose")) { + result.verbose = true; + continue; + } + index += 1; + if (index == args.len) return error.MissingArgument; + const value = args[index]; + if (std.mem.eql(u8, flag, "--bind")) { + result.bind = value; + } else if (std.mem.eql(u8, flag, "--port")) { + result.port = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--threads")) { + result.threads = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--buffer-count")) { + result.buffer_count = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--buffer-size")) { + result.buffer_size = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--exchange-count")) { + result.exchange_count = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--dtls-sessions")) { + result.dtls_sessions = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--max-requests")) { + result.capacity = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--session-timeout-ms")) { + result.timeout_ms = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--rate-limit-ips")) { + result.rate_ips = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--rate-limit")) { + result.rate = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--rate-burst")) { + result.burst = try std.fmt.parseInt(u16, value, 10); + } else if (std.mem.eql(u8, flag, "--psk-file")) { + result.psk_file = value; + } else if (std.mem.eql(u8, flag, "--allocator")) { + result.allocator = std.meta.stringToEnum(@TypeOf(result.allocator), value) orelse return error.InvalidAllocator; + } else if (std.mem.eql(u8, flag, "--cpu-affinity")) { + if (result.cpu_count != 0) return error.DuplicateAffinity; + var parts = std.mem.splitScalar(u8, value, ','); + while (parts.next()) |part| { + if (part.len == 0 or result.cpu_count == result.cpus.len) return error.InvalidAffinity; + result.cpus[result.cpu_count] = try std.fmt.parseInt(u16, part, 10); + result.cpu_count += 1; + } + } else return error.UnknownArgument; + } + if (result.threads == 0 or result.threads > 256 or result.port == 0 or result.buffer_count < 4 or result.buffer_count > 8192 or !std.math.isPowerOfTwo(result.buffer_count) or result.buffer_size < 64 or result.buffer_size > 65535 or result.exchange_count == 0 or result.exchange_count > 16384 or result.rate_ips > 16384 or result.dtls_sessions == 0 or result.dtls_sessions > 65536 or result.capacity == 0 or result.capacity > protocol.max_requests or result.timeout_ms < 1000 or (result.rate_ips != 0 and (result.rate == 0 or result.burst == 0))) return error.InvalidConfig; + _ = try coap.platform.parseIp(result.bind, result.port); + return result; +} +var verbose_logging = false; +pub const std_options: std.Options = .{ .logFn = benchmarkLog }; +fn benchmarkLog(comptime level: std.log.Level, comptime scope: @EnumLiteral(), comptime format: []const u8, args: anytype) void { + const routine_warning = std.mem.startsWith(u8, format, "multishot recv failed"); + if (level == .err or (level == .warn and !routine_warning) or verbose_logging) + std.log.defaultLog(level, scope, format, args); +} +var active_server: std.atomic.Value(?*coap.Server) = .init(null); +fn terminate(_: std.posix.SIG) callconv(.c) void { + if (active_server.load(.acquire)) |server| server.stop(); +} + +pub fn main(init: std.process.Init.Minimal) !void { + var arguments: [64][]const u8 = undefined; + var iterator = init.args.iterate(); + _ = iterator.next(); + var count: usize = 0; + while (iterator.next()) |arg| { + if (count == arguments.len) return error.TooManyArguments; + arguments[count] = arg; + count += 1; + } + const config = parse(arguments[0..count]) catch |err| { + std.debug.print("bench-server: {s}\n{s}", .{ @errorName(err), usage }); + return err; + }; + if (config.help) { + std.debug.print("{s}", .{usage}); + return; + } + verbose_logging = config.verbose; + var debug: std.heap.DebugAllocator(.{}) = .init; + defer _ = debug.deinit(); + const allocator = switch (config.allocator) { + .smp => std.heap.smp_allocator, + .debug => debug.allocator(), + }; + var credentials: ?Credentials = if (config.psk_file) |path| try Credentials.load(path) else null; + defer if (credentials) |*c| c.clear(); + var io = std.Io.Threaded.init(allocator, .{}); + defer io.deinit(); + var context = try measurement.Context.init(allocator, config.capacity, (@as(usize, config.threads) + 1) * 6, config.timeout_ms, io.io()); + defer context.deinit(); + var server = try coap.Server.initContext(allocator, .{ + .bind_address = config.bind, + .port = config.port, + .thread_count = config.threads, + .buffer_count = config.buffer_count, + .buffer_size = config.buffer_size, + .exchange_count = config.exchange_count, + .dtls_session_count = config.dtls_sessions, + .rate_limit_ip_count = config.rate_ips, + .rate_limit_tokens_per_sec = config.rate, + .rate_limit_burst = config.burst, + .cpu_affinity = if (config.cpu_count > 0) config.cpus[0..config.cpu_count] else null, + .psk = if (credentials) |*c| c.psk() else null, + }, measurement.Context.handle, &context); + defer server.deinit(); + active_server.store(&server, .release); + defer active_server.store(null, .release); + const action: std.posix.Sigaction = .{ .handler = .{ .handler = terminate }, .mask = @splat(0), .flags = 0 }; + var old_int: std.posix.Sigaction = undefined; + var old_term: std.posix.Sigaction = undefined; + std.posix.sigaction(.INT, &action, &old_int); + defer std.posix.sigaction(.INT, &old_int, null); + std.posix.sigaction(.TERM, &action, &old_term); + defer std.posix.sigaction(.TERM, &old_term, null); + std.debug.print("bench-server: bind={s}:{d} workers={d} allocator={s} NON capacity={d} lease={d}ms rate-limit-ips={d} DTLS={}\n", .{ config.bind, server.config.port, config.threads, @tagName(config.allocator), config.capacity, config.timeout_ms, config.rate_ips, credentials != null }); + std.debug.print(" effective buffers={d}x{d} exchanges={d} DTLS-sessions/worker={d} affinity={s}\n", .{ + server.config.buffer_count, server.config.buffer_size, server.config.exchange_count, + server.config.dtls_session_count, if (config.cpu_count == 0) "OS scheduled" else "requested (best-effort)", + }); + if (credentials == null) std.debug.print("Plain benchmark controls are unauthenticated; use a trusted network/WireGuard.\n", .{}); + try server.run(); +} +test "server CLI bounds match ring and u16 hash-table capacities" { + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--buffer-count", "3" })); + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--buffer-count", "600" })); + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--buffer-count", "16384" })); + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--exchange-count", "16385" })); + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--rate-limit-ips", "16385" })); + _ = try parse(&.{ "--buffer-count", "8192", "--exchange-count", "16384", "--rate-limit-ips", "16384" }); +} + +test "server CLI validates bounds and explicit settings" { + const c = try parse(&.{ "--threads", "8", "--bind", "::1", "--cpu-affinity", "0,1,2", "--max-requests", "1000" }); + try std.testing.expectEqual(@as(u16, 8), c.threads); + try std.testing.expectEqual(@as(usize, 3), c.cpu_count); + try std.testing.expectError(error.InvalidConfig, parse(&.{ "--threads", "0" })); + try std.testing.expectError(error.InvalidAffinity, parse(&.{ "--cpu-affinity", "0," })); + try std.testing.expectError(error.MissingArgument, parse(&.{"--port"})); +} diff --git a/bench/tests.zig b/bench/tests.zig new file mode 100644 index 0000000..3ce2702 --- /dev/null +++ b/bench/tests.zig @@ -0,0 +1,7 @@ +test { + _ = @import("protocol.zig"); + _ = @import("credentials.zig"); + _ = @import("measurement.zig"); + _ = @import("server.zig"); + _ = @import("remote.zig"); +} diff --git a/build.zig b/build.zig index 6001cee..251111d 100644 --- a/build.zig +++ b/build.zig @@ -66,6 +66,19 @@ pub fn build(b: *std.Build) void { const bench_step = b.step("bench", "Run benchmark client"); bench_step.dependOn(&run_bench.step); + // Standalone configurable benchmark server (build only, not run). + const server_mod = b.createModule(.{ + .root_source_file = b.path("bench/server.zig"), + .target = target, + .optimize = optimize, + }); + server_mod.addImport("coap", lib_mod); + server_mod.addImport("coapz", coapz_dep.module("coapz")); + const server_exe = b.addExecutable(.{ .name = "bench-server", .root_module = server_mod }); + const install_server = b.addInstallArtifact(server_exe, .{}); + b.getInstallStep().dependOn(&install_server.step); + b.step("bench-server", "Build the standalone benchmark server").dependOn(&install_server.step); + // Tests const lib_tests = b.addTest(.{ .root_module = lib_mod, @@ -73,6 +86,16 @@ pub fn build(b: *std.Build) void { const run_lib_tests = b.addRunArtifact(lib_tests); const test_step = b.step("test", "Run unit tests"); test_step.dependOn(&run_lib_tests.step); + const bench_test_mod = b.createModule(.{ + .root_source_file = b.path("bench/tests.zig"), + .target = target, + .optimize = optimize, + }); + bench_test_mod.addImport("coap", lib_mod); + bench_test_mod.addImport("coapz", coapz_dep.module("coapz")); + const bench_tests = b.addTest(.{ .root_module = bench_test_mod }); + const run_bench_tests = b.addRunArtifact(bench_tests); + test_step.dependOn(&run_bench_tests.step); // Documentation const docs = lib.getEmittedDocs(); diff --git a/docs/REMOTE_BENCHMARKS.md b/docs/REMOTE_BENCHMARKS.md new file mode 100644 index 0000000..3d0bfd2 --- /dev/null +++ b/docs/REMOTE_BENCHMARKS.md @@ -0,0 +1,177 @@ +# Two-machine benchmarks + +Linux and Zig 0.16.0 are required. `bench-server` is a dedicated echo/measurement +server, not an example application server. Allow sufficient locked-memory budget +for its io_uring workers (`ulimit -l`; local verification/CI use 8192 KiB). An +insufficient budget can fail startup with `SystemResources`, even after buffer +count clamping; do not treat such a run as a zero-throughput sample. + +It supports IPv4/IPv6, plain CoAP or +DTLS-PSK, and one active measurement session at a time. Configuration is CLI-only. + +## Plain CoAP over WireGuard + +On the server machine, bind explicitly to its WireGuard address: + +```sh +zig build bench-server -Doptimize=ReleaseFast +./zig-out/bin/bench-server --bind --port 5683 \ + --threads 8 --allocator smp --exchange-count 4096 +``` + +This target **builds/installs only** the server; it does not start it. The default +bind is `127.0.0.1`, default worker count is 8, and rate limiting is disabled by +default. Plain controls are unauthenticated: keep this server on WireGuard or +another trusted, access-controlled network. Do not expose it publicly. + +On the client machine: + +```sh +# Completed CON responses over the client's measurement interval. +zig build bench -Doptimize=ReleaseFast -- \ + --no-server --host --port 5683 \ + --threads 8 --multi-only --multi-window 256 \ + --plain-only --con-only --allocator smp --count 3000000 + +# Unique NON requests received at the remote handler, including explicit loss. +zig build bench -Doptimize=ReleaseFast -- \ + --no-server --host --port 5683 \ + --threads 8 --multi-only --plain-only --non-only \ + --allocator smp --count 3000000 +``` + +These run the existing 0/100/1000-byte payload matrix. Use `--payload-only 1000` +for one payload. `--window` controls single-thread CON concurrency (default 256); +`--multi-window` controls **each** multi-thread client's CON concurrency (default +64). Both accept 1..16384 (bounded by the client's internal hash table). Eight clients with window 256 allow at most 2048 +outstanding requests. Client `--threads` does not configure remote server workers. +NON remains an unpaced fixed-count burst; window settings do not pace it. + +Each client obtains a fresh session before warmup and releases it afterward. +A concurrent benchmark is rejected rather than resetting someone else's counters. +Crashed clients leave an inactivity lease (default 30 seconds); valid measured +traffic and owner/nonce-checked control operations refresh activity. An abandoned idle +lease is reclaimed on the next control operation. Finish is idempotent until a new +run replaces it. Stop the server with SIGINT or SIGTERM. + +## DTLS credentials + +The standalone server requires a credentials file to enable DTLS. Generate one +on a trusted machine and copy it securely to both machines, readable only by the +benchmark user: + +```sh +umask 077 +python3 - <<'PY' > bench-psk.json +import json, secrets +print(json.dumps({"identity": "remote-benchmark", "key_hex": secrets.token_hex(32)})) +PY +``` + +The exact JSON fields are `identity` (1..128 UTF-8 bytes) and `key_hex` (16..64 key +bytes, hexadecimal encoded). Unknown/duplicate fields and malformed credentials +are rejected. The file is bounded to 2048 bytes. Secrets are not accepted as CLI +values or printed; owned parsing/key buffers are cleared after use. + +```sh +# Server machine; one server instance is either plain or DTLS. +./zig-out/bin/bench-server --bind --port 5684 \ + --threads 8 --allocator smp --exchange-count 4096 --psk-file bench-psk.json + +# Client machine; replace --con-only with --non-only for receipt accounting. +zig build bench -Doptimize=ReleaseFast -- \ + --no-server --host --port 5684 \ + --threads 8 --multi-only --multi-window 256 --allocator smp \ + --dtls-only --con-only --psk-file bench-psk.json --count 3000000 +``` + +The server/library switches port 5683 to 5684 when PSK is enabled; specify 5684 or +a different custom port explicitly. In the client, `--psk-file` changes DTLS +credentials but does **not** select the transport: use `--dtls-only`. Without a +file the client retains the public embedded-test PSK; it is not a deployment key. +Control and measured traffic use the same transport/PSK. The DTLS control client +also consumes a server session slot. For IPv6 use literal addresses (`::1` for +loopback); do not put brackets around the CLI address. + +## Server controls + +`./zig-out/bin/bench-server --help` lists defaults and bounds. Available settings: + +- `--bind`, `--port`, `--threads` (1..256), `--allocator smp|debug`. +- `--buffer-count`, `--buffer-size`: application datagram buffers **per worker**, + not UDP socket queue byte sizes. Buffer count must be a power of two, 4..8192. + The library may clamp it to memlock availability; startup reports effective configuration. +- `--exchange-count`: per-worker CON response cache/exchange capacity, 1..16384. +- `--dtls-sessions`: per-worker DTLS session capacity (startup allocation). +- `--cpu-affinity A,B,...`: optional best-effort CPU placement, cyclic across + workers. Defaults to OS scheduling; invalid/unavailable affinity is warned. +- `--max-requests`: receipt bitmap/run-count capacity, 1..16000000 (default + 16000000). Both CON and NON session counts must fit this bound. +- `--session-timeout-ms`: inactive lease, minimum 1000 ms (default 30000). +- `--rate-limit-ips`: tracked IP capacity, 0..16384; 0 disables (default 0). + `--rate-limit` and `--rate-burst` select per-IP tokens/sec and burst when enabled. +- `--psk-file`: enable DTLS with file credentials. +- `--verbose`: include routine transport logs, which can perturb measurements. + Fatal/configuration warnings remain visible without it; repetitive receive-error + logging from unconsumed NON replies is suppressed by default. + +Receipt storage is allocated once at startup: a bounded atomic bitset (2 MB at +16 million IDs), per-worker counters, and control reply buffers. No measurement +handler allocation or growable per-run container is used. A fixed worker-slot +budget includes library worker restarts; exhaustion is reported as tracking +failure, never disguised as network loss. These guarantees describe measurement +storage, not a new allocation guarantee for the underlying server or raw APIs. + +## Reading NON results + +Each remote NON case emits one `REMOTE_BENCHMARK {...}` JSON line (stderr, like +the existing benchmark output), containing transport, client threads, payload, +attempted requests, successful local sends, unique handler receipts, missing, +duplicates, local errors, and `server_elapsed_ns`. + +- `missing = sent - received`; local errors are separate, not delivery failures. +- Warmup and control requests are excluded. IDs are globally unique within the + run, not per socket. A run-prefix/ID token and coordinator IP exclude unrelated + traffic; per-worker counters and a shared atomic bitset suppress duplicates. +- After sending, the client polls until all expected requests arrive or receipt + activity is quiet for 250 ms, with a 5-second drain deadline. Finish freezes the + final snapshot. Missing means **not received before this finite cutoff**, not + an infinite-time loss proof or localization of the drop layer. +- NON table `req/s` is unique handler receipts divided by the **server-local + first-to-last matching receipt interval**, including duplicate arrival timing. + It is not sender send rate, a synchronized cross-machine wall interval, or a + loss-free sustainable capacity. Zero-length receive intervals report zero rate. +- NON replies are echoed but not drained/validated by the data clients. These are + request-delivery results, not reply-delivery results. UDP error counters can + include those unconsumed replies. + +Remote NON refuses unsupported servers, malformed accounting, capacity overflow, +tracking failure, overlapping sessions, or unsuccessful drain/control operations; +it never falls back to claiming local send rate as delivery. CON can fall back to +a generic echo server that reports unsupported control, with an explicit warning +that it has no benchmark lease. The plain CON generator validates correlated +content responses and ignores duplicate/unrelated replies; it is still the +existing raw UDP generator, not the public CON retransmission engine. DTLS CON +uses the public client. Plain timing includes worker startup/warmup; DTLS uses the +existing synchronized post-warmup gate. Use long counts and record these shapes. + +There is no timed-run controller, pacing/search for an acceptable loss threshold, +remote host automation, or WireGuard setup. Record build mode, both allocators, +worker/window counts, RTT, MTU, tunnel bandwidth, CPU utilization and loss. A +1 Gbit/s tunnel may bottleneck before the server, especially with 1000-byte +payloads. Concurrent requests needed to fill it depend on RTT, not just cores. + +## Verification + +```sh +zig build test +zig build test -Doptimize=ReleaseSafe +zig build -Doptimize=ReleaseFast +python3 bench/remote-smoke.py +``` + +The smoke script uses local separate server/client processes, ephemeral custom +PSKs, both IP families/transports, single/multi clients, lease/capacity/overlap +checks, and a generic echo peer with duplicate replies across the 16-bit MID wrap. +It is functional verification, **not** a two-machine or WireGuard performance +measurement. Pass an alternate binary directory as its first argument if needed.