authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-08-01 16:26:37-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-08-01 16:26:37-04:00
loge3ae2cfb5243e7255bf4dbcc8a9b7e77a31e9d45
tree73dc936ec76d72e8ddeb49e57926716f967d18a2
parent3c8d4e04ea000d087af4e77331340db1c8b1cef3

add std.event.RwLock and a few more std changes

* add std.event.RwLock and std.event.RwLocked * std.debug.warn does its printing locked * add std.Mutex, however it's currently implemented as a spinlock * rename std.event.Group.cancelAll to std.event.Group.deinit and change the docs and assumptions. * add std.HashMap.clone

13 files changed, 422 insertions(+), 18 deletions(-)

CMakeLists.txt+3
......@@ -466,6 +466,8 @@ set(ZIG_STD_FILES
466466 "event/lock.zig"
467467 "event/locked.zig"
468468 "event/loop.zig"
469 "event/rwlock.zig"
470 "event/rwlocked.zig"
469471 "event/tcp.zig"
470472 "fmt/errol/enum3.zig"
471473 "fmt/errol/index.zig"
......@@ -554,6 +556,7 @@ set(ZIG_STD_FILES
554556 "math/tanh.zig"
555557 "math/trunc.zig"
556558 "mem.zig"
559 "mutex.zig"
557560 "net.zig"
558561 "os/child_process.zig"
559562 "os/darwin.zig"
std/debug/index.zig+3
......@@ -23,7 +23,10 @@ pub const runtime_safety = switch (builtin.mode) {
2323var stderr_file: os.File = undefined;
2424var stderr_file_out_stream: io.FileOutStream = undefined;
2525var stderr_stream: ?*io.OutStream(io.FileOutStream.Error) = null;
26var stderr_mutex = std.Mutex.init();
2627pub fn warn(comptime fmt: []const u8, args: ...) void {
28 const held = stderr_mutex.acquire();
29 defer held.release();
2730 const stderr = getStderrStream() catch return;
2831 stderr.print(fmt, args) catch return;
2932}
std/event.zig+4
......@@ -3,6 +3,8 @@ pub const Future = @import("event/future.zig").Future;
33pub const Group = @import("event/group.zig").Group;
44pub const Lock = @import("event/lock.zig").Lock;
55pub const Locked = @import("event/locked.zig").Locked;
6pub const RwLock = @import("event/rwlock.zig").Lock;
7pub const RwLocked = @import("event/rwlocked.zig").RwLocked;
68pub const Loop = @import("event/loop.zig").Loop;
79pub const fs = @import("event/fs.zig");
810pub const tcp = @import("event/tcp.zig");
......@@ -14,6 +16,8 @@ test "import event tests" {
1416 _ = @import("event/group.zig");
1517 _ = @import("event/lock.zig");
1618 _ = @import("event/locked.zig");
19 _ = @import("event/rwlock.zig");
20 _ = @import("event/rwlocked.zig");
1721 _ = @import("event/loop.zig");
1822 _ = @import("event/tcp.zig");
1923}
std/event/channel.zig+4
......@@ -116,6 +116,10 @@ pub fn Channel(comptime T: type) type {
116116 return result;
117117 }
118118
119 fn getOrNull(self: *SelfChannel) ?T {
120 TODO();
121 }
122
119123 fn dispatch(self: *SelfChannel) void {
120124 // set the "need dispatch" flag
121125 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
std/event/fs.zig+3-1
......@@ -253,7 +253,9 @@ pub async fn openReadWrite(
253253}
254254
255255/// This abstraction helps to close file handles in defer expressions
256/// without suspending. Start a CloseOperation before opening a file.
256/// without the possibility of failure and without the use of suspend points.
257/// Start a `CloseOperation` before opening a file, so that you can defer
258/// `CloseOperation.deinit`.
257259pub const CloseOperation = struct {
258260 loop: *event.Loop,
259261 have_fd: bool,
std/event/group.zig+13-15
......@@ -29,6 +29,17 @@ pub fn Group(comptime ReturnType: type) type {
2929 };
3030 }
3131
32 /// Cancel all the outstanding promises. Can be called even if wait was already called.
33 pub fn deinit(self: *Self) void {
34 while (self.coro_stack.pop()) |node| {
35 cancel node.data;
36 }
37 while (self.alloc_stack.pop()) |node| {
38 cancel node.data;
39 self.lock.loop.allocator.destroy(node);
40 }
41 }
42
3243 /// Add a promise to the group. Thread-safe.
3344 pub fn add(self: *Self, handle: promise->ReturnType) (error{OutOfMemory}!void) {
3445 const node = try self.lock.loop.allocator.create(Stack.Node{
......@@ -88,7 +99,7 @@ pub fn Group(comptime ReturnType: type) type {
8899 await node.data;
89100 } else {
90101 (await node.data) catch |err| {
91 self.cancelAll();
102 self.deinit();
92103 return err;
93104 };
94105 }
......@@ -100,25 +111,12 @@ pub fn Group(comptime ReturnType: type) type {
100111 await handle;
101112 } else {
102113 (await handle) catch |err| {
103 self.cancelAll();
114 self.deinit();
104115 return err;
105116 };
106117 }
107118 }
108119 }
109
110 /// Cancel all the outstanding promises. May only be called if wait was never called.
111 /// TODO These should be `cancelasync` not `cancel`.
112 /// See https://github.com/ziglang/zig/issues/1261
113 pub fn cancelAll(self: *Self) void {
114 while (self.coro_stack.pop()) |node| {
115 cancel node.data;
116 }
117 while (self.alloc_stack.pop()) |node| {
118 cancel node.data;
119 self.lock.loop.allocator.destroy(node);
120 }
121 }
122120 };
123121}
124122
std/event/lock.zig+1
......@@ -9,6 +9,7 @@ const Loop = std.event.Loop;
99/// Thread-safe async/await lock.
1010/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
1111/// are resumed when the lock is released, in order.
12/// Allows only one actor to hold the lock.
1213pub const Lock = struct {
1314 loop: *Loop,
1415 shared_bit: u8, // TODO make this a bool
std/event/rwlock.zig created+292
......@@ -0,0 +1,292 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const mem = std.mem;
5const AtomicRmwOp = builtin.AtomicRmwOp;
6const AtomicOrder = builtin.AtomicOrder;
7const Loop = std.event.Loop;
8
9/// Thread-safe async/await lock.
10/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
11/// are resumed when the lock is released, in order.
12/// Many readers can hold the lock at the same time; however locking for writing is exclusive.
13pub const RwLock = struct {
14 loop: *Loop,
15 shared_state: u8, // TODO make this an enum
16 writer_queue: Queue,
17 reader_queue: Queue,
18 writer_queue_empty_bit: u8, // TODO make this a bool
19 reader_queue_empty_bit: u8, // TODO make this a bool
20 reader_lock_count: usize,
21
22 const State = struct {
23 const Unlocked = 0;
24 const WriteLock = 1;
25 const ReadLock = 2;
26 };
27
28 const Queue = std.atomic.Queue(promise);
29
30 pub const HeldRead = struct {
31 lock: *RwLock,
32
33 pub fn release(self: HeldRead) void {
34 // If other readers still hold the lock, we're done.
35 if (@atomicRmw(usize, &self.lock.reader_lock_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst) != 1) {
36 return;
37 }
38
39 _ = @atomicRmw(u8, &self.lock.reader_queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
40 if (@cmpxchgStrong(u8, &self.lock.shared_state, State.ReadLock, State.Unlocked, AtomicOrder.SeqCst, AtomicOrder.SeqCst) != null) {
41 // Didn't unlock. Someone else's problem.
42 return;
43 }
44
45 self.lock.commonPostUnlock();
46 }
47 };
48
49 pub const HeldWrite = struct {
50 lock: *RwLock,
51
52 pub fn release(self: HeldWrite) void {
53 // See if we can leave it locked for writing, and pass the lock to the next writer
54 // in the queue to grab the lock.
55 if (self.lock.writer_queue.get()) |node| {
56 self.lock.loop.onNextTick(node);
57 return;
58 }
59
60 // We need to release the write lock. Check if any readers are waiting to grab the lock.
61 if (@atomicLoad(u8, &self.lock.reader_queue_empty_bit, AtomicOrder.SeqCst) == 0) {
62 // Switch to a read lock.
63 _ = @atomicRmw(u8, &self.lock.shared_state, AtomicRmwOp.Xchg, State.ReadLock, AtomicOrder.SeqCst);
64 while (self.lock.reader_queue.get()) |node| {
65 self.lock.loop.onNextTick(node);
66 }
67 return;
68 }
69
70 _ = @atomicRmw(u8, &self.lock.writer_queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
71 _ = @atomicRmw(u8, &self.lock.shared_state, AtomicRmwOp.Xchg, State.Unlocked, AtomicOrder.SeqCst);
72
73 self.lock.commonPostUnlock();
74 }
75 };
76
77 pub fn init(loop: *Loop) RwLock {
78 return RwLock{
79 .loop = loop,
80 .shared_state = State.Unlocked,
81 .writer_queue = Queue.init(),
82 .writer_queue_empty_bit = 1,
83 .reader_queue = Queue.init(),
84 .reader_queue_empty_bit = 1,
85 .reader_lock_count = 0,
86 };
87 }
88
89 /// Must be called when not locked. Not thread safe.
90 /// All calls to acquire() and release() must complete before calling deinit().
91 pub fn deinit(self: *RwLock) void {
92 assert(self.shared_state == State.Unlocked);
93 while (self.writer_queue.get()) |node| cancel node.data;
94 while (self.reader_queue.get()) |node| cancel node.data;
95 }
96
97 pub async fn acquireRead(self: *RwLock) HeldRead {
98 _ = @atomicRmw(usize, &self.reader_lock_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
99
100 suspend |handle| {
101 // TODO explicitly put this memory in the coroutine frame #1194
102 var my_tick_node = Loop.NextTickNode{
103 .data = handle,
104 .next = undefined,
105 };
106
107 self.reader_queue.put(&my_tick_node);
108
109 // At this point, we are in the reader_queue, so we might have already been resumed and this coroutine
110 // frame might be destroyed. For the rest of the suspend block we cannot access the coroutine frame.
111
112 // We set this bit so that later we can rely on the fact, that if reader_queue_empty_bit is 1,
113 // some actor will attempt to grab the lock.
114 _ = @atomicRmw(u8, &self.reader_queue_empty_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
115
116 // Here we don't care if we are the one to do the locking or if it was already locked for reading.
117 const have_read_lock = if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.ReadLock, AtomicOrder.SeqCst, AtomicOrder.SeqCst)) |old_state| old_state == State.ReadLock else true;
118 if (have_read_lock) {
119 // Give out all the read locks.
120 if (self.reader_queue.get()) |first_node| {
121 while (self.reader_queue.get()) |node| {
122 self.loop.onNextTick(node);
123 }
124 resume first_node.data;
125 }
126 }
127 }
128 return HeldRead{ .lock = self };
129 }
130
131 pub async fn acquireWrite(self: *RwLock) HeldWrite {
132 suspend |handle| {
133 // TODO explicitly put this memory in the coroutine frame #1194
134 var my_tick_node = Loop.NextTickNode{
135 .data = handle,
136 .next = undefined,
137 };
138
139 self.writer_queue.put(&my_tick_node);
140
141 // At this point, we are in the writer_queue, so we might have already been resumed and this coroutine
142 // frame might be destroyed. For the rest of the suspend block we cannot access the coroutine frame.
143
144 // We set this bit so that later we can rely on the fact, that if writer_queue_empty_bit is 1,
145 // some actor will attempt to grab the lock.
146 _ = @atomicRmw(u8, &self.writer_queue_empty_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
147
148 // Here we must be the one to acquire the write lock. It cannot already be locked.
149 if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.WriteLock, AtomicOrder.SeqCst, AtomicOrder.SeqCst) == null) {
150 // We now have a write lock.
151 if (self.writer_queue.get()) |node| {
152 // Whether this node is us or someone else, we tail resume it.
153 resume node.data;
154 }
155 }
156 }
157 return HeldWrite{ .lock = self };
158 }
159
160 fn commonPostUnlock(self: *RwLock) void {
161 while (true) {
162 // There might be a writer_queue item or a reader_queue item
163 // If we check and both are empty, we can be done, because the other actors will try to
164 // obtain the lock.
165 // But if there's a writer_queue item or a reader_queue item,
166 // we are the actor which must loop and attempt to grab the lock again.
167 if (@atomicLoad(u8, &self.writer_queue_empty_bit, AtomicOrder.SeqCst) == 0) {
168 if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.WriteLock, AtomicOrder.SeqCst, AtomicOrder.SeqCst) != null) {
169 // We did not obtain the lock. Great, the queues are someone else's problem.
170 return;
171 }
172 // If there's an item in the writer queue, give them the lock, and we're done.
173 if (self.writer_queue.get()) |node| {
174 self.loop.onNextTick(node);
175 return;
176 }
177 // Release the lock again.
178 _ = @atomicRmw(u8, &self.writer_queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
179 _ = @atomicRmw(u8, &self.shared_state, AtomicRmwOp.Xchg, State.Unlocked, AtomicOrder.SeqCst);
180 continue;
181 }
182
183 if (@atomicLoad(u8, &self.reader_queue_empty_bit, AtomicOrder.SeqCst) == 0) {
184 if (@cmpxchgStrong(u8, &self.shared_state, State.Unlocked, State.ReadLock, AtomicOrder.SeqCst, AtomicOrder.SeqCst) != null) {
185 // We did not obtain the lock. Great, the queues are someone else's problem.
186 return;
187 }
188 // If there are any items in the reader queue, give out all the reader locks, and we're done.
189 if (self.reader_queue.get()) |first_node| {
190 self.loop.onNextTick(first_node);
191 while (self.reader_queue.get()) |node| {
192 self.loop.onNextTick(node);
193 }
194 return;
195 }
196 // Release the lock again.
197 _ = @atomicRmw(u8, &self.reader_queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
198 if (@cmpxchgStrong(u8, &self.shared_state, State.ReadLock, State.Unlocked, AtomicOrder.SeqCst, AtomicOrder.SeqCst) != null) {
199 // Didn't unlock. Someone else's problem.
200 return;
201 }
202 continue;
203 }
204 return;
205 }
206 }
207};
208
209test "std.event.RwLock" {
210 var da = std.heap.DirectAllocator.init();
211 defer da.deinit();
212
213 const allocator = &da.allocator;
214
215 var loop: Loop = undefined;
216 try loop.initMultiThreaded(allocator);
217 defer loop.deinit();
218
219 var lock = RwLock.init(&loop);
220 defer lock.deinit();
221
222 const handle = try async<allocator> testLock(&loop, &lock);
223 defer cancel handle;
224 loop.run();
225
226 const expected_result = [1]i32{shared_it_count * @intCast(i32, shared_test_data.len)} ** shared_test_data.len;
227 assert(mem.eql(i32, shared_test_data, expected_result));
228}
229
230async fn testLock(loop: *Loop, lock: *RwLock) void {
231 // TODO explicitly put next tick node memory in the coroutine frame #1194
232 suspend |p| {
233 resume p;
234 }
235
236 var read_nodes: [100]Loop.NextTickNode = undefined;
237 for (read_nodes) |*read_node| {
238 read_node.data = async readRunner(lock) catch @panic("out of memory");
239 loop.onNextTick(read_node);
240 }
241
242 var write_nodes: [shared_it_count]Loop.NextTickNode = undefined;
243 for (write_nodes) |*write_node| {
244 write_node.data = async writeRunner(lock) catch @panic("out of memory");
245 loop.onNextTick(write_node);
246 }
247
248 for (write_nodes) |*write_node| {
249 await @ptrCast(promise->void, write_node.data);
250 }
251 for (read_nodes) |*read_node| {
252 await @ptrCast(promise->void, read_node.data);
253 }
254}
255
256const shared_it_count = 10;
257var shared_test_data = [1]i32{0} ** 10;
258var shared_test_index: usize = 0;
259var shared_count: usize = 0;
260
261async fn writeRunner(lock: *RwLock) void {
262 suspend; // resumed by onNextTick
263
264 var i: usize = 0;
265 while (i < shared_test_data.len) : (i += 1) {
266 std.os.time.sleep(0, 100000);
267 const lock_promise = async lock.acquireWrite() catch @panic("out of memory");
268 const handle = await lock_promise;
269 defer handle.release();
270
271 shared_count += 1;
272 while (shared_test_index < shared_test_data.len) : (shared_test_index += 1) {
273 shared_test_data[shared_test_index] = shared_test_data[shared_test_index] + 1;
274 }
275 shared_test_index = 0;
276 }
277}
278
279async fn readRunner(lock: *RwLock) void {
280 suspend; // resumed by onNextTick
281 std.os.time.sleep(0, 1);
282
283 var i: usize = 0;
284 while (i < shared_test_data.len) : (i += 1) {
285 const lock_promise = async lock.acquireRead() catch @panic("out of memory");
286 const handle = await lock_promise;
287 defer handle.release();
288
289 assert(shared_test_index == 0);
290 assert(shared_test_data[i] == @intCast(i32, shared_count));
291 }
292}
std/event/rwlocked.zig created+58
......@@ -0,0 +1,58 @@
1const std = @import("../index.zig");
2const RwLock = std.event.RwLock;
3const Loop = std.event.Loop;
4
5/// Thread-safe async/await RW lock that protects one piece of data.
6/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
7/// are resumed when the lock is released, in order.
8pub fn RwLocked(comptime T: type) type {
9 return struct {
10 lock: RwLock,
11 locked_data: T,
12
13 const Self = this;
14
15 pub const HeldReadLock = struct {
16 value: *const T,
17 held: RwLock.HeldRead,
18
19 pub fn release(self: HeldReadLock) void {
20 self.held.release();
21 }
22 };
23
24 pub const HeldWriteLock = struct {
25 value: *T,
26 held: RwLock.HeldWrite,
27
28 pub fn release(self: HeldWriteLock) void {
29 self.held.release();
30 }
31 };
32
33 pub fn init(loop: *Loop, data: T) Self {
34 return Self{
35 .lock = RwLock.init(loop),
36 .locked_data = data,
37 };
38 }
39
40 pub fn deinit(self: *Self) void {
41 self.lock.deinit();
42 }
43
44 pub async fn acquireRead(self: *Self) HeldReadLock {
45 return HeldReadLock{
46 .held = await (async self.lock.acquireRead() catch unreachable),
47 .value = &self.locked_data,
48 };
49 }
50
51 pub async fn acquireWrite(self: *Self) HeldWriteLock {
52 return HeldWriteLock{
53 .held = await (async self.lock.acquireWrite() catch unreachable),
54 .value = &self.locked_data,
55 };
56 }
57 };
58}
std/event/tcp.zig+2-2
......@@ -61,7 +61,7 @@ pub const Server = struct {
6161
6262 /// Stop listening
6363 pub fn close(self: *Server) void {
64 self.loop.removeFd(self.sockfd.?);
64 self.loop.linuxRemoveFd(self.sockfd.?);
6565 std.os.close(self.sockfd.?);
6666 }
6767
......@@ -116,7 +116,7 @@ pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File
116116 errdefer std.os.close(sockfd);
117117
118118 try std.os.posixConnectAsync(sockfd, &address.os_addr);
119 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT);
119 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
120120 try std.os.posixGetSockOptConnectError(sockfd);
121121
122122 return std.os.File.openHandle(sockfd);
std/hash_map.zig+10
......@@ -163,6 +163,16 @@ pub fn HashMap(comptime K: type, comptime V: type, comptime hash: fn (key: K) u3
163163 };
164164 }
165165
166 pub fn clone(self: Self) !Self {
167 var other = Self.init(self.allocator);
168 try other.initCapacity(self.entries.len);
169 var it = self.iterator();
170 while (it.next()) |entry| {
171 try other.put(entry.key, entry.value);
172 }
173 return other;
174 }
175
166176 fn initCapacity(hm: *Self, capacity: usize) !void {
167177 hm.entries = try hm.allocator.alloc(Entry, capacity);
168178 hm.size = 0;
std/index.zig+2
......@@ -9,6 +9,7 @@ pub const LinkedList = @import("linked_list.zig").LinkedList;
99pub const IntrusiveLinkedList = @import("linked_list.zig").IntrusiveLinkedList;
1010pub const SegmentedList = @import("segmented_list.zig").SegmentedList;
1111pub const DynLib = @import("dynamic_library.zig").DynLib;
12pub const Mutex = @import("mutex.zig").Mutex;
1213
1314pub const atomic = @import("atomic/index.zig");
1415pub const base64 = @import("base64.zig");
......@@ -48,6 +49,7 @@ test "std" {
4849 _ = @import("hash_map.zig");
4950 _ = @import("linked_list.zig");
5051 _ = @import("segmented_list.zig");
52 _ = @import("mutex.zig");
5153
5254 _ = @import("base64.zig");
5355 _ = @import("build.zig");
std/mutex.zig created+27
......@@ -0,0 +1,27 @@
1const std = @import("index.zig");
2const builtin = @import("builtin");
3const AtomicOrder = builtin.AtomicOrder;
4const AtomicRmwOp = builtin.AtomicRmwOp;
5const assert = std.debug.assert;
6
7/// TODO use syscalls instead of a spinlock
8pub const Mutex = struct {
9 lock: u8, // TODO use a bool
10
11 pub const Held = struct {
12 mutex: *Mutex,
13
14 pub fn release(self: Held) void {
15 assert(@atomicRmw(u8, &self.mutex.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
16 }
17 };
18
19 pub fn init() Mutex {
20 return Mutex{ .lock = 0 };
21 }
22
23 pub fn acquire(self: *Mutex) Held {
24 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
25 return Held{ .mutex = self };
26 }
27};