zig/lib/std /
event/loop.zig
const std = @import("../std.zig");
const builtin = @import("builtin");
const assert = std.debug.assert;
const testing = std.testing;
const mem = std.mem;
const os = std.os;
const windows = os.windows;
const maxInt = std.math.maxInt;
const Thread = std.Thread;
const Atomic = std.atomic.Atomic;
const is_windows = builtin.os.tag == .windows;
Loop
pub const Loop = struct {
next_tick_queue: std.atomic.Queue(anyframe),
os_data: OsData,
final_resume_node: ResumeNode,
pending_event_count: usize,
extra_threads: []Thread,
fs_thread: Thread,
fs_queue: std.atomic.Queue(Request),
fs_end_request: Request.Node,
fs_thread_wakeup: std.Thread.ResetEvent,
arena: std.heap.ArenaAllocator,
delay_queue: DelayQueue,
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.tag) {
.windows => windows.OVERLAPPED{
.Internal = 0,
.InternalHigh = 0,
.DUMMYUNIONNAME = .{
.DUMMYSTRUCTNAME = .{
.Offset = 0,
.OffsetHigh = 0,
},
},
.hEvent = null,
},
else => {},
};
pub const Overlapped = @TypeOf(overlapped_init);
pub const Id = enum {
basic,
stop,
event_fd,
};
pub const EventFd = switch (builtin.os.tag) {
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventFd,
.linux => struct {
base: ResumeNode,
epoll_op: u32,
eventfd: i32,
},
.windows => struct {
base: ResumeNode,
completion_key: usize,
},
else => struct {},
};
const KEventFd = struct {
base: ResumeNode,
kevent: os.Kevent,
};
pub const Basic = switch (builtin.os.tag) {
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventBasic,
.linux => struct {
base: ResumeNode,
},
.windows => struct {
base: ResumeNode,
},
else => @compileError("unsupported OS"),
};
const KEventBasic = struct {
base: ResumeNode,
kev: os.Kevent,
};
};
pub const Instance = switch (std.options.io_mode) {
.blocking => @TypeOf(null),
.evented => ?*Loop,
};
pub const instance = std.options.event_loop;
var global_instance_state: Loop = undefined;
pub const default_instance = switch (std.options.io_mode) {
.blocking => null,
.evented => &global_instance_state,
};
pub const Mode = enum {
single_threaded,
multi_threaded,
};
pub const default_mode = .multi_threaded;
init()
TODO change this to a pool of configurable number of threads and rename it to be not file-system-specific. it will become a thread pool for turning non-CPU-bound blocking things into async things. A fallback for any missing OS-specific API. For resources that have the same lifetime as the Loop. This is only used by Loop for the thread pool and associated resources. State which manages frames that are sleeping on timers Pre-allocated eventfds. All permanently active. This is how Loop sends promises to be resumed on other threads. 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) !void {
if (builtin.single_threaded or std.options.event_loop_mode == .single_threaded) {
return self.initSingleThreaded();
} else {
return self.initMultiThreaded();
}
}
initSingleThreaded()
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) !void {
return self.initThreadPool(1);
}
initMultiThreaded()
After initialization, call run(). This is the same as initThreadPool using Thread.getCpuCount to determine the thread pool size. 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) !void {
if (builtin.single_threaded)
@compileError("initMultiThreaded unavailable when building in single-threaded mode");
const core_count = try Thread.getCpuCount();
return self.initThreadPool(core_count);
}
initThreadPool()
Thread count is the total thread count. The thread pool size will be max(thread_count - 1, 0)
pub fn initThreadPool(self: *Loop, thread_count: usize) !void {
self.* = Loop{
.arena = std.heap.ArenaAllocator.init(std.heap.page_allocator),
.pending_event_count = 1,
.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 = .stop,
.handle = undefined,
.overlapped = ResumeNode.overlapped_init,
},
.fs_end_request = .{ .data = .{ .msg = .end, .finish = .no_action } },
.fs_queue = std.atomic.Queue(Request).init(),
.fs_thread = undefined,
.fs_thread_wakeup = .{},
.delay_queue = undefined,
};
errdefer self.arena.deinit();
// 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 = @max(extra_thread_count, 1);
self.eventfd_resume_nodes = try self.arena.allocator().alloc(
std.atomic.Stack(ResumeNode.EventFd).Node,
resume_node_count,
);
self.extra_threads = try self.arena.allocator().alloc(Thread, extra_thread_count);
try self.initOsData(extra_thread_count);
errdefer self.deinitOsData();
if (!builtin.single_threaded) {
self.fs_thread = try Thread.spawn(.{}, posixFsRun, .{self});
}
errdefer if (!builtin.single_threaded) {
self.posixFsRequest(&self.fs_end_request);
self.fs_thread.join();
};
if (!builtin.single_threaded)
try self.delay_queue.init();
}
deinit()
pub fn deinit(self: *Loop) void {
self.deinitOsData();
self.arena.deinit();
self.* = undefined;
}
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 {
nosuspend switch (builtin.os.tag) {
.linux => {
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 = .event_fd,
.handle = undefined,
.overlapped = ResumeNode.overlapped_init,
},
.eventfd = try os.eventfd(1, os.linux.EFD.CLOEXEC | os.linux.EFD.NONBLOCK),
.epoll_op = os.linux.EPOLL.CTL_ADD,
},
.next = undefined,
};
self.available_eventfd_resume_nodes.push(eventfd_node);
}
self.os_data.epollfd = try os.epoll_create1(os.linux.EPOLL.CLOEXEC);
errdefer os.close(self.os_data.epollfd);
self.os_data.final_eventfd = try os.eventfd(0, os.linux.EFD.CLOEXEC | os.linux.EFD.NONBLOCK);
errdefer os.close(self.os_data.final_eventfd);
self.os_data.final_eventfd_event = os.linux.epoll_event{
.events = os.linux.EPOLL.IN,
.data = os.linux.epoll_data{ .ptr = @intFromPtr(&self.final_resume_node) },
};
try os.epoll_ctl(
self.os_data.epollfd,
os.linux.EPOLL.CTL_ADD,
self.os_data.final_eventfd,
&self.os_data.final_eventfd_event,
);
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
const amt = os.write(self.os_data.final_eventfd, &wakeup_bytes) catch unreachable;
assert(amt == wakeup_bytes.len);
while (extra_thread_index != 0) {
extra_thread_index -= 1;
self.extra_threads[extra_thread_index].join();
}
}
while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
}
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly => {
self.os_data.kqfd = try os.kqueue();
errdefer os.close(self.os_data.kqfd);
const empty_kevs = &[0]os.Kevent{};
for (self.eventfd_resume_nodes, 0..) |*eventfd_node, i| {
eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
.data = ResumeNode.EventFd{
.base = ResumeNode{
.id = .event_fd,
.handle = undefined,
.overlapped = ResumeNode.overlapped_init,
},
// this one is for sending events
.kevent = os.Kevent{
.ident = i,
.filter = os.system.EVFILT_USER,
.flags = os.system.EV_CLEAR | os.system.EV_ADD | os.system.EV_DISABLE,
.fflags = 0,
.data = 0,
.udata = @intFromPtr(&eventfd_node.data.base),
},
},
.next = undefined,
};
self.available_eventfd_resume_nodes.push(eventfd_node);
const kevent_array = @as(*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.system.EV_CLEAR | os.system.EV_ENABLE;
eventfd_node.data.kevent.fflags = os.system.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.system.EVFILT_USER,
.flags = os.system.EV_ADD | os.system.EV_DISABLE,
.fflags = 0,
.data = 0,
.udata = @intFromPtr(&self.final_resume_node),
};
const final_kev_arr = @as(*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.system.EV_ENABLE;
self.os_data.final_kevent.fflags = os.system.NOTE_TRIGGER;
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].join();
}
}
while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
}
},
.openbsd => {
self.os_data.kqfd = try os.kqueue();
errdefer os.close(self.os_data.kqfd);
const empty_kevs = &[0]os.Kevent{};
for (self.eventfd_resume_nodes, 0..) |*eventfd_node, i| {
eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
.data = ResumeNode.EventFd{
.base = ResumeNode{
.id = .event_fd,
.handle = undefined,
.overlapped = ResumeNode.overlapped_init,
},
// this one is for sending events
.kevent = os.Kevent{
.ident = i,
.filter = os.system.EVFILT_TIMER,
.flags = os.system.EV_CLEAR | os.system.EV_ADD | os.system.EV_DISABLE | os.system.EV_ONESHOT,
.fflags = 0,
.data = 0,
.udata = @intFromPtr(&eventfd_node.data.base),
},
},
.next = undefined,
};
self.available_eventfd_resume_nodes.push(eventfd_node);
const kevent_array = @as(*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.system.EV_CLEAR | os.system.EV_ENABLE;
}
// 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.system.EVFILT_TIMER,
.flags = os.system.EV_ADD | os.system.EV_ONESHOT | os.system.EV_DISABLE,
.fflags = 0,
.data = 0,
.udata = @intFromPtr(&self.final_resume_node),
};
const final_kev_arr = @as(*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.system.EV_ENABLE;
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].join();
}
}
while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
}
},
.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| {
eventfd_node.* = std.atomic.Stack(ResumeNode.EventFd).Node{
.data = ResumeNode.EventFd{
.base = ResumeNode{
.id = .event_fd,
.handle = undefined,
.overlapped = ResumeNode.overlapped_init,
},
// this one is for sending events
.completion_key = @intFromPtr(&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].join();
}
}
while (extra_thread_index < extra_thread_count) : (extra_thread_index += 1) {
self.extra_threads[extra_thread_index] = try Thread.spawn(.{}, workerRun, .{self});
}
},
else => {},
};
}
fn deinitOsData(self: *Loop) void {
nosuspend switch (builtin.os.tag) {
.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);
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
os.close(self.os_data.kqfd);
},
.windows => {
windows.CloseHandle(self.os_data.io_port);
},
else => {},
};
}
linuxAddFd()
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.linux.EPOLL.ET == os.linux.EPOLL.ET);
self.beginOneEvent();
errdefer self.finishOneEvent();
try self.linuxModFd(
fd,
os.linux.EPOLL.CTL_ADD,
flags,
resume_node,
);
}
linuxModFd()
pub fn linuxModFd(self: *Loop, fd: i32, op: u32, flags: u32, resume_node: *ResumeNode) !void {
assert(flags & os.linux.EPOLL.ET == os.linux.EPOLL.ET);
var ev = os.linux.epoll_event{
.events = flags,
.data = os.linux.epoll_data{ .ptr = @intFromPtr(resume_node) },
};
try os.epoll_ctl(self.os_data.epollfd, op, fd, &ev);
}
linuxRemoveFd()
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();
}
linuxWaitFd()
pub fn linuxWaitFd(self: *Loop, fd: i32, flags: u32) void {
assert(flags & os.linux.EPOLL.ET == os.linux.EPOLL.ET);
assert(flags & os.linux.EPOLL.ONESHOT == os.linux.EPOLL.ONESHOT);
var resume_node = ResumeNode.Basic{
.base = ResumeNode{
.id = .basic,
.handle = @frame(),
.overlapped = ResumeNode.overlapped_init,
},
};
var need_to_delete = true;
defer if (need_to_delete) self.linuxRemoveFd(fd);
suspend {
self.linuxAddFd(fd, &resume_node.base, flags) catch |err| switch (err) {
error.FileDescriptorNotRegistered => unreachable,
error.OperationCausesCircularLoop => unreachable,
error.FileDescriptorIncompatibleWithEpoll => unreachable,
error.FileDescriptorAlreadyPresentInSet => unreachable, // evented writes to the same fd is not thread-safe
error.SystemResources,
error.UserResourceLimitReached,
error.Unexpected,
=> {
need_to_delete = false;
// Fall back to a blocking poll(). Ideally this codepath is never hit, since
// epoll should be just fine. But this is better than incorrect behavior.
var poll_flags: i16 = 0;
if ((flags & os.linux.EPOLL.IN) != 0) poll_flags |= os.POLL.IN;
if ((flags & os.linux.EPOLL.OUT) != 0) poll_flags |= os.POLL.OUT;
var pfd = [1]os.pollfd{os.pollfd{
.fd = fd,
.events = poll_flags,
.revents = undefined,
}};
_ = os.poll(&pfd, -1) catch |poll_err| switch (poll_err) {
error.NetworkSubsystemFailed => unreachable, // only possible on windows
error.SystemResources,
error.Unexpected,
=> {
// Even poll() didn't work. The best we can do now is sleep for a
// small duration and then hope that something changed.
std.time.sleep(1 * std.time.ns_per_ms);
},
};
resume @frame();
},
};
}
}
waitUntilFdReadable()
pub fn waitUntilFdReadable(self: *Loop, fd: os.fd_t) void {
switch (builtin.os.tag) {
.linux => {
self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.IN);
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_READ, os.system.EV_ONESHOT);
},
else => @compileError("Unsupported OS"),
}
}
waitUntilFdWritable()
pub fn waitUntilFdWritable(self: *Loop, fd: os.fd_t) void {
switch (builtin.os.tag) {
.linux => {
self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.OUT);
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_WRITE, os.system.EV_ONESHOT);
},
else => @compileError("Unsupported OS"),
}
}
waitUntilFdWritableOrReadable()
pub fn waitUntilFdWritableOrReadable(self: *Loop, fd: os.fd_t) void {
switch (builtin.os.tag) {
.linux => {
self.linuxWaitFd(fd, os.linux.EPOLL.ET | os.linux.EPOLL.ONESHOT | os.linux.EPOLL.OUT | os.linux.EPOLL.IN);
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_READ, os.system.EV_ONESHOT);
self.bsdWaitKev(@as(usize, @intCast(fd)), os.system.EVFILT_WRITE, os.system.EV_ONESHOT);
},
else => @compileError("Unsupported OS"),
}
}
bsdWaitKev()
pub fn bsdWaitKev(self: *Loop, ident: usize, filter: i16, flags: u16) void {
var resume_node = ResumeNode.Basic{
.base = ResumeNode{
.id = .basic,
.handle = @frame(),
.overlapped = ResumeNode.overlapped_init,
},
.kev = undefined,
};
defer {
// If the kevent was set to be ONESHOT, it doesn't need to be deleted manually.
if (flags & os.system.EV_ONESHOT != 0) {
self.bsdRemoveKev(ident, filter);
}
}
suspend {
self.bsdAddKev(&resume_node, ident, filter, flags) catch unreachable;
}
}
bsdAddKev()
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, flags: u16) !void {
self.beginOneEvent();
errdefer self.finishOneEvent();
var kev = [1]os.Kevent{os.Kevent{
.ident = ident,
.filter = filter,
.flags = os.system.EV_ADD | os.system.EV_ENABLE | os.system.EV_CLEAR | flags,
.fflags = 0,
.data = 0,
.udata = @intFromPtr(&resume_node.base),
}};
const empty_kevs = &[0]os.Kevent{};
_ = try os.kevent(self.os_data.kqfd, &kev, empty_kevs, null);
}
bsdRemoveKev()
pub fn bsdRemoveKev(self: *Loop, ident: usize, filter: i16) void {
var kev = [1]os.Kevent{os.Kevent{
.ident = ident,
.filter = filter,
.flags = os.system.EV_DELETE,
.fflags = 0,
.data = 0,
.udata = 0,
}};
const empty_kevs = &[0]os.Kevent{};
_ = os.kevent(self.os_data.kqfd, &kev, 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.tag) {
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
const kevent_array = @as(*const [1]os.Kevent, &eventfd_node.kevent);
const empty_kevs = &[0]os.Kevent{};
_ = 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.linux.EPOLL.ONESHOT | os.linux.EPOLL.IN | os.linux.EPOLL.OUT |
os.linux.EPOLL.ET;
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"),
}
}
}
onNextTick()
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();
}
cancelOnNextTick()
pub fn cancelOnNextTick(self: *Loop, node: *NextTickNode) void {
if (self.next_tick_queue.remove(node)) {
self.finishOneEvent();
}
}
run()
pub fn run(self: *Loop) void {
self.finishOneEvent(); // the reference we start with
self.workerRun();
if (!builtin.single_threaded) {
switch (builtin.os.tag) {
.linux,
.macos,
.ios,
.tvos,
.watchos,
.freebsd,
.netbsd,
.dragonfly,
.openbsd,
=> self.fs_thread.join(),
else => {},
}
}
for (self.extra_threads) |extra_thread| {
extra_thread.join();
}
self.delay_queue.deinit();
}
runDetached()
Runs the provided function asynchronously. The function's frame is allocated with allocator and freed when the function returns. func must return void and it can be an async function. Yields to the event loop, running the function on the next tick.
pub fn runDetached(self: *Loop, alloc: mem.Allocator, comptime func: anytype, args: anytype) error{OutOfMemory}!void {
if (!std.io.is_async) @compileError("Can't use runDetached in non-async mode!");
if (@TypeOf(@call(.{}, func, args)) != void) {
@compileError("`func` must not have a return value");
}
const Wrapper = struct {
const Args = @TypeOf(args);
fn run(func_args: Args, loop: *Loop, allocator: mem.Allocator) void {
loop.beginOneEvent();
loop.yield();
@call(.{}, func, func_args); // compile error when called with non-void ret type
suspend {
loop.finishOneEvent();
allocator.destroy(@frame());
}
}
};
var run_frame = try alloc.create(@Frame(Wrapper.run));
run_frame.* = async Wrapper.run(args, self, alloc);
}
yield()
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);
}
}
startCpuBoundOperation()
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();
}
}
beginOneEvent()
call finishOneEvent when done
pub fn beginOneEvent(self: *Loop) void {
_ = @atomicRmw(usize, &self.pending_event_count, .Add, 1, .SeqCst);
}
finishOneEvent()
pub fn finishOneEvent(self: *Loop) void {
nosuspend {
const prev = @atomicRmw(usize, &self.pending_event_count, .Sub, 1, .SeqCst);
if (prev != 1) return;
// cause all the threads to stop
self.posixFsRequest(&self.fs_end_request);
switch (builtin.os.tag) {
.linux => {
// writing to the eventfd will only wake up one thread, thus multiple writes
// are needed to wakeup all the threads
var i: usize = 0;
while (i < self.extra_threads.len + 1) : (i += 1) {
// writing 8 bytes to an eventfd cannot fail
const amt = os.write(self.os_data.final_eventfd, &wakeup_bytes) catch unreachable;
assert(amt == wakeup_bytes.len);
}
return;
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
const final_kevent = @as(*const [1]os.Kevent, &self.os_data.final_kevent);
const empty_kevs = &[0]os.Kevent{};
// 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"),
}
}
}
sleep()
pub fn sleep(self: *Loop, nanoseconds: u64) void {
if (builtin.single_threaded)
@compileError("TODO: integrate timers with epoll/kevent/iocp for single-threaded");
suspend {
const now = self.delay_queue.timer.read();
var entry: DelayQueue.Waiters.Entry = undefined;
entry.init(@frame(), now + nanoseconds);
self.delay_queue.waiters.insert(&entry);
// Speculatively wake up the timer thread when we add a new entry.
// If the timer thread is sleeping on a longer entry, we need to
// interrupt it so that our entry can be expired in time.
self.delay_queue.event.set();
}
}
const DelayQueue = struct {
timer: std.time.Timer,
waiters: Waiters,
thread: std.Thread,
event: std.Thread.ResetEvent,
is_running: Atomic(bool),
fn init(self: *DelayQueue) !void {
self.* = DelayQueue{
.timer = try std.time.Timer.start(),
.waiters = DelayQueue.Waiters{
.entries = std.atomic.Queue(anyframe).init(),
},
.thread = undefined,
.event = .{},
.is_running = Atomic(bool).init(true),
};
// Must be after init so that it can read the other state, such as `is_running`.
self.thread = try std.Thread.spawn(.{}, DelayQueue.run, .{self});
}
fn deinit(self: *DelayQueue) void {
self.is_running.store(false, .SeqCst);
self.event.set();
self.thread.join();
}
fn run(self: *DelayQueue) void {
const loop = @fieldParentPtr(Loop, "delay_queue", self);
while (self.is_running.load(.SeqCst)) {
self.event.reset();
const now = self.timer.read();
if (self.waiters.popExpired(now)) |entry| {
loop.onNextTick(&entry.node);
continue;
}
if (self.waiters.nextExpire()) |expires| {
if (now >= expires)
continue;
self.event.timedWait(expires - now) catch {};
} else {
self.event.wait();
}
}
}
// TODO: use a tickless hierarchical timer wheel:
// https://github.com/wahern/timeout/
const Waiters = struct {
entries: std.atomic.Queue(anyframe),
const Entry = struct {
node: NextTickNode,
expires: u64,
fn init(self: *Entry, frame: anyframe, expires: u64) void {
self.node.data = frame;
self.expires = expires;
}
};
fn insert(self: *Waiters, entry: *Entry) void {
self.entries.put(&entry.node);
}
fn popExpired(self: *Waiters, now: u64) ?*Entry {
const entry = self.peekExpiringEntry() orelse return null;
if (entry.expires > now)
return null;
assert(self.entries.remove(&entry.node));
return entry;
}
fn nextExpire(self: *Waiters) ?u64 {
const entry = self.peekExpiringEntry() orelse return null;
return entry.expires;
}
fn peekExpiringEntry(self: *Waiters) ?*Entry {
self.entries.mutex.lock();
defer self.entries.mutex.unlock();
// starting from the head
var head = self.entries.head orelse return null;
// traverse the list of waiting entries to
// find the Node with the smallest `expires` field
var min = head;
while (head.next) |node| {
const minEntry = @fieldParentPtr(Entry, "node", min);
const nodeEntry = @fieldParentPtr(Entry, "node", node);
if (nodeEntry.expires < minEntry.expires)
min = node;
head = node;
}
return @fieldParentPtr(Entry, "node", min);
}
};
};
accept()
Initialize the delay queue by spawning the timer thread and starting any timer resources. Entry point for the timer thread which waits for timer entries to expire and reschedules them. Registers the entry into the queue of waiting frames Dequeues one expired event relative to now Returns an estimate for the amount of time to wait until the next waiting entry expires. ------- I/0 APIs -------
pub fn accept(
self: *Loop,
sockfd: os.socket_t,
addr: *os.sockaddr,
addr_size: *os.socklen_t,
flags: u32,
) os.AcceptError!os.socket_t {
while (true) {
return os.accept(sockfd, addr, addr_size, flags | os.SOCK.NONBLOCK) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(sockfd);
continue;
},
else => return err,
};
}
}
connect()
This argument is a socket that has been created with socket, bound to a local address with bind, and is listening for connections after a listen. This argument is a pointer to a sockaddr structure. This structure is filled in with the address of the peer socket, as known to the communications layer. The exact format of the address returned addr is determined by the socket's address family (see socket and the respective protocol man pages). This argument is a value-result argument: the caller must initialize it to contain the size (in bytes) of the structure pointed to by addr; on return it will contain the actual size of the peer address.
The returned address is truncated if the buffer provided is too small; in this case, addr_size will return a value greater than was supplied to the call. The following values can be bitwise ORed in flags to obtain different behavior: * SOCK.CLOEXEC - Set the close-on-exec (FD_CLOEXEC) flag on the new file descriptor. See the description of the O.CLOEXEC flag in open for reasons why this may be useful.
pub fn connect(self: *Loop, sockfd: os.socket_t, sock_addr: *const os.sockaddr, len: os.socklen_t) os.ConnectError!void {
os.connect(sockfd, sock_addr, len) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(sockfd);
return os.getsockoptError(sockfd);
},
else => return err,
};
}
openZ()
Performs an async os.open using a separate thread.
pub fn openZ(self: *Loop, file_path: [*:0]const u8, flags: u32, mode: os.mode_t) os.OpenError!os.fd_t {
var req_node = Request.Node{
.data = .{
.msg = .{
.open = .{
.path = file_path,
.flags = flags,
.mode = mode,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.open.result;
}
openatZ()
Performs an async os.opent using a separate thread.
pub fn openatZ(self: *Loop, fd: os.fd_t, file_path: [*:0]const u8, flags: u32, mode: os.mode_t) os.OpenError!os.fd_t {
var req_node = Request.Node{
.data = .{
.msg = .{
.openat = .{
.fd = fd,
.path = file_path,
.flags = flags,
.mode = mode,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.openat.result;
}
close()
Performs an async os.close using a separate thread.
pub fn close(self: *Loop, fd: os.fd_t) void {
var req_node = Request.Node{
.data = .{
.msg = .{ .close = .{ .fd = fd } },
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
}
read()
Performs an async os.read using a separate thread. fd must block and not return EAGAIN.
pub fn read(self: *Loop, fd: os.fd_t, buf: []u8, simulate_evented: bool) os.ReadError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.read = .{
.fd = fd,
.buf = buf,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.read.result;
} else {
while (true) {
return os.read(fd, buf) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(fd);
continue;
},
else => return err,
};
}
}
}
readv()
Performs an async os.readv using a separate thread. fd must block and not return EAGAIN.
pub fn readv(self: *Loop, fd: os.fd_t, iov: []const os.iovec, simulate_evented: bool) os.ReadError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.readv = .{
.fd = fd,
.iov = iov,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.readv.result;
} else {
while (true) {
return os.readv(fd, iov) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(fd);
continue;
},
else => return err,
};
}
}
}
pread()
Performs an async os.pread using a separate thread. fd must block and not return EAGAIN.
pub fn pread(self: *Loop, fd: os.fd_t, buf: []u8, offset: u64, simulate_evented: bool) os.PReadError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.pread = .{
.fd = fd,
.buf = buf,
.offset = offset,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.pread.result;
} else {
while (true) {
return os.pread(fd, buf, offset) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(fd);
continue;
},
else => return err,
};
}
}
}
preadv()
Performs an async os.preadv using a separate thread. fd must block and not return EAGAIN.
pub fn preadv(self: *Loop, fd: os.fd_t, iov: []const os.iovec, offset: u64, simulate_evented: bool) os.ReadError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.preadv = .{
.fd = fd,
.iov = iov,
.offset = offset,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.preadv.result;
} else {
while (true) {
return os.preadv(fd, iov, offset) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(fd);
continue;
},
else => return err,
};
}
}
}
write()
Performs an async os.write using a separate thread. fd must block and not return EAGAIN.
pub fn write(self: *Loop, fd: os.fd_t, bytes: []const u8, simulate_evented: bool) os.WriteError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.write = .{
.fd = fd,
.bytes = bytes,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.write.result;
} else {
while (true) {
return os.write(fd, bytes) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(fd);
continue;
},
else => return err,
};
}
}
}
writev()
Performs an async os.writev using a separate thread. fd must block and not return EAGAIN.
pub fn writev(self: *Loop, fd: os.fd_t, iov: []const os.iovec_const, simulate_evented: bool) os.WriteError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.writev = .{
.fd = fd,
.iov = iov,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.writev.result;
} else {
while (true) {
return os.writev(fd, iov) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(fd);
continue;
},
else => return err,
};
}
}
}
pwrite()
Performs an async os.pwrite using a separate thread. fd must block and not return EAGAIN.
pub fn pwrite(self: *Loop, fd: os.fd_t, bytes: []const u8, offset: u64, simulate_evented: bool) os.PerformsWriteError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.pwrite = .{
.fd = fd,
.bytes = bytes,
.offset = offset,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.pwrite.result;
} else {
while (true) {
return os.pwrite(fd, bytes, offset) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(fd);
continue;
},
else => return err,
};
}
}
}
pwritev()
Performs an async os.pwritev using a separate thread. fd must block and not return EAGAIN.
pub fn pwritev(self: *Loop, fd: os.fd_t, iov: []const os.iovec_const, offset: u64, simulate_evented: bool) os.PWriteError!usize {
if (simulate_evented) {
var req_node = Request.Node{
.data = .{
.msg = .{
.pwritev = .{
.fd = fd,
.iov = iov,
.offset = offset,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.pwritev.result;
} else {
while (true) {
return os.pwritev(fd, iov, offset) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(fd);
continue;
},
else => return err,
};
}
}
}
sendto()
pub fn sendto(
self: *Loop,
sockfd: os.fd_t,
buf: []const u8,
flags: u32,
dest_addr: ?*const os.sockaddr,
addrlen: os.socklen_t,
) os.SendToError!usize {
while (true) {
return os.sendto(sockfd, buf, flags, dest_addr, addrlen) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdWritable(sockfd);
continue;
},
else => return err,
};
}
}
recvfrom()
The file descriptor of the sending socket. Message to send.
pub fn recvfrom(
self: *Loop,
sockfd: os.fd_t,
buf: []u8,
flags: u32,
src_addr: ?*os.sockaddr,
addrlen: ?*os.socklen_t,
) os.RecvFromError!usize {
while (true) {
return os.recvfrom(sockfd, buf, flags, src_addr, addrlen) catch |err| switch (err) {
error.WouldBlock => {
self.waitUntilFdReadable(sockfd);
continue;
},
else => return err,
};
}
}
faccessatZ()
Performs an async os.faccessatZ using a separate thread. fd must block and not return EAGAIN.
pub fn faccessatZ(
self: *Loop,
dirfd: os.fd_t,
path_z: [*:0]const u8,
mode: u32,
flags: u32,
) os.AccessError!void {
var req_node = Request.Node{
.data = .{
.msg = .{
.faccessat = .{
.dirfd = dirfd,
.path = path_z,
.mode = mode,
.flags = flags,
.result = undefined,
},
},
.finish = .{ .tick_node = .{ .data = @frame() } },
},
};
suspend {
self.posixFsRequest(&req_node);
}
return req_node.data.msg.faccessat.result;
}
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.tag) {
.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 = @as(*ResumeNode, @ptrFromInt(ev.data.ptr));
const handle = resume_node.handle;
const resume_node_id = resume_node.id;
switch (resume_node_id) {
.basic => {},
.stop => return,
.event_fd => {
const event_fd_node = @fieldParentPtr(ResumeNode.EventFd, "base", resume_node);
event_fd_node.epoll_op = os.linux.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 == .event_fd) {
self.finishOneEvent();
}
}
},
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => {
var eventlist: [1]os.Kevent = undefined;
const empty_kevs = &[0]os.Kevent{};
const count = os.kevent(self.os_data.kqfd, empty_kevs, eventlist[0..], null) catch unreachable;
for (eventlist[0..count]) |ev| {
const resume_node = @as(*ResumeNode, @ptrFromInt(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,
.event_fd => {
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 == .event_fd) {
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;
};
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,
.event_fd => {
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: *Request.Node) void {
self.beginOneEvent(); // finished in posixFsRun after processing the msg
self.fs_queue.put(request_node);
self.fs_thread_wakeup.set();
}
fn posixFsCancel(self: *Loop, request_node: *Request.Node) void {
if (self.fs_queue.remove(request_node)) {
self.finishOneEvent();
}
}
fn posixFsRun(self: *Loop) void {
nosuspend while (true) {
self.fs_thread_wakeup.reset();
while (self.fs_queue.get()) |node| {
switch (node.data.msg) {
.end => return,
.read => |*msg| {
msg.result = os.read(msg.fd, msg.buf);
},
.readv => |*msg| {
msg.result = os.readv(msg.fd, msg.iov);
},
.write => |*msg| {
msg.result = os.write(msg.fd, msg.bytes);
},
.writev => |*msg| {
msg.result = os.writev(msg.fd, msg.iov);
},
.pwrite => |*msg| {
msg.result = os.pwrite(msg.fd, msg.bytes, msg.offset);
},
.pwritev => |*msg| {
msg.result = os.pwritev(msg.fd, msg.iov, msg.offset);
},
.pread => |*msg| {
msg.result = os.pread(msg.fd, msg.buf, msg.offset);
},
.preadv => |*msg| {
msg.result = os.preadv(msg.fd, msg.iov, msg.offset);
},
.open => |*msg| {
if (is_windows) unreachable; // TODO
msg.result = os.openZ(msg.path, msg.flags, msg.mode);
},
.openat => |*msg| {
if (is_windows) unreachable; // TODO
msg.result = os.openatZ(msg.fd, msg.path, msg.flags, msg.mode);
},
.faccessat => |*msg| {
msg.result = os.faccessatZ(msg.dirfd, msg.path, msg.mode, msg.flags);
},
.close => |*msg| os.close(msg.fd),
}
switch (node.data.finish) {
.tick_node => |*tick_node| self.onNextTick(tick_node),
.no_action => {},
}
self.finishOneEvent();
}
self.fs_thread_wakeup.wait();
};
}
const OsData = switch (builtin.os.tag) {
.linux => LinuxOsData,
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KEventData,
.windows => struct {
io_port: windows.HANDLE,
extra_thread_count: usize,
},
else => struct {},
};
const KEventData = struct {
kqfd: i32,
final_kevent: os.Kevent,
};
const LinuxOsData = struct {
epollfd: i32,
final_eventfd: i32,
final_eventfd_event: os.linux.epoll_event,
};
pub const Request = struct {
msg: Msg,
finish: Finish,
pub const Node = std.atomic.Queue(Request).Node;
pub const Finish = union(enum) {
tick_node: Loop.NextTickNode,
no_action,
};
pub const Msg = union(enum) {
read: Read,
readv: ReadV,
write: Write,
writev: WriteV,
pwrite: PWrite,
pwritev: PWriteV,
pread: PRead,
preadv: PReadV,
open: Open,
openat: OpenAt,
close: Close,
faccessat: FAccessAt,
end,
pub const Read = struct {
fd: os.fd_t,
buf: []u8,
result: Error!usize,
pub const Error = os.ReadError;
};
pub const ReadV = struct {
fd: os.fd_t,
iov: []const os.iovec,
result: Error!usize,
pub const Error = os.ReadError;
};
pub const Write = struct {
fd: os.fd_t,
bytes: []const u8,
result: Error!usize,
pub const Error = os.WriteError;
};
pub const WriteV = struct {
fd: os.fd_t,
iov: []const os.iovec_const,
result: Error!usize,
pub const Error = os.WriteError;
};
pub const PWrite = struct {
fd: os.fd_t,
bytes: []const u8,
offset: usize,
result: Error!usize,
pub const Error = os.PWriteError;
};
pub const PWriteV = struct {
fd: os.fd_t,
iov: []const os.iovec_const,
offset: usize,
result: Error!usize,
pub const Error = os.PWriteError;
};
pub const PRead = struct {
fd: os.fd_t,
buf: []u8,
offset: usize,
result: Error!usize,
pub const Error = os.PReadError;
};
pub const PReadV = struct {
fd: os.fd_t,
iov: []const os.iovec,
offset: usize,
result: Error!usize,
pub const Error = os.PReadError;
};
pub const Open = struct {
path: [*:0]const u8,
flags: u32,
mode: os.mode_t,
result: Error!os.fd_t,
pub const Error = os.OpenError;
};
pub const OpenAt = struct {
fd: os.fd_t,
path: [*:0]const u8,
flags: u32,
mode: os.mode_t,
result: Error!os.fd_t,
pub const Error = os.OpenError;
};
pub const Close = struct {
fd: os.fd_t,
};
pub const FAccessAt = struct {
dirfd: os.fd_t,
path: [*:0]const u8,
mode: u32,
flags: u32,
result: Error!void,
pub const Error = os.AccessError;
};
};
};
};
Test:
std.event.Loop - basic
special - means the fs thread should exit
test "std.event.Loop - basic" {
// https://github.com/ziglang/zig/issues/1908
if (builtin.single_threaded) return error.SkipZigTest;
if (true) {
// https://github.com/ziglang/zig/issues/4922
return error.SkipZigTest;
}
var loop: Loop = undefined;
try loop.initMultiThreaded();
defer loop.deinit();
loop.run();
}
fn testEventLoop() i32 {
return 1234;
}
fn testEventLoop2(h: anyframe->i32, did_it: *bool) void {
const value = await h;
try testing.expect(value == 1234);
did_it.* = true;
}
var testRunDetachedData: usize = 0;
Test:
std.event.Loop - runDetached
test "std.event.Loop - runDetached" {
// https://github.com/ziglang/zig/issues/1908
if (builtin.single_threaded) return error.SkipZigTest;
if (!std.io.is_async) return error.SkipZigTest;
if (true) {
// https://github.com/ziglang/zig/issues/4922
return error.SkipZigTest;
}
var loop: Loop = undefined;
try loop.initMultiThreaded();
defer loop.deinit();
// Schedule the execution, won't actually start until we start the
// event loop.
try loop.runDetached(std.testing.allocator, testRunDetached, .{});
// Now we can start the event loop. The function will return only
// after all tasks have been completed, allowing us to synchronize
// with the previous runDetached.
loop.run();
try testing.expect(testRunDetachedData == 1);
}
fn testRunDetached() void {
testRunDetachedData += 1;
}
Test:
std.event.Loop - sleep
test "std.event.Loop - sleep" {
// https://github.com/ziglang/zig/issues/1908
if (builtin.single_threaded) return error.SkipZigTest;
if (!std.io.is_async) return error.SkipZigTest;
const frames = try testing.allocator.alloc(@Frame(testSleep), 10);
defer testing.allocator.free(frames);
const wait_time = 100 * std.time.ns_per_ms;
var sleep_count: usize = 0;
for (frames) |*frame|
frame.* = async testSleep(wait_time, &sleep_count);
for (frames) |*frame|
await frame;
try testing.expect(sleep_count == frames.len);
}
fn testSleep(wait_ns: u64, sleep_count: *usize) void {
Loop.instance.?.sleep(wait_ns);
_ = @atomicRmw(usize, sleep_count, .Add, 1, .SeqCst);
}