From 14899db412f8404a97115a7d6a029a59bfc75d2d Mon Sep 17 00:00:00 2001 From: Endel Dreyer Date: Sun, 13 Sep 2026 22:11:41 -0300 Subject: [PATCH] iocp: add poll, through the AFD driver The IOCP backend had no readiness operation, so Stream.poll was a compile error on Windows and a loop that waits for a socket to become readable could not run there. Poll now submits IOCTL_AFD_POLL on a per-loop \Device\Afd handle associated with the completion port. That is the mechanism beneath WSAPoll, and what wepoll and mio use, so the result arrives like any other overlapped I/O and cancels through NtCancelIoFileEx. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YXLKbmpkRS5DedisNs9VNp --- src/backend/iocp.zig | 127 +++++++++++++++++++++++++++++++++++++++ src/watcher/stream.zig | 21 ++++--- src/windows.zig | 133 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 274 insertions(+), 7 deletions(-) diff --git a/src/backend/iocp.zig b/src/backend/iocp.zig index cc8506ea..c031b577 100644 --- a/src/backend/iocp.zig +++ b/src/backend/iocp.zig @@ -57,6 +57,9 @@ pub const Loop = struct { /// Heap of timers. timers: TimerHeap = .{ .context = {} }, + /// The AFD driver handle poll requests go through, opened on first use. + afd_handle: ?windows.HANDLE = null, + /// Cached time cached_now: u64, @@ -97,6 +100,7 @@ pub const Loop = struct { /// Deinitialize the loop, this closes the handle to the Completion Port. Any events that were /// unprocessed are lost -- their callbacks will never be called. pub fn deinit(self: *Loop) void { + if (self.afd_handle) |h| windows.CloseHandle(h); windows.CloseHandle(self.iocp_handle); } @@ -726,6 +730,20 @@ pub const Loop = struct { break :action .{ .submitted = {} }; }, + .poll => |*v| action: { + const afd_handle = self.afd() catch |err| break :action .{ .result = .{ .poll = err } }; + const base = windows.afd.baseHandle(v.fd) catch |err| break :action .{ .result = .{ .poll = err } }; + v.info = .{ + .timeout = std.math.maxInt(i64), + .number_of_handles = 1, + .exclusive = 0, + .handles = .{.{ .handle = base, .events = v.events, .status = .SUCCESS }}, + }; + windows.afd.poll(afd_handle, &v.info, &completion.overlapped) catch |err| + break :action .{ .result = .{ .poll = err } }; + break :action .{ .submitted = {} }; + }, + .timer => |*v| action: { v.c = completion; self.timers.insert(v); @@ -844,6 +862,12 @@ pub const Loop = struct { } }, + .poll => { + if (completion.flags.state == .active) { + cancel_result.?.* = windows.afd.cancel(self.afd_handle.?, &completion.overlapped); + } + }, + else => @panic("Not implemented"), } } @@ -864,6 +888,17 @@ pub const Loop = struct { }; } + /// The AFD handle, opened and associated with the port the first time a + /// poll needs it. + fn afd(self: *Loop) PollError!windows.HANDLE { + if (self.afd_handle) |h| return h; + const h = try windows.afd.open(); + errdefer windows.CloseHandle(h); + self.associate_fd(h) catch return error.Unexpected; + self.afd_handle = h; + return h; + } + /// Associate a handler to the internal completion port. /// This has to be done only once per handle so we delegate the responsibility to the caller. pub fn associate_fd(self: Loop, fd: windows.HANDLE) !void { @@ -1179,6 +1214,12 @@ pub const Completion = struct { .async_wait => .{ .async_wait = {} }, .job_object => self.result.?, + + .poll => .{ .poll = switch (windows.afd.completionStatus(&self.overlapped)) { + .SUCCESS => {}, + .CANCELLED => error.Canceled, + else => |status| std.os.windows.unexpectedStatus(status), + } }, }; } @@ -1248,8 +1289,20 @@ pub const OperationType = enum { /// Receive a notification from a job object associated with a completion port job_object, + + /// Wait for a socket to become ready, through the AFD driver. + poll, }; +/// What a read poll waits for: data, or anything that means a read will no +/// longer block. +pub const poll_read_events = windows.afd.POLL_RECEIVE | + windows.afd.POLL_ACCEPT | + windows.afd.POLL_DISCONNECT | + windows.afd.POLL_ABORT | + windows.afd.POLL_LOCAL_CLOSE | + windows.afd.POLL_CONNECT_FAIL; + /// All the supported operations of this event loop. These are always /// backend-specific and therefore the structure and types change depending /// on the underlying system in use. The high level operations are @@ -1345,6 +1398,16 @@ pub const Operation = union(OperationType) { /// Do not use this, it is used internally. associated: bool = false, }, + + poll: struct { + fd: windows.HANDLE, + + /// AFD_POLL_* events to wait for. + events: u32, + + /// The request while AFD holds it. Do not use this, it is used internally. + info: windows.afd.PollInfo = undefined, + }, }; /// The result type based on the operation type. For a callback, the @@ -1367,12 +1430,18 @@ pub const Result = union(OperationType) { cancel: CancelError!void, async_wait: AsyncError!void, job_object: JobObjectError!JobObjectResult, + poll: PollError!void, }; pub const CancelError = error{ Unexpected, }; +pub const PollError = error{ + Canceled, + Unexpected, +}; + pub const AcceptError = error{ AddressFamilyNotSupported, ProcessFdQuotaExceeded, @@ -2349,6 +2418,64 @@ test "iocp: recv cancellation" { try testing.expectError(error.Canceled, recv_result.recv); } +test "iocp: poll" { + const mem = std.mem; + const testing = std.testing; + + var loop = try Loop.init(.{}); + defer loop.deinit(); + + const address = try net.Address.parseIp4("127.0.0.1", 3132); + const socket = try windows.WSASocketW(windows.ws2_32.AF.INET, windows.ws2_32.SOCK.DGRAM, windows.ws2_32.IPPROTO.UDP, null, 0, windows.ws2_32.WSA_FLAG_OVERLAPPED); + defer iocpClose(socket); + try iocpSetsockopt(asSocket(socket), windows.ws2_32.SOL.SOCKET, windows.ws2_32.SO.REUSEADDR, &mem.toBytes(@as(c_int, 1))); + try iocpBind(asSocket(socket), &address.any, address.getOsSockLen()); + + const Poll = struct { + result: ?(PollError!void) = null, + + fn callback(ud: ?*anyopaque, _: *Loop, _: *Completion, r: Result) CallbackAction { + const self: *@This() = @ptrCast(@alignCast(ud.?)); + self.result = r.poll; + return .disarm; + } + }; + + // Nothing to read, so the poll stays pending until it is canceled. + var canceled: Poll = .{}; + var c_canceled: Completion = .{ + .op = .{ .poll = .{ .fd = socket, .events = poll_read_events } }, + .userdata = &canceled, + .callback = Poll.callback, + }; + loop.add(&c_canceled); + try loop.run(.no_wait); + try testing.expect(canceled.result == null); + + var c_cancel: Completion = .{ .op = .{ .cancel = .{ .c = &c_canceled } } }; + loop.add(&c_cancel); + try loop.run(.until_done); + try testing.expectError(error.Canceled, canceled.result.?); + + // A datagram arriving completes it. + var readable: Poll = .{}; + var c_readable: Completion = .{ + .op = .{ .poll = .{ .fd = socket, .events = poll_read_events } }, + .userdata = &readable, + .callback = Poll.callback, + }; + loop.add(&c_readable); + try loop.run(.no_wait); + try testing.expect(readable.result == null); + + const sender = try windows.WSASocketW(windows.ws2_32.AF.INET, windows.ws2_32.SOCK.DGRAM, windows.ws2_32.IPPROTO.UDP, null, 0, windows.ws2_32.WSA_FLAG_OVERLAPPED); + defer iocpClose(sender); + try testing.expectEqual(4, windows.ws2_32.sendto(asSocket(sender), "ping", 4, 0, &address.any, @intCast(address.getOsSockLen()))); + + try loop.run(.until_done); + try readable.result.?; +} + test "iocp: accept cancellation" { const mem = std.mem; const testing = std.testing; diff --git a/src/watcher/stream.zig b/src/watcher/stream.zig index c00e631e..f8fbe636 100644 --- a/src/watcher/stream.zig +++ b/src/watcher/stream.zig @@ -60,12 +60,12 @@ pub fn Shared(comptime xev: type) type { pub const PollError = switch (xev.backend) { .io_uring, .epoll, + .iocp, => xev.Sys.PollError, .kqueue, => xev.ReadError, - .iocp, .wasi_poll, => error{}, }; @@ -75,8 +75,9 @@ pub fn Shared(comptime xev: type) type { read = switch (xev.backend) { .io_uring => std.posix.POLL.IN, .epoll => std.os.linux.EPOLL.IN, + .iocp => xev.Sys.poll_read_events, .kqueue => 0, // doesn't matter - .iocp, .wasi_poll => 0, // invalid + .wasi_poll => 0, // invalid }, fn fromResult( @@ -86,6 +87,7 @@ pub fn Shared(comptime xev: type) type { return switch (xev.backend) { .io_uring, .epoll, + .iocp, => if (result.poll) |_| @enumFromInt(c.op.poll.events) else |err| @@ -96,7 +98,6 @@ pub fn Shared(comptime xev: type) type { else => unreachable, }, - .iocp, .wasi_poll, => @compileError("poll not supported on this backend"), }; @@ -221,8 +222,8 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ // Do not add the methods for poll if the backend doesn't support it. switch (xev.backend) { - .io_uring, .epoll, .kqueue => {}, - .iocp, .wasi_poll => return struct {}, + .io_uring, .epoll, .kqueue, .iocp => {}, + .wasi_poll => return struct {}, } return struct { @@ -263,6 +264,13 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ }, } }, + .iocp => .{ .poll = .{ + .fd = self.fd, + .events = switch (event) { + .read => xev.Sys.poll_read_events, + }, + } }, + .kqueue => switch (options.read) { .none => unreachable, @@ -277,7 +285,6 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ } }, }, - .iocp, .wasi_poll, => @compileError("poll not supported on this backend"), }, @@ -292,11 +299,11 @@ fn Pollable(comptime xev: type, comptime T: type, comptime options: Options) typ const fd: Self = switch (xev.backend) { .io_uring, .epoll, + .iocp, => T.initFd(c_inner.op.poll.fd), .kqueue => T.initFd(c_inner.op.read.fd), - .iocp, .wasi_poll, => @compileError("poll not supported on this backend"), }; diff --git a/src/windows.zig b/src/windows.zig index 8ddbe3a3..ef81dfc6 100644 --- a/src/windows.zig +++ b/src/windows.zig @@ -444,6 +444,29 @@ pub const ws2_32 = struct { optlen: i32, ) callconv(.winapi) i32; + pub extern "ws2_32" fn sendto( + s: SOCKET, + buf: [*]const u8, + len: i32, + flags: i32, + to: *const posix.sockaddr, + tolen: i32, + ) callconv(.winapi) i32; + + pub extern "ws2_32" fn WSAIoctl( + s: SOCKET, + dwIoControlCode: DWORD, + lpvInBuffer: ?*const anyopaque, + cbInBuffer: DWORD, + lpvOutBuffer: ?*anyopaque, + cbOutBuffer: DWORD, + lpcbBytesReturned: *DWORD, + lpOverlapped: ?*OVERLAPPED, + lpCompletionRoutine: ?LPWSAOVERLAPPED_COMPLETION_ROUTINE, + ) callconv(.winapi) i32; + + pub const SIO_BASE_HANDLE: DWORD = 0x48000022; + pub extern "mswsock" fn AcceptEx( sListenSocket: SOCKET, sAcceptSocket: SOCKET, @@ -456,6 +479,116 @@ pub const ws2_32 = struct { ) callconv(.winapi) BOOL; }; +/// Socket readiness through the AFD driver: the mechanism underneath `WSAPoll`, +/// used directly so a poll can complete on a completion port like any other +/// overlapped I/O. std has the IOCTL code but not the request structures. +pub const afd = struct { + pub const POLL_RECEIVE: u32 = 0x0001; + pub const POLL_RECEIVE_EXPEDITED: u32 = 0x0002; + pub const POLL_SEND: u32 = 0x0004; + pub const POLL_DISCONNECT: u32 = 0x0008; + pub const POLL_ABORT: u32 = 0x0010; + pub const POLL_LOCAL_CLOSE: u32 = 0x0020; + pub const POLL_ACCEPT: u32 = 0x0080; + pub const POLL_CONNECT_FAIL: u32 = 0x0100; + + pub const PollHandleInfo = extern struct { + handle: HANDLE, + events: win.ULONG, + status: win.NTSTATUS, + }; + + pub const PollInfo = extern struct { + timeout: win.LARGE_INTEGER, + number_of_handles: win.ULONG, + exclusive: win.ULONG, + handles: [1]PollHandleInfo, + }; + + /// Open a handle to the AFD driver to send poll requests through. Any + /// name under `\Device\Afd` will do; the suffix only shows in handle + /// listings. + pub fn open() error{Unexpected}!HANDLE { + var handle: HANDLE = undefined; + var iosb: win.IO_STATUS_BLOCK = undefined; + const status = win.ntdll.NtCreateFile( + &handle, + .{ .STANDARD = .{ .SYNCHRONIZE = true } }, + &.{ .ObjectName = @constCast(&win.UNICODE_STRING.init(win.AFD.DEVICE_NAME ++ .{ '\\', 'x', 'e', 'v' })) }, + &iosb, + null, + .{}, + .{ .READ = true, .WRITE = true }, + .OPEN, + .{ .IO = .ASYNCHRONOUS }, + null, + 0, + ); + if (status != .SUCCESS) return win.unexpectedStatus(status); + return handle; + } + + /// The handle AFD knows a socket by. A layered service provider can wrap + /// the one the application holds, and AFD rejects those. + pub fn baseHandle(sock: HANDLE) error{Unexpected}!HANDLE { + var base: HANDLE = undefined; + var returned: DWORD = 0; + const rc = ws2_32.WSAIoctl( + @ptrCast(sock), + ws2_32.SIO_BASE_HANDLE, + null, + 0, + @ptrCast(&base), + @sizeOf(HANDLE), + &returned, + null, + null, + ); + if (rc != 0) return unexpectedWSAError(ws2_32.WSAGetLastError()); + return base; + } + + /// Start a poll. It completes on the port `afd_handle` is associated with: + /// `overlapped` is passed as the APC context, which comes back as the + /// entry's `lpOverlapped`, and as the IO_STATUS_BLOCK, whose two fields + /// OVERLAPPED begins with. `info` is read and written in place, so it has + /// to stay put until then. + pub fn poll(afd_handle: HANDLE, info: *PollInfo, overlapped: *OVERLAPPED) error{Unexpected}!void { + const status = win.ntdll.NtDeviceIoControlFile( + afd_handle, + null, + null, + overlapped, + @ptrCast(overlapped), + win.IOCTL.AFD.POLL, + info, + @sizeOf(PollInfo), + info, + @sizeOf(PollInfo), + ); + switch (status) { + // Either way the result arrives on the port. + .SUCCESS, .PENDING => {}, + else => return win.unexpectedStatus(status), + } + } + + /// Cancel a pending poll. It still completes, with STATUS_CANCELLED. + pub fn cancel(afd_handle: HANDLE, overlapped: *OVERLAPPED) error{Unexpected}!void { + var iosb: win.IO_STATUS_BLOCK = undefined; + switch (win.ntdll.NtCancelIoFileEx(afd_handle, @ptrCast(overlapped), &iosb)) { + // NOT_FOUND: it completed first, and its result is already queued. + .SUCCESS, .NOT_FOUND => {}, + else => |status| return win.unexpectedStatus(status), + } + } + + /// The final status of a completed poll. + pub fn completionStatus(overlapped: *const OVERLAPPED) win.NTSTATUS { + return @enumFromInt(@as(u32, @truncate(overlapped.Internal))); + } +}; + // --- High-level wrapper functions --- pub fn unexpectedWSAError(err: ws2_32.WinsockError) error{Unexpected} {