authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2020-09-28 17:30:36-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2020-09-28 17:30:36-04:00
loga0c0f9ead53dab4f558dbb51cc8a49961fc6984f
tree8261d216d0db8c5d46c010c0035e0aeb66391c6f
parent5c6cd5e2c9e8b2d0feb0026bad7c201035a175b4
parent468a4bf0b443067b9d4a0bf68ea7e675a9bca727
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #6441 from kprotty/lock

New std.event.Lock implementation

1 files changed, 90 insertions(+), 115 deletions(-)

lib/std/event/lock.zig+90-115
...@@ -16,107 +16,107 @@ const Loop = std.event.Loop;...@@ -16,107 +16,107 @@ const Loop = std.event.Loop;
16/// Allows only one actor to hold the lock.16/// Allows only one actor to hold the lock.
17/// TODO: make this API also work in blocking I/O mode.17/// TODO: make this API also work in blocking I/O mode.
18pub const Lock = struct {18pub const Lock = struct {
19 shared: bool,19 mutex: std.Mutex = std.Mutex{},
20 queue: Queue,20 head: usize = UNLOCKED,
21 queue_empty: bool,
2221
23 const Queue = std.atomic.Queue(anyframe);22 const UNLOCKED = 0;
23 const LOCKED = 1;
2424
25 const global_event_loop = Loop.instance orelse25 const global_event_loop = Loop.instance orelse
26 @compileError("std.event.Lock currently only works with event-based I/O");26 @compileError("std.event.Lock currently only works with event-based I/O");
2727
28 pub const Held = struct {28 const Waiter = struct {
29 lock: *Lock,29 // forced Waiter alignment to ensure it doesn't clash with LOCKED
3030 next: ?*Waiter align(2),
31 pub fn release(self: Held) void {31 tail: *Waiter,
32 // Resume the next item from the queue.32 node: Loop.NextTickNode,
33 if (self.lock.queue.get()) |node| {33 };
34 global_event_loop.onNextTick(node);
35 return;
36 }
37
38 // We need to release the lock.
39 @atomicStore(bool, &self.lock.queue_empty, true, .SeqCst);
40 @atomicStore(bool, &self.lock.shared, false, .SeqCst);
41
42 // There might be a queue item. If we know the queue is empty, we can be done,
43 // because the other actor will try to obtain the lock.
44 // But if there's a queue item, we are the actor which must loop and attempt
45 // to grab the lock again.
46 if (@atomicLoad(bool, &self.lock.queue_empty, .SeqCst)) {
47 return;
48 }
49
50 while (true) {
51 if (@atomicRmw(bool, &self.lock.shared, .Xchg, true, .SeqCst)) {
52 // We did not obtain the lock. Great, the queue is someone else's problem.
53 return;
54 }
55
56 // Resume the next item from the queue.
57 if (self.lock.queue.get()) |node| {
58 global_event_loop.onNextTick(node);
59 return;
60 }
61
62 // Release the lock again.
63 @atomicStore(bool, &self.lock.queue_empty, true, .SeqCst);
64 @atomicStore(bool, &self.lock.shared, false, .SeqCst);
6534
66 // Find out if we can be done.35 pub fn acquire(self: *Lock) Held {
67 if (@atomicLoad(bool, &self.lock.queue_empty, .SeqCst)) {36 const held = self.mutex.acquire();
68 return;37
69 }38 // self.head transitions from multiple stages depending on the value:
70 }39 // UNLOCKED -> LOCKED:
40 // acquire Lock ownership when theres no waiters
41 // LOCKED -> <Waiter head ptr>:
42 // Lock is already owned, enqueue first Waiter
43 // <head ptr> -> <head ptr>:
44 // Lock is owned with pending waiters. Push our waiter to the queue.
45
46 if (self.head == UNLOCKED) {
47 self.head = LOCKED;
48 held.release();
49 return Held{ .lock = self };
71 }50 }
72 };
7351
74 pub fn init() Lock {52 var waiter: Waiter = undefined;
75 return Lock{53 waiter.next = null;
76 .shared = false,54 waiter.tail = &waiter;
77 .queue = Queue.init(),
78 .queue_empty = true,
79 };
80 }
8155
82 pub fn initLocked() Lock {56 const head = switch (self.head) {
83 return Lock{57 UNLOCKED => unreachable,
84 .shared = true,58 LOCKED => null,
85 .queue = Queue.init(),59 else => @intToPtr(*Waiter, self.head),
86 .queue_empty = true,
87 };60 };
88 }
89
90 /// Must be called when not locked. Not thread safe.
91 /// All calls to acquire() and release() must complete before calling deinit().
92 pub fn deinit(self: *Lock) void {
93 assert(!self.shared);
94 while (self.queue.get()) |node| resume node.data;
95 }
9661
97 pub fn acquire(self: *Lock) callconv(.Async) Held {62 if (head) |h| {
98 var my_tick_node = Loop.NextTickNode.init(@frame());63 h.tail.next = &waiter;
64 h.tail = &waiter;
65 } else {
66 self.head = @ptrToInt(&waiter);
67 }
9968
100 errdefer _ = self.queue.remove(&my_tick_node); // TODO test canceling an acquire
101 suspend {69 suspend {
102 self.queue.put(&my_tick_node);70 waiter.node = Loop.NextTickNode{
10371 .prev = undefined,
104 // At this point, we are in the queue, so we might have already been resumed.72 .next = undefined,
73 .data = @frame(),
74 };
75 held.release();
76 }
10577
106 // We set this bit so that later we can rely on the fact, that if queue_empty == true, some actor78 return Held{ .lock = self };
107 // will attempt to grab the lock.79 }
108 @atomicStore(bool, &self.queue_empty, false, .SeqCst);
10980
110 if (!@atomicRmw(bool, &self.shared, .Xchg, true, .SeqCst)) {81 pub const Held = struct {
111 if (self.queue.get()) |node| {82 lock: *Lock,
112 // Whether this node is us or someone else, we tail resume it.83
113 resume node.data;84 pub fn release(self: Held) void {
85 const waiter = blk: {
86 const held = self.lock.mutex.acquire();
87 defer held.release();
88
89 // self.head goes through the reverse transition from acquire():
90 // <head ptr> -> <new head ptr>:
91 // pop a waiter from the queue to give Lock ownership when theres still others pending
92 // <head ptr> -> LOCKED:
93 // pop the laster waiter from the queue, while also giving it lock ownership when awaken
94 // LOCKED -> UNLOCKED:
95 // last lock owner releases lock while no one else is waiting for it
96
97 switch (self.lock.head) {
98 UNLOCKED => {
99 unreachable; // Lock unlocked while unlocking
100 },
101 LOCKED => {
102 self.lock.head = UNLOCKED;
103 break :blk null;
104 },
105 else => {
106 const waiter = @intToPtr(*Waiter, self.lock.head);
107 self.lock.head = if (waiter.next == null) LOCKED else @ptrToInt(waiter.next);
108 if (waiter.next) |next|
109 next.tail = waiter.tail;
110 break :blk waiter;
111 },
114 }112 }
113 };
114
115 if (waiter) |w| {
116 global_event_loop.onNextTick(&w.node);
115 }117 }
116 }118 }
117119 };
118 return Held{ .lock = self };
119 }
120};120};
121121
122test "std.event.Lock" {122test "std.event.Lock" {
...@@ -128,41 +128,16 @@ test "std.event.Lock" {...@@ -128,41 +128,16 @@ test "std.event.Lock" {
128 // TODO https://github.com/ziglang/zig/issues/3251128 // TODO https://github.com/ziglang/zig/issues/3251
129 if (builtin.os.tag == .freebsd) return error.SkipZigTest;129 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
130130
131 // TODO this file has bit-rotted. repair it131 var lock = Lock{};
132 if (true) return error.SkipZigTest;132 testLock(&lock);
133
134 var lock = Lock.init();
135 defer lock.deinit();
136
137 _ = async testLock(&lock);
138133
139 const expected_result = [1]i32{3 * @intCast(i32, shared_test_data.len)} ** shared_test_data.len;134 const expected_result = [1]i32{3 * @intCast(i32, shared_test_data.len)} ** shared_test_data.len;
140 testing.expectEqualSlices(i32, &expected_result, &shared_test_data);135 testing.expectEqualSlices(i32, &expected_result, &shared_test_data);
141}136}
142fn testLock(lock: *Lock) callconv(.Async) void {137fn testLock(lock: *Lock) void {
143 var handle1 = async lockRunner(lock);138 var handle1 = async lockRunner(lock);
144 var tick_node1 = Loop.NextTickNode{
145 .prev = undefined,
146 .next = undefined,
147 .data = &handle1,
148 };
149 Loop.instance.?.onNextTick(&tick_node1);
150
151 var handle2 = async lockRunner(lock);139 var handle2 = async lockRunner(lock);
152 var tick_node2 = Loop.NextTickNode{
153 .prev = undefined,
154 .next = undefined,
155 .data = &handle2,
156 };
157 Loop.instance.?.onNextTick(&tick_node2);
158
159 var handle3 = async lockRunner(lock);140 var handle3 = async lockRunner(lock);
160 var tick_node3 = Loop.NextTickNode{
161 .prev = undefined,
162 .next = undefined,
163 .data = &handle3,
164 };
165 Loop.instance.?.onNextTick(&tick_node3);
166141
167 await handle1;142 await handle1;
168 await handle2;143 await handle2;
...@@ -171,13 +146,13 @@ fn testLock(lock: *Lock) callconv(.Async) void {...@@ -171,13 +146,13 @@ fn testLock(lock: *Lock) callconv(.Async) void {
171146
172var shared_test_data = [1]i32{0} ** 10;147var shared_test_data = [1]i32{0} ** 10;
173var shared_test_index: usize = 0;148var shared_test_index: usize = 0;
174fn lockRunner(lock: *Lock) callconv(.Async) void {149
175 suspend; // resumed by onNextTick150fn lockRunner(lock: *Lock) void {
151 Lock.global_event_loop.yield();
176152
177 var i: usize = 0;153 var i: usize = 0;
178 while (i < shared_test_data.len) : (i += 1) {154 while (i < shared_test_data.len) : (i += 1) {
179 var lock_frame = async lock.acquire();155 const handle = lock.acquire();
180 const handle = await lock_frame;
181 defer handle.release();156 defer handle.release();
182157
183 shared_test_index = 0;158 shared_test_index = 0;