authorgravatar for justus@klausecker.deJustus Klausecker <justus@klausecker.de> 2026-04-17 13:03:55+02:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-17 19:19:02+02:00
loga43973b3361946e233ea679ff6628b1bfc4579d1
tree906fe69c00fd346eeefed129ac45ddfd325e646c
parent21980c82f48f239e50239d5af264706d15829268

std.Io.Semaphore: add `waitTimeout`

Returns `error.Timeout` if provided timeout expires before a permit is available. Also adds/reworks tests for all wait functions.

1 files changed, 112 insertions(+), 17 deletions(-)

lib/std/Io/Semaphore.zig+112-17
...@@ -15,14 +15,42 @@ cond: Io.Condition = .init,...@@ -15,14 +15,42 @@ cond: Io.Condition = .init,
15/// It is OK to initialize this field to any value.15/// It is OK to initialize this field to any value.
16permits: usize = 0,16permits: usize = 0,
1717
18/// Blocks until a `permit` is available and consumes a single one.
19/// Unblocks without consuming a `permit` when canceled.
20///
21/// See also:
22/// * `waitTimeout`
23/// * `waitUncancelable`
18pub fn wait(s: *Semaphore, io: Io) Io.Cancelable!void {24pub fn wait(s: *Semaphore, io: Io) Io.Cancelable!void {
25 s.waitTimeout(io, .none) catch |err| switch (err) {
26 error.Timeout => unreachable,
27 error.Canceled => |e| return e,
28 };
29}
30
31pub const WaitTimeoutError = Io.Cancelable || Io.Timeout.Error;
32
33/// Blocks until a `permit` is available and consumes a single one.
34/// Unblocks without consuming a `permit` when canceled or when the provided
35/// timeout expires before a `permit` is available.
36///
37/// See also:
38/// * `wait`
39/// * `waitUncancelable`
40pub fn waitTimeout(s: *Semaphore, io: Io, timeout: Io.Timeout) WaitTimeoutError!void {
41 const deadline = timeout.toDeadline(io);
19 try s.mutex.lock(io);42 try s.mutex.lock(io);
20 defer s.mutex.unlock(io);43 defer s.mutex.unlock(io);
21 while (s.permits == 0) try s.cond.wait(io, &s.mutex);44 while (s.permits == 0) try s.cond.waitTimeout(io, &s.mutex, deadline);
22 s.permits -= 1;45 s.permits -= 1;
23 if (s.permits > 0) s.cond.signal(io);46 if (s.permits > 0) s.cond.signal(io);
24}47}
2548
49/// Blocks until a `permit` is available and consumes a single one.
50///
51/// See also:
52/// * `wait`
53/// * `waitTimeout`
26pub fn waitUncancelable(s: *Semaphore, io: Io) void {54pub fn waitUncancelable(s: *Semaphore, io: Io) void {
27 s.mutex.lockUncancelable(io);55 s.mutex.lockUncancelable(io);
28 defer s.mutex.unlock(io);56 defer s.mutex.unlock(io);
...@@ -31,6 +59,7 @@ pub fn waitUncancelable(s: *Semaphore, io: Io) void {...@@ -31,6 +59,7 @@ pub fn waitUncancelable(s: *Semaphore, io: Io) void {
31 if (s.permits > 0) s.cond.signal(io);59 if (s.permits > 0) s.cond.signal(io);
32}60}
3361
62/// Makes an additional `permit` available.
34pub fn post(s: *Semaphore, io: Io) void {63pub fn post(s: *Semaphore, io: Io) void {
35 s.mutex.lockUncancelable(io);64 s.mutex.lockUncancelable(io);
36 defer s.mutex.unlock(io);65 defer s.mutex.unlock(io);
...@@ -39,27 +68,93 @@ pub fn post(s: *Semaphore, io: Io) void {...@@ -39,27 +68,93 @@ pub fn post(s: *Semaphore, io: Io) void {
39 s.cond.signal(io);68 s.cond.signal(io);
40}69}
4170
42test Semaphore {71test wait {
43 if (builtin.single_threaded) return error.SkipZigTest;
44 const io = testing.io;72 const io = testing.io;
4573
46 const TestContext = struct {74 const Context = struct {
47 sem: *Semaphore,75 sem: Semaphore = .{ .permits = 1 },
48 n: *i32,76 n: u32 = 0,
77
49 fn worker(ctx: *@This()) !void {78 fn worker(ctx: *@This()) !void {
50 try ctx.sem.wait(io);79 try ctx.sem.wait(io);
51 ctx.n.* += 1;80 ctx.n += 1;
52 ctx.sem.post(io);81 ctx.sem.post(io);
53 }82 }
54 };83 };
55 const num_threads = 3;84
56 var sem: Semaphore = .{ .permits = 1 };85 var ctx: Context = .{};
57 var threads: [num_threads]std.Thread = undefined;86
58 var n: i32 = 0;87 var group: Io.Group = .init;
59 var ctx = TestContext{ .sem = &sem, .n = &n };88 defer group.cancel(io);
6089
61 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, TestContext.worker, .{&ctx});90 const num_workers = 3;
62 for (threads) |t| t.join();91 for (0..num_workers) |_| group.async(io, Context.worker, .{&ctx});
63 try sem.wait(io);92
64 try testing.expect(n == num_threads);93 try group.await(io);
94 try testing.expectEqual(num_workers, ctx.n);
95}
96
97test waitTimeout {
98 const io = testing.io;
99
100 const Context = struct {
101 ready: Io.Event = .unset,
102 sem: Semaphore = .{ .permits = 0 },
103 value: u32 = 0,
104
105 fn worker(ctx: *@This()) !void {
106 defer ctx.ready.set(io);
107
108 try testing.expectError(error.Timeout, ctx.sem.waitTimeout(io, .{ .duration = .{
109 .raw = .fromMilliseconds(1),
110 .clock = .awake,
111 } }));
112 try testing.expectEqual(0, ctx.value);
113
114 ctx.ready.set(io);
115
116 while (ctx.value == 0) try ctx.sem.wait(io);
117 try testing.expectEqual(1, ctx.value);
118 }
119 };
120
121 var ctx: Context = .{};
122
123 var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) {
124 error.ConcurrencyUnavailable => return error.SkipZigTest,
125 };
126 defer future.cancel(io) catch {};
127
128 try ctx.ready.wait(io);
129
130 ctx.value = 1;
131 ctx.sem.post(io);
132
133 try future.await(io);
134}
135
136test waitUncancelable {
137 const io = testing.io;
138
139 const Context = struct {
140 sem: Semaphore = .{ .permits = 1 },
141 n: u32 = 0,
142
143 fn worker(ctx: *@This()) !void {
144 ctx.sem.waitUncancelable(io);
145 ctx.n += 1;
146 ctx.sem.post(io);
147 }
148 };
149
150 var ctx: Context = .{};
151
152 var group: Io.Group = .init;
153 defer group.cancel(io);
154
155 const num_workers = 3;
156 for (0..num_workers) |_| group.async(io, Context.worker, .{&ctx});
157
158 try group.await(io);
159 try testing.expectEqual(num_workers, ctx.n);
65}160}