//! Daemon latency probe (`zig build bench -- daemon [--files N] [--rounds N]`).
//!
//! Seeds a sub-manifest k=7 store with `--files` files (default 10,000)
//! through the daemon itself, then times, request written to reply read over
//! the real socket:
//!
//! * `feed` with the current cursor on an unchanged store: the nothing-new
//! answer a client polls with, targeted at 10 ms;
//! * a one-file commit through the resident daemon, targeted at 30 ms;
//! * `status`, for scale.
//!
//! Each is reported as median and p99 over `--rounds` requests (default 200).
//! The targets are printed beside the numbers, not asserted: a loaded machine
//! is not a regression.
//!
//! The store lives in a fresh directory under `.zig-cache/tmp/` and is
//! removed afterwards; the daemon is the ReleaseFast `zit` the bench step
//! builds, so the numbers are what a release binary does.
const std = @import("std");
const Io = std.Io;
const net = std.Io.net;
const build_options = @import("build_options");
const protocol = @import("zit").daemon_protocol;
pub const Error = error{
OutOfMemory,
BadArgs,
/// The store could not be prepared, or the daemon refused a request.
Setup,
Spawn,
DaemonNeverListened,
} || Io.Writer.Error;
const feed_target_ms: f64 = 10;
const commit_target_ms: f64 = 30;
const max_files: usize = 200_000;
const max_rounds: usize = 100_000;
const Config = struct {
files: usize = 10_000,
rounds: usize = 200,
};
fn parseArgs(args: []const []const u8) Error!Config {
var cfg: Config = .{};
var i: usize = 2;
while (i < args.len) : (i += 2) {
if (i + 1 >= args.len) return error.BadArgs;
const v = std.fmt.parseInt(usize, args[i + 1], 10) catch return error.BadArgs;
if (std.mem.eql(u8, args[i], "--files")) {
if (v == 0 or v > max_files) return error.BadArgs;
cfg.files = v;
} else if (std.mem.eql(u8, args[i], "--rounds")) {
if (v == 0 or v > max_rounds) return error.BadArgs;
cfg.rounds = v;
} else return error.BadArgs;
}
return cfg;
}
pub fn run(gpa: std.mem.Allocator, io: Io, args: []const []const u8, out: *Io.Writer) Error!void {
const cfg = try parseArgs(args);
var arena_state = std.heap.ArenaAllocator.init(gpa);
defer arena_state.deinit();
const a = arena_state.allocator();
var nonce: u64 = undefined;
io.random(std.mem.asBytes(&nonce));
const dir_rel = try std.fmt.allocPrint(a, ".zig-cache/tmp/zit-daemon-bench-{x}", .{nonce});
const cwd = Io.Dir.cwd();
cwd.createDirPath(io, dir_rel) catch return error.Setup;
defer cwd.deleteTree(io, dir_rel) catch {};
var dir = cwd.openDir(io, dir_rel, .{}) catch return error.Setup;
defer dir.close(io);
const exe = try absExe(a, io);
try runOnce(a, io, dir, &.{ exe, "init", "-q", "--submanifest", "7", "store" });
var child = try spawnDaemon(a, io, dir, exe);
defer child.kill(io);
const socket = try std.fmt.allocPrint(a, "{s}/s/d.sock", .{dir_rel});
var c = try Client.open(a, io, socket);
defer c.stream.close(io);
try seed(a, io, &c, cfg.files);
const st = try c.call(a, .{ .id = 1, .op = .status }, &.{});
const cursor = st.object.get("cursor").?.string;
const feeds = try a.alloc(u64, cfg.rounds);
for (feeds, 0..) |*t, i| {
const start = Io.Timestamp.now(io, .awake);
const v = try c.call(a, .{ .id = @intCast(i), .op = .{ .feed = cursor } }, &.{});
t.* = elapsedNs(io, start);
if (v.object.get("changes").?.array.items.len != 0) return error.Setup;
}
const statuses = try a.alloc(u64, cfg.rounds);
for (statuses, 0..) |*t, i| {
const start = Io.Timestamp.now(io, .awake);
_ = try c.call(a, .{ .id = @intCast(i), .op = .status }, &.{});
t.* = elapsedNs(io, start);
}
const commits = try a.alloc(u64, cfg.rounds);
for (commits, 0..) |*t, i| {
const name = try fileName(a, i * 7919 % cfg.files);
const bytes = try std.fmt.allocPrint(a, "{s} revision {d}\n", .{ name, i });
const req: protocol.Request = .{ .id = @intCast(i), .op = .{ .commit = .{
.user = "bench",
.comment = "one file",
.time = .{ .epoch = 1_800_000_000 + @as(i64, @intCast(i)) + 1, .ms = 0 },
.edits = &.{.{ .put = .{ .name = name, .size = bytes.len } }},
} } };
const start = Io.Timestamp.now(io, .awake);
const v = try c.call(a, req, &.{bytes});
t.* = elapsedNs(io, start);
if (!std.mem.eql(u8, v.object.get("outcome").?.string, "recorded")) return error.Setup;
}
try out.print("daemon latency, k=7 store of {d} files, {d} rounds each (request to reply over the socket)\n", .{ cfg.files, cfg.rounds });
try report(out, "feed, nothing new", feeds, feed_target_ms);
try report(out, "status", statuses, null);
try report(out, "commit, one file", commits, commit_target_ms);
}
fn report(out: *Io.Writer, what: []const u8, ns: []u64, target_ms: ?f64) Io.Writer.Error!void {
std.mem.sort(u64, ns, {}, std.sort.asc(u64));
const median = ms(percentile(ns, 50));
const p99 = ms(percentile(ns, 99));
try out.print(" {s:<20} median {d:>8.3} ms p99 {d:>8.3} ms", .{ what, median, p99 });
if (target_ms) |t| try out.print(" target {d} ms: {s}", .{ t, if (p99 <= t) "met at p99" else if (median <= t) "met at median" else "missed" });
try out.writeByte('\n');
}
fn ms(ns: u64) f64 {
return @as(f64, @floatFromInt(ns)) / std.time.ns_per_ms;
}
/// The nearest-rank percentile of an ascending, non-empty sample.
fn percentile(sorted: []const u64, p: u64) u64 {
std.debug.assert(sorted.len != 0 and p <= 100);
const rank = (p * sorted.len + 99) / 100;
return sorted[@max(rank, 1) - 1];
}
fn elapsedNs(io: Io, since: Io.Timestamp) u64 {
const d = since.durationTo(Io.Timestamp.now(io, .awake)).toNanoseconds();
return if (d <= 0) 0 else @intCast(d);
}
/// Spread over directories the way a source tree is, so k=7 partitions it.
fn fileName(a: std.mem.Allocator, i: usize) error{OutOfMemory}![]const u8 {
return std.fmt.allocPrint(a, "d{d:0>3}/f{d:0>5}.txt", .{ i % 211, i });
}
/// One commit of every file, in batches small enough for one header each.
fn seed(a: std.mem.Allocator, io: Io, c: *Client, files: usize) Error!void {
const per_commit: usize = 5000;
var at: usize = 0;
var round: i64 = 0;
while (at < files) : (round += 1) {
const n = @min(per_commit, files - at);
const edits = try a.alloc(protocol.Edit, n);
const trailers = try a.alloc([]const u8, n);
for (edits, trailers, 0..) |*e, *t, j| {
const name = try fileName(a, at + j);
t.* = try std.fmt.allocPrint(a, "{s}\nseeded content for the daemon benchmark\n", .{name});
e.* = .{ .put = .{ .name = name, .size = t.len } };
}
const v = try c.call(a, .{ .id = 0, .op = .{ .commit = .{
.user = "bench",
.comment = "seed",
.time = .{ .epoch = 1_700_000_000 + round, .ms = 0 },
.edits = edits,
} } }, trailers);
_ = v;
at += n;
}
_ = io;
}
fn absExe(a: std.mem.Allocator, io: Io) Error![]const u8 {
const cwd_abs = std.process.currentPathAlloc(io, a) catch return error.Setup;
return std.fs.path.resolve(a, &.{ cwd_abs, build_options.zit_exe });
}
fn runOnce(a: std.mem.Allocator, io: Io, dir: Io.Dir, argv: []const []const u8) Error!void {
const res = std.process.run(a, io, .{ .argv = argv, .cwd = .{ .dir = dir } }) catch return error.Spawn;
switch (res.term) {
.exited => |code| if (code == 0) return,
else => {},
}
return error.Setup;
}
fn spawnDaemon(a: std.mem.Allocator, io: Io, dir: Io.Dir, exe: []const u8) Error!std.process.Child {
var child = std.process.spawn(io, .{
.argv = &.{ exe, "daemon", "-R", "store", "--socket", "s/d.sock" },
.cwd = .{ .dir = dir },
.stdin = .ignore,
.stdout = .pipe,
.stderr = .inherit,
}) catch return error.Spawn;
errdefer child.kill(io);
var buf: [256]u8 = undefined;
var r = child.stdout.?.readerStreaming(io, &buf);
_ = r.interface.takeDelimiterInclusive('\n') catch return error.DaemonNeverListened;
_ = a;
return child;
}
const Client = struct {
stream: net.Stream,
reader: net.Stream.Reader,
writer: net.Stream.Writer,
fn open(a: std.mem.Allocator, io: Io, socket: []const u8) Error!Client {
const addr = net.UnixAddress.init(socket) catch return error.Setup;
const stream = addr.connect(io) catch return error.Setup;
const rbuf = try a.alloc(u8, 16 << 20);
const wbuf = try a.alloc(u8, 1 << 16);
return .{ .stream = stream, .reader = stream.reader(io, rbuf), .writer = stream.writer(io, wbuf) };
}
fn call(c: *Client, a: std.mem.Allocator, req: protocol.Request, trailers: []const []const u8) Error!std.json.Value {
protocol.writeRequest(&c.writer.interface, req) catch return error.Setup;
for (trailers) |t| c.writer.interface.writeAll(t) catch return error.Setup;
c.writer.interface.flush() catch return error.Setup;
const line = c.reader.interface.takeDelimiterInclusive('\n') catch return error.Setup;
const v = std.json.parseFromSliceLeaky(std.json.Value, a, line, .{}) catch return error.Setup;
if (v != .object or v.object.get("ok") == null or !v.object.get("ok").?.bool) return error.Setup;
return v;
}
};
test "percentile: nearest rank over an ascending sample" {
const s = [_]u64{ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };
try std.testing.expectEqual(@as(u64, 5), percentile(&s, 50));
try std.testing.expectEqual(@as(u64, 10), percentile(&s, 99));
try std.testing.expectEqual(@as(u64, 1), percentile(&s, 0));
try std.testing.expectEqual(@as(u64, 7), percentile(&[_]u64{7}, 99));
}
test "parseArgs: defaults, values, and refusals" {
const d = try parseArgs(&.{ "bench", "daemon" });
try std.testing.expectEqual(@as(usize, 10_000), d.files);
const v = try parseArgs(&.{ "bench", "daemon", "--files", "12000", "--rounds", "50" });
try std.testing.expectEqual(@as(usize, 12000), v.files);
try std.testing.expectEqual(@as(usize, 50), v.rounds);
try std.testing.expectError(error.BadArgs, parseArgs(&.{ "bench", "daemon", "--files" }));
try std.testing.expectError(error.BadArgs, parseArgs(&.{ "bench", "daemon", "--files", "0" }));
try std.testing.expectError(error.BadArgs, parseArgs(&.{ "bench", "daemon", "--bogus", "1" }));
}