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