Merge pull request #3315 from ziglang/mv-std-lib

Move std/ to lib/std/

1 files changed, 923 insertions, 0 deletions

diff --git a/lib/std/event/loop.zig b/lib/std/event/loop.zig
new file mode 100644
index 0000000000..d0d36abc0c
--- /dev/null
+++ b/lib/std/event/loop.zig
@@ -0,0 +1,923 @@
+const std = @import("../std.zig");
+const builtin = @import("builtin");
+const root = @import("root");
+const assert = std.debug.assert;
+const testing = std.testing;
+const mem = std.mem;
+const AtomicRmwOp = builtin.AtomicRmwOp;
+const AtomicOrder = builtin.AtomicOrder;
+const fs = std.event.fs;
+const os = std.os;
+const windows = os.windows;
+const maxInt = std.math.maxInt;
+const Thread = std.Thread;
+
+pub const Loop = struct {
+    allocator: *mem.Allocator,
+    next_tick_queue: std.atomic.Queue(anyframe),
+    os_data: OsData,
+    final_resume_node: ResumeNode,
+    pending_event_count: usize,
+    extra_threads: []*Thread,
+
+    // pre-allocated eventfds. all permanently active.
+    // this is how we send promises to be resumed on other threads.
+    available_eventfd_resume_nodes: std.atomic.Stack(ResumeNode.EventFd),
+    eventfd_resume_nodes: []std.atomic.Stack(ResumeNode.EventFd).Node,
+
+    pub const NextTickNode = std.atomic.Queue(anyframe).Node;
+
+    pub const ResumeNode = struct {
+        id: Id,
+        handle: anyframe,
+        overlapped: Overlapped,
+
+        pub const overlapped_init = switch (builtin.os) {
+            .windows => windows.OVERLAPPED{
+                .Internal = 0,
+                .InternalHigh = 0,
+                .Offset = 0,
+                .OffsetHigh = 0,
+                .hEvent = null,
+            },
+            else => {},
+        };
+        pub const Overlapped = @typeOf(overlapped_init);
+
+        pub const Id = enum {
+            Basic,
+            Stop,
+            EventFd,
+        };
+
+        pub const EventFd = switch (builtin.os) {
+            .macosx, .freebsd, .netbsd => KEventFd,
+            .linux => struct {
+                base: ResumeNode,
+                epoll_op: u32,
+                eventfd: i32,
+            },
+            .windows => struct {
+                base: ResumeNode,
+                completion_key: usize,
+            },
+            else => @compileError("unsupported OS"),
+        };
+
+        const KEventFd = struct {
+            base: ResumeNode,
+            kevent: os.Kevent,
+        };
+
+        pub const Basic = switch (builtin.os) {
+            .macosx, .freebsd, .netbsd => KEventBasic,
+            .linux => struct {
+                base: ResumeNode,
+            },
+            .windows => struct {
+                base: ResumeNode,
+            },
+            else => @compileError("unsupported OS"),
+        };
+
+        const KEventBasic = struct {
+            base: ResumeNode,
+            kev: os.Kevent,
+        };
+    };
+
+    var global_instance_state: Loop = undefined;
+    const default_instance: ?*Loop = switch (std.io.mode) {
+        .blocking => null,
+        .evented => &global_instance_state,
+    };
+    pub const instance: ?*Loop = if (@hasDecl(root, "event_loop")) root.event_loop else default_instance;
+
+    /// TODO copy elision / named return values so that the threads referencing *Loop
+    /// have the correct pointer value.
+    /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
+    pub fn init(self: *Loop, allocator: *mem.Allocator) !void {
+        if (builtin.single_threaded) {
+            return self.initSingleThreaded(allocator);
+        } else {
+            return self.initMultiThreaded(allocator);
+        }
+    }
+
+    /// After initialization, call run().
+    /// TODO copy elision / named return values so that the threads referencing *Loop
+    /// have the correct pointer value.
+    /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
+    pub fn initSingleThreaded(self: *Loop, allocator: *mem.Allocator) !void {
+        return self.initInternal(allocator, 1);
+    }
+
+    /// The allocator must be thread-safe because we use it for multiplexing
+    /// async functions onto kernel threads.
+    /// After initialization, call run().
+    /// TODO copy elision / named return values so that the threads referencing *Loop
+    /// have the correct pointer value.
+    /// https://github.com/ziglang/zig/issues/2761 and https://github.com/ziglang/zig/issues/2765
+    pub fn initMultiThreaded(self: *Loop, allocator: *mem.Allocator) !void {
+        if (builtin.single_threaded) @compileError("initMultiThreaded unavailable when building in single-threaded mode");
+        const core_count = try Thread.cpuCount();
+        return self.initInternal(allocator, core_count);
+    }
+
+    /// Thread count is the total thread count. The thread pool size will be
+    /// max(thread_count - 1, 0)
+    fn initInternal(self: *Loop, allocator: *mem.Allocator, thread_count: usize) !void {
+        self.* = Loop{
+            .pending_event_count = 1,
+            .allocator = allocator,
+            .os_data = undefined,
+            .next_tick_queue = std.atomic.Queue(anyframe).init(),
+            .extra_threads = undefined,
+            .available_eventfd_resume_nodes = std.atomic.Stack(ResumeNode.EventFd).init(),
+            .eventfd_resume_nodes = undefined,
+            .final_resume_node = ResumeNode{
+                .id = ResumeNode.Id.Stop,
+                .handle = undefined,
+                .overlapped = ResumeNode.overlapped_init,
+            },
+        };
+        // We need at least one of these in case the fs thread wants to use onNextTick
+        const extra_thread_count = thread_count - 1;
+        const resume_node_count = std.math.max(extra_thread_count, 1);
+        self.eventfd_resume_nodes = try self.allocator.alloc(
+            std.atomic.Stack(ResumeNode.EventFd).Node,
+            resume_node_count,
+        );
+        errdefer self.allocator.free(self.eventfd_resume_nodes);
+
+        self.extra_threads = try self.allocator.alloc(*Thread, extra_thread_count);
+        errdefer self.allocator.free(self.extra_threads);
+
+        try self.initOsData(extra_thread_count);
+        errdefer self.deinitOsData();
+    }
+
+    pub fn deinit(self: *Loop) void {
+        self.deinitOsData();
+        self.allocator.free(self.extra_threads);
+    }
+
+    const InitOsDataError = os.EpollCreateError || mem.Allocator.Error || os.EventFdError ||
+        Thread.SpawnError || os.EpollCtlError || os.KEventError ||
+        windows.CreateIoCompletionPortError;
+
+    const wakeup_bytes = [_]u8{0x1} ** 8;
+
+    fn initOsData(self: *Loop, extra_thread_count: usize) InitOsDataError!void {
+        switch (builtin.os) {
+            .linux => {
+                self.os_data.fs_queue = std.atomic.Queue(fs.Request).init();
+                self.os_data.fs_queue_item = 0;
+                // we need another thread for the file system because Linux does not have an async
+                // file system I/O API.
+                self.os_data.fs_end_request = fs.RequestNode{
+                    .prev = undefined,
+                    .next = undefined,
+                    .data = fs.Request{
+                        .msg = fs.Request.Msg.End,
+                        .finish = fs.Request.Finish.NoAction,
+                    },
+                };
+
+                errdefer {
+                    while (self.available_eventfd_resume_nodes.pop()) |node| os.close(node.data.eventfd);
+                }
+                for (self.eventfd_resume_nodes) |*eventfd_node| {
+                    eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
+                        .data = ResumeNode.EventFd{
+                            .base = ResumeNode{
+                                .id = .EventFd,
+                                .handle = undefined,
+                                .overlapped = ResumeNode.overlapped_init,
+                            },
+                            .eventfd = try os.eventfd(1, os.EFD_CLOEXEC | os.EFD_NONBLOCK),
+                            .epoll_op = os.EPOLL_CTL_ADD,
+                        },
+                        .next = undefined,
+                    };
+                    self.available_eventfd_resume_nodes.push(eventfd_node);
+                }
+
+                self.os_data.epollfd = try os.epoll_create1(os.EPOLL_CLOEXEC);
+                errdefer os.close(self.os_data.epollfd);
+
+                self.os_data.final_eventfd = try os.eventfd(0, os.EFD_CLOEXEC | os.EFD_NONBLOCK);
+                errdefer os.close(self.os_data.final_eventfd);
+
+                self.os_data.final_eventfd_event = os.epoll_event{
+                    .events = os.EPOLLIN,
+                    .data = os.epoll_data{ .ptr = @ptrToInt(&self.final_resume_node) },
+                };
+                try os.epoll_ctl(
+                    self.os_data.epollfd,
+                    os.EPOLL_CTL_ADD,
+                    self.os_data.final_eventfd,
+                    &self.os_data.final_eventfd_event,
+                );
+
+                self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
+                errdefer {
+                    self.posixFsRequest(&self.os_data.fs_end_request);
+                    self.os_data.fs_thread.wait();
+                }
+
+                if (builtin.single_threaded) {
+                    assert(extra_thread_count == 0);
+                    return;
+                }
+
+                var extra_thread_index: usize = 0;
+                errdefer {
+                    // writing 8 bytes to an eventfd cannot fail
+                    os.write(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
+                    while (extra_thread_index != 0) {
+                        extra_thread_index -= 1;
+                        self.extra_threads[extra_thread_index].wait();
+                    }
+                }
+                while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
+                    self.extra_threads[extra_thread_index] = try Thread.spawn(self, workerRun);
+                }
+            },
+            .macosx, .freebsd, .netbsd => {
+                self.os_data.kqfd = try os.kqueue();
+                errdefer os.close(self.os_data.kqfd);
+
+                self.os_data.fs_kqfd = try os.kqueue();
+                errdefer os.close(self.os_data.fs_kqfd);
+
+                self.os_data.fs_queue = std.atomic.Queue(fs.Request).init();
+                // we need another thread for the file system because Darwin does not have an async
+                // file system I/O API.
+                self.os_data.fs_end_request = fs.RequestNode{
+                    .prev = undefined,
+                    .next = undefined,
+                    .data = fs.Request{
+                        .msg = fs.Request.Msg.End,
+                        .finish = fs.Request.Finish.NoAction,
+                    },
+                };
+
+                const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+
+                for (self.eventfd_resume_nodes) |*eventfd_node, i| {
+                    eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
+                        .data = ResumeNode.EventFd{
+                            .base = ResumeNode{
+                                .id = ResumeNode.Id.EventFd,
+                                .handle = undefined,
+                                .overlapped = ResumeNode.overlapped_init,
+                            },
+                            // this one is for sending events
+                            .kevent = os.Kevent{
+                                .ident = i,
+                                .filter = os.EVFILT_USER,
+                                .flags = os.EV_CLEAR | os.EV_ADD | os.EV_DISABLE,
+                                .fflags = 0,
+                                .data = 0,
+                                .udata = @ptrToInt(&eventfd_node.data.base),
+                            },
+                        },
+                        .next = undefined,
+                    };
+                    self.available_eventfd_resume_nodes.push(eventfd_node);
+                    const kevent_array = (*const [1]os.Kevent)(&eventfd_node.data.kevent);
+                    _ = try os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null);
+                    eventfd_node.data.kevent.flags = os.EV_CLEAR | os.EV_ENABLE;
+                    eventfd_node.data.kevent.fflags = os.NOTE_TRIGGER;
+                }
+
+                // Pre-add so that we cannot get error.SystemResources
+                // later when we try to activate it.
+                self.os_data.final_kevent = os.Kevent{
+                    .ident = extra_thread_count,
+                    .filter = os.EVFILT_USER,
+                    .flags = os.EV_ADD | os.EV_DISABLE,
+                    .fflags = 0,
+                    .data = 0,
+                    .udata = @ptrToInt(&self.final_resume_node),
+                };
+                const final_kev_arr = (*const [1]os.Kevent)(&self.os_data.final_kevent);
+                _ = try os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null);
+                self.os_data.final_kevent.flags = os.EV_ENABLE;
+                self.os_data.final_kevent.fflags = os.NOTE_TRIGGER;
+
+                self.os_data.fs_kevent_wake = os.Kevent{
+                    .ident = 0,
+                    .filter = os.EVFILT_USER,
+                    .flags = os.EV_ADD | os.EV_ENABLE,
+                    .fflags = os.NOTE_TRIGGER,
+                    .data = 0,
+                    .udata = undefined,
+                };
+
+                self.os_data.fs_kevent_wait = os.Kevent{
+                    .ident = 0,
+                    .filter = os.EVFILT_USER,
+                    .flags = os.EV_ADD | os.EV_CLEAR,
+                    .fflags = 0,
+                    .data = 0,
+                    .udata = undefined,
+                };
+
+                self.os_data.fs_thread = try Thread.spawn(self, posixFsRun);
+                errdefer {
+                    self.posixFsRequest(&self.os_data.fs_end_request);
+                    self.os_data.fs_thread.wait();
+                }
+
+                if (builtin.single_threaded) {
+                    assert(extra_thread_count == 0);
+                    return;
+                }
+
+                var extra_thread_index: usize = 0;
+                errdefer {
+                    _ = os.kevent(self.os_data.kqfd, final_kev_arr, empty_kevs, null) catch unreachable;
+                    while (extra_thread_index != 0) {
+                        extra_thread_index -= 1;
+                        self.extra_threads[extra_thread_index].wait();
+                    }
+                }
+                while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
+                    self.extra_threads[extra_thread_index] = try Thread.spawn(self, workerRun);
+                }
+            },
+            .windows => {
+                self.os_data.io_port = try windows.CreateIoCompletionPort(
+                    windows.INVALID_HANDLE_VALUE,
+                    null,
+                    undefined,
+                    maxInt(windows.DWORD),
+                );
+                errdefer windows.CloseHandle(self.os_data.io_port);
+
+                for (self.eventfd_resume_nodes) |*eventfd_node, i| {
+                    eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
+                        .data = ResumeNode.EventFd{
+                            .base = ResumeNode{
+                                .id = ResumeNode.Id.EventFd,
+                                .handle = undefined,
+                                .overlapped = ResumeNode.overlapped_init,
+                            },
+                            // this one is for sending events
+                            .completion_key = @ptrToInt(&eventfd_node.data.base),
+                        },
+                        .next = undefined,
+                    };
+                    self.available_eventfd_resume_nodes.push(eventfd_node);
+                }
+
+                if (builtin.single_threaded) {
+                    assert(extra_thread_count == 0);
+                    return;
+                }
+
+                var extra_thread_index: usize = 0;
+                errdefer {
+                    var i: usize = 0;
+                    while (i < extra_thread_index) : (i += 1) {
+                        while (true) {
+                            const overlapped = &self.final_resume_node.overlapped;
+                            windows.PostQueuedCompletionStatus(self.os_data.io_port, undefined, undefined, overlapped) catch continue;
+                            break;
+                        }
+                    }
+                    while (extra_thread_index != 0) {
+                        extra_thread_index -= 1;
+                        self.extra_threads[extra_thread_index].wait();
+                    }
+                }
+                while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
+                    self.extra_threads[extra_thread_index] = try Thread.spawn(self, workerRun);
+                }
+            },
+            else => {},
+        }
+    }
+
+    fn deinitOsData(self: *Loop) void {
+        switch (builtin.os) {
+            .linux => {
+                os.close(self.os_data.final_eventfd);
+                while (self.available_eventfd_resume_nodes.pop()) |node| os.close(node.data.eventfd);
+                os.close(self.os_data.epollfd);
+                self.allocator.free(self.eventfd_resume_nodes);
+            },
+            .macosx, .freebsd, .netbsd => {
+                os.close(self.os_data.kqfd);
+                os.close(self.os_data.fs_kqfd);
+            },
+            .windows => {
+                windows.CloseHandle(self.os_data.io_port);
+            },
+            else => {},
+        }
+    }
+
+    /// resume_node must live longer than the anyframe that it holds a reference to.
+    /// flags must contain EPOLLET
+    pub fn linuxAddFd(self: *Loop, fd: i32, resume_node: *ResumeNode, flags: u32) !void {
+        assert(flags & os.EPOLLET == os.EPOLLET);
+        self.beginOneEvent();
+        errdefer self.finishOneEvent();
+        try self.linuxModFd(
+            fd,
+            os.EPOLL_CTL_ADD,
+            flags,
+            resume_node,
+        );
+    }
+
+    pub fn linuxModFd(self: *Loop, fd: i32, op: u32, flags: u32, resume_node: *ResumeNode) !void {
+        assert(flags & os.EPOLLET == os.EPOLLET);
+        var ev = os.linux.epoll_event{
+            .events = flags,
+            .data = os.linux.epoll_data{ .ptr = @ptrToInt(resume_node) },
+        };
+        try os.epoll_ctl(self.os_data.epollfd, op, fd, &ev);
+    }
+
+    pub fn linuxRemoveFd(self: *Loop, fd: i32) void {
+        os.epoll_ctl(self.os_data.epollfd, os.linux.EPOLL_CTL_DEL, fd, null) catch {};
+        self.finishOneEvent();
+    }
+
+    pub fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) !void {
+        defer self.linuxRemoveFd(fd);
+        suspend {
+            var resume_node = ResumeNode.Basic{
+                .base = ResumeNode{
+                    .id = .Basic,
+                    .handle = @frame(),
+                    .overlapped = ResumeNode.overlapped_init,
+                },
+            };
+            try self.linuxAddFd(fd, &resume_node.base, flags);
+        }
+    }
+
+    pub fn waitUntilFdReadable(self: *Loop, fd: os.fd_t) !void {
+        return self.linuxWaitFd(fd, os.EPOLLET | os.EPOLLIN);
+    }
+
+    pub async fn bsdWaitKev(self: *Loop, ident: usize, filter: i16, fflags: u32) !os.Kevent {
+        var resume_node = ResumeNode.Basic{
+            .base = ResumeNode{
+                .id = ResumeNode.Id.Basic,
+                .handle = @frame(),
+                .overlapped = ResumeNode.overlapped_init,
+            },
+            .kev = undefined,
+        };
+        defer self.bsdRemoveKev(ident, filter);
+        suspend {
+            try self.bsdAddKev(&resume_node, ident, filter, fflags);
+        }
+        return resume_node.kev;
+    }
+
+    /// resume_node must live longer than the anyframe that it holds a reference to.
+    pub fn bsdAddKev(self: *Loop, resume_node: *ResumeNode.Basic, ident: usize, filter: i16, fflags: u32) !void {
+        self.beginOneEvent();
+        errdefer self.finishOneEvent();
+        var kev = os.Kevent{
+            .ident = ident,
+            .filter = filter,
+            .flags = os.EV_ADD | os.EV_ENABLE | os.EV_CLEAR,
+            .fflags = fflags,
+            .data = 0,
+            .udata = @ptrToInt(&resume_node.base),
+        };
+        const kevent_array = (*const [1]os.Kevent)(&kev);
+        const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+        _ = try os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null);
+    }
+
+    pub fn bsdRemoveKev(self: *Loop, ident: usize, filter: i16) void {
+        var kev = os.Kevent{
+            .ident = ident,
+            .filter = filter,
+            .flags = os.EV_DELETE,
+            .fflags = 0,
+            .data = 0,
+            .udata = 0,
+        };
+        const kevent_array = (*const [1]os.Kevent)(&kev);
+        const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+        _ = os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null) catch undefined;
+        self.finishOneEvent();
+    }
+
+    fn dispatch(self: *Loop) void {
+        while (self.available_eventfd_resume_nodes.pop()) |resume_stack_node| {
+            const next_tick_node = self.next_tick_queue.get() orelse {
+                self.available_eventfd_resume_nodes.push(resume_stack_node);
+                return;
+            };
+            const eventfd_node = &resume_stack_node.data;
+            eventfd_node.base.handle = next_tick_node.data;
+            switch (builtin.os) {
+                .macosx, .freebsd, .netbsd => {
+                    const kevent_array = (*const [1]os.Kevent)(&eventfd_node.kevent);
+                    const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+                    _ = os.kevent(self.os_data.kqfd, kevent_array, empty_kevs, null) catch {
+                        self.next_tick_queue.unget(next_tick_node);
+                        self.available_eventfd_resume_nodes.push(resume_stack_node);
+                        return;
+                    };
+                },
+                .linux => {
+                    // the pending count is already accounted for
+                    const epoll_events = os.EPOLLONESHOT | os.linux.EPOLLIN | os.linux.EPOLLOUT |
+                        os.linux.EPOLLET;
+                    self.linuxModFd(
+                        eventfd_node.eventfd,
+                        eventfd_node.epoll_op,
+                        epoll_events,
+                        &eventfd_node.base,
+                    ) catch {
+                        self.next_tick_queue.unget(next_tick_node);
+                        self.available_eventfd_resume_nodes.push(resume_stack_node);
+                        return;
+                    };
+                },
+                .windows => {
+                    windows.PostQueuedCompletionStatus(
+                        self.os_data.io_port,
+                        undefined,
+                        undefined,
+                        &eventfd_node.base.overlapped,
+                    ) catch {
+                        self.next_tick_queue.unget(next_tick_node);
+                        self.available_eventfd_resume_nodes.push(resume_stack_node);
+                        return;
+                    };
+                },
+                else => @compileError("unsupported OS"),
+            }
+        }
+    }
+
+    /// Bring your own linked list node. This means it can't fail.
+    pub fn onNextTick(self: *Loop, node: *NextTickNode) void {
+        self.beginOneEvent(); // finished in dispatch()
+        self.next_tick_queue.put(node);
+        self.dispatch();
+    }
+
+    pub fn cancelOnNextTick(self: *Loop, node: *NextTickNode) void {
+        if (self.next_tick_queue.remove(node)) {
+            self.finishOneEvent();
+        }
+    }
+
+    pub fn run(self: *Loop) void {
+        self.finishOneEvent(); // the reference we start with
+
+        self.workerRun();
+
+        switch (builtin.os) {
+            .linux,
+            .macosx,
+            .freebsd,
+            .netbsd,
+            => self.os_data.fs_thread.wait(),
+            else => {},
+        }
+
+        for (self.extra_threads) |extra_thread| {
+            extra_thread.wait();
+        }
+    }
+
+    /// This is equivalent to function call, except it calls `startCpuBoundOperation` first.
+    pub fn call(comptime func: var, args: ...) @typeOf(func).ReturnType {
+        startCpuBoundOperation();
+        return func(args);
+    }
+
+    /// Yielding lets the event loop run, starting any unstarted async operations.
+    /// Note that async operations automatically start when a function yields for any other reason,
+    /// for example, when async I/O is performed. This function is intended to be used only when
+    /// CPU bound tasks would be waiting in the event loop but never get started because no async I/O
+    /// is performed.
+    pub fn yield(self: *Loop) void {
+        suspend {
+            var my_tick_node = NextTickNode{
+                .prev = undefined,
+                .next = undefined,
+                .data = @frame(),
+            };
+            self.onNextTick(&my_tick_node);
+        }
+    }
+
+    /// If the build is multi-threaded and there is an event loop, then it calls `yield`. Otherwise,
+    /// does nothing.
+    pub fn startCpuBoundOperation() void {
+        if (builtin.single_threaded) {
+            return;
+        } else if (instance) |event_loop| {
+            event_loop.yield();
+        }
+    }
+
+    /// call finishOneEvent when done
+    pub fn beginOneEvent(self: *Loop) void {
+        _ = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Add, 1, AtomicOrder.SeqCst);
+    }
+
+    pub fn finishOneEvent(self: *Loop) void {
+        const prev = @atomicRmw(usize, &self.pending_event_count, AtomicRmwOp.Sub, 1, AtomicOrder.SeqCst);
+        if (prev == 1) {
+            // cause all the threads to stop
+            switch (builtin.os) {
+                .linux => {
+                    self.posixFsRequest(&self.os_data.fs_end_request);
+                    // writing 8 bytes to an eventfd cannot fail
+                    os.write(self.os_data.final_eventfd, wakeup_bytes) catch unreachable;
+                    return;
+                },
+                .macosx, .freebsd, .netbsd => {
+                    self.posixFsRequest(&self.os_data.fs_end_request);
+                    const final_kevent = (*const [1]os.Kevent)(&self.os_data.final_kevent);
+                    const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+                    // cannot fail because we already added it and this just enables it
+                    _ = os.kevent(self.os_data.kqfd, final_kevent, empty_kevs, null) catch unreachable;
+                    return;
+                },
+                .windows => {
+                    var i: usize = 0;
+                    while (i < self.extra_threads.len + 1) : (i += 1) {
+                        while (true) {
+                            const overlapped = &self.final_resume_node.overlapped;
+                            windows.PostQueuedCompletionStatus(self.os_data.io_port, undefined, undefined, overlapped) catch continue;
+                            break;
+                        }
+                    }
+                    return;
+                },
+                else => @compileError("unsupported OS"),
+            }
+        }
+    }
+
+    fn workerRun(self: *Loop) void {
+        while (true) {
+            while (true) {
+                const next_tick_node = self.next_tick_queue.get() orelse break;
+                self.dispatch();
+                resume next_tick_node.data;
+                self.finishOneEvent();
+            }
+
+            switch (builtin.os) {
+                .linux => {
+                    // only process 1 event so we don't steal from other threads
+                    var events: [1]os.linux.epoll_event = undefined;
+                    const count = os.epoll_wait(self.os_data.epollfd, events[0..], -1);
+                    for (events[0..count]) |ev| {
+                        const resume_node = @intToPtr(*ResumeNode, ev.data.ptr);
+                        const handle = resume_node.handle;
+                        const resume_node_id = resume_node.id;
+                        switch (resume_node_id) {
+                            .Basic => {},
+                            .Stop => return,
+                            .EventFd => {
+                                const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
+                                event_fd_node.epoll_op = os.EPOLL_CTL_MOD;
+                                const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
+                                self.available_eventfd_resume_nodes.push(stack_node);
+                            },
+                        }
+                        resume handle;
+                        if (resume_node_id == ResumeNode.Id.EventFd) {
+                            self.finishOneEvent();
+                        }
+                    }
+                },
+                .macosx, .freebsd, .netbsd => {
+                    var eventlist: [1]os.Kevent = undefined;
+                    const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+                    const count = os.kevent(self.os_data.kqfd, empty_kevs, eventlist[0..], null) catch unreachable;
+                    for (eventlist[0..count]) |ev| {
+                        const resume_node = @intToPtr(*ResumeNode, ev.udata);
+                        const handle = resume_node.handle;
+                        const resume_node_id = resume_node.id;
+                        switch (resume_node_id) {
+                            .Basic => {
+                                const basic_node = @fieldParentPtr(ResumeNode.Basic, "base", resume_node);
+                                basic_node.kev = ev;
+                            },
+                            .Stop => return,
+                            .EventFd => {
+                                const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
+                                const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
+                                self.available_eventfd_resume_nodes.push(stack_node);
+                            },
+                        }
+                        resume handle;
+                        if (resume_node_id == ResumeNode.Id.EventFd) {
+                            self.finishOneEvent();
+                        }
+                    }
+                },
+                .windows => {
+                    var completion_key: usize = undefined;
+                    const overlapped = while (true) {
+                        var nbytes: windows.DWORD = undefined;
+                        var overlapped: ?*windows.OVERLAPPED = undefined;
+                        switch (windows.GetQueuedCompletionStatus(self.os_data.io_port, &nbytes, &completion_key, &overlapped, windows.INFINITE)) {
+                            .Aborted => return,
+                            .Normal => {},
+                            .EOF => {},
+                            .Cancelled => continue,
+                        }
+                        if (overlapped) |o| break o;
+                    } else unreachable; // TODO else unreachable should not be necessary
+                    const resume_node = @fieldParentPtr(ResumeNode, "overlapped", overlapped);
+                    const handle = resume_node.handle;
+                    const resume_node_id = resume_node.id;
+                    switch (resume_node_id) {
+                        .Basic => {},
+                        .Stop => return,
+                        .EventFd => {
+                            const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
+                            const stack_node = @fieldParentPtr(std.atomic.Stack(ResumeNode.EventFd).Node, "data", event_fd_node);
+                            self.available_eventfd_resume_nodes.push(stack_node);
+                        },
+                    }
+                    resume handle;
+                    self.finishOneEvent();
+                },
+                else => @compileError("unsupported OS"),
+            }
+        }
+    }
+
+    fn posixFsRequest(self: *Loop, request_node: *fs.RequestNode) void {
+        self.beginOneEvent(); // finished in posixFsRun after processing the msg
+        self.os_data.fs_queue.put(request_node);
+        switch (builtin.os) {
+            .macosx, .freebsd, .netbsd => {
+                const fs_kevs = (*const [1]os.Kevent)(&self.os_data.fs_kevent_wake);
+                const empty_kevs = ([*]os.Kevent)(undefined)[0..0];
+                _ = os.kevent(self.os_data.fs_kqfd, fs_kevs, empty_kevs, null) catch unreachable;
+            },
+            .linux => {
+                _ = @atomicRmw(i32, &self.os_data.fs_queue_item, AtomicRmwOp.Xchg, 1, AtomicOrder.SeqCst);
+                const rc = os.linux.futex_wake(&self.os_data.fs_queue_item, os.linux.FUTEX_WAKE, 1);
+                switch (os.linux.getErrno(rc)) {
+                    0 => {},
+                    os.EINVAL => unreachable,
+                    else => unreachable,
+                }
+            },
+            else => @compileError("Unsupported OS"),
+        }
+    }
+
+    fn posixFsCancel(self: *Loop, request_node: *fs.RequestNode) void {
+        if (self.os_data.fs_queue.remove(request_node)) {
+            self.finishOneEvent();
+        }
+    }
+
+    fn posixFsRun(self: *Loop) void {
+        while (true) {
+            if (builtin.os == .linux) {
+                _ = @atomicRmw(i32, &self.os_data.fs_queue_item, .Xchg, 0, .SeqCst);
+            }
+            while (self.os_data.fs_queue.get()) |node| {
+                switch (node.data.msg) {
+                    .End => return,
+                    .WriteV => |*msg| {
+                        msg.result = os.writev(msg.fd, msg.iov);
+                    },
+                    .PWriteV => |*msg| {
+                        msg.result = os.pwritev(msg.fd, msg.iov, msg.offset);
+                    },
+                    .PReadV => |*msg| {
+                        msg.result = os.preadv(msg.fd, msg.iov, msg.offset);
+                    },
+                    .Open => |*msg| {
+                        msg.result = os.openC(msg.path.ptr, msg.flags, msg.mode);
+                    },
+                    .Close => |*msg| os.close(msg.fd),
+                    .WriteFile => |*msg| blk: {
+                        const flags = os.O_LARGEFILE | os.O_WRONLY | os.O_CREAT |
+                            os.O_CLOEXEC | os.O_TRUNC;
+                        const fd = os.openC(msg.path.ptr, flags, msg.mode) catch |err| {
+                            msg.result = err;
+                            break :blk;
+                        };
+                        defer os.close(fd);
+                        msg.result = os.write(fd, msg.contents);
+                    },
+                }
+                switch (node.data.finish) {
+                    .TickNode => |*tick_node| self.onNextTick(tick_node),
+                    .DeallocCloseOperation => |close_op| {
+                        self.allocator.destroy(close_op);
+                    },
+                    .NoAction => {},
+                }
+                self.finishOneEvent();
+            }
+            switch (builtin.os) {
+                .linux => {
+                    const rc = os.linux.futex_wait(&self.os_data.fs_queue_item, os.linux.FUTEX_WAIT, 0, null);
+                    switch (os.linux.getErrno(rc)) {
+                        0, os.EINTR, os.EAGAIN => continue,
+                        else => unreachable,
+                    }
+                },
+                .macosx, .freebsd, .netbsd => {
+                    const fs_kevs = (*const [1]os.Kevent)(&self.os_data.fs_kevent_wait);
+                    var out_kevs: [1]os.Kevent = undefined;
+                    _ = os.kevent(self.os_data.fs_kqfd, fs_kevs, out_kevs[0..], null) catch unreachable;
+                },
+                else => @compileError("Unsupported OS"),
+            }
+        }
+    }
+
+    const OsData = switch (builtin.os) {
+        .linux => LinuxOsData,
+        .macosx, .freebsd, .netbsd => KEventData,
+        .windows => struct {
+            io_port: windows.HANDLE,
+            extra_thread_count: usize,
+        },
+        else => struct {},
+    };
+
+    const KEventData = struct {
+        kqfd: i32,
+        final_kevent: os.Kevent,
+        fs_kevent_wake: os.Kevent,
+        fs_kevent_wait: os.Kevent,
+        fs_thread: *Thread,
+        fs_kqfd: i32,
+        fs_queue: std.atomic.Queue(fs.Request),
+        fs_end_request: fs.RequestNode,
+    };
+
+    const LinuxOsData = struct {
+        epollfd: i32,
+        final_eventfd: i32,
+        final_eventfd_event: os.linux.epoll_event,
+        fs_thread: *Thread,
+        fs_queue_item: i32,
+        fs_queue: std.atomic.Queue(fs.Request),
+        fs_end_request: fs.RequestNode,
+    };
+};
+
+test "std.event.Loop - basic" {
+    // https://github.com/ziglang/zig/issues/1908
+    if (builtin.single_threaded) return error.SkipZigTest;
+
+    const allocator = std.heap.direct_allocator;
+
+    var loop: Loop = undefined;
+    try loop.initMultiThreaded(allocator);
+    defer loop.deinit();
+
+    loop.run();
+}
+
+test "std.event.Loop - call" {
+    // https://github.com/ziglang/zig/issues/1908
+    if (builtin.single_threaded) return error.SkipZigTest;
+
+    const allocator = std.heap.direct_allocator;
+
+    var loop: Loop = undefined;
+    try loop.initMultiThreaded(allocator);
+    defer loop.deinit();
+
+    var did_it = false;
+    var handle = async Loop.call(testEventLoop);
+    var handle2 = async Loop.call(testEventLoop2, &handle, &did_it);
+
+    loop.run();
+
+    testing.expect(did_it);
+}
+
+async fn testEventLoop() i32 {
+    return 1234;
+}
+
+async fn testEventLoop2(h: anyframe->i32, did_it: *bool) void {
+    const value = await h;
+    testing.expect(value == 1234);
+    did_it.* = true;
+}