authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-09 22:22:44-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-09 22:22:44-04:00
logb6eb404831e44a92b4841459068f4fbe9c753541
treec2c391448f8ed155377df7d45987618fc3cd9cae
parentccef60a64033a25dbe2351c27f28257546b2ae5b

organize std.event into directories

7 files changed, 1277 insertions(+), 1222 deletions(-)

CMakeLists.txt+5
...@@ -458,6 +458,11 @@ set(ZIG_STD_FILES...@@ -458,6 +458,11 @@ set(ZIG_STD_FILES
458 "elf.zig"458 "elf.zig"
459 "empty.zig"459 "empty.zig"
460 "event.zig"460 "event.zig"
461 "event/channel.zig"
462 "event/lock.zig"
463 "event/locked.zig"
464 "event/loop.zig"
465 "event/tcp.zig"
461 "fmt/errol/enum3.zig"466 "fmt/errol/enum3.zig"
462 "fmt/errol/index.zig"467 "fmt/errol/index.zig"
463 "fmt/errol/lookup.zig"468 "fmt/errol/lookup.zig"
std/event.zig+12-1222
...@@ -1,1223 +1,13 @@...@@ -1,1223 +1,13 @@
1const std = @import("index.zig");1pub const Locked = @import("event/locked.zig").Locked;
2const builtin = @import("builtin");2pub const Loop = @import("event/loop.zig").Loop;
3const assert = std.debug.assert;3pub const Lock = @import("event/lock.zig").Lock;
4const event = this;4pub const tcp = @import("event/tcp.zig");
5const mem = std.mem;5pub const Channel = @import("event/channel.zig").Channel;
6const posix = std.os.posix;6
7const windows = std.os.windows;7test "import event tests" {
8const AtomicRmwOp = builtin.AtomicRmwOp;8 _ = @import("event/locked.zig");
9const AtomicOrder = builtin.AtomicOrder;9 _ = @import("event/loop.zig");
1010 _ = @import("event/lock.zig");
11pub const TcpServer = struct {11 _ = @import("event/tcp.zig");
12 handleRequestFn: async<*mem.Allocator> fn (*TcpServer, *const std.net.Address, *const std.os.File) void,12 _ = @import("event/channel.zig");
13
14 loop: *Loop,
15 sockfd: ?i32,
16 accept_coro: ?promise,
17 listen_address: std.net.Address,
18
19 waiting_for_emfile_node: PromiseNode,
20 listen_resume_node: event.Loop.ResumeNode,
21
22 const PromiseNode = std.LinkedList(promise).Node;
23
24 pub fn init(loop: *Loop) TcpServer {
25 // TODO can't initialize handler coroutine here because we need well defined copy elision
26 return TcpServer{
27 .loop = loop,
28 .sockfd = null,
29 .accept_coro = null,
30 .handleRequestFn = undefined,
31 .waiting_for_emfile_node = undefined,
32 .listen_address = undefined,
33 .listen_resume_node = event.Loop.ResumeNode{
34 .id = event.Loop.ResumeNode.Id.Basic,
35 .handle = undefined,
36 },
37 };
38 }
39
40 pub fn listen(
41 self: *TcpServer,
42 address: *const std.net.Address,
43 handleRequestFn: async<*mem.Allocator> fn (*TcpServer, *const std.net.Address, *const std.os.File) void,
44 ) !void {
45 self.handleRequestFn = handleRequestFn;
46
47 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
48 errdefer std.os.close(sockfd);
49 self.sockfd = sockfd;
50
51 try std.os.posixBind(sockfd, &address.os_addr);
52 try std.os.posixListen(sockfd, posix.SOMAXCONN);
53 self.listen_address = std.net.Address.initPosix(try std.os.posixGetSockName(sockfd));
54
55 self.accept_coro = try async<self.loop.allocator> TcpServer.handler(self);
56 errdefer cancel self.accept_coro.?;
57
58 self.listen_resume_node.handle = self.accept_coro.?;
59 try self.loop.addFd(sockfd, &self.listen_resume_node);
60 errdefer self.loop.removeFd(sockfd);
61 }
62
63 /// Stop listening
64 pub fn close(self: *TcpServer) void {
65 self.loop.removeFd(self.sockfd.?);
66 std.os.close(self.sockfd.?);
67 }
68
69 pub fn deinit(self: *TcpServer) void {
70 if (self.accept_coro) |accept_coro| cancel accept_coro;
71 if (self.sockfd) |sockfd| std.os.close(sockfd);
72 }
73
74 pub async fn handler(self: *TcpServer) void {
75 while (true) {
76 var accepted_addr: std.net.Address = undefined;
77 if (std.os.posixAccept(self.sockfd.?, &accepted_addr.os_addr, posix.SOCK_NONBLOCK | posix.SOCK_CLOEXEC)) |accepted_fd| {
78 var socket = std.os.File.openHandle(accepted_fd);
79 _ = async<self.loop.allocator> self.handleRequestFn(self, accepted_addr, socket) catch |err| switch (err) {
80 error.OutOfMemory => {
81 socket.close();
82 continue;
83 },
84 };
85 } else |err| switch (err) {
86 error.WouldBlock => {
87 suspend; // we will get resumed by epoll_wait in the event loop
88 continue;
89 },
90 error.ProcessFdQuotaExceeded => {
91 errdefer std.os.emfile_promise_queue.remove(&self.waiting_for_emfile_node);
92 suspend |p| {
93 self.waiting_for_emfile_node = PromiseNode.init(p);
94 std.os.emfile_promise_queue.append(&self.waiting_for_emfile_node);
95 }
96 continue;
97 },
98 error.ConnectionAborted, error.FileDescriptorClosed => continue,
99
100 error.PageFault => unreachable,
101 error.InvalidSyscall => unreachable,
102 error.FileDescriptorNotASocket => unreachable,
103 error.OperationNotSupported => unreachable,
104
105 error.SystemFdQuotaExceeded, error.SystemResources, error.ProtocolFailure, error.BlockedByFirewall, error.Unexpected => {
106 @panic("TODO handle this error");
107 },
108 }
109 }
110 }
111};
112
113pub const Loop = struct {
114 allocator: *mem.Allocator,
115 next_tick_queue: std.atomic.QueueMpsc(promise),
116 os_data: OsData,
117 final_resume_node: ResumeNode,
118 dispatch_lock: u8, // TODO make this a bool
119 pending_event_count: usize,
120 extra_threads: []*std.os.Thread,
121
122 // pre-allocated eventfds. all permanently active.
123 // this is how we send promises to be resumed on other threads.
124 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
125 eventfd_resume_nodes: []std.atomic.Stack(ResumeNode.EventFd).Node,
126
127 pub const NextTickNode = std.atomic.QueueMpsc(promise).Node;
128
129 pub const ResumeNode = struct {
130 id: Id,
131 handle: promise,
132
133 pub const Id = enum {
134 Basic,
135 Stop,
136 EventFd,
137 };
138
139 pub const EventFd = switch (builtin.os) {
140 builtin.Os.macosx => MacOsEventFd,
141 builtin.Os.linux => struct {
142 base: ResumeNode,
143 epoll_op: u32,
144 eventfd: i32,
145 },
146 builtin.Os.windows => struct {
147 base: ResumeNode,
148 completion_key: usize,
149 },
150 else => @compileError("unsupported OS"),
151 };
152
153 const MacOsEventFd = struct {
154 base: ResumeNode,
155 kevent: posix.Kevent,
156 };
157 };
158
159 /// After initialization, call run().
160 /// TODO copy elision / named return values so that the threads referencing *Loop
161 /// have the correct pointer value.
162 fn initSingleThreaded(self: *Loop, allocator: *mem.Allocator) !void {
163 return self.initInternal(allocator, 1);
164 }
165
166 /// The allocator must be thread-safe because we use it for multiplexing
167 /// coroutines onto kernel threads.
168 /// After initialization, call run().
169 /// TODO copy elision / named return values so that the threads referencing *Loop
170 /// have the correct pointer value.
171 fn initMultiThreaded(self: *Loop, allocator: *mem.Allocator) !void {
172 const core_count = try std.os.cpuCount(allocator);
173 return self.initInternal(allocator, core_count);
174 }
175
176 /// Thread count is the total thread count. The thread pool size will be
177 /// max(thread_count - 1, 0)
178 fn initInternal(self: *Loop, allocator: *mem.Allocator, thread_count: usize) !void {
179 self.* = Loop{
180 .pending_event_count = 0,
181 .allocator = allocator,
182 .os_data = undefined,
183 .next_tick_queue = std.atomic.QueueMpsc(promise).init(),
184 .dispatch_lock = 1, // start locked so threads go directly into epoll wait
185 .extra_threads = undefined,
186 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
187 .eventfd_resume_nodes = undefined,
188 .final_resume_node = ResumeNode{
189 .id = ResumeNode.Id.Stop,
190 .handle = undefined,
191 },
192 };
193 const extra_thread_count = thread_count - 1;
194 self.eventfd_resume_nodes = try self.allocator.alloc(
195 std.atomic.Stack(ResumeNode.EventFd).Node,
196 extra_thread_count,
197 );
198 errdefer self.allocator.free(self.eventfd_resume_nodes);
199
200 self.extra_threads = try self.allocator.alloc(*std.os.Thread, extra_thread_count);
201 errdefer self.allocator.free(self.extra_threads);
202
203 try self.initOsData(extra_thread_count);
204 errdefer self.deinitOsData();
205 }
206
207 /// must call stop before deinit
208 pub fn deinit(self: *Loop) void {
209 self.deinitOsData();
210 self.allocator.free(self.extra_threads);
211 }
212
213 const InitOsDataError = std.os.LinuxEpollCreateError || mem.Allocator.Error || std.os.LinuxEventFdError ||
214 std.os.SpawnThreadError || std.os.LinuxEpollCtlError || std.os.BsdKEventError ||
215 std.os.WindowsCreateIoCompletionPortError;
216
217 const wakeup_bytes = []u8{0x1} ** 8;
218
219 fn initOsData(self: *Loop, extra_thread_count: usize) InitOsDataError!void {
220 switch (builtin.os) {
221 builtin.Os.linux => {
222 errdefer {
223 while (self.available_eventfd_resume_nodes.pop()) |node| std.os.close(node.data.eventfd);
224 }
225 for (self.eventfd_resume_nodes) |*eventfd_node| {
226 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
227 .data = ResumeNode.EventFd{
228 .base = ResumeNode{
229 .id = ResumeNode.Id.EventFd,
230 .handle = undefined,
231 },
232 .eventfd = try std.os.linuxEventFd(1, posix.EFD_CLOEXEC | posix.EFD_NONBLOCK),
233 .epoll_op = posix.EPOLL_CTL_ADD,
234 },
235 .next = undefined,
236 };
237 self.available_eventfd_resume_nodes.push(eventfd_node);
238 }
239
240 self.os_data.epollfd = try std.os.linuxEpollCreate(posix.EPOLL_CLOEXEC);
241 errdefer std.os.close(self.os_data.epollfd);
242
243 self.os_data.final_eventfd = try std.os.linuxEventFd(0, posix.EFD_CLOEXEC | posix.EFD_NONBLOCK);
244 errdefer std.os.close(self.os_data.final_eventfd);
245
246 self.os_data.final_eventfd_event = posix.epoll_event{
247 .events = posix.EPOLLIN,
248 .data = posix.epoll_data{ .ptr = @ptrToInt(&self.final_resume_node) },
249 };
250 try std.os.linuxEpollCtl(
251 self.os_data.epollfd,
252 posix.EPOLL_CTL_ADD,
253 self.os_data.final_eventfd,
254 &self.os_data.final_eventfd_event,
255 );
256
257 var extra_thread_index: usize = 0;
258 errdefer {
259 // writing 8 bytes to an eventfd cannot fail
260 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
261 while (extra_thread_index != 0) {
262 extra_thread_index -= 1;
263 self.extra_threads[extra_thread_index].wait();
264 }
265 }
266 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
267 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
268 }
269 },
270 builtin.Os.macosx => {
271 self.os_data.kqfd = try std.os.bsdKQueue();
272 errdefer std.os.close(self.os_data.kqfd);
273
274 self.os_data.kevents = try self.allocator.alloc(posix.Kevent, extra_thread_count);
275 errdefer self.allocator.free(self.os_data.kevents);
276
277 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
278
279 for (self.eventfd_resume_nodes) |*eventfd_node, i| {
280 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
281 .data = ResumeNode.EventFd{
282 .base = ResumeNode{
283 .id = ResumeNode.Id.EventFd,
284 .handle = undefined,
285 },
286 // this one is for sending events
287 .kevent = posix.Kevent{
288 .ident = i,
289 .filter = posix.EVFILT_USER,
290 .flags = posix.EV_CLEAR | posix.EV_ADD | posix.EV_DISABLE,
291 .fflags = 0,
292 .data = 0,
293 .udata = @ptrToInt(&eventfd_node.data.base),
294 },
295 },
296 .next = undefined,
297 };
298 self.available_eventfd_resume_nodes.push(eventfd_node);
299 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.data.kevent);
300 _ = try std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null);
301 eventfd_node.data.kevent.flags = posix.EV_CLEAR | posix.EV_ENABLE;
302 eventfd_node.data.kevent.fflags = posix.NOTE_TRIGGER;
303 // this one is for waiting for events
304 self.os_data.kevents[i] = posix.Kevent{
305 .ident = i,
306 .filter = posix.EVFILT_USER,
307 .flags = 0,
308 .fflags = 0,
309 .data = 0,
310 .udata = @ptrToInt(&eventfd_node.data.base),
311 };
312 }
313
314 // Pre-add so that we cannot get error.SystemResources
315 // later when we try to activate it.
316 self.os_data.final_kevent = posix.Kevent{
317 .ident = extra_thread_count,
318 .filter = posix.EVFILT_USER,
319 .flags = posix.EV_ADD | posix.EV_DISABLE,
320 .fflags = 0,
321 .data = 0,
322 .udata = @ptrToInt(&self.final_resume_node),
323 };
324 const kevent_array = (*[1]posix.Kevent)(&self.os_data.final_kevent);
325 _ = try std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null);
326 self.os_data.final_kevent.flags = posix.EV_ENABLE;
327 self.os_data.final_kevent.fflags = posix.NOTE_TRIGGER;
328
329 var extra_thread_index: usize = 0;
330 errdefer {
331 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch unreachable;
332 while (extra_thread_index != 0) {
333 extra_thread_index -= 1;
334 self.extra_threads[extra_thread_index].wait();
335 }
336 }
337 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
338 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
339 }
340 },
341 builtin.Os.windows => {
342 self.os_data.extra_thread_count = extra_thread_count;
343
344 self.os_data.io_port = try std.os.windowsCreateIoCompletionPort(
345 windows.INVALID_HANDLE_VALUE,
346 null,
347 undefined,
348 undefined,
349 );
350 errdefer std.os.close(self.os_data.io_port);
351
352 for (self.eventfd_resume_nodes) |*eventfd_node, i| {
353 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
354 .data = ResumeNode.EventFd{
355 .base = ResumeNode{
356 .id = ResumeNode.Id.EventFd,
357 .handle = undefined,
358 },
359 // this one is for sending events
360 .completion_key = @ptrToInt(&eventfd_node.data.base),
361 },
362 .next = undefined,
363 };
364 self.available_eventfd_resume_nodes.push(eventfd_node);
365 }
366
367 var extra_thread_index: usize = 0;
368 errdefer {
369 var i: usize = 0;
370 while (i < extra_thread_index) : (i += 1) {
371 while (true) {
372 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
373 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
374 break;
375 }
376 }
377 while (extra_thread_index != 0) {
378 extra_thread_index -= 1;
379 self.extra_threads[extra_thread_index].wait();
380 }
381 }
382 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
383 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
384 }
385 },
386 else => {},
387 }
388 }
389
390 fn deinitOsData(self: *Loop) void {
391 switch (builtin.os) {
392 builtin.Os.linux => {
393 std.os.close(self.os_data.final_eventfd);
394 while (self.available_eventfd_resume_nodes.pop()) |node| std.os.close(node.data.eventfd);
395 std.os.close(self.os_data.epollfd);
396 self.allocator.free(self.eventfd_resume_nodes);
397 },
398 builtin.Os.macosx => {
399 self.allocator.free(self.os_data.kevents);
400 std.os.close(self.os_data.kqfd);
401 },
402 builtin.Os.windows => {
403 std.os.close(self.os_data.io_port);
404 },
405 else => {},
406 }
407 }
408
409 /// resume_node must live longer than the promise that it holds a reference to.
410 pub fn addFd(self: *Loop, fd: i32, resume_node: *ResumeNode) !void {
411 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
412 errdefer {
413 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
414 }
415 try self.modFd(
416 fd,
417 posix.EPOLL_CTL_ADD,
418 std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET,
419 resume_node,
420 );
421 }
422
423 pub fn modFd(self: *Loop, fd: i32, op: u32, events: u32, resume_node: *ResumeNode) !void {
424 var ev = std.os.linux.epoll_event{
425 .events = events,
426 .data = std.os.linux.epoll_data{ .ptr = @ptrToInt(resume_node) },
427 };
428 try std.os.linuxEpollCtl(self.os_data.epollfd, op, fd, &ev);
429 }
430
431 pub fn removeFd(self: *Loop, fd: i32) void {
432 self.removeFdNoCounter(fd);
433 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
434 }
435
436 fn removeFdNoCounter(self: *Loop, fd: i32) void {
437 std.os.linuxEpollCtl(self.os_data.epollfd, std.os.linux.EPOLL_CTL_DEL, fd, undefined) catch {};
438 }
439
440 pub async fn waitFd(self: *Loop, fd: i32) !void {
441 defer self.removeFd(fd);
442 suspend |p| {
443 // TODO explicitly put this memory in the coroutine frame #1194
444 var resume_node = ResumeNode{
445 .id = ResumeNode.Id.Basic,
446 .handle = p,
447 };
448 try self.addFd(fd, &resume_node);
449 }
450 }
451
452 /// Bring your own linked list node. This means it can't fail.
453 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
454 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
455 self.next_tick_queue.put(node);
456 }
457
458 pub fn run(self: *Loop) void {
459 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
460 self.workerRun();
461 for (self.extra_threads) |extra_thread| {
462 extra_thread.wait();
463 }
464 }
465
466 fn workerRun(self: *Loop) void {
467 start_over: while (true) {
468 if (@atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) == 0) {
469 while (self.next_tick_queue.get()) |next_tick_node| {
470 const handle = next_tick_node.data;
471 if (self.next_tick_queue.isEmpty()) {
472 // last node, just resume it
473 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
474 resume handle;
475 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
476 continue :start_over;
477 }
478
479 // non-last node, stick it in the epoll/kqueue set so that
480 // other threads can get to it
481 if (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {
482 const eventfd_node = &resume_stack_node.data;
483 eventfd_node.base.handle = handle;
484 switch (builtin.os) {
485 builtin.Os.macosx => {
486 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.kevent);
487 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
488 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch {
489 // fine, we didn't need it anyway
490 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
491 self.available_eventfd_resume_nodes.push(resume_stack_node);
492 resume handle;
493 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
494 continue :start_over;
495 };
496 },
497 builtin.Os.linux => {
498 // the pending count is already accounted for
499 const epoll_events = posix.EPOLLONESHOT | std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET;
500 self.modFd(eventfd_node.eventfd, eventfd_node.epoll_op, epoll_events, &eventfd_node.base) catch {
501 // fine, we didn't need it anyway
502 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
503 self.available_eventfd_resume_nodes.push(resume_stack_node);
504 resume handle;
505 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
506 continue :start_over;
507 };
508 },
509 builtin.Os.windows => {
510 // this value is never dereferenced but we need it to be non-null so that
511 // the consumer code can decide whether to read the completion key.
512 // it has to do this for normal I/O, so we match that behavior here.
513 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
514 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, eventfd_node.completion_key, overlapped) catch {
515 // fine, we didn't need it anyway
516 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
517 self.available_eventfd_resume_nodes.push(resume_stack_node);
518 resume handle;
519 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
520 continue :start_over;
521 };
522 },
523 else => @compileError("unsupported OS"),
524 }
525 } else {
526 // threads are too busy, can't add another eventfd to wake one up
527 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
528 resume handle;
529 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
530 continue :start_over;
531 }
532 }
533
534 const pending_event_count = @atomicLoad(usize, &self.pending_event_count, AtomicOrder.SeqCst);
535 if (pending_event_count == 0) {
536 // cause all the threads to stop
537 switch (builtin.os) {
538 builtin.Os.linux => {
539 // writing 8 bytes to an eventfd cannot fail
540 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
541 return;
542 },
543 builtin.Os.macosx => {
544 const final_kevent = (*[1]posix.Kevent)(&self.os_data.final_kevent);
545 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
546 // cannot fail because we already added it and this just enables it
547 _ = std.os.bsdKEvent(self.os_data.kqfd, final_kevent, eventlist, null) catch unreachable;
548 return;
549 },
550 builtin.Os.windows => {
551 var i: usize = 0;
552 while (i < self.os_data.extra_thread_count) : (i += 1) {
553 while (true) {
554 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
555 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
556 break;
557 }
558 }
559 return;
560 },
561 else => @compileError("unsupported OS"),
562 }
563 }
564
565 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
566 }
567
568 switch (builtin.os) {
569 builtin.Os.linux => {
570 // only process 1 event so we don't steal from other threads
571 var events: [1]std.os.linux.epoll_event = undefined;
572 const count = std.os.linuxEpollWait(self.os_data.epollfd, events[0..], -1);
573 for (events[0..count]) |ev| {
574 const resume_node = @intToPtr(*ResumeNode, ev.data.ptr);
575 const handle = resume_node.handle;
576 const resume_node_id = resume_node.id;
577 switch (resume_node_id) {
578 ResumeNode.Id.Basic => {},
579 ResumeNode.Id.Stop => return,
580 ResumeNode.Id.EventFd => {
581 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
582 event_fd_node.epoll_op = posix.EPOLL_CTL_MOD;
583 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
584 self.available_eventfd_resume_nodes.push(stack_node);
585 },
586 }
587 resume handle;
588 if (resume_node_id == ResumeNode.Id.EventFd) {
589 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
590 }
591 }
592 },
593 builtin.Os.macosx => {
594 var eventlist: [1]posix.Kevent = undefined;
595 const count = std.os.bsdKEvent(self.os_data.kqfd, self.os_data.kevents, eventlist[0..], null) catch unreachable;
596 for (eventlist[0..count]) |ev| {
597 const resume_node = @intToPtr(*ResumeNode, ev.udata);
598 const handle = resume_node.handle;
599 const resume_node_id = resume_node.id;
600 switch (resume_node_id) {
601 ResumeNode.Id.Basic => {},
602 ResumeNode.Id.Stop => return,
603 ResumeNode.Id.EventFd => {
604 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
605 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
606 self.available_eventfd_resume_nodes.push(stack_node);
607 },
608 }
609 resume handle;
610 if (resume_node_id == ResumeNode.Id.EventFd) {
611 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
612 }
613 }
614 },
615 builtin.Os.windows => {
616 var completion_key: usize = undefined;
617 while (true) {
618 var nbytes: windows.DWORD = undefined;
619 var overlapped: ?*windows.OVERLAPPED = undefined;
620 switch (std.os.windowsGetQueuedCompletionStatus(self.os_data.io_port, &nbytes, &completion_key, &overlapped, windows.INFINITE)) {
621 std.os.WindowsWaitResult.Aborted => return,
622 std.os.WindowsWaitResult.Normal => {},
623 }
624 if (overlapped != null) break;
625 }
626 const resume_node = @intToPtr(*ResumeNode, completion_key);
627 const handle = resume_node.handle;
628 const resume_node_id = resume_node.id;
629 switch (resume_node_id) {
630 ResumeNode.Id.Basic => {},
631 ResumeNode.Id.Stop => return,
632 ResumeNode.Id.EventFd => {
633 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
634 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
635 self.available_eventfd_resume_nodes.push(stack_node);
636 },
637 }
638 resume handle;
639 if (resume_node_id == ResumeNode.Id.EventFd) {
640 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
641 }
642 },
643 else => @compileError("unsupported OS"),
644 }
645 }
646 }
647
648 const OsData = switch (builtin.os) {
649 builtin.Os.linux => struct {
650 epollfd: i32,
651 final_eventfd: i32,
652 final_eventfd_event: std.os.linux.epoll_event,
653 },
654 builtin.Os.macosx => MacOsData,
655 builtin.Os.windows => struct {
656 io_port: windows.HANDLE,
657 extra_thread_count: usize,
658 },
659 else => struct {},
660 };
661
662 const MacOsData = struct {
663 kqfd: i32,
664 final_kevent: posix.Kevent,
665 kevents: []posix.Kevent,
666 };
667};
668
669/// many producer, many consumer, thread-safe, lock-free, runtime configurable buffer size
670/// when buffer is empty, consumers suspend and are resumed by producers
671/// when buffer is full, producers suspend and are resumed by consumers
672pub fn Channel(comptime T: type) type {
673 return struct {
674 loop: *Loop,
675
676 getters: std.atomic.QueueMpsc(GetNode),
677 putters: std.atomic.QueueMpsc(PutNode),
678 get_count: usize,
679 put_count: usize,
680 dispatch_lock: u8, // TODO make this a bool
681 need_dispatch: u8, // TODO make this a bool
682
683 // simple fixed size ring buffer
684 buffer_nodes: []T,
685 buffer_index: usize,
686 buffer_len: usize,
687
688 const SelfChannel = this;
689 const GetNode = struct {
690 ptr: *T,
691 tick_node: *Loop.NextTickNode,
692 };
693 const PutNode = struct {
694 data: T,
695 tick_node: *Loop.NextTickNode,
696 };
697
698 /// call destroy when done
699 pub fn create(loop: *Loop, capacity: usize) !*SelfChannel {
700 const buffer_nodes = try loop.allocator.alloc(T, capacity);
701 errdefer loop.allocator.free(buffer_nodes);
702
703 const self = try loop.allocator.create(SelfChannel{
704 .loop = loop,
705 .buffer_len = 0,
706 .buffer_nodes = buffer_nodes,
707 .buffer_index = 0,
708 .dispatch_lock = 0,
709 .need_dispatch = 0,
710 .getters = std.atomic.QueueMpsc(GetNode).init(),
711 .putters = std.atomic.QueueMpsc(PutNode).init(),
712 .get_count = 0,
713 .put_count = 0,
714 });
715 errdefer loop.allocator.destroy(self);
716
717 return self;
718 }
719
720 /// must be called when all calls to put and get have suspended and no more calls occur
721 pub fn destroy(self: *SelfChannel) void {
722 while (self.getters.get()) |get_node| {
723 cancel get_node.data.tick_node.data;
724 }
725 while (self.putters.get()) |put_node| {
726 cancel put_node.data.tick_node.data;
727 }
728 self.loop.allocator.free(self.buffer_nodes);
729 self.loop.allocator.destroy(self);
730 }
731
732 /// puts a data item in the channel. The promise completes when the value has been added to the
733 /// buffer, or in the case of a zero size buffer, when the item has been retrieved by a getter.
734 pub async fn put(self: *SelfChannel, data: T) void {
735 // TODO should be able to group memory allocation failure before first suspend point
736 // so that the async invocation catches it
737 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
738 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
739
740 suspend |handle| {
741 var my_tick_node = Loop.NextTickNode{
742 .next = undefined,
743 .data = handle,
744 };
745 var queue_node = std.atomic.QueueMpsc(PutNode).Node{
746 .data = PutNode{
747 .tick_node = &my_tick_node,
748 .data = data,
749 },
750 .next = undefined,
751 };
752 self.putters.put(&queue_node);
753 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
754
755 self.loop.onNextTick(dispatch_tick_node_ptr);
756 }
757 }
758
759 /// await this function to get an item from the channel. If the buffer is empty, the promise will
760 /// complete when the next item is put in the channel.
761 pub async fn get(self: *SelfChannel) T {
762 // TODO should be able to group memory allocation failure before first suspend point
763 // so that the async invocation catches it
764 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
765 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
766
767 // TODO integrate this function with named return values
768 // so we can get rid of this extra result copy
769 var result: T = undefined;
770 suspend |handle| {
771 var my_tick_node = Loop.NextTickNode{
772 .next = undefined,
773 .data = handle,
774 };
775 var queue_node = std.atomic.QueueMpsc(GetNode).Node{
776 .data = GetNode{
777 .ptr = &result,
778 .tick_node = &my_tick_node,
779 },
780 .next = undefined,
781 };
782 self.getters.put(&queue_node);
783 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
784
785 self.loop.onNextTick(dispatch_tick_node_ptr);
786 }
787 return result;
788 }
789
790 async fn dispatch(self: *SelfChannel, tick_node_ptr: **Loop.NextTickNode) void {
791 // resumed by onNextTick
792 suspend |handle| {
793 var tick_node = Loop.NextTickNode{
794 .data = handle,
795 .next = undefined,
796 };
797 tick_node_ptr.* = &tick_node;
798 }
799
800 // set the "need dispatch" flag
801 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
802
803 lock: while (true) {
804 // set the lock flag
805 const prev_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
806 if (prev_lock != 0) return;
807
808 // clear the need_dispatch flag since we're about to do it
809 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
810
811 while (true) {
812 one_dispatch: {
813 // later we correct these extra subtractions
814 var get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
815 var put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
816
817 // transfer self.buffer to self.getters
818 while (self.buffer_len != 0) {
819 if (get_count == 0) break :one_dispatch;
820
821 const get_node = &self.getters.get().?.data;
822 get_node.ptr.* = self.buffer_nodes[self.buffer_index -% self.buffer_len];
823 self.loop.onNextTick(get_node.tick_node);
824 self.buffer_len -= 1;
825
826 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
827 }
828
829 // direct transfer self.putters to self.getters
830 while (get_count != 0 and put_count != 0) {
831 const get_node = &self.getters.get().?.data;
832 const put_node = &self.putters.get().?.data;
833
834 get_node.ptr.* = put_node.data;
835 self.loop.onNextTick(get_node.tick_node);
836 self.loop.onNextTick(put_node.tick_node);
837
838 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
839 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
840 }
841
842 // transfer self.putters to self.buffer
843 while (self.buffer_len != self.buffer_nodes.len and put_count != 0) {
844 const put_node = &self.putters.get().?.data;
845
846 self.buffer_nodes[self.buffer_index] = put_node.data;
847 self.loop.onNextTick(put_node.tick_node);
848 self.buffer_index +%= 1;
849 self.buffer_len += 1;
850
851 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
852 }
853 }
854
855 // undo the extra subtractions
856 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
857 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
858
859 // clear need-dispatch flag
860 const need_dispatch = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
861 if (need_dispatch != 0) continue;
862
863 const my_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
864 assert(my_lock != 0);
865
866 // we have to check again now that we unlocked
867 if (@atomicLoad(u8, &self.need_dispatch, AtomicOrder.SeqCst) != 0) continue :lock;
868
869 return;
870 }
871 }
872 }
873 };
874}
875
876pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {
877 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733
878
879 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
880 errdefer std.os.close(sockfd);
881
882 try std.os.posixConnectAsync(sockfd, &address.os_addr);
883 try await try async loop.waitFd(sockfd);
884 try std.os.posixGetSockOptConnectError(sockfd);
885
886 return std.os.File.openHandle(sockfd);
887}
888
889test "listen on a port, send bytes, receive bytes" {
890 if (builtin.os != builtin.Os.linux) {
891 // TODO build abstractions for other operating systems
892 return;
893 }
894 const MyServer = struct {
895 tcp_server: TcpServer,
896
897 const Self = this;
898 async<*mem.Allocator> fn handler(tcp_server: *TcpServer, _addr: *const std.net.Address, _socket: *const std.os.File) void {
899 const self = @fieldParentPtr(Self, "tcp_server", tcp_server);
900 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
901 defer socket.close();
902 // TODO guarantee elision of this allocation
903 const next_handler = async errorableHandler(self, _addr, socket) catch unreachable;
904 (await next_handler) catch |err| {
905 std.debug.panic("unable to handle connection: {}\n", err);
906 };
907 suspend |p| {
908 cancel p;
909 }
910 }
911 async fn errorableHandler(self: *Self, _addr: *const std.net.Address, _socket: *const std.os.File) !void {
912 const addr = _addr.*; // TODO https://github.com/ziglang/zig/issues/733
913 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
914
915 var adapter = std.io.FileOutStream.init(&socket);
916 var stream = &adapter.stream;
917 try stream.print("hello from server\n");
918 }
919 };
920
921 const ip4addr = std.net.parseIp4("127.0.0.1") catch unreachable;
922 const addr = std.net.Address.initIp4(ip4addr, 0);
923
924 var loop: Loop = undefined;
925 try loop.initSingleThreaded(std.debug.global_allocator);
926 var server = MyServer{ .tcp_server = TcpServer.init(&loop) };
927 defer server.tcp_server.deinit();
928 try server.tcp_server.listen(addr, MyServer.handler);
929
930 const p = try async<std.debug.global_allocator> doAsyncTest(&loop, server.tcp_server.listen_address, &server.tcp_server);
931 defer cancel p;
932 loop.run();
933}
934
935async fn doAsyncTest(loop: *Loop, address: *const std.net.Address, server: *TcpServer) void {
936 errdefer @panic("test failure");
937
938 var socket_file = try await try async event.connect(loop, address);
939 defer socket_file.close();
940
941 var buf: [512]u8 = undefined;
942 const amt_read = try socket_file.read(buf[0..]);
943 const msg = buf[0..amt_read];
944 assert(mem.eql(u8, msg, "hello from server\n"));
945 server.close();
946}
947
948test "std.event.Channel" {
949 var da = std.heap.DirectAllocator.init();
950 defer da.deinit();
951
952 const allocator = &da.allocator;
953
954 var loop: Loop = undefined;
955 // TODO make a multi threaded test
956 try loop.initSingleThreaded(allocator);
957 defer loop.deinit();
958
959 const channel = try Channel(i32).create(&loop, 0);
960 defer channel.destroy();
961
962 const handle = try async<allocator> testChannelGetter(&loop, channel);
963 defer cancel handle;
964
965 const putter = try async<allocator> testChannelPutter(channel);
966 defer cancel putter;
967
968 loop.run();
969}
970
971async fn testChannelGetter(loop: *Loop, channel: *Channel(i32)) void {
972 errdefer @panic("test failed");
973
974 const value1_promise = try async channel.get();
975 const value1 = await value1_promise;
976 assert(value1 == 1234);
977
978 const value2_promise = try async channel.get();
979 const value2 = await value2_promise;
980 assert(value2 == 4567);
981}
982
983async fn testChannelPutter(channel: *Channel(i32)) void {
984 await (async channel.put(1234) catch @panic("out of memory"));
985 await (async channel.put(4567) catch @panic("out of memory"));
986}
987
988/// Thread-safe async/await lock.
989/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
990/// are resumed when the lock is released, in order.
991pub const Lock = struct {
992 loop: *Loop,
993 shared_bit: u8, // TODO make this a bool
994 queue: Queue,
995 queue_empty_bit: u8, // TODO make this a bool
996
997 const Queue = std.atomic.QueueMpsc(promise);
998
999 pub const Held = struct {
1000 lock: *Lock,
1001
1002 pub fn release(self: Held) void {
1003 // Resume the next item from the queue.
1004 if (self.lock.queue.get()) |node| {
1005 self.lock.loop.onNextTick(node);
1006 return;
1007 }
1008
1009 // We need to release the lock.
1010 _ = @atomicRmw(u8, &self.lock.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
1011 _ = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
1012
1013 // There might be a queue item. If we know the queue is empty, we can be done,
1014 // because the other actor will try to obtain the lock.
1015 // But if there's a queue item, we are the actor which must loop and attempt
1016 // to grab the lock again.
1017 if (@atomicLoad(u8, &self.lock.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
1018 return;
1019 }
1020
1021 while (true) {
1022 const old_bit = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
1023 if (old_bit != 0) {
1024 // We did not obtain the lock. Great, the queue is someone else's problem.
1025 return;
1026 }
1027
1028 // Resume the next item from the queue.
1029 if (self.lock.queue.get()) |node| {
1030 self.lock.loop.onNextTick(node);
1031 return;
1032 }
1033
1034 // Release the lock again.
1035 _ = @atomicRmw(u8, &self.lock.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
1036 _ = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
1037
1038 // Find out if we can be done.
1039 if (@atomicLoad(u8, &self.lock.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
1040 return;
1041 }
1042 }
1043 }
1044 };
1045
1046 pub fn init(loop: *Loop) Lock {
1047 return Lock{
1048 .loop = loop,
1049 .shared_bit = 0,
1050 .queue = Queue.init(),
1051 .queue_empty_bit = 1,
1052 };
1053 }
1054
1055 /// Must be called when not locked. Not thread safe.
1056 /// All calls to acquire() and release() must complete before calling deinit().
1057 pub fn deinit(self: *Lock) void {
1058 assert(self.shared_bit == 0);
1059 while (self.queue.get()) |node| cancel node.data;
1060 }
1061
1062 pub async fn acquire(self: *Lock) Held {
1063 s: suspend |handle| {
1064 // TODO explicitly put this memory in the coroutine frame #1194
1065 var my_tick_node = Loop.NextTickNode{
1066 .data = handle,
1067 .next = undefined,
1068 };
1069
1070 self.queue.put(&my_tick_node);
1071
1072 // At this point, we are in the queue, so we might have already been resumed and this coroutine
1073 // frame might be destroyed. For the rest of the suspend block we cannot access the coroutine frame.
1074
1075 // We set this bit so that later we can rely on the fact, that if queue_empty_bit is 1, some actor
1076 // will attempt to grab the lock.
1077 _ = @atomicRmw(u8, &self.queue_empty_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
1078
1079 while (true) {
1080 const old_bit = @atomicRmw(u8, &self.shared_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
1081 if (old_bit != 0) {
1082 // We did not obtain the lock. Trust that our queue entry will resume us, and allow
1083 // suspend to complete.
1084 break;
1085 }
1086 // We got the lock. However we might have already been resumed from the queue.
1087 if (self.queue.get()) |node| {
1088 // Whether this node is us or someone else, we tail resume it.
1089 resume node.data;
1090 break;
1091 } else {
1092 // We already got resumed, and there are none left in the queue, which means that
1093 // we aren't even supposed to hold the lock right now.
1094 _ = @atomicRmw(u8, &self.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
1095 _ = @atomicRmw(u8, &self.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
1096
1097 // There might be a queue item. If we know the queue is empty, we can be done,
1098 // because the other actor will try to obtain the lock.
1099 // But if there's a queue item, we are the actor which must loop and attempt
1100 // to grab the lock again.
1101 if (@atomicLoad(u8, &self.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
1102 break;
1103 } else {
1104 continue;
1105 }
1106 }
1107 unreachable;
1108 }
1109 }
1110
1111 return Held{ .lock = self };
1112 }
1113};
1114
1115/// Thread-safe async/await lock that protects one piece of data.
1116/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
1117/// are resumed when the lock is released, in order.
1118pub fn Locked(comptime T: type) type {
1119 return struct {
1120 lock: Lock,
1121 private_data: T,
1122
1123 const Self = this;
1124
1125 pub const HeldLock = struct {
1126 value: *T,
1127 held: Lock.Held,
1128
1129 pub fn release(self: HeldLock) void {
1130 self.held.release();
1131 }
1132 };
1133
1134 pub fn init(loop: *Loop, data: T) Self {
1135 return Self{
1136 .lock = Lock.init(loop),
1137 .private_data = data,
1138 };
1139 }
1140
1141 pub fn deinit(self: *Self) void {
1142 self.lock.deinit();
1143 }
1144
1145 pub async fn acquire(self: *Self) HeldLock {
1146 return HeldLock{
1147 // TODO guaranteed allocation elision
1148 .held = await (async self.lock.acquire() catch unreachable),
1149 .value = &self.private_data,
1150 };
1151 }
1152 };
1153}
1154
1155test "std.event.Lock" {
1156 var da = std.heap.DirectAllocator.init();
1157 defer da.deinit();
1158
1159 const allocator = &da.allocator;
1160
1161 var loop: Loop = undefined;
1162 try loop.initMultiThreaded(allocator);
1163 defer loop.deinit();
1164
1165 var lock = Lock.init(&loop);
1166 defer lock.deinit();
1167
1168 const handle = try async<allocator> testLock(&loop, &lock);
1169 defer cancel handle;
1170 loop.run();
1171
1172 assert(mem.eql(i32, shared_test_data, [1]i32{3 * @intCast(i32, shared_test_data.len)} ** shared_test_data.len));
1173}
1174
1175async fn testLock(loop: *Loop, lock: *Lock) void {
1176 // TODO explicitly put next tick node memory in the coroutine frame #1194
1177 suspend |p| {
1178 resume p;
1179 }
1180 const handle1 = async lockRunner(lock) catch @panic("out of memory");
1181 var tick_node1 = Loop.NextTickNode{
1182 .next = undefined,
1183 .data = handle1,
1184 };
1185 loop.onNextTick(&tick_node1);
1186
1187 const handle2 = async lockRunner(lock) catch @panic("out of memory");
1188 var tick_node2 = Loop.NextTickNode{
1189 .next = undefined,
1190 .data = handle2,
1191 };
1192 loop.onNextTick(&tick_node2);
1193
1194 const handle3 = async lockRunner(lock) catch @panic("out of memory");
1195 var tick_node3 = Loop.NextTickNode{
1196 .next = undefined,
1197 .data = handle3,
1198 };
1199 loop.onNextTick(&tick_node3);
1200
1201 await handle1;
1202 await handle2;
1203 await handle3;
1204}
1205
1206var shared_test_data = [1]i32{0} ** 10;
1207var shared_test_index: usize = 0;
1208
1209async fn lockRunner(lock: *Lock) void {
1210 suspend; // resumed by onNextTick
1211
1212 var i: usize = 0;
1213 while (i < shared_test_data.len) : (i += 1) {
1214 const lock_promise = async lock.acquire() catch @panic("out of memory");
1215 const handle = await lock_promise;
1216 defer handle.release();
1217
1218 shared_test_index = 0;
1219 while (shared_test_index < shared_test_data.len) : (shared_test_index += 1) {
1220 shared_test_data[shared_test_index] = shared_test_data[shared_test_index] + 1;
1221 }
1222 }
1223}13}
std/event/channel.zig created+254
...@@ -0,0 +1,254 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const AtomicRmwOp = builtin.AtomicRmwOp;
5const AtomicOrder = builtin.AtomicOrder;
6const Loop = std.event.Loop;
7
8/// many producer, many consumer, thread-safe, lock-free, runtime configurable buffer size
9/// when buffer is empty, consumers suspend and are resumed by producers
10/// when buffer is full, producers suspend and are resumed by consumers
11pub fn Channel(comptime T: type) type {
12 return struct {
13 loop: *Loop,
14
15 getters: std.atomic.QueueMpsc(GetNode),
16 putters: std.atomic.QueueMpsc(PutNode),
17 get_count: usize,
18 put_count: usize,
19 dispatch_lock: u8, // TODO make this a bool
20 need_dispatch: u8, // TODO make this a bool
21
22 // simple fixed size ring buffer
23 buffer_nodes: []T,
24 buffer_index: usize,
25 buffer_len: usize,
26
27 const SelfChannel = this;
28 const GetNode = struct {
29 ptr: *T,
30 tick_node: *Loop.NextTickNode,
31 };
32 const PutNode = struct {
33 data: T,
34 tick_node: *Loop.NextTickNode,
35 };
36
37 /// call destroy when done
38 pub fn create(loop: *Loop, capacity: usize) !*SelfChannel {
39 const buffer_nodes = try loop.allocator.alloc(T, capacity);
40 errdefer loop.allocator.free(buffer_nodes);
41
42 const self = try loop.allocator.create(SelfChannel{
43 .loop = loop,
44 .buffer_len = 0,
45 .buffer_nodes = buffer_nodes,
46 .buffer_index = 0,
47 .dispatch_lock = 0,
48 .need_dispatch = 0,
49 .getters = std.atomic.QueueMpsc(GetNode).init(),
50 .putters = std.atomic.QueueMpsc(PutNode).init(),
51 .get_count = 0,
52 .put_count = 0,
53 });
54 errdefer loop.allocator.destroy(self);
55
56 return self;
57 }
58
59 /// must be called when all calls to put and get have suspended and no more calls occur
60 pub fn destroy(self: *SelfChannel) void {
61 while (self.getters.get()) |get_node| {
62 cancel get_node.data.tick_node.data;
63 }
64 while (self.putters.get()) |put_node| {
65 cancel put_node.data.tick_node.data;
66 }
67 self.loop.allocator.free(self.buffer_nodes);
68 self.loop.allocator.destroy(self);
69 }
70
71 /// puts a data item in the channel. The promise completes when the value has been added to the
72 /// buffer, or in the case of a zero size buffer, when the item has been retrieved by a getter.
73 pub async fn put(self: *SelfChannel, data: T) void {
74 // TODO should be able to group memory allocation failure before first suspend point
75 // so that the async invocation catches it
76 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
77 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
78
79 suspend |handle| {
80 var my_tick_node = Loop.NextTickNode{
81 .next = undefined,
82 .data = handle,
83 };
84 var queue_node = std.atomic.QueueMpsc(PutNode).Node{
85 .data = PutNode{
86 .tick_node = &my_tick_node,
87 .data = data,
88 },
89 .next = undefined,
90 };
91 self.putters.put(&queue_node);
92 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
93
94 self.loop.onNextTick(dispatch_tick_node_ptr);
95 }
96 }
97
98 /// await this function to get an item from the channel. If the buffer is empty, the promise will
99 /// complete when the next item is put in the channel.
100 pub async fn get(self: *SelfChannel) T {
101 // TODO should be able to group memory allocation failure before first suspend point
102 // so that the async invocation catches it
103 var dispatch_tick_node_ptr: *Loop.NextTickNode = undefined;
104 _ = async self.dispatch(&dispatch_tick_node_ptr) catch unreachable;
105
106 // TODO integrate this function with named return values
107 // so we can get rid of this extra result copy
108 var result: T = undefined;
109 suspend |handle| {
110 var my_tick_node = Loop.NextTickNode{
111 .next = undefined,
112 .data = handle,
113 };
114 var queue_node = std.atomic.QueueMpsc(GetNode).Node{
115 .data = GetNode{
116 .ptr = &result,
117 .tick_node = &my_tick_node,
118 },
119 .next = undefined,
120 };
121 self.getters.put(&queue_node);
122 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
123
124 self.loop.onNextTick(dispatch_tick_node_ptr);
125 }
126 return result;
127 }
128
129 async fn dispatch(self: *SelfChannel, tick_node_ptr: **Loop.NextTickNode) void {
130 // resumed by onNextTick
131 suspend |handle| {
132 var tick_node = Loop.NextTickNode{
133 .data = handle,
134 .next = undefined,
135 };
136 tick_node_ptr.* = &tick_node;
137 }
138
139 // set the "need dispatch" flag
140 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
141
142 lock: while (true) {
143 // set the lock flag
144 const prev_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
145 if (prev_lock != 0) return;
146
147 // clear the need_dispatch flag since we're about to do it
148 _ = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
149
150 while (true) {
151 one_dispatch: {
152 // later we correct these extra subtractions
153 var get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
154 var put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
155
156 // transfer self.buffer to self.getters
157 while (self.buffer_len != 0) {
158 if (get_count == 0) break :one_dispatch;
159
160 const get_node = &self.getters.get().?.data;
161 get_node.ptr.* = self.buffer_nodes[self.buffer_index -% self.buffer_len];
162 self.loop.onNextTick(get_node.tick_node);
163 self.buffer_len -= 1;
164
165 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
166 }
167
168 // direct transfer self.putters to self.getters
169 while (get_count != 0 and put_count != 0) {
170 const get_node = &self.getters.get().?.data;
171 const put_node = &self.putters.get().?.data;
172
173 get_node.ptr.* = put_node.data;
174 self.loop.onNextTick(get_node.tick_node);
175 self.loop.onNextTick(put_node.tick_node);
176
177 get_count = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
178 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
179 }
180
181 // transfer self.putters to self.buffer
182 while (self.buffer_len != self.buffer_nodes.len and put_count != 0) {
183 const put_node = &self.putters.get().?.data;
184
185 self.buffer_nodes[self.buffer_index] = put_node.data;
186 self.loop.onNextTick(put_node.tick_node);
187 self.buffer_index +%= 1;
188 self.buffer_len += 1;
189
190 put_count = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
191 }
192 }
193
194 // undo the extra subtractions
195 _ = @atomicRmw(usize, &self.get_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
196 _ = @atomicRmw(usize, &self.put_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
197
198 // clear need-dispatch flag
199 const need_dispatch = @atomicRmw(u8, &self.need_dispatch, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
200 if (need_dispatch != 0) continue;
201
202 const my_lock = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
203 assert(my_lock != 0);
204
205 // we have to check again now that we unlocked
206 if (@atomicLoad(u8, &self.need_dispatch, AtomicOrder.SeqCst) != 0) continue :lock;
207
208 return;
209 }
210 }
211 }
212 };
213}
214
215test "std.event.Channel" {
216 var da = std.heap.DirectAllocator.init();
217 defer da.deinit();
218
219 const allocator = &da.allocator;
220
221 var loop: Loop = undefined;
222 // TODO make a multi threaded test
223 try loop.initSingleThreaded(allocator);
224 defer loop.deinit();
225
226 const channel = try Channel(i32).create(&loop, 0);
227 defer channel.destroy();
228
229 const handle = try async<allocator> testChannelGetter(&loop, channel);
230 defer cancel handle;
231
232 const putter = try async<allocator> testChannelPutter(channel);
233 defer cancel putter;
234
235 loop.run();
236}
237
238async fn testChannelGetter(loop: *Loop, channel: *Channel(i32)) void {
239 errdefer @panic("test failed");
240
241 const value1_promise = try async channel.get();
242 const value1 = await value1_promise;
243 assert(value1 == 1234);
244
245 const value2_promise = try async channel.get();
246 const value2 = await value2_promise;
247 assert(value2 == 4567);
248}
249
250async fn testChannelPutter(channel: *Channel(i32)) void {
251 await (async channel.put(1234) catch @panic("out of memory"));
252 await (async channel.put(4567) catch @panic("out of memory"));
253}
254
std/event/lock.zig created+204
...@@ -0,0 +1,204 @@
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.
12pub const Lock = struct {
13 loop: *Loop,
14 shared_bit: u8, // TODO make this a bool
15 queue: Queue,
16 queue_empty_bit: u8, // TODO make this a bool
17
18 const Queue = std.atomic.QueueMpsc(promise);
19
20 pub const Held = struct {
21 lock: *Lock,
22
23 pub fn release(self: Held) void {
24 // Resume the next item from the queue.
25 if (self.lock.queue.get()) |node| {
26 self.lock.loop.onNextTick(node);
27 return;
28 }
29
30 // We need to release the lock.
31 _ = @atomicRmw(u8, &self.lock.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
32 _ = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
33
34 // There might be a queue item. If we know the queue is empty, we can be done,
35 // because the other actor will try to obtain the lock.
36 // But if there's a queue item, we are the actor which must loop and attempt
37 // to grab the lock again.
38 if (@atomicLoad(u8, &self.lock.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
39 return;
40 }
41
42 while (true) {
43 const old_bit = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
44 if (old_bit != 0) {
45 // We did not obtain the lock. Great, the queue is someone else's problem.
46 return;
47 }
48
49 // Resume the next item from the queue.
50 if (self.lock.queue.get()) |node| {
51 self.lock.loop.onNextTick(node);
52 return;
53 }
54
55 // Release the lock again.
56 _ = @atomicRmw(u8, &self.lock.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
57 _ = @atomicRmw(u8, &self.lock.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
58
59 // Find out if we can be done.
60 if (@atomicLoad(u8, &self.lock.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
61 return;
62 }
63 }
64 }
65 };
66
67 pub fn init(loop: *Loop) Lock {
68 return Lock{
69 .loop = loop,
70 .shared_bit = 0,
71 .queue = Queue.init(),
72 .queue_empty_bit = 1,
73 };
74 }
75
76 /// Must be called when not locked. Not thread safe.
77 /// All calls to acquire() and release() must complete before calling deinit().
78 pub fn deinit(self: *Lock) void {
79 assert(self.shared_bit == 0);
80 while (self.queue.get()) |node| cancel node.data;
81 }
82
83 pub async fn acquire(self: *Lock) Held {
84 s: suspend |handle| {
85 // TODO explicitly put this memory in the coroutine frame #1194
86 var my_tick_node = Loop.NextTickNode{
87 .data = handle,
88 .next = undefined,
89 };
90
91 self.queue.put(&my_tick_node);
92
93 // At this point, we are in the queue, so we might have already been resumed and this coroutine
94 // frame might be destroyed. For the rest of the suspend block we cannot access the coroutine frame.
95
96 // We set this bit so that later we can rely on the fact, that if queue_empty_bit is 1, some actor
97 // will attempt to grab the lock.
98 _ = @atomicRmw(u8, &self.queue_empty_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
99
100 while (true) {
101 const old_bit = @atomicRmw(u8, &self.shared_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
102 if (old_bit != 0) {
103 // We did not obtain the lock. Trust that our queue entry will resume us, and allow
104 // suspend to complete.
105 break;
106 }
107 // We got the lock. However we might have already been resumed from the queue.
108 if (self.queue.get()) |node| {
109 // Whether this node is us or someone else, we tail resume it.
110 resume node.data;
111 break;
112 } else {
113 // We already got resumed, and there are none left in the queue, which means that
114 // we aren't even supposed to hold the lock right now.
115 _ = @atomicRmw(u8, &self.queue_empty_bit, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
116 _ = @atomicRmw(u8, &self.shared_bit, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
117
118 // There might be a queue item. If we know the queue is empty, we can be done,
119 // because the other actor will try to obtain the lock.
120 // But if there's a queue item, we are the actor which must loop and attempt
121 // to grab the lock again.
122 if (@atomicLoad(u8, &self.queue_empty_bit, AtomicOrder.SeqCst) == 1) {
123 break;
124 } else {
125 continue;
126 }
127 }
128 unreachable;
129 }
130 }
131
132 return Held{ .lock = self };
133 }
134};
135
136test "std.event.Lock" {
137 var da = std.heap.DirectAllocator.init();
138 defer da.deinit();
139
140 const allocator = &da.allocator;
141
142 var loop: Loop = undefined;
143 try loop.initMultiThreaded(allocator);
144 defer loop.deinit();
145
146 var lock = Lock.init(&loop);
147 defer lock.deinit();
148
149 const handle = try async<allocator> testLock(&loop, &lock);
150 defer cancel handle;
151 loop.run();
152
153 assert(mem.eql(i32, shared_test_data, [1]i32{3 * @intCast(i32, shared_test_data.len)} ** shared_test_data.len));
154}
155
156async fn testLock(loop: *Loop, lock: *Lock) void {
157 // TODO explicitly put next tick node memory in the coroutine frame #1194
158 suspend |p| {
159 resume p;
160 }
161 const handle1 = async lockRunner(lock) catch @panic("out of memory");
162 var tick_node1 = Loop.NextTickNode{
163 .next = undefined,
164 .data = handle1,
165 };
166 loop.onNextTick(&tick_node1);
167
168 const handle2 = async lockRunner(lock) catch @panic("out of memory");
169 var tick_node2 = Loop.NextTickNode{
170 .next = undefined,
171 .data = handle2,
172 };
173 loop.onNextTick(&tick_node2);
174
175 const handle3 = async lockRunner(lock) catch @panic("out of memory");
176 var tick_node3 = Loop.NextTickNode{
177 .next = undefined,
178 .data = handle3,
179 };
180 loop.onNextTick(&tick_node3);
181
182 await handle1;
183 await handle2;
184 await handle3;
185}
186
187var shared_test_data = [1]i32{0} ** 10;
188var shared_test_index: usize = 0;
189
190async fn lockRunner(lock: *Lock) void {
191 suspend; // resumed by onNextTick
192
193 var i: usize = 0;
194 while (i < shared_test_data.len) : (i += 1) {
195 const lock_promise = async lock.acquire() catch @panic("out of memory");
196 const handle = await lock_promise;
197 defer handle.release();
198
199 shared_test_index = 0;
200 while (shared_test_index < shared_test_data.len) : (shared_test_index += 1) {
201 shared_test_data[shared_test_index] = shared_test_data[shared_test_index] + 1;
202 }
203 }
204}
std/event/locked.zig created+42
...@@ -0,0 +1,42 @@
1const std = @import("../index.zig");
2const Lock = std.event.Lock;
3
4/// Thread-safe async/await lock that protects one piece of data.
5/// Does not make any syscalls - coroutines which are waiting for the lock are suspended, and
6/// are resumed when the lock is released, in order.
7pub fn Locked(comptime T: type) type {
8 return struct {
9 lock: Lock,
10 private_data: T,
11
12 const Self = this;
13
14 pub const HeldLock = struct {
15 value: *T,
16 held: Lock.Held,
17
18 pub fn release(self: HeldLock) void {
19 self.held.release();
20 }
21 };
22
23 pub fn init(loop: *Loop, data: T) Self {
24 return Self{
25 .lock = Lock.init(loop),
26 .private_data = data,
27 };
28 }
29
30 pub fn deinit(self: *Self) void {
31 self.lock.deinit();
32 }
33
34 pub async fn acquire(self: *Self) HeldLock {
35 return HeldLock{
36 // TODO guaranteed allocation elision
37 .held = await (async self.lock.acquire() catch unreachable),
38 .value = &self.private_data,
39 };
40 }
41 };
42}
std/event/loop.zig created+577
...@@ -0,0 +1,577 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const mem = std.mem;
5const posix = std.os.posix;
6const windows = std.os.windows;
7const AtomicRmwOp = builtin.AtomicRmwOp;
8const AtomicOrder = builtin.AtomicOrder;
9
10pub const Loop = struct {
11 allocator: *mem.Allocator,
12 next_tick_queue: std.atomic.QueueMpsc(promise),
13 os_data: OsData,
14 final_resume_node: ResumeNode,
15 dispatch_lock: u8, // TODO make this a bool
16 pending_event_count: usize,
17 extra_threads: []*std.os.Thread,
18
19 // pre-allocated eventfds. all permanently active.
20 // this is how we send promises to be resumed on other threads.
21 available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
22 eventfd_resume_nodes: []std.atomic.Stack(ResumeNode.EventFd).Node,
23
24 pub const NextTickNode = std.atomic.QueueMpsc(promise).Node;
25
26 pub const ResumeNode = struct {
27 id: Id,
28 handle: promise,
29
30 pub const Id = enum {
31 Basic,
32 Stop,
33 EventFd,
34 };
35
36 pub const EventFd = switch (builtin.os) {
37 builtin.Os.macosx => MacOsEventFd,
38 builtin.Os.linux => struct {
39 base: ResumeNode,
40 epoll_op: u32,
41 eventfd: i32,
42 },
43 builtin.Os.windows => struct {
44 base: ResumeNode,
45 completion_key: usize,
46 },
47 else => @compileError("unsupported OS"),
48 };
49
50 const MacOsEventFd = struct {
51 base: ResumeNode,
52 kevent: posix.Kevent,
53 };
54 };
55
56 /// After initialization, call run().
57 /// TODO copy elision / named return values so that the threads referencing *Loop
58 /// have the correct pointer value.
59 fn initSingleThreaded(self: *Loop, allocator: *mem.Allocator) !void {
60 return self.initInternal(allocator, 1);
61 }
62
63 /// The allocator must be thread-safe because we use it for multiplexing
64 /// coroutines onto kernel threads.
65 /// After initialization, call run().
66 /// TODO copy elision / named return values so that the threads referencing *Loop
67 /// have the correct pointer value.
68 fn initMultiThreaded(self: *Loop, allocator: *mem.Allocator) !void {
69 const core_count = try std.os.cpuCount(allocator);
70 return self.initInternal(allocator, core_count);
71 }
72
73 /// Thread count is the total thread count. The thread pool size will be
74 /// max(thread_count - 1, 0)
75 fn initInternal(self: *Loop, allocator: *mem.Allocator, thread_count: usize) !void {
76 self.* = Loop{
77 .pending_event_count = 0,
78 .allocator = allocator,
79 .os_data = undefined,
80 .next_tick_queue = std.atomic.QueueMpsc(promise).init(),
81 .dispatch_lock = 1, // start locked so threads go directly into epoll wait
82 .extra_threads = undefined,
83 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
84 .eventfd_resume_nodes = undefined,
85 .final_resume_node = ResumeNode{
86 .id = ResumeNode.Id.Stop,
87 .handle = undefined,
88 },
89 };
90 const extra_thread_count = thread_count - 1;
91 self.eventfd_resume_nodes = try self.allocator.alloc(
92 std.atomic.Stack(ResumeNode.EventFd).Node,
93 extra_thread_count,
94 );
95 errdefer self.allocator.free(self.eventfd_resume_nodes);
96
97 self.extra_threads = try self.allocator.alloc(*std.os.Thread, extra_thread_count);
98 errdefer self.allocator.free(self.extra_threads);
99
100 try self.initOsData(extra_thread_count);
101 errdefer self.deinitOsData();
102 }
103
104 /// must call stop before deinit
105 pub fn deinit(self: *Loop) void {
106 self.deinitOsData();
107 self.allocator.free(self.extra_threads);
108 }
109
110 const InitOsDataError = std.os.LinuxEpollCreateError || mem.Allocator.Error || std.os.LinuxEventFdError ||
111 std.os.SpawnThreadError || std.os.LinuxEpollCtlError || std.os.BsdKEventError ||
112 std.os.WindowsCreateIoCompletionPortError;
113
114 const wakeup_bytes = []u8{0x1} ** 8;
115
116 fn initOsData(self: *Loop, extra_thread_count: usize) InitOsDataError!void {
117 switch (builtin.os) {
118 builtin.Os.linux => {
119 errdefer {
120 while (self.available_eventfd_resume_nodes.pop()) |node| std.os.close(node.data.eventfd);
121 }
122 for (self.eventfd_resume_nodes) |*eventfd_node| {
123 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
124 .data = ResumeNode.EventFd{
125 .base = ResumeNode{
126 .id = ResumeNode.Id.EventFd,
127 .handle = undefined,
128 },
129 .eventfd = try std.os.linuxEventFd(1, posix.EFD_CLOEXEC | posix.EFD_NONBLOCK),
130 .epoll_op = posix.EPOLL_CTL_ADD,
131 },
132 .next = undefined,
133 };
134 self.available_eventfd_resume_nodes.push(eventfd_node);
135 }
136
137 self.os_data.epollfd = try std.os.linuxEpollCreate(posix.EPOLL_CLOEXEC);
138 errdefer std.os.close(self.os_data.epollfd);
139
140 self.os_data.final_eventfd = try std.os.linuxEventFd(0, posix.EFD_CLOEXEC | posix.EFD_NONBLOCK);
141 errdefer std.os.close(self.os_data.final_eventfd);
142
143 self.os_data.final_eventfd_event = posix.epoll_event{
144 .events = posix.EPOLLIN,
145 .data = posix.epoll_data{ .ptr = @ptrToInt(&self.final_resume_node) },
146 };
147 try std.os.linuxEpollCtl(
148 self.os_data.epollfd,
149 posix.EPOLL_CTL_ADD,
150 self.os_data.final_eventfd,
151 &self.os_data.final_eventfd_event,
152 );
153
154 var extra_thread_index: usize = 0;
155 errdefer {
156 // writing 8 bytes to an eventfd cannot fail
157 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
158 while (extra_thread_index != 0) {
159 extra_thread_index -= 1;
160 self.extra_threads[extra_thread_index].wait();
161 }
162 }
163 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
164 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
165 }
166 },
167 builtin.Os.macosx => {
168 self.os_data.kqfd = try std.os.bsdKQueue();
169 errdefer std.os.close(self.os_data.kqfd);
170
171 self.os_data.kevents = try self.allocator.alloc(posix.Kevent, extra_thread_count);
172 errdefer self.allocator.free(self.os_data.kevents);
173
174 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
175
176 for (self.eventfd_resume_nodes) |*eventfd_node, i| {
177 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
178 .data = ResumeNode.EventFd{
179 .base = ResumeNode{
180 .id = ResumeNode.Id.EventFd,
181 .handle = undefined,
182 },
183 // this one is for sending events
184 .kevent = posix.Kevent{
185 .ident = i,
186 .filter = posix.EVFILT_USER,
187 .flags = posix.EV_CLEAR | posix.EV_ADD | posix.EV_DISABLE,
188 .fflags = 0,
189 .data = 0,
190 .udata = @ptrToInt(&eventfd_node.data.base),
191 },
192 },
193 .next = undefined,
194 };
195 self.available_eventfd_resume_nodes.push(eventfd_node);
196 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.data.kevent);
197 _ = try std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null);
198 eventfd_node.data.kevent.flags = posix.EV_CLEAR | posix.EV_ENABLE;
199 eventfd_node.data.kevent.fflags = posix.NOTE_TRIGGER;
200 // this one is for waiting for events
201 self.os_data.kevents[i] = posix.Kevent{
202 .ident = i,
203 .filter = posix.EVFILT_USER,
204 .flags = 0,
205 .fflags = 0,
206 .data = 0,
207 .udata = @ptrToInt(&eventfd_node.data.base),
208 };
209 }
210
211 // Pre-add so that we cannot get error.SystemResources
212 // later when we try to activate it.
213 self.os_data.final_kevent = posix.Kevent{
214 .ident = extra_thread_count,
215 .filter = posix.EVFILT_USER,
216 .flags = posix.EV_ADD | posix.EV_DISABLE,
217 .fflags = 0,
218 .data = 0,
219 .udata = @ptrToInt(&self.final_resume_node),
220 };
221 const kevent_array = (*[1]posix.Kevent)(&self.os_data.final_kevent);
222 _ = try std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null);
223 self.os_data.final_kevent.flags = posix.EV_ENABLE;
224 self.os_data.final_kevent.fflags = posix.NOTE_TRIGGER;
225
226 var extra_thread_index: usize = 0;
227 errdefer {
228 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch unreachable;
229 while (extra_thread_index != 0) {
230 extra_thread_index -= 1;
231 self.extra_threads[extra_thread_index].wait();
232 }
233 }
234 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
235 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
236 }
237 },
238 builtin.Os.windows => {
239 self.os_data.extra_thread_count = extra_thread_count;
240
241 self.os_data.io_port = try std.os.windowsCreateIoCompletionPort(
242 windows.INVALID_HANDLE_VALUE,
243 null,
244 undefined,
245 undefined,
246 );
247 errdefer std.os.close(self.os_data.io_port);
248
249 for (self.eventfd_resume_nodes) |*eventfd_node, i| {
250 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
251 .data = ResumeNode.EventFd{
252 .base = ResumeNode{
253 .id = ResumeNode.Id.EventFd,
254 .handle = undefined,
255 },
256 // this one is for sending events
257 .completion_key = @ptrToInt(&eventfd_node.data.base),
258 },
259 .next = undefined,
260 };
261 self.available_eventfd_resume_nodes.push(eventfd_node);
262 }
263
264 var extra_thread_index: usize = 0;
265 errdefer {
266 var i: usize = 0;
267 while (i < extra_thread_index) : (i += 1) {
268 while (true) {
269 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
270 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
271 break;
272 }
273 }
274 while (extra_thread_index != 0) {
275 extra_thread_index -= 1;
276 self.extra_threads[extra_thread_index].wait();
277 }
278 }
279 while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
280 self.extra_threads[extra_thread_index] = try std.os.spawnThread(self, workerRun);
281 }
282 },
283 else => {},
284 }
285 }
286
287 fn deinitOsData(self: *Loop) void {
288 switch (builtin.os) {
289 builtin.Os.linux => {
290 std.os.close(self.os_data.final_eventfd);
291 while (self.available_eventfd_resume_nodes.pop()) |node| std.os.close(node.data.eventfd);
292 std.os.close(self.os_data.epollfd);
293 self.allocator.free(self.eventfd_resume_nodes);
294 },
295 builtin.Os.macosx => {
296 self.allocator.free(self.os_data.kevents);
297 std.os.close(self.os_data.kqfd);
298 },
299 builtin.Os.windows => {
300 std.os.close(self.os_data.io_port);
301 },
302 else => {},
303 }
304 }
305
306 /// resume_node must live longer than the promise that it holds a reference to.
307 pub fn addFd(self: *Loop, fd: i32, resume_node: *ResumeNode) !void {
308 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
309 errdefer {
310 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
311 }
312 try self.modFd(
313 fd,
314 posix.EPOLL_CTL_ADD,
315 std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET,
316 resume_node,
317 );
318 }
319
320 pub fn modFd(self: *Loop, fd: i32, op: u32, events: u32, resume_node: *ResumeNode) !void {
321 var ev = std.os.linux.epoll_event{
322 .events = events,
323 .data = std.os.linux.epoll_data{ .ptr = @ptrToInt(resume_node) },
324 };
325 try std.os.linuxEpollCtl(self.os_data.epollfd, op, fd, &ev);
326 }
327
328 pub fn removeFd(self: *Loop, fd: i32) void {
329 self.removeFdNoCounter(fd);
330 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
331 }
332
333 fn removeFdNoCounter(self: *Loop, fd: i32) void {
334 std.os.linuxEpollCtl(self.os_data.epollfd, std.os.linux.EPOLL_CTL_DEL, fd, undefined) catch {};
335 }
336
337 pub async fn waitFd(self: *Loop, fd: i32) !void {
338 defer self.removeFd(fd);
339 suspend |p| {
340 // TODO explicitly put this memory in the coroutine frame #1194
341 var resume_node = ResumeNode{
342 .id = ResumeNode.Id.Basic,
343 .handle = p,
344 };
345 try self.addFd(fd, &resume_node);
346 }
347 }
348
349 /// Bring your own linked list node. This means it can't fail.
350 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
351 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
352 self.next_tick_queue.put(node);
353 }
354
355 pub fn run(self: *Loop) void {
356 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
357 self.workerRun();
358 for (self.extra_threads) |extra_thread| {
359 extra_thread.wait();
360 }
361 }
362
363 fn workerRun(self: *Loop) void {
364 start_over: while (true) {
365 if (@atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) == 0) {
366 while (self.next_tick_queue.get()) |next_tick_node| {
367 const handle = next_tick_node.data;
368 if (self.next_tick_queue.isEmpty()) {
369 // last node, just resume it
370 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
371 resume handle;
372 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
373 continue :start_over;
374 }
375
376 // non-last node, stick it in the epoll/kqueue set so that
377 // other threads can get to it
378 if (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {
379 const eventfd_node = &resume_stack_node.data;
380 eventfd_node.base.handle = handle;
381 switch (builtin.os) {
382 builtin.Os.macosx => {
383 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.kevent);
384 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
385 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch {
386 // fine, we didn't need it anyway
387 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
388 self.available_eventfd_resume_nodes.push(resume_stack_node);
389 resume handle;
390 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
391 continue :start_over;
392 };
393 },
394 builtin.Os.linux => {
395 // the pending count is already accounted for
396 const epoll_events = posix.EPOLLONESHOT | std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET;
397 self.modFd(eventfd_node.eventfd, eventfd_node.epoll_op, epoll_events, &eventfd_node.base) catch {
398 // fine, we didn't need it anyway
399 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
400 self.available_eventfd_resume_nodes.push(resume_stack_node);
401 resume handle;
402 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
403 continue :start_over;
404 };
405 },
406 builtin.Os.windows => {
407 // this value is never dereferenced but we need it to be non-null so that
408 // the consumer code can decide whether to read the completion key.
409 // it has to do this for normal I/O, so we match that behavior here.
410 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
411 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, eventfd_node.completion_key, overlapped) catch {
412 // fine, we didn't need it anyway
413 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
414 self.available_eventfd_resume_nodes.push(resume_stack_node);
415 resume handle;
416 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
417 continue :start_over;
418 };
419 },
420 else => @compileError("unsupported OS"),
421 }
422 } else {
423 // threads are too busy, can't add another eventfd to wake one up
424 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
425 resume handle;
426 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
427 continue :start_over;
428 }
429 }
430
431 const pending_event_count = @atomicLoad(usize, &self.pending_event_count, AtomicOrder.SeqCst);
432 if (pending_event_count == 0) {
433 // cause all the threads to stop
434 switch (builtin.os) {
435 builtin.Os.linux => {
436 // writing 8 bytes to an eventfd cannot fail
437 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
438 return;
439 },
440 builtin.Os.macosx => {
441 const final_kevent = (*[1]posix.Kevent)(&self.os_data.final_kevent);
442 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
443 // cannot fail because we already added it and this just enables it
444 _ = std.os.bsdKEvent(self.os_data.kqfd, final_kevent, eventlist, null) catch unreachable;
445 return;
446 },
447 builtin.Os.windows => {
448 var i: usize = 0;
449 while (i < self.os_data.extra_thread_count) : (i += 1) {
450 while (true) {
451 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
452 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
453 break;
454 }
455 }
456 return;
457 },
458 else => @compileError("unsupported OS"),
459 }
460 }
461
462 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
463 }
464
465 switch (builtin.os) {
466 builtin.Os.linux => {
467 // only process 1 event so we don't steal from other threads
468 var events: [1]std.os.linux.epoll_event = undefined;
469 const count = std.os.linuxEpollWait(self.os_data.epollfd, events[0..], -1);
470 for (events[0..count]) |ev| {
471 const resume_node = @intToPtr(*ResumeNode, ev.data.ptr);
472 const handle = resume_node.handle;
473 const resume_node_id = resume_node.id;
474 switch (resume_node_id) {
475 ResumeNode.Id.Basic => {},
476 ResumeNode.Id.Stop => return,
477 ResumeNode.Id.EventFd => {
478 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
479 event_fd_node.epoll_op = posix.EPOLL_CTL_MOD;
480 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
481 self.available_eventfd_resume_nodes.push(stack_node);
482 },
483 }
484 resume handle;
485 if (resume_node_id == ResumeNode.Id.EventFd) {
486 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
487 }
488 }
489 },
490 builtin.Os.macosx => {
491 var eventlist: [1]posix.Kevent = undefined;
492 const count = std.os.bsdKEvent(self.os_data.kqfd, self.os_data.kevents, eventlist[0..], null) catch unreachable;
493 for (eventlist[0..count]) |ev| {
494 const resume_node = @intToPtr(*ResumeNode, ev.udata);
495 const handle = resume_node.handle;
496 const resume_node_id = resume_node.id;
497 switch (resume_node_id) {
498 ResumeNode.Id.Basic => {},
499 ResumeNode.Id.Stop => return,
500 ResumeNode.Id.EventFd => {
501 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
502 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
503 self.available_eventfd_resume_nodes.push(stack_node);
504 },
505 }
506 resume handle;
507 if (resume_node_id == ResumeNode.Id.EventFd) {
508 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
509 }
510 }
511 },
512 builtin.Os.windows => {
513 var completion_key: usize = undefined;
514 while (true) {
515 var nbytes: windows.DWORD = undefined;
516 var overlapped: ?*windows.OVERLAPPED = undefined;
517 switch (std.os.windowsGetQueuedCompletionStatus(self.os_data.io_port, &nbytes, &completion_key, &overlapped, windows.INFINITE)) {
518 std.os.WindowsWaitResult.Aborted => return,
519 std.os.WindowsWaitResult.Normal => {},
520 }
521 if (overlapped != null) break;
522 }
523 const resume_node = @intToPtr(*ResumeNode, completion_key);
524 const handle = resume_node.handle;
525 const resume_node_id = resume_node.id;
526 switch (resume_node_id) {
527 ResumeNode.Id.Basic => {},
528 ResumeNode.Id.Stop => return,
529 ResumeNode.Id.EventFd => {
530 const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
531 const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
532 self.available_eventfd_resume_nodes.push(stack_node);
533 },
534 }
535 resume handle;
536 if (resume_node_id == ResumeNode.Id.EventFd) {
537 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
538 }
539 },
540 else => @compileError("unsupported OS"),
541 }
542 }
543 }
544
545 const OsData = switch (builtin.os) {
546 builtin.Os.linux => struct {
547 epollfd: i32,
548 final_eventfd: i32,
549 final_eventfd_event: std.os.linux.epoll_event,
550 },
551 builtin.Os.macosx => MacOsData,
552 builtin.Os.windows => struct {
553 io_port: windows.HANDLE,
554 extra_thread_count: usize,
555 },
556 else => struct {},
557 };
558
559 const MacOsData = struct {
560 kqfd: i32,
561 final_kevent: posix.Kevent,
562 kevents: []posix.Kevent,
563 };
564};
565
566test "std.event.Loop - basic" {
567 //var da = std.heap.DirectAllocator.init();
568 //defer da.deinit();
569
570 //const allocator = &da.allocator;
571
572 //var loop: Loop = undefined;
573 //try loop.initMultiThreaded(allocator);
574 //defer loop.deinit();
575
576 //loop.run();
577}
std/event/tcp.zig created+183
...@@ -0,0 +1,183 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const event = std.event;
5const mem = std.mem;
6const posix = std.os.posix;
7const windows = std.os.windows;
8const Loop = std.event.Loop;
9
10pub const Server = struct {
11 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
12
13 loop: *Loop,
14 sockfd: ?i32,
15 accept_coro: ?promise,
16 listen_address: std.net.Address,
17
18 waiting_for_emfile_node: PromiseNode,
19 listen_resume_node: event.Loop.ResumeNode,
20
21 const PromiseNode = std.LinkedList(promise).Node;
22
23 pub fn init(loop: *Loop) Server {
24 // TODO can't initialize handler coroutine here because we need well defined copy elision
25 return Server{
26 .loop = loop,
27 .sockfd = null,
28 .accept_coro = null,
29 .handleRequestFn = undefined,
30 .waiting_for_emfile_node = undefined,
31 .listen_address = undefined,
32 .listen_resume_node = event.Loop.ResumeNode{
33 .id = event.Loop.ResumeNode.Id.Basic,
34 .handle = undefined,
35 },
36 };
37 }
38
39 pub fn listen(
40 self: *Server,
41 address: *const std.net.Address,
42 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
43 ) !void {
44 self.handleRequestFn = handleRequestFn;
45
46 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
47 errdefer std.os.close(sockfd);
48 self.sockfd = sockfd;
49
50 try std.os.posixBind(sockfd, &address.os_addr);
51 try std.os.posixListen(sockfd, posix.SOMAXCONN);
52 self.listen_address = std.net.Address.initPosix(try std.os.posixGetSockName(sockfd));
53
54 self.accept_coro = try async<self.loop.allocator> Server.handler(self);
55 errdefer cancel self.accept_coro.?;
56
57 self.listen_resume_node.handle = self.accept_coro.?;
58 try self.loop.addFd(sockfd, &self.listen_resume_node);
59 errdefer self.loop.removeFd(sockfd);
60 }
61
62 /// Stop listening
63 pub fn close(self: *Server) void {
64 self.loop.removeFd(self.sockfd.?);
65 std.os.close(self.sockfd.?);
66 }
67
68 pub fn deinit(self: *Server) void {
69 if (self.accept_coro) |accept_coro| cancel accept_coro;
70 if (self.sockfd) |sockfd| std.os.close(sockfd);
71 }
72
73 pub async fn handler(self: *Server) void {
74 while (true) {
75 var accepted_addr: std.net.Address = undefined;
76 if (std.os.posixAccept(self.sockfd.?, &accepted_addr.os_addr, posix.SOCK_NONBLOCK | posix.SOCK_CLOEXEC)) |accepted_fd| {
77 var socket = std.os.File.openHandle(accepted_fd);
78 _ = async<self.loop.allocator> self.handleRequestFn(self, accepted_addr, socket) catch |err| switch (err) {
79 error.OutOfMemory => {
80 socket.close();
81 continue;
82 },
83 };
84 } else |err| switch (err) {
85 error.WouldBlock => {
86 suspend; // we will get resumed by epoll_wait in the event loop
87 continue;
88 },
89 error.ProcessFdQuotaExceeded => {
90 errdefer std.os.emfile_promise_queue.remove(&self.waiting_for_emfile_node);
91 suspend |p| {
92 self.waiting_for_emfile_node = PromiseNode.init(p);
93 std.os.emfile_promise_queue.append(&self.waiting_for_emfile_node);
94 }
95 continue;
96 },
97 error.ConnectionAborted, error.FileDescriptorClosed => continue,
98
99 error.PageFault => unreachable,
100 error.InvalidSyscall => unreachable,
101 error.FileDescriptorNotASocket => unreachable,
102 error.OperationNotSupported => unreachable,
103
104 error.SystemFdQuotaExceeded, error.SystemResources, error.ProtocolFailure, error.BlockedByFirewall, error.Unexpected => {
105 @panic("TODO handle this error");
106 },
107 }
108 }
109 }
110};
111
112pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {
113 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733
114
115 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
116 errdefer std.os.close(sockfd);
117
118 try std.os.posixConnectAsync(sockfd, &address.os_addr);
119 try await try async loop.waitFd(sockfd);
120 try std.os.posixGetSockOptConnectError(sockfd);
121
122 return std.os.File.openHandle(sockfd);
123}
124
125test "listen on a port, send bytes, receive bytes" {
126 if (builtin.os != builtin.Os.linux) {
127 // TODO build abstractions for other operating systems
128 return;
129 }
130 const MyServer = struct {
131 tcp_server: Server,
132
133 const Self = this;
134 async<*mem.Allocator> fn handler(tcp_server: *Server, _addr: *const std.net.Address, _socket: *const std.os.File) void {
135 const self = @fieldParentPtr(Self, "tcp_server", tcp_server);
136 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
137 defer socket.close();
138 // TODO guarantee elision of this allocation
139 const next_handler = async errorableHandler(self, _addr, socket) catch unreachable;
140 (await next_handler) catch |err| {
141 std.debug.panic("unable to handle connection: {}\n", err);
142 };
143 suspend |p| {
144 cancel p;
145 }
146 }
147 async fn errorableHandler(self: *Self, _addr: *const std.net.Address, _socket: *const std.os.File) !void {
148 const addr = _addr.*; // TODO https://github.com/ziglang/zig/issues/733
149 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
150
151 var adapter = std.io.FileOutStream.init(&socket);
152 var stream = &adapter.stream;
153 try stream.print("hello from server\n");
154 }
155 };
156
157 const ip4addr = std.net.parseIp4("127.0.0.1") catch unreachable;
158 const addr = std.net.Address.initIp4(ip4addr, 0);
159
160 var loop: Loop = undefined;
161 try loop.initSingleThreaded(std.debug.global_allocator);
162 var server = MyServer{ .tcp_server = Server.init(&loop) };
163 defer server.tcp_server.deinit();
164 try server.tcp_server.listen(addr, MyServer.handler);
165
166 const p = try async<std.debug.global_allocator> doAsyncTest(&loop, server.tcp_server.listen_address, &server.tcp_server);
167 defer cancel p;
168 loop.run();
169}
170
171async fn doAsyncTest(loop: *Loop, address: *const std.net.Address, server: *Server) void {
172 errdefer @panic("test failure");
173
174 var socket_file = try await try async connect(loop, address);
175 defer socket_file.close();
176
177 var buf: [512]u8 = undefined;
178 const amt_read = try socket_file.read(buf[0..]);
179 const msg = buf[0..amt_read];
180 assert(mem.eql(u8, msg, "hello from server\n"));
181 server.close();
182}
183