authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-12-17 20:04:02-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2019-12-17 20:04:02-05:00
log4d54e9a4fbb899a18f1d7b9e83bbb65f0973a0cb
tree43a293be46eef6d84725cea2e86b9ce5759f7a61
parentd8499f7abe43ec641027eb7f94b41906c9bf5cca
parentc9122964436b16dd44a5fb8dfd92f0768ad6fef3
signature Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #3918 from kprotty/lock_fix

Synchronization primitive improvements

3 files changed, 490 insertions(+), 410 deletions(-)

lib/std/mutex.zig+179-85
...@@ -1,12 +1,13 @@...@@ -1,12 +1,13 @@
1const std = @import("std.zig");1const std = @import("std.zig");
2const builtin = @import("builtin");2const builtin = @import("builtin");
3const os = std.os;
3const testing = std.testing;4const testing = std.testing;
5const SpinLock = std.SpinLock;
4const ResetEvent = std.ResetEvent;6const ResetEvent = std.ResetEvent;
57
6/// Lock may be held only once. If the same thread8/// Lock may be held only once. If the same thread
7/// tries to acquire the same mutex twice, it deadlocks.9/// tries to acquire the same mutex twice, it deadlocks.
8/// This type supports static initialization and is based off of Webkit's WTF Lock (via rust parking_lot)10/// This type supports static initialization and is at most `@sizeOf(usize)` in size.
9/// https://github.com/Amanieu/parking_lot/blob/master/core/src/word_lock.rs
10/// When an application is built in single threaded release mode, all the functions are11/// When an application is built in single threaded release mode, all the functions are
11/// no-ops. In single threaded debug mode, there is deadlock detection.12/// no-ops. In single threaded debug mode, there is deadlock detection.
12pub const Mutex = if (builtin.single_threaded)13pub const Mutex = if (builtin.single_threaded)
...@@ -24,136 +25,229 @@ pub const Mutex = if (builtin.single_threaded)...@@ -24,136 +25,229 @@ pub const Mutex = if (builtin.single_threaded)
24 }25 }
25 }26 }
26 };27 };
28
27 pub fn init() Mutex {29 pub fn init() Mutex {
28 return Mutex{ .lock = lock_init };30 return Mutex{ .lock = lock_init };
29 }31 }
30 pub fn deinit(self: *Mutex) void {}
3132
32 pub fn acquire(self: *Mutex) Held {33 pub fn deinit(self: *Mutex) void {
33 if (std.debug.runtime_safety and self.lock) {34 self.* = undefined;
34 @panic("deadlock detected");35 }
36
37 pub fn tryAcquire(self: *Mutex) ?Held {
38 if (std.debug.runtime_safety) {
39 if (self.lock) return null;
40 self.lock = true;
35 }41 }
36 return Held{ .mutex = self };42 return Held{ .mutex = self };
37 }43 }
38 }
39else
40 struct {
41 state: usize,
4244
43 const MUTEX_LOCK: usize = 1 << 0;45 pub fn acquire(self: *Mutex) Held {
44 const QUEUE_LOCK: usize = 1 << 1;46 return self.tryAcquire() orelse @panic("deadlock detected");
45 const QUEUE_MASK: usize = ~(MUTEX_LOCK | QUEUE_LOCK);47 }
46 const QueueNode = std.atomic.Stack(ResetEvent).Node;48 }
4749else if (builtin.os == .windows)
48 /// number of iterations to spin yielding the cpu50 // https://locklessinc.com/articles/keyed_events/
49 const SPIN_CPU = 4;51 extern union {
5052 locked: u8,
51 /// number of iterations to spin in the cpu yield loop53 waiters: u32,
52 const SPIN_CPU_COUNT = 30;
5354
54 /// number of iterations to spin yielding the thread55 const WAKE = 1 << 8;
55 const SPIN_THREAD = 1;56 const WAIT = 1 << 9;
5657
57 pub fn init() Mutex {58 pub fn init() Mutex {
58 return Mutex{ .state = 0 };59 return Mutex{ .waiters = 0 };
59 }60 }
6061
61 pub fn deinit(self: *Mutex) void {62 pub fn deinit(self: *Mutex) void {
62 self.* = undefined;63 self.* = undefined;
63 }64 }
6465
66 pub fn tryAcquire(self: *Mutex) ?Held {
67 if (@atomicRmw(u8, &self.locked, .Xchg, 1, .Acquire) != 0)
68 return null;
69 return Held{ .mutex = self };
70 }
71
72 pub fn acquire(self: *Mutex) Held {
73 return self.tryAcquire() orelse self.acquireSlow();
74 }
75
76 fn acquireSlow(self: *Mutex) Held {
77 @setCold(true);
78 while (true) : (SpinLock.loopHint(1)) {
79 const waiters = @atomicLoad(u32, &self.waiters, .Monotonic);
80
81 // try and take lock if unlocked
82 if ((waiters & 1) == 0) {
83 if (@atomicRmw(u8, &self.locked, .Xchg, 1, .Acquire) == 0)
84 return Held{ .mutex = self };
85
86 // otherwise, try and update the waiting count.
87 // then unset the WAKE bit so that another unlocker can wake up a thread.
88 } else if (@cmpxchgWeak(u32, &self.waiters, waiters, (waiters + WAIT) | 1, .Monotonic, .Monotonic) == null) {
89 ResetEvent.OsEvent.Futex.wait(@ptrCast(*i32, &self.waiters), undefined, null) catch unreachable;
90 _ = @atomicRmw(u32, &self.waiters, .Sub, WAKE, .Monotonic);
91 }
92 }
93 }
94
65 pub const Held = struct {95 pub const Held = struct {
66 mutex: *Mutex,96 mutex: *Mutex,
6797
68 pub fn release(self: Held) void {98 pub fn release(self: Held) void {
69 // since MUTEX_LOCK is the first bit, we can use (.Sub) instead of (.And, ~MUTEX_LOCK).99 // unlock without a rmw/cmpxchg instruction
70 // this is because .Sub may be implemented more efficiently than the latter100 @atomicStore(u8, @ptrCast(*u8, &self.mutex.locked), 0, .Release);
71 // (e.g. `lock xadd` vs `cmpxchg` loop on x86)101
72 const state = @atomicRmw(usize, &self.mutex.state, .Sub, MUTEX_LOCK, .Release);102 while (true) : (SpinLock.loopHint(1)) {
73 if ((state & QUEUE_MASK) != 0 and (state & QUEUE_LOCK) == 0) {103 const waiters = @atomicLoad(u32, &self.mutex.waiters, .Monotonic);
74 self.mutex.releaseSlow(state);104
105 // no one is waiting
106 if (waiters < WAIT) return;
107 // someone grabbed the lock and will do the wake instead
108 if (waiters & 1 != 0) return;
109 // someone else is currently waking up
110 if (waiters & WAKE != 0) return;
111
112 // try to decrease the waiter count & set the WAKE bit meaning a thread is waking up
113 if (@cmpxchgWeak(u32, &self.mutex.waiters, waiters, waiters - WAIT + WAKE, .Release, .Monotonic) == null)
114 return ResetEvent.OsEvent.Futex.wake(@ptrCast(*i32, &self.mutex.waiters));
75 }115 }
76 }116 }
77 };117 };
118 }
119else if (builtin.link_libc or builtin.os == .linux)
120 // stack-based version of https://github.com/Amanieu/parking_lot/blob/master/core/src/word_lock.rs
121 struct {
122 state: usize,
78123
79 pub fn acquire(self: *Mutex) Held {124 /// number of times to spin trying to acquire the lock.
80 // fast path close to SpinLock fast path125 /// https://webkit.org/blog/6161/locking-in-webkit/
81 if (@cmpxchgWeak(usize, &self.state, 0, MUTEX_LOCK, .Acquire, .Monotonic)) |current_state| {126 const SPIN_COUNT = 40;
82 self.acquireSlow(current_state);127
83 }128 const MUTEX_LOCK: usize = 1 << 0;
129 const QUEUE_LOCK: usize = 1 << 1;
130 const QUEUE_MASK: usize = ~(MUTEX_LOCK | QUEUE_LOCK);
131
132 const Node = struct {
133 next: ?*Node,
134 event: ResetEvent,
135 };
136
137 pub fn init() Mutex {
138 return Mutex{ .state = 0 };
139 }
140
141 pub fn deinit(self: *Mutex) void {
142 self.* = undefined;
143 }
144
145 pub fn tryAcquire(self: *Mutex) ?Held {
146 if (@cmpxchgWeak(usize, &self.state, 0, MUTEX_LOCK, .Acquire, .Monotonic) != null)
147 return null;
84 return Held{ .mutex = self };148 return Held{ .mutex = self };
85 }149 }
86150
87 fn acquireSlow(self: *Mutex, current_state: usize) void {151 pub fn acquire(self: *Mutex) Held {
88 var spin: usize = 0;152 return self.tryAcquire() orelse {
89 var state = current_state;153 self.acquireSlow();
154 return Held{ .mutex = self };
155 };
156 }
157
158 fn acquireSlow(self: *Mutex) void {
159 // inlining the fast path and hiding *Slow()
160 // calls behind a @setCold(true) appears to
161 // improve performance in release builds.
162 @setCold(true);
90 while (true) {163 while (true) {
91164
92 // try and acquire the lock if unlocked165 // try and spin for a bit to acquire the mutex if theres currently no queue
93 if ((state & MUTEX_LOCK) == 0) {166 var spin_count: u32 = SPIN_COUNT;
94 state = @cmpxchgWeak(usize, &self.state, state, state | MUTEX_LOCK, .Acquire, .Monotonic) orelse return;167 var state = @atomicLoad(usize, &self.state, .Monotonic);
95 continue;168 while (spin_count != 0) : (spin_count -= 1) {
169 if (state & MUTEX_LOCK == 0) {
170 _ = @cmpxchgWeak(usize, &self.state, state, state | MUTEX_LOCK, .Acquire, .Monotonic) orelse return;
171 } else if (state & QUEUE_MASK == 0) {
172 break;
173 }
174 SpinLock.yield();
175 state = @atomicLoad(usize, &self.state, .Monotonic);
96 }176 }
97177
98 // spin only if the waiting queue isn't empty and when it hasn't spun too much already178 // create the ResetEvent node on the stack
99 if ((state & QUEUE_MASK) == 0 and spin < SPIN_CPU + SPIN_THREAD) {179 // (faster than threadlocal on platforms like OSX)
100 if (spin < SPIN_CPU) {180 var node: Node = undefined;
101 std.SpinLock.yield(SPIN_CPU_COUNT);181 node.event = ResetEvent.init();
182 defer node.event.deinit();
183
184 // we've spun too long, try and add our node to the LIFO queue.
185 // if the mutex becomes available in the process, try and grab it instead.
186 while (true) {
187 if (state & MUTEX_LOCK == 0) {
188 _ = @cmpxchgWeak(usize, &self.state, state, state | MUTEX_LOCK, .Acquire, .Monotonic) orelse return;
102 } else {189 } else {
103 std.os.sched_yield() catch std.time.sleep(0);190 node.next = @intToPtr(?*Node, state & QUEUE_MASK);
191 const new_state = @ptrToInt(&node) | (state & ~QUEUE_MASK);
192 _ = @cmpxchgWeak(usize, &self.state, state, new_state, .Release, .Monotonic) orelse {
193 node.event.wait();
194 break;
195 };
104 }196 }
197 SpinLock.yield();
105 state = @atomicLoad(usize, &self.state, .Monotonic);198 state = @atomicLoad(usize, &self.state, .Monotonic);
106 continue;
107 }199 }
108
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 }200 }
124 }201 }
125202
126 fn releaseSlow(self: *Mutex, current_state: usize) void {203 pub const Held = struct {
127 // grab the QUEUE_LOCK in order to signal a waiting queue node's event.204 mutex: *Mutex,
128 var state = current_state;205
129 while (true) {206 pub fn release(self: Held) void {
130 if ((state & QUEUE_LOCK) != 0 or (state & QUEUE_MASK) == 0)207 // first, remove the lock bit so another possibly parallel acquire() can succeed.
208 // use .Sub since it can be usually compiled down more efficiency
209 // (`lock sub` on x86) vs .And ~MUTEX_LOCK (`lock cmpxchg` loop on x86)
210 const state = @atomicRmw(usize, &self.mutex.state, .Sub, MUTEX_LOCK, .Release);
211
212 // if the LIFO queue isnt locked and it has a node, try and wake up the node.
213 if ((state & QUEUE_LOCK) == 0 and (state & QUEUE_MASK) != 0)
214 self.mutex.releaseSlow();
215 }
216 };
217
218 fn releaseSlow(self: *Mutex) void {
219 @setCold(true);
220
221 // try and lock the LFIO queue to pop a node off,
222 // stopping altogether if its already locked or the queue is empty
223 var state = @atomicLoad(usize, &self.state, .Monotonic);
224 while (true) : (SpinLock.loopHint(1)) {
225 if (state & QUEUE_LOCK != 0 or state & QUEUE_MASK == 0)
131 return;226 return;
132 state = @cmpxchgWeak(usize, &self.state, state, state | QUEUE_LOCK, .Acquire, .Monotonic) orelse break;227 state = @cmpxchgWeak(usize, &self.state, state, state | QUEUE_LOCK, .Acquire, .Monotonic) orelse break;
133 }228 }
134229
135 while (true) {230 // acquired the QUEUE_LOCK, try and pop a node to wake it.
136 // barrier needed to observe incoming state changes231 // if the mutex is locked, then unset QUEUE_LOCK and let
137 defer @fence(.Acquire);232 // the thread who holds the mutex do the wake-up on unlock()
138233 while (true) : (SpinLock.loopHint(1)) {
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) {234 if ((state & MUTEX_LOCK) != 0) {
142 state = @cmpxchgWeak(usize, &self.state, state, state & ~QUEUE_LOCK, .Release, .Monotonic) orelse return;235 state = @cmpxchgWeak(usize, &self.state, state, state & ~QUEUE_LOCK, .Release, .Acquire) orelse return;
143 continue;236 } else {
237 const node = @intToPtr(*Node, state & QUEUE_MASK);
238 const new_state = @ptrToInt(node.next);
239 state = @cmpxchgWeak(usize, &self.state, state, new_state, .Release, .Acquire) orelse {
240 node.event.set();
241 return;
242 };
144 }243 }
145
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 };
154 }244 }
155 }245 }
156 };246 }
247
248// for platforms without a known OS blocking
249// primitive, default to SpinLock for correctness
250else SpinLock;
157251
158const TestContext = struct {252const TestContext = struct {
159 mutex: *Mutex,253 mutex: *Mutex,
lib/std/reset_event.zig+265-287
...@@ -1,8 +1,8 @@...@@ -1,8 +1,8 @@
1const std = @import("std.zig");1const std = @import("std.zig");
2const builtin = @import("builtin");2const builtin = @import("builtin");
3const testing = std.testing;3const testing = std.testing;
4const SpinLock = std.SpinLock;
4const assert = std.debug.assert;5const assert = std.debug.assert;
5const Backoff = std.SpinLock.Backoff;
6const c = std.c;6const c = std.c;
7const os = std.os;7const os = std.os;
8const time = std.time;8const time = std.time;
...@@ -14,13 +14,20 @@ const windows = os.windows;...@@ -14,13 +14,20 @@ const windows = os.windows;
14pub const ResetEvent = struct {14pub const ResetEvent = struct {
15 os_event: OsEvent,15 os_event: OsEvent,
1616
17 pub const OsEvent =
18 if (builtin.single_threaded)
19 DebugEvent
20 else if (builtin.link_libc and builtin.os != .windows and builtin.os != .linux)
21 PosixEvent
22 else
23 AtomicEvent;
24
17 pub fn init() ResetEvent {25 pub fn init() ResetEvent {
18 return ResetEvent{ .os_event = OsEvent.init() };26 return ResetEvent{ .os_event = OsEvent.init() };
19 }27 }
2028
21 pub fn deinit(self: *ResetEvent) void {29 pub fn deinit(self: *ResetEvent) void {
22 self.os_event.deinit();30 self.os_event.deinit();
23 self.* = undefined;
24 }31 }
2532
26 /// Returns whether or not the event is currenetly set33 /// Returns whether or not the event is currenetly set
...@@ -29,308 +36,116 @@ pub const ResetEvent = struct {...@@ -29,308 +36,116 @@ pub const ResetEvent = struct {
29 }36 }
3037
31 /// Sets the event if not already set and38 /// Sets the event if not already set and
32 /// wakes up AT LEAST one thread waiting the event.39 /// wakes up at least one thread waiting the event.
33 /// Returns whether or not a thread was woken up.40 pub fn set(self: *ResetEvent) void {
34 pub fn set(self: *ResetEvent, auto_reset: bool) bool {41 return self.os_event.set();
35 return self.os_event.set(auto_reset);
36 }42 }
3743
38 /// Resets the event to its original, unset state.44 /// Resets the event to its original, unset state.
39 /// Returns whether or not the event was currently set before un-setting.45 pub fn reset(self: *ResetEvent) void {
40 pub fn reset(self: *ResetEvent) bool {
41 return self.os_event.reset();46 return self.os_event.reset();
42 }47 }
4348
44 const WaitError = error{49 /// Wait for the event to be set by blocking the current thread.
45 /// The thread blocked longer than the maximum time specified.50 pub fn wait(self: *ResetEvent) void {
46 TimedOut,51 return self.os_event.wait(null) catch unreachable;
47 };52 }
4853
49 /// Wait for the event to be set by blocking the current thread.54 /// Wait for the event to be set by blocking the current thread.
50 /// Optionally provided timeout in nanoseconds which throws an55 /// A timeout in nanoseconds can be provided as a hint for how
51 /// `error.TimedOut` if the thread blocked AT LEAST longer than specified.56 /// long the thread should block on the unset event before throwind error.TimedOut.
52 /// Returns whether or not the thread blocked from the event being unset at the time of calling.57 pub fn timedWait(self: *ResetEvent, timeout_ns: u64) !void {
53 pub fn wait(self: *ResetEvent, timeout_ns: ?u64) WaitError!bool {
54 return self.os_event.wait(timeout_ns);58 return self.os_event.wait(timeout_ns);
55 }59 }
56};60};
5761
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 {62const DebugEvent = struct {
65 is_set: @TypeOf(set_init),63 is_set: bool,
66
67 const set_init = if (std.debug.runtime_safety) false else {};
6864
69 pub fn init() DebugEvent {65 fn init() DebugEvent {
70 return DebugEvent{ .is_set = set_init };66 return DebugEvent{ .is_set = false };
71 }67 }
7268
73 pub fn deinit(self: *DebugEvent) void {69 fn deinit(self: *DebugEvent) void {
74 self.* = undefined;70 self.* = undefined;
75 }71 }
7672
77 pub fn isSet(self: *DebugEvent) bool {73 fn isSet(self: *DebugEvent) bool {
78 if (!std.debug.runtime_safety)
79 return true;
80 return self.is_set;74 return self.is_set;
81 }75 }
8276
83 pub fn set(self: *DebugEvent, auto_reset: bool) bool {77 fn reset(self: *DebugEvent) void {
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;78 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 }79 }
18380
184 fn wait(ptr: *const u32, expected: u32, timeout: ?u64) ResetEvent.WaitError!void {81 fn set(self: *DebugEvent) void {
185 var ts: linux.timespec = undefined;82 self.is_set = true;
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 }83 }
205});
20684
207const WindowsEvent = AtomicEvent(struct {85 fn wait(self: *DebugEvent, timeout: ?u64) !void {
208 fn wake(ptr: *const u32) void {86 if (self.is_set)
209 if (getEventHandle()) |handle| {87 return;
210 const key = @ptrCast(*const c_void, ptr);88 if (timeout != null)
211 const rc = windows.ntdll.NtReleaseKeyedEvent(handle, key, windows.FALSE, null);89 return error.TimedOut;
212 assert(rc == 0);90 @panic("deadlock detected");
213 }
214 }91 }
21592};
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});
27693
277const PosixEvent = struct {94const PosixEvent = struct {
278 state: u32,95 is_set: bool,
279 cond: c.pthread_cond_t,96 cond: c.pthread_cond_t,
280 mutex: c.pthread_mutex_t,97 mutex: c.pthread_mutex_t,
28198
282 const IS_SET: u32 = 1;99 fn init() PosixEvent {
283
284 pub fn init() PosixEvent {
285 return PosixEvent{100 return PosixEvent{
286 .state = 0,101 .is_set = false,
287 .cond = c.PTHREAD_COND_INITIALIZER,102 .cond = c.PTHREAD_COND_INITIALIZER,
288 .mutex = c.PTHREAD_MUTEX_INITIALIZER,103 .mutex = c.PTHREAD_MUTEX_INITIALIZER,
289 };104 };
290 }105 }
291106
292 pub fn deinit(self: *PosixEvent) void {107 fn deinit(self: *PosixEvent) void {
293 // On dragonfly, the destroy functions return EINVAL if they were initialized statically.108 // on dragonfly, *destroy() functions can return EINVAL
109 // for statically initialized pthread structures
110 const err = if (builtin.os == .dragonfly) os.EINVAL else 0;
111
294 const retm = c.pthread_mutex_destroy(&self.mutex);112 const retm = c.pthread_mutex_destroy(&self.mutex);
295 assert(retm == 0 or retm == (if (builtin.os == .dragonfly) os.EINVAL else 0));113 assert(retm == 0 or retm == err);
296 const retc = c.pthread_cond_destroy(&self.cond);114 const retc = c.pthread_cond_destroy(&self.cond);
297 assert(retc == 0 or retc == (if (builtin.os == .dragonfly) os.EINVAL else 0));115 assert(retc == 0 or retc == err);
298 }116 }
299117
300 pub fn isSet(self: *PosixEvent) bool {118 fn isSet(self: *PosixEvent) bool {
301 assert(c.pthread_mutex_lock(&self.mutex) == 0);119 assert(c.pthread_mutex_lock(&self.mutex) == 0);
302 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);120 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
303121
304 return self.state == IS_SET;122 return self.is_set;
305 }123 }
306124
307 pub fn reset(self: *PosixEvent) bool {125 fn reset(self: *PosixEvent) void {
308 assert(c.pthread_mutex_lock(&self.mutex) == 0);126 assert(c.pthread_mutex_lock(&self.mutex) == 0);
309 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);127 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
310128
311 const was_set = self.state == IS_SET;129 self.is_set = false;
312 self.state = 0;
313 return was_set;
314 }130 }
315131
316 pub fn set(self: *PosixEvent, auto_reset: bool) bool {132 fn set(self: *PosixEvent) void {
317 assert(c.pthread_mutex_lock(&self.mutex) == 0);133 assert(c.pthread_mutex_lock(&self.mutex) == 0);
318 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);134 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
319135
320 const had_waiter = self.state > IS_SET;136 if (!self.is_set) {
321 self.state = if (auto_reset) 0 else IS_SET;137 self.is_set = true;
322 if (had_waiter) {
323 assert(c.pthread_cond_signal(&self.cond) == 0);138 assert(c.pthread_cond_signal(&self.cond) == 0);
324 }139 }
325 return had_waiter;
326 }140 }
327141
328 pub fn wait(self: *PosixEvent, timeout: ?u64) ResetEvent.WaitError!bool {142 fn wait(self: *PosixEvent, timeout: ?u64) !void {
329 assert(c.pthread_mutex_lock(&self.mutex) == 0);143 assert(c.pthread_mutex_lock(&self.mutex) == 0);
330 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);144 defer assert(c.pthread_mutex_unlock(&self.mutex) == 0);
331145
332 if (self.state == IS_SET)146 // quick guard before possibly calling time syscalls below
333 return false;147 if (self.is_set)
148 return;
334149
335 var ts: os.timespec = undefined;150 var ts: os.timespec = undefined;
336 if (timeout) |timeout_ns| {151 if (timeout) |timeout_ns| {
...@@ -349,85 +164,248 @@ const PosixEvent = struct {...@@ -349,85 +164,248 @@ const PosixEvent = struct {
349 ts.tv_nsec = @intCast(@TypeOf(ts.tv_nsec), @mod(timeout_abs, time.second));164 ts.tv_nsec = @intCast(@TypeOf(ts.tv_nsec), @mod(timeout_abs, time.second));
350 }165 }
351166
352 var dummy_value: u32 = undefined;167 while (!self.is_set) {
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) {168 const rc = switch (timeout == null) {
358 true => c.pthread_cond_wait(&self.cond, &self.mutex),169 true => c.pthread_cond_wait(&self.cond, &self.mutex),
359 else => c.pthread_cond_timedwait(&self.cond, &self.mutex, &ts),170 else => c.pthread_cond_timedwait(&self.cond, &self.mutex, &ts),
360 };171 };
361 // TODO: rc appears to be the positive error code making os.errno() always return 0 on linux172 switch (rc) {
362 switch (std.math.max(@as(c_int, os.errno(rc)), rc)) {
363 0 => {},173 0 => {},
364 os.ETIMEDOUT => return ResetEvent.WaitError.TimedOut,174 os.ETIMEDOUT => return error.TimedOut,
365 os.EINVAL => unreachable,175 os.EINVAL => unreachable,
366 os.EPERM => unreachable,176 os.EPERM => unreachable,
367 else => unreachable,177 else => unreachable,
368 }178 }
369 }179 }
370 return true;
371 }180 }
372};181};
373182
374test "std.ResetEvent" {183const AtomicEvent = struct {
375 // TODO184 state: State,
376 if (builtin.single_threaded)185
377 return error.SkipZigTest;186 const State = enum(i32) {
187 Empty,
188 Waiting,
189 Signaled,
190 };
191
192 fn init() AtomicEvent {
193 return AtomicEvent{ .state = .Empty };
194 }
195
196 fn deinit(self: *AtomicEvent) void {
197 self.* = undefined;
198 }
199
200 fn isSet(self: *AtomicEvent) bool {
201 return @atomicLoad(State, &self.state, .Acquire) == .Signaled;
202 }
203
204 fn reset(self: *AtomicEvent) void {
205 @atomicStore(State, &self.state, .Empty, .Monotonic);
206 }
207
208 fn set(self: *AtomicEvent) void {
209 if (@atomicRmw(State, &self.state, .Xchg, .Signaled, .Release) == .Waiting)
210 Futex.wake(@ptrCast(*i32, &self.state));
211 }
212
213 fn wait(self: *AtomicEvent, timeout: ?u64) !void {
214 var state = @atomicLoad(State, &self.state, .Monotonic);
215 while (state == .Empty) {
216 state = @cmpxchgWeak(State, &self.state, .Empty, .Waiting, .Acquire, .Monotonic) orelse
217 return Futex.wait(@ptrCast(*i32, &self.state), @enumToInt(State.Waiting), timeout);
218 }
219 }
220
221 pub const Futex = switch (builtin.os) {
222 .windows => WindowsFutex,
223 .linux => LinuxFutex,
224 else => SpinFutex,
225 };
226
227 const SpinFutex = struct {
228 fn wake(ptr: *i32) void {}
229
230 fn wait(ptr: *i32, expected: i32, timeout: ?u64) !void {
231 // TODO: handle platforms where a monotonic timer isnt available
232 var timer: time.Timer = undefined;
233 if (timeout != null)
234 timer = time.Timer.start() catch unreachable;
235
236 while (@atomicLoad(i32, ptr, .Acquire) == expected) {
237 SpinLock.yield();
238 if (timeout) |timeout_ns| {
239 if (timer.read() >= timeout_ns)
240 return error.TimedOut;
241 }
242 }
243 }
244 };
245
246 const LinuxFutex = struct {
247 fn wake(ptr: *i32) void {
248 const rc = linux.futex_wake(ptr, linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
249 assert(linux.getErrno(rc) == 0);
250 }
251
252 fn wait(ptr: *i32, expected: i32, timeout: ?u64) !void {
253 var ts: linux.timespec = undefined;
254 var ts_ptr: ?*linux.timespec = null;
255 if (timeout) |timeout_ns| {
256 ts_ptr = &ts;
257 ts.tv_sec = @intCast(isize, timeout_ns / time.ns_per_s);
258 ts.tv_nsec = @intCast(isize, timeout_ns % time.ns_per_s);
259 }
260
261 while (@atomicLoad(i32, ptr, .Acquire) == expected) {
262 const rc = linux.futex_wait(ptr, linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, expected, ts_ptr);
263 switch (linux.getErrno(rc)) {
264 0 => continue,
265 os.ETIMEDOUT => return error.TimedOut,
266 os.EINTR => continue,
267 os.EAGAIN => return,
268 else => unreachable,
269 }
270 }
271 }
272 };
273
274 const WindowsFutex = struct {
275 pub fn wake(ptr: *i32) void {
276 const handle = getEventHandle() orelse return SpinFutex.wake(ptr);
277 const key = @ptrCast(*const c_void, ptr);
278 const rc = windows.ntdll.NtReleaseKeyedEvent(handle, key, windows.FALSE, null);
279 assert(rc == 0);
280 }
281
282 pub fn wait(ptr: *i32, expected: i32, timeout: ?u64) !void {
283 const handle = getEventHandle() orelse return SpinFutex.wait(ptr, expected, timeout);
284
285 // NT uses timeouts in units of 100ns with negative value being relative
286 var timeout_ptr: ?*windows.LARGE_INTEGER = null;
287 var timeout_value: windows.LARGE_INTEGER = undefined;
288 if (timeout) |timeout_ns| {
289 timeout_ptr = &timeout_value;
290 timeout_value = -@intCast(windows.LARGE_INTEGER, timeout_ns / 100);
291 }
292
293 // NtWaitForKeyedEvent doesnt have spurious wake-ups
294 const key = @ptrCast(*const c_void, ptr);
295 const rc = windows.ntdll.NtWaitForKeyedEvent(handle, key, windows.FALSE, timeout_ptr);
296 switch (rc) {
297 windows.WAIT_TIMEOUT => return error.TimedOut,
298 windows.WAIT_OBJECT_0 => {},
299 else => unreachable,
300 }
301 }
378302
303 var event_handle: usize = EMPTY;
304 const EMPTY = ~@as(usize, 0);
305 const LOADING = EMPTY - 1;
306
307 pub fn getEventHandle() ?windows.HANDLE {
308 var handle = @atomicLoad(usize, &event_handle, .Monotonic);
309 while (true) {
310 switch (handle) {
311 EMPTY => handle = @cmpxchgWeak(usize, &event_handle, EMPTY, LOADING, .Acquire, .Monotonic) orelse {
312 const handle_ptr = @ptrCast(*windows.HANDLE, &handle);
313 const access_mask = windows.GENERIC_READ | windows.GENERIC_WRITE;
314 if (windows.ntdll.NtCreateKeyedEvent(handle_ptr, access_mask, null, 0) != 0)
315 handle = 0;
316 @atomicStore(usize, &event_handle, handle, .Monotonic);
317 return @intToPtr(?windows.HANDLE, handle);
318 },
319 LOADING => {
320 SpinLock.yield();
321 handle = @atomicLoad(usize, &event_handle, .Monotonic);
322 },
323 else => {
324 return @intToPtr(?windows.HANDLE, handle);
325 },
326 }
327 }
328 }
329 };
330};
331
332test "std.ResetEvent" {
379 var event = ResetEvent.init();333 var event = ResetEvent.init();
380 defer event.deinit();334 defer event.deinit();
381335
382 // test event setting336 // test event setting
383 testing.expect(event.isSet() == false);337 testing.expect(event.isSet() == false);
384 testing.expect(event.set(false) == false);338 event.set();
385 testing.expect(event.isSet() == true);339 testing.expect(event.isSet() == true);
386340
387 // test event resetting341 // test event resetting
388 testing.expect(event.reset() == true);342 event.reset();
389 testing.expect(event.isSet() == false);343 testing.expect(event.isSet() == false);
390 testing.expect(event.reset() == false);
391344
392 // test cross thread signaling345 // test event waiting (non-blocking)
393 const Context = struct {346 event.set();
394 event: ResetEvent,347 event.wait();
395 value: u128,348 try event.timedWait(1);
396349
397 fn receiver(self: *@This()) void {350 // test cross-thread signaling
398 // wait for the sender to notify us with updated value351 if (builtin.single_threaded)
399 assert(self.value == 0);352 return;
400 assert((self.event.wait(1 * time.second) catch unreachable) == true);
401 assert(self.value == 1);
402353
403 // wait for sender to sleep, then notify it of new value354 const Context = struct {
404 time.sleep(50 * time.millisecond);355 const Self = @This();
405 self.value = 2;356
406 assert(self.event.set(false) == true);357 value: u128,
358 in: ResetEvent,
359 out: ResetEvent,
360
361 fn init() Self {
362 return Self{
363 .value = 0,
364 .in = ResetEvent.init(),
365 .out = ResetEvent.init(),
366 };
407 }367 }
408368
409 fn sender(self: *@This()) !void {369 fn deinit(self: *Self) void {
410 // wait for the receiver() to start wait()'ing370 self.in.deinit();
411 time.sleep(50 * time.millisecond);371 self.out.deinit();
372 self.* = undefined;
373 }
412374
413 // update value to 1 and notify the receiver()375 fn sender(self: *Self) void {
414 assert(self.value == 0);376 // update value and signal input
377 testing.expect(self.value == 0);
415 self.value = 1;378 self.value = 1;
416 assert(self.event.set(true) == true);379 self.in.set();
417380
418 // wait for the receiver to update the value & notify us381 // wait for receiver to update value and signal output
419 assert((try self.event.wait(1 * time.second)) == true);382 self.out.wait();
420 assert(self.value == 2);383 testing.expect(self.value == 2);
384
385 // update value and signal final input
386 self.value = 3;
387 self.in.set();
421 }388 }
422 };
423389
424 _ = event.reset();390 fn receiver(self: *Self) void {
425 var context = Context{391 // wait for sender to update value and signal input
426 .event = event,392 self.in.wait();
427 .value = 0,393 assert(self.value == 1);
394
395 // update value and signal output
396 self.in.reset();
397 self.value = 2;
398 self.out.set();
399
400 // wait for sender to update value and signal final input
401 self.in.wait();
402 assert(self.value == 3);
403 }
428 };404 };
429405
430 var receiver = try std.Thread.spawn(&context, Context.receiver);406 var context = Context.init();
407 defer context.deinit();
408 const receiver = try std.Thread.spawn(&context, Context.receiver);
431 defer receiver.wait();409 defer receiver.wait();
432 try context.sender();410 context.sender();
433}411}
lib/std/spinlock.zig+46-38
...@@ -1,69 +1,77 @@...@@ -1,69 +1,77 @@
1const std = @import("std.zig");1const std = @import("std.zig");
2const builtin = @import("builtin");2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const time = std.time;
5const os = std.os;
63
7pub const SpinLock = struct {4pub const SpinLock = struct {
8 lock: u8, // TODO use a bool or enum5 state: State,
6
7 const State = enum(u8) {
8 Unlocked,
9 Locked,
10 };
911
10 pub const Held = struct {12 pub const Held = struct {
11 spinlock: *SpinLock,13 spinlock: *SpinLock,
1214
13 pub fn release(self: Held) void {15 pub fn release(self: Held) void {
14 @atomicStore(u8, &self.spinlock.lock, 0, .Release);16 @atomicStore(State, &self.spinlock.state, .Unlocked, .Release);
15 }17 }
16 };18 };
1719
18 pub fn init() SpinLock {20 pub fn init() SpinLock {
19 return SpinLock{ .lock = 0 };21 return SpinLock{ .state = .Unlocked };
20 }22 }
2123
22 pub fn acquire(self: *SpinLock) Held {24 pub fn deinit(self: *SpinLock) void {
23 var backoff = Backoff.init();25 self.* = undefined;
24 while (@atomicRmw(u8, &self.lock, .Xchg, 1, .Acquire) != 0)
25 backoff.yield();
26 return Held{ .spinlock = self };
27 }26 }
2827
29 pub fn yield(iterations: usize) void {28 pub fn tryAcquire(self: *SpinLock) ?Held {
30 var i = iterations;29 return switch (@atomicRmw(State, &self.state, .Xchg, .Locked, .Acquire)) {
31 while (i != 0) : (i -= 1) {30 .Unlocked => Held{ .spinlock = self },
32 switch (builtin.arch) {31 .Locked => null,
33 .i386, .x86_64 => asm volatile ("pause"),32 };
34 .arm, .aarch64 => asm volatile ("yield"),
35 else => time.sleep(0),
36 }
37 }
38 }33 }
3934
40 /// Provides a method to incrementally yield longer each time its called.35 pub fn acquire(self: *SpinLock) Held {
41 pub const Backoff = struct {36 while (true) {
42 iteration: usize,37 return self.tryAcquire() orelse {
38 yield();
39 continue;
40 };
41 }
42 }
4343
44 pub fn init() @This() {44 pub fn yield() void {
45 return @This(){ .iteration = 0 };45 // On native windows, SwitchToThread is too expensive,
46 // and yielding for 380-410 iterations was found to be
47 // a nice sweet spot. Posix systems on the other hand,
48 // especially linux, perform better by yielding the thread.
49 switch (builtin.os) {
50 .windows => loopHint(400),
51 else => std.os.sched_yield() catch loopHint(1),
46 }52 }
53 }
4754
48 /// Modified hybrid yielding from55 /// Hint to the cpu that execution is spinning
49 /// http://www.1024cores.net/home/lock-free-algorithms/tricks/spinning56 /// for the given amount of iterations.
50 pub fn yield(self: *@This()) void {57 pub fn loopHint(iterations: usize) void {
51 defer self.iteration +%= 1;58 var i = iterations;
52 if (self.iteration < 20) {59 while (i != 0) : (i -= 1) {
53 SpinLock.yield(self.iteration);60 switch (builtin.arch) {
54 } else if (self.iteration < 24) {61 // these instructions use a memory clobber as they
55 os.sched_yield() catch time.sleep(1);62 // flush the pipeline of any speculated reads/writes.
56 } else if (self.iteration < 26) {63 .i386, .x86_64 => asm volatile ("pause" ::: "memory"),
57 time.sleep(1 * time.millisecond);64 .arm, .aarch64 => asm volatile ("yield" ::: "memory"),
58 } else {65 else => std.os.sched_yield() catch {},
59 time.sleep(10 * time.millisecond);
60 }66 }
61 }67 }
62 };68 }
63};69};
6470
65test "spinlock" {71test "spinlock" {
66 var lock = SpinLock.init();72 var lock = SpinLock.init();
73 defer lock.deinit();
74
67 const held = lock.acquire();75 const held = lock.acquire();
68 defer held.release();76 defer held.release();
69}77}