authorgravatar for kbutcher6200@gmail.comkprotty <kbutcher6200@gmail.com> 2019-11-23 10:22:16-06:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-11-26 20:40:04-05:00
log9bce97a479e70f2d8e09047c9a0c93690cd8fd99
tree08420296f604a6de1f2c48308fd170ced5838125
parent8ecd6c4d8c021f7778b4959bdf75204dfd2d1946
signaturelock-open Commit is signed but in an unrecognized format.

Start on ResetEvent

2 files changed, 341 insertions(+), 0 deletions(-)

lib/std/c.zig+1
......@@ -220,6 +220,7 @@ pub extern "c" fn pthread_mutex_destroy(mutex: *pthread_mutex_t) c_int;
220220
221221pub const PTHREAD_COND_INITIALIZER = pthread_cond_t{};
222222pub extern "c" fn pthread_cond_wait(noalias cond: *pthread_cond_t, noalias mutex: *pthread_mutex_t) c_int;
223pub extern "c" fn pthread_cond_timedwait(noalias: cond: *pthread_cond_t, noalias: mutex: *pthread_mutex_t, noalias abstime: *const timespec) c_int;
223224pub extern "c" fn pthread_cond_signal(cond: *pthread_cond_t) c_int;
224225pub extern "c" fn pthread_cond_destroy(cond: *pthread_cond_t) c_int;
225226
lib/std/reset_event.zig created+340
......@@ -0,0 +1,340 @@
1const std = @import("std.zig");
2const builtin = @import("builtin");
3const testing = std.testing;
4const assert = std.debug.assert;
5const Backoff = std.SpinLock.Backoff;
6const c = std.c;
7const time = std.time;
8const linux = std.os.linux;
9const windows = std.os.windows;
10
11/// A resource object which supports blocking until signaled.
12/// Once finished, the `deinit()` method should be called for correctness.
13pub const ResetEvent = struct {
14 os_event: OsEvent,
15
16 pub fn init() ResetEvent {
17 return ResetEvent{ .os_event = OsEvent.init() };
18 }
19
20 pub fn deinit(self: *ResetEvent) void {
21 self.os_event.deinit();
22 self.* = undefined;
23 }
24
25 /// Returns whether or not the event is currenetly set
26 pub fn isSet(self: *const ResetEvent) bool {
27 return self.os_event.isSet();
28 }
29
30 /// Sets the event if not already set and
31 /// wakes up AT LEAST one thread waiting the event.
32 /// Returns whether or not a thread was woken up.
33 pub fn set(self: *ResetEvent) bool {
34 return self.os_event.set();
35 }
36
37 /// Resets the event to its original, unset state.
38 /// Returns whether or not the event was currently set before un-setting.
39 pub fn reset(self: *ResetEvent) bool {
40 return self.os_event.reset();
41 }
42
43 const WaitError = error{
44 /// The thread blocked longer than the maximum time specified.
45 TimedOut,
46 };
47
48 /// Wait for the event to be set by blocking the current thread.
49 /// Optionally provided timeout in nanoseconds which throws an
50 /// `error.TimedOut` if the thread blocked AT LEAST longer than specified.
51 /// Returns whether or not the thread blocked from the event being unset at the time of calling.
52 pub fn wait(self: *ResetEvent, timeout_ns: ?u64) WaitError!bool {
53 return self.os_event.wait(timeout_ns);
54 }
55};
56
57const OsEvent = if (builtin.single_threaded) DebugEvent else switch (builtin.os) {
58 .windows => WindowsEvent,
59 .linux => if (builtin.link_libc) PosixEvent else LinuxEvent,
60 else => if (builtin.link_libc) PosixEvent else SpinEvent,
61};
62
63const DebugEvent = struct {
64 is_set: @typeOf(set_init),
65
66 const set_init = if (std.debug.runtime_safety) false else {};
67
68 pub fn init() DebugEvent {
69 return DebugEvent{ .is_set = set_init };
70 }
71
72 pub fn deinit(self: *DebugEvent) void {
73 self.* = undefined;
74 }
75
76 pub fn isSet(self: *const DebugEvent) bool {
77 if (!std.debug.runtime_safety)
78 return true;
79 return self.is_set;
80 }
81
82 pub fn set(self: *DebugEvent) bool {
83 if (std.debug.runtime_safety)
84 self.is_set = true;
85 return false;
86 }
87
88 pub fn reset(self: *DebugEvent) bool {
89 if (!std.debug.runtime_safety)
90 return false;
91 const was_set = self.is_set;
92 self.is_set = false;
93 return was_set;
94 }
95
96 pub fn wait(self: *DebugEvent, timeout: ?u64) ResetEvent.WaitError!bool {
97 if (std.debug.runtime_safety and !self.is_set)
98 @panic("deadlock detected");
99 return ResetEvent.WaitError.TimedOut;
100 }
101};
102
103fn EventState(comptime TagType: type) type {
104 return enum(TagType) {
105 Empty,
106 Waiting,
107 Signaled,
108 };
109}
110
111const SpinEvent = struct {
112 state: State,
113
114 const State = EventState(u8);
115
116 pub fn init() SpinEvent {
117 return SpinEvent{ .state = .Empty };
118 }
119
120 pub fn deinit(self: *SpinEvent) void {
121 self.* = undefined;
122 }
123
124 pub fn isSet(self: *const SpinEvent) bool {
125 return @atomicLoad(State, &self.state, .Acquire) == .Signaled;
126 }
127
128 pub fn set(self: *SpinEvent) bool {
129 return @atomicRmw(State, &self.state, .Xchg, .Signaled, .Release) == .Waiting;
130 }
131
132 pub fn reset(self: *SpinEvent) bool {
133 return @atomicRmw(State, &self.state, .Xchg, .Empty, .Monotonic) == .Signaled;
134 }
135
136 pub fn wait(self: *SpinEvent, timeout: ?u64) ResetEvent.WaitError!bool {
137 var state = @atomicLoad(State, &self.state, .Monotonic);
138 while (true) {
139 switch (state) {
140 .Empty => state = @cmpxchgWeak(State, &self.state, state, .Waiting, .Acquire, .Monotonic) orelse break,
141 .Waiting => break,
142 .Signaled => return false,
143 }
144 }
145
146 // TODO: handle case for time.Timer.start() fails
147 var spin = Backoff.init();
148 var timer = if (timeout == null) null else time.Timer.start() catch unreachable;
149 while (@atomicLoad(State, &self.state, .Monotonic) == .Waiting) {
150 spin.yield();
151 if (timeout) |timeout_ns| {
152 if (timer.?.read() > timeout_ns)
153 return ResetEvent.WaitError.TimedOut;
154 }
155 }
156 return true;
157 }
158};
159
160const LinuxEvent = struct {
161 state: State,
162
163 const State = EventState(i32);
164
165 pub fn init() LinuxEvent {
166 return LinuxEvent{ .state = .Empty };
167 }
168
169 pub fn deinit(self: *LinuxEvent) void {
170 self.* = undefined;
171 }
172
173 pub fn isSet(self: *const LinuxEvent) bool {
174 return @atomicLoad(State, &self.state, .Acquire) == .Signaled;
175 }
176
177 pub fn set(self: *LinuxEvent) bool {
178 if (@atomicRmw(State, &self.state, .Xchg, .Signaled, .Release) != .Waiting)
179 return false;
180 const rc = linux.futex_wake(@ptrCast(*const i32, &self.state), linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
181 assert(linux.getErrno(rc) == 0);
182 return true;
183 }
184
185 pub fn reset(self: *LinuxEvent) bool {
186 return @atomicRmw(State, &self.state, .Xchg, .Empty, .Monotonic) == .Signaled;
187 }
188
189 pub fn wait(self: *LinuxEvent, timeout: ?u64) ResetEvent.WaitError!bool {
190 var state = @atomicLoad(State, &self.state, .Monotonic);
191 while (true) {
192 switch (state) {
193 .Empty => state = @cmpxchgWeak(State, &self.state, .Empty, .Waiting, .Acquire, .Monotonic) orelse break,
194 .Waiting => break,
195 .Signaled => return false,
196 }
197 }
198
199 var ts: linux.timespec = undefined;
200 var ts_ptr: ?*linux.timespec = null;
201 if (timeout) |timeout_ns| {
202 ts_ptr = &ts;
203 ts.tv_sec = @intCast(isize, timeout_ns / time.ns_per_s);
204 ts.tv_nsec = @intCast(isize, timeout_ns % time.ns_per_s);
205 }
206
207 while (@atomicLoad(State, &self.state, .Monotonic) == .Waiting) {
208 const rc = linux.futex_wait(@ptrCast(*const i32, &self.state), linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, @enumToInt(State.Waiting), ts_ptr);
209 switch (linux.getErrno(rc)) {
210 0, linux.EINTR => continue,
211 linux.EAGAIN => break,
212 linux.ETIMEDOUT => return ResetEvent.WaitError.TimedOut,
213 else => unreachable,
214 }
215 }
216 }
217};
218
219const PosixEvent = struct {
220 state: State,
221 cond: c.pthread_cond_t,
222 mutex: c.pthread_mutex_t,
223
224 const State = EventState(u8);
225
226 pub fn init() PosixEvent {
227 return PosixEvent{
228 .state = .Empty,
229 .cond = c.PTHREAD_COND_INITIALIZER,
230 .mutex = c.PTHREAD_MUTEX_INITIALIZER,
231 };
232 }
233
234 pub fn deinit(self: *PosixEvent) void {
235 // On dragonfly, the destroy functions return EINVAL if they were initialized statically.
236 const retm = c.pthread_mutex_destroy(&self.mutex);
237 assert(retm == 0 or retm == (if (builtin.os == .dragonfly) std.os.EINVAL else 0));
238 const retc = c.pthread_cond_destroy(&self.cond);
239 assert(retc == 0 or retc == (if (builtin.os == .dragonfly) std.os.EINVAL else 0));
240 self.* = undefined;
241 }
242
243 pub fn isSet(self: *const PosixEvent) bool {
244 assert(c.pthread_mutex_lock(&self.mutex) == 0);
245 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
246
247 return self.state == .Signaled;
248 }
249
250 pub fn set(self: *PosixEvent) bool {
251 assert(c.pthread_mutex_lock(&self.mutex) == 0);
252 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
253
254 const woken = self.state == .Waiting;
255 self.state = .Signaled;
256 return woken;
257 }
258
259 pub fn reset(self: *PosixEvent) bool {
260 assert(c.pthread_mutex_lock(&self.mutex) == 0);
261 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
262
263 const was_set = self.state == .Signaled;
264 self.state = .Empty;
265 return was_set;
266 }
267
268 pub fn wait(self: *PosixEvent, timeout: ?u64) ResetEvent.WaitError!bool {
269 assert(c.pthread_mutex_lock(&self.mutex) == 0);
270 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
271
272 if (self.state == .Signaled)
273 return false;
274
275 var ts: std.os.timespec = undefined;
276 var ts_ptr = &ts;
277 if (timeout) |timeout_ns| {
278 var tv: std.os.timeval = undefined;
279 assert(c.gettimeofday(&tv, null) == 0);
280 ts.tv_sec = @intCast(isize, tv.tv_sec + (timeout_ns / time.ns_per_s));
281 ts.tv_nsec = @intCast(isize, (tv.tv_usec * time.microsecond) + (timeout_ns % time.ns_per_s));
282 }
283
284 self.state = .Waiting;
285 while (self.state == .Waiting) {
286 const rc = switch (timeout == null) {
287 true => c.pthread_cond_wait(&self.cond, &self.mutex),
288 else => c.pthread_cond_timedwait(&self.cond, &self.mutex, ts_ptr),
289 };
290 assert(rc == 0);
291 }
292 }
293};
294
295const WindowsEvent = struct {
296 state: State,
297
298 const State = EventState(u32);
299
300 pub fn init() WindowsEvent {
301 return WindowsEvent{ .state = .Empty };
302 }
303
304 pub fn deinit(self: *WindowsEvent) void {
305 self.* = undefined;
306 }
307
308 pub fn isSet(self: *const WindowsEvent) bool {
309 return @atomicLoad(State, &self.state, .Acquire) == .Signaled;
310 }
311
312 pub fn set(self: *WindowsEvent) bool {
313 if (@atomicRmw(State, &self.state, .Xchg, .Signaled, .Release) != .Waiting)
314 return false;
315
316 if (getEventHandle()) |handle| {
317 const key = @ptrCast(*const c_void, &self.state);
318 const rc = windows.ntdll.NtReleaseKeyedEvent(handle, key, windows.FALSE, null);
319 assert(rc == 0);
320 }
321 return true;
322 }
323
324 pub fn reset(self: *WindowsEvent) bool {
325 return @atomicRmw(State, &self.state, .Xchg, .Empty, .Monotonic) == .Signaled;
326 }
327
328 pub fn wait(self: *WindowsEvent, timeout: ?u64) ResetEvent.WaitError!bool {
329 var state = @atomicLoad(State, &self.state, .Monotonic);
330 while (true) {
331 switch (state) {
332 .Empty => state = @cmpxchgWeak(State, &self.state, .Empty, .Waiting, .Acquire, .Monotonic) orelse break,
333 .Waiting => break,
334 .Signaled => return false,
335 }
336 }
337
338 const timeout_ms = if (timeout @intCast(windows.LARGE_INTEGER, )
339 }
340};