authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-11-26 21:08:25-05:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-11-26 21:08:25-05:00
log63300a21ddf4cfe209a39796c6d7ea7773e14fd6
treeef8094a6f8bbfd6cc4753c936714496954c69698
parent8ecd6c4d8c021f7778b4959bdf75204dfd2d1946
parentff445814cbf909db79193ce5815279eb074246fe
signaturelock-open Commit is signed but in an unrecognized format.

Merge branch 'kprotty-event_lock'

closes #3751

5 files changed, 520 insertions(+), 228 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/mutex.zig+85-47
......@@ -1,13 +1,12 @@
11const std = @import("std.zig");
22const builtin = @import("builtin");
33const testing = std.testing;
4const SpinLock = std.SpinLock;
5const ThreadParker = std.ThreadParker;
4const ResetEvent = std.ResetEvent;
65
76/// Lock may be held only once. If the same thread
87/// tries to acquire the same mutex twice, it deadlocks.
9/// This type supports static initialization and is based off of Golang 1.13 runtime.lock_futex:
10/// https://github.com/golang/go/blob/master/src/runtime/lock_futex.go
8/// This type supports static initialization and is based off of Webkit's WTF Lock (via rust parking_lot)
9/// https://github.com/Amanieu/parking_lot/blob/master/core/src/word_lock.rs
1110/// When an application is built in single threaded release mode, all the functions are
1211/// no-ops. In single threaded debug mode, there is deadlock detection.
1312pub const Mutex = if (builtin.single_threaded)
......@@ -39,80 +38,119 @@ pub const Mutex = if (builtin.single_threaded)
3938 }
4039else
4140 struct {
42 state: State, // TODO: make this an enum
43 parker: ThreadParker,
41 state: usize,
4442
45 const State = enum(u32) {
46 Unlocked,
47 Sleeping,
48 Locked,
49 };
43 const MUTEX_LOCK: usize = 1 << 0;
44 const QUEUE_LOCK: usize = 1 << 1;
45 const QUEUE_MASK: usize = ~(MUTEX_LOCK | QUEUE_LOCK);
46 const QueueNode = std.atomic.Stack(ResetEvent).Node;
5047
5148 /// number of iterations to spin yielding the cpu
5249 const SPIN_CPU = 4;
5350
54 /// number of iterations to perform in the cpu yield loop
51 /// number of iterations to spin in the cpu yield loop
5552 const SPIN_CPU_COUNT = 30;
5653
5754 /// number of iterations to spin yielding the thread
5855 const SPIN_THREAD = 1;
5956
6057 pub fn init() Mutex {
61 return Mutex{
62 .state = .Unlocked,
63 .parker = ThreadParker.init(),
64 };
58 return Mutex{ .state = 0 };
6559 }
6660
6761 pub fn deinit(self: *Mutex) void {
68 self.parker.deinit();
62 self.* = undefined;
6963 }
7064
7165 pub const Held = struct {
7266 mutex: *Mutex,
7367
7468 pub fn release(self: Held) void {
75 switch (@atomicRmw(State, &self.mutex.state, .Xchg, .Unlocked, .Release)) {
76 .Locked => {},
77 .Sleeping => self.mutex.parker.unpark(@ptrCast(*const u32, &self.mutex.state)),
78 .Unlocked => unreachable, // unlocking an unlocked mutex
79 else => unreachable, // should never be anything else
69 // since MUTEX_LOCK is the first bit, we can use (.Sub) instead of (.And, ~MUTEX_LOCK).
70 // this is because .Sub may be implemented more efficiently than the latter
71 // (e.g. `lock xadd` vs `cmpxchg` loop on x86)
72 const state = @atomicRmw(usize, &self.mutex.state, .Sub, MUTEX_LOCK, .Release);
73 if ((state & QUEUE_MASK) != 0 and (state & QUEUE_LOCK) == 0) {
74 self.mutex.releaseSlow(state);
8075 }
8176 }
8277 };
8378
8479 pub fn acquire(self: *Mutex) Held {
85 // Try and speculatively grab the lock.
86 // If it fails, the state is either Locked or Sleeping
87 // depending on if theres a thread stuck sleeping below.
88 var state = @atomicRmw(State, &self.state, .Xchg, .Locked, .Acquire);
89 if (state == .Unlocked)
90 return Held{ .mutex = self };
80 // fast path close to SpinLock fast path
81 if (@cmpxchgWeak(usize, &self.state, 0, MUTEX_LOCK, .Acquire, .Monotonic)) |current_state| {
82 self.acquireSlow(current_state);
83 }
84 return Held{ .mutex = self };
85 }
9186
87 fn acquireSlow(self: *Mutex, current_state: usize) void {
88 var spin: usize = 0;
89 var state = current_state;
9290 while (true) {
93 // try and acquire the lock using cpu spinning on failure
94 var spin: usize = 0;
95 while (spin < SPIN_CPU) : (spin += 1) {
96 var value = @atomicLoad(State, &self.state, .Monotonic);
97 while (value == .Unlocked)
98 value = @cmpxchgWeak(State, &self.state, .Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
99 SpinLock.yield(SPIN_CPU_COUNT);
91
92 // try and acquire the lock if unlocked
93 if ((state & MUTEX_LOCK) == 0) {
94 state = @cmpxchgWeak(usize, &self.state, state, state | MUTEX_LOCK, .Acquire, .Monotonic) orelse return;
95 continue;
96 }
97
98 // spin only if the waiting queue isn't empty and when it hasn't spun too much already
99 if ((state & QUEUE_MASK) == 0 and spin < SPIN_CPU + SPIN_THREAD) {
100 if (spin < SPIN_CPU) {
101 std.SpinLock.yield(SPIN_CPU_COUNT);
102 } else {
103 std.os.sched_yield() catch std.time.sleep(0);
104 }
105 state = @atomicLoad(usize, &self.state, .Monotonic);
106 continue;
100107 }
101108
102 // try and acquire the lock using thread rescheduling on failure
103 spin = 0;
104 while (spin < SPIN_THREAD) : (spin += 1) {
105 var value = @atomicLoad(State, &self.state, .Monotonic);
106 while (value == .Unlocked)
107 value = @cmpxchgWeak(State, &self.state, .Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
108 std.os.sched_yield() catch std.time.sleep(1);
109 // thread should block, try and add this event to the waiting queue
110 var node = QueueNode{
111 .next = @intToPtr(?*QueueNode, state & QUEUE_MASK),
112 .data = ResetEvent.init(),
113 };
114 defer node.data.deinit();
115 const new_state = @ptrToInt(&node) | (state & ~QUEUE_MASK);
116 state = @cmpxchgWeak(usize, &self.state, state, new_state, .Release, .Monotonic) orelse {
117 // node is in the queue, wait until a `held.release()` wakes us up.
118 _ = node.data.wait(null) catch unreachable;
119 spin = 0;
120 state = @atomicLoad(usize, &self.state, .Monotonic);
121 continue;
122 };
123 }
124 }
125
126 fn releaseSlow(self: *Mutex, current_state: usize) void {
127 // grab the QUEUE_LOCK in order to signal a waiting queue node's event.
128 var state = current_state;
129 while (true) {
130 if ((state & QUEUE_LOCK) != 0 or (state & QUEUE_MASK) == 0)
131 return;
132 state = @cmpxchgWeak(usize, &self.state, state, state | QUEUE_LOCK, .Acquire, .Monotonic) orelse break;
133 }
134
135 while (true) {
136 // barrier needed to observe incoming state changes
137 defer @fence(.Acquire);
138
139 // the mutex is currently locked. try to unset the QUEUE_LOCK and let the locker wake up the next node.
140 // avoids waking up multiple sleeping threads which try to acquire the lock again which increases contention.
141 if ((state & MUTEX_LOCK) != 0) {
142 state = @cmpxchgWeak(usize, &self.state, state, state & ~QUEUE_LOCK, .Release, .Monotonic) orelse return;
143 continue;
109144 }
110145
111 // failed to acquire the lock, go to sleep until woken up by `Held.release()`
112 if (@atomicRmw(State, &self.state, .Xchg, .Sleeping, .Acquire) == .Unlocked)
113 return Held{ .mutex = self };
114 state = .Sleeping;
115 self.parker.park(@ptrCast(*const u32, &self.state), @enumToInt(State.Sleeping));
146 // try to pop the top node on the waiting queue stack to wake it up
147 // while at the same time unsetting the QUEUE_LOCK.
148 const node = @intToPtr(*QueueNode, state & QUEUE_MASK);
149 const new_state = @ptrToInt(node.next) | (state & MUTEX_LOCK);
150 state = @cmpxchgWeak(usize, &self.state, state, new_state, .Release, .Monotonic) orelse {
151 _ = node.data.set(false);
152 return;
153 };
116154 }
117155 }
118156 };
lib/std/parker.zig deleted-180
......@@ -1,180 +0,0 @@
1const std = @import("std.zig");
2const builtin = @import("builtin");
3const time = std.time;
4const testing = std.testing;
5const assert = std.debug.assert;
6const SpinLock = std.SpinLock;
7const linux = std.os.linux;
8const windows = std.os.windows;
9
10pub const ThreadParker = switch (builtin.os) {
11 .linux => if (builtin.link_libc) PosixParker else LinuxParker,
12 .windows => WindowsParker,
13 else => if (builtin.link_libc) PosixParker else SpinParker,
14};
15
16const SpinParker = struct {
17 pub fn init() SpinParker {
18 return SpinParker{};
19 }
20 pub fn deinit(self: *SpinParker) void {}
21
22 pub fn unpark(self: *SpinParker, ptr: *const u32) void {}
23
24 pub fn park(self: *SpinParker, ptr: *const u32, expected: u32) void {
25 var backoff = SpinLock.Backoff.init();
26 while (@atomicLoad(u32, ptr, .Acquire) == expected)
27 backoff.yield();
28 }
29};
30
31const LinuxParker = struct {
32 pub fn init() LinuxParker {
33 return LinuxParker{};
34 }
35 pub fn deinit(self: *LinuxParker) void {}
36
37 pub fn unpark(self: *LinuxParker, ptr: *const u32) void {
38 const rc = linux.futex_wake(@ptrCast(*const i32, ptr), linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
39 assert(linux.getErrno(rc) == 0);
40 }
41
42 pub fn park(self: *LinuxParker, ptr: *const u32, expected: u32) void {
43 const value = @intCast(i32, expected);
44 while (@atomicLoad(u32, ptr, .Acquire) == expected) {
45 const rc = linux.futex_wait(@ptrCast(*const i32, ptr), linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, value, null);
46 switch (linux.getErrno(rc)) {
47 0, linux.EAGAIN => return,
48 linux.EINTR => continue,
49 linux.EINVAL => unreachable,
50 else => continue,
51 }
52 }
53 }
54};
55
56const WindowsParker = struct {
57 waiters: u32,
58
59 pub fn init() WindowsParker {
60 return WindowsParker{ .waiters = 0 };
61 }
62 pub fn deinit(self: *WindowsParker) void {}
63
64 pub fn unpark(self: *WindowsParker, ptr: *const u32) void {
65 const key = @ptrCast(*const c_void, ptr);
66 const handle = getEventHandle() orelse return;
67
68 var waiting = @atomicLoad(u32, &self.waiters, .Monotonic);
69 while (waiting != 0) {
70 waiting = @cmpxchgWeak(u32, &self.waiters, waiting, waiting - 1, .Acquire, .Monotonic) orelse {
71 const rc = windows.ntdll.NtReleaseKeyedEvent(handle, key, windows.FALSE, null);
72 assert(rc == 0);
73 return;
74 };
75 }
76 }
77
78 pub fn park(self: *WindowsParker, ptr: *const u32, expected: u32) void {
79 var spin = SpinLock.Backoff.init();
80 const ev_handle = getEventHandle();
81 const key = @ptrCast(*const c_void, ptr);
82
83 while (@atomicLoad(u32, ptr, .Monotonic) == expected) {
84 if (ev_handle) |handle| {
85 _ = @atomicRmw(u32, &self.waiters, .Add, 1, .Release);
86 const rc = windows.ntdll.NtWaitForKeyedEvent(handle, key, windows.FALSE, null);
87 assert(rc == 0);
88 } else {
89 spin.yield();
90 }
91 }
92 }
93
94 var event_handle = std.lazyInit(windows.HANDLE);
95
96 fn getEventHandle() ?windows.HANDLE {
97 if (event_handle.get()) |handle_ptr|
98 return handle_ptr.*;
99 defer event_handle.resolve();
100
101 const access_mask = windows.GENERIC_READ | windows.GENERIC_WRITE;
102 if (windows.ntdll.NtCreateKeyedEvent(&event_handle.data, access_mask, null, 0) != 0)
103 return null;
104 return event_handle.data;
105 }
106};
107
108const PosixParker = struct {
109 cond: c.pthread_cond_t,
110 mutex: c.pthread_mutex_t,
111
112 const c = std.c;
113
114 pub fn init() PosixParker {
115 return PosixParker{
116 .cond = c.PTHREAD_COND_INITIALIZER,
117 .mutex = c.PTHREAD_MUTEX_INITIALIZER,
118 };
119 }
120
121 pub fn deinit(self: *PosixParker) void {
122 // On dragonfly, the destroy functions return EINVAL if they were initialized statically.
123 const retm = c.pthread_mutex_destroy(&self.mutex);
124 assert(retm == 0 or retm == (if (builtin.os == .dragonfly) os.EINVAL else 0));
125 const retc = c.pthread_cond_destroy(&self.cond);
126 assert(retc == 0 or retc == (if (builtin.os == .dragonfly) os.EINVAL else 0));
127 }
128
129 pub fn unpark(self: *PosixParker, ptr: *const u32) void {
130 assert(c.pthread_mutex_lock(&self.mutex) == 0);
131 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
132 assert(c.pthread_cond_signal(&self.cond) == 0);
133 }
134
135 pub fn park(self: *PosixParker, ptr: *const u32, expected: u32) void {
136 assert(c.pthread_mutex_lock(&self.mutex) == 0);
137 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
138 while (@atomicLoad(u32, ptr, .Acquire) == expected)
139 assert(c.pthread_cond_wait(&self.cond, &self.mutex) == 0);
140 }
141};
142
143test "std.ThreadParker" {
144 if (builtin.single_threaded)
145 return error.SkipZigTest;
146
147 const Context = struct {
148 parker: ThreadParker,
149 data: u32,
150
151 fn receiver(self: *@This()) void {
152 self.parker.park(&self.data, 0); // receives 1
153 assert(@atomicRmw(u32, &self.data, .Xchg, 2, .SeqCst) == 1); // sends 2
154 self.parker.unpark(&self.data); // wakes up waiters on 2
155 self.parker.park(&self.data, 2); // receives 3
156 assert(@atomicRmw(u32, &self.data, .Xchg, 4, .SeqCst) == 3); // sends 4
157 self.parker.unpark(&self.data); // wakes up waiters on 4
158 }
159
160 fn sender(self: *@This()) void {
161 assert(@atomicRmw(u32, &self.data, .Xchg, 1, .SeqCst) == 0); // sends 1
162 self.parker.unpark(&self.data); // wakes up waiters on 1
163 self.parker.park(&self.data, 1); // receives 2
164 assert(@atomicRmw(u32, &self.data, .Xchg, 3, .SeqCst) == 2); // sends 3
165 self.parker.unpark(&self.data); // wakes up waiters on 3
166 self.parker.park(&self.data, 3); // receives 4
167 }
168 };
169
170 var context = Context{
171 .parker = ThreadParker.init(),
172 .data = 0,
173 };
174 defer context.parker.deinit();
175
176 var receiver = try std.Thread.spawn(&context, Context.receiver);
177 defer receiver.wait();
178
179 context.sender();
180}
lib/std/reset_event.zig created+433
......@@ -0,0 +1,433 @@
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 os = std.os;
8const time = std.time;
9const linux = os.linux;
10const windows = os.windows;
11
12/// A resource object which supports blocking until signaled.
13/// Once finished, the `deinit()` method should be called for correctness.
14pub const ResetEvent = struct {
15 os_event: OsEvent,
16
17 pub fn init() ResetEvent {
18 return ResetEvent{ .os_event = OsEvent.init() };
19 }
20
21 pub fn deinit(self: *ResetEvent) void {
22 self.os_event.deinit();
23 self.* = undefined;
24 }
25
26 /// Returns whether or not the event is currenetly set
27 pub fn isSet(self: *ResetEvent) bool {
28 return self.os_event.isSet();
29 }
30
31 /// Sets the event if not already set and
32 /// wakes up AT LEAST one thread waiting the event.
33 /// Returns whether or not a thread was woken up.
34 pub fn set(self: *ResetEvent, auto_reset: bool) bool {
35 return self.os_event.set(auto_reset);
36 }
37
38 /// Resets the event to its original, unset state.
39 /// Returns whether or not the event was currently set before un-setting.
40 pub fn reset(self: *ResetEvent) bool {
41 return self.os_event.reset();
42 }
43
44 const WaitError = error{
45 /// The thread blocked longer than the maximum time specified.
46 TimedOut,
47 };
48
49 /// Wait for the event to be set by blocking the current thread.
50 /// Optionally provided timeout in nanoseconds which throws an
51 /// `error.TimedOut` if the thread blocked AT LEAST longer than specified.
52 /// Returns whether or not the thread blocked from the event being unset at the time of calling.
53 pub fn wait(self: *ResetEvent, timeout_ns: ?u64) WaitError!bool {
54 return self.os_event.wait(timeout_ns);
55 }
56};
57
58const OsEvent = if (builtin.single_threaded) DebugEvent else switch (builtin.os) {
59 .windows => WindowsEvent,
60 .linux => if (builtin.link_libc) PosixEvent else LinuxEvent,
61 else => if (builtin.link_libc) PosixEvent else SpinEvent,
62};
63
64const DebugEvent = struct {
65 is_set: @typeOf(set_init),
66
67 const set_init = if (std.debug.runtime_safety) false else {};
68
69 pub fn init() DebugEvent {
70 return DebugEvent{ .is_set = set_init };
71 }
72
73 pub fn deinit(self: *DebugEvent) void {
74 self.* = undefined;
75 }
76
77 pub fn isSet(self: *DebugEvent) bool {
78 if (!std.debug.runtime_safety)
79 return true;
80 return self.is_set;
81 }
82
83 pub fn set(self: *DebugEvent, auto_reset: bool) bool {
84 if (std.debug.runtime_safety)
85 self.is_set = !auto_reset;
86 return false;
87 }
88
89 pub fn reset(self: *DebugEvent) bool {
90 if (!std.debug.runtime_safety)
91 return false;
92 const was_set = self.is_set;
93 self.is_set = false;
94 return was_set;
95 }
96
97 pub fn wait(self: *DebugEvent, timeout: ?u64) ResetEvent.WaitError!bool {
98 if (std.debug.runtime_safety and !self.is_set)
99 @panic("deadlock detected");
100 return ResetEvent.WaitError.TimedOut;
101 }
102};
103
104fn AtomicEvent(comptime FutexImpl: type) type {
105 return struct {
106 state: u32,
107
108 const IS_SET: u32 = 1 << 0;
109 const WAIT_MASK = ~IS_SET;
110
111 pub const Self = @This();
112 pub const Futex = FutexImpl;
113
114 pub fn init() Self {
115 return Self{ .state = 0 };
116 }
117
118 pub fn deinit(self: *Self) void {
119 self.* = undefined;
120 }
121
122 pub fn isSet(self: *const Self) bool {
123 const state = @atomicLoad(u32, &self.state, .Acquire);
124 return (state & IS_SET) != 0;
125 }
126
127 pub fn reset(self: *Self) bool {
128 const old_state = @atomicRmw(u32, &self.state, .Xchg, 0, .Monotonic);
129 return (old_state & IS_SET) != 0;
130 }
131
132 pub fn set(self: *Self, auto_reset: bool) bool {
133 const new_state = if (auto_reset) 0 else IS_SET;
134 const old_state = @atomicRmw(u32, &self.state, .Xchg, new_state, .Release);
135 if ((old_state & WAIT_MASK) == 0) {
136 return false;
137 }
138
139 Futex.wake(&self.state);
140 return true;
141 }
142
143 pub fn wait(self: *Self, timeout: ?u64) ResetEvent.WaitError!bool {
144 var dummy_value: u32 = undefined;
145 const wait_token = @truncate(u32, @ptrToInt(&dummy_value));
146
147 var state = @atomicLoad(u32, &self.state, .Monotonic);
148 while (true) {
149 if ((state & IS_SET) != 0)
150 return false;
151 state = @cmpxchgWeak(u32, &self.state, state, wait_token, .Acquire, .Monotonic) orelse break;
152 }
153
154 try Futex.wait(&self.state, wait_token, timeout);
155 return true;
156 }
157 };
158}
159
160const SpinEvent = AtomicEvent(struct {
161 fn wake(ptr: *const u32) void {}
162
163 fn wait(ptr: *const u32, expected: u32, timeout: ?u64) ResetEvent.WaitError!void {
164 // TODO: handle platforms where time.Timer.start() fails
165 var spin = Backoff.init();
166 var timer = if (timeout == null) null else time.Timer.start() catch unreachable;
167 while (@atomicLoad(u32, ptr, .Acquire) == expected) {
168 spin.yield();
169 if (timeout) |timeout_ns| {
170 if (timer.?.read() > timeout_ns)
171 return ResetEvent.WaitError.TimedOut;
172 }
173 }
174 }
175});
176
177const LinuxEvent = AtomicEvent(struct {
178 fn wake(ptr: *const u32) void {
179 const key = @ptrCast(*const i32, ptr);
180 const rc = linux.futex_wake(key, linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
181 assert(linux.getErrno(rc) == 0);
182 }
183
184 fn wait(ptr: *const u32, expected: u32, timeout: ?u64) ResetEvent.WaitError!void {
185 var ts: linux.timespec = undefined;
186 var ts_ptr: ?*linux.timespec = null;
187 if (timeout) |timeout_ns| {
188 ts_ptr = &ts;
189 ts.tv_sec = @intCast(isize, timeout_ns / time.ns_per_s);
190 ts.tv_nsec = @intCast(isize, timeout_ns % time.ns_per_s);
191 }
192
193 const key = @ptrCast(*const i32, ptr);
194 const key_expect = @bitCast(i32, expected);
195 while (@atomicLoad(i32, key, .Acquire) == key_expect) {
196 const rc = linux.futex_wait(key, linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, key_expect, ts_ptr);
197 switch (linux.getErrno(rc)) {
198 0, linux.EAGAIN => break,
199 linux.EINTR => continue,
200 linux.ETIMEDOUT => return ResetEvent.WaitError.TimedOut,
201 else => unreachable,
202 }
203 }
204 }
205});
206
207const WindowsEvent = AtomicEvent(struct {
208 fn wake(ptr: *const u32) void {
209 if (getEventHandle()) |handle| {
210 const key = @ptrCast(*const c_void, ptr);
211 const rc = windows.ntdll.NtReleaseKeyedEvent(handle, key, windows.FALSE, null);
212 assert(rc == 0);
213 }
214 }
215
216 fn wait(ptr: *const u32, expected: u32, timeout: ?u64) ResetEvent.WaitError!void {
217 // fallback to spinlock if NT Keyed Events arent available
218 const handle = getEventHandle() orelse {
219 return SpinEvent.Futex.wait(ptr, expected, timeout);
220 };
221
222 // NT uses timeouts in units of 100ns with negative value being relative
223 var timeout_ptr: ?*windows.LARGE_INTEGER = null;
224 var timeout_value: windows.LARGE_INTEGER = undefined;
225 if (timeout) |timeout_ns| {
226 timeout_ptr = &timeout_value;
227 timeout_value = -@intCast(windows.LARGE_INTEGER, timeout_ns / 100);
228 }
229
230 // NtWaitForKeyedEvent doesnt have spurious wake-ups
231 if (@atomicLoad(u32, ptr, .Acquire) == expected) {
232 const key = @ptrCast(*const c_void, ptr);
233 const rc = windows.ntdll.NtWaitForKeyedEvent(handle, key, windows.FALSE, timeout_ptr);
234 switch (rc) {
235 0 => {},
236 windows.WAIT_TIMEOUT => return ResetEvent.WaitError.TimedOut,
237 else => unreachable,
238 }
239 }
240 }
241
242 var keyed_state = State.Uninitialized;
243 var keyed_handle: ?windows.HANDLE = null;
244
245 const State = enum(u8) {
246 Uninitialized,
247 Intializing,
248 Initialized,
249 };
250
251 fn getEventHandle() ?windows.HANDLE {
252 var spin = Backoff.init();
253 var state = @atomicLoad(State, &keyed_state, .Monotonic);
254
255 while (true) {
256 switch (state) {
257 .Initialized => {
258 return keyed_handle;
259 },
260 .Intializing => {
261 spin.yield();
262 state = @atomicLoad(State, &keyed_state, .Acquire);
263 },
264 .Uninitialized => state = @cmpxchgWeak(State, &keyed_state, state, .Intializing, .Acquire, .Monotonic) orelse {
265 var handle: windows.HANDLE = undefined;
266 const access_mask = windows.GENERIC_READ | windows.GENERIC_WRITE;
267 if (windows.ntdll.NtCreateKeyedEvent(&handle, access_mask, null, 0) == 0)
268 keyed_handle = handle;
269 @atomicStore(State, &keyed_state, .Initialized, .Release);
270 return keyed_handle;
271 },
272 }
273 }
274 }
275});
276
277const PosixEvent = struct {
278 state: u32,
279 cond: c.pthread_cond_t,
280 mutex: c.pthread_mutex_t,
281
282 const IS_SET: u32 = 1;
283
284 pub fn init() PosixEvent {
285 return PosixEvent{
286 .state = .0,
287 .cond = c.PTHREAD_COND_INITIALIZER,
288 .mutex = c.PTHREAD_MUTEX_INITIALIZER,
289 };
290 }
291
292 pub fn deinit(self: *PosixEvent) void {
293 // On dragonfly, the destroy functions return EINVAL if they were initialized statically.
294 const retm = c.pthread_mutex_destroy(&self.mutex);
295 assert(retm == 0 or retm == (if (builtin.os == .dragonfly) os.EINVAL else 0));
296 const retc = c.pthread_cond_destroy(&self.cond);
297 assert(retc == 0 or retc == (if (builtin.os == .dragonfly) os.EINVAL else 0));
298 }
299
300 pub fn isSet(self: *PosixEvent) bool {
301 assert(c.pthread_mutex_lock(&self.mutex) == 0);
302 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
303
304 return self.state == IS_SET;
305 }
306
307 pub fn reset(self: *PosixEvent) bool {
308 assert(c.pthread_mutex_lock(&self.mutex) == 0);
309 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
310
311 const was_set = self.state == IS_SET;
312 self.state = 0;
313 return was_set;
314 }
315
316 pub fn set(self: *PosixEvent, auto_reset: bool) bool {
317 assert(c.pthread_mutex_lock(&self.mutex) == 0);
318 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
319
320 const had_waiter = self.state > IS_SET;
321 self.state = if (auto_reset) 0 else IS_SET;
322 if (had_waiter) {
323 assert(c.pthread_cond_signal(&self.cond) == 0);
324 }
325 return had_waiter;
326 }
327
328 pub fn wait(self: *PosixEvent, timeout: ?u64) ResetEvent.WaitError!bool {
329 assert(c.pthread_mutex_lock(&self.mutex) == 0);
330 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
331
332 if (self.state == IS_SET)
333 return false;
334
335 var ts: os.timespec = undefined;
336 if (timeout) |timeout_ns| {
337 var timeout_abs = timeout_ns;
338 if (comptime std.Target.current.isDarwin()) {
339 var tv: os.darwin.timeval = undefined;
340 assert(os.darwin.gettimeofday(&tv, null) == 0);
341 timeout_abs += @intCast(u64, tv.tv_sec) * time.second;
342 timeout_abs += @intCast(u64, tv.tv_usec) * time.microsecond;
343 } else {
344 os.clock_gettime(os.CLOCK_REALTIME, &ts) catch unreachable;
345 timeout_abs += @intCast(u64, ts.tv_sec) * time.second;
346 timeout_abs += @intCast(u64, ts.tv_nsec);
347 }
348 ts.tv_sec = @intCast(@typeOf(ts.tv_sec), @divFloor(timeout_abs, time.second));
349 ts.tv_nsec = @intCast(@typeOf(ts.tv_nsec), @mod(timeout_abs, time.second));
350 }
351
352 var dummy_value: u32 = undefined;
353 var wait_token = @truncate(u32, @ptrToInt(&dummy_value));
354 self.state = wait_token;
355
356 while (self.state == wait_token) {
357 const rc = switch (timeout == null) {
358 true => c.pthread_cond_wait(&self.cond, &self.mutex),
359 else => c.pthread_cond_timedwait(&self.cond, &self.mutex, &ts),
360 };
361 // TODO: rc appears to be the positive error code making os.errno() always return 0 on linux
362 switch (std.math.max(@as(c_int, os.errno(rc)), rc)) {
363 0 => {},
364 os.ETIMEDOUT => return ResetEvent.WaitError.TimedOut,
365 os.EINVAL => unreachable,
366 os.EPERM => unreachable,
367 else => unreachable,
368 }
369 }
370 return true;
371 }
372};
373
374test "std.ResetEvent" {
375 // TODO
376 if (builtin.single_threaded)
377 return error.SkipZigTest;
378
379 var event = ResetEvent.init();
380 defer event.deinit();
381
382 // test event setting
383 testing.expect(event.isSet() == false);
384 testing.expect(event.set(false) == false);
385 testing.expect(event.isSet() == true);
386
387 // test event resetting
388 testing.expect(event.reset() == true);
389 testing.expect(event.isSet() == false);
390 testing.expect(event.reset() == false);
391
392 // test cross thread signaling
393 const Context = struct {
394 event: ResetEvent,
395 value: u128,
396
397 fn receiver(self: *@This()) void {
398 // wait for the sender to notify us with updated value
399 assert(self.value == 0);
400 assert((self.event.wait(1 * time.second) catch unreachable) == true);
401 assert(self.value == 1);
402
403 // wait for sender to sleep, then notify it of new value
404 time.sleep(50 * time.millisecond);
405 self.value = 2;
406 assert(self.event.set(false) == true);
407 }
408
409 fn sender(self: *@This()) !void {
410 // wait for the receiver() to start wait()'ing
411 time.sleep(50 * time.millisecond);
412
413 // update value to 1 and notify the receiver()
414 assert(self.value == 0);
415 self.value = 1;
416 assert(self.event.set(true) == true);
417
418 // wait for the receiver to update the value & notify us
419 assert((try self.event.wait(1 * time.second)) == true);
420 assert(self.value == 2);
421 }
422 };
423
424 _ = event.reset();
425 var context = Context{
426 .event = event,
427 .value = 0,
428 };
429
430 var receiver = try std.Thread.spawn(&context, Context.receiver);
431 defer receiver.wait();
432 try context.sender();
433}
\ No newline at end of file
lib/std/std.zig+1-1
......@@ -16,6 +16,7 @@ pub const PackedIntSlice = @import("packed_int_array.zig").PackedIntSlice;
1616pub const PackedIntSliceEndian = @import("packed_int_array.zig").PackedIntSliceEndian;
1717pub const PriorityQueue = @import("priority_queue.zig").PriorityQueue;
1818pub const Progress = @import("progress.zig").Progress;
19pub const ResetEvent = @import("reset_event.zig").ResetEvent;
1920pub const SegmentedList = @import("segmented_list.zig").SegmentedList;
2021pub const SinglyLinkedList = @import("linked_list.zig").SinglyLinkedList;
2122pub const SpinLock = @import("spinlock.zig").SpinLock;
......@@ -23,7 +24,6 @@ pub const StringHashMap = @import("hash_map.zig").StringHashMap;
2324pub const TailQueue = @import("linked_list.zig").TailQueue;
2425pub const Target = @import("target.zig").Target;
2526pub const Thread = @import("thread.zig").Thread;
26pub const ThreadParker = @import("parker.zig").ThreadParker;
2727
2828pub const atomic = @import("atomic.zig");
2929pub const base64 = @import("base64.zig");