From d821446cf92b9b8974ec4a3e5d7be883d1741b4e Mon Sep 17 00:00:00 2001 From: Lukas Lalinsky Date: Thu, 19 Feb 2026 12:07:54 +0100 Subject: [PATCH 1/3] Implement `Condition.waitTimeout` I'd have preferred if `vtable.futexWait` returned `error.Timeout`, since all the OS-level APIs provide it. However, if I keep the vtable untouched, I had to determine the timeout case by post-checking the deadline. It's fine functionally, but one extra syscall that be avoided at cost of changing the vtable and all the futex implementations. --- lib/std/Io.zig | 34 +++++++++++++++++++++++++--------- lib/std/Io/test.zig | 44 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 9 deletions(-) diff --git a/lib/std/Io.zig b/lib/std/Io.zig index 03b33dae06d069496a800737c6891ab560e8db9f..0db3087abcf370cf46846057db4ba94253c513ea 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -1682,19 +1682,27 @@ pub const Condition = struct { }; pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void { - try waitInner(cond, io, mutex, false); + waitInner(cond, io, mutex, .{ .timeout = .none }) catch |err| switch (err) { + error.Timeout => unreachable, + error.Canceled => return error.Canceled, + }; + } + + pub fn waitTimeout(cond: *Condition, io: Io, mutex: *Mutex, timeout: Timeout) (Cancelable || Timeout.Error)!void { + return waitInner(cond, io, mutex, .{ .timeout = timeout.toDeadline(io) }); } /// Same as `wait`, except does not introduce a cancelation point. /// /// For a description of cancelation and cancelation points, see `Future.cancel`. pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void { - waitInner(cond, io, mutex, true) catch |err| switch (err) { + waitInner(cond, io, mutex, .uncancelable) catch |err| switch (err) { + error.Timeout => unreachable, error.Canceled => unreachable, }; } - fn waitInner(cond: *Condition, io: Io, mutex: *Mutex, uncancelable: bool) Cancelable!void { + fn waitInner(cond: *Condition, io: Io, mutex: *Mutex, mode: union(enum) { uncancelable, timeout: Timeout }) (Cancelable || Timeout.Error)!void { var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load { @@ -1706,10 +1714,10 @@ pub const Condition = struct { defer mutex.lockUncancelable(io); while (true) { - const result = if (uncancelable) - io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch) - else - io.futexWait(u32, &cond.epoch.raw, epoch); + const result = switch (mode) { + .uncancelable => io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch), + .timeout => |t| io.futexWaitTimeout(u32, &cond.epoch.raw, epoch, t), + }; epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod @@ -1729,13 +1737,21 @@ pub const Condition = struct { } // There are no more signals available; this was a spurious wakeup or an error. If it - // was an error, we will remove ourselves as a waiter and return that error. Otherwise, - // we'll loop back to the futex wait. + // was an error, we will remove ourselves as a waiter and return that error. If a + // timeout was specified and the deadline has passed, we remove ourselves as a waiter + // and return `error.Timeout`. Otherwise, we'll loop back to the futex wait. result catch |err| { const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic); assert(prev_state.waiters > 0); // underflow caused by illegal state return err; }; + if (mode == .timeout and mode.timeout != .none) { + if (mode.timeout.deadline.untilNow(io).raw.nanoseconds >= 0) { + const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic); + assert(prev_state.waiters > 0); // underflow caused by illegal state + return error.Timeout; + } + } } } diff --git a/lib/std/Io/test.zig b/lib/std/Io/test.zig index bd6b433d058540ec95225662a0a51038f005a971..37c6b78c39559e8ad1b279a650a78e848a33e52a 100644 --- a/lib/std/Io/test.zig +++ b/lib/std/Io/test.zig @@ -971,3 +971,47 @@ test "Select.cancel with no tasks, no deadlock" { var select: Io.Select(U) = .init(io, &.{}); try expectEqual(null, select.cancel()); } + +test "Condition" { + if (builtin.single_threaded) return error.SkipZigTest; + const io = testing.io; + + const TestContext = struct { + ready: Io.Event = .unset, + mutex: Io.Mutex = .init, + cond: Io.Condition = .init, + value: u32 = 0, + + fn worker(ctx: *@This()) !void { + defer ctx.ready.set(io); + + try ctx.mutex.lock(io); + defer ctx.mutex.unlock(io); + + try expectError(error.Timeout, ctx.cond.waitTimeout(io, &ctx.mutex, .{ .duration = .{ + .raw = .fromMilliseconds(1), + .clock = .awake, + } })); + try expectEqual(0, ctx.value); + + ctx.ready.set(io); + + while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex); + try expectEqual(1, ctx.value); + } + }; + + var ctx: TestContext = .{}; + + var future = try io.concurrent(TestContext.worker, .{&ctx}); + defer future.cancel(io) catch {}; + + try ctx.ready.wait(io); + + try ctx.mutex.lock(io); + ctx.value = 1; + ctx.mutex.unlock(io); + ctx.cond.signal(io); + + try future.await(io); +} -- 2.54.0 From 078185a54baf6093d3f43cd5b51c32322b34af75 Mon Sep 17 00:00:00 2001 From: Andrew Kelley Date: Thu, 16 Apr 2026 13:59:40 -0700 Subject: [PATCH 2/3] std.Io: still run Condition test on single-threaded this will still work under Evented for example --- lib/std/Io/test.zig | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/lib/std/Io/test.zig b/lib/std/Io/test.zig index 37c6b78c39559e8ad1b279a650a78e848a33e52a..cd40a7782be92755f130f79937a1b4325cf353c8 100644 --- a/lib/std/Io/test.zig +++ b/lib/std/Io/test.zig @@ -973,10 +973,9 @@ test "Select.cancel with no tasks, no deadlock" { } test "Condition" { - if (builtin.single_threaded) return error.SkipZigTest; const io = testing.io; - const TestContext = struct { + const Context = struct { ready: Io.Event = .unset, mutex: Io.Mutex = .init, cond: Io.Condition = .init, @@ -1001,9 +1000,11 @@ test "Condition" { } }; - var ctx: TestContext = .{}; + var ctx: Context = .{}; - var future = try io.concurrent(TestContext.worker, .{&ctx}); + var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; defer future.cancel(io) catch {}; try ctx.ready.wait(io); -- 2.54.0 From c0763b5e257ac7c0d42877d17a2af2db0910221e Mon Sep 17 00:00:00 2001 From: Andrew Kelley Date: Thu, 16 Apr 2026 14:51:25 -0700 Subject: [PATCH 3/3] std.Io.Condition: separate wait impls for clarity also: * add docs * add test coverage for waitUncancelable * explicit error set declaration WaitTimeoutError --- lib/std/Io.zig | 79 +++++++++++++++++++++++++++++++++------------ lib/std/Io/test.zig | 45 +++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 21 deletions(-) diff --git a/lib/std/Io.zig b/lib/std/Io.zig index 0db3087abcf370cf46846057db4ba94253c513ea..b44cbe8b4ea8fa025e66dbfa6eda4b8092e7cdce 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -1681,28 +1681,29 @@ pub const Condition = struct { .epoch = .init(0), }; + /// Blocks until the condition is signaled or canceled. + /// + /// See also: + /// * `waitUncancelable` + /// * `waitTimeout` pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void { - waitInner(cond, io, mutex, .{ .timeout = .none }) catch |err| switch (err) { + waitTimeout(cond, io, mutex, .none) catch |err| switch (err) { error.Timeout => unreachable, - error.Canceled => return error.Canceled, + error.Canceled => |e| return e, }; } - pub fn waitTimeout(cond: *Condition, io: Io, mutex: *Mutex, timeout: Timeout) (Cancelable || Timeout.Error)!void { - return waitInner(cond, io, mutex, .{ .timeout = timeout.toDeadline(io) }); - } + pub const WaitTimeoutError = Cancelable || Timeout.Error; - /// Same as `wait`, except does not introduce a cancelation point. + /// Blocks until the condition is signaled, canceled, or the provided + /// timeout expires. /// - /// For a description of cancelation and cancelation points, see `Future.cancel`. - pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void { - waitInner(cond, io, mutex, .uncancelable) catch |err| switch (err) { - error.Timeout => unreachable, - error.Canceled => unreachable, - }; - } + /// See also: + /// * `wait` + /// * `waitUncancelable` + pub fn waitTimeout(cond: *Condition, io: Io, mutex: *Mutex, timeout: Timeout) WaitTimeoutError!void { + const deadline = timeout.toDeadline(io); - fn waitInner(cond: *Condition, io: Io, mutex: *Mutex, mode: union(enum) { uncancelable, timeout: Timeout }) (Cancelable || Timeout.Error)!void { var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load { @@ -1714,10 +1715,7 @@ pub const Condition = struct { defer mutex.lockUncancelable(io); while (true) { - const result = switch (mode) { - .uncancelable => io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch), - .timeout => |t| io.futexWaitTimeout(u32, &cond.epoch.raw, epoch, t), - }; + const result = io.futexWaitTimeout(u32, &cond.epoch.raw, epoch, deadline); epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod @@ -1745,13 +1743,54 @@ pub const Condition = struct { assert(prev_state.waiters > 0); // underflow caused by illegal state return err; }; - if (mode == .timeout and mode.timeout != .none) { - if (mode.timeout.deadline.untilNow(io).raw.nanoseconds >= 0) { + switch (deadline) { + .none => {}, + .deadline => |d| if (d.untilNow(io).raw.nanoseconds >= 0) { const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic); assert(prev_state.waiters > 0); // underflow caused by illegal state return error.Timeout; + }, + .duration => unreachable, + } + } + } + + /// Same as `wait`, except does not introduce a cancelation point. + /// + /// See `Future.cancel` for a description of cancelation points. + pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void { + var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load + + { + const prev_state = cond.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic); + assert(prev_state.waiters < math.maxInt(u16)); // overflow caused by too many waiters + } + + mutex.unlock(io); + defer mutex.lockUncancelable(io); + + while (true) { + io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch); + + epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod + + // Even on error, try to consume a pending signal first. Otherwise a race might + // cause a signal to get stuck in the state with no corresponding waiter. + { + var prev_state = cond.state.load(.monotonic); + while (prev_state.signals > 0) { + prev_state = cond.state.cmpxchgWeak(prev_state, .{ + .waiters = prev_state.waiters - 1, + .signals = prev_state.signals - 1, + }, .acquire, .monotonic) orelse { + // We successfully consumed a signal. + return; + }; } } + + // There are no more signals available; this was a spurious wakeup, + // so we'll loop back to the futex wait. } } diff --git a/lib/std/Io/test.zig b/lib/std/Io/test.zig index cd40a7782be92755f130f79937a1b4325cf353c8..20426acb9c82f357bdff97940148028cddd91d2c 100644 --- a/lib/std/Io/test.zig +++ b/lib/std/Io/test.zig @@ -972,7 +972,7 @@ test "Select.cancel with no tasks, no deadlock" { try expectEqual(null, select.cancel()); } -test "Condition" { +test "Condition.waitTimeout" { const io = testing.io; const Context = struct { @@ -1016,3 +1016,46 @@ test "Condition" { try future.await(io); } + +test "Condition.waitUncancelable" { + const io = testing.io; + + const Context = struct { + ready: Io.Event = .unset, + mutex: Io.Mutex = .init, + cond: Io.Condition = .init, + value: u32 = 0, + + fn worker(ctx: *@This()) !void { + defer ctx.ready.set(io); + + try ctx.mutex.lock(io); + defer ctx.mutex.unlock(io); + + try expectEqual(0, ctx.value); + + ctx.ready.set(io); + + ctx.cond.waitUncancelable(io, &ctx.mutex); + + while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex); + try expectEqual(1, ctx.value); + } + }; + + var ctx: Context = .{}; + + var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + }; + defer future.cancel(io) catch {}; + + try ctx.ready.wait(io); + + try ctx.mutex.lock(io); + ctx.value = 1; + ctx.mutex.unlock(io); + ctx.cond.signal(io); + + try future.await(io); +} -- 2.54.0