diff --git a/lib/compiler/Maker.zig b/lib/compiler/Maker.zig index 2339de5ff1e3612029d7be22480e212e0592f998..e33605d5dc9a3dac4359f5413a24c3feba7de76a 100644 --- a/lib/compiler/Maker.zig +++ b/lib/compiler/Maker.zig @@ -11,6 +11,7 @@ const File = std.Io.File; const Io = std.Io; const Dir = std.Io.Dir; const Path = std.Build.Cache.Path; +const Reader = std.Io.Reader; const Writer = std.Io.Writer; const assert = std.debug.assert; const fatal = std.process.fatal; @@ -19,6 +20,8 @@ const log = std.log; const mem = std.mem; const process = std.process; const Color = std.zig.Color; +const Client = std.zig.Client; +const Server = std.zig.Server; const EnvVar = std.zig.EnvVar; const default_local_zig_cache_basename = std.zig.default_local_zig_cache_basename; const stringToEnum = std.meta.stringToEnum; @@ -51,10 +54,14 @@ max_rss_mutex: Io.Mutex, skip_oom_steps: bool, unit_test_timeout_ns: ?u64, watch: bool, +protocol_server: ?*AvoidableServer, +protocol_server_mutex: Io.Mutex, web_server: ?*AvoidableWebServer, /// Allocated into `gpa`. memory_blocked_steps: std.ArrayList(Configuration.Step.Index), /// Allocated into `gpa`. +initial_steps: std.array_hash_map.Auto(Configuration.Step.Index, void), +/// Allocated into `gpa`. step_stack: std.array_hash_map.Auto(Configuration.Step.Index, void), pkg_config: PkgConfig, @@ -67,6 +74,7 @@ var stdio_buffer_allocation: [256]u8 = undefined; var stdout_writer_allocation: Io.File.Writer = undefined; var debug_maker_leaks: bool = false; +const AvoidableServer = if (builtin.single_threaded) void else Server; const AvoidableWebServer = if (builtin.single_threaded) void else WebServer; const is_debug_mode = builtin.mode == .debug; @@ -216,6 +224,7 @@ pub fn main(init: process.Init.Minimal) !void { var watch = false; var fuzz: ?Fuzz.Mode = null; var debounce_interval_ms: u16 = 50; + var listen: bool = false; var webui_listen: ?Io.net.IpAddress = null; var debug_pkg_config = false; var run_args: ?[]const []const u8 = null; @@ -422,6 +431,8 @@ pub fn main(init: process.Init.Minimal) !void { next_arg, err, }); }; + } else if (mem.eql(u8, arg, "--listen=-")) { + listen = true; } else if (mem.eql(u8, arg, "--webui")) { if (webui_listen == null) webui_listen = .{ .ip6 = .loopback(0) }; } else if (mem.startsWith(u8, arg, "--webui=")) { @@ -559,7 +570,7 @@ pub fn main(init: process.Init.Minimal) !void { } const early_exit_mode = fetch_only or help_menu or steps_menu or print_configuration != .none; - const server_mode = !early_exit_mode and (watch or webui_listen != null or fuzz != null); + const server_mode = !early_exit_mode and (watch or webui_listen != null or fuzz != null or listen); process.raiseFileDescriptorLimit(); @@ -667,6 +678,25 @@ pub fn main(init: process.Init.Minimal) !void { break :ws &web_server_allocation; } else null; + var stdin_buffer: [256]u8 = undefined; + var stdout_buffer: [256]u8 = undefined; + var stdin_reader = Io.File.stdin().reader(io, &stdin_buffer); + var stdout_writer = Io.File.stdout().writer(io, &stdout_buffer); + + var protocol_server_allocation: AvoidableServer = undefined; + const protocol_server: ?*AvoidableServer = if (listen) s: { + if (builtin.single_threaded) fatal("--listen is not yet supported on single-threaded hosts", .{}); + if (watch) fatal("using '--watch' and '--listen' together is not supported", .{}); + if (fuzz != null) fatal("using '--fuzz' and '--listen' together is not supported", .{}); + if (step_names.items.len > 0) fatal("build steps must be provided over the protocol instead of using CLI arguments", .{}); + protocol_server_allocation = .{ + .in = &stdin_reader.interface, + .out = &stdout_writer.interface, + }; + try serveBSPHandshake(&protocol_server_allocation); + break :s &protocol_server_allocation; + } else null; + while (true) { // If this fails, we can still start the server and wait for user // to request a rebuild. If it returns error.FailedButCacheIntact @@ -737,16 +767,25 @@ pub fn main(init: process.Init.Minimal) !void { .watch = watch, .web_server = web_server, + .protocol_server = protocol_server, + .protocol_server_mutex = .init, .memory_blocked_steps = .empty, + .initial_steps = .empty, .step_stack = .empty, .pkg_config = .{ .debug = debug_pkg_config }, .error_style = error_style, .multiline_errors = multiline_errors, - .summary = summary orelse if (watch or webui_listen != null) .new else .failures, + .summary = summary orelse if (listen) + .none + else if (watch or webui_listen != null) + .new + else + .failures, }; defer { maker.memory_blocked_steps.deinit(gpa); + maker.initial_steps.deinit(gpa); maker.step_stack.deinit(gpa); } @@ -755,7 +794,91 @@ pub fn main(init: process.Init.Minimal) !void { maker.max_rss_is_default = true; } - maker.prepare(step_names.items) catch |err| switch (err) { + if (protocol_server) |s| { + try s.serveStringMessage(.bsp_configuration, try arena.print("{f}", .{scanned_config.path})); + + var w: ?Watch = null; + + const Event = union(enum) { + message: Reader.Error!Client.Message.Header, + fs_event: if (Watch.have_impl) @typeInfo(@TypeOf(Watch.wait)).@"fn".return_type.? else noreturn, + }; + + var select_buffer: [2]Event = undefined; + var select: Io.Select(Event) = .init(io, &select_buffer); + defer select.cancelDiscard(); + + try select.concurrent(.message, Server.receiveMessage, .{s}); + + var in_debounce = false; + loop: switch (try select.await()) { + .message => |payload| { + const header: Client.Message.Header = try payload; + switch (header.tag) { + .exit => { + cleanExit(io, &scanned_config); + process.exit(0); + }, + .bsp_build_steps => { + // Cancel existing file watching + select.cancelDiscard(); + in_debounce = false; + + const body = try s.in.takeStruct(Client.Message.BuildSteps, .little); + const steps = try s.in.readSliceEndianAlloc(gpa, Configuration.Step.Index, body.step_count, .little); + defer gpa.free(steps); + if (body.flags.watch and !Watch.have_impl) fatal("file watching is unavailable", .{}); + + try select.concurrent(.message, Server.receiveMessage, .{s}); + + maker.watch = body.flags.watch; + maker.prepare(steps) catch |err| switch (err) { + error.DependencyLoopDetected, error.InsufficientMemory => { + // TODO handle DependencyLoopDetected as error.FailedButCacheIntact + // and handle InsufficientMemory as error.AlreadyReported + _ = io.lockStderr(&.{}, graph.stderr_mode) catch {}; + process.exit(1); + }, + else => |e| return e, + }; + + try maker.makeSteps(main_progress_node, null); + + if (body.flags.watch) { + if (!Watch.have_impl) unreachable; + if (w == null) w = try .init(&maker); + + try w.?.update(maker.step_stack.keys()); + try select.concurrent(.fs_event, Watch.wait, .{ &w.?, if (in_debounce) .{ .ms = debounce_interval_ms } else .none }); + } + + continue :loop try select.await(); + }, + else => fatal("unsupported message: {t}", .{header.tag}), + } + }, + .fs_event => |payload| { + if (!Watch.have_impl) unreachable; + switch (try payload) { + .timeout => { + assert(in_debounce); + markFailedStepsDirty(&maker); + try maker.makeSteps(main_progress_node, null); + in_debounce = false; + }, + .dirty => in_debounce = true, + .clean => {}, + } + try select.concurrent(.fs_event, Watch.wait, .{ &w.?, if (in_debounce) .{ .ms = debounce_interval_ms } else .none }); + continue :loop try select.await(); + }, + } + } + + const initial_steps = try maker.resolveTopLevelSteps(step_names.items); + defer gpa.free(initial_steps); + + maker.prepare(initial_steps) catch |err| switch (err) { error.DependencyLoopDetected, error.InsufficientMemory => { // TODO handle DependencyLoopDetected as error.FailedButCacheIntact // and handle InsufficientMemory as error.AlreadyReported @@ -780,18 +903,7 @@ pub fn main(init: process.Init.Minimal) !void { error.WriteFailed => return stderr.file_writer.err.?, }; }) { - if (web_server) |ws| ws.startBuild(); - - try maker.makeStepNames(step_names.items, main_progress_node, fuzz); - - if (web_server) |ws| { - if (fuzz) |mode| if (mode != .forever) fatal( - "error: limited fuzzing is not implemented yet for --webui", - .{}, - ); - - ws.finishBuild(.{ .fuzz = fuzz != null }); - } + try maker.makeSteps(main_progress_node, fuzz); if (web_server) |ws| { const c = &scanned_config.configuration; @@ -856,6 +968,9 @@ pub fn main(init: process.Init.Minimal) !void { _ = io.lockStderr(&.{}, graph.stderr_mode) catch {}; process.exit(1); } + if (protocol_server != null) { + fatal("(zig build system) TODO send error messages to client when build.zig compilation fails", .{}); + } if (watch and can_fs_watch) { fatal("(zig build system) TODO set up fs watching even when build.zig compilation fails", .{}); } else { @@ -2022,11 +2137,37 @@ pub fn stepByIndex(maker: *const Maker, i: Configuration.Step.Index) *Step { return &maker.steps[@backingInt(i)]; } -fn prepare(maker: *Maker, step_names: []const []const u8) !void { +fn resolveTopLevelSteps(maker: *Maker, step_names: []const []const u8) ![]const Configuration.Step.Index { + const gpa = maker.gpa; + const c = &maker.scanned_config.configuration; + + if (step_names.len == 0) { + return try gpa.dupe(Configuration.Step.Index, &.{c.default_step}); + } + + var result: std.array_hash_map.Auto(Configuration.Step.Index, void) = .empty; + defer result.deinit(gpa); + + try result.ensureTotalCapacity(gpa, step_names.len); + + for (0..step_names.len) |i| { + const step_name = step_names[step_names.len - i - 1]; + const s = maker.scanned_config.top_level_steps.get(step_name) orelse { + log.info("to list available steps: zig build -l", .{}); + fatal("no such step: {s}", .{step_name}); + }; + result.putAssumeCapacity(s, {}); + } + + return try gpa.dupe(Configuration.Step.Index, result.keys()); +} + +fn prepare(maker: *Maker, step_indices: []const Configuration.Step.Index) !void { const gpa = maker.gpa; const graph = maker.graph; const arena = graph.arena; const seed: u32 = graph.random_seed; + const initial_steps = &maker.initial_steps; const step_stack = &maker.step_stack; const c = &maker.scanned_config.configuration; @@ -2035,18 +2176,15 @@ fn prepare(maker: *Maker, step_names: []const []const u8) !void { step.* = .{ .extended = .init(step_index.ptr(c).flags(c).tag) }; } - if (step_names.len == 0) { - try step_stack.put(gpa, c.default_step, {}); - } else { - try step_stack.ensureUnusedCapacity(gpa, step_names.len); - for (0..step_names.len) |i| { - const step_name = step_names[step_names.len - i - 1]; - const s = maker.scanned_config.top_level_steps.get(step_name) orelse { - log.info("to list available steps: zig build -l", .{}); - fatal("no such step: {s}", .{step_name}); - }; - step_stack.putAssumeCapacity(s, {}); - } + try initial_steps.ensureUnusedCapacity(gpa, step_indices.len); + try step_stack.ensureUnusedCapacity(gpa, step_indices.len); + + initial_steps.clearRetainingCapacity(); + step_stack.clearRetainingCapacity(); + + for (step_indices) |step| { + initial_steps.putAssumeCapacity(step, {}); + step_stack.putAssumeCapacity(step, {}); } const starting_steps = try arena.dupe(Configuration.Step.Index, step_stack.keys()); @@ -2095,9 +2233,8 @@ fn prepare(maker: *Maker, step_names: []const []const u8) !void { } } -fn makeStepNames( +fn makeSteps( maker: *Maker, - step_names: []const []const u8, parent_progress_node: std.Progress.Node, fuzz: ?Fuzz.Mode, ) !void { @@ -2108,6 +2245,12 @@ fn makeStepNames( const top_level_steps = &maker.scanned_config.top_level_steps; const c = &maker.scanned_config.configuration; + if (maker.web_server) |ws| ws.startBuild(); + + if (maker.protocol_server) |s| { + try s.serveBodylessMessage(.bsp_build_started); + } + { // Collect the initial set of tasks (those with no outstanding dependencies) into a buffer, // then spawn them. The buffer is so that we don't race with `makeStep` and end up thinking @@ -2133,6 +2276,19 @@ fn makeStepNames( try group.await(io); } + if (maker.web_server) |ws| { + if (fuzz) |mode| if (mode != .forever) fatal( + "error: limited fuzzing is not implemented yet for --webui", + .{}, + ); + + ws.finishBuild(.{ .fuzz = fuzz != null }); + } + + if (maker.protocol_server) |s| { + try s.serveBodylessMessage(.bsp_build_completed); + } + assert(maker.memory_blocked_steps.items.len == 0); var test_pass_count: usize = 0; @@ -2285,7 +2441,7 @@ fn makeStepNames( defer step_stack_copy.deinit(gpa); var print_node: PrintNode = .{ .parent = null }; - if (step_names.len == 0) { + if (maker.initial_steps.count() == 0) { print_node.last = true; printTreeStep(maker, c.default_step, t, &print_node, &step_stack_copy) catch |err| switch (err) { error.Canceled => |e| return e, @@ -2293,10 +2449,10 @@ fn makeStepNames( }; } else { const last_index = if (maker.summary == .all) top_level_steps.count() else blk: { - var i: usize = step_names.len; + var i: usize = maker.initial_steps.count(); while (i > 0) { i -= 1; - const step_index = top_level_steps.get(step_names[i]).?; + const step_index = maker.initial_steps.keys()[i]; const step = maker.stepByIndex(step_index); const found = switch (maker.summary) { .all, .line, .none => unreachable, @@ -2307,8 +2463,7 @@ fn makeStepNames( } break :blk top_level_steps.count(); }; - for (step_names, 0..) |step_name, i| { - const step_index = top_level_steps.get(step_name).?; + for (maker.initial_steps.keys(), 0..) |step_index, i| { print_node.last = i + 1 == last_index; printTreeStep(maker, step_index, t, &print_node, &step_stack_copy) catch |err| switch (err) { error.Canceled => |e| return e, @@ -2319,7 +2474,7 @@ fn makeStepNames( w.writeByte('\n') catch {}; } - if (maker.watch or maker.web_server != null) return; + if (maker.watch or maker.web_server != null or maker.protocol_server != null) return; const code: u8 = code: { if (failure_count == 0) break :code 0; // success @@ -2394,6 +2549,15 @@ fn makeStep( defer step_prog_node.end(); if (maker.web_server) |ws| ws.updateStepStatus(step_index, .wip); + if (maker.protocol_server) |s| { + maker.protocol_server_mutex.lockUncancelable(io); + defer maker.protocol_server_mutex.unlock(io); + + s.serveU32Message( + .bsp_step_started, + @backingInt(step_index), + ) catch @panic("TODO propagate error when failing to send protocol message"); + } const new_state: Step.State = for (deps) |dep_index| { const dep_make_step = maker.stepByIndex(dep_index); @@ -2419,7 +2583,7 @@ fn makeStep( @atomicStore(Step.State, &make_step.state, new_state, .monotonic); - switch (new_state) { + const success = switch (new_state) { .precheck_unstarted => unreachable, .precheck_started => unreachable, .precheck_done => unreachable, @@ -2427,17 +2591,37 @@ fn makeStep( .failure, .dependency_failure, .skipped_oom, - => { - if (maker.web_server) |ws| ws.updateStepStatus(step_index, .failure); - std.Progress.setStatus(.failure_working); - }, + => false, .success, .skipped, - => { - if (maker.web_server) |ws| ws.updateStepStatus(step_index, .success); - }, + => true, + }; + + if (maker.web_server) |ws| { + ws.updateStepStatus(step_index, if (success) .success else .failure); } + if (maker.protocol_server != null) { + maker.protocol_server_mutex.lockUncancelable(io); + defer maker.protocol_server_mutex.unlock(io); + + const status: Server.Message.BuildStepCompleted.Status = switch (new_state) { + .precheck_unstarted => unreachable, + .precheck_started => unreachable, + .precheck_done => unreachable, + .success => .success, + .failure, .dependency_failure => .failure, + .skipped => .skipped, + .skipped_oom => .skipped_oom, + }; + serveBuildStepCompleted( + maker, + step_index, + status, + ) catch |err| std.debug.panic("TODO propagate error when failing to send protocol message: {t}", .{err}); + } + + if (!success) std.Progress.setStatus(.failure_working); } // No matter the result, we want to display error/warning messages. @@ -2992,6 +3176,50 @@ fn cleanTmpFiles(maker: *Maker, steps: []const Configuration.Step.Index) void { } } +fn serveBSPHandshake(s: *const std.zig.Server) !void { + const handshake_header: Server.Message.Handshake = .{ + .version = Server.build_system_version, + .flags = .{ + .file_system_watch_supported = Watch.have_impl, + }, + }; + try s.serveMessageHeader(.{ + .tag = .bsp_handshake, + .bytes_len = @sizeOf(Server.Message.Handshake), + }); + try s.out.writeStruct(handshake_header, .little); + try s.out.flush(); +} + +fn serveBuildStepCompleted( + maker: *Maker, + step_index: Configuration.Step.Index, + status: Server.Message.BuildStepCompleted.Status, +) !void { + const s: *Server = maker.protocol_server.?; + const step = maker.stepByIndex(step_index); + const error_bundle = step.result_error_bundle; + + const body: Server.Message.BuildStepCompleted = .{ + .step_index = step_index, + .status = status, + .error_bundle = .{ + .extra_len = @intCast(error_bundle.extra.len), + .string_bytes_len = @intCast(error_bundle.string_bytes.len), + }, + }; + const eb_bytes_len = @sizeOf(u32) * error_bundle.extra.len + error_bundle.string_bytes.len; + const bytes_len = @sizeOf(Server.Message.BuildStepCompleted) + eb_bytes_len; + try s.serveMessageHeader(.{ + .tag = .bsp_step_completed, + .bytes_len = @intCast(bytes_len), + }); + try s.out.writeStruct(body, .little); + try s.out.writeSliceEndian(u32, error_bundle.extra, .little); + try s.out.writeAll(error_bundle.string_bytes); + try s.out.flush(); +} + fn initStdoutWriter(io: Io) *Writer { stdout_writer_allocation = Io.File.stdout().writerStreaming(io, &stdio_buffer_allocation); return &stdout_writer_allocation.interface; diff --git a/lib/compiler/Maker/Step.zig b/lib/compiler/Maker/Step.zig index 812613928eaa9134dd249e4fc2ece38065cad01c..b8c5992cce243ed5b7f5fb0cdfc1b6b9d92f7fdb 100644 --- a/lib/compiler/Maker/Step.zig +++ b/lib/compiler/Maker/Step.zig @@ -561,24 +561,26 @@ fn zigProcessUpdate(step_index: Configuration.Step.Index, maker: *Maker, zp: *Zi var result: ?Path = null; var eos_err: error{EndOfStream}!void = {}; - const stdout = zp.multi_reader.fileReader(0); + var client: std.zig.Client = .{ + .in = zp.multi_reader.reader(0), + .out = undefined, + }; while (true) { - const Header = std.zig.Server.Message.Header; - const header = stdout.interface.takeStruct(Header, .little) catch |err| switch (err) { - error.EndOfStream => break, - error.ReadFailed => return stdout.err.?, - }; - const body = stdout.interface.take(header.bytes_len) catch |err| switch (err) { + const header = client.receiveMessageWithMultiReader(&zp.multi_reader, .none) catch |err| switch (err) { + error.Timeout => unreachable, error.EndOfStream => |e| { + if (client.in.bufferedLen() == 0) break; // Better to report the crash with stderr below, but we set // this in case the child exits successfully while violating // this protocol. eos_err = e; break; }, - error.ReadFailed => return stdout.err.?, + else => |e| return e, }; + const body = client.in.take(header.bytes_len) catch unreachable; + switch (header.tag) { .zig_version => { if (!std.mem.eql(u8, builtin.zig_version_string, body)) { diff --git a/lib/compiler/Maker/Step/Run.zig b/lib/compiler/Maker/Step/Run.zig index 0ac652096de65af81a50d25fb94532c805d0205c..56fea332016567473fb6a3fc6c385e92b6c28f91 100644 --- a/lib/compiler/Maker/Step/Run.zig +++ b/lib/compiler/Maker/Step/Run.zig @@ -384,13 +384,23 @@ fn waitZigTest( var sub_prog_node: ?std.Progress.Node = null; defer if (sub_prog_node) |n| n.end(); + const stdout = multi_reader.reader(0); + const stderr = multi_reader.reader(1); + + var stdin_writer = child.stdin.?.writerStreaming(io, &.{}); + + var client: std.zig.Client = .{ + .in = stdout, + .out = &stdin_writer.interface, + }; + if (opt_metadata.*) |*md| { // Previous unit test process died or was killed; we're continuing where it left off - requestNextTest(io, child.stdin.?, md, &sub_prog_node) catch |err| return .{ .write_failed = err }; + requestNextTest(&client, md, &sub_prog_node) catch |err| return .{ .write_failed = err }; } else { // Running unit tests normally run.fuzz_tests.clearRetainingCapacity(); - sendMessage(io, child.stdin.?, .query_test_metadata) catch |err| return .{ .write_failed = err }; + client.serveBodylessMessage(.query_test_metadata) catch |err| return .{ .write_failed = err }; } var active_test_index: ?u32 = null; @@ -410,10 +420,6 @@ fn waitZigTest( .raw = .fromNanoseconds(ns), } else null; - const stdout = multi_reader.reader(0); - const stderr = multi_reader.reader(1); - const Header = std.zig.Server.Message.Header; - while (true) { const timeout: Io.Timeout = t: { const opt_duration = if (active_test_index == null) response_timeout else test_timeout; @@ -421,46 +427,20 @@ fn waitZigTest( break :t .{ .deadline = last_update.addDuration(duration) }; }; - // This block is exited when `stdout` contains enough bytes for a `Header`. - header_ready: { - if (stdout.buffered().len >= @sizeOf(Header)) { - // We already have one, no need to poll! - break :header_ready; - } - - multi_reader.fill(64, timeout) catch |err| switch (err) { - error.Timeout => return .{ .timeout = .{ - .active_test_index = active_test_index, - .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), - } }, - error.EndOfStream => return .{ .no_poll = .{ - .active_test_index = active_test_index, - .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), - } }, - else => |e| return e, - }; - - continue; - } - // There is definitely a header available now -- read it. - const header = stdout.takeStruct(Header, .little) catch unreachable; - - while (stdout.buffered().len < header.bytes_len) { - multi_reader.fill(64, timeout) catch |err| switch (err) { - error.Timeout => return .{ .timeout = .{ - .active_test_index = active_test_index, - .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), - } }, - error.EndOfStream => return .{ .no_poll = .{ - .active_test_index = active_test_index, - .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), - } }, - else => |e| return e, - }; - } - - const body = stdout.take(header.bytes_len) catch unreachable; + const header = client.receiveMessageWithMultiReader(multi_reader, timeout) catch |err| switch (err) { + error.Timeout => return .{ .timeout = .{ + .active_test_index = active_test_index, + .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), + } }, + error.EndOfStream => return .{ .no_poll = .{ + .active_test_index = active_test_index, + .ns_elapsed = @intCast(last_update.untilNow(io).raw.nanoseconds), + } }, + else => |e| return e, + }; + const body = client.in.take(header.bytes_len) catch unreachable; var body_r: std.Io.Reader = .fixed(body); + switch (header.tag) { .zig_version => { if (!std.mem.eql(u8, builtin.zig_version_string, body)) return step.fail( @@ -500,7 +480,7 @@ fn waitZigTest( active_test_index = null; last_update = .now(io, .awake); - requestNextTest(io, child.stdin.?, &opt_metadata.*.?, &sub_prog_node) catch |err| return .{ .write_failed = err }; + requestNextTest(&client, &opt_metadata.*.?, &sub_prog_node) catch |err| return .{ .write_failed = err }; }, .test_started => { active_test_index = opt_metadata.*.?.next_index - 1; @@ -551,7 +531,7 @@ fn waitZigTest( md.ns_per_test[tr_hdr.index] = @intCast(last_update.durationTo(now).raw.nanoseconds); last_update = now; - requestNextTest(io, child.stdin.?, md, &sub_prog_node) catch |err| return .{ .write_failed = err }; + requestNextTest(&client, md, &sub_prog_node) catch |err| return .{ .write_failed = err }; }, else => {}, // ignore other messages } @@ -697,17 +677,18 @@ const FuzzTestRunner = struct { for (0.., f.instances) |id, *instance| { const id32: u32 = @intCast(id); + var writer = instance.child.stdin.?.writerStreaming(io, &.{}); + const client: std.zig.Client = .{ + .in = undefined, + .out = &writer.interface, + }; (switch (f.ctx.fuzz.mode) { - .forever => sendRunFuzzTestMessage( - io, - instance.child.stdin.?, + .forever => client.serveRunFuzzTestMessage( run.fuzz_tests.items, .forever, id32, ), - .limit => |limit| sendRunFuzzTestMessage( - io, - instance.child.stdin.?, + .limit => |limit| client.serveRunFuzzTestMessage( run.fuzz_tests.items, .iterations, limit.amount, @@ -1315,7 +1296,7 @@ pub const CachedTestMetadata = struct { } }; -fn requestNextTest(io: Io, in: Io.File, metadata: *TestMetadata, sub_prog_node: *?std.Progress.Node) !void { +fn requestNextTest(client: *std.zig.Client, metadata: *TestMetadata, sub_prog_node: *?std.Progress.Node) !void { while (metadata.next_index < metadata.names.len) { const i = metadata.next_index; metadata.next_index += 1; @@ -1326,76 +1307,11 @@ fn requestNextTest(io: Io, in: Io.File, metadata: *TestMetadata, sub_prog_node: if (sub_prog_node.*) |n| n.end(); sub_prog_node.* = metadata.prog_node.start(name, 0); - try sendRunTestMessage(io, in, .run_test, i); + try client.serveRunTest(i); return; } else { metadata.next_index = std.math.maxInt(u32); // indicate that all tests are done - try sendMessage(io, in, .exit); - } -} - -fn sendMessage(io: Io, file: Io.File, tag: std.zig.Client.Message.Tag) !void { - const header: std.zig.Client.Message.Header = .{ - .tag = tag, - .bytes_len = 0, - }; - var w = file.writerStreaming(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; -} - -fn sendRunTestMessage(io: Io, file: Io.File, tag: std.zig.Client.Message.Tag, index: u32) !void { - const header: std.zig.Client.Message.Header = .{ - .tag = tag, - .bytes_len = 4, - }; - var w = file.writerStreaming(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - w.interface.writeInt(u32, index, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; -} - -fn sendRunFuzzTestMessage( - io: Io, - file: Io.File, - test_names: []const []const u8, - kind: std.Build.abi.fuzz.LimitKind, - amount_or_instance: u64, -) !void { - const header: std.zig.Client.Message.Header = .{ - .tag = .start_fuzzing, - .bytes_len = 1 + 8 + 4 + count: { - var c: u32 = @intCast(test_names.len * 4); - for (test_names) |name| { - c += @intCast(name.len); - } - break :count c; - }, - }; - var w = file.writerStreaming(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - w.interface.writeByte(@backingInt(kind)) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - w.interface.writeInt(u64, amount_or_instance, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - w.interface.writeInt(u32, @intCast(test_names.len), .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - for (test_names) |test_name| { - w.interface.writeInt(u32, @intCast(test_name.len), .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; - w.interface.writeAll(test_name) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; + try client.serveBodylessMessage(.exit); } } @@ -2285,25 +2201,35 @@ fn spawnChildAndCollect( assert(conf_run.flags.stdio != .inherit); break :s .pipe; } else switch (conf_run.flags.stdio) { - .infer_from_args => if (has_side_effects) .inherit else .ignore, + .infer_from_args => if (maker.protocol_server == null and has_side_effects) .inherit else .ignore, .inherit => .inherit, .check => .ignore, .zig_test => .pipe, }, .stdout = if (conf_run.captured_stdout.value != null) .pipe else switch (conf_run.flags.stdio) { - .infer_from_args => if (has_side_effects) .inherit else .ignore, + .infer_from_args => if (maker.protocol_server == null and has_side_effects) .inherit else .ignore, .inherit => .inherit, .check => if (checksContainStdout(&conf_run)) .pipe else .ignore, .zig_test => .pipe, }, .stderr = if (conf_run.captured_stderr.value != null) .pipe else switch (conf_run.flags.stdio) { - .infer_from_args => if (has_side_effects) .inherit else .pipe, - .inherit => .inherit, + .infer_from_args => if (maker.protocol_server == null and has_side_effects) .inherit else .pipe, + .inherit => if (maker.protocol_server == null) .inherit else .pipe, .check => .pipe, .zig_test => .pipe, }, }; + if (maker.protocol_server != null) { + if (spawn_options.stdin == .inherit) { + return step.fail(maker, "Cannot inherit stdin when running through over the build system protocol", .{}); + } + if (spawn_options.stdout == .inherit) { + return step.fail(maker, "Cannot inherit stdout when running through over the build system protocol", .{}); + } + assert(spawn_options.stderr != .inherit); + } + if (conf_run.flags.stdio == .zig_test) { try setColorEnvironmentVariables(&conf_run, environ_map, graph.stderr_mode.?); const started: Io.Clock.Timestamp = .now(io, .awake); diff --git a/lib/compiler/objcopy.zig b/lib/compiler/objcopy.zig index 857299a60e16f410c3eb54ca6be8382c33b99567..3e6967779598067003281d655f07c519def94340 100644 --- a/lib/compiler/objcopy.zig +++ b/lib/compiler/objcopy.zig @@ -214,11 +214,11 @@ fn cmdObjCopy(arena: Allocator, io: Io, args: []const []const u8) !void { if (listen) { var stdin_reader = Io.File.stdin().reader(io, &stdin_buffer); var stdout_writer = Io.File.stdout().writer(io, &stdout_buffer); - var server = try Server.init(.{ + var server: Server = .{ .in = &stdin_reader.interface, .out = &stdout_writer.interface, - .zig_version = builtin.zig_version_string, - }); + }; + try server.serveStringMessage(.zig_version, builtin.zig_version_string); var seen_update = false; while (true) { diff --git a/lib/compiler/std-docs.zig b/lib/compiler/std-docs.zig index f53123ae925a7fb4d997e1b8c1a5463624077780..0ed2c5bf186c2c91730b1aa4aa1755babdab0be2 100644 --- a/lib/compiler/std-docs.zig +++ b/lib/compiler/std-docs.zig @@ -346,29 +346,39 @@ fn buildWasmBinary( multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ child.stdout.?, child.stderr.? }); defer multi_reader.deinit(); - try sendMessage(io, child.stdin.?, .update); - try sendMessage(io, child.stdin.?, .exit); + const stdout = multi_reader.reader(0); + + var stdin_buffer: [256]u8 = undefined; + var stdin_writer = child.stdin.?.writerStreaming(io, &stdin_buffer); + + var client: std.zig.Client = .{ + .in = stdout, + .out = &stdin_writer.interface, + }; + + try client.serveMessageHeader(.{ .tag = .update, .bytes_len = 0 }); + try client.serveMessageHeader(.{ .tag = .exit, .bytes_len = 0 }); + try client.out.flush(); var result: ?Cache.Path = null; var result_error_bundle = std.zig.ErrorBundle.empty; - const stdout = multi_reader.fileReader(0); - const MessageHeader = std.zig.Server.Message.Header; - var eos_err: error{EndOfStream}!void = {}; while (true) { - const header = stdout.interface.takeStruct(MessageHeader, .little) catch |err| switch (err) { - error.EndOfStream => break, - error.ReadFailed => return stdout.err.?, - }; - const body = stdout.interface.take(header.bytes_len) catch |err| switch (err) { + const header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Timeout => unreachable, error.EndOfStream => |e| { + if (client.in.bufferedLen() == 0) break; + // Better to report the crash with stderr below, but we set + // this in case the child exits successfully while violating + // this protocol. eos_err = e; break; }, - error.ReadFailed => return stdout.err.?, + else => |e| return e, }; + const body = client.in.take(header.bytes_len) catch unreachable; switch (header.tag) { .zig_version => { @@ -435,17 +445,6 @@ fn buildWasmBinary( }; } -fn sendMessage(io: Io, file: Io.File, tag: std.zig.Client.Message.Tag) !void { - const header: std.zig.Client.Message.Header = .{ - .tag = tag, - .bytes_len = 0, - }; - var w = file.writer(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { - error.WriteFailed => return w.err.?, - }; -} - fn openBrowserTab(io: Io, url: []const u8) !void { // Until https://github.com/ziglang/zig/issues/19205 is implemented, we // spawn and then leak a concurrent task for this child process. diff --git a/lib/compiler/test_runner.zig b/lib/compiler/test_runner.zig index 2827d32c496e12c3a2bf7c481b54afa50eb477fe..0fd5afad694b6aa417fb7cdfcf7305614574d883 100644 --- a/lib/compiler/test_runner.zig +++ b/lib/compiler/test_runner.zig @@ -78,11 +78,11 @@ fn mainServer(init: std.process.Init.Minimal) !void { @disableInstrumentation(); stdin_reader = .initStreaming(.stdin(), runner_threaded_io, &stdin_buffer); stdout_writer = .initStreaming(.stdout(), runner_threaded_io, &stdout_buffer); - var server = try std.zig.Server.init(.{ + var server: std.zig.Server = .{ .in = &stdin_reader.interface, .out = &stdout_writer.interface, - .zig_version = builtin.zig_version_string, - }); + }; + try server.serveStringMessage(.zig_version, builtin.zig_version_string); while (true) { const hdr = try server.receiveMessage(); diff --git a/lib/std/Io/Reader.zig b/lib/std/Io/Reader.zig index 68cae96c836d7cab147c327afd06588c118d784c..966af7879c2488a2b8447f6398aa036a79aba263 100644 --- a/lib/std/Io/Reader.zig +++ b/lib/std/Io/Reader.zig @@ -718,7 +718,7 @@ pub inline fn readSliceEndian( endian: std.builtin.Endian, ) Error!void { try readSliceAll(r, @ptrCast(buffer)); - if (native_endian != endian) for (buffer) |*elem| std.mem.byteSwapAllFields(Elem, elem); + if (native_endian != endian) std.mem.byteSwapAllElements(Elem, buffer); } pub const ReadAllocError = Error || Allocator.Error; @@ -734,8 +734,7 @@ pub inline fn readSliceEndianAlloc( ) ReadAllocError![]Elem { const dest = try allocator.alloc(Elem, len); errdefer allocator.free(dest); - try readSliceAll(r, @ptrCast(dest)); - if (native_endian != endian) for (dest) |*elem| std.mem.byteSwapAllFields(Elem, elem); + try r.readSliceEndian(Elem, dest, endian); return dest; } @@ -1227,8 +1226,7 @@ pub inline fn takeStruct(r: *Reader, comptime T: type, endian: std.builtin.Endia .auto => @compileError("ill-defined memory layout"), .@"extern" => { var res: T = undefined; - try r.readSliceAll(std.mem.asBytes(&res)); - if (native_endian != endian) std.mem.byteSwapAllFields(T, &res); + try r.readSliceEndian(T, (&res)[0..1], endian); return res; }, .@"packed" => { diff --git a/lib/std/mem.zig b/lib/std/mem.zig index 9529f73b93586014b6338eef2ecd780461b9cd37..46e8347edec959724ef407bfb1c3b6981b3e9142 100644 --- a/lib/std/mem.zig +++ b/lib/std/mem.zig @@ -2215,33 +2215,54 @@ test writeVarPackedInt { try testing.expectEqual(T{ .a = 1, .b = value, .c = 4 }, st); } -/// Swap the byte order of all the members of the fields of a struct -/// (Changing their endianness) -pub fn byteSwapAllFields(comptime S: type, ptr: *S) void { - byteSwapAllFieldsAligned(S, .of(S), ptr); +/// Deprecated: use `byteSwap` instead. +pub const byteSwapAllFields = byteSwap; + +/// Deprecated: use `byteSwapAligned` instead. +pub const byteSwapAllFieldsAligned = byteSwapAligned; + +/// Reverses the byte order. +/// Handles structs, unions, arrays, enums, floats, and integers recursively. +/// The order of extern struct fields and array elements remains unchanged and +/// will be byte swapped recursively. +/// Useful for converting between little-endian and big-endian representations. +pub fn byteSwap(comptime S: type, ptr: *S) void { + byteSwapAligned(S, .of(S), ptr); } -/// Swap the byte order of all the members of the fields of a struct -/// (Changing their endianness) -pub fn byteSwapAllFieldsAligned(comptime S: type, comptime a: Alignment, ptr: *align(a.toByteUnits()) S) void { +/// Reverses the byte order. +/// Handles structs, unions, arrays, enums, floats, and integers recursively. +/// The order of extern struct fields and array elements remains unchanged and +/// will be byte swapped recursively. +/// Useful for converting between little-endian and big-endian representations. +pub fn byteSwapAligned( + comptime S: type, + comptime a: Alignment, + ptr: *align(a.toByteUnits()) S, +) void { switch (@typeInfo(S)) { .@"struct" => |@"struct"| { if (@"struct".backing_integer) |Int| { ptr.* = @bitCast(@byteSwap(@as(Int, @bitCast(ptr.*)))); - } else inline for (@"struct".field_types, @"struct".field_names, @"struct".field_attrs) |f_type, f_name, f_attr| { - switch (@typeInfo(f_type)) { - .@"struct" => byteSwapAllFieldsAligned(f_type, .fromByteUnits(f_attr.@"align" orelse @alignOf(f_type)), &@field(ptr, f_name)), - .@"union", .array => byteSwapAllFieldsAligned(f_type, .fromByteUnits(f_attr.@"align" orelse @alignOf(f_type)), &@field(ptr, f_name)), - .@"enum" => { - @field(ptr, f_name) = @fromBackingInt(@intCast(@byteSwap(@backingInt(@field(ptr, f_name))))); - }, - .bool => {}, - .float => |float| { - @field(ptr, f_name) = @bitCast(@byteSwap(@as(@Int(.unsigned, float.bits), @bitCast(@field(ptr, f_name))))); - }, - else => { - @field(ptr, f_name) = @byteSwap(@field(ptr, f_name)); - }, + } else { + if (@"struct".layout != .@"extern") { + @compileError("byteSwapAligned expects a packed or extern struct"); + } + inline for (@"struct".field_types, @"struct".field_names, @"struct".field_attrs) |f_type, f_name, f_attr| { + switch (@typeInfo(f_type)) { + .@"struct" => byteSwapAligned(f_type, .fromByteUnits(f_attr.@"align" orelse @alignOf(f_type)), &@field(ptr, f_name)), + .@"union", .array => byteSwapAligned(f_type, .fromByteUnits(f_attr.@"align" orelse @alignOf(f_type)), &@field(ptr, f_name)), + .@"enum" => { + @field(ptr, f_name) = @fromBackingInt(@byteSwap(@backingInt(@field(ptr, f_name)))); + }, + .bool => {}, + .float => |float| { + @field(ptr, f_name) = @bitCast(@byteSwap(@as(@Int(.unsigned, float.bits), @bitCast(@field(ptr, f_name))))); + }, + else => { + @field(ptr, f_name) = @byteSwap(@field(ptr, f_name)); + }, + } } } }, @@ -2249,7 +2270,7 @@ pub fn byteSwapAllFieldsAligned(comptime S: type, comptime a: Alignment, ptr: *a ptr.* = @bitCast(@byteSwap(@as(Int, @bitCast(ptr.*)))); } else { if (@"union".layout != .@"extern") { - @compileError("byteSwapAllFields expects a packed or extern union"); + @compileError("byteSwapAligned expects a packed or extern union"); } const first_size = @bitSizeOf(@"union".field_types[0]); @@ -2266,13 +2287,21 @@ pub fn byteSwapAllFieldsAligned(comptime S: type, comptime a: Alignment, ptr: *a .array => |array| { byteSwapAllElements(array.child, ptr); }, + .@"enum" => { + ptr.* = @fromBackingInt(@byteSwap(@backingInt(ptr.*))); + }, + .bool => {}, + .float => |float| { + const int_repr: @Int(.unsigned, float.bits) = @bitCast(ptr.*); + ptr.* = @bitCast(@byteSwap(int_repr)); + }, else => { ptr.* = @byteSwap(ptr.*); }, } } -test byteSwapAllFields { +test byteSwap { const T = extern struct { f0: u8, f1: u16, @@ -2304,6 +2333,9 @@ test byteSwapAllFields { } align(4), f2: u32, }; + const E = enum(u32) { + _, + }; var s = T{ .f0 = 0x12, .f1 = 0x1234, @@ -2327,10 +2359,14 @@ test byteSwapAllFields { .f1 = .{ .f0 = 0x123456789ABCDEF0 }, .f2 = 0x87654321, }; - byteSwapAllFields(T, &s); - byteSwapAllFields(K, &k); - byteSwapAllFields(P, &p); - byteSwapAllFields(A, &a); + var e: E = @fromBackingInt(0x12345678); + var f: f32 = @bitCast(@as(u32, 0x4640e400)); + byteSwap(T, &s); + byteSwap(K, &k); + byteSwap(P, &p); + byteSwap(A, &a); + byteSwap(E, &e); + byteSwap(f32, &f); try std.testing.expectEqual(T{ .f0 = 0x12, .f1 = 0x3412, @@ -2354,28 +2390,15 @@ test byteSwapAllFields { .f1 = .{ .f0 = 0xF0DEBC9A78563412 }, .f2 = 0x21436587, }, a); + try std.testing.expectEqual(@as(E, @fromBackingInt(0x78563412)), e); + try std.testing.expectEqual(@as(f32, @bitCast(@as(u32, 0x00e44046))), f); } /// Reverses the byte order of all elements in a slice. /// Handles structs, unions, arrays, enums, floats, and integers recursively. /// Useful for converting between little-endian and big-endian representations. pub fn byteSwapAllElements(comptime Elem: type, slice: []Elem) void { - for (slice) |*elem| { - switch (@typeInfo(@TypeOf(elem.*))) { - .@"struct", .@"union", .array => byteSwapAllFields(@TypeOf(elem.*), elem), - .@"enum" => { - elem.* = @fromBackingInt(@intCast(@byteSwap(@backingInt(elem.*)))); - }, - .bool => {}, - .float => |float| { - const int_repr: @Int(.unsigned, float.bits) = @bitCast(elem.*); - elem.* = @bitCast(@byteSwap(int_repr)); - }, - else => { - elem.* = @byteSwap(elem.*); - }, - } - } + for (slice) |*elem| byteSwap(Elem, elem); } /// Returns an iterator that iterates over the slices of `buffer` that are not diff --git a/lib/std/zig.zig b/lib/std/zig.zig index 3a456af0ca2a16978dd405886ecbeac1b273d609..4a5232f27b506faa9205badfb2ef194e96583e64 100644 --- a/lib/std/zig.zig +++ b/lib/std/zig.zig @@ -1658,31 +1658,32 @@ pub fn buildExeSubprocess( }; defer child.kill(io); - var stderr_task = io.concurrent(readStreamAlloc, .{ gpa, io, child.stderr.?, .unlimited }) catch - @panic("TODO use multireader instead"); - defer if (stderr_task.cancel(io)) |slice| gpa.free(slice) else |_| {}; + var multi_reader_buffer: Io.File.MultiReader.Buffer(2) = undefined; + var multi_reader: Io.File.MultiReader = undefined; + multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ child.stdout.?, child.stderr.? }); + defer multi_reader.deinit(); + const stdout = multi_reader.reader(0); + const stderr = multi_reader.reader(1); - var stdout_buffer: [512]u8 = undefined; - var stdout_reader: Io.File.Reader = .initStreaming(child.stdout.?, io, &stdout_buffer); - const stdout = &stdout_reader.interface; + var stdin_buffer: [8]u8 = undefined; + var stdin_writer = child.stdin.?.writerStreaming(io, &stdin_buffer); - { - var w = child.stdin.?.writer(io, &.{}); - w.interface.writeStruct(Client.Message.Header{ .tag = .update, .bytes_len = 0 }, .little) catch |err| switch (err) { - error.WriteFailed => { - log.err("{t} writing to command: {f}", .{ w.err.?, cmd }); - return error.AlreadyReported; - }, - }; - w.interface.writeStruct(Client.Message.Header{ .tag = .exit, .bytes_len = 0 }, .little) catch |err| switch (err) { - error.WriteFailed => { - log.err("{t} writing to command: {f}", .{ w.err.?, cmd }); - return error.AlreadyReported; - }, - }; - } + var client: Client = .{ + .in = stdout, + .out = &stdin_writer.interface, + }; - const Header = Server.Message.Header; + (blk: { + client.serveMessageHeader(.{ .tag = .update, .bytes_len = 0 }) catch |err| break :blk err; + client.serveMessageHeader(.{ .tag = .exit, .bytes_len = 0 }) catch |err| break :blk err; + client.out.flush() catch |err| break :blk err; + }) catch |err| switch (err) { + error.WriteFailed => { + if (stdin_writer.err.? == error.Canceled) return error.Canceled; + log.err("{t} writing to command: {f}", .{ stdin_writer.err.?, cmd }); + return error.AlreadyReported; + }, + }; var result: ?Cache.Path = null; defer if (result) |r| gpa.free(r.sub_path); @@ -1690,33 +1691,29 @@ pub fn buildExeSubprocess( var result_error_bundle: ErrorBundle = .empty; defer result_error_bundle.deinit(gpa); - var body_buffer: std.ArrayList(u8) = .empty; - defer body_buffer.deinit(gpa); - var received_fs_inputs = false; var cache_hit = false; + var eos_err: error{EndOfStream}!void = {}; + while (true) { - const header = stdout.takeStruct(Header, .little) catch |err| switch (err) { - error.ReadFailed => { - log.err("{t} reading from command: {f}", .{ stdout_reader.err.?, cmd }); - return error.AlreadyReported; - }, - error.EndOfStream => break, - }; - body_buffer.clearRetainingCapacity(); - stdout.appendExact(gpa, &body_buffer, header.bytes_len) catch |err| switch (err) { - error.ReadFailed => { - log.err("{t} reading from command: {f}", .{ stdout_reader.err.?, cmd }); - return error.AlreadyReported; + const header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Timeout => unreachable, + error.EndOfStream => |e| { + if (client.in.bufferedLen() == 0) break; + // Better to report the crash with stderr below, but we set + // this in case the child exits successfully while violating + // this protocol. + eos_err = e; + break; }, - error.OutOfMemory => |e| return e, - error.EndOfStream => { - log.err("unexpected end of stream from command: {f}", .{cmd}); + error.Canceled, error.OutOfMemory => |e| return e, + else => |e| { + log.err("{t} reading from command: {f}", .{ e, cmd }); return error.AlreadyReported; }, }; - const body = body_buffer.items; + const body = stdout.take(header.bytes_len) catch unreachable; switch (header.tag) { .zig_version => { @@ -1767,16 +1764,15 @@ pub fn buildExeSubprocess( } } - const stderr_contents = stderr_task.await(io) catch |err| switch (err) { - error.Canceled, error.OutOfMemory => |e| return e, - else => |e| c: { - log.warn("{t} reading stderr from command: {f}", .{ e, cmd }); - break :c ""; - }, - }; + const stderr_contents = stderr.buffered(); if (stderr_contents.len > 0) log.warn("unexpected stderr from {s} command:\n{s}", .{ options.argv[0], stderr_contents }); + eos_err catch { + log.err("unexpected end of stream from command: {f}", .{cmd}); + return error.AlreadyReported; + }; + // Send EOF to stdin. child.stdin.?.close(io); child.stdin = null; @@ -1834,14 +1830,6 @@ pub fn buildExeSubprocess( }; } -fn readStreamAlloc(gpa: Allocator, io: Io, file: Io.File, limit: Io.Limit) ![]u8 { - var file_reader: Io.File.Reader = .initStreaming(file, io, &.{}); - return file_reader.interface.allocRemaining(gpa, limit) catch |err| switch (err) { - error.ReadFailed => return file_reader.err.?, - else => |e| return e, - }; -} - test { _ = Ast; _ = AstRlAnnotate; diff --git a/lib/std/zig/Client.zig b/lib/std/zig/Client.zig index fe50f2314a0b68b5bd9d6510efec6d4146477237..cedda191b99042affaba76eec838fd9f332abdbb 100644 --- a/lib/std/zig/Client.zig +++ b/lib/std/zig/Client.zig @@ -1,3 +1,18 @@ +const Client = @This(); + +const std = @import("std"); +const Io = std.Io; +const Allocator = std.mem.Allocator; +const assert = std.debug.assert; +const Configuration = std.Build.Configuration; +const OutMessage = std.zig.Client.Message; +const InMessage = std.zig.Server.Message; +const Reader = Io.Reader; +const Writer = Io.Writer; + +in: *Reader, +out: *Writer, + pub const Message = struct { pub const Header = extern struct { tag: Tag, @@ -46,11 +61,120 @@ pub const Message = struct { /// The message body has the same format as in Server. new_fuzz_input, + /// Asks the server to run a list of steps. + /// Body is a `BuildSteps`. + /// This message only applies to the build system protocol. + bsp_build_steps = 0x80000000, + _, }; + /// Trailing: + /// * step_indices: [step_count]std.Build.Configuration.Step.Index, + pub const BuildSteps = extern struct { + step_count: u32, + flags: Flags, + + pub const Flags = packed struct(u32) { + /// Can only be enabled when the server declared support for file + /// watching. + watch: bool, + reserved: u31 = 0, + }; + }; + comptime { - const std = @import("std"); - std.debug.assert(@sizeOf(std.Build.abi.fuzz.LimitKind) == 1); + assert(@sizeOf(std.Build.abi.fuzz.LimitKind) == 1); } }; + +pub fn receiveMessage(c: *const Client) Reader.Error!InMessage.Header { + return c.in.takeStruct(InMessage.Header, .little); +} + +/// Assumes that `c.in` is a reader in `multi_reader`. +/// Guarantees that the response body will be buffered in `c.in` on success. +pub fn receiveMessageWithMultiReader( + c: *Client, + multi_reader: *Io.File.MultiReader, + timeout: Io.Timeout, +) (Io.File.MultiReader.Error || Io.Timeout.Error)!InMessage.Header { + while (c.in.bufferedLen() < @sizeOf(InMessage.Header)) { + multi_reader.fill(64, timeout) catch |err| switch (err) { + error.Canceled, + error.Timeout, + error.ConcurrencyUnavailable, + error.EndOfStream, + => |e| return e, + }; + } + const header = c.in.takeStruct(InMessage.Header, .little) catch unreachable; + while (c.in.bufferedLen() < header.bytes_len) { + try multi_reader.fill(header.bytes_len - c.in.bufferedLen(), timeout); + } + try multi_reader.checkAnyError(); + return header; +} + +/// Don't forget to flush! +pub fn serveMessageHeader(c: *const Client, header: OutMessage.Header) Writer.Error!void { + try c.out.writeStruct(header, .little); +} + +pub fn serveBodylessMessage(c: *const Client, tag: OutMessage.Tag) Writer.Error!void { + try c.serveMessageHeader(.{ .tag = tag, .bytes_len = 0 }); + try c.out.flush(); +} + +pub fn serveRunTest(c: *const Client, index: u32) !void { + try c.serveMessageHeader(.{ + .tag = .run_test, + .bytes_len = @sizeOf(u32), + }); + try c.out.writeInt(u32, index, .little); + try c.out.flush(); +} + +pub fn serveRunFuzzTestMessage( + c: *const Client, + test_names: []const []const u8, + kind: std.Build.abi.fuzz.LimitKind, + amount_or_instance: u64, +) !void { + try c.serveMessageHeader(.{ + .tag = .start_fuzzing, + .bytes_len = 1 + 8 + 4 + count: { + var bytes_len: u32 = @intCast(test_names.len * 4); + for (test_names) |name| { + bytes_len += @intCast(name.len); + } + break :count bytes_len; + }, + }); + try c.out.writeByte(@backingInt(kind)); + try c.out.writeInt(u64, amount_or_instance, .little); + try c.out.writeInt(u32, @intCast(test_names.len), .little); + for (test_names) |test_name| { + try c.out.writeInt(u32, @intCast(test_name.len), .little); + try c.out.writeAll(test_name); + } + try c.out.flush(); +} + +pub fn serveBuildSteps( + c: *const Client, + steps: []const Configuration.Step.Index, + flags: OutMessage.BuildSteps.Flags, +) !void { + try c.serveMessageHeader(.{ + .tag = .bsp_build_steps, + .bytes_len = @intCast(@sizeOf(OutMessage.BuildSteps) + steps.len * @sizeOf(Configuration.Step.Index)), + }); + const body: OutMessage.BuildSteps = .{ + .step_count = @intCast(steps.len), + .flags = flags, + }; + try c.out.writeStruct(body, .little); + try c.out.writeSliceEndian(Configuration.Step.Index, steps, .little); + try c.out.flush(); +} diff --git a/lib/std/zig/Server.zig b/lib/std/zig/Server.zig index cf43cb0af2822cf416868dd1eba76bcb06b791a7..1f6d208084abdcc9d93ef33929e36aa0239549af 100644 --- a/lib/std/zig/Server.zig +++ b/lib/std/zig/Server.zig @@ -1,12 +1,8 @@ const Server = @This(); -const builtin = @import("builtin"); - const std = @import("std"); const Allocator = std.mem.Allocator; const assert = std.debug.assert; -const native_endian = builtin.target.cpu.arch.endian(); -const need_bswap = native_endian != .little; const Cache = std.Build.Cache; const OutMessage = std.zig.Server.Message; const InMessage = std.zig.Client.Message; @@ -16,6 +12,14 @@ const Writer = std.Io.Writer; in: *Reader, out: *Writer, +/// The ABI version of the build system protocol. Will be bumped whenever a +/// backwards incompatible changes to the protocol is made. +/// +/// Does not apply to the internal compiler protocol or test runner. +/// +/// See `version` in `Message.Handshake`. +pub const build_system_version: u32 = 1; + pub const Message = struct { pub const Header = extern struct { tag: Tag, @@ -70,9 +74,62 @@ pub const Message = struct { /// Body is a TimeReport. time_report, + /// The first message sent by the server over the build system protocol. + /// Body is a `Handshake`. + /// This message only applies to the build system protocol. + bsp_handshake = 0x80000000, + /// Notifies that a new configuration file is available. + /// Body is a cwd relative path to the configuration file. + /// This message only applies to the build system protocol. + bsp_configuration, + /// Does not have a body. + /// This message only applies to the build system protocol. + bsp_build_started, + /// Does not have a body. + /// This message only applies to the build system protocol. + bsp_build_completed, + /// Body is a `Configuration.Step.Index`. + /// This message only applies to the build system protocol. + bsp_step_started, + /// Body is a `BuildStepCompleted`. + /// This message only applies to the build system protocol. + bsp_step_completed, + _, }; + /// Trailing: + /// * base_paths: BasePaths, + pub const Handshake = extern struct { + /// See `build_system_version`. + version: u32, + flags: Flags, + + pub const Flags = packed struct(u32) { + file_system_watch_supported: bool, + _: u31 = 0, + }; + }; + + /// Trailing: + /// * error_bundle: ErrorBundle, + pub const BuildStepCompleted = extern struct { + step_index: std.Build.Configuration.Step.Index, + status: Status, + error_bundle: ErrorBundle, + // TODO result_error_msgs + // TODO result_stderr + // TODO result_peak_rss + // TODO result_duration_ns + + pub const Status = enum(u32) { + success, + failure, + skipped, + skipped_oom, + }; + }; + pub const PathPrefix = enum(u8) { cwd, zig_lib, @@ -140,21 +197,6 @@ pub const Message = struct { }; }; -pub const Options = struct { - in: *Reader, - out: *Writer, - zig_version: []const u8, -}; - -pub fn init(options: Options) !Server { - var s: Server = .{ - .in = options.in, - .out = options.out, - }; - try s.serveStringMessage(.zig_version, options.zig_version); - return s; -} - pub fn receiveMessage(s: *Server) !InMessage.Header { return s.in.takeStruct(InMessage.Header, .little); } @@ -183,6 +225,11 @@ pub fn serveMessageHeader(s: *const Server, header: OutMessage.Header) !void { try s.out.writeStruct(header, .little); } +pub fn serveBodylessMessage(s: *const Server, tag: OutMessage.Tag) Writer.Error!void { + try s.serveMessageHeader(.{ .tag = tag, .bytes_len = 0 }); + try s.out.flush(); +} + pub fn serveU32Message(s: *const Server, tag: OutMessage.Tag, int: u32) !void { try serveMessageHeader(s, .{ .tag = tag, diff --git a/src/Compilation.zig b/src/Compilation.zig index 0952e9f7dc05456c4a8d4e6cbc35fa17d14ef484..16d290c1931760a0441b18d7480678b0068218e5 100644 --- a/src/Compilation.zig +++ b/src/Compilation.zig @@ -6028,26 +6028,30 @@ fn spawnZigRc( multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ child.stdout.?, child.stderr.? }); defer multi_reader.deinit(); - const stdout = multi_reader.fileReader(0); - const MessageHeader = std.zig.Server.Message.Header; + const stdout = multi_reader.reader(0); var eos_err: error{EndOfStream}!void = {}; + var client: std.zig.Client = .{ + .in = stdout, + .out = undefined, + }; + while (true) { - const header = stdout.interface.takeStruct(MessageHeader, .little) catch |err| switch (err) { - error.EndOfStream => break, - error.ReadFailed => return stdout.err.?, - }; - const body = stdout.interface.take(header.bytes_len) catch |err| switch (err) { + const header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Timeout => unreachable, error.EndOfStream => |e| { + if (client.in.bufferedLen() == 0) break; // Better to report the crash with stderr below, but we set // this in case the child exits successfully while violating // this protocol. eos_err = e; break; }, - error.ReadFailed => return stdout.err.?, + else => |e| return e, }; + const body = client.in.take(header.bytes_len) catch unreachable; + switch (header.tag) { // We expect exactly one ErrorBundle, and if any error_bundle header is // sent then it's a fatal error. diff --git a/src/main.zig b/src/main.zig index 7a69c9ba65d5c1da8f687f2cb2a6d7ab7eb28698..4faeaebcecd927c8bdd3468ebc32453a74fc9c18 100644 --- a/src/main.zig +++ b/src/main.zig @@ -4297,11 +4297,8 @@ fn serve( const gpa = comp.gpa; const io = comp.io; - var server = try Server.init(.{ - .in = in, - .out = out, - .zig_version = build_options.version, - }); + var server: Server = .{ .in = in, .out = out }; + try server.serveStringMessage(.zig_version, build_options.version); var child_pid: ?std.process.Child.Id = null; diff --git a/test/standalone/build.zig b/test/standalone/build.zig index f14d17562d4715ca75340d67b6b2ce5b7344b73d..fa1c4186abcd7da988acfcda5ca30ed966d62a80 100644 --- a/test/standalone/build.zig +++ b/test/standalone/build.zig @@ -31,6 +31,7 @@ pub fn build(b: *std.Build) void { const tools_target = b.resolveTargetQuery(.{}); for ([_][]const u8{ // Alphabetically sorted. No need to build `tools/spirv/grammar.zig`. + "../../tools/bsp.zig", "../../tools/check_mingw.zig", "../../tools/dump-cov.zig", "../../tools/fetch_them_macos_headers.zig", diff --git a/tools/bsp.zig b/tools/bsp.zig new file mode 100644 index 0000000000000000000000000000000000000000..bab0bc2c70afb95e99eb51ccf29be357ce4a678f --- /dev/null +++ b/tools/bsp.zig @@ -0,0 +1,242 @@ +//! CLI tool to interface with the build system protocol (zig build --listen=-) + +const std = @import("std"); +const Io = std.Io; +const Allocator = std.mem.Allocator; +const Configuration = std.Build.Configuration; +const Client = std.zig.Client; +const Server = std.zig.Server; +const log = std.log.scoped(.bsp); + +pub fn main(init: std.process.Init) !void { + const io = init.io; + const gpa = init.gpa; + const arena = init.arena.allocator(); + + var maker_args: std.ArrayList([]const u8) = .empty; + + const args = try init.minimal.args.toSlice(arena); + for (args[1..]) |arg| { + try maker_args.append(arena, try arena.dupe(u8, arg)); + } + if (maker_args.items.len < 1) try maker_args.append(arena, "zig"); + if (maker_args.items.len < 2) try maker_args.append(arena, "build"); + if (!std.mem.eql(u8, maker_args.last().?.*, "--listen=-")) try maker_args.append(arena, "--listen=-"); + + log.debug("cmd: {f}", .{std.zig.SubprocessCommand{ + .argv = maker_args.items, + }}); + + var child_process = std.process.spawn(io, .{ + .argv = maker_args.items, + .stdin = .pipe, + .stdout = .pipe, + .stderr = .pipe, + }) catch |err| std.debug.panic("failed to spawn process: {}", .{err}); + errdefer child_process.kill(io); + + var multi_reader_buffer: Io.File.MultiReader.Buffer(2) = undefined; + var multi_reader: Io.File.MultiReader = undefined; + defer multi_reader.deinit(); + multi_reader.init( + gpa, + io, + multi_reader_buffer.toStreams(), + &.{ child_process.stdout.?, child_process.stderr.? }, + ); + const client_stdout = multi_reader.reader(0); + const client_stderr = multi_reader.reader(1); + + var client_stdout_buffer: [256]u8 = undefined; + var client_stdout_writer = child_process.stdin.?.writerStreaming(io, &client_stdout_buffer); + + var client: Client = .{ + .in = client_stdout, + .out = &client_stdout_writer.interface, + }; + + const err = blk: { + const handshake: Server.Message.Handshake = handshake: { + const header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Canceled, error.ConcurrencyUnavailable => |e| return e, + error.Timeout => unreachable, + else => |e| { + log.err("failed to receive message: {t}", .{err}); + break :blk e; + }, + }; + const body = client_stdout.take(header.bytes_len) catch unreachable; + log.debug("received {f} ({d} bytes)", .{ fmtEnum(header.tag), body.len }); + + if (header.tag != .bsp_handshake) { + log.err("received unexpected message: {f}", .{fmtEnum(header.tag)}); + return error.UnexpectedMessage; + } + + var r: Io.Reader = .fixed(body); + break :handshake try r.takeStruct(Server.Message.Handshake, .little); + }; + _ = handshake; + + var conf_arena_allocator: std.heap.ArenaAllocator = .init(gpa); + defer conf_arena_allocator.deinit(); + const conf_arena = conf_arena_allocator.allocator(); + + const configuration = configuration: { + const header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Canceled, error.ConcurrencyUnavailable => |e| return e, + error.Timeout => unreachable, + else => |e| { + log.err("failed to receive message: {t}", .{err}); + break :blk e; + }, + }; + const body = client_stdout.take(header.bytes_len) catch unreachable; + log.debug("received {t} ({d} bytes)", .{ header.tag, body.len }); + + if (header.tag != .bsp_configuration) { + log.err("received unexpected message: {f}", .{fmtEnum(header.tag)}); + return error.UnexpectedMessage; + } + + const configuration_path = body; + var file = Io.Dir.cwd().openFile(io, configuration_path, .{}) catch |err| + std.debug.panic("failed to open configuration file {q}: {t}", .{ configuration_path, err }); + defer file.close(io); + break :configuration Configuration.loadFile(conf_arena, io, file) catch |err| + std.debug.panic("failed to load configuration file {q}: {t}", .{ configuration_path, err }); + }; + const c = &configuration; + + var top_level_steps: std.array_hash_map.String(Configuration.Step.Index) = .empty; + defer top_level_steps.deinit(gpa); + + for (c.steps, 0..) |*conf_step, step_index_usize| { + if (conf_step.owner != .root) continue; + const step_index: Configuration.Step.Index = @fromBackingInt(@intCast(step_index_usize)); + const flags = conf_step.flags(c); + if (flags.tag != .top_level) continue; + const name = step_index.ptr(c).name.slice(c); + try top_level_steps.putNoClobber(gpa, name, step_index); + } + + std.debug.print("Steps:\n", .{}); + for (top_level_steps.keys()) |name| { + std.debug.print(" - {q}\n", .{name}); + } + std.debug.print( + \\Available Commands: + \\ - build [step names / step indices] + \\ - watch [step names / step indices] + \\ - exit + \\ + , .{}); + + var stdin_reader_buffer: [256]u8 = undefined; + var stdin_reader = Io.File.stdin().reader(io, &stdin_reader_buffer); + const stdin = &stdin_reader.interface; + + while (true) { + try Io.File.stdout().writeStreamingAll(io, "> "); + const command = try stdin.takeDelimiterExclusive('\n'); + stdin.toss(1); + if (std.mem.startsWith(u8, command, "build") or + std.mem.startsWith(u8, command, "watch")) + { + var steps: std.ArrayList(Configuration.Step.Index) = .empty; + defer steps.deinit(gpa); + + const watch = std.mem.startsWith(u8, command, "watch"); + + if (std.mem.cutPrefix(u8, command, "build ") orelse + std.mem.cutPrefix(u8, command, "watch ")) |command_args| + { + var it = std.mem.tokenizeScalar(u8, command_args, ' '); + while (it.next()) |arg| { + const step: Configuration.Step.Index = + if (std.fmt.parseInt(u32, arg, 10)) |i| + @fromBackingInt(i) + else |_| + top_level_steps.get(arg) orelse std.debug.panic("unexpected step name or index", .{}); + try steps.append(gpa, step); + } + } + + if (steps.items.len < 1) { + try steps.append(gpa, c.default_step); + } + + try client.serveBuildSteps(steps.items, .{ .watch = watch }); + + while (true) { + const header: Server.Message.Header = client.receiveMessageWithMultiReader(&multi_reader, .none) catch |err| switch (err) { + error.Canceled, error.ConcurrencyUnavailable => |e| return e, + error.Timeout => unreachable, + else => |e| { + log.err("failed to receive message: {t}", .{err}); + break :blk e; + }, + }; + const body = client_stdout.take(header.bytes_len) catch unreachable; + log.debug("received {f} ({d} bytes)", .{ fmtEnum(header.tag), body.len }); + + switch (header.tag) { + .bsp_build_started => {}, + .bsp_build_completed => if (!watch) break, + .bsp_step_started => {}, + .bsp_step_completed => {}, + .bsp_configuration => @panic("TODO"), + else => std.debug.panic("received unexpected message: {f}", .{fmtEnum(header.tag)}), + } + } + continue; + } else if (std.mem.eql(u8, command, "exit")) { + try client.serveBodylessMessage(.exit); + break; + } else { + log.err("unknown command: {q}", .{command}); + continue; + } + } + }; + + try multi_reader.fillRemaining(.none); + + if (client_stderr.bufferedLen() > 0) { + log.err("stderr:\n{s}\n", .{client_stderr.buffered()}); + } + + try err; + + const term = try child_process.wait(io); + + if (!term.success()) { + log.err("maker {f}", .{term}); + } +} + +const FormatEnum = union(enum) { + named: []const u8, + unnamed: usize, + + pub fn format( + e: FormatEnum, + writer: *std.Io.Writer, + ) std.Io.Writer.Error!void { + switch (e) { + .named => |name| { + try writer.writeByte('.'); + try writer.writeAll(name); + }, + .unnamed => |number| try writer.print("0x{x}", .{number}), + } + } +}; + +fn fmtEnum(e: anytype) FormatEnum { + if (std.enums.tagName(@TypeOf(e), e)) |name| { + return .{ .named = name }; + } else { + return .{ .unnamed = @backingInt(e) }; + } +} diff --git a/tools/incr-check.zig b/tools/incr-check.zig index 89c14ce1e7f60a710e863349025d3803f2dc37bb..cbc1ec659409eadefd89d516c8816ed36f92d4e5 100644 --- a/tools/incr-check.zig +++ b/tools/incr-check.zig @@ -305,21 +305,23 @@ const Eval = struct { fn check(eval: *Eval, mr: *Io.File.MultiReader, update: Case.Update, prog_node: std.Progress.Node) !void { const arena = eval.arena; - const stdout = mr.fileReader(0); - const stderr = &mr.fileReader(1).interface; - const Header = std.zig.Server.Message.Header; + const stdout = mr.reader(0); + const stderr = mr.reader(1); + + var client: std.zig.Client = .{ + .in = stdout, + .out = undefined, + }; while (true) { - const header = stdout.interface.takeStruct(Header, .little) catch |err| switch (err) { - error.EndOfStream => break, - error.ReadFailed => return stdout.err.?, - }; - const body = stdout.interface.take(header.bytes_len) catch |err| switch (err) { + const header = client.receiveMessageWithMultiReader(mr, .none) catch |err| switch (err) { + error.Timeout => unreachable, // If this panic triggers it might be helpful to rework this // code to print the stderr from the abnormally terminated child. error.EndOfStream => @panic("unexpected mid-message end of stream"), - error.ReadFailed => return stdout.err.?, + else => |e| return e, }; + const body = client.in.take(header.bytes_len) catch unreachable; switch (header.tag) { .error_bundle => { @@ -605,12 +607,13 @@ const Eval = struct { fn requestUpdate(eval: *Eval) !void { const io = eval.io; - const header: std.zig.Client.Message.Header = .{ - .tag = .update, - .bytes_len = 0, + + var w = eval.child.stdin.?.writerStreaming(io, &.{}); + var client: std.zig.Client = .{ + .in = undefined, + .out = &w.interface, }; - var w = eval.child.stdin.?.writer(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { + client.serveBodylessMessage(.update) catch |err| switch (err) { error.WriteFailed => return w.err.?, }; } @@ -618,22 +621,23 @@ const Eval = struct { fn end(eval: *Eval, mr: *Io.File.MultiReader) !void { requestExit(eval.child, eval); - const stdout = mr.fileReader(0); - const Header = std.zig.Server.Message.Header; + var client: std.zig.Client = .{ + .in = mr.reader(0), + .out = undefined, + }; while (true) { - const header = stdout.interface.takeStruct(Header, .little) catch |err| switch (err) { - error.EndOfStream => break, - error.ReadFailed => return stdout.err.?, - }; - stdout.interface.discardAll(header.bytes_len) catch |err| switch (err) { - error.ReadFailed => return stdout.err.?, - error.EndOfStream => |e| return e, + const header = client.receiveMessageWithMultiReader(mr, .none) catch |err| switch (err) { + error.Timeout => unreachable, + error.EndOfStream => |e| { + if (client.in.bufferedLen() == 0) break; + return e; + }, + else => |e| return e, }; + try client.in.discardAll(header.bytes_len); } - try mr.fillRemaining(.none); - const stderr = mr.reader(1).buffered(); if (stderr.len > 0) eval.fatal("unexpected stderr:\n{s}", .{stderr}); } @@ -899,12 +903,12 @@ fn requestExit(child: *std.process.Child, eval: *Eval) void { if (child.stdin == null) return; const io = eval.io; - const header: std.zig.Client.Message.Header = .{ - .tag = .exit, - .bytes_len = 0, + var w = eval.child.stdin.?.writerStreaming(io, &.{}); + var client: std.zig.Client = .{ + .in = undefined, + .out = &w.interface, }; - var w = eval.child.stdin.?.writer(io, &.{}); - w.interface.writeStruct(header, .little) catch |err| switch (err) { + client.serveBodylessMessage(.exit) catch |err| switch (err) { error.WriteFailed => switch (w.err.?) { error.BrokenPipe => {}, else => |e| eval.fatal("failed to send exit: {t}", .{e}),