zzstoatzz.io
This repository has no description
1//! platform readiness poller: epoll (linux) / kqueue (darwin, bsd).
2//! level-triggered on both platforms — handlers drain until EAGAIN.
3//! written against zig 0.16.0; syscalls via std.c (libc required).
4const std = @import("std");
5const builtin = @import("builtin");
6const c = std.c;
7
8pub const Interest = struct {
9 read: bool = false,
10 write: bool = false,
11};
12
13pub const Event = struct {
14 token: u64,
15 readable: bool,
16 writable: bool,
17 hup: bool,
18};
19
20pub const Poller = switch (builtin.os.tag) {
21 .linux => Epoll,
22 .macos, .ios, .tvos, .watchos, .visionos, .freebsd, .netbsd, .openbsd, .dragonfly => Kqueue,
23 else => @compileError("zio poller: unsupported os"),
24};
25
26pub const Epoll = struct {
27 epfd: c.fd_t,
28
29 const EPOLL = std.os.linux.EPOLL;
30
31 pub fn init() !Epoll {
32 const epfd = c.epoll_create1(EPOLL.CLOEXEC);
33 if (epfd < 0) return error.PollerInitFailed;
34 return .{ .epfd = epfd };
35 }
36
37 pub fn deinit(p: *Epoll) void {
38 _ = c.close(p.epfd);
39 }
40
41 pub fn add(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
42 var ev: c.epoll_event = .{
43 .events = eventsFrom(interest),
44 .data = .{ .u64 = token },
45 };
46 if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) != 0) return error.PollerAddFailed;
47 }
48
49 pub fn del(p: *Epoll, fd: c.fd_t) void {
50 _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null);
51 }
52
53 pub fn mod(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
54 var ev: c.epoll_event = .{
55 .events = eventsFrom(interest),
56 .data = .{ .u64 = token },
57 };
58 if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) != 0) return error.PollerModFailed;
59 }
60
61 /// arm a single edge: fires once, then disarms until re-armed.
62 pub fn armOneshot(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void {
63 var ev: c.epoll_event = .{
64 .events = eventsFrom(interest) | EPOLL.ONESHOT,
65 .data = .{ .u64 = token },
66 };
67 // fd may or may not be registered; try MOD first (common: re-arm), fall back to ADD.
68 if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) == 0) return;
69 if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) == 0) return;
70 return error.PollerArmFailed;
71 }
72
73 /// fills `out` with ready events; blocks up to timeout_ms. returns slice of out.
74 pub fn wait(p: *Epoll, out: []Event, timeout_ms: i32) []Event {
75 var raw: [max_batch]c.epoll_event = undefined;
76 const cap: c_uint = @intCast(@min(out.len, raw.len));
77 const n = c.epoll_wait(p.epfd, &raw, cap, timeout_ms);
78 if (n <= 0) return out[0..0];
79 const count: usize = @intCast(n);
80 for (raw[0..count], out[0..count]) |r, *e| {
81 e.* = .{
82 .token = r.data.u64,
83 .readable = r.events & EPOLL.IN != 0,
84 .writable = r.events & EPOLL.OUT != 0,
85 .hup = r.events & (EPOLL.HUP | EPOLL.ERR | EPOLL.RDHUP) != 0,
86 };
87 }
88 return out[0..count];
89 }
90
91 fn eventsFrom(interest: Interest) u32 {
92 var events: u32 = EPOLL.RDHUP;
93 if (interest.read) events |= EPOLL.IN;
94 if (interest.write) events |= EPOLL.OUT;
95 return events;
96 }
97};
98
99pub const Kqueue = struct {
100 kq: c.fd_t,
101
102 pub fn init() !Kqueue {
103 const kq = c.kqueue();
104 if (kq < 0) return error.PollerInitFailed;
105 return .{ .kq = kq };
106 }
107
108 pub fn deinit(p: *Kqueue) void {
109 _ = c.close(p.kq);
110 }
111
112 pub fn add(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
113 // both filters are always registered; interest toggles ENABLE/DISABLE.
114 // kqueue EV_ADD on an existing (ident,filter) updates it, so mod == add.
115 const changes = [2]c.Kevent{
116 kev(fd, c.EVFILT.READ, interest.read, token),
117 kev(fd, c.EVFILT.WRITE, interest.write, token),
118 };
119 var none: [0]c.Kevent = .{};
120 const rc = c.kevent(p.kq, &changes, changes.len, &none, 0, null);
121 if (rc < 0) return error.PollerAddFailed;
122 }
123
124 pub fn mod(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
125 return p.add(fd, interest, token) catch error.PollerModFailed;
126 }
127
128 /// arm a single edge: fires once, then disarms until re-armed.
129 pub fn armOneshot(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void {
130 var changes: [1]c.Kevent = undefined;
131 const filter: @FieldType(c.Kevent, "filter") = if (interest.read) c.EVFILT.READ else c.EVFILT.WRITE;
132 changes[0] = .{
133 .ident = @intCast(fd),
134 .filter = filter,
135 .flags = c.EV.ADD | c.EV.ENABLE | c.EV.ONESHOT,
136 .fflags = 0,
137 .data = 0,
138 .udata = token,
139 };
140 var none: [0]c.Kevent = .{};
141 if (c.kevent(p.kq, &changes, changes.len, &none, 0, null) < 0) return error.PollerArmFailed;
142 }
143
144 pub fn del(p: *Kqueue, fd: c.fd_t) void {
145 // closing the fd removes its kevents automatically; explicit EV_DELETE
146 // would race with close-initiated removal, so rely on close semantics.
147 _ = p;
148 _ = fd;
149 }
150
151 pub fn wait(p: *Kqueue, out: []Event, timeout_ms: i32) []Event {
152 var raw: [max_batch]c.Kevent = undefined;
153 const cap: c_int = @intCast(@min(out.len, raw.len));
154 const ts: c.timespec = .{
155 .sec = @divTrunc(timeout_ms, 1000),
156 .nsec = @as(@FieldType(c.timespec, "nsec"), @intCast(@mod(timeout_ms, 1000))) * 1_000_000,
157 };
158 const no_changes: [0]c.Kevent = .{};
159 const n = c.kevent(p.kq, &no_changes, 0, &raw, cap, &ts);
160 if (n <= 0) return out[0..0];
161 const count: usize = @intCast(n);
162 for (raw[0..count], out[0..count]) |r, *e| {
163 e.* = .{
164 .token = r.udata,
165 .readable = r.filter == c.EVFILT.READ,
166 .writable = r.filter == c.EVFILT.WRITE,
167 .hup = r.flags & c.EV.EOF != 0,
168 };
169 }
170 return out[0..count];
171 }
172
173 fn kev(fd: c.fd_t, filter: @FieldType(c.Kevent, "filter"), enabled: bool, token: u64) c.Kevent {
174 return .{
175 .ident = @intCast(fd),
176 .filter = filter,
177 .flags = c.EV.ADD | @as(@FieldType(c.Kevent, "flags"), if (enabled) c.EV.ENABLE else c.EV.DISABLE),
178 .fflags = 0,
179 .data = 0,
180 .udata = token,
181 };
182 }
183};
184
185pub const max_batch = 256;