From c706d22f743fc1bac8001a195d228c14dcf3bd59 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Wed, 19 Aug 2026 19:16:51 -0700 Subject: [PATCH 01/10] nbio: fix lost lake-up and stale op.l on windows --- core/nbio/impl_windows.odin | 25 +++++++------------------ 1 file changed, 7 insertions(+), 18 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index 162092a80..d9e43f378 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -25,11 +25,6 @@ _Event_Loop :: struct { thread: win.HANDLE, completed: queue.Queue(^Operation), completed_oob: Multi_Producer_Single_Consumer, - state: enum { - Working, - Waking, - Sleeping, - }, } @(private="package") @@ -176,12 +171,8 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { } if actual_timeout > 0 { - sync.atomic_store_explicit(&l.state, .Sleeping, .Release) - - // There could be a race condition where we go sleeping at the same time as things get queued - // and a wakeup isn't done because the state is not .Sleeping yet. - // So after sleeping we first check our queues. - + // Work may have been queued after the drain at the top of this tick, + // so check the queues once more before blocking. for { op := (^Operation)(mpsc_dequeue(&l.queue)) if op == nil { break } @@ -209,8 +200,6 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { } } - sync.atomic_store_explicit(&l.state, .Working, .Relaxed) - if actual_timeout > 0 { // We may have just waited some time, lets update the current time. l.now = time.now() @@ -228,7 +217,7 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { handle_completed(op) } else { op_l := op.l - for !mpsc_enqueue(&op.l.completed_oob, op) { + for !mpsc_enqueue(&op_l.completed_oob, op) { warn("oob queue filled up, QUEUE_SIZE may need increasing") _wake_up(op_l) win.SwitchToThread() @@ -770,10 +759,10 @@ _associate_socket :: proc(socket: Any_Socket, l: ^Event_Loop) -> Association_Err @(private="package") _wake_up :: proc(l: ^Event_Loop) { - _, exchanged := sync.atomic_compare_exchange_strong(&l.state, .Sleeping, .Waking) - if exchanged { - win.QueueUserAPC(proc "system" (Parameter: win.ULONG_PTR) {}, l.thread, 0) - } + // Unconditional: an APC queued before the loop enters its alertable wait stays + // pending and is delivered as soon as that wait begins. Queueing it only once the + // loop is already asleep drops every wake sent while it is still awake. + win.QueueUserAPC(proc "system" (Parameter: win.ULONG_PTR) {}, l.thread, 0) } @(private="package") From 2719a089ec943eafe7ffcaab2204640101f94310 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Thu, 20 Aug 2026 20:43:08 -0700 Subject: [PATCH 02/10] recompute timeouts --- core/nbio/impl_windows.odin | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index d9e43f378..2af88afb9 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -160,15 +160,7 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { if pool.num_outstanding(&l.operation_pool) == 0 { return nil } - actual_timeout := win.INFINITE - if queue.len(l.completed) > 0 || mpsc_count(&l.completed_oob) > 0 { - actual_timeout = 0 - } else if timeout >= 0 { - actual_timeout = win.DWORD(timeout / time.Millisecond) - } - if nt, ok := next_timeout.?; ok { - actual_timeout = min(actual_timeout, win.DWORD(nt / time.Millisecond)) - } + actual_timeout := compute_timeout(l, timeout, next_timeout) if actual_timeout > 0 { // Work may have been queued after the drain at the top of this tick, @@ -184,6 +176,10 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { if op == nil { break } handle_completed(op) } + + // The drains can add timeouts, and `timeout_exec` only puts those in + // `l.timeouts` without posting anything + actual_timeout = compute_timeout(l, timeout, check_timeouts(l)) } for { @@ -237,6 +233,19 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { return nil + compute_timeout :: proc(l: ^Event_Loop, timeout: time.Duration, next_timeout: Maybe(time.Duration)) -> win.DWORD { + actual: win.DWORD = win.INFINITE + if queue.len(l.completed) > 0 || mpsc_count(&l.completed_oob) > 0 { + actual = 0 + } else if timeout >= 0 { + actual = win.DWORD(timeout / time.Millisecond) + } + if nt, ok := next_timeout.?; ok { + actual = min(actual, win.DWORD(nt / time.Millisecond)) + } + return actual + } + check_timeouts :: proc(l: ^Event_Loop) -> (expires: Maybe(time.Duration)) { curr := l.now From ae56bf54564bd88018bc0cf4984dd8194d6a637c Mon Sep 17 00:00:00 2001 From: kalsprite Date: Thu, 20 Aug 2026 23:04:26 -0700 Subject: [PATCH 03/10] nbio(windows): clear the OVERLAPPED when arming a poll; nbio(windows): tear down poll waits before their event; nbio(windows): clear the socket event record on poll completion --- core/nbio/impl_windows.odin | 39 ++++++++++++++++++++++++++++++++++--- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index 2af88afb9..a8c99e05b 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -676,7 +676,7 @@ _remove :: proc(target: ^Operation) { switch target.type { case .Poll: - win.UnregisterWaitEx(target.poll._impl.wait_handle, nil) + win.UnregisterWaitEx(target.poll._impl.wait_handle, win.INVALID_HANDLE_VALUE) target.poll._impl.wait_handle = nil ok := win.PostQueuedCompletionStatus( @@ -1500,9 +1500,14 @@ sendfile_callback :: proc(op: ^Operation) -> Op_Result { return .Done } +// Bit indices into `WSANETWORKEVENTS.iErrorCode`, corresponding to the `FD_*` masks. +FD_READ_BIT :: 0 +FD_WRITE_BIT :: 1 + @(require_results) poll_exec :: proc(op: ^Operation) -> Op_Result { assert(op.type == .Poll) + op._impl.over = {} // Operations are recycled, clear stale state from a previous use. events: i32 = win.FD_CLOSE switch op.poll.event { @@ -1565,12 +1570,40 @@ poll_exec :: proc(op: ^Operation) -> Op_Result { poll_callback :: proc(op: ^Operation) { assert(op.type == .Poll) + // Clear the socket's internal network event record, and find out what actually + // fired. Without this the record stays set after an event is reported, so the next + // `WSAEventSelect` on that socket signals its event object immediately from the + // stale record. That completes a poll for a readiness that never happened, and the + // send/recv the caller then makes fails with WOULDBLOCK. if op._impl.over.hEvent != nil { - win.WSACloseEvent(op._impl.over.hEvent) + nev: win.WSANETWORKEVENTS + sk := win.SOCKET(net.any_socket_to_socket(op.poll.socket)) + if win.WSAEnumNetworkEvents(sk, op._impl.over.hEvent, &nev) == 0 { + bit: uint + switch op.poll.event { + case .Receive: bit = FD_READ_BIT + case .Send: bit = FD_WRITE_BIT + } + + // Only downgrade a result that is still `Ready`; `wait_callback` may have + // already set `.Timeout`. + if op.poll.result == nil && nev.lNetworkEvents & (i32(1) << bit) != 0 && nev.iErrorCode[bit] != 0 { + op.poll.result = .Error + } + } } + // Tear down in the reverse order of `poll_exec`: stop `wait_callback` from running + // before the event it waits on goes away. `INVALID_HANDLE_VALUE` waits for an + // in-flight callback to return, so it can't touch `op` after it is recycled. if op.poll._impl.wait_handle != nil { - win.UnregisterWaitEx(op.poll._impl.wait_handle, nil) + win.UnregisterWaitEx(op.poll._impl.wait_handle, win.INVALID_HANDLE_VALUE) + op.poll._impl.wait_handle = nil + } + + if op._impl.over.hEvent != nil { + win.WSACloseEvent(op._impl.over.hEvent) + op._impl.over.hEvent = nil } if op.poll.result != nil { From bcf2dcd2f0afeac70bb28ba88f0299c1ae0b11f9 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Thu, 20 Aug 2026 23:39:21 -0700 Subject: [PATCH 04/10] nbio(windows): don't complete synchronously-failed overlapped ops twice --- core/nbio/impl_windows.odin | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index a8c99e05b..de16a5a69 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -923,6 +923,9 @@ accept_exec :: proc(op: ^Operation) -> Op_Result { return .Pending } else if op._impl.over.Internal == nil { op.accept.err = net._accept_error() + } else { + link_timeout(op, op.accept.expires) + return .Pending } } @@ -1024,6 +1027,9 @@ dial_exec :: proc(op: ^Operation) -> (result: Op_Result) { return .Pending } else if op._impl.over.Internal == nil { op.dial.err = net._dial_error() + } else { + link_timeout(op, op.dial.expires) + return .Pending } } @@ -1083,6 +1089,13 @@ read_exec :: proc(op: ^Operation) -> Op_Result { return .Pending } op.read.err = FS_Error(err) + } else { + // The read completed synchronously with a failure status. `FILE_SKIP_COMPLETION_PORT_ON_SUCCESS` + // only suppresses the completion packet on success, so one is still queued for + // this. Returning `.Done` here would complete the operation a second time, on an + // Operation that has already been recycled into the pool. + link_timeout(op, op.read.expires) + return .Pending } } @@ -1157,6 +1170,9 @@ write_exec :: proc(op: ^Operation) -> Op_Result { return .Pending } op.write.err = FS_Error(err) + } else { + link_timeout(op, op.write.expires) + return .Pending } } @@ -1250,6 +1266,9 @@ recv_exec :: proc(op: ^Operation) -> Op_Result { case TCP_Socket: op.recv.err = net._tcp_recv_error() case UDP_Socket: op.recv.err = net._udp_recv_error() } + } else { + link_timeout(op, op.recv.expires) + return .Pending } } @@ -1368,6 +1387,9 @@ send_exec :: proc(op: ^Operation) -> Op_Result { case TCP_Socket: op.send.err = net._tcp_send_error() case UDP_Socket: op.send.err = net._udp_send_error() } + } else { + link_timeout(op, op.send.expires) + return .Pending } } @@ -1457,6 +1479,9 @@ sendfile_exec :: proc(op: ^Operation) -> Op_Result { return .Pending } else if op._impl.over.Internal == nil { op.sendfile.err = net._tcp_send_error() + } else { + link_timeout(op, op.sendfile.expires) + return .Pending } } From 6e690afaf1d455cf29c7997839f2a39e9d226a42 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 00:09:43 -0700 Subject: [PATCH 05/10] revert wake change --- core/nbio/impl_windows.odin | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index de16a5a69..e1e0f2ebc 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -25,6 +25,11 @@ _Event_Loop :: struct { thread: win.HANDLE, completed: queue.Queue(^Operation), completed_oob: Multi_Producer_Single_Consumer, + state: enum { + Working, + Waking, + Sleeping, + }, } @(private="package") @@ -163,8 +168,11 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { actual_timeout := compute_timeout(l, timeout, next_timeout) if actual_timeout > 0 { - // Work may have been queued after the drain at the top of this tick, - // so check the queues once more before blocking. + sync.atomic_store_explicit(&l.state, .Sleeping, .Release) + + // There could be a race condition where we go sleeping at the same time as things get queued + // and a wakeup isn't done because the state is not .Sleeping yet. + // So after sleeping we first check our queues. for { op := (^Operation)(mpsc_dequeue(&l.queue)) if op == nil { break } @@ -196,6 +204,8 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { } } + sync.atomic_store_explicit(&l.state, .Working, .Relaxed) + if actual_timeout > 0 { // We may have just waited some time, lets update the current time. l.now = time.now() @@ -768,10 +778,10 @@ _associate_socket :: proc(socket: Any_Socket, l: ^Event_Loop) -> Association_Err @(private="package") _wake_up :: proc(l: ^Event_Loop) { - // Unconditional: an APC queued before the loop enters its alertable wait stays - // pending and is delivered as soon as that wait begins. Queueing it only once the - // loop is already asleep drops every wake sent while it is still awake. - win.QueueUserAPC(proc "system" (Parameter: win.ULONG_PTR) {}, l.thread, 0) + _, exchanged := sync.atomic_compare_exchange_strong(&l.state, .Sleeping, .Waking) + if exchanged { + win.QueueUserAPC(proc "system" (Parameter: win.ULONG_PTR) {}, l.thread, 0) + } } @(private="package") From d2da33235d3a63b418097bc42812ada9696f7672 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 09:18:57 -0700 Subject: [PATCH 06/10] nbio(windows): handle queued work in the tick that was woken for it; nbio(tests): let wake_up tolerate a tick that returns without progress --- core/nbio/impl_windows.odin | 14 ++++++++++++++ tests/core/nbio/nbio.odin | 7 +++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index e1e0f2ebc..7357f7c18 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -241,6 +241,20 @@ __tick :: proc(l: ^Event_Loop, timeout: time.Duration) -> (err: General_Error) { actual_timeout = 0 } + // A wake, or another loop routing a completion to us, can leave work queued. + // Handle it here instead of waiting for the caller to tick again. + for { + op := (^Operation)(mpsc_dequeue(&l.queue)) + if op == nil { break } + _exec(op) + } + + for { + op := (^Operation)(mpsc_dequeue(&l.completed_oob)) + if op == nil { break } + handle_completed(op) + } + return nil compute_timeout :: proc(l: ^Event_Loop, timeout: time.Duration, next_timeout: Maybe(time.Duration)) -> win.DWORD { diff --git a/tests/core/nbio/nbio.odin b/tests/core/nbio/nbio.odin index 6121d1ac7..5d43fd814 100644 --- a/tests/core/nbio/nbio.odin +++ b/tests/core/nbio/nbio.odin @@ -244,8 +244,11 @@ wake_up :: proc(t: ^testing.T) { }, context) defer thread.destroy(thr) - // Should block forever until the thread calling wake_up will make it return. - ev(t, nbio.tick(), nil) + // A tick can return without progress; loop until the wake is observed. + // A lost wake would block here forever and trip the fail timeout. + for !hit { + ev(t, nbio.tick(), nil) + } e(t, hit) nbio.remove(accept) From 1b4e14fbc549e532fe5e2fc614ab66502e849c93 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 20:43:33 -0700 Subject: [PATCH 07/10] restore fail timeout --- tests/core/nbio/net.odin | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/core/nbio/net.odin b/tests/core/nbio/net.odin index 77e44f73b..f0b8226c6 100644 --- a/tests/core/nbio/net.odin +++ b/tests/core/nbio/net.odin @@ -245,7 +245,7 @@ And it tests big send/recv buffers being handled properly. @(test) poll :: proc(t: ^testing.T) { if event_loop_guard(t) { -// testing.set_fail_timeout(t, time.Minute) + testing.set_fail_timeout(t, time.Minute) can_recv: bool From 7bb55ba45b4c1d463e1f096fcde35e15052c5fa9 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 21:31:49 -0700 Subject: [PATCH 08/10] nbio(windows): poll sockets through AFD --- core/nbio/impl_windows.odin | 224 +++++++++++++++++++++--------------- core/nbio/ops.odin | 12 ++ core/sys/windows/ntdll.odin | 13 +++ tests/core/nbio/remove.odin | 17 ++- 4 files changed, 167 insertions(+), 99 deletions(-) diff --git a/core/nbio/impl_windows.odin b/core/nbio/impl_windows.odin index 7357f7c18..4d202620e 100644 --- a/core/nbio/impl_windows.odin +++ b/core/nbio/impl_windows.odin @@ -19,6 +19,38 @@ import win "core:sys/windows" @(private="package") _FULLY_SUPPORTED :: true +// Poll is driven by AFD, the socket driver underneath winsock. +// `WSAEventSelect` is edge triggered (`FD_WRITE` is only recorded again after a +// send fails with WOULDBLOCK) and neither `select` nor `WSAPoll` reports send +// buffer space, so neither can give the level triggered readiness `poll` promises. +// AFD also completes on the IOCP, which makes a poll an ordinary overlapped operation. +IOCTL_AFD_POLL :: 0x00012024 +SIO_BASE_HANDLE :: win.DWORD(0x48000022) + +AFD_POLL_RECEIVE :: 0x0001 +AFD_POLL_RECEIVE_EXPEDITED :: 0x0002 +AFD_POLL_SEND :: 0x0004 +AFD_POLL_DISCONNECT :: 0x0008 +AFD_POLL_ABORT :: 0x0010 +AFD_POLL_LOCAL_CLOSE :: 0x0020 +AFD_POLL_ACCEPT :: 0x0080 +AFD_POLL_CONNECT_FAIL :: 0x0100 + +AFD_Poll_Handle_Info :: struct { + handle: win.HANDLE, + events: win.ULONG, + status: win.NTSTATUS, +} + +AFD_Poll_Info :: struct { + timeout: i64, + number_of_handles: win.ULONG, + exclusive: win.ULONG, + handles: [1]AFD_Poll_Handle_Info, +} + +afd_device_name := [?]u16{'\\','D','e','v','i','c','e','\\','A','f','d','\\','E','n','d','p','o','i','n','t'} + @(private="package") _Event_Loop :: struct { timeouts: avl.Tree(^Operation), @@ -88,7 +120,7 @@ _Timeout :: struct { @(private="package") _Poll :: struct { - wait_handle: win.HANDLE, + info: AFD_Poll_Info, } @(private="package") @@ -699,19 +731,6 @@ _remove :: proc(target: ^Operation) { target._impl.timeout = (^Operation)(REMOVED) switch target.type { - case .Poll: - win.UnregisterWaitEx(target.poll._impl.wait_handle, win.INVALID_HANDLE_VALUE) - target.poll._impl.wait_handle = nil - - ok := win.PostQueuedCompletionStatus( - g.iocp, - 0, - 0, - &target._impl.over, - ) - ensure(ok == true, "unexpected PostQueuedCompletionStatus error") - return - case .Timeout: if avl.remove_value(&target.l.timeouts, target) { debug("removed timeout directly") @@ -727,6 +746,17 @@ _remove :: proc(target: ^Operation) { // Synchronous ops, picked up in handler. return + case .Poll: + // The poll may have completed already, with its completion queued but not yet + // handled, `NOT_FOUND` is expected rather than exceptional. + if !win.CancelIoEx(g.afd, &target._impl.over) { + #partial switch win.System_Error(win.GetLastError()) { + case .NOT_FOUND: + // nop + case: assert(false, "unexpected CancelIoEx error") + } + } + case .Accept, .Dial, .Read, .Recv, .Send, .Write, .Send_File: if is_pending(target._impl.over) { handle := operation_handle(target) @@ -824,6 +854,7 @@ g: struct{ mu: sync.Mutex, refs: int, iocp: win.HANDLE, + afd: win.HANDLE, err: General_Error, } @@ -839,6 +870,36 @@ g_ref :: proc() -> General_Error { if g.iocp == nil { g.err = General_Error(win.GetLastError()) } + + if g.err != nil { return g.err } + + // A handle on the socket driver, used to poll sockets for readiness. + iosb: win.IO_STATUS_BLOCK + status := win.NtCreateFile( + &g.afd, + win.SYNCHRONIZE, + &{ + Length = size_of(win.OBJECT_ATTRIBUTES), + ObjectName = &{ + Length = u16(len(afd_device_name)*2), + MaximumLength = u16(len(afd_device_name)*2), + Buffer = raw_data(afd_device_name[:]), + }, + }, + &iosb, + nil, + 0, + win.FILE_SHARE_READ|win.FILE_SHARE_WRITE, + win.FILE_OPEN, + 0, + nil, + 0, + ) + if syserr := win.System_Error(win.RtlNtStatusToDosError(status)); syserr != .SUCCESS { + g.err = General_Error(syserr) + } else if win.CreateIoCompletionPort(g.afd, g.iocp, 0, 0) != g.iocp { + g.err = General_Error(win.GetLastError()) + } } sync.atomic_add(&g.refs, 1) @@ -850,6 +911,7 @@ g_unref :: proc() { sync.guard(&g.mu) if sync.atomic_sub(&g.refs, 1) == 1 { + if g.afd != nil { win.CloseHandle(g.afd) } win.CloseHandle(g.iocp) g.err = nil } @@ -872,7 +934,7 @@ operation_handle :: proc(op: ^Operation) -> win.HANDLE { case .Recv: return win.HANDLE(uintptr(net.any_socket_to_socket(op.recv.socket))) case .Send: return win.HANDLE(uintptr(net.any_socket_to_socket(op.send.socket))) case .Send_File: return win.HANDLE(uintptr(net.any_socket_to_socket(op.sendfile.socket))) - case .Poll: return win.HANDLE(uintptr(net.any_socket_to_socket(op.poll.socket))) + case .Poll: return g.afd case .Stat: return win.HANDLE(uintptr(op.stat.handle)) case .Timeout, .Open, ._Splice, ._Link_Timeout, ._Remove, .None: @@ -1549,120 +1611,94 @@ sendfile_callback :: proc(op: ^Operation) -> Op_Result { return .Done } -// Bit indices into `WSANETWORKEVENTS.iErrorCode`, corresponding to the `FD_*` masks. -FD_READ_BIT :: 0 -FD_WRITE_BIT :: 1 - @(require_results) poll_exec :: proc(op: ^Operation) -> Op_Result { assert(op.type == .Poll) op._impl.over = {} // Operations are recycled, clear stale state from a previous use. - events: i32 = win.FD_CLOSE + events: win.ULONG = AFD_POLL_ABORT|AFD_POLL_DISCONNECT|AFD_POLL_LOCAL_CLOSE|AFD_POLL_CONNECT_FAIL switch op.poll.event { - case .Send: events |= win.FD_WRITE|win.FD_CONNECT - case .Receive: events |= win.FD_READ|win.FD_ACCEPT + case .Receive: events |= AFD_POLL_RECEIVE|AFD_POLL_RECEIVE_EXPEDITED|AFD_POLL_ACCEPT + case .Send: events |= AFD_POLL_SEND case: op.poll.result = .Invalid_Argument return .Done } - op._impl.over.hEvent = win.WSACreateEvent() - if win.WSAEventSelect( + // AFD needs the socket underneath any layered service providers. + base: win.SOCKET + bytes: win.DWORD + if win.WSAIoctl( win.SOCKET(net.any_socket_to_socket(op.poll.socket)), - op._impl.over.hEvent, - events, + SIO_BASE_HANDLE, + nil, 0, + &base, size_of(base), + &bytes, nil, nil, ) != 0 { - #partial switch win.System_Error(win.GetLastError()) { + #partial switch win.System_Error(win.WSAGetLastError()) { case .WSAEINVAL, .WSAENOTSOCK: op.poll.result = .Invalid_Argument case: op.poll.result = .Error } return .Done } - timeout := win.INFINITE + // A negative timeout is relative, in 100ns units. + timeout := max(i64) if op.poll.expires != {} { diff := max(0, time.diff(op.l.now, op.poll.expires)) - timeout = win.DWORD(diff / time.Millisecond) + timeout = -i64(diff / 100) } - ok := win.RegisterWaitForSingleObject( - &op.poll._impl.wait_handle, - op._impl.over.hEvent, - wait_callback, - op, - timeout, - win.WT_EXECUTEINWAITTHREAD|win.WT_EXECUTEONLYONCE, + op.poll._impl.info = { + timeout = timeout, + number_of_handles = 1, + handles = {{handle = win.HANDLE(uintptr(base)), events = events}}, + } + + // The OVERLAPPED doubles as the IO_STATUS_BLOCK, their first two fields line up. + status := win.NtDeviceIoControlFile( + g.afd, + nil, + nil, + &op._impl.over, + win.PIO_STATUS_BLOCK(rawptr(&op._impl.over)), + IOCTL_AFD_POLL, + &op.poll._impl.info, + size_of(AFD_Poll_Info), + &op.poll._impl.info, + size_of(AFD_Poll_Info), ) - ensure(ok == true, "unexpected RegisterWaitForSingleObject error") - return .Pending - - wait_callback :: proc "system" (lpParameter: win.PVOID, TimerOrWaitFired: win.BOOLEAN) { - op := (^Operation)(lpParameter) - assert_contextless(op.type == .Poll) - - if TimerOrWaitFired { - op.poll.result = .Timeout - } - - ok := win.PostQueuedCompletionStatus( - g.iocp, - 0, - 0, - &op._impl.over, - ) - ensure_contextless(ok == true, "unexpected PostQueuedCompletionStatus error") + // The AFD handle is not set to skip completion on success, so a completion is + // queued even when this finishes synchronously. + #partial switch win.System_Error(win.RtlNtStatusToDosError(status)) { + case .SUCCESS, .IO_PENDING: + return .Pending + case: + op.poll.result = .Error + return .Done } } poll_callback :: proc(op: ^Operation) { assert(op.type == .Poll) - // Clear the socket's internal network event record, and find out what actually - // fired. Without this the record stays set after an event is reported, so the next - // `WSAEventSelect` on that socket signals its event object immediately from the - // stale record. That completes a poll for a readiness that never happened, and the - // send/recv the caller then makes fails with WOULDBLOCK. - if op._impl.over.hEvent != nil { - nev: win.WSANETWORKEVENTS - sk := win.SOCKET(net.any_socket_to_socket(op.poll.socket)) - if win.WSAEnumNetworkEvents(sk, op._impl.over.hEvent, &nev) == 0 { - bit: uint - switch op.poll.event { - case .Receive: bit = FD_READ_BIT - case .Send: bit = FD_WRITE_BIT - } - - // Only downgrade a result that is still `Ready`; `wait_callback` may have - // already set `.Timeout`. - if op.poll.result == nil && nev.lNetworkEvents & (i32(1) << bit) != 0 && nev.iErrorCode[bit] != 0 { - op.poll.result = .Error - } - } - } - - // Tear down in the reverse order of `poll_exec`: stop `wait_callback` from running - // before the event it waits on goes away. `INVALID_HANDLE_VALUE` waits for an - // in-flight callback to return, so it can't touch `op` after it is recycled. - if op.poll._impl.wait_handle != nil { - win.UnregisterWaitEx(op.poll._impl.wait_handle, win.INVALID_HANDLE_VALUE) - op.poll._impl.wait_handle = nil - } - - if op._impl.over.hEvent != nil { - win.WSACloseEvent(op._impl.over.hEvent) - op._impl.over.hEvent = nil - } - if op.poll.result != nil { return } - _, err := get_result(op._impl.over) - #partial switch err { - case .SUCCESS: - case: + // AFD reports a timeout by coming back with no handles. + if op.poll._impl.info.number_of_handles == 0 { + op.poll.result = .Timeout + return + } + + if _, err := get_result(op._impl.over); err != .SUCCESS { + op.poll.result = .Error + return + } + + if op.poll._impl.info.handles[0].events & (AFD_POLL_ABORT|AFD_POLL_CONNECT_FAIL) != 0 { op.poll.result = .Error } } diff --git a/core/nbio/ops.odin b/core/nbio/ops.odin index 382dca747..6af266e72 100644 --- a/core/nbio/ops.odin +++ b/core/nbio/ops.odin @@ -1626,6 +1626,9 @@ Poll a socket for readiness. NOTE: this is provided to help with "legacy" APIs that require polling behavior. If you can avoid it and use the other procs in this package, do so. +NOTE: on Windows only one poll per socket is delivered, a second poll on the same +socket does not complete. + Any user data can be set on the returned operation's `user_data` field. Polymorphic variants for type safe user data are available under `poll_poly`, `poll_poly2`, and `poll_poly3`. @@ -1656,6 +1659,9 @@ Poll a socket for readiness. NOTE: this is provided to help with "legacy" APIs that require polling behavior. If you can avoid it and use the other procs in this package, do so. +NOTE: on Windows only one poll per socket is delivered, a second poll on the same +socket does not complete. + This procedure uses polymorphism for type safe user data up to a certain size. Inputs: @@ -1690,6 +1696,9 @@ Poll a socket for readiness. NOTE: this is provided to help with "legacy" APIs that require polling behavior. If you can avoid it and use the other procs in this package, do so. +NOTE: on Windows only one poll per socket is delivered, a second poll on the same +socket does not complete. + This procedure uses polymorphism for type safe user data up to a certain size. Inputs: @@ -1725,6 +1734,9 @@ Poll a socket for readiness. NOTE: this is provided to help with "legacy" APIs that require polling behavior. If you can avoid it and use the other procs in this package, do so. +NOTE: on Windows only one poll per socket is delivered, a second poll on the same +socket does not complete. + This procedure uses polymorphism for type safe user data up to a certain size. Inputs: diff --git a/core/sys/windows/ntdll.odin b/core/sys/windows/ntdll.odin index 41deaa1c4..a78228642 100644 --- a/core/sys/windows/ntdll.odin +++ b/core/sys/windows/ntdll.odin @@ -7,6 +7,19 @@ foreign import ntdll_lib "system:ntdll.lib" foreign ntdll_lib { RtlGetVersion :: proc(lpVersionInformation: ^OSVERSIONINFOEXW) -> NTSTATUS --- + NtDeviceIoControlFile :: proc( + FileHandle: HANDLE, + Event: HANDLE, + ApcRoutine: PIO_APC_ROUTINE, + ApcContext: rawptr, + IoStatusBlock: PIO_STATUS_BLOCK, + IoControlCode: ULONG, + InputBuffer: rawptr, + InputBufferLength: ULONG, + OutputBuffer: rawptr, + OutputBufferLength: ULONG, + ) -> NTSTATUS --- + NtQueryInformationProcess :: proc( ProcessHandle: HANDLE, diff --git a/tests/core/nbio/remove.odin b/tests/core/nbio/remove.odin index 063c2cf58..37d32c5e1 100644 --- a/tests/core/nbio/remove.odin +++ b/tests/core/nbio/remove.odin @@ -212,13 +212,18 @@ remove_multiple_poll :: proc(t: ^testing.T) { if event_loop_guard(t) { testing.set_fail_timeout(t, time.Minute) - sock, ep := open_next_available_local_port(t) - defer nbio.close(sock) + // Two sockets rather than two polls on one socket: only one poll per socket is + // delivered on Windows, and what this tests is removal, not that. + removed_sock, removed_ep := open_next_available_local_port(t) + defer nbio.close(removed_sock) + + kept_sock, kept_ep := open_next_available_local_port(t) + defer nbio.close(kept_sock) hit: bool - first := nbio.poll(sock, .Receive, on_poll) - nbio.poll_poly2(sock, .Receive, t, &hit, on_poll2) + first := nbio.poll(removed_sock, .Receive, on_poll) + nbio.poll_poly2(kept_sock, .Receive, t, &hit, on_poll2) on_poll :: proc(op: ^nbio.Operation) { log.error("shouldn't be called") @@ -235,7 +240,9 @@ remove_multiple_poll :: proc(t: ^testing.T) { ev(t, nbio.tick(0), nil) - nbio.dial_poly(ep, t, on_dial) + // Make both readable, the removed poll must still not fire. + nbio.dial_poly(removed_ep, t, on_dial) + nbio.dial_poly(kept_ep, t, on_dial) on_dial :: proc(op: ^nbio.Operation, t: ^testing.T) { ev(t, op.dial.err, nil) From d29b2cadbd1bb3709f789f02e8ddfbc9078b98f0 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 21:58:40 -0700 Subject: [PATCH 09/10] nbio(tests): fill the socket until sending blocks --- tests/core/nbio/net.odin | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/tests/core/nbio/net.odin b/tests/core/nbio/net.odin index f0b8226c6..544fb6f58 100644 --- a/tests/core/nbio/net.odin +++ b/tests/core/nbio/net.odin @@ -302,13 +302,25 @@ poll :: proc(t: ^testing.T) { on_poll1 :: proc(op: ^nbio.Operation, t: ^testing.T, can_recv: ^bool) { ev(t, op.poll.result, nil) - // Send 4 GB of data, which in my experience causes a Would_Block error because we filled up the internal buffer. + // Fill the socket until sending actually blocks. How much that takes depends + // on the machine's socket buffers, so keep sending rather than assuming a + // fixed amount does it. Nothing is reading yet, so this terminates. buf, mem_err := make([]byte, mem.Gigabyte*4, context.temp_allocator) ev(t, mem_err, nil) // Use `core:net` as example external code that doesn't care about the event loop. net.set_blocking(op.poll.socket, false) - n, send_err := net.send(op.poll.socket, buf) + + n: int + send_err: net.Network_Error + for _ in 0..<16 { + sent: int + sent, send_err = net.send(op.poll.socket, buf) + n += sent + if send_err != nil { + break + } + } ev(t, send_err, net.TCP_Send_Error.Would_Block) log.debugf("blocking after %M", n) From fe52f23b502c7ca7132dd16a13674685e5865679 Mon Sep 17 00:00:00 2001 From: kalsprite Date: Fri, 21 Aug 2026 15:25:18 -0700 Subject: [PATCH 10/10] tests/core/nbio/remove.odin --- tests/core/nbio/remove.odin | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/tests/core/nbio/remove.odin b/tests/core/nbio/remove.odin index 37d32c5e1..fe7884c28 100644 --- a/tests/core/nbio/remove.odin +++ b/tests/core/nbio/remove.odin @@ -58,7 +58,12 @@ immediate_remove_of_sendfile :: proc(t: ^testing.T) { } on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) { - ev(t, op.recv.err, nil) + // The server cancelled a sendfile that had already put bytes on the wire and + // then closed, which ends the connection with a reset rather than gracefully + // often enough that both have to be accepted here. + if op.recv.err != nil { + ev(t, op.recv.err, net.TCP_Recv_Error.Connection_Closed) + } nbio.close(op.recv.socket.(net.TCP_Socket)) } @@ -126,7 +131,12 @@ immediate_remove_of_sendfile_without_stat :: proc(t: ^testing.T) { } on_recv :: proc(op: ^nbio.Operation, t: ^testing.T) { - ev(t, op.recv.err, nil) + // The server cancelled a sendfile that had already put bytes on the wire and + // then closed, which ends the connection with a reset rather than gracefully + // often enough that both have to be accepted here. + if op.recv.err != nil { + ev(t, op.recv.err, net.TCP_Recv_Error.Connection_Closed) + } nbio.close(op.recv.socket.(net.TCP_Socket)) }