//! 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;
//! * the read ops a history view makes: `log` of one file, a page of 20
//! rows, `annotate` of it, and `diff` of it between two check-ins with
//! its hunks. The file is first given `history_revisions` versions
//! among the one-file commits of other files, so the walk has rows to
//! list and check-ins to skip.
//!
//! 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.
//!
//! `--scaling` instead times one-file commits on one store at
//! `scaling_small` files and again after growing it to `scaling_large`, and
//! fails unless the large median is under `scaling_ratio_max` times the
//! small: a commit must cost what it changes, not what the store holds. A
//! ratio, so a loaded machine slows both sides alike.
//!
//! 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,
/// `--scaling`: the large store's commit median broke the ratio.
CommitCostGrows,
} || Io.Writer.Error;
const feed_target_ms: f64 = 10;
/// What a history view should cost against the CLI's 170-640 ms per call
/// on a large store, which reopens the store each time.
const read_target_ms: f64 = 30;
const history_revisions: usize = 60;
const commit_target_ms: f64 = 30;
const max_files: usize = 200_000;
const max_rounds: usize = 100_000;
const scaling_small: usize = 2_000;
const scaling_large: usize = 42_000;
const scaling_ratio_max: f64 = 2;
const Config = struct {
files: usize = 10_000,
rounds: usize = 200,
scaling: bool = false,
};
fn parseArgs(args: []const []const u8) Error!Config {
var cfg: Config = .{};
var i: usize = 2;
while (i < args.len) : (i += 2) {
if (std.mem.eql(u8, args[i], "--scaling")) {
cfg.scaling = true;
i -= 1;
continue;
}
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);
if (cfg.scaling) return scaling(a, io, &c, cfg.rounds, out);
try seed(a, io, &c, 0, cfg.files, 1_700_000_000);
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 timeCommits(a, io, &c, cfg.files, cfg.rounds, 0, 1_800_000_000);
// One file's history, interleaved with other files' commits.
const path = try fileName(a, 4242 % cfg.files);
var hashes: [2][]const u8 = undefined;
for (0..history_revisions) |r| {
const other = try fileName(a, (r * 104729 + 1) % cfg.files);
for ([_][]const u8{ path, other }, 0..) |name, j| {
var body: std.ArrayList(u8) = .empty;
for (0..r + 1) |line| try body.print(a, "{s} line {d} of revision {d}\n", .{ name, line, r });
const v = try c.call(a, .{ .id = 0, .op = .{ .commit = .{
.user = "bench",
.comment = "history",
.time = .{ .epoch = 1_900_000_000 + @as(i64, @intCast(2 * r + j)), .ms = 0 },
.edits = &.{.{ .put = .{ .name = name, .size = body.items.len } }},
} } }, &.{body.items});
if (j == 0 and r + 2 >= history_revisions) hashes[r + 2 - history_revisions] = try a.dupe(u8, v.object.get("hash").?.string);
}
}
const logs = try a.alloc(u64, cfg.rounds);
for (logs, 0..) |*t, i| {
const start = Io.Timestamp.now(io, .awake);
const v = try c.call(a, .{ .id = @intCast(i), .op = .{ .log = .{ .path = path, .limit = 20 } } }, &.{});
t.* = elapsedNs(io, start);
if (v.object.get("checkins").?.array.items.len != 20) return error.Setup;
}
const annotates = try a.alloc(u64, cfg.rounds);
for (annotates, 0..) |*t, i| {
const start = Io.Timestamp.now(io, .awake);
const v = try c.call(a, .{ .id = @intCast(i), .op = .{ .annotate = .{ .path = path } } }, &.{});
t.* = elapsedNs(io, start);
if (v.object.get("lines").?.array.items.len != history_revisions) return error.Setup;
}
const diffs = try a.alloc(u64, cfg.rounds);
for (diffs, 0..) |*t, i| {
const start = Io.Timestamp.now(io, .awake);
const v = try c.call(a, .{ .id = @intCast(i), .op = .{ .diff = .{ .from = hashes[0], .to = hashes[1], .path = path } } }, &.{});
t.* = elapsedNs(io, start);
if (v.object.get("files").?.array.items.len != 1) 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);
try report(out, "log -n 20, one file", logs, read_target_ms);
try report(out, "annotate, one file", annotates, read_target_ms);
try report(out, "diff, one file", diffs, read_target_ms);
}
/// One-file commits, each request to reply, on a store of `files` files.
/// `first` keeps the contents of several runs on one store apart, and
/// `epoch` their D times.
fn timeCommits(a: std.mem.Allocator, io: Io, c: *Client, files: usize, rounds: usize, first: usize, epoch: i64) Error![]u64 {
const commits = try a.alloc(u64, rounds);
for (commits, first..) |*t, i| {
const name = try fileName(a, i * 7919 % 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 = epoch + @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;
}
return commits;
}
fn scaling(a: std.mem.Allocator, io: Io, c: *Client, rounds: usize, out: *Io.Writer) Error!void {
try seed(a, io, c, 0, scaling_small, 1_700_000_000);
const small = try timeCommits(a, io, c, scaling_small, rounds, 0, 1_800_000_000);
try seed(a, io, c, scaling_small, scaling_large, 1_810_000_000);
const large = try timeCommits(a, io, c, scaling_large, rounds, rounds, 1_820_000_000);
try out.print("daemon commit scaling, one k=7 store grown from {d} to {d} files, {d} commits each\n", .{ scaling_small, scaling_large, rounds });
try report(out, "commit, small store", small, null);
try report(out, "commit, large store", large, null);
const ratio = ms(percentile(large, 50)) / ms(@max(percentile(small, 50), 1));
try out.print(" median ratio {d:.2} (must be under {d})\n", .{ ratio, scaling_ratio_max });
try out.flush();
if (ratio >= scaling_ratio_max) return error.CommitCostGrows;
}
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 `[from, files)`, in batches small enough for
/// one header each, dated from `epoch` on.
fn seed(a: std.mem.Allocator, io: Io, c: *Client, from: usize, files: usize, epoch: i64) Error!void {
const per_commit: usize = 5000;
var at: usize = from;
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 = epoch + 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" }));
const s = try parseArgs(&.{ "bench", "daemon", "--scaling", "--rounds", "30" });
try std.testing.expect(s.scaling);
try std.testing.expectEqual(@as(usize, 30), s.rounds);
}