Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364//! platform readiness poller: epoll (linux) / kqueue (darwin, bsd).//! level-triggered on both platforms — handlers drain until EAGAIN.//! written against zig 0.16.0; syscalls via std.c (libc required).const std = @import("std");const builtin = @import("builtin");const c = std.c;
pub const Interest = struct { read: bool = false, write: bool = false,};
pub const Event = struct { token: u64, readable: bool, writable: bool, hup: bool, /// the fd this readiness is for. Delivery is destructive -- it clears the /// token and drops the registration -- so a caller that cannot consume /// the event needs the fd back in order to re-arm it. fd: c.fd_t = -1,};
pub const Poller = switch (builtin.os.tag) { .linux => Epoll, .macos, .ios, .tvos, .watchos, .visionos, .freebsd, .netbsd, .openbsd, .dragonfly => Kqueue, else => @compileError("zio poller: unsupported os"),};
pub const Epoll = struct { epfd: c.fd_t, /// diagnostics: arming is allowed to fail (fd closed, watch limit) and /// the failure is otherwise invisible from outside the process. arm_failures: u64 = 0, last_arm_errno: i32 = 0, allocator: std.mem.Allocator, /// epoll holds ONE registration per fd, so per-direction one-shot arms /// live here: independent read/write tokens, merged kernel interest. /// without this, a writer fiber arming an fd replaces a parked reader's /// interest and token — a lost wake (proven by the arm-collision test). armed: std.AutoHashMapUnmanaged(c.fd_t, Armed) = .empty,
const Armed = struct { read_token: ?u64 = null, write_token: ?u64 = null }; /// tags kernel data words that refer to the `armed` table (vs literal /// tokens from `add`, e.g. the wake pipe). const TAG: u64 = 1 << 63;
const EPOLL = std.os.linux.EPOLL;
pub fn init(allocator: std.mem.Allocator) !Epoll { const epfd = c.epoll_create1(EPOLL.CLOEXEC); if (epfd < 0) return error.PollerInitFailed; return .{ .epfd = epfd, .allocator = allocator }; }
/// remove `fd`'s registration and return whoever was parked on it. /// called when the fd is about to be closed: the kernel drops closed fds /// from the epoll set silently, so a fiber parked on one would never be /// woken again. pub fn takeWaiters(p: *Epoll, fd: c.fd_t) [2]?u64 { const e = p.armed.fetchRemove(fd) orelse return .{ null, null }; _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null); return .{ e.value.read_token, e.value.write_token }; }
/// number of fds with at least one armed direction (diagnostic). pub fn registrationCount(p: *const Epoll) usize { return p.armed.count(); }
pub fn deinit(p: *Epoll) void { p.armed.deinit(p.allocator); _ = c.close(p.epfd); }
pub fn add(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void { var ev: c.epoll_event = .{ .events = eventsFrom(interest), .data = .{ .u64 = token }, }; if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) != 0) return error.PollerAddFailed; }
pub fn del(p: *Epoll, fd: c.fd_t) void { _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null); _ = p.armed.remove(fd); }
pub fn mod(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void { var ev: c.epoll_event = .{ .events = eventsFrom(interest), .data = .{ .u64 = token }, }; if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) != 0) return error.PollerModFailed; }
/// arm a single direction: fires once for that direction, then disarms /// until re-armed. the other direction's arm (if any) is preserved. pub fn armOneshot(p: *Epoll, fd: c.fd_t, interest: Interest, token: u64) !void { const gop = p.armed.getOrPut(p.allocator, fd) catch return error.PollerArmFailed; if (!gop.found_existing) gop.value_ptr.* = .{}; if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token;
var ev: c.epoll_event = .{ .events = mergedEvents(gop.value_ptr.*), .data = .{ .u64 = TAG | @as(u64, @intCast(@as(u32, @bitCast(fd)))) }, }; if (c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev) == 0) return; // MOD failed: the fd was not registered — it is fresh (or was closed // and the number reused). any other-direction token in the table is // from a previous fd lifetime; drop it so a dead fiber's arm cannot // shadow this registration. gop.value_ptr.* = .{}; if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token; ev.events = mergedEvents(gop.value_ptr.*); if (c.epoll_ctl(p.epfd, EPOLL.CTL_ADD, fd, &ev) == 0) return; // both MOD and ADD failed. MOD=ENOENT means "not registered"; ADD // failing then is usually ENOSPC/ENOMEM — the epoll watch limit. // record it: a silent arm failure turns a parked fiber into a spin, // and a spinning accept loop starves the whole scheduler. p.last_arm_errno = @backingInt(c.errno(@as(isize, -1))); p.arm_failures +%= 1; _ = p.armed.remove(fd); return error.PollerArmFailed; }
/// fills `out` with ready events; blocks up to timeout_ms. returns slice /// of out. a single fd can yield two events (read + write waiters), so /// callers size `out` at 2x their raw batch. pub fn wait(p: *Epoll, out: []Event, timeout_ms: i32) []Event { var raw: [max_batch]c.epoll_event = undefined; const cap: c_uint = @intCast(@min(out.len / 2, raw.len)); const n = c.epoll_wait(p.epfd, &raw, cap, timeout_ms); if (n <= 0) return out[0..0]; const count: usize = @intCast(n); var emitted: usize = 0; for (raw[0..count]) |r| { const readable = r.events & EPOLL.IN != 0; const writable = r.events & EPOLL.OUT != 0; const hup = r.events & (EPOLL.HUP | EPOLL.ERR | EPOLL.RDHUP) != 0; if (r.data.u64 & TAG == 0) { out[emitted] = .{ .token = r.data.u64, .readable = readable, .writable = writable, .hup = hup, .fd = -1 }; emitted += 1; continue; } const fd: c.fd_t = @bitCast(@as(u32, @truncate(r.data.u64))); const entry = p.armed.getPtr(fd) orelse { // stale kernel registration with no table entry: remove it // so it cannot fire again. _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null); continue; }; if ((readable or hup) and entry.read_token != null) { out[emitted] = .{ .token = entry.read_token.?, .readable = true, .writable = false, .hup = hup, .fd = fd }; entry.read_token = null; emitted += 1; } if ((writable or hup) and entry.write_token != null) { out[emitted] = .{ .token = entry.write_token.?, .readable = false, .writable = true, .hup = hup, .fd = fd }; entry.write_token = null; emitted += 1; } // userspace one-shot: narrow the interest, and DELETE the // registration once no direction is armed. an empty registration // left as RDHUP-only keeps firing for every dead socket // (level-triggered), and a few hundred churned connections turn // every epoll batch into token-less hangup noise that starves // real events — observed as the accept loop going deaf while // everything else stayed healthy. if (entry.read_token == null and entry.write_token == null) { _ = c.epoll_ctl(p.epfd, EPOLL.CTL_DEL, fd, null); _ = p.armed.remove(fd); } else { var ev: c.epoll_event = .{ .events = mergedEvents(entry.*), .data = .{ .u64 = TAG | @as(u64, @intCast(@as(u32, @bitCast(fd)))) }, }; _ = c.epoll_ctl(p.epfd, EPOLL.CTL_MOD, fd, &ev); } } return out[0..emitted]; }
fn mergedEvents(a: Armed) u32 { var events: u32 = EPOLL.RDHUP; if (a.read_token != null) events |= EPOLL.IN; if (a.write_token != null) events |= EPOLL.OUT; return events; }
fn eventsFrom(interest: Interest) u32 { var events: u32 = EPOLL.RDHUP; if (interest.read) events |= EPOLL.IN; if (interest.write) events |= EPOLL.OUT; return events; }};
pub const Kqueue = struct { kq: c.fd_t, arm_failures: u64 = 0, last_arm_errno: i32 = 0, allocator: std.mem.Allocator, /// mirrors the kernel's per-direction one-shot arms. The kernel does not /// need it (filters are independent), but close-while-parked does: /// closing an fd silently removes its kevents, so without a userspace /// record of who is parked there the fiber is stranded forever — the /// same deafness class the epoll table already guards against. armed: std.AutoHashMapUnmanaged(c.fd_t, Armed) = .empty,
const Armed = struct { read_token: ?u64 = null, write_token: ?u64 = null };
pub fn init(allocator: std.mem.Allocator) !Kqueue { const kq = c.kqueue(); if (kq < 0) return error.PollerInitFailed; return .{ .kq = kq, .allocator = allocator }; }
/// remove `fd`'s arms and return whoever was parked on it — called when /// the fd is about to be closed. The kernel removes the kevents itself /// on close; dropping the table entry here keeps the mirror honest. pub fn takeWaiters(p: *Kqueue, fd: c.fd_t) [2]?u64 { const e = p.armed.fetchRemove(fd) orelse return .{ null, null }; return .{ e.value.read_token, e.value.write_token }; }
pub fn registrationCount(p: *const Kqueue) usize { return p.armed.count(); }
pub fn deinit(p: *Kqueue) void { p.armed.deinit(p.allocator); _ = c.close(p.kq); }
pub fn add(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void { // both filters are always registered; interest toggles ENABLE/DISABLE. // kqueue EV_ADD on an existing (ident,filter) updates it, so mod == add. const changes = [2]c.Kevent{ kev(fd, c.EVFILT.READ, interest.read, token), kev(fd, c.EVFILT.WRITE, interest.write, token), }; var none: [0]c.Kevent = .{}; const rc = c.kevent(p.kq, &changes, changes.len, &none, 0, null); if (rc < 0) return error.PollerAddFailed; }
pub fn mod(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void { return p.add(fd, interest, token) catch error.PollerModFailed; }
/// arm a single edge: fires once, then disarms until re-armed. pub fn armOneshot(p: *Kqueue, fd: c.fd_t, interest: Interest, token: u64) !void { const gop = p.armed.getOrPut(p.allocator, fd) catch return error.PollerArmFailed; if (!gop.found_existing) gop.value_ptr.* = .{}; if (interest.read) gop.value_ptr.read_token = token else gop.value_ptr.write_token = token; var changes: [1]c.Kevent = undefined; const filter: @FieldType(c.Kevent, "filter") = if (interest.read) c.EVFILT.READ else c.EVFILT.WRITE; changes[0] = .{ .ident = @intCast(fd), .filter = filter, .flags = c.EV.ADD | c.EV.ENABLE | c.EV.ONESHOT, .fflags = 0, .data = 0, .udata = token, }; var none: [0]c.Kevent = .{}; if (c.kevent(p.kq, &changes, changes.len, &none, 0, null) < 0) { if (interest.read) gop.value_ptr.read_token = null else gop.value_ptr.write_token = null; if (gop.value_ptr.read_token == null and gop.value_ptr.write_token == null) _ = p.armed.remove(fd); return error.PollerArmFailed; } }
pub fn del(p: *Kqueue, fd: c.fd_t) void { // closing the fd removes its kevents automatically; explicit EV_DELETE // would race with close-initiated removal, so rely on close semantics. _ = p; _ = fd; }
pub fn wait(p: *Kqueue, out: []Event, timeout_ms: i32) []Event { var raw: [max_batch]c.Kevent = undefined; const cap: c_int = @intCast(@min(out.len, raw.len)); const ts: c.timespec = .{ .sec = @divTrunc(timeout_ms, 1000), .nsec = @as(@FieldType(c.timespec, "nsec"), @intCast(@mod(timeout_ms, 1000))) * 1_000_000, }; const no_changes: [0]c.Kevent = .{}; const n = c.kevent(p.kq, &no_changes, 0, &raw, cap, &ts); if (n <= 0) return out[0..0]; const count: usize = @intCast(n); for (raw[0..count], out[0..count]) |r, *e| { const fd: c.fd_t = @intCast(r.ident); e.* = .{ .token = r.udata, .readable = r.filter == c.EVFILT.READ, .writable = r.filter == c.EVFILT.WRITE, .hup = r.flags & c.EV.EOF != 0, .fd = fd, }; // EV_ONESHOT consumed the kernel arm; mirror that here. Literal // tokens from add() (the wake pipe) are not in the table. if (p.armed.getPtr(fd)) |entry| { if (e.readable) entry.read_token = null else entry.write_token = null; if (entry.read_token == null and entry.write_token == null) _ = p.armed.remove(fd); } } return out[0..count]; }
fn kev(fd: c.fd_t, filter: @FieldType(c.Kevent, "filter"), enabled: bool, token: u64) c.Kevent { return .{ .ident = @intCast(fd), .filter = filter, .flags = c.EV.ADD | @as(@FieldType(c.Kevent, "flags"), if (enabled) c.EV.ENABLE else c.EV.DISABLE), .fflags = 0, .data = 0, .udata = token, }; }};
pub const max_batch = 256;
test "epoll: a delivered readiness event destroys the arm and is never redelivered" { if (builtin.os.tag != .linux) return error.SkipZigTest; var p = try Epoll.init(std.testing.allocator); defer p.deinit();
var fds: [2]c.fd_t = undefined; try std.testing.expect(c.socketpair(c.AF.UNIX, c.SOCK.STREAM, 0, &fds) == 0); defer _ = c.close(fds[0]); defer _ = c.close(fds[1]);
const token: u64 = 0xBEEF; try p.armOneshot(fds[0], .{ .read = true }, token); try std.testing.expectEqual(@as(usize, 1), p.registrationCount());
// peer writes then closes: readable + RDHUP, the websocket close-frame case try std.testing.expectEqual(@as(isize, 3), c.write(fds[1], "bye", 3)); _ = c.close(fds[1]);
var out: [max_batch * 2]Event = undefined; const first = p.wait(&out, 50); try std.testing.expectEqual(@as(usize, 1), first.len); try std.testing.expectEqual(token, first[0].token);
// delivery already destroyed the registration try std.testing.expectEqual(@as(usize, 0), p.registrationCount());
// the caller drops this event (the fiber it named was not parked). // Nothing re-arms. The bytes are still unread and the peer is gone, but // the poller will never mention this fd again -- it is now invisible. const second = p.wait(&out, 50); try std.testing.expectEqual(@as(usize, 0), second.len);
var pending: c_int = 0; _ = c.ioctl(fds[0], @as(c_int, 0x541B), &pending); // FIONREAD try std.testing.expectEqual(@as(c_int, 3), pending);}