authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-10-03 14:55:12-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-10-03 14:55:12-04:00
log3f13a59cbc4235a5abcc2ca35bbc3f172336fd61
treeabfac8c81ff70fee75d81c07aae4ac4080218e00
parent66cb75d1148fffdd161e7829b9e27aa52f0f1616
signature Commit is signed but in an unrecognized format.

better mutex implementation

based on Ulrich Drepper's "Futexes are tricky" paper, Mutex, Take 3 also includes tests

1 files changed, 61 insertions(+), 20 deletions(-)

std/mutex.zig+61-20
......@@ -8,6 +8,8 @@ const linux = std.os.linux;
88
99/// Lock may be held only once. If the same thread
1010/// tries to acquire the same mutex twice, it deadlocks.
11/// The Linux implementation is based on mutex3 from
12/// https://www.akkadia.org/drepper/futex.pdf
1113pub const Mutex = struct {
1214 /// 0: unlocked
1315 /// 1: locked, no waiters
......@@ -25,12 +27,10 @@ pub const Mutex = struct {
2527
2628 pub fn release(self: Held) void {
2729 if (builtin.os == builtin.Os.linux) {
28 // Always unlock. If the previous state was Locked-No-Waiters, then we're done.
29 // Otherwise, wake a waiter up.
30 const prev = @atomicRmw(i32, &self.mutex.linux_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.Release);
31 if (prev != 1) {
32 assert(prev == 2);
33 const rc = linux.futex_wake(&self.mutex.linux_lock, linux.FUTEX_WAKE, 1);
30 const c = @atomicRmw(i32, &self.mutex.linux_lock, AtomicRmwOp.Sub, 1, AtomicOrder.Release);
31 if (c != 1) {
32 _ = @atomicRmw(i32, &self.mutex.linux_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.Release);
33 const rc = linux.futex_wake(&self.mutex.linux_lock, linux.FUTEX_WAKE | linux.FUTEX_PRIVATE_FLAG, 1);
3434 switch (linux.getErrno(rc)) {
3535 0 => {},
3636 linux.EINVAL => unreachable,
......@@ -52,21 +52,18 @@ pub const Mutex = struct {
5252
5353 pub fn acquire(self: *Mutex) Held {
5454 if (builtin.os == builtin.Os.linux) {
55 // First try to go from Unlocked to Locked-No-Waiters. If this succeeds, no syscalls are needed.
56 // Otherwise, we need to be in the Locked-With-Waiters state. If we are already in that state,
57 // proceed to futex_wait. Otherwise, try to go from Locked-No-Waiters to Locked-With-Waiters.
58 // If that succeeds, proceed to futex_wait. Otherwise start the whole loop over again.
59 while (@cmpxchgWeak(i32, &self.linux_lock, 0, 1, AtomicOrder.Acquire, AtomicOrder.Monotonic)) |l| {
60 if (l == 2 or
61 @cmpxchgWeak(i32, &self.linux_lock, 1, 2, AtomicOrder.Acquire, AtomicOrder.Monotonic) == null)
62 {
63 const rc = linux.futex_wait(&self.linux_lock, linux.FUTEX_WAIT, 2, null);
64 switch (linux.getErrno(rc)) {
65 0, linux.EINTR, linux.EAGAIN => continue,
66 linux.EINVAL => unreachable,
67 else => unreachable,
68 }
55 var c = @cmpxchgWeak(i32, &self.linux_lock, 0, 1, AtomicOrder.Acquire, AtomicOrder.Monotonic) orelse
56 return Held{ .mutex = self };
57 if (c != 2)
58 c = @atomicRmw(i32, &self.linux_lock, AtomicRmwOp.Xchg, 2, AtomicOrder.Acquire);
59 while (c != 0) {
60 const rc = linux.futex_wait(&self.linux_lock, linux.FUTEX_WAIT | linux.FUTEX_PRIVATE_FLAG, 2, null);
61 switch (linux.getErrno(rc)) {
62 0, linux.EINTR, linux.EAGAIN => {},
63 linux.EINVAL => unreachable,
64 else => unreachable,
6965 }
66 c = @atomicRmw(i32, &self.linux_lock, AtomicRmwOp.Xchg, 2, AtomicOrder.Acquire);
7067 }
7168 } else {
7269 _ = self.spin_lock.acquire();
......@@ -74,3 +71,47 @@ pub const Mutex = struct {
7471 return Held{ .mutex = self };
7572 }
7673};
74
75const Context = struct {
76 mutex: *Mutex,
77 data: i128,
78
79 const incr_count = 10000;
80};
81
82test "std.Mutex" {
83 var direct_allocator = std.heap.DirectAllocator.init();
84 defer direct_allocator.deinit();
85
86 var plenty_of_memory = try direct_allocator.allocator.alloc(u8, 300 * 1024);
87 defer direct_allocator.allocator.free(plenty_of_memory);
88
89 var fixed_buffer_allocator = std.heap.ThreadSafeFixedBufferAllocator.init(plenty_of_memory);
90 var a = &fixed_buffer_allocator.allocator;
91
92 var mutex = Mutex.init();
93 var context = Context{
94 .mutex = &mutex,
95 .data = 0,
96 };
97
98 const thread_count = 10;
99 var threads: [thread_count]*std.os.Thread = undefined;
100 for (threads) |*t| {
101 t.* = try std.os.spawnThread(&context, worker);
102 }
103 for (threads) |t|
104 t.wait();
105
106 std.debug.assertOrPanic(context.data == thread_count * Context.incr_count);
107}
108
109fn worker(ctx: *Context) void {
110 var i: usize = 0;
111 while (i != Context.incr_count) : (i += 1) {
112 const held = ctx.mutex.acquire();
113 defer held.release();
114
115 ctx.data += 1;
116 }
117}