FuzzTestRunner

const FuzzTestRunner = struct

File

Code

const FuzzTestRunner = struct {
    run: *Run,
    run_index: Configuration.Step.Index,
    ctx: FuzzContext,
    coverage_id: ?u64,

    instances: []Instance,
    /// The indexes of this are layed out such that it is effectively an array
    /// of `[instances.len][3]Io.Operation.Storage` of stdin, stdout, stderr.
    batch: Io.Batch,
    /// LIFO. Stream of message bodies trailed by PendingBroadcastFooter.
    pending_broadcasts: std.ArrayList(u8),
    broadcast: std.ArrayList(u8),
    broadcast_undelivered: u32,

    const Instance = struct {
        child: process.Child,
        message: std.ArrayListAligned(u8, .@"4"),
        broadcast_written: usize,
        stderr: std.ArrayList(u8),
        stdin_vec: [1][]u8,
        stdout_vec: [1][]u8,
        stderr_vec: [1][]u8,
        progress_node: std.Progress.Node,

        fn messageHeader(instance: *Instance) InHeader {
            assert(instance.message.items.len >= @sizeOf(InHeader));
            const header_ptr: *InHeader = @ptrCast(instance.message.items);
            var header = header_ptr.*;
            if (std.builtin.Endian.native != .little) {
                std.mem.byteSwapAllFields(InHeader, &header);
            }
            return header;
        }
    };

    const PendingBroadcastFooter = struct {
        from_id: u32,
        body_len: u32,
    };

    const InHeader = std.zig.Server.Message.Header;
    const OutHeader = std.zig.Client.Message.Header;

    const stdin_i = 0;
    const stdout_i = 1;
    const stderr_i = 2;

    fn init(
        run: *Run,
        run_index: Configuration.Step.Index,
        ctx: FuzzContext,
        progress_node: std.Progress.Node,
        spawn_options: process.SpawnOptions,
    ) !FuzzTestRunner {
        const maker = ctx.fuzz.maker;
        const graph = maker.graph;
        const gpa = maker.gpa;
        const io = graph.io;

        const n_instances = switch (ctx.fuzz.mode) {
            .forever => graph.max_jobs orelse @min(
                std.Thread.getCpuCount() catch 1,
                (std.math.maxInt(u32) - 2) / 3,
            ),
            .limit => 1,
        };
        const instances = try gpa.alloc(Instance, n_instances);
        errdefer gpa.free(instances);
        const batch_storage = try gpa.alloc(Io.Operation.Storage, instances.len * 3);
        errdefer gpa.free(batch_storage);

        @memset(instances, .{
            .child = undefined,
            .message = .empty,
            .broadcast_written = undefined,
            .stderr = .empty,
            .stdin_vec = undefined,
            .stdout_vec = undefined,
            .stderr_vec = undefined,
            .progress_node = undefined,
        });
        for (0.., instances) |id, *instance| {
            errdefer for (instances[0..id]) |*spawned| {
                spawned.child.kill(io);
                spawned.progress_node.end();
            };
            instance.child = try process.spawn(io, spawn_options);
            instance.progress_node = progress_node.start("starting fuzzer", 0);
        }

        return .{
            .run = run,
            .run_index = run_index,
            .ctx = ctx,
            .coverage_id = null,

            .instances = instances,
            .batch = .init(batch_storage),
            .pending_broadcasts = .empty,
            .broadcast = .empty,
            .broadcast_undelivered = 0,
        };
    }

    fn deinit(f: *FuzzTestRunner) void {
        const maker = f.ctx.fuzz.maker;
        const run_index = f.run_index;

        const graph = maker.graph;
        const gpa = maker.gpa;
        const io = graph.io;
        const step = maker.stepByIndex(run_index);

        f.batch.cancel(io);
        gpa.free(f.batch.storage);
        var total_rss: usize = 0;
        for (f.instances) |*instance| {
            instance.child.kill(io);
            instance.message.deinit(gpa);
            instance.stderr.deinit(gpa);
            instance.progress_node.end();
            total_rss += instance.child.resource_usage_statistics.getMaxRss() orelse 0;
        }
        step.result_peak_rss = @max(step.result_peak_rss, total_rss);
        gpa.free(f.instances);
    }

    fn startInstances(f: *FuzzTestRunner) !void {
        const maker = f.ctx.fuzz.maker;
        const run_index = f.run_index;
        const run = f.run;

        const graph = maker.graph;
        const io = graph.io;
        const step = maker.stepByIndex(run_index);

        for (0.., f.instances) |id, *instance| {
            const id32: u32 = @intCast(id);
            (switch (f.ctx.fuzz.mode) {
                .forever => sendRunFuzzTestMessage(
                    io,
                    instance.child.stdin.?,
                    run.fuzz_tests.items,
                    .forever,
                    id32,
                ),
                .limit => |limit| sendRunFuzzTestMessage(
                    io,
                    instance.child.stdin.?,
                    run.fuzz_tests.items,
                    .iterations,
                    limit.amount,
                ),
            }) catch |write_err| {
                // The runner unexpectedly closed stdin, which means it crashed during initialization.
                // Clean up everything and wait for the child to exit.
                instance.child.stdin.?.close(io);
                instance.child.stdin = null;
                const term = try instance.child.wait(io);
                return step.fail(
                    maker,
                    "unable to write stdin ({t}); test process unexpectedly {f}",
                    .{ write_err, fmtTerm(term) },
                );
            };

            try f.addStdoutRead(id32, @sizeOf(InHeader));
            try f.addStderrRead(id32);
        }
    }

    fn listen(f: *FuzzTestRunner) !void {
        const maker = f.ctx.fuzz.maker;
        const graph = maker.graph;
        const io = graph.io;

        while (true) {
            try f.batch.awaitConcurrent(io, .none);
            while (f.batch.next()) |completion| {
                const id = completion.index / 3;
                const result = completion.result;
                switch (completion.index % 3) {
                    0 => try f.completeStdinWrite(id, result.file_write_streaming catch |e| switch (e) {
                        // Avoid calling `instanceEos` until EndOfStream is seen with stderr so
                        // that all stderr is collected.
                        error.BrokenPipe => continue,
                        else => |write_e| return write_e,
                    }),
                    1 => try f.completeStdoutRead(id, result.file_read_streaming catch |e| switch (e) {
                        // Avoid calling `instanceEos` until EndOfStream is seen with stderr so
                        // that all stderr is collected.
                        error.EndOfStream => continue,
                        else => |read_e| return read_e,
                    }),
                    2 => try f.completeStderrRead(id, result.file_read_streaming catch |e| switch (e) {
                        error.EndOfStream => return f.instanceEos(id),
                        else => |read_e| return read_e,
                    }),
                    else => unreachable,
                }
            }
        }
    }

    fn completeStdoutRead(f: *FuzzTestRunner, id: u32, n: usize) !void {
        const maker = f.ctx.fuzz.maker;
        const instance = &f.instances[id];
        const run_index = f.run_index;
        const run = f.run;

        const graph = maker.graph;
        const gpa = maker.gpa;
        const io = graph.io;
        const step = maker.stepByIndex(run_index);

        instance.message.items.len += n;
        const total_read = instance.message.items.len;
        if (total_read < @sizeOf(InHeader)) {
            try f.addStdoutRead(id, @sizeOf(InHeader));
            return;
        }

        const header = instance.messageHeader();
        const body = instance.message.items[@sizeOf(InHeader)..];
        if (body.len != header.bytes_len) {
            try f.addStdoutRead(id, @sizeOf(InHeader) + header.bytes_len);
            return;
        }

        switch (header.tag) {
            .zig_version => {
                if (!std.mem.eql(u8, builtin.zig_version_string, body)) return step.fail(
                    maker,
                    "zig version mismatch build runner vs compiler: '{s}' vs '{s}'",
                    .{ builtin.zig_version_string, body },
                );
            },
            .coverage_id => {
                var body_r: Io.Reader = .fixed(body);
                f.coverage_id = body_r.takeInt(u64, .little) catch unreachable;
                const cumulative_runs = body_r.takeInt(u64, .little) catch unreachable;
                const cumulative_unique = body_r.takeInt(u64, .little) catch unreachable;
                const cumulative_coverage = body_r.takeInt(u64, .little) catch unreachable;

                const fuzz = f.ctx.fuzz;
                fuzz.queue_mutex.lockUncancelable(io);
                defer fuzz.queue_mutex.unlock(io);
                try fuzz.msg_queue.append(gpa, .{ .coverage = .{
                    .id = f.coverage_id.?,
                    .cumulative = .{
                        .runs = cumulative_runs,
                        .unique = cumulative_unique,
                        .coverage = cumulative_coverage,
                    },
                    .run = run_index,
                } });
                fuzz.queue_cond.signal(io);
            },
            .fuzz_start_addr => {
                var body_r: Io.Reader = .fixed(body);
                const fuzz = f.ctx.fuzz;
                const addr = body_r.takeInt(u64, .little) catch unreachable;

                fuzz.queue_mutex.lockUncancelable(io);
                defer fuzz.queue_mutex.unlock(io);
                try fuzz.msg_queue.append(gpa, .{ .entry_point = .{
                    .addr = addr,
                    .coverage_id = f.coverage_id.?,
                } });
                fuzz.queue_cond.signal(io);
            },
            .fuzz_test_change => {
                const test_i = std.mem.readInt(u32, body[0..4], .little);
                instance.progress_node.setName(run.fuzz_tests.items[test_i]);
            },
            .broadcast_fuzz_input => {
                if (f.instances.len == 1) {
                    // No other processes to broadcast to.
                } else if (f.broadcast_undelivered == 0) {
                    try f.instanceBroadcast(id, body);
                } else {
                    const footer: PendingBroadcastFooter = .{
                        .from_id = id,
                        .body_len = @intCast(body.len),
                    };
                    // There is another broadcast in progress so add this one to the queue.
                    const size = @sizeOf(PendingBroadcastFooter) + body.len;
                    try f.pending_broadcasts.ensureUnusedCapacity(gpa, size);
                    f.pending_broadcasts.appendSliceAssumeCapacity(body);
                    f.pending_broadcasts.appendSliceAssumeCapacity(@ptrCast(&footer));
                }
            },
            else => {}, // ignore other messages
        }

        instance.message.clearRetainingCapacity();
        try f.addStdoutRead(id, @sizeOf(InHeader));
    }

    fn completeStderrRead(f: *FuzzTestRunner, id: u32, n: usize) !void {
        const instance = &f.instances[id];
        instance.stderr.items.len += n;
        try f.addStderrRead(id);
    }

    fn completeStdinWrite(f: *FuzzTestRunner, id: u32, n: usize) !void {
        const instance = &f.instances[id];

        instance.broadcast_written += n;
        if (instance.broadcast_written == f.broadcast.items.len) {
            f.broadcast_undelivered -= 1;
            if (f.broadcast_undelivered == 0) {
                try f.broadcastComplete();
            }
        } else {
            f.addStdinWrite(id);
        }
    }

    fn addStdoutRead(f: *FuzzTestRunner, id: u32, end: usize) !void {
        const maker = f.ctx.fuzz.maker;
        const gpa = maker.gpa;
        const instance = &f.instances[id];

        try instance.message.ensureTotalCapacity(gpa, end);
        const start = instance.message.items.len;
        instance.stdout_vec = .{instance.message.allocatedSlice()[start..end]};
        f.batch.addAt(id * 3 + stdout_i, .{ .file_read_streaming = .{
            .file = instance.child.stdout.?,
            .data = &instance.stdout_vec,
        } });
    }

    fn addStderrRead(f: *FuzzTestRunner, id: u32) !void {
        const maker = f.ctx.fuzz.maker;
        const gpa = maker.gpa;
        const instance = &f.instances[id];

        try instance.stderr.ensureUnusedCapacity(gpa, 1);
        instance.stderr_vec = .{instance.stderr.unusedCapacitySlice()};
        f.batch.addAt(id * 3 + stderr_i, .{ .file_read_streaming = .{
            .file = instance.child.stderr.?,
            .data = &instance.stderr_vec,
        } });
    }

    fn addStdinWrite(f: *FuzzTestRunner, id: u32) void {
        const instance = &f.instances[id];

        assert(f.broadcast.items.len != instance.broadcast_written);
        instance.stdin_vec = .{f.broadcast.items[instance.broadcast_written..]};
        f.batch.addAt(id * 3 + stdin_i, .{ .file_write_streaming = .{
            .file = instance.child.stdin.?,
            .data = &instance.stdin_vec,
        } });
    }

    fn instanceEos(f: *FuzzTestRunner, id: u32) !void {
        const maker = f.ctx.fuzz.maker;
        const gpa = maker.gpa;
        const instance = &f.instances[id];
        const run_index = f.run_index;

        const graph = maker.graph;
        const io = graph.io;
        const step = maker.stepByIndex(run_index);

        instance.child.stdin.?.close(io);
        instance.child.stdin = null;
        const term = try instance.child.wait(io);
        if (!termMatches(.{ .exited = 0 }, term)) {
            step.takeResultStderr(gpa, try f.mergedStderr(gpa));
            try f.saveCrash(id, term);
            return step.fail(maker, "test process unexpectedly {f}", .{fmtTerm(term)});
        }
    }

    fn saveCrash(f: *FuzzTestRunner, id: u32, term: process.Child.Term) !void {
        const fuzz = f.ctx.fuzz;
        const run_index = f.run_index;
        const run = f.run;

        const maker = fuzz.maker;
        const step = maker.stepByIndex(run_index);
        const graph = maker.graph;
        const io = graph.io;
        const cache_root = graph.local_cache_root;

        if (f.coverage_id == null) return;

        // Search for the input file corresponding to the instance
        const InputHeader = std.Build.abi.fuzz.MmapInputHeader;
        var in_r_buf: [@sizeOf(InputHeader)]u8 = undefined;
        var in_r: Io.File.Reader = undefined;
        var in_f: Io.File = undefined;
        var in_name_buf: [12]u8 = undefined;
        var in_name: []const u8 = undefined;
        var i: u32 = 0;
        const header: InputHeader = while (true) : ({
            if (i == std.math.maxInt(u32)) return;
            i += 1;
        }) {
            const name_prefix = "f" ++ Dir.path.sep_str ++ "in";
            in_name = std.fmt.bufPrint(&in_name_buf, name_prefix ++ "{x}", .{i}) catch unreachable;
            in_f = cache_root.handle.openFile(io, in_name, .{
                .lock = .exclusive,
                .lock_nonblocking = true,
            }) catch |e| switch (e) {
                error.FileNotFound => return,
                error.WouldBlock => continue, // Can not be from
                // the crashed instance since it is still locked.
                else => return step.fail(maker, "failed to open file '{f}{s}': {t}", .{
                    cache_root, in_name, e,
                }),
            };

            in_r = in_f.readerStreaming(io, &in_r_buf);
            const header = in_r.interface.takeStruct(InputHeader, .little) catch |e| {
                in_f.close(io);
                switch (e) {
                    error.ReadFailed => return step.fail(maker, "failed to read file '{f}{s}': {t}", .{
                        cache_root, in_name, in_r.err.?,
                    }),
                    error.EndOfStream => continue,
                }
            };

            if (header.pc_digest == f.coverage_id.? and
                header.instance_id == id and
                header.test_i < run.fuzz_tests.items.len)
            {
                break header;
            }

            in_f.close(io);
        };
        defer in_f.close(io);

        // Save it to a seperate file
        const crash_name = "f" ++ Dir.path.sep_str ++ "crash";
        const out = cache_root.handle.createFile(io, crash_name, .{
            .lock = .exclusive, // Multiple run steps could have found a crash at the same time
        }) catch |e| return step.fail(maker, "failed to create file '{f}{s}': {t}", .{
            cache_root, crash_name, e,
        });
        defer out.close(io);

        var out_w_buf: [512]u8 = undefined;
        var out_w = out.writerStreaming(io, &out_w_buf);
        _ = out_w.interface.sendFileAll(&in_r, .limited(header.len)) catch |e| switch (e) {
            error.ReadFailed => return step.fail(maker, "failed to read file '{f}{s}': {t}", .{
                cache_root, in_name, in_r.err.?,
            }),
            error.WriteFailed => return step.fail(maker, "failed to write file '{f}{s}': {t}", .{
                cache_root, crash_name, out_w.err.?,
            }),
        };

        return step.fail(maker, "test '{s}' {f}; input saved to '{f}{s}'", .{
            run.fuzz_tests.items[header.test_i],
            fmtTerm(term),
            cache_root,
            crash_name,
        });
    }

    fn instanceBroadcast(f: *FuzzTestRunner, from_id: u32, bytes: []const u8) !void {
        assert(f.instances.len > 1);
        assert(f.broadcast_undelivered == 0); // no other broadcast is progress
        assert(f.broadcast.items.len == 0);
        assert(from_id < f.instances.len);

        const maker = f.ctx.fuzz.maker;
        const gpa = maker.gpa;

        var out_header: OutHeader = .{
            .tag = .new_fuzz_input,
            .bytes_len = @intCast(bytes.len),
        };
        if (std.builtin.Endian.native != .little) {
            std.mem.byteSwapAllFields(OutHeader, &out_header);
        }
        try f.broadcast.ensureTotalCapacity(gpa, @sizeOf(OutHeader) + bytes.len);
        f.broadcast.appendSliceAssumeCapacity(@ptrCast(&out_header));
        f.broadcast.appendSliceAssumeCapacity(bytes);

        f.broadcast_undelivered = @intCast(f.instances.len - 1);
        for (0.., f.instances) |to_id, *instance| {
            if (to_id == from_id) continue;
            instance.broadcast_written = 0;
            f.addStdinWrite(@intCast(to_id));
        }
    }

    fn broadcastComplete(f: *FuzzTestRunner) !void {
        assert(f.instances.len > 1);
        assert(f.broadcast_undelivered == 0);
        f.broadcast.clearRetainingCapacity();

        const pending = &f.pending_broadcasts;
        if (pending.items.len != 0) {
            // Another broadcast is pending; copy it over to `broadcast`

            const footer_len = @sizeOf(PendingBroadcastFooter);
            const footer_bytes = pending.items[pending.items.len - footer_len ..];
            const footer: *align(1) PendingBroadcastFooter = @ptrCast(footer_bytes);
            pending.items.len -= footer_len;

            const body = pending.items[pending.items.len - footer.body_len ..];
            try f.instanceBroadcast(footer.from_id, body);
            pending.items.len -= body.len;
        }
    }

    fn mergedStderr(f: *FuzzTestRunner, gpa: Allocator) Allocator.Error![]const u8 {
        // Collect any available stderr
        while (f.batch.next()) |completion| {
            if (completion.index % 3 != 2) continue;
            const len = completion.result.file_read_streaming catch continue;
            f.instances[completion.index / 3].stderr.items.len += len;
        }

        var stderr_len: usize = 0;
        for (f.instances) |*instance| stderr_len += instance.stderr.items.len;
        const stderr = try gpa.alloc(u8, stderr_len);

        stderr_len = 0;
        for (f.instances) |*instance| {
            @memcpy(stderr[stderr_len..][0..instance.stderr.items.len], instance.stderr.items);
            stderr_len += instance.stderr.items.len;
        }
        return stderr;
    }
}