const std = @import("std");
const repeat = @import("zit").repeat;
const builtin = @import("builtin");
const zit = @import("zit");
const storage = zit.storage;
const logstore = zit.logstore;
const cli = @import("zit").cli;
const cmd_sync = @import("cmd/sync.zig");
const net = std.Io.net;
const delta = zit.delta;
pub const Error = error{
OutOfMemory,
BadArgs,
Arithmetic,
Io,
Storage,
Corrupt,
Spawn,
ServerNeverBound,
CloneFailed,
WriteFailed,
ProcessFailed,
};
const min_width: u64 = 1;
const max_width: u64 = 200_000;
const min_file_size: u64 = 16;
const max_file_size: u64 = 64 * 1024;
const max_commits: u64 = 64;
const max_delta_depth: u64 = 16;
const max_reps: u64 = 7;
const max_changed_files: u64 = 64;
const Config = struct {
seed: u64 = 1,
width: u64 = 20_000,
file_size: u64 = 64,
commits: u64 = 20,
changed_files: u64 = 1,
delta_depth: u64 = 0,
reps: u64 = 7,
revision: []const u8 = "",
};
fn parseArgs(args: []const []const u8) Error!Config {
if (args.len < 2) return error.BadArgs;
var cfg = Config{};
var index: usize = 2;
while (index < args.len) : (index += 2) {
if (index + 1 >= args.len) return error.BadArgs;
const flag = args[index];
const value = args[index + 1];
if (std.mem.eql(u8, flag, "--width")) {
cfg.width = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--file-size")) {
cfg.file_size = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--commits")) {
cfg.commits = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--changed-files")) {
cfg.changed_files = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--delta-depth")) {
cfg.delta_depth = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--reps")) {
cfg.reps = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--seed")) {
cfg.seed = std.fmt.parseInt(u64, value, 0) catch return error.BadArgs;
} else if (std.mem.eql(u8, flag, "--revision")) {
cfg.revision = value;
} else return error.BadArgs;
}
try validateConfig(cfg);
return cfg;
}
fn validateConfig(cfg: Config) Error!void {
if (cfg.width < min_width or cfg.width > max_width) return error.BadArgs;
if (cfg.file_size < min_file_size or cfg.file_size > max_file_size) return error.BadArgs;
if (cfg.commits < 1 or cfg.commits > max_commits) return error.BadArgs;
if (cfg.changed_files < 1 or cfg.changed_files > cfg.width or cfg.changed_files > max_changed_files) return error.BadArgs;
if (cfg.delta_depth > max_delta_depth) return error.BadArgs;
if (cfg.reps < 1 or cfg.reps > max_reps) return error.BadArgs;
if (cfg.revision.len == 0 or cfg.revision.len > 256) return error.BadArgs;
if (std.mem.eql(u8, cfg.revision, "unknown")) return error.BadArgs;
}
fn churnIndex(cfg: Config, commit: u64, changed: u64) u64 {
const width = cfg.width;
const start = (cfg.seed % width +
((commit % width) * (cfg.changed_files % width)) % width) % width;
return (start + changed) % width;
}
fn fillFile(seed: u64, file_index: u64, version: u64, bytes: []u8) void {
var prng = std.Random.DefaultPrng.init(std.hash.Wyhash.hash(seed, std.mem.asBytes(&file_index)));
prng.random().bytes(bytes);
std.mem.writeInt(u64, bytes[0..8], file_index, .little);
std.mem.writeInt(u64, bytes[8..16], version, .little);
}
fn corpusFingerprint(cfg: Config) [32]u8 {
var hasher = std.crypto.hash.sha2.Sha256.init(.{});
var words: [5]u64 = .{ cfg.seed, cfg.width, cfg.file_size, cfg.commits, cfg.changed_files };
hasher.update(std.mem.sliceAsBytes(&words));
var bytes: [64 * 1024]u8 = undefined;
const file_bytes = bytes[0..@intCast(cfg.file_size)];
var index: u64 = 0;
while (index < cfg.width) : (index += 1) {
fillFile(cfg.seed, index, 0, file_bytes);
hasher.update(file_bytes);
}
var commit: u64 = 0;
while (commit < cfg.commits) : (commit += 1) {
var changed: u64 = 0;
while (changed < cfg.changed_files) : (changed += 1) {
const file_index = churnIndex(cfg, commit, changed);
fillFile(cfg.seed, file_index, commit + 1, file_bytes);
hasher.update(file_bytes);
}
}
var digest: [32]u8 = undefined;
hasher.final(&digest);
return digest;
}
const CostEntries = struct {
entries: []zit.artifact.FCard,
names: []u8,
fn deinit(self: CostEntries, gpa: std.mem.Allocator) void {
gpa.free(self.entries);
gpa.free(self.names);
}
};
fn costEntries(gpa: std.mem.Allocator, cfg: Config) Error!CostEntries {
const count: usize = @intCast(cfg.width);
const name_width: usize = 32;
const names_len = std.math.mul(usize, count, name_width) catch return error.Arithmetic;
const entries = gpa.alloc(zit.artifact.FCard, count) catch return error.OutOfMemory;
errdefer gpa.free(entries);
const names = gpa.alloc(u8, names_len) catch return error.OutOfMemory;
errdefer gpa.free(names);
for (entries, 0..) |*entry, index| {
const begin = std.math.mul(usize, index, name_width) catch return error.Arithmetic;
const filename = std.fmt.bufPrint(names[begin .. begin + name_width], "file-{d}", .{index}) catch return error.Arithmetic;
entry.* = .{
.filename = filename,
.hash = repeat("0", 64),
.perm = null,
.old_name = null,
.mode = .regular,
};
}
std.mem.sort(zit.artifact.FCard, entries, {}, struct {
fn lessThan(_: void, a: zit.artifact.FCard, b: zit.artifact.FCard) bool {
return std.mem.order(u8, a.filename, b.filename) == .lt;
}
}.lessThan);
return .{ .entries = entries, .names = names };
}
fn optimalK(gpa: std.mem.Allocator, cfg: Config) Error!u4 {
const corpus = try costEntries(gpa, cfg);
defer corpus.deinit(gpa);
var costs: [12]zit.submanifest.Cost = undefined;
var k: u4 = 1;
while (true) : (k += 1) {
costs[@as(usize, @intCast(k - 1))] = zit.submanifest.cost(
gpa,
corpus.entries,
k,
.sha3_256,
) catch return error.OutOfMemory;
if (k == 12) break;
}
return (zit.submanifest.best(&costs) orelse return error.Corrupt).k;
}
fn artifactCount(cfg: Config) Error!u64 {
const changed = std.math.mul(u64, cfg.commits, cfg.changed_files) catch return error.Arithmetic;
return std.math.add(u64, cfg.width, changed) catch return error.Arithmetic;
}
fn phaseTotal(phases: []const u64) Error!u64 {
var total: u64 = 0;
for (phases) |phase| {
total = std.math.add(u64, total, phase) catch return error.Arithmetic;
}
return total;
}
const Arm = enum { flat_default, tree_no_advice, tree_default };
fn armOrder(repetition: usize) [3]Arm {
return switch (repetition % 3) {
0 => .{ .flat_default, .tree_no_advice, .tree_default },
1 => .{ .tree_no_advice, .tree_default, .flat_default },
else => .{ .tree_default, .flat_default, .tree_no_advice },
};
}
const CommitMeasurements = struct {
wall_ns: [max_commits]u64 = @as([max_commits]u64, @splat(0)),
cpu_ns: [max_commits]u64 = @as([max_commits]u64, @splat(0)),
len: usize = 0,
artifacts: u64 = 0,
store_bytes: u64 = 0,
};
const ChangedPaths = struct {
paths: [max_changed_files][]u8 = undefined,
indexes: [max_changed_files]u64 = undefined,
len: usize = 0,
fn deinit(self: *ChangedPaths, gpa: std.mem.Allocator) void {
for (self.paths[0..self.len]) |path| gpa.free(path);
self.len = 0;
}
};
const CommitArgv = struct {
args: [8 + max_changed_files][]const u8 = undefined,
len: usize,
};
fn rewriteSchedule(gpa: std.mem.Allocator, cfg: Config, commit: u64) Error!ChangedPaths {
var schedule = ChangedPaths{};
errdefer schedule.deinit(gpa);
const count: usize = @intCast(cfg.changed_files);
var changed: usize = 0;
while (changed < count) : (changed += 1) {
const index = churnIndex(cfg, commit, @intCast(changed));
schedule.indexes[changed] = index;
schedule.paths[changed] = std.fmt.allocPrint(gpa, "file-{d}", .{index}) catch return error.OutOfMemory;
schedule.len = changed + 1;
}
return schedule;
}
fn commitArgv(arm: Arm, executable_path: []const u8, schedule: *const ChangedPaths) CommitArgv {
std.debug.assert(schedule.len <= max_changed_files);
var result = CommitArgv{ .len = 7 + schedule.len };
result.args[0..7].* = .{ executable_path, "commit", "-q", "-m", "measure", "--user", "bench" };
if (arm == .tree_no_advice) {
result.args[7] = "--no-submanifest-advice";
result.len += 1;
}
const path_start = result.len - schedule.len;
for (schedule.paths[0..schedule.len], 0..) |path, index| {
result.args[path_start + index] = path;
}
return result;
}
const CloneMeasurements = struct {
wall_ns: [max_reps]u64 = @as([max_reps]u64, @splat(0)),
receive_ns: [max_reps]u64 = @as([max_reps]u64, @splat(0)),
persist_ns: [max_reps]u64 = @as([max_reps]u64, @splat(0)),
config_ns: [max_reps]u64 = @as([max_reps]u64, @splat(0)),
warm_cache_ns: [max_reps]u64 = @as([max_reps]u64, @splat(0)),
};
fn appendNumbers(buffer: []u8, used: *usize, values: []const u64) std.fmt.BufPrintError!void {
used.* += (try std.fmt.bufPrint(buffer[used.*..], "[", .{})).len;
for (values, 0..) |value, index| {
used.* += (try std.fmt.bufPrint(
buffer[used.*..],
"{s}{d}",
.{ if (index == 0) "" else ",", value },
)).len;
}
used.* += (try std.fmt.bufPrint(buffer[used.*..], "]", .{})).len;
}
fn appendJsonString(buffer: []u8, used: *usize, value: []const u8) std.fmt.BufPrintError!void {
const hex = "0123456789abcdef";
used.* += (try std.fmt.bufPrint(buffer[used.*..], "\"", .{})).len;
for (value) |byte| {
switch (byte) {
'"', '\\' => used.* += (try std.fmt.bufPrint(buffer[used.*..], "\\{c}", .{byte})).len,
0...0x1f, 0x80...0xff => used.* += (try std.fmt.bufPrint(
buffer[used.*..],
"\\u00{c}{c}",
.{ hex[byte >> 4], hex[byte & 0x0f] },
)).len,
else => {
if (buffer.len - used.* < 1) return error.NoSpaceLeft;
buffer[used.*] = byte;
used.* += 1;
},
}
}
used.* += (try std.fmt.bufPrint(buffer[used.*..], "\"", .{})).len;
}
fn jsonLine(
buffer: []u8,
cfg: Config,
executable_sha256: []const u8,
commits: CommitMeasurements,
clones: CloneMeasurements,
phase_totals_ns: [3]u64,
nodes: u64,
arm: Arm,
) (Error || std.fmt.BufPrintError)![]const u8 {
var used: usize = 0;
const corpus_hex = std.fmt.bytesToHex(corpusFingerprint(cfg), .lower);
const wall_ns = try phaseTotal(clones.wall_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], "{{\"revision\":", .{})).len;
try appendJsonString(buffer, &used, cfg.revision);
used += (try std.fmt.bufPrint(
buffer[used..],
",\"optimize\":\"{s}\",\"seed\":{d},\"corpus_fingerprint\":\"{s}\"," ++
"\"width\":{d},\"file_size\":{d},\"commits\":{d},\"changed_files\":{d},\"delta_depth\":{d},\"reps\":{d}," ++
"\"submanifests\":{s},\"arm\":\"{s}\",\"executable_sha256\":\"{s}\",\"os\":\"{s}\",\"arch\":\"{s}\"," ++
"\"commit_wall_ns\":",
.{ @tagName(builtin.mode), cfg.seed, corpus_hex, cfg.width, cfg.file_size, cfg.commits, cfg.changed_files, cfg.delta_depth, cfg.reps, if (arm != .flat_default) "true" else "false", @tagName(arm), executable_sha256, @tagName(builtin.os.tag), @tagName(builtin.cpu.arch) },
)).len;
try appendNumbers(buffer, &used, commits.wall_ns[0..commits.len]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"commit_cpu_ns\":", .{})).len;
try appendNumbers(buffer, &used, commits.cpu_ns[0..commits.len]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"clone_wall_ns\":", .{})).len;
try appendNumbers(buffer, &used, clones.wall_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"receive_ns\":", .{})).len;
try appendNumbers(buffer, &used, clones.receive_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"persist_ns\":", .{})).len;
try appendNumbers(buffer, &used, clones.persist_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"config_ns\":", .{})).len;
try appendNumbers(buffer, &used, clones.config_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"warm_cache_ns\":", .{})).len;
try appendNumbers(buffer, &used, clones.warm_cache_ns[0..@intCast(cfg.reps)]);
used += (try std.fmt.bufPrint(buffer[used..], ",\"phase_totals_ns\":", .{})).len;
if (buffer.len - used < 1) return error.NoSpaceLeft;
buffer[used] = '{';
used += 1;
used += (try std.fmt.bufPrint(
buffer[used..],
"\"corpus\":{d},\"resolve\":{d},\"persist\":{d},\"wall\":{d}",
.{ phase_totals_ns[0], phase_totals_ns[1], phase_totals_ns[2], wall_ns },
)).len;
if (buffer.len - used < 1) return error.NoSpaceLeft;
buffer[used] = '}';
used += 1;
used += (try std.fmt.bufPrint(
buffer[used..],
",\"artifacts\":{d},\"store_bytes\":{d}," ++
"\"cold_warm_protocol\":\"first clone per spawned server is cold; later clones are warm\"," ++
"\"server_lifecycle\":\"one spawned server per arm\"," ++
"\"operation\":\"runCloneObserved\",\"operation_counters\":",
.{ commits.artifacts, commits.store_bytes },
)).len;
buffer[used] = '{';
used += 1;
used += (try std.fmt.bufPrint(
buffer[used..],
"\"clone\":{d},\"nodes\":{d}",
.{ clones.wall_ns[0..@intCast(cfg.reps)].len, nodes },
)).len;
if (buffer.len - used < 3) return error.NoSpaceLeft;
buffer[used] = '}';
buffer[used + 1] = '}';
buffer[used + 2] = '\n';
return buffer[0 .. used + 3];
}
pub const RunError = Error || std.fmt.BufPrintError || std.Io.Writer.Error;
const executable_hash_buffer_len: usize = 64 * 1024;
const max_executable_bytes: u64 = 1 << 30;
pub fn run(
gpa: std.mem.Allocator,
io: std.Io,
args: []const []const u8,
out: *std.Io.Writer,
zit_exe: []const u8,
) RunError!void {
const cfg = try parseArgs(args);
const cwd = std.process.currentPathAlloc(io, gpa) catch return error.Io;
defer gpa.free(cwd);
const executable_path = std.fs.path.resolve(gpa, &.{ cwd, zit_exe }) catch return error.OutOfMemory;
defer gpa.free(executable_path);
const executable_sha256 = try executableSha256(io, executable_path);
const executable_hex = std.fmt.bytesToHex(executable_sha256, .lower);
const nodes = try artifactCount(cfg);
var flat_clones = CloneMeasurements{};
var no_advice_clones = CloneMeasurements{};
var tree_clones = CloneMeasurements{};
var flat_commits = CommitMeasurements{};
var no_advice_commits = CommitMeasurements{};
var tree_commits = CommitMeasurements{};
var flat_phases = [_]u64{ 0, 0, 0 };
var no_advice_phases = [_]u64{ 0, 0, 0 };
var tree_phases = [_]u64{ 0, 0, 0 };
var flat = try startLiveArm(gpa, io, cfg, .flat_default, executable_path, &flat_clones, &flat_phases, &flat_commits);
defer flat.deinit(gpa, io);
var no_advice = try startLiveArm(gpa, io, cfg, .tree_no_advice, executable_path, &no_advice_clones, &no_advice_phases, &no_advice_commits);
defer no_advice.deinit(gpa, io);
var tree = try startLiveArm(gpa, io, cfg, .tree_default, executable_path, &tree_clones, &tree_phases, &tree_commits);
defer tree.deinit(gpa, io);
const reps: usize = @intCast(cfg.reps);
for (0..reps) |repetition| {
for (armOrder(repetition)) |arm| {
switch (arm) {
.flat_default => try flat.clone(gpa, io, repetition, out),
.tree_no_advice => try no_advice.clone(gpa, io, repetition, out),
.tree_default => try tree.clone(gpa, io, repetition, out),
}
}
}
var json_buffer: [16 * 1024]u8 = undefined;
try out.writeAll(try jsonLine(&json_buffer, cfg, &executable_hex, flat_commits, flat_clones, flat_phases, nodes, .flat_default));
try out.writeAll(try jsonLine(&json_buffer, cfg, &executable_hex, no_advice_commits, no_advice_clones, no_advice_phases, nodes, .tree_no_advice));
try out.writeAll(try jsonLine(&json_buffer, cfg, &executable_hex, tree_commits, tree_clones, tree_phases, nodes, .tree_default));
try out.flush();
}
fn addPhases(total: *[3]u64, next: [3]u64) Error!void {
for (total, next) |*sum, value| {
sum.* = std.math.add(u64, sum.*, value) catch return error.Arithmetic;
}
}
fn nowNs(io: std.Io) u64 {
const ns = std.Io.Clock.Timestamp.now(io, .awake).raw.nanoseconds;
if (ns < 0) return 0;
return @intCast(ns);
}
fn elapsedNs(start_ns: u64, end_ns: u64) u64 {
std.debug.assert(end_ns >= start_ns);
return end_ns - start_ns;
}
fn executableSha256(io: std.Io, path: []const u8) Error![32]u8 {
var file = std.Io.Dir.openFileAbsolute(io, path, .{ .mode = .read_only }) catch return error.Io;
defer file.close(io);
const stat = file.stat(io) catch return error.Io;
if (stat.size > max_executable_bytes) return error.BadArgs;
var hasher = std.crypto.hash.sha2.Sha256.init(.{});
var block: [executable_hash_buffer_len]u8 = undefined;
var offset: u64 = 0;
while (offset < stat.size) {
const remaining = stat.size - offset;
const next: usize = @intCast(@min(remaining, @as(u64, block.len)));
const read = file.readPositionalAll(io, block[0..next], offset) catch return error.Io;
if (read != next) return error.Io;
hasher.update(block[0..next]);
offset = std.math.add(u64, offset, @intCast(next)) catch return error.Arithmetic;
}
var digest: [32]u8 = undefined;
hasher.final(&digest);
return digest;
}
fn childCpuNs() u64 {
const usage = std.posix.getrusage(-1);
const user_seconds: u64 = @intCast(usage.utime.sec);
const system_seconds: u64 = @intCast(usage.stime.sec);
const user_us: u64 = @intCast(usage.utime.usec);
const system_us: u64 = @intCast(usage.stime.usec);
const seconds = std.math.add(u64, user_seconds, system_seconds) catch return 0;
const microseconds = std.math.add(u64, user_us, system_us) catch return 0;
const seconds_ns = std.math.mul(u64, seconds, std.time.ns_per_s) catch return 0;
const micros_ns = std.math.mul(u64, microseconds, std.time.ns_per_us) catch return 0;
return std.math.add(u64, seconds_ns, micros_ns) catch 0;
}
const clone_phase_count: usize = @typeInfo(cmd_sync.ClonePhase).@"enum".field_names.len;
const CloneObserver = struct {
io: std.Io,
started_ns: [clone_phase_count]?u64 = @as([clone_phase_count]?u64, @splat(null)),
totals_ns: [clone_phase_count]u64 = @as([clone_phase_count]u64, @splat(0)),
counts: [clone_phase_count]u64 = @as([clone_phase_count]u64, @splat(0)),
fn callback(context: *anyopaque, phase: cmd_sync.ClonePhase, event: cmd_sync.ClonePhaseEvent) void {
const self: *@This() = @ptrCast(@alignCast(context));
const index = @backingInt(phase);
switch (event) {
.begin => {
std.debug.assert(self.started_ns[index] == null);
self.started_ns[index] = nowNs(self.io);
},
.end => {
const start = self.started_ns[index].?;
self.totals_ns[index] += elapsedNs(start, nowNs(self.io));
self.started_ns[index] = null;
self.counts[index] += 1;
},
}
}
};
fn prepareSource(
gpa: std.mem.Allocator,
io: std.Io,
cfg: Config,
arm: Arm,
commits: *CommitMeasurements,
executable_path: []const u8,
) Error![]u8 {
const tree_k: ?u4 = if (arm == .flat_default) null else try optimalK(gpa, cfg);
const path = std.fmt.allocPrint(gpa, "/tmp/zit-submanifest-{d}-{s}", .{ nowNs(io), @tagName(arm) }) catch return error.OutOfMemory;
errdefer gpa.free(path);
std.Io.Dir.createDirAbsolute(io, path, .default_dir) catch return error.Io;
errdefer std.Io.Dir.cwd().deleteTree(io, path) catch {};
try runZit(io, executable_path, path, arm, "init", &.{ executable_path, "init", path });
if (tree_k) |k| {
var k_buf: [3]u8 = undefined;
const k_text = std.fmt.bufPrint(&k_buf, "{d}", .{k}) catch return error.Arithmetic;
try runZit(io, executable_path, path, arm, "submanifest-enable", &.{
executable_path, "submanifest", "enable", "--k", k_text,
});
}
try writeWorkingFiles(gpa, io, path, cfg);
try runZit(io, executable_path, path, arm, "add-initial", &.{ executable_path, "add", "." });
try runZit(io, executable_path, path, arm, "commit-initial", &.{
executable_path, "commit", "-q", "-m", "initial", "--user", "bench",
});
const bytes = gpa.alloc(u8, @intCast(cfg.file_size)) catch return error.OutOfMemory;
defer gpa.free(bytes);
var commit: u64 = 0;
while (commit < cfg.commits) : (commit += 1) {
var changed_paths = try rewriteWorkingFiles(gpa, io, path, cfg, commit, bytes);
errdefer changed_paths.deinit(gpa);
const argv = commitArgv(arm, executable_path, &changed_paths);
const wall_start = nowNs(io);
const cpu_start = childCpuNs();
try runZit(io, executable_path, path, arm, "commit-measured", argv.args[0..argv.len]);
const commit_index: usize = @intCast(commit);
commits.wall_ns[commit_index] = elapsedNs(wall_start, nowNs(io));
const cpu_end = childCpuNs();
if (cpu_end < cpu_start) return error.Arithmetic;
commits.cpu_ns[commit_index] = cpu_end - cpu_start;
commits.len = commit_index + 1;
changed_paths.deinit(gpa);
}
const zit_path = std.fmt.allocPrint(gpa, "{s}/.zit", .{path}) catch return error.OutOfMemory;
defer gpa.free(zit_path);
var zit_file = std.Io.Dir.openFileAbsolute(io, zit_path, .{ .mode = .read_write }) catch return error.Io;
defer zit_file.close(io);
const zit_size = (zit_file.stat(io) catch return error.Io).size;
var fs = storage.FileStorage.init(io, zit_file, zit_size);
var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch return error.Storage;
defer log.deinit(gpa);
const delta_last = try appendRetainedDeltaChain(gpa, &log, cfg);
if (cfg.delta_depth != 0 and log.depthOf(delta_last) != @as(u8, @intCast(cfg.delta_depth))) return error.Corrupt;
commits.artifacts = log.artifactCount();
if (cfg.delta_depth != 0 and log.maxArtifactDeltaDepth() < @as(u8, @intCast(cfg.delta_depth))) return error.Corrupt;
if (tree_k) |k| {
var config = (log.readConfig(gpa) catch return error.Storage) orelse return error.Corrupt;
defer config.deinit(gpa);
if (config.config.submanifest_k != k) return error.Corrupt;
if (config.config.submanifest_settled_cost_shape == null) return error.Corrupt;
}
commits.store_bytes = (zit_file.stat(io) catch return error.Io).size;
return path;
}
fn appendRetainedDeltaChain(gpa: std.mem.Allocator, log: *logstore.LogStore, cfg: Config) Error!logstore.Rid {
var base: [256]u8 = undefined;
fillFile(cfg.seed, cfg.width, 0, &base);
var hash: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(&base, &hash, .{});
var rid = log.put(&hash, &base) catch return error.Storage;
var depth: u64 = 0;
while (depth < cfg.delta_depth) : (depth += 1) {
var child = base;
child[8 + @as(usize, @intCast(depth))] ^= 0xa5;
const raw = delta.create(gpa, &base, &child) catch return error.OutOfMemory;
defer gpa.free(raw);
std.crypto.hash.sha3.Sha3_256.hash(&child, &hash, .{});
rid = log.putDelta(&hash, rid, raw) catch return error.Storage;
base = child;
}
log.commit() catch return error.Storage;
return rid;
}
fn runZit(
io: std.Io,
executable_path: []const u8,
cwd: []const u8,
arm: Arm,
stage: []const u8,
argv: []const []const u8,
) Error!void {
var child = std.process.spawn(io, .{
.argv = argv,
.cwd = .{ .path = cwd },
.stdin = .ignore,
.stdout = .ignore,
.stderr = .ignore,
}) catch |err| {
std.debug.print("submanifest arm={s} stage={s} spawn={s} exe={s}\n", .{
@tagName(arm), stage, @errorName(err), executable_path,
});
return error.Spawn;
};
defer child.kill(io);
const term = child.wait(io) catch |err| {
std.debug.print("submanifest arm={s} stage={s} wait={s} exe={s}\n", .{
@tagName(arm), stage, @errorName(err), executable_path,
});
return error.Spawn;
};
switch (term) {
.exited => |code| if (code != 0) {
std.debug.print("submanifest arm={s} stage={s} exit={d} exe={s}\n", .{
@tagName(arm), stage, code, executable_path,
});
return error.ProcessFailed;
},
else => {
std.debug.print("submanifest arm={s} stage={s} terminated exe={s}\n", .{
@tagName(arm), stage, executable_path,
});
return error.ProcessFailed;
},
}
}
fn writeWorkingFiles(gpa: std.mem.Allocator, io: std.Io, path: []const u8, cfg: Config) Error!void {
const bytes = gpa.alloc(u8, @intCast(cfg.file_size)) catch return error.OutOfMemory;
defer gpa.free(bytes);
var index: u64 = 0;
while (index < cfg.width) : (index += 1) {
fillFile(cfg.seed, index, 0, bytes);
try writeWorkingFile(gpa, io, path, index, bytes);
}
}
fn rewriteWorkingFiles(
gpa: std.mem.Allocator,
io: std.Io,
path: []const u8,
cfg: Config,
commit: u64,
bytes: []u8,
) Error!ChangedPaths {
var schedule = try rewriteSchedule(gpa, cfg, commit);
errdefer schedule.deinit(gpa);
for (schedule.indexes[0..schedule.len]) |index| {
fillFile(cfg.seed, index, commit + 1, bytes);
try writeWorkingFile(gpa, io, path, index, bytes);
}
return schedule;
}
fn writeWorkingFile(gpa: std.mem.Allocator, io: std.Io, path: []const u8, index: u64, bytes: []const u8) Error!void {
const file_path = std.fmt.allocPrint(gpa, "{s}/file-{d}", .{ path, index }) catch return error.OutOfMemory;
defer gpa.free(file_path);
var file = std.Io.Dir.createFileAbsolute(io, file_path, .{}) catch return error.Io;
defer file.close(io);
file.writePositionalAll(io, bytes, 0) catch return error.Io;
}
fn waitForServer(io: std.Io, port: u16) Error!void {
const addr = net.IpAddress.parseIp4("127.0.0.1", port) catch return error.ServerNeverBound;
const deadline = nowNs(io) + 30_000_000_000;
while (nowNs(io) < deadline) {
if (addr.connect(io, .{ .mode = .stream })) |stream| {
stream.close(io);
return;
} else |_| {}
io.sleep(.{ .nanoseconds = 500_000 }, .awake) catch return error.Io;
}
return error.ServerNeverBound;
}
const LiveArm = struct {
source: []u8,
url: []u8,
child: std.process.Child,
clones: *CloneMeasurements,
phase_totals: *[3]u64,
fn clone(self: *LiveArm, gpa: std.mem.Allocator, io: std.Io, repetition: usize, diagnostic: *std.Io.Writer) Error!void {
try cloneOne(gpa, io, self.url, repetition, diagnostic, self.clones, self.phase_totals);
}
fn deinit(self: *LiveArm, gpa: std.mem.Allocator, io: std.Io) void {
self.child.kill(io);
std.Io.Dir.cwd().deleteTree(io, self.source) catch {};
gpa.free(self.url);
gpa.free(self.source);
}
};
fn startLiveArm(
gpa: std.mem.Allocator,
io: std.Io,
cfg: Config,
arm: Arm,
executable_path: []const u8,
clones: *CloneMeasurements,
phase_totals: *[3]u64,
commits: *CommitMeasurements,
) Error!LiveArm {
const source = try prepareSource(gpa, io, cfg, arm, commits, executable_path);
errdefer {
std.Io.Dir.cwd().deleteTree(io, source) catch {};
gpa.free(source);
}
const port = try freshPort(io);
const port_text = std.fmt.allocPrint(gpa, "{d}", .{port}) catch return error.OutOfMemory;
defer gpa.free(port_text);
var child = std.process.spawn(io, .{
.argv = &.{ executable_path, "serve", "--port", port_text },
.cwd = .{ .path = source },
.stdout = .ignore,
.stderr = .ignore,
}) catch return error.Spawn;
errdefer child.kill(io);
try waitForServer(io, port);
const url = std.fmt.allocPrint(gpa, "http://127.0.0.1:{d}", .{port}) catch return error.OutOfMemory;
return .{ .source = source, .url = url, .child = child, .clones = clones, .phase_totals = phase_totals };
}
fn cloneOne(
gpa: std.mem.Allocator,
io: std.Io,
url: []const u8,
index: usize,
diagnostic: *std.Io.Writer,
clones: *CloneMeasurements,
phase_totals: *[3]u64,
) Error!void {
const target = std.fmt.allocPrint(gpa, "/tmp/zit-submanifest-clone-{d}-{d}", .{ nowNs(io), index }) catch return error.OutOfMemory;
defer gpa.free(target);
defer std.Io.Dir.cwd().deleteTree(io, target) catch {};
var observer = CloneObserver{ .io = io };
const callback = cmd_sync.CloneObserver{ .context = &observer, .callback = CloneObserver.callback };
var output_buffer: [512]u8 = undefined;
var error_buffer: [512]u8 = undefined;
var output = std.Io.Writer.fixed(&output_buffer);
var errors = std.Io.Writer.fixed(&error_buffer);
const streams: cli.Streams = .{ .out = &output, .err = &errors };
var arena = std.heap.ArenaAllocator.init(gpa);
defer arena.deinit();
const wall_start = nowNs(io);
if (cmd_sync.runCloneObserved(gpa, io, arena.allocator(), &.{ url, target }, streams, &callback) != .ok) {
try diagnostic.writeAll(errors.buffered());
return error.CloneFailed;
}
const wall_end = nowNs(io);
if (wall_end < wall_start) return error.Arithmetic;
const grouped = try groupClonePhases(observer.totals_ns);
const wall_ns = wall_end - wall_start;
try validateCloneAccounting(wall_ns, grouped);
clones.wall_ns[index] = wall_ns;
clones.receive_ns[index] = grouped[1];
clones.persist_ns[index] = observer.totals_ns[@backingInt(cmd_sync.ClonePhase.persist)];
clones.config_ns[index] = observer.totals_ns[@backingInt(cmd_sync.ClonePhase.config)];
clones.warm_cache_ns[index] = observer.totals_ns[@backingInt(cmd_sync.ClonePhase.warm_cache)];
try addPhases(phase_totals, grouped);
}
fn freshPort(io: std.Io) Error!u16 {
const addr = net.IpAddress.parseIp4("127.0.0.1", 0) catch return error.ServerNeverBound;
var server = addr.listen(io, .{}) catch return error.ServerNeverBound;
defer server.deinit(io);
const port = server.socket.address.getPort();
if (port == 0) return error.ServerNeverBound;
return port;
}
fn groupClonePhases(phases: [clone_phase_count]u64) Error![3]u64 {
const transport = std.math.add(
u64,
phases[@backingInt(cmd_sync.ClonePhase.transport)],
phases[@backingInt(cmd_sync.ClonePhase.initialize)],
) catch return error.Arithmetic;
const tail = std.math.add(
u64,
phases[@backingInt(cmd_sync.ClonePhase.persist)],
phases[@backingInt(cmd_sync.ClonePhase.config)],
) catch return error.Arithmetic;
const warm = std.math.add(
u64,
tail,
phases[@backingInt(cmd_sync.ClonePhase.warm_cache)],
) catch return error.Arithmetic;
return .{ transport, phases[@backingInt(cmd_sync.ClonePhase.exchange)], warm };
}
fn validateCloneAccounting(wall_ns: u64, phases: [3]u64) Error!void {
const phase_ns = try phaseTotal(&phases);
if (phase_ns > wall_ns) return error.Corrupt;
const residual_ns = wall_ns - phase_ns;
if (residual_ns > wall_ns / 20) return error.Corrupt;
}
test "submanifest parser accepts complete measurement CLI and production width" {
const argv = [_][]const u8{
"bench", "submanifest", "--width", "200000", "--file-size", "64", "--commits", "20",
"--changed-files", "1", "--delta-depth", "0", "--reps", "7", "--seed", "1",
"--revision", "307d7287-local-submanifest",
};
const cfg = try parseArgs(&argv);
try std.testing.expectEqual(@as(u64, 200_000), cfg.width);
try std.testing.expectEqualStrings("307d7287-local-submanifest", cfg.revision);
}
test "submanifest parser rejects missing or unknown provenance" {
const missing = [_][]const u8{ "bench", "submanifest" };
try std.testing.expectError(error.BadArgs, parseArgs(&missing));
const unknown = [_][]const u8{ "bench", "submanifest", "--revision", "unknown" };
try std.testing.expectError(error.BadArgs, parseArgs(&unknown));
}
test "submanifest parser accepts retained delta depths" {
const argv = [_][]const u8{
"bench", "submanifest", "--delta-depth", "4", "--commits", "4", "--revision", "local",
};
const cfg = try parseArgs(&argv);
try std.testing.expectEqual(@as(u64, 4), cfg.delta_depth);
}
test "submanifest parser bounds retained delta depth" {
const accepted = [_][]const u8{ "bench", "submanifest", "--delta-depth", "16", "--revision", "test" };
try std.testing.expectEqual(@as(u64, max_delta_depth), (try parseArgs(&accepted)).delta_depth);
const rejected = [_][]const u8{ "bench", "submanifest", "--delta-depth", "17", "--revision", "test" };
try std.testing.expectError(error.BadArgs, parseArgs(&rejected));
}
test "submanifest parser enforces nonperiodic payload minimum" {
const rejected = [_][]const u8{ "bench", "submanifest", "--file-size", "15", "--revision", "test" };
try std.testing.expectError(error.BadArgs, parseArgs(&rejected));
const accepted = [_][]const u8{ "bench", "submanifest", "--file-size", "16", "--revision", "test" };
try std.testing.expectEqual(@as(u64, min_file_size), (try parseArgs(&accepted)).file_size);
}
test "submanifest minimum file payload encodes every bounded version" {
var v0: [min_file_size]u8 = undefined;
var v1: [min_file_size]u8 = undefined;
var v2: [min_file_size]u8 = undefined;
var vmax: [min_file_size]u8 = undefined;
fillFile(1, 7, 0, &v0);
fillFile(1, 7, 1, &v1);
fillFile(1, 7, 2, &v2);
fillFile(1, 7, max_commits, &vmax);
try std.testing.expectEqual(@as(u64, 7), std.mem.readInt(u64, v2[0..8], .little));
try std.testing.expectEqual(max_commits, std.mem.readInt(u64, vmax[8..16], .little));
try std.testing.expect(!std.mem.eql(u8, &v0, &v1));
try std.testing.expect(!std.mem.eql(u8, &v1, &v2));
try std.testing.expect(!std.mem.eql(u8, &v2, &vmax));
}
test "submanifest sparse schedule drives rewritten paths and commit argv" {
const cfg = Config{
.seed = 9,
.width = 10,
.changed_files = 3,
.revision = "test",
};
var scheduled = try rewriteSchedule(std.testing.allocator, cfg, 4);
defer scheduled.deinit(std.testing.allocator);
try std.testing.expectEqual(@as(usize, 3), scheduled.len);
try std.testing.expectEqualSlices(u64, &.{ 1, 2, 3 }, scheduled.indexes[0..scheduled.len]);
try std.testing.expectEqualStrings("file-1", scheduled.paths[0]);
try std.testing.expectEqualStrings("file-2", scheduled.paths[1]);
try std.testing.expectEqualStrings("file-3", scheduled.paths[2]);
const argv = commitArgv(.tree_no_advice, "/tmp/zit", &scheduled);
try std.testing.expectEqual(@as(usize, 11), argv.len);
try std.testing.expectEqualStrings("--no-submanifest-advice", argv.args[7]);
for (scheduled.paths[0..scheduled.len], 0..) |path, index| {
try std.testing.expectEqualStrings(path, argv.args[8 + index]);
}
}
test "submanifest repetitions rotate arm order deterministically" {
try std.testing.expectEqualSlices(Arm, &.{ .flat_default, .tree_no_advice, .tree_default }, &armOrder(0));
try std.testing.expectEqualSlices(Arm, &.{ .tree_no_advice, .tree_default, .flat_default }, &armOrder(1));
try std.testing.expectEqualSlices(Arm, &.{ .tree_default, .flat_default, .tree_no_advice }, &armOrder(2));
try std.testing.expectEqualSlices(Arm, &.{ .flat_default, .tree_no_advice, .tree_default }, &armOrder(3));
}
test "submanifest tree pin equals the production cost winner" {
const cfg = Config{ .width = 64, .file_size = 64, .commits = 2, .changed_files = 1, .revision = "test" };
const selected = try optimalK(std.testing.allocator, cfg);
const corpus = try costEntries(std.testing.allocator, cfg);
defer corpus.deinit(std.testing.allocator);
var costs: [12]zit.submanifest.Cost = undefined;
var k: u4 = 1;
while (true) : (k += 1) {
costs[@as(usize, @intCast(k - 1))] = try zit.submanifest.cost(
std.testing.allocator,
corpus.entries,
k,
.sha3_256,
);
if (k == 12) break;
}
try std.testing.expectEqual((zit.submanifest.best(&costs) orelse unreachable).k, selected);
}
test "submanifest corpus fingerprint is deterministic and seed-specific" {
const cfg = Config{ .width = 10, .file_size = 64, .commits = 2, .changed_files = 1, .revision = "test" };
const a = corpusFingerprint(cfg);
const b = corpusFingerprint(cfg);
try std.testing.expectEqualSlices(u8, &a, &b);
const other = corpusFingerprint(.{ .seed = 2, .width = 10, .file_size = 64, .commits = 2, .changed_files = 1, .revision = "test" });
try std.testing.expect(!std.mem.eql(u8, &a, &other));
}
test "submanifest clone accounting compares phases to independent wall time" {
try validateCloneAccounting(100, .{ 11, 17, 67 });
try std.testing.expectError(error.Corrupt, validateCloneAccounting(100, .{ 11, 17, 66 }));
try std.testing.expectError(error.Corrupt, validateCloneAccounting(100, .{ 11, 17, 73 }));
}
test "submanifest JSONL escapes revision and records corpus provenance" {
var buf: [2048]u8 = undefined;
const cfg = Config{ .width = 10, .file_size = 64, .commits = 2, .changed_files = 1, .reps = 2, .revision = "local\"\\\n" };
var commits = CommitMeasurements{ .len = 2, .artifacts = 13, .store_bytes = 99 };
commits.wall_ns[0] = 1;
commits.wall_ns[1] = 2;
var clones = CloneMeasurements{};
clones.wall_ns[0] = 3;
clones.wall_ns[1] = 5;
const line = try jsonLine(
&buf,
cfg,
"0000000000000000000000000000000000000000000000000000000000000000",
commits,
clones,
.{ 1, 2, 3 },
13,
.tree_default,
);
for ([_][]const u8{
"\"optimize\"", "\"seed\"", "\"corpus_fingerprint\"", "\"reps\"", "\"width\"", "\"file_size\"", "\"commits\"", "\"changed_files\"",
"\"delta_depth\"", "\"submanifests\":true", "\"commit_wall_ns\"", "\"commit_cpu_ns\"", "\"clone_wall_ns\":[3,5]", "\"receive_ns\"", "\"persist_ns\"", "\"config_ns\"",
"\"warm_cache_ns\"", "\"artifacts\"", "\"store_bytes\"", "\"executable_sha256\"",
}) |field| try std.testing.expect(std.mem.indexOf(u8, line, field) != null);
var parsed = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, line, .{});
defer parsed.deinit();
try std.testing.expectEqualStrings(cfg.revision, parsed.value.object.get("revision").?.string);
}