authorgravatar for kbutcher6200@gmail.comkprotty <kbutcher6200@gmail.com> 2019-11-05 16:14:43-06:00
committergravatar for kbutcher6200@gmail.comkprotty <kbutcher6200@gmail.com> 2019-11-05 16:14:43-06:00
logc9db420a09ad894c1e521d1bdf1c99a3424aa9a6
tree7007925a1ade7c0b0536d31fcf649ffbcdf3a5d1
parent465ebf494df070ddb7eab9ba7787aacec394b1cc

Adaptive Mutex:

- uses std.ThreadParker - supports static initialization (deprecates StaticallyInitializedMutex)

3 files changed, 102 insertions(+), 77 deletions(-)

lib/std/mutex.zig+63-74
......@@ -1,19 +1,13 @@
11const std = @import("std.zig");
22const builtin = @import("builtin");
3const AtomicOrder = builtin.AtomicOrder;
4const AtomicRmwOp = builtin.AtomicRmwOp;
53const testing = std.testing;
64const SpinLock = std.SpinLock;
7const linux = std.os.linux;
8const windows = std.os.windows;
5const ThreadParker = std.ThreadParker;
96
107/// Lock may be held only once. If the same thread
118/// tries to acquire the same mutex twice, it deadlocks.
12/// This type must be initialized at runtime, and then deinitialized when no
13/// longer needed, to free resources.
14/// If you need static initialization, use std.StaticallyInitializedMutex.
15/// The Linux implementation is based on mutex3 from
16/// https://www.akkadia.org/drepper/futex.pdf
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
1711/// When an application is built in single threaded release mode, all the functions are
1812/// no-ops. In single threaded debug mode, there is deadlock detection.
1913pub const Mutex = if (builtin.single_threaded)
......@@ -43,83 +37,78 @@ pub const Mutex = if (builtin.single_threaded)
4337 return Held{ .mutex = self };
4438 }
4539 }
46else switch (builtin.os) {
47 builtin.Os.linux => struct {
48 /// 0: unlocked
49 /// 1: locked, no waiters
50 /// 2: locked, one or more waiters
51 lock: i32,
52
53 pub const Held = struct {
54 mutex: *Mutex,
55
56 pub fn release(self: Held) void {
57 const c = @atomicRmw(i32, &self.mutex.lock, AtomicRmwOp.Sub, 1, AtomicOrder.Release);
58 if (c != 1) {
59 _ = @atomicRmw(i32, &self.mutex.lock, AtomicRmwOp.Xchg, 0, AtomicOrder.Release);
60 const rc = linux.futex_wake(&self.mutex.lock, linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
61 switch (linux.getErrno(rc)) {
62 0 => {},
63 linux.EINVAL => unreachable,
64 else => unreachable,
65 }
66 }
67 }
40else struct {
41 state: u32, // TODO: make this an enum
42 parker: ThreadParker,
43
44 const Unlocked = 0;
45 const Sleeping = 1;
46 const Locked = 2;
47
48 /// number of iterations to spin yielding the cpu
49 const SpinCpu = 4;
50 /// number of iterations to perform in the cpu yield loop
51 const SpinCpuCount = 30;
52 /// number of iterations to spin yielding the thread
53 const SpinThread = 1;
54
55 pub fn init() Mutex {
56 return Mutex{
57 .state = Unlocked,
58 .parker = ThreadParker.init(),
6859 };
60 }
6961
70 pub fn init() Mutex {
71 return Mutex{ .lock = 0 };
72 }
62 pub fn deinit(self: *Mutex) void {
63 self.parker.deinit();
64 }
7365
74 pub fn deinit(self: *Mutex) void {}
66 pub const Held = struct {
67 mutex: *Mutex,
7568
76 pub fn acquire(self: *Mutex) Held {
77 var c = @cmpxchgWeak(i32, &self.lock, 0, 1, AtomicOrder.Acquire, AtomicOrder.Monotonic) orelse
78 return Held{ .mutex = self };
79 if (c != 2)
80 c = @atomicRmw(i32, &self.lock, AtomicRmwOp.Xchg, 2, AtomicOrder.Acquire);
81 while (c != 0) {
82 const rc = linux.futex_wait(&self.lock, linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, 2, null);
83 switch (linux.getErrno(rc)) {
84 0, linux.EINTR, linux.EAGAIN => {},
85 linux.EINVAL => unreachable,
86 else => unreachable,
87 }
88 c = @atomicRmw(i32, &self.lock, AtomicRmwOp.Xchg, 2, AtomicOrder.Acquire);
69 pub fn release(self: Held) void {
70 switch (@atomicRmw(u32, &self.mutex.state, .Xchg, Unlocked, .Release)) {
71 Locked => {},
72 Sleeping => self.mutex.parker.unpark(&self.mutex.state),
73 Unlocked => unreachable, // unlocking an unlocked mutex
74 else => unreachable, // should never be anything else
8975 }
90 return Held{ .mutex = self };
9176 }
92 },
93 // TODO once https://github.com/ziglang/zig/issues/287 (copy elision) is solved, we can make a
94 // better implementation of this. The problem is we need the init() function to have access to
95 // the address of the CRITICAL_SECTION, and then have it not move.
96 builtin.Os.windows => std.StaticallyInitializedMutex,
97 else => struct {
98 /// TODO better implementation than spin lock.
99 /// When changing this, one must also change the corresponding
100 /// std.StaticallyInitializedMutex code, since it aliases this type,
101 /// under the assumption that it works both statically and at runtime.
102 lock: SpinLock,
77 };
10378
104 pub const Held = struct {
105 mutex: *Mutex,
79 pub fn acquire(self: *Mutex) Held {
80 // Try and speculatively grab the lock.
81 // If it fails, the state is either Locked or Sleeping
82 // depending on if theres a thread stuck sleeping below.
83 var state = @atomicRmw(u32, &self.state, .Xchg, Locked, .Acquire);
84 if (state == Unlocked)
85 return Held{ .mutex = self };
10686
107 pub fn release(self: Held) void {
108 SpinLock.Held.release(SpinLock.Held{ .spinlock = &self.mutex.lock });
87 while (true) {
88 // try and acquire the lock using cpu spinning on failure
89 for (([SpinCpu]void)(undefined)) |_| {
90 var value = @atomicLoad(u32, &self.state, .Monotonic);
91 while (value == Unlocked)
92 value = @cmpxchgWeak(u32, &self.state, Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
93 for (([SpinCpuCount]void)(undefined)) |_|
94 SpinLock.yieldCpu();
10995 }
110 };
11196
112 pub fn init() Mutex {
113 return Mutex{ .lock = SpinLock.init() };
114 }
115
116 pub fn deinit(self: *Mutex) void {}
97 // try and acquire the lock using thread rescheduling on failure
98 for (([SpinThread]void)(undefined)) |_| {
99 var value = @atomicLoad(u32, &self.state, .Monotonic);
100 while (value == Unlocked)
101 value = @cmpxchgWeak(u32, &self.state, Unlocked, state, .Acquire, .Monotonic) orelse return Held{ .mutex = self };
102 SpinLock.yieldThread();
103 }
117104
118 pub fn acquire(self: *Mutex) Held {
119 _ = self.lock.acquire();
120 return Held{ .mutex = self };
105 // failed to acquire the lock, go to sleep until woken up by `Held.release()`
106 if (@atomicRmw(u32, &self.state, .Xchg, Sleeping, .Acquire) == Unlocked)
107 return Held{ .mutex = self };
108 state = Sleeping;
109 self.parker.park(&self.state, Sleeping);
121110 }
122 },
111 }
123112};
124113
125114const TestContext = struct {
lib/std/parker.zig+37-1
......@@ -159,7 +159,7 @@ const WindowsParker = struct {
159159 const key = @ptrCast(*const c_void, ptr);
160160 var waiting = @atomicLoad(u32, waiters, .Acquire);
161161 while (waiting != 0) {
162 waiting = @cmpxchgWeak(u32, waiters, waiting, waiting - 1, .AcqRel, .Monotonic) orelse {
162 waiting = @cmpxchgWeak(u32, waiters, waiting, waiting - 1, .Acquire, .Monotonic) orelse {
163163 const rc = windows.ntdll.NtReleaseKeyedEvent(self.handle, key, windows.FALSE, null);
164164 assert(rc == 0);
165165 return;
......@@ -338,3 +338,39 @@ const PosixParker = struct {
338338 else => unreachable,
339339 };
340340};
341
342test "std.ThreadParker" {
343 const Context = struct {
344 parker: ThreadParker,
345 data: u32,
346
347 fn receiver(self: *@This()) void {
348 self.parker.park(&self.data, 0); // receives 1
349 assert(@atomicRmw(u32, &self.data, .Xchg, 2, .SeqCst) == 1); // sends 2
350 self.parker.unpark(&self.data); // wakes up waiters on 2
351 self.parker.park(&self.data, 2); // receives 3
352 assert(@atomicRmw(u32, &self.data, .Xchg, 4, .SeqCst) == 3); // sends 4
353 self.parker.unpark(&self.data); // wakes up waiters on 4
354 }
355
356 fn sender(self: *@This()) void {
357 assert(@atomicRmw(u32, &self.data, .Xchg, 1, .SeqCst) == 0); // sends 1
358 self.parker.unpark(&self.data); // wakes up waiters on 1
359 self.parker.park(&self.data, 1); // receives 2
360 assert(@atomicRmw(u32, &self.data, .Xchg, 3, .SeqCst) == 2); // sends 3
361 self.parker.unpark(&self.data); // wakes up waiters on 3
362 self.parker.park(&self.data, 3); // receives 4
363 }
364 };
365
366 var context = Context{
367 .parker = ThreadParker.init(),
368 .data = 0,
369 };
370 defer context.parker.deinit();
371
372 var receiver = try std.Thread.spawn(&context, Context.receiver);
373 defer receiver.wait();
374
375 context.sender();
376}
\ No newline at end of file
lib/std/spinlock.zig+2-2
......@@ -28,7 +28,7 @@ pub const SpinLock = struct {
2828 return Held{ .spinlock = self };
2929 }
3030
31 fn yieldCpu() void {
31 pub fn yieldCpu() void {
3232 switch (builtin.arch) {
3333 .i386, .x86_64 => asm volatile("pause" ::: "memory"),
3434 .arm, .aarch64 => asm volatile("yield"),
......@@ -36,7 +36,7 @@ pub const SpinLock = struct {
3636 }
3737 }
3838
39 fn yieldThread() void {
39 pub fn yieldThread() void {
4040 switch (builtin.os) {
4141 .linux => assert(linux.syscall0(linux.SYS_sched_yield) == 0),
4242 .windows => _ = windows.kernel32.SwitchToThread(),