authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-17 15:17:06-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-07-17 15:17:44-04:00
loga9ab528e348a3f07dfaffcdcb1031062b1bfe8bb
treee9fe9d13b7d5c42579298bff53433a819cc2fe14
parentecf8da00c53b20085cc32e84030caf32e8b3e16b

std.event.Loop.onNextTick dispatches work to waiting threads

2 files changed, 113 insertions(+), 106 deletions(-)

std/atomic/queue.zig+14
...@@ -51,6 +51,20 @@ pub fn Queue(comptime T: type) type {...@@ -51,6 +51,20 @@ pub fn Queue(comptime T: type) type {
51 return head;51 return head;
52 }52 }
5353
54 pub fn unget(self: *Self, node: *Node) void {
55 while (@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) != 0) {}
56 defer assert(@atomicRmw(u8, &self.lock, builtin.AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst) == 1);
57
58 const opt_head = self.head;
59 self.head = node;
60 if (opt_head) |head| {
61 head.next = node;
62 } else {
63 assert(self.tail == null);
64 self.tail = node;
65 }
66 }
67
54 pub fn isEmpty(self: *Self) bool {68 pub fn isEmpty(self: *Self) bool {
55 return @atomicLoad(?*Node, &self.head, builtin.AtomicOrder.SeqCst) != null;69 return @atomicLoad(?*Node, &self.head, builtin.AtomicOrder.SeqCst) != null;
56 }70 }
std/event/loop.zig+99-106
...@@ -12,7 +12,6 @@ pub const Loop = struct {...@@ -12,7 +12,6 @@ pub const Loop = struct {
12 next_tick_queue: std.atomic.Queue(promise),12 next_tick_queue: std.atomic.Queue(promise),
13 os_data: OsData,13 os_data: OsData,
14 final_resume_node: ResumeNode,14 final_resume_node: ResumeNode,
15 dispatch_lock: u8, // TODO make this a bool
16 pending_event_count: usize,15 pending_event_count: usize,
17 extra_threads: []*std.os.Thread,16 extra_threads: []*std.os.Thread,
1817
...@@ -74,11 +73,10 @@ pub const Loop = struct {...@@ -74,11 +73,10 @@ pub const Loop = struct {
74 /// max(thread_count - 1, 0)73 /// max(thread_count - 1, 0)
75 fn initInternal(self: *Loop, allocator: *mem.Allocator, thread_count: usize) !void {74 fn initInternal(self: *Loop, allocator: *mem.Allocator, thread_count: usize) !void {
76 self.* = Loop{75 self.* = Loop{
77 .pending_event_count = 0,76 .pending_event_count = 1,
78 .allocator = allocator,77 .allocator = allocator,
79 .os_data = undefined,78 .os_data = undefined,
80 .next_tick_queue = std.atomic.Queue(promise).init(),79 .next_tick_queue = std.atomic.Queue(promise).init(),
81 .dispatch_lock = 1, // start locked so threads go directly into epoll wait
82 .extra_threads = undefined,80 .extra_threads = undefined,
83 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),81 .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
84 .eventfd_resume_nodes = undefined,82 .eventfd_resume_nodes = undefined,
...@@ -306,7 +304,7 @@ pub const Loop = struct {...@@ -306,7 +304,7 @@ pub const Loop = struct {
306 pub fn addFd(self: *Loop, fd: i32, resume_node: *ResumeNode) !void {304 pub fn addFd(self: *Loop, fd: i32, resume_node: *ResumeNode) !void {
307 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);305 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
308 errdefer {306 errdefer {
309 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);307 self.finishOneEvent();
310 }308 }
311 try self.modFd(309 try self.modFd(
312 fd,310 fd,
...@@ -326,7 +324,7 @@ pub const Loop = struct {...@@ -326,7 +324,7 @@ pub const Loop = struct {
326324
327 pub fn removeFd(self: *Loop, fd: i32) void {325 pub fn removeFd(self: *Loop, fd: i32) void {
328 self.removeFdNoCounter(fd);326 self.removeFdNoCounter(fd);
329 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);327 self.finishOneEvent();
330 }328 }
331329
332 fn removeFdNoCounter(self: *Loop, fd: i32) void {330 fn removeFdNoCounter(self: *Loop, fd: i32) void {
...@@ -345,14 +343,70 @@ pub const Loop = struct {...@@ -345,14 +343,70 @@ pub const Loop = struct {
345 }343 }
346 }344 }
347345
346 fn dispatch(self: *Loop) void {
347 while (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {
348 const next_tick_node = self.next_tick_queue.get() orelse {
349 self.available_eventfd_resume_nodes.push(resume_stack_node);
350 return;
351 };
352 const eventfd_node = &resume_stack_node.data;
353 eventfd_node.base.handle = next_tick_node.data;
354 switch (builtin.os) {
355 builtin.Os.macosx => {
356 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.kevent);
357 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
358 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch {
359 self.next_tick_queue.unget(next_tick_node);
360 self.available_eventfd_resume_nodes.push(resume_stack_node);
361 return;
362 };
363 },
364 builtin.Os.linux => {
365 // the pending count is already accounted for
366 const epoll_events = posix.EPOLLONESHOT | std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT |
367 std.os.linux.EPOLLET;
368 self.modFd(
369 eventfd_node.eventfd,
370 eventfd_node.epoll_op,
371 epoll_events,
372 &eventfd_node.base,
373 ) catch {
374 self.next_tick_queue.unget(next_tick_node);
375 self.available_eventfd_resume_nodes.push(resume_stack_node);
376 return;
377 };
378 },
379 builtin.Os.windows => {
380 // this value is never dereferenced but we need it to be non-null so that
381 // the consumer code can decide whether to read the completion key.
382 // it has to do this for normal I/O, so we match that behavior here.
383 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
384 std.os.windowsPostQueuedCompletionStatus(
385 self.os_data.io_port,
386 undefined,
387 eventfd_node.completion_key,
388 overlapped,
389 ) catch {
390 self.next_tick_queue.unget(next_tick_node);
391 self.available_eventfd_resume_nodes.push(resume_stack_node);
392 return;
393 };
394 },
395 else => @compileError("unsupported OS"),
396 }
397 }
398 }
399
348 /// Bring your own linked list node. This means it can't fail.400 /// Bring your own linked list node. This means it can't fail.
349 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {401 pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
350 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);402 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
351 self.next_tick_queue.put(node);403 self.next_tick_queue.put(node);
404 self.dispatch();
352 }405 }
353406
354 pub fn run(self: *Loop) void {407 pub fn run(self: *Loop) void {
355 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);408 self.finishOneEvent(); // the reference we start with
409
356 self.workerRun();410 self.workerRun();
357 for (self.extra_threads) |extra_thread| {411 for (self.extra_threads) |extra_thread| {
358 extra_thread.wait();412 extra_thread.wait();
...@@ -396,106 +450,45 @@ pub const Loop = struct {...@@ -396,106 +450,45 @@ pub const Loop = struct {
396 }450 }
397 }451 }
398452
399 fn workerRun(self: *Loop) void {453 fn finishOneEvent(self: *Loop) void {
400 start_over: while (true) {454 if (@atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst) == 1) {
401 if (@atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst) == 0) {455 // cause all the threads to stop
402 while (self.next_tick_queue.get()) |next_tick_node| {456 switch (builtin.os) {
403 const handle = next_tick_node.data;457 builtin.Os.linux => {
404 if (self.next_tick_queue.isEmpty()) {458 // writing 8 bytes to an eventfd cannot fail
405 // last node, just resume it459 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
406 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);460 return;
407 resume handle;461 },
408 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);462 builtin.Os.macosx => {
409 continue :start_over;463 const final_kevent = (*[1]posix.Kevent)(&self.os_data.final_kevent);
410 }464 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
411465 // cannot fail because we already added it and this just enables it
412 // non-last node, stick it in the epoll/kqueue set so that466 _ = std.os.bsdKEvent(self.os_data.kqfd, final_kevent, eventlist, null) catch unreachable;
413 // other threads can get to it467 return;
414 if (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {468 },
415 const eventfd_node = &resume_stack_node.data;469 builtin.Os.windows => {
416 eventfd_node.base.handle = handle;470 var i: usize = 0;
417 switch (builtin.os) {471 while (i < self.os_data.extra_thread_count) : (i += 1) {
418 builtin.Os.macosx => {472 while (true) {
419 const kevent_array = (*[1]posix.Kevent)(&eventfd_node.kevent);473 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
420 const eventlist = ([*]posix.Kevent)(undefined)[0..0];474 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
421 _ = std.os.bsdKEvent(self.os_data.kqfd, kevent_array, eventlist, null) catch {475 break;
422 // fine, we didn't need it anyway
423 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
424 self.available_eventfd_resume_nodes.push(resume_stack_node);
425 resume handle;
426 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
427 continue :start_over;
428 };
429 },
430 builtin.Os.linux => {
431 // the pending count is already accounted for
432 const epoll_events = posix.EPOLLONESHOT | std.os.linux.EPOLLIN | std.os.linux.EPOLLOUT | std.os.linux.EPOLLET;
433 self.modFd(eventfd_node.eventfd, eventfd_node.epoll_op, epoll_events, &eventfd_node.base) catch {
434 // fine, we didn't need it anyway
435 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
436 self.available_eventfd_resume_nodes.push(resume_stack_node);
437 resume handle;
438 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
439 continue :start_over;
440 };
441 },
442 builtin.Os.windows => {
443 // this value is never dereferenced but we need it to be non-null so that
444 // the consumer code can decide whether to read the completion key.
445 // it has to do this for normal I/O, so we match that behavior here.
446 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
447 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, eventfd_node.completion_key, overlapped) catch {
448 // fine, we didn't need it anyway
449 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
450 self.available_eventfd_resume_nodes.push(resume_stack_node);
451 resume handle;
452 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
453 continue :start_over;
454 };
455 },
456 else => @compileError("unsupported OS"),
457 }476 }
458 } else {
459 // threads are too busy, can't add another eventfd to wake one up
460 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);
461 resume handle;
462 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
463 continue :start_over;
464 }477 }
465 }478 return;
466479 },
467 const pending_event_count = @atomicLoad(usize, &self.pending_event_count, AtomicOrder.SeqCst);480 else => @compileError("unsupported OS"),
468 if (pending_event_count == 0) {481 }
469 // cause all the threads to stop482 }
470 switch (builtin.os) {483 }
471 builtin.Os.linux => {
472 // writing 8 bytes to an eventfd cannot fail
473 std.os.posixWrite(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
474 return;
475 },
476 builtin.Os.macosx => {
477 const final_kevent = (*[1]posix.Kevent)(&self.os_data.final_kevent);
478 const eventlist = ([*]posix.Kevent)(undefined)[0..0];
479 // cannot fail because we already added it and this just enables it
480 _ = std.os.bsdKEvent(self.os_data.kqfd, final_kevent, eventlist, null) catch unreachable;
481 return;
482 },
483 builtin.Os.windows => {
484 var i: usize = 0;
485 while (i < self.os_data.extra_thread_count) : (i += 1) {
486 while (true) {
487 const overlapped = @intToPtr(?*windows.OVERLAPPED, 0x1);
488 std.os.windowsPostQueuedCompletionStatus(self.os_data.io_port, undefined, @ptrToInt(&self.final_resume_node), overlapped) catch continue;
489 break;
490 }
491 }
492 return;
493 },
494 else => @compileError("unsupported OS"),
495 }
496 }
497484
498 _ = @atomicRmw(u8, &self.dispatch_lock, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);485 fn workerRun(self: *Loop) void {
486 while (true) {
487 while (true) {
488 const next_tick_node = self.next_tick_queue.get() orelse break;
489 self.dispatch();
490 resume next_tick_node.data;
491 self.finishOneEvent();
499 }492 }
500493
501 switch (builtin.os) {494 switch (builtin.os) {
...@@ -519,7 +512,7 @@ pub const Loop = struct {...@@ -519,7 +512,7 @@ pub const Loop = struct {
519 }512 }
520 resume handle;513 resume handle;
521 if (resume_node_id == ResumeNode.Id.EventFd) {514 if (resume_node_id == ResumeNode.Id.EventFd) {
522 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);515 self.finishOneEvent();
523 }516 }
524 }517 }
525 },518 },
...@@ -541,7 +534,7 @@ pub const Loop = struct {...@@ -541,7 +534,7 @@ pub const Loop = struct {
541 }534 }
542 resume handle;535 resume handle;
543 if (resume_node_id == ResumeNode.Id.EventFd) {536 if (resume_node_id == ResumeNode.Id.EventFd) {
544 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);537 self.finishOneEvent();
545 }538 }
546 }539 }
547 },540 },
...@@ -570,7 +563,7 @@ pub const Loop = struct {...@@ -570,7 +563,7 @@ pub const Loop = struct {
570 }563 }
571 resume handle;564 resume handle;
572 if (resume_node_id == ResumeNode.Id.EventFd) {565 if (resume_node_id == ResumeNode.Id.EventFd) {
573 _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);566 self.finishOneEvent();
574 }567 }
575 },568 },
576 else => @compileError("unsupported OS"),569 else => @compileError("unsupported OS"),