authorgravatar for 45520026+kprotty@users.noreply.github.comkprotty <45520026+kprotty@users.noreply.github.com> 2020-09-27 14:05:38-05:00
committergravatar for 45520026+kprotty@users.noreply.github.comkprotty <45520026+kprotty@users.noreply.github.com> 2020-09-27 14:05:38-05:00
loga31d9f92f282a878836a3ecac5a48d5f4037868c
treea8fc0335698ab4b3fc2aa71eb1cf91a5cb1c9d5c
parent8794ce6f79886e5ebbf0476d56917e219b52c561

new std.event.Lock implementation

1 files changed, 72 insertions(+), 114 deletions(-)

lib/std/event/lock.zig+72-114
......@@ -16,107 +16,90 @@ const Loop = std.event.Loop;
1616/// Allows only one actor to hold the lock.
1717/// TODO: make this API also work in blocking I/O mode.
1818pub const Lock = struct {
19 shared: bool,
20 queue: Queue,
21 queue_empty: bool,
19 mutex: std.Mutex = std.Mutex{},
20 head: usize = UNLOCKED,
2221
23 const Queue = std.atomic.Queue(anyframe);
22 const UNLOCKED = 0;
23 const LOCKED = 69;
2424
2525 const global_event_loop = Loop.instance orelse
2626 @compileError("std.event.Lock currently only works with event-based I/O");
2727
28 pub const Held = struct {
29 lock: *Lock,
30
31 pub fn release(self: Held) void {
32 // Resume the next item from the queue.
33 if (self.lock.queue.get()) |node| {
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 }
28 const Waiter = struct {
29 next: ?*Waiter,
30 tail: *Waiter,
31 node: Loop.NextTickNode,
32 };
6133
62 // Release the lock again.
63 @atomicStore(bool, &self.lock.queue_empty, true, .SeqCst);
64 @atomicStore(bool, &self.lock.shared, false, .SeqCst);
34 pub fn acquire(self: *Lock) Held {
35 const held = self.mutex.acquire();
6536
66 // Find out if we can be done.
67 if (@atomicLoad(bool, &self.lock.queue_empty, .SeqCst)) {
68 return;
69 }
70 }
37 if (self.head == UNLOCKED) {
38 self.head = LOCKED;
39 held.release();
40 return Held{ .lock = self };
7141 }
72 };
7342
74 pub fn init() Lock {
75 return Lock{
76 .shared = false,
77 .queue = Queue.init(),
78 .queue_empty = true,
79 };
80 }
43 var waiter: Waiter = undefined;
44 waiter.next = null;
45 waiter.tail = &waiter;
8146
82 pub fn initLocked() Lock {
83 return Lock{
84 .shared = true,
85 .queue = Queue.init(),
86 .queue_empty = true,
47 const head = switch (self.head) {
48 UNLOCKED => unreachable,
49 LOCKED => null,
50 else => @intToPtr(?*Waiter, self.head),
8751 };
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 }
9652
97 pub fn acquire(self: *Lock) callconv(.Async) Held {
98 var my_tick_node = Loop.NextTickNode.init(@frame());
53 if (head) |h| {
54 h.tail.next = &waiter;
55 h.tail = &waiter;
56 } else {
57 self.head = @ptrToInt(&waiter);
58 }
9959
100 errdefer _ = self.queue.remove(&my_tick_node); // TODO test canceling an acquire
10160 suspend {
102 self.queue.put(&my_tick_node);
103
104 // At this point, we are in the queue, so we might have already been resumed.
61 waiter.node = Loop.NextTickNode{
62 .prev = undefined,
63 .next = undefined,
64 .data = @frame(),
65 };
66 held.release();
67 }
10568
106 // We set this bit so that later we can rely on the fact, that if queue_empty == true, some actor
107 // will attempt to grab the lock.
108 @atomicStore(bool, &self.queue_empty, false, .SeqCst);
69 return Held{ .lock = self };
70 }
10971
110 if (!@atomicRmw(bool, &self.shared, .Xchg, true, .SeqCst)) {
111 if (self.queue.get()) |node| {
112 // Whether this node is us or someone else, we tail resume it.
113 resume node.data;
72 pub const Held = struct {
73 lock: *Lock,
74
75 pub fn release(self: Held) void {
76 const waiter = blk: {
77 const held = self.lock.mutex.acquire();
78 defer held.release();
79
80 switch (self.lock.head) {
81 UNLOCKED => {
82 std.debug.panic("Lock unlocked when already unlocked", .{});
83 },
84 LOCKED => {
85 self.lock.head = UNLOCKED;
86 break :blk null;
87 },
88 else => {
89 const waiter = @intToPtr(*Waiter, self.lock.head);
90 self.lock.head = if (waiter.next == null) LOCKED else @ptrToInt(waiter.next);
91 if (waiter.next) |next|
92 next.tail = waiter.tail;
93 break :blk waiter;
94 },
11495 }
96 };
97
98 if (waiter) |w| {
99 global_event_loop.onNextTick(&w.node);
115100 }
116101 }
117
118 return Held{ .lock = self };
119 }
102 };
120103};
121104
122105test "std.event.Lock" {
......@@ -128,41 +111,16 @@ test "std.event.Lock" {
128111 // TODO https://github.com/ziglang/zig/issues/3251
129112 if (builtin.os.tag == .freebsd) return error.SkipZigTest;
130113
131 // TODO this file has bit-rotted. repair it
132 if (true) return error.SkipZigTest;
133
134 var lock = Lock.init();
135 defer lock.deinit();
136
137 _ = async testLock(&lock);
114 var lock = Lock{};
115 testLock(&lock);
138116
139117 const expected_result = [1]i32{3 * @intCast(i32, shared_test_data.len)} ** shared_test_data.len;
140118 testing.expectEqualSlices(i32, &expected_result, &shared_test_data);
141119}
142fn testLock(lock: *Lock) callconv(.Async) void {
120fn testLock(lock: *Lock) void {
143121 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
151122 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
159123 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);
166124
167125 await handle1;
168126 await handle2;
......@@ -171,13 +129,13 @@ fn testLock(lock: *Lock) callconv(.Async) void {
171129
172130var shared_test_data = [1]i32{0} ** 10;
173131var shared_test_index: usize = 0;
174fn lockRunner(lock: *Lock) callconv(.Async) void {
175 suspend; // resumed by onNextTick
132
133fn lockRunner(lock: *Lock) void {
134 Lock.global_event_loop.yield();
176135
177136 var i: usize = 0;
178137 while (i < shared_test_data.len) : (i += 1) {
179 var lock_frame = async lock.acquire();
180 const handle = await lock_frame;
138 const handle = lock.acquire();
181139 defer handle.release();
182140
183141 shared_test_index = 0;