diff --git a/lib/compiler/build_runner.zig b/lib/compiler/build_runner.zig index 752f1fdae3dd78e3c6c483b604bbe1b22aefdbcb..7efee4bac4c78f0cc2b1db807180cf57d7cd271b 100644 --- a/lib/compiler/build_runner.zig +++ b/lib/compiler/build_runner.zig @@ -503,7 +503,7 @@ pub fn main(init: process.Init.Minimal) !void { var run: Run = .{ .gpa = gpa, - .max_rss = max_rss, + .available_rss = max_rss, .max_rss_is_default = false, .max_rss_mutex = .init, .skip_oom_steps = skip_oom_steps, @@ -514,7 +514,6 @@ pub fn main(init: process.Init.Minimal) !void { .memory_blocked_steps = .empty, .step_stack = .empty, - .claimed_rss = 0, .error_style = error_style, .multiline_errors = multiline_errors, .summary = summary orelse if (watch or webui_listen != null) .line else .failures, @@ -524,8 +523,8 @@ pub fn main(init: process.Init.Minimal) !void { run.step_stack.deinit(gpa); } - if (run.max_rss == 0) { - run.max_rss = process.totalSystemMemory() catch std.math.maxInt(u64); + if (run.available_rss == 0) { + run.available_rss = process.totalSystemMemory() catch std.math.maxInt(u64); run.max_rss_is_default = true; } @@ -595,6 +594,7 @@ pub fn main(init: process.Init.Minimal) !void { .rebuild => { for (run.step_stack.keys()) |step| { step.state = .precheck_done; + step.pending_deps = @intCast(step.dependencies.items.len); step.reset(gpa); } continue :rebuild; @@ -637,7 +637,7 @@ pub fn main(init: process.Init.Minimal) !void { fn markFailedStepsDirty(gpa: Allocator, all_steps: []const *Step) void { for (all_steps) |step| switch (step.state) { - .dependency_failure, .failure, .skipped => step.recursiveReset(gpa), + .dependency_failure, .failure, .skipped => _ = step.invalidateResult(gpa), else => continue, }; // Now that all dirty steps have been found, the remaining steps that @@ -658,7 +658,8 @@ fn countSubProcesses(all_steps: []const *Step) usize { const Run = struct { gpa: Allocator, - max_rss: u64, + + available_rss: usize, max_rss_is_default: bool, max_rss_mutex: Io.Mutex, skip_oom_steps: bool, @@ -670,7 +671,6 @@ const Run = struct { /// Allocated into `gpa`. step_stack: std.AutoArrayHashMapUnmanaged(*Step, void), - claimed_rss: usize, error_style: ErrorStyle, multiline_errors: MultilineErrors, summary: Summary, @@ -715,12 +715,15 @@ fn prepare( var any_problems = false; for (step_stack.keys()) |s| { if (s.max_rss == 0) continue; - if (s.max_rss > run.max_rss) { + if (s.max_rss > run.available_rss) { if (run.skip_oom_steps) { s.state = .skipped_oom; + for (s.dependants.items) |dependant| { + dependant.pending_deps -= 1; + } } else { std.debug.print("{s}{s}: this step declares an upper bound of {d} bytes of memory, exceeding the available {d} bytes of memory\n", .{ - s.owner.dep_prefix, s.name, s.max_rss, run.max_rss, + s.owner.dep_prefix, s.name, s.max_rss, run.available_rss, }); any_problems = true; } @@ -747,23 +750,26 @@ fn runStepNames( const step_stack = &run.step_stack; { + // 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 + // a step is initial when it actually became ready due to an earlier initial step. + var initial_set: std.ArrayList(*Step) = .empty; + defer initial_set.deinit(gpa); + try initial_set.ensureUnusedCapacity(gpa, step_stack.count()); + for (step_stack.keys()) |s| { + if (s.state == .precheck_done and s.pending_deps == 0) { + initial_set.appendAssumeCapacity(s); + } + } + const step_prog = parent_prog_node.start("steps", step_stack.count()); defer step_prog.end(); var group: Io.Group = .init; defer group.cancel(io); - - // Here we spawn the initial set of tasks with a nice heuristic - - // dependency order. Each worker when it finishes a step will then - // check whether it should run any dependants. - const steps_slice = step_stack.keys(); - for (0..steps_slice.len) |i| { - const step = steps_slice[steps_slice.len - i - 1]; - if (step.state == .skipped_oom) continue; - - group.async(io, workerMakeOneStep, .{ &group, b, step, step_prog, run }); - } - + // Start working on all of the initial steps... + for (initial_set.items) |s| try stepReady(&group, b, s, step_prog, run); + // ...and `makeStep` will trigger every other step when their last dependency finishes. try group.await(io); } @@ -798,17 +804,7 @@ fn runStepNames( switch (s.state) { .precheck_unstarted => unreachable, .precheck_started => unreachable, - .running => unreachable, - .precheck_done => { - // precheck_done is equivalent to dependency_failure in the case of - // transitive dependencies. For example: - // A -> B -> C (failure) - // B will be marked as dependency_failure, while A may never be queued, and thus - // remain in the initial state of precheck_done. - s.state = .dependency_failure; - if (run.web_server) |*ws| ws.updateStepStatus(s, .failure); - pending_count += 1; - }, + .precheck_done => unreachable, .dependency_failure => pending_count += 1, .success => success_count += 1, .skipped, .skipped_oom => skipped_count += 1, @@ -1008,7 +1004,6 @@ fn printStepStatus(s: *Step, stderr: Io.Terminal, run: *const Run) !void { .precheck_unstarted => unreachable, .precheck_started => unreachable, .precheck_done => unreachable, - .running => unreachable, .dependency_failure => { try stderr.setColor(.dim); @@ -1067,16 +1062,16 @@ fn printStepStatus(s: *Step, stderr: Io.Terminal, run: *const Run) !void { } try writer.writeAll("\n"); }, - .skipped, .skipped_oom => |skip| { + .skipped => { try stderr.setColor(.yellow); - try writer.writeAll(" skipped"); - if (skip == .skipped_oom) { - try writer.writeAll(" (not enough memory)"); - try stderr.setColor(.dim); - try writer.print(" upper bound of {d} exceeded runner limit ({d})", .{ s.max_rss, run.max_rss }); - try stderr.setColor(.yellow); - } - try writer.writeAll("\n"); + try writer.writeAll(" skipped\n"); + try stderr.setColor(.reset); + }, + .skipped_oom => { + try stderr.setColor(.yellow); + try writer.writeAll(" skipped (not enough memory)"); + try stderr.setColor(.dim); + try writer.print(" upper bound of {d} exceeded runner limit ({d})\n", .{ s.max_rss, run.available_rss }); try stderr.setColor(.reset); }, .failure => { @@ -1250,10 +1245,10 @@ fn printTreeStep( /// Each step has its dependencies traversed in random order, this accomplishes /// two things: /// - `step_stack` will be in randomized-depth-first order, so the build runner -/// spawns steps in a random (but optimized) order +/// spawns initial steps in a random order /// - each step's `dependants` list is also filled in a random order, so that -/// when it finishes executing in `workerMakeOneStep`, it spawns next steps -/// to run in random order +/// when it finishes executing in `makeStep`, it spawns next steps to run in +/// random order fn constructGraphAndCheckForDependencyLoop( gpa: Allocator, b: *std.Build, @@ -1290,12 +1285,12 @@ fn constructGraphAndCheckForDependencyLoop( } s.state = .precheck_done; + s.pending_deps = @intCast(s.dependencies.items.len); }, .precheck_done => {}, // These don't happen until we actually run the step graph. .dependency_failure => unreachable, - .running => unreachable, .success => unreachable, .failure => unreachable, .skipped => unreachable, @@ -1303,148 +1298,136 @@ fn constructGraphAndCheckForDependencyLoop( } } -fn workerMakeOneStep( +/// Runs the "make" function of the single step `s`, updates its state, and then spawns newly-ready +/// dependant steps in `group`. If `s` makes an RSS claim (i.e. `s.max_rss != 0`), the caller must +/// have already subtracted this value from `run.available_rss`. This function will release the RSS +/// claim (i.e. add `s.max_rss` back into `run.available_rss`) and queue any viable memory-blocked +/// steps after "make" completes for `s`. +fn makeStep( group: *Io.Group, b: *std.Build, s: *Step, - prog_node: std.Progress.Node, + root_prog_node: std.Progress.Node, run: *Run, -) void { +) Io.Cancelable!void { const graph = b.graph; const io = graph.io; const gpa = run.gpa; - // First, check the conditions for running this step. If they are not met, - // then we return without doing the step, relying on another worker to - // queue this step up again when dependencies are met. - for (s.dependencies.items) |dep| { - switch (@atomicLoad(Step.State, &dep.state, .seq_cst)) { - .success, .skipped => continue, - .failure, .dependency_failure, .skipped_oom => { - @atomicStore(Step.State, &s.state, .dependency_failure, .seq_cst); - if (run.web_server) |*ws| ws.updateStepStatus(s, .failure); - return; - }, - .precheck_done, .running => { - // dependency is not finished yet. - return; - }, - .precheck_unstarted => unreachable, - .precheck_started => unreachable, - } - } - - if (s.max_rss != 0) { - run.max_rss_mutex.lockUncancelable(io); - defer run.max_rss_mutex.unlock(io); - - // Avoid running steps twice. - if (s.state != .precheck_done) { - // Another worker got the job. - return; - } - - const new_claimed_rss = run.claimed_rss + s.max_rss; - if (new_claimed_rss > run.max_rss) { - // Running this step right now could possibly exceed the allotted RSS. - // Add this step to the queue of memory-blocked steps. - run.memory_blocked_steps.append(gpa, s) catch @panic("OOM"); - return; - } - - run.claimed_rss = new_claimed_rss; - s.state = .running; - } else { - // Avoid running steps twice. - if (@cmpxchgStrong(Step.State, &s.state, .precheck_done, .running, .seq_cst, .seq_cst) != null) { - // Another worker got the job. - return; - } - } - - const sub_prog_node = prog_node.start(s.name, 0); - defer sub_prog_node.end(); - - if (run.web_server) |*ws| ws.updateStepStatus(s, .wip); - - const make_result = s.make(.{ - .progress_node = sub_prog_node, - .watch = run.watch, - .web_server = if (run.web_server) |*ws| ws else null, - .unit_test_timeout_ns = run.unit_test_timeout_ns, - .gpa = gpa, - }); - - // No matter the result, we want to display error/warning messages. - const show_compile_errors = s.result_error_bundle.errorMessageCount() > 0; - const show_error_msgs = s.result_error_msgs.items.len > 0; - const show_stderr = s.result_stderr.len > 0; - if (show_error_msgs or show_compile_errors or show_stderr) { - const stderr = io.lockStderr(&stdio_buffer_allocation, graph.stderr_mode) catch |err| switch (err) { - error.Canceled => return, - }; - defer io.unlockStderr(); - printErrorMessages(gpa, s, .{}, stderr.terminal(), run.error_style, run.multiline_errors) catch {}; - } - - handle_result: { - if (make_result) |_| { - @atomicStore(Step.State, &s.state, .success, .seq_cst); - if (run.web_server) |*ws| ws.updateStepStatus(s, .success); + { + const step_prog_node = root_prog_node.start(s.name, 0); + defer step_prog_node.end(); + + if (run.web_server) |*ws| ws.updateStepStatus(s, .wip); + + const new_state: Step.State = for (s.dependencies.items) |dep| { + switch (@atomicLoad(Step.State, &dep.state, .monotonic)) { + .precheck_unstarted => unreachable, + .precheck_started => unreachable, + .precheck_done => unreachable, + + .failure, + .dependency_failure, + .skipped_oom, + => break .dependency_failure, + + .success, .skipped => {}, + } + } else if (s.make(.{ + .progress_node = step_prog_node, + .watch = run.watch, + .web_server = if (run.web_server) |*ws| ws else null, + .unit_test_timeout_ns = run.unit_test_timeout_ns, + .gpa = gpa, + })) state: { + break :state .success; } else |err| switch (err) { - error.MakeFailed => { - @atomicStore(Step.State, &s.state, .failure, .seq_cst); + error.MakeFailed => .failure, + error.MakeSkipped => .skipped, + }; + + @atomicStore(Step.State, &s.state, new_state, .monotonic); + + switch (new_state) { + .precheck_unstarted => unreachable, + .precheck_started => unreachable, + .precheck_done => unreachable, + + .failure, + .dependency_failure, + .skipped_oom, + => { if (run.web_server) |*ws| ws.updateStepStatus(s, .failure); std.Progress.setStatus(.failure_working); - break :handle_result; }, - error.MakeSkipped => { - @atomicStore(Step.State, &s.state, .skipped, .seq_cst); + + .success, + .skipped, + => { if (run.web_server) |*ws| ws.updateStepStatus(s, .success); }, } + } - // Successful completion of a step, so we queue up its dependants as well. - for (s.dependants.items) |dep| { - group.async(io, workerMakeOneStep, .{ group, b, dep, prog_node, run }); - } + // No matter the result, we want to display error/warning messages. + if (s.result_error_bundle.errorMessageCount() > 0 or + s.result_error_msgs.items.len > 0 or + s.result_stderr.len > 0) + { + const stderr = try io.lockStderr(&stdio_buffer_allocation, graph.stderr_mode); + defer io.unlockStderr(); + printErrorMessages(gpa, s, .{}, stderr.terminal(), run.error_style, run.multiline_errors) catch {}; } - // If this is a step that claims resources, we must now queue up other - // steps that are waiting for resources. if (s.max_rss != 0) { - var dispatch_deps: std.ArrayList(*Step) = .empty; - defer dispatch_deps.deinit(gpa); + var dispatch_set: std.ArrayList(*Step) = .empty; + defer dispatch_set.deinit(gpa); + // Release our RSS claim and kick off some blocked steps if possible. We use `dispatch_set` + // as a staging buffer to avoid recursing into `makeStep` while `run.max_rss_mutex` is held. { - run.max_rss_mutex.lockUncancelable(io); + try run.max_rss_mutex.lock(io); defer run.max_rss_mutex.unlock(io); - - dispatch_deps.ensureUnusedCapacity(gpa, run.memory_blocked_steps.items.len) catch @panic("OOM"); - - // Give the memory back to the scheduler. - run.claimed_rss -= s.max_rss; - // Avoid kicking off too many tasks that we already know will not have - // enough resources. - var remaining = run.max_rss - run.claimed_rss; - var i: usize = 0; - for (run.memory_blocked_steps.items) |dep| { - assert(dep.max_rss != 0); - if (dep.max_rss <= remaining) { - remaining -= dep.max_rss; - dispatch_deps.appendAssumeCapacity(dep); - } else { - run.memory_blocked_steps.items[i] = dep; - i += 1; - } + run.available_rss += s.max_rss; + dispatch_set.ensureUnusedCapacity(gpa, run.memory_blocked_steps.items.len) catch @panic("OOM"); + while (run.memory_blocked_steps.getLastOrNull()) |candidate| { + if (run.available_rss < candidate.max_rss) break; + assert(run.memory_blocked_steps.pop() == candidate); + dispatch_set.appendAssumeCapacity(candidate); } - run.memory_blocked_steps.shrinkRetainingCapacity(i); } - for (dispatch_deps.items) |dep| { - // Must be called without max_rss_mutex held in case it executes recursively. - group.async(io, workerMakeOneStep, .{ group, b, dep, prog_node, run }); + for (dispatch_set.items) |candidate| { + group.async(io, makeStep, .{ group, b, candidate, root_prog_node, run }); } } + + for (s.dependants.items) |dependant| { + // `.acq_rel` synchronizes with itself to ensure all dependencies' final states are visible when this hits 0. + if (@atomicRmw(u32, &dependant.pending_deps, .Sub, 1, .acq_rel) == 1) { + try stepReady(group, b, dependant, root_prog_node, run); + } + } +} + +fn stepReady( + group: *Io.Group, + b: *std.Build, + s: *Step, + root_prog_node: std.Progress.Node, + run: *Run, +) !void { + const io = b.graph.io; + if (s.max_rss != 0) { + try run.max_rss_mutex.lock(io); + defer run.max_rss_mutex.unlock(io); + if (run.available_rss < s.max_rss) { + // Running this step right now could possibly exceed the allotted RSS. + run.memory_blocked_steps.append(run.gpa, s) catch @panic("OOM"); + return; + } + run.available_rss -= s.max_rss; + } + group.async(io, makeStep, .{ group, b, s, root_prog_node, run }); } pub fn printErrorMessages( diff --git a/lib/std/Build/Step.zig b/lib/std/Build/Step.zig index ec931306cd86e102deb3ce477045fdfeb78d0f18..24e00bea5ed8629d0fbd2580169687ad0274ba10 100644 --- a/lib/std/Build/Step.zig +++ b/lib/std/Build/Step.zig @@ -29,7 +29,6 @@ dependants: ArrayList(*Step), /// retain previous value, or update. inputs: Inputs, -state: State, /// Set this field to declare an upper bound on the amount of bytes of memory it will /// take to run the step. Zero means no limit. /// @@ -51,6 +50,9 @@ state: State, /// total system memory available. max_rss: usize, +state: State, +pending_deps: u32, + result_error_msgs: ArrayList([]const u8), result_error_bundle: std.zig.ErrorBundle, result_stderr: []const u8, @@ -129,7 +131,6 @@ pub const State = enum { /// file system inputs have been modified, meaning that the step needs to /// be re-evaluated. precheck_done, - running, dependency_failure, success, failure, @@ -242,6 +243,7 @@ pub fn init(options: StepOptions) Step { .dependants = .empty, .inputs = Inputs.init, .state = .precheck_unstarted, + .pending_deps = undefined, // initialized by build runner .max_rss = options.max_rss, .debug_stack_trace = blk: { const addr_buf = arena.alloc(usize, options.owner.debug_stack_frames_count) catch @panic("OOM"); @@ -980,14 +982,17 @@ pub fn reset(step: *Step, gpa: Allocator) void { } /// Implementation detail of file watching. Prepares the step for being re-evaluated. -pub fn recursiveReset(step: *Step, gpa: Allocator) void { - assert(step.state != .precheck_done); +/// Returns `true` if the step was newly invalidated, `false` if it was already invalidated. +pub fn invalidateResult(step: *Step, gpa: Allocator) bool { + if (step.state == .precheck_done) return false; + assert(step.pending_deps == 0); step.state = .precheck_done; step.reset(gpa); - for (step.dependants.items) |dep| { - if (dep.state == .precheck_done) continue; - dep.recursiveReset(gpa); + for (step.dependants.items) |dependant| { + _ = dependant.invalidateResult(gpa); + dependant.pending_deps += 1; } + return true; } test { diff --git a/lib/std/Build/Watch.zig b/lib/std/Build/Watch.zig index 8e8ef732a6b72724c7b15feb028fe4fe0515ff72..c4ac62b216604a89ed3a91c078bbe17bca8826ce 100644 --- a/lib/std/Build/Watch.zig +++ b/lib/std/Build/Watch.zig @@ -901,7 +901,7 @@ fn markAllFilesDirty(w: *Watch, gpa: Allocator) void { }; for (reaction_set.values()) |step_set| { for (step_set.keys()) |step| { - step.recursiveReset(gpa); + _ = step.invalidateResult(gpa); } } } @@ -910,10 +910,7 @@ fn markAllFilesDirty(w: *Watch, gpa: Allocator) void { fn markStepSetDirty(gpa: Allocator, step_set: *StepSet, any_dirty: bool) bool { var this_any_dirty = false; for (step_set.keys()) |step| { - if (step.state != .precheck_done) { - step.recursiveReset(gpa); - this_any_dirty = true; - } + if (step.invalidateResult(gpa)) this_any_dirty = true; } return any_dirty or this_any_dirty; } diff --git a/lib/std/Build/Watch/FsEvents.zig b/lib/std/Build/Watch/FsEvents.zig index b23805c7ee14a459a69adea27ebe68be3371276f..c9d2ed219c7e926b3cefd80b61d5ca5bf8945e02 100644 --- a/lib/std/Build/Watch/FsEvents.zig +++ b/lib/std/Build/Watch/FsEvents.zig @@ -349,13 +349,17 @@ fn eventCallback( false => { if (fse.watch_paths.get(event_path)) |steps| { assert(steps.len > 0); - for (steps) |s| dirtyStep(s, gpa, &any_dirty); + for (steps) |s| { + if (s.invalidateResult(gpa)) any_dirty = true; + } } if (std.fs.path.dirname(event_path)) |event_dirname| { // Modifying '/foo/bar' triggers the watch on '/foo'. if (fse.watch_paths.get(event_dirname)) |steps| { assert(steps.len > 0); - for (steps) |s| dirtyStep(s, gpa, &any_dirty); + for (steps) |s| { + if (s.invalidateResult(gpa)) any_dirty = true; + } } } }, @@ -368,7 +372,9 @@ fn eventCallback( const changed_path = std.fs.path.dirname(event_path) orelse event_path; for (fse.watch_paths.keys(), fse.watch_paths.values()) |watching_path, steps| { if (dirStartsWith(watching_path, changed_path)) { - for (steps) |s| dirtyStep(s, gpa, &any_dirty); + for (steps) |s| { + if (s.invalidateResult(gpa)) any_dirty = true; + } } } }, @@ -379,11 +385,6 @@ fn eventCallback( _ = dispatch_semaphore_signal(fse.waiting_semaphore); } } -fn dirtyStep(s: *std.Build.Step, gpa: Allocator, any_dirty: *bool) void { - if (s.state == .precheck_done) return; - s.recursiveReset(gpa); - any_dirty.* = true; -} fn dirStartsWith(path: []const u8, prefix: []const u8) bool { if (std.mem.eql(u8, path, prefix)) return true; if (!std.mem.startsWith(u8, path, prefix)) return false;