archives/zig
1
0
Fork 0
mirror of https://codeberg.org/ziglang/zig.git synced 2026-03-08 02:24:33 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

build runner: refactor step evaluation logic

Browse source 
The previous logic was made really messy by the fact that upon entry to
the step eval worker, the step may not be ready to run, we may be racing
with other workers doing the same check, and we had already acquired our
RSS requirement even though we might not run. It also required iterating
all dependencies each time we were called to check whether we were even
ready to run yet.

A much better strategy is for each step to have an atomic counter
representing how many of its dependencies are yet to complete. When a
step completes (successfully or otherwise), it decrements this value for
all of its dependants, and if it drops any to 0, it schedules that step
to run. This means each step is scheduled exactly once, and only when
all of its dependencies have finished, reducing redundant checks and
hence contention. If the step being scheduled needs to claim RSS which
isn't available, then it is instead added to `memory_blocked_steps`,
which is iterated by the step worker after a step with an RSS claim
finishes.

This logic is more concise than before, simpler to understand, generally
more efficient, and fixes a bug in the RSS tracking. Also, as a nice
side effect, it should also play a little bit nicer with `Io.Threaded`'s
scheduling strategy, because we no longer spawn extremely short-lived
tasks all the time as we previously did.

Resolves: https://codeberg.org/ziglang/zig/issues/30742
This commit is contained in:
Matthew Lugg 2026-01-08 19:00:11 +00:00 • committed by mlugg
parent 04e73d03bd
commit aebd84b525
4 changed files with 168 additions and 182 deletions
Download patch file Download diff file Expand all files Collapse all files

307
lib/compiler/build_runner.zig
View file

@ -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;
},
{
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 => .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);
},
.success,
.skipped,
=> {
if (run.web_server) |*ws| ws.updateStepStatus(s, .success);
},
}
}
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,
};
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 {};
}
handle_result: {
if (make_result) |_| {
@atomicStore(Step.State, &s.state, .success, .seq_cst);
if (run.web_server) |*ws| ws.updateStepStatus(s, .success);
} else |err| switch (err) {
error.MakeFailed => {
@atomicStore(Step.State, &s.state, .failure, .seq_cst);
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);
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 });
}
}
// 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(

19
lib/std/Build/Step.zig
View file

@ -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 {

7
lib/std/Build/Watch.zig
View file

@ -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;
}

17
lib/std/Build/Watch/FsEvents.zig
View file

@ -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;