authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-16 23:19:13-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2019-08-16 23:19:13-04:00
loge24cc2e77b741fe49e738acc497fbedf2998008c
tree73c1fa20a1f0e30bcfe6759fe2f9ef6f967768dc
parent3dce41b61a20c23c494a485b1d3092e7c67dd97f
signaturelock-open Commit is signed but in an unrecognized format.

std.event.Loop: fix not waking up after file system I/O

for single threaded event loops

2 files changed, 11 insertions(+), 9 deletions(-)

std/c.zig+1
...@@ -69,6 +69,7 @@ pub extern "c" fn raise(sig: c_int) c_int;...@@ -69,6 +69,7 @@ pub extern "c" fn raise(sig: c_int) c_int;
69pub extern "c" fn read(fd: fd_t, buf: [*]u8, nbyte: usize) isize;69pub extern "c" fn read(fd: fd_t, buf: [*]u8, nbyte: usize) isize;
70pub extern "c" fn pread(fd: fd_t, buf: [*]u8, nbyte: usize, offset: u64) isize;70pub extern "c" fn pread(fd: fd_t, buf: [*]u8, nbyte: usize, offset: u64) isize;
71pub extern "c" fn preadv(fd: c_int, iov: [*]const iovec, iovcnt: c_uint, offset: usize) isize;71pub extern "c" fn preadv(fd: c_int, iov: [*]const iovec, iovcnt: c_uint, offset: usize) isize;
72pub extern "c" fn writev(fd: c_int, iov: [*]const iovec_const, iovcnt: c_uint) isize;
72pub extern "c" fn pwritev(fd: c_int, iov: [*]const iovec_const, iovcnt: c_uint, offset: usize) isize;73pub extern "c" fn pwritev(fd: c_int, iov: [*]const iovec_const, iovcnt: c_uint, offset: usize) isize;
73pub extern "c" fn stat(noalias path: [*]const u8, noalias buf: *Stat) c_int;74pub extern "c" fn stat(noalias path: [*]const u8, noalias buf: *Stat) c_int;
74pub extern "c" fn write(fd: fd_t, buf: [*]const u8, nbyte: usize) isize;75pub extern "c" fn write(fd: fd_t, buf: [*]const u8, nbyte: usize) isize;
std/event/loop.zig+10-9
...@@ -149,13 +149,14 @@ pub const Loop = struct {...@@ -149,13 +149,14 @@ pub const Loop = struct {
149 .overlapped = ResumeNode.overlapped_init,149 .overlapped = ResumeNode.overlapped_init,
150 },150 },
151 };151 };
152 const extra_thread_count = thread_count - 1;152 // We need an extra one of these in case the fs thread wants to use onNextTick
153 self.eventfd_resume_nodes = try self.allocator.alloc(153 self.eventfd_resume_nodes = try self.allocator.alloc(
154 std.atomic.Stack(ResumeNode.EventFd).Node,154 std.atomic.Stack(ResumeNode.EventFd).Node,
155 extra_thread_count,155 thread_count,
156 );156 );
157 errdefer self.allocator.free(self.eventfd_resume_nodes);157 errdefer self.allocator.free(self.eventfd_resume_nodes);
158158
159 const extra_thread_count = thread_count - 1;
159 self.extra_threads = try self.allocator.alloc(*Thread, extra_thread_count);160 self.extra_threads = try self.allocator.alloc(*Thread, extra_thread_count);
160 errdefer self.allocator.free(self.extra_threads);161 errdefer self.allocator.free(self.extra_threads);
161162
...@@ -197,7 +198,7 @@ pub const Loop = struct {...@@ -197,7 +198,7 @@ pub const Loop = struct {
197 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{198 eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
198 .data = ResumeNode.EventFd{199 .data = ResumeNode.EventFd{
199 .base = ResumeNode{200 .base = ResumeNode{
200 .id = ResumeNode.Id.EventFd,201 .id = .EventFd,
201 .handle = undefined,202 .handle = undefined,
202 .overlapped = ResumeNode.overlapped_init,203 .overlapped = ResumeNode.overlapped_init,
203 },204 },
...@@ -454,12 +455,12 @@ pub const Loop = struct {...@@ -454,12 +455,12 @@ pub const Loop = struct {
454 self.finishOneEvent();455 self.finishOneEvent();
455 }456 }
456457
457 pub async fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) !void {458 pub fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) !void {
458 defer self.linuxRemoveFd(fd);459 defer self.linuxRemoveFd(fd);
459 suspend {460 suspend {
460 var resume_node = ResumeNode.Basic{461 var resume_node = ResumeNode.Basic{
461 .base = ResumeNode{462 .base = ResumeNode{
462 .id = ResumeNode.Id.Basic,463 .id = .Basic,
463 .handle = @frame(),464 .handle = @frame(),
464 .overlapped = ResumeNode.overlapped_init,465 .overlapped = ResumeNode.overlapped_init,
465 },466 },
...@@ -793,8 +794,8 @@ pub const Loop = struct {...@@ -793,8 +794,8 @@ pub const Loop = struct {
793794
794 fn posixFsRun(self: *Loop) void {795 fn posixFsRun(self: *Loop) void {
795 while (true) {796 while (true) {
796 if (builtin.os == builtin.Os.linux) {797 if (builtin.os == .linux) {
797 _ = @atomicRmw(i32, &self.os_data.fs_queue_item, AtomicRmwOp.Xchg, 0, AtomicOrder.SeqCst);798 _ = @atomicRmw(i32, &self.os_data.fs_queue_item, .Xchg, 0, .SeqCst);
798 }799 }
799 while (self.os_data.fs_queue.get()) |node| {800 while (self.os_data.fs_queue.get()) |node| {
800 switch (node.data.msg) {801 switch (node.data.msg) {
...@@ -833,14 +834,14 @@ pub const Loop = struct {...@@ -833,14 +834,14 @@ pub const Loop = struct {
833 self.finishOneEvent();834 self.finishOneEvent();
834 }835 }
835 switch (builtin.os) {836 switch (builtin.os) {
836 builtin.Os.linux => {837 .linux => {
837 const rc = os.linux.futex_wait(&self.os_data.fs_queue_item, os.linux.FUTEX_WAIT, 0, null);838 const rc = os.linux.futex_wait(&self.os_data.fs_queue_item, os.linux.FUTEX_WAIT, 0, null);
838 switch (os.linux.getErrno(rc)) {839 switch (os.linux.getErrno(rc)) {
839 0, os.EINTR, os.EAGAIN => continue,840 0, os.EINTR, os.EAGAIN => continue,
840 else => unreachable,841 else => unreachable,
841 }842 }
842 },843 },
843 builtin.Os.macosx, builtin.Os.freebsd, builtin.Os.netbsd => {844 .macosx, .freebsd, .netbsd => {
844 const fs_kevs = (*const [1]os.Kevent)(&self.os_data.fs_kevent_wait);845 const fs_kevs = (*const [1]os.Kevent)(&self.os_data.fs_kevent_wait);
845 var out_kevs: [1]os.Kevent = undefined;846 var out_kevs: [1]os.Kevent = undefined;
846 _ = os.kevent(self.os_data.fs_kqfd, fs_kevs, out_kevs[0..], null) catch unreachable;847 _ = os.kevent(self.os_data.fs_kqfd, fs_kevs, out_kevs[0..], null) catch unreachable;