authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-17 19:02:38+02:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-17 19:02:38+02:00
log21980c82f48f239e50239d5af264706d15829268
treed57a87dcda895fd99f4ee9d2608cf8b79df17abd
parentadb1f3efc21adad67fe2638c889178401ca602a1
parentc0763b5e257ac7c0d42877d17a2af2db0910221e

Merge pull request 'Implement `Condition.waitTimeout`' (#31278) from lukasl/zig:condition-timeout into master

Reviewed-on: https://codeberg.org/ziglang/zig/pulls/31278 Reviewed-by: Andrew Kelley <andrew@ziglang.org>

2 files changed, 158 insertions(+), 15 deletions(-)

lib/std/Io.zig+70-15
...@@ -1681,20 +1681,29 @@ pub const Condition = struct {...@@ -1681,20 +1681,29 @@ pub const Condition = struct {
1681 .epoch = .init(0),1681 .epoch = .init(0),
1682 };1682 };
16831683
1684 /// Blocks until the condition is signaled or canceled.
1685 ///
1686 /// See also:
1687 /// * `waitUncancelable`
1688 /// * `waitTimeout`
1684 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {1689 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) Cancelable!void {
1685 try waitInner(cond, io, mutex, false);1690 waitTimeout(cond, io, mutex, .none) catch |err| switch (err) {
1691 error.Timeout => unreachable,
1692 error.Canceled => |e| return e,
1693 };
1686 }1694 }
16871695
1688 /// Same as `wait`, except does not introduce a cancelation point.1696 pub const WaitTimeoutError = Cancelable || Timeout.Error;
1697
1698 /// Blocks until the condition is signaled, canceled, or the provided
1699 /// timeout expires.
1689 ///1700 ///
1690 /// For a description of cancelation and cancelation points, see `Future.cancel`.1701 /// See also:
1691 pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void {1702 /// * `wait`
1692 waitInner(cond, io, mutex, true) catch |err| switch (err) {1703 /// * `waitUncancelable`
1693 error.Canceled => unreachable,1704 pub fn waitTimeout(cond: *Condition, io: Io, mutex: *Mutex, timeout: Timeout) WaitTimeoutError!void {
1694 };1705 const deadline = timeout.toDeadline(io);
1695 }
16961706
1697 fn waitInner(cond: *Condition, io: Io, mutex: *Mutex, uncancelable: bool) Cancelable!void {
1698 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load1707 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
16991708
1700 {1709 {
...@@ -1706,10 +1715,7 @@ pub const Condition = struct {...@@ -1706,10 +1715,7 @@ pub const Condition = struct {
1706 defer mutex.lockUncancelable(io);1715 defer mutex.lockUncancelable(io);
17071716
1708 while (true) {1717 while (true) {
1709 const result = if (uncancelable)1718 const result = io.futexWaitTimeout(u32, &cond.epoch.raw, epoch, deadline);
1710 io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch)
1711 else
1712 io.futexWait(u32, &cond.epoch.raw, epoch);
17131719
1714 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod1720 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod
17151721
...@@ -1729,13 +1735,62 @@ pub const Condition = struct {...@@ -1729,13 +1735,62 @@ pub const Condition = struct {
1729 }1735 }
17301736
1731 // There are no more signals available; this was a spurious wakeup or an error. If it1737 // There are no more signals available; this was a spurious wakeup or an error. If it
1732 // was an error, we will remove ourselves as a waiter and return that error. Otherwise,1738 // was an error, we will remove ourselves as a waiter and return that error. If a
1733 // we'll loop back to the futex wait.1739 // timeout was specified and the deadline has passed, we remove ourselves as a waiter
1740 // and return `error.Timeout`. Otherwise, we'll loop back to the futex wait.
1734 result catch |err| {1741 result catch |err| {
1735 const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic);1742 const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic);
1736 assert(prev_state.waiters > 0); // underflow caused by illegal state1743 assert(prev_state.waiters > 0); // underflow caused by illegal state
1737 return err;1744 return err;
1738 };1745 };
1746 switch (deadline) {
1747 .none => {},
1748 .deadline => |d| if (d.untilNow(io).raw.nanoseconds >= 0) {
1749 const prev_state = cond.state.fetchSub(.{ .waiters = 1, .signals = 0 }, .monotonic);
1750 assert(prev_state.waiters > 0); // underflow caused by illegal state
1751 return error.Timeout;
1752 },
1753 .duration => unreachable,
1754 }
1755 }
1756 }
1757
1758 /// Same as `wait`, except does not introduce a cancelation point.
1759 ///
1760 /// See `Future.cancel` for a description of cancelation points.
1761 pub fn waitUncancelable(cond: *Condition, io: Io, mutex: *Mutex) void {
1762 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
1763
1764 {
1765 const prev_state = cond.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic);
1766 assert(prev_state.waiters < math.maxInt(u16)); // overflow caused by too many waiters
1767 }
1768
1769 mutex.unlock(io);
1770 defer mutex.lockUncancelable(io);
1771
1772 while (true) {
1773 io.futexWaitUncancelable(u32, &cond.epoch.raw, epoch);
1774
1775 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod
1776
1777 // Even on error, try to consume a pending signal first. Otherwise a race might
1778 // cause a signal to get stuck in the state with no corresponding waiter.
1779 {
1780 var prev_state = cond.state.load(.monotonic);
1781 while (prev_state.signals > 0) {
1782 prev_state = cond.state.cmpxchgWeak(prev_state, .{
1783 .waiters = prev_state.waiters - 1,
1784 .signals = prev_state.signals - 1,
1785 }, .acquire, .monotonic) orelse {
1786 // We successfully consumed a signal.
1787 return;
1788 };
1789 }
1790 }
1791
1792 // There are no more signals available; this was a spurious wakeup,
1793 // so we'll loop back to the futex wait.
1739 }1794 }
1740 }1795 }
17411796
lib/std/Io/test.zig+88
...@@ -971,3 +971,91 @@ test "Select.cancel with no tasks, no deadlock" {...@@ -971,3 +971,91 @@ test "Select.cancel with no tasks, no deadlock" {
971 var select: Io.Select(U) = .init(io, &.{});971 var select: Io.Select(U) = .init(io, &.{});
972 try expectEqual(null, select.cancel());972 try expectEqual(null, select.cancel());
973}973}
974
975test "Condition.waitTimeout" {
976 const io = testing.io;
977
978 const Context = struct {
979 ready: Io.Event = .unset,
980 mutex: Io.Mutex = .init,
981 cond: Io.Condition = .init,
982 value: u32 = 0,
983
984 fn worker(ctx: *@This()) !void {
985 defer ctx.ready.set(io);
986
987 try ctx.mutex.lock(io);
988 defer ctx.mutex.unlock(io);
989
990 try expectError(error.Timeout, ctx.cond.waitTimeout(io, &ctx.mutex, .{ .duration = .{
991 .raw = .fromMilliseconds(1),
992 .clock = .awake,
993 } }));
994 try expectEqual(0, ctx.value);
995
996 ctx.ready.set(io);
997
998 while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex);
999 try expectEqual(1, ctx.value);
1000 }
1001 };
1002
1003 var ctx: Context = .{};
1004
1005 var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) {
1006 error.ConcurrencyUnavailable => return error.SkipZigTest,
1007 };
1008 defer future.cancel(io) catch {};
1009
1010 try ctx.ready.wait(io);
1011
1012 try ctx.mutex.lock(io);
1013 ctx.value = 1;
1014 ctx.mutex.unlock(io);
1015 ctx.cond.signal(io);
1016
1017 try future.await(io);
1018}
1019
1020test "Condition.waitUncancelable" {
1021 const io = testing.io;
1022
1023 const Context = struct {
1024 ready: Io.Event = .unset,
1025 mutex: Io.Mutex = .init,
1026 cond: Io.Condition = .init,
1027 value: u32 = 0,
1028
1029 fn worker(ctx: *@This()) !void {
1030 defer ctx.ready.set(io);
1031
1032 try ctx.mutex.lock(io);
1033 defer ctx.mutex.unlock(io);
1034
1035 try expectEqual(0, ctx.value);
1036
1037 ctx.ready.set(io);
1038
1039 ctx.cond.waitUncancelable(io, &ctx.mutex);
1040
1041 while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex);
1042 try expectEqual(1, ctx.value);
1043 }
1044 };
1045
1046 var ctx: Context = .{};
1047
1048 var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) {
1049 error.ConcurrencyUnavailable => return error.SkipZigTest,
1050 };
1051 defer future.cancel(io) catch {};
1052
1053 try ctx.ready.wait(io);
1054
1055 try ctx.mutex.lock(io);
1056 ctx.value = 1;
1057 ctx.mutex.unlock(io);
1058 ctx.cond.signal(io);
1059
1060 try future.await(io);
1061}