From 57dcc861024a0f6b27e4495528c95716d99d80cf Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Thu, 9 Jul 2026 00:39:32 -0400 Subject: [PATCH 1/6] fix: preserve queued UDP datagrams on overflow When the fixed internal AF_INET/AF_INET6 receive queue is full, discard the incoming datagram and preserve the order of datagrams already accepted. Keep AF_UNIX on its existing bounded path because reliable Unix datagrams need a separate sender-backpressure design. This changes socket overflow semantics without adding structural ABI entries. It is quarantined in Batch 2 for the planned ABI 16-to-17 semantic reconciliation. --- crates/kernel/src/syscalls.rs | 79 ++++++++++++++++++++++++++++++++++- 1 file changed, 78 insertions(+), 1 deletion(-) diff --git a/crates/kernel/src/syscalls.rs b/crates/kernel/src/syscalls.rs index c2037a712..dbd3cc5dc 100644 --- a/crates/kernel/src/syscalls.rs +++ b/crates/kernel/src/syscalls.rs @@ -6541,6 +6541,19 @@ fn udp_socket_accepts_datagram( } fn udp_queue_datagram(sock: &mut crate::socket::SocketInfo, datagram: crate::socket::Datagram) { + // This is a fixed internal queue limit; SO_RCVBUF is currently advisory. + // UDP is unreliable, so a full receive queue drops the incoming datagram + // while preserving the order of datagrams that were already accepted. + if sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT { + return; + } + sock.dgram_queue.push(datagram); +} + +fn unix_queue_datagram(sock: &mut crate::socket::SocketInfo, datagram: crate::socket::Datagram) { + // Preserve the existing bounded AF_UNIX behavior. Unlike UDP, Unix + // datagrams are reliable on Linux and ultimately need sender backpressure + // rather than silent tail-drop; that requires a separate blocking design. if sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT { sock.dgram_queue.remove(0); } @@ -6943,7 +6956,7 @@ fn unix_dgram_send_to_sock( { return Err(Errno::ECONNREFUSED); } - udp_queue_datagram(target, datagram); + unix_queue_datagram(target, datagram); Ok(buf.len()) } @@ -20385,6 +20398,70 @@ mod tests { } } + #[test] + fn test_udp_recv_queue_tail_drops_on_overflow() { + use wasm_posix_shared::socket::*; + + let mut proc = Process::new(1); + let mut host = MockHostIO::new(); + + // Receiver bound to 127.0.0.1:ephemeral. + let recv_fd = sys_socket(&mut proc, &mut host, AF_INET, SOCK_DGRAM, 0).unwrap(); + let mut baddr = [0u8; 16]; + baddr[0] = 2; + sys_bind(&mut proc, &mut host, recv_fd, &baddr).unwrap(); + let mut gsa = [0u8; 16]; + sys_getsockname(&proc, recv_fd, &mut gsa).unwrap(); + let port = [gsa[2], gsa[3]]; + + // Sender. + let send_fd = sys_socket(&mut proc, &mut host, AF_INET, SOCK_DGRAM, 0).unwrap(); + let mut saddr = [0u8; 16]; + saddr[0] = 2; + sys_bind(&mut proc, &mut host, send_fd, &saddr).unwrap(); + + // Send LIMIT + 2 sequence-numbered datagrams to 127.0.0.1:port. + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT + 2 { + let mut dest = [0u8; 16]; + dest[0] = 2; + dest[2] = port[0]; + dest[3] = port[1]; + dest[4] = 127; + dest[7] = 1; + assert_eq!( + sys_sendto( + &mut proc, + &mut host, + send_fd, + &(sequence as u32).to_le_bytes(), + 0, + &dest, + ) + .unwrap(), + 4, + ); + } + + // The already-queued datagrams survive in order. + for expected in 0..UDP_DATAGRAM_QUEUE_LIMIT { + let mut buf = [0u8; 4]; + let mut from = [0u8; 16]; + let (n, from_len) = + sys_recvfrom(&mut proc, &mut host, recv_fd, &mut buf, 0, &mut from).unwrap(); + assert_eq!(n, 4); + assert_eq!(from_len, 16); + assert_eq!(u32::from_le_bytes(buf), expected as u32); + } + + // The two incoming datagrams sent after the queue filled were dropped. + let mut buf = [0u8; 4]; + let mut from = [0u8; 16]; + assert_eq!( + sys_recvfrom(&mut proc, &mut host, recv_fd, &mut buf, 0, &mut from).unwrap_err(), + Errno::EAGAIN, + ); + } + // ── Threading tests ────────────────────────────────────────────── #[test] From 72e3ada15d5996d75d5638674542ff7498259c32 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Fri, 10 Jul 2026 22:10:53 -0400 Subject: [PATCH 2/6] docs: state datagram overflow boundaries --- docs/architecture.md | 8 ++++++++ docs/posix-status.md | 3 ++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/docs/architecture.md b/docs/architecture.md index 0bdd4f55b..86d6eb0bf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -506,6 +506,14 @@ Offset Size Field User-visible networking is POSIX-first. Guest programs call normal AF_UNIX, AF_INET, and partial AF_INET6 socket syscalls (`socket`, `bind`, `connect`, `listen`, `accept`, `send`, `recv`, `sendto`, `recvfrom`, `poll`, and `select`). The Rust kernel owns the socket file descriptors, datagram queues, stream listener state, loopback routing, and errno behavior. Host transports plug in below that layer through `NetworkIO`; they are backends, not the userspace-visible abstraction. +AF_INET and AF_INET6 receive queues are currently bounded at 128 datagrams per +socket. Once that fixed internal queue is full, a newly arriving UDP datagram +is dropped and the already-queued datagrams retain their order. `SO_RCVBUF` +is stored and reported but does not size this queue yet. AF_UNIX datagrams use +the same bounded storage, but their overflow path still evicts the oldest +message instead of applying the sender backpressure required for Linux-style +reliable Unix datagrams. + Loopback addresses are scoped to one Kandelo machine, but not every socket path is machine-wide yet. IPv4 and IPv6 loopback TCP and AF_UNIX streams have explicit cross-process paths. Current in-kernel IPv4/IPv6 loopback datagrams, AF_UNIX datagrams, and IPv4 multicast delivery are confined to the sending process. Forked sockets retain their kernel-local bind reservations and local lookup targets, but host-backed UDP endpoint registrations are not yet shared or transferred between processes. AF_INET6 represents `sockaddr_in6`, supports `::`/`::1`, and models dual-stack wildcard stream-port reservation, but it has no external or virtual-network IPv6 transport and no IPv6 multicast delivery. AF_INET6 datagrams therefore report `IPV6_V6ONLY=1`; disabling it fails until dual-stack datagram routing exists. Routed virtual IPv4 addresses are explicit backend addresses. For example, the browser network lab attaches separate machines to addresses such as `10.88.0.2`, `10.88.0.3`, and `10.88.0.4`; traffic to `127.0.0.1` stays inside one machine, while traffic to those virtual addresses can cross machines through the backend. diff --git a/docs/posix-status.md b/docs/posix-status.md index e62197a9b..2b2c458c3 100644 --- a/docs/posix-status.md +++ b/docs/posix-status.md @@ -235,7 +235,7 @@ shortcuts. | `accept()` / `accept4()` | Partial | AF_INET TCP delegates to the active HostIO networking backend; AF_UNIX and AF_INET6 loopback streams return connected sockets from kernel queues. A dual-stack IPv6 listener reports IPv4 peers as IPv4-mapped `sockaddr_in6`. Linux-style accept does not inherit O_NONBLOCK; accept4 applies SOCK_NONBLOCK and SOCK_CLOEXEC explicitly and rejects other flags before consuming a pending connection. Datagram accept rejects as unsupported. | | `connect()` | Partial | AF_UNIX streams support same- and cross-process pathname or abstract-namespace listeners. AF_UNIX datagrams deliver to a registered peer only within the same process; a missing, wrong-type, or cross-process peer returns ECONNREFUSED until machine-wide datagram routing exists. AF_INET TCP is host-backed and works over Node external TCP or the browser local virtual-network backend. AF_INET UDP connect stores the peer, auto-binds an ephemeral local port when needed, filters receives to the connected peer, and supports AF_UNSPEC unconnect. AF_INET6 streams support same- and cross-process `::1`; AF_INET6 datagrams are process-local and report `IPV6_V6ONLY=1` because dual-stack datagram routing is not implemented. Non-loopback IPv6 fails with EADDRNOTAVAIL for streams and ENETUNREACH for datagrams. External raw UDP also returns ENETUNREACH without another HostIO transport. | | `send()` / `recv()` | Partial | Unix domain streams and datagrams, AF_INET/AF_INET6 TCP streams, and connected AF_INET/AF_INET6 UDP preserve their socket-family addressing and datagram boundaries. TCP send/recv works over Node external TCP and the local virtual-network backend. Datagram MSG_PEEK and MSG_DONTWAIT are handled through recvfrom. Normal TCP close drains queued bytes before FIN and EOF; no transport invents a fixed post-FIN write count. A send rejected by a closed/reset stream returns EPIPE and raises SIGPIPE, while direct host/virtual handles may preserve ECONNRESET; accepted pipe-bridged resets currently surface as EOF/EPIPE. MSG_NOSIGNAL suppresses SIGPIPE without changing the errno. | -| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | +| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` does not size that fixed queue yet. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | | `setsockopt()` / `getsockopt()` | Partial | SOL_SOCKET exposes SO_TYPE, SO_DOMAIN, SO_ERROR, SO_ACCEPTCONN, SO_RCVBUF, and SO_SNDBUF; SO_REUSEADDR affects UDP bind conflicts. SO_LINGER uses `struct linger`; its disabled form is stored, while enabling timed or reset-style linger returns EOPNOTSUPP until every transport supports the close mode. SO_BINDTODEVICE validates `lo`/`eth0`, supports empty-name unbind, and constrains bind/connect/send routing. TCP_CONGESTION uses a string layout and accepts only the modeled `cubic` policy; selecting unimplemented algorithms fails. IPv4 multicast membership/source-filter options drive process-local loopback delivery. IPV6_V6ONLY controls pre-bind stream dual-stack behavior; AF_INET6 datagrams truthfully remain V6-only. Other accepted IPv6 multicast options are stored but do not provide IPv6 multicast transport. | | `shutdown()` | Partial | SHUT_RD, SHUT_WR, and SHUT_RDWR transitions are idempotent within a process and release each owned pipe/host reference once. UDP write shutdown returns EPIPE on datagram send; read shutdown is EOF-like for recv/poll. Fork-inherited sockets still clone shutdown flags per process instead of sharing one socket-wide shutdown state, and the external host ABI has no half-shutdown operation. | | `select()` | Partial | Wrapper around poll(). Converts fd_set bitmasks to pollfd array. Timeout supported via polling loop. | @@ -405,6 +405,7 @@ Systematic audit of all subsystems against POSIX specifications. Gaps are catego | **setpgid() self-only** | process | Only supports setting own pgid. Setting another process's pgid returns ESRCH. | | ~~**realpath() no symlink resolution**~~ | filesystem | **Resolved.** Now resolves symlinks via iterative lstat/readlink with ELOOP after 40 resolutions. | | **Socket options partially no-op** | socket | SO_REUSEADDR affects UDP bind conflicts. SO_KEEPALIVE, SO_BROADCAST, SO_RCVTIMEO, SO_SNDTIMEO, and TCP_NODELAY are accepted/stored but have limited or no effect on data transfer. Enabled SO_LINGER is rejected rather than stored as a no-op. | +| **AF_UNIX datagram overflow is lossy** | socket | AF_UNIX datagrams preserve message boundaries and order under ordinary load, but their fixed 128-message receive queue currently evicts the oldest message on overflow instead of blocking the sender (or returning EAGAIN for a nonblocking send) as a reliable Unix datagram transport should. | | **POLLERR partial** | I/O multiplex | poll() reports UDP pending socket errors and stream shutdown/error cases. Some edge cases remain implementation-defined. | | **pread/pwrite not multi-process safe** | I/O | Uses save/seek/read/restore pattern — safe only when no other process shares the OFD, but races with shared OFDs across processes. | | ~~**brk not inherited on fork**~~ | memory | **Resolved.** Program break serialized/deserialized in fork state. (`exec` reset is intentional per POSIX; host re-installs from new program's `__heap_base`.) | From f5f71d48832de6ab96e54d0700140c156c8ebca0 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Fri, 10 Jul 2026 22:15:11 -0400 Subject: [PATCH 3/6] fix: backpressure full Unix datagram queues Unix datagrams are reliable, so a full receive queue now returns EAGAIN into the host blocking-retry path instead of discarding a message. Connected senders also stop advertising POLLOUT until their peer drains space. --- crates/kernel/src/syscalls.rs | 127 +++++++++++++++++++++++++++++++--- 1 file changed, 119 insertions(+), 8 deletions(-) diff --git a/crates/kernel/src/syscalls.rs b/crates/kernel/src/syscalls.rs index dbd3cc5dc..d32111469 100644 --- a/crates/kernel/src/syscalls.rs +++ b/crates/kernel/src/syscalls.rs @@ -6550,14 +6550,18 @@ fn udp_queue_datagram(sock: &mut crate::socket::SocketInfo, datagram: crate::soc sock.dgram_queue.push(datagram); } -fn unix_queue_datagram(sock: &mut crate::socket::SocketInfo, datagram: crate::socket::Datagram) { - // Preserve the existing bounded AF_UNIX behavior. Unlike UDP, Unix - // datagrams are reliable on Linux and ultimately need sender backpressure - // rather than silent tail-drop; that requires a separate blocking design. +fn unix_queue_datagram( + sock: &mut crate::socket::SocketInfo, + datagram: crate::socket::Datagram, +) -> Result<(), Errno> { + // AF_UNIX datagrams are reliable. EAGAIN enters the host's ordinary + // blocking-write retry path, while O_NONBLOCK or MSG_DONTWAIT exposes it + // directly to the caller. if sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT { - sock.dgram_queue.remove(0); + return Err(Errno::EAGAIN); } sock.dgram_queue.push(datagram); + Ok(()) } fn udp_take_socket_error(proc: &mut Process, sock_idx: usize) -> Result<(), Errno> { @@ -6956,7 +6960,7 @@ fn unix_dgram_send_to_sock( { return Err(Errno::ECONNREFUSED); } - unix_queue_datagram(target, datagram); + unix_queue_datagram(target, datagram)?; Ok(buf.len()) } @@ -9349,7 +9353,19 @@ fn poll_check(proc: &mut Process, host: &mut dyn HostIO, fds: &mut [WasmPollFd]) { revents |= POLLIN; } - if pollfd.events & POLLOUT != 0 && !sock.shut_wr { + let unix_peer_queue_full = sock.domain + == crate::socket::SocketDomain::Unix + && sock.state == SocketState::Connected + && sock + .peer_idx + .and_then(|peer_idx| proc.sockets.get(peer_idx)) + .is_some_and(|peer| { + peer.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT + }); + if pollfd.events & POLLOUT != 0 + && !sock.shut_wr + && !unix_peer_queue_full + { revents |= POLLOUT; } } @@ -20108,6 +20124,99 @@ mod tests { unsafe { crate::unix_socket::global_unix_socket_registry() }.cleanup_process(9025); } + #[test] + fn test_unix_dgram_full_queue_backpressures_without_reordering() { + let _lock = UNIX_REGISTRY_LOCK.lock().unwrap(); + let mut proc = Process::new(9052); + let mut host = MockHostIO::new(); + use wasm_posix_shared::socket::*; + + let recv_fd = sys_socket(&mut proc, &mut host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); + let mut addr = [0u8; 64]; + addr[0] = AF_UNIX as u8; + let path = b"/tmp/udg-overflow.sock"; + addr[2..2 + path.len()].copy_from_slice(path); + let addr = &addr[..2 + path.len() + 1]; + sys_bind(&mut proc, &mut host, recv_fd, addr).unwrap(); + + let send_fd = sys_socket(&mut proc, &mut host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); + sys_connect(&mut proc, &mut host, send_fd, addr).unwrap(); + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT { + assert_eq!( + sys_write( + &mut proc, + &mut host, + send_fd, + &(sequence as u32).to_le_bytes(), + ) + .unwrap(), + 4, + ); + } + assert_eq!( + sys_write( + &mut proc, + &mut host, + send_fd, + &(UDP_DATAGRAM_QUEUE_LIMIT as u32).to_le_bytes(), + ) + .unwrap_err(), + Errno::EAGAIN, + ); + + use wasm_posix_shared::poll::POLLOUT; + let mut pollfd = WasmPollFd { + fd: send_fd, + events: POLLOUT, + revents: 0, + }; + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 0, + ); + assert_eq!(pollfd.revents, 0); + + let mut buf = [0u8; 4]; + assert_eq!(sys_read(&mut proc, &mut host, recv_fd, &mut buf).unwrap(), 4); + assert_eq!(u32::from_le_bytes(buf), 0); + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 1, + ); + assert_ne!(pollfd.revents & POLLOUT, 0); + assert_eq!( + sys_write( + &mut proc, + &mut host, + send_fd, + &(UDP_DATAGRAM_QUEUE_LIMIT as u32).to_le_bytes(), + ) + .unwrap(), + 4, + ); + + for expected in 1..=UDP_DATAGRAM_QUEUE_LIMIT { + assert_eq!(sys_read(&mut proc, &mut host, recv_fd, &mut buf).unwrap(), 4); + assert_eq!(u32::from_le_bytes(buf), expected as u32); + } + + sys_close(&mut proc, &mut host, send_fd).unwrap(); + sys_close(&mut proc, &mut host, recv_fd).unwrap(); + unsafe { crate::unix_socket::global_unix_socket_registry() }.cleanup_process(9052); + } + #[test] fn test_unix_dgram_rejects_stream_target() { let _lock = UNIX_REGISTRY_LOCK.lock().unwrap(); @@ -20402,7 +20511,7 @@ mod tests { fn test_udp_recv_queue_tail_drops_on_overflow() { use wasm_posix_shared::socket::*; - let mut proc = Process::new(1); + let mut proc = Process::new(9051); let mut host = MockHostIO::new(); // Receiver bound to 127.0.0.1:ephemeral. @@ -20460,6 +20569,8 @@ mod tests { sys_recvfrom(&mut proc, &mut host, recv_fd, &mut buf, 0, &mut from).unwrap_err(), Errno::EAGAIN, ); + sys_close(&mut proc, &mut host, send_fd).unwrap(); + sys_close(&mut proc, &mut host, recv_fd).unwrap(); } // ── Threading tests ────────────────────────────────────────────── From 95df61f82d96dfa62e3140dddc2629e790342bd1 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Fri, 10 Jul 2026 22:15:11 -0400 Subject: [PATCH 4/6] docs: record reliable Unix datagram backpressure --- docs/architecture.md | 6 +++--- docs/posix-status.md | 3 +-- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 86d6eb0bf..e1bbcefd7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -510,9 +510,9 @@ AF_INET and AF_INET6 receive queues are currently bounded at 128 datagrams per socket. Once that fixed internal queue is full, a newly arriving UDP datagram is dropped and the already-queued datagrams retain their order. `SO_RCVBUF` is stored and reported but does not size this queue yet. AF_UNIX datagrams use -the same bounded storage, but their overflow path still evicts the oldest -message instead of applying the sender backpressure required for Linux-style -reliable Unix datagrams. +the same bounded storage but are reliable: a full receive queue makes the send +enter the host's blocking retry path, or returns `EAGAIN` immediately for an +`O_NONBLOCK` or `MSG_DONTWAIT` send, without discarding queued messages. Loopback addresses are scoped to one Kandelo machine, but not every socket path is machine-wide yet. IPv4 and IPv6 loopback TCP and AF_UNIX streams have explicit cross-process paths. Current in-kernel IPv4/IPv6 loopback datagrams, AF_UNIX datagrams, and IPv4 multicast delivery are confined to the sending process. Forked sockets retain their kernel-local bind reservations and local lookup targets, but host-backed UDP endpoint registrations are not yet shared or transferred between processes. AF_INET6 represents `sockaddr_in6`, supports `::`/`::1`, and models dual-stack wildcard stream-port reservation, but it has no external or virtual-network IPv6 transport and no IPv6 multicast delivery. AF_INET6 datagrams therefore report `IPV6_V6ONLY=1`; disabling it fails until dual-stack datagram routing exists. diff --git a/docs/posix-status.md b/docs/posix-status.md index 2b2c458c3..9a17faacf 100644 --- a/docs/posix-status.md +++ b/docs/posix-status.md @@ -235,7 +235,7 @@ shortcuts. | `accept()` / `accept4()` | Partial | AF_INET TCP delegates to the active HostIO networking backend; AF_UNIX and AF_INET6 loopback streams return connected sockets from kernel queues. A dual-stack IPv6 listener reports IPv4 peers as IPv4-mapped `sockaddr_in6`. Linux-style accept does not inherit O_NONBLOCK; accept4 applies SOCK_NONBLOCK and SOCK_CLOEXEC explicitly and rejects other flags before consuming a pending connection. Datagram accept rejects as unsupported. | | `connect()` | Partial | AF_UNIX streams support same- and cross-process pathname or abstract-namespace listeners. AF_UNIX datagrams deliver to a registered peer only within the same process; a missing, wrong-type, or cross-process peer returns ECONNREFUSED until machine-wide datagram routing exists. AF_INET TCP is host-backed and works over Node external TCP or the browser local virtual-network backend. AF_INET UDP connect stores the peer, auto-binds an ephemeral local port when needed, filters receives to the connected peer, and supports AF_UNSPEC unconnect. AF_INET6 streams support same- and cross-process `::1`; AF_INET6 datagrams are process-local and report `IPV6_V6ONLY=1` because dual-stack datagram routing is not implemented. Non-loopback IPv6 fails with EADDRNOTAVAIL for streams and ENETUNREACH for datagrams. External raw UDP also returns ENETUNREACH without another HostIO transport. | | `send()` / `recv()` | Partial | Unix domain streams and datagrams, AF_INET/AF_INET6 TCP streams, and connected AF_INET/AF_INET6 UDP preserve their socket-family addressing and datagram boundaries. TCP send/recv works over Node external TCP and the local virtual-network backend. Datagram MSG_PEEK and MSG_DONTWAIT are handled through recvfrom. Normal TCP close drains queued bytes before FIN and EOF; no transport invents a fixed post-FIN write count. A send rejected by a closed/reset stream returns EPIPE and raises SIGPIPE, while direct host/virtual handles may preserve ECONNRESET; accepted pipe-bridged resets currently surface as EOF/EPIPE. MSG_NOSIGNAL suppresses SIGPIPE without changing the errno. | -| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` does not size that fixed queue yet. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | +| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` does not size that fixed queue yet. AF_UNIX uses the same bound but preserves reliable delivery: a full queue blocks a blocking send through host retry and returns EAGAIN for `O_NONBLOCK`/`MSG_DONTWAIT`. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | | `setsockopt()` / `getsockopt()` | Partial | SOL_SOCKET exposes SO_TYPE, SO_DOMAIN, SO_ERROR, SO_ACCEPTCONN, SO_RCVBUF, and SO_SNDBUF; SO_REUSEADDR affects UDP bind conflicts. SO_LINGER uses `struct linger`; its disabled form is stored, while enabling timed or reset-style linger returns EOPNOTSUPP until every transport supports the close mode. SO_BINDTODEVICE validates `lo`/`eth0`, supports empty-name unbind, and constrains bind/connect/send routing. TCP_CONGESTION uses a string layout and accepts only the modeled `cubic` policy; selecting unimplemented algorithms fails. IPv4 multicast membership/source-filter options drive process-local loopback delivery. IPV6_V6ONLY controls pre-bind stream dual-stack behavior; AF_INET6 datagrams truthfully remain V6-only. Other accepted IPv6 multicast options are stored but do not provide IPv6 multicast transport. | | `shutdown()` | Partial | SHUT_RD, SHUT_WR, and SHUT_RDWR transitions are idempotent within a process and release each owned pipe/host reference once. UDP write shutdown returns EPIPE on datagram send; read shutdown is EOF-like for recv/poll. Fork-inherited sockets still clone shutdown flags per process instead of sharing one socket-wide shutdown state, and the external host ABI has no half-shutdown operation. | | `select()` | Partial | Wrapper around poll(). Converts fd_set bitmasks to pollfd array. Timeout supported via polling loop. | @@ -405,7 +405,6 @@ Systematic audit of all subsystems against POSIX specifications. Gaps are catego | **setpgid() self-only** | process | Only supports setting own pgid. Setting another process's pgid returns ESRCH. | | ~~**realpath() no symlink resolution**~~ | filesystem | **Resolved.** Now resolves symlinks via iterative lstat/readlink with ELOOP after 40 resolutions. | | **Socket options partially no-op** | socket | SO_REUSEADDR affects UDP bind conflicts. SO_KEEPALIVE, SO_BROADCAST, SO_RCVTIMEO, SO_SNDTIMEO, and TCP_NODELAY are accepted/stored but have limited or no effect on data transfer. Enabled SO_LINGER is rejected rather than stored as a no-op. | -| **AF_UNIX datagram overflow is lossy** | socket | AF_UNIX datagrams preserve message boundaries and order under ordinary load, but their fixed 128-message receive queue currently evicts the oldest message on overflow instead of blocking the sender (or returning EAGAIN for a nonblocking send) as a reliable Unix datagram transport should. | | **POLLERR partial** | I/O multiplex | poll() reports UDP pending socket errors and stream shutdown/error cases. Some edge cases remain implementation-defined. | | **pread/pwrite not multi-process safe** | I/O | Uses save/seek/read/restore pattern — safe only when no other process shares the OFD, but races with shared OFDs across processes. | | ~~**brk not inherited on fork**~~ | memory | **Resolved.** Program break serialized/deserialized in fork state. (`exec` reset is intentional per POSIX; host re-installs from new program's `__heap_base`.) | From b6d7ffbd2f4f1b1c6120da9a6b0b1dab1be55e96 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Fri, 10 Jul 2026 22:21:34 -0400 Subject: [PATCH 5/6] test: exercise datagram overflow through the guest ABI --- .../udp.expect/queue-overflow-tail-drop.posix | 1 + .../unix-dgram-queue-backpressure.posix | 1 + .../udp/queue-overflow-tail-drop.c | 70 ++++++++++++++ .../udp/unix-dgram-queue-backpressure.c | 92 +++++++++++++++++++ 4 files changed, 164 insertions(+) create mode 100644 tests/sortix/os-test-local/udp.expect/queue-overflow-tail-drop.posix create mode 100644 tests/sortix/os-test-local/udp.expect/unix-dgram-queue-backpressure.posix create mode 100644 tests/sortix/os-test-local/udp/queue-overflow-tail-drop.c create mode 100644 tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c diff --git a/tests/sortix/os-test-local/udp.expect/queue-overflow-tail-drop.posix b/tests/sortix/os-test-local/udp.expect/queue-overflow-tail-drop.posix new file mode 100644 index 000000000..9766475a4 --- /dev/null +++ b/tests/sortix/os-test-local/udp.expect/queue-overflow-tail-drop.posix @@ -0,0 +1 @@ +ok diff --git a/tests/sortix/os-test-local/udp.expect/unix-dgram-queue-backpressure.posix b/tests/sortix/os-test-local/udp.expect/unix-dgram-queue-backpressure.posix new file mode 100644 index 000000000..9766475a4 --- /dev/null +++ b/tests/sortix/os-test-local/udp.expect/unix-dgram-queue-backpressure.posix @@ -0,0 +1 @@ +ok diff --git a/tests/sortix/os-test-local/udp/queue-overflow-tail-drop.c b/tests/sortix/os-test-local/udp/queue-overflow-tail-drop.c new file mode 100644 index 000000000..e3f04a0e2 --- /dev/null +++ b/tests/sortix/os-test-local/udp/queue-overflow-tail-drop.c @@ -0,0 +1,70 @@ +/* + * Fill Kandelo's documented 128-datagram UDP receive queue, then verify that + * later arrivals are dropped without evicting or reordering accepted data. + */ + +#include "udp.h" + +#include + +int main(void) +{ + int recv_fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + if ( recv_fd < 0 ) + err(1, "receiver socket"); + + struct sockaddr_in recv_addr; + memset(&recv_addr, 0, sizeof(recv_addr)); + recv_addr.sin_family = AF_INET; + recv_addr.sin_addr.s_addr = htobe32(INADDR_LOOPBACK); + if ( bind(recv_fd, (const struct sockaddr*) &recv_addr, + sizeof(recv_addr)) < 0 ) + err(1, "receiver bind"); + + socklen_t recv_addr_len = sizeof(recv_addr); + if ( getsockname(recv_fd, (struct sockaddr*) &recv_addr, + &recv_addr_len) < 0 ) + err(1, "receiver getsockname"); + + int send_fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + if ( send_fd < 0 ) + err(1, "sender socket"); + + for ( uint32_t sequence = 0; sequence < 130; sequence++ ) + { + uint32_t payload = htobe32(sequence); + ssize_t amount = sendto(send_fd, &payload, sizeof(payload), 0, + (const struct sockaddr*) &recv_addr, + recv_addr_len); + if ( amount < 0 ) + err(1, "sendto"); + if ( amount != (ssize_t) sizeof(payload) ) + errx(1, "sendto returned %zi", amount); + } + + for ( uint32_t expected = 0; expected < 128; expected++ ) + { + uint32_t payload = 0; + ssize_t amount = recv(recv_fd, &payload, sizeof(payload), + MSG_DONTWAIT); + if ( amount < 0 ) + err(1, "recv sequence %u", expected); + if ( amount != (ssize_t) sizeof(payload) ) + errx(1, "recv returned %zi", amount); + uint32_t actual = be32toh(payload); + if ( actual != expected ) + errx(1, "sequence %u arrived as %u", expected, actual); + } + + uint32_t payload = 0; + errno = 0; + if ( recv(recv_fd, &payload, sizeof(payload), MSG_DONTWAIT) != -1 ) + errx(1, "overflow datagram unexpectedly remained queued"); + if ( errno != EAGAIN && errno != EWOULDBLOCK ) + err(1, "final recv"); + + if ( close(send_fd) < 0 || close(recv_fd) < 0 ) + err(1, "close"); + puts("ok"); + return 0; +} diff --git a/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c b/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c new file mode 100644 index 000000000..42869b983 --- /dev/null +++ b/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c @@ -0,0 +1,92 @@ +/* + * AF_UNIX datagrams are reliable: a full receive queue must make a + * nonblocking sender report EAGAIN without discarding or reordering messages. + */ + +#include "udp.h" + +#include +#include +#include + +int main(void) +{ + const char* path = "/tmp/kandelo-unix-dgram-overflow.sock"; + if ( unlink(path) < 0 && errno != ENOENT ) + err(1, "unlink before bind"); + + int recv_fd = socket(AF_UNIX, SOCK_DGRAM, 0); + if ( recv_fd < 0 ) + err(1, "receiver socket"); + + struct sockaddr_un addr; + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1); + socklen_t addr_len = + (socklen_t) (offsetof(struct sockaddr_un, sun_path) + + strlen(addr.sun_path) + 1); + if ( bind(recv_fd, (const struct sockaddr*) &addr, addr_len) < 0 ) + err(1, "receiver bind"); + + int send_fd = socket(AF_UNIX, SOCK_DGRAM, 0); + if ( send_fd < 0 ) + err(1, "sender socket"); + if ( connect(send_fd, (const struct sockaddr*) &addr, addr_len) < 0 ) + err(1, "sender connect"); + int flags = fcntl(send_fd, F_GETFL); + if ( flags < 0 || fcntl(send_fd, F_SETFL, flags | O_NONBLOCK) < 0 ) + err(1, "sender nonblock"); + + for ( uint32_t sequence = 0; sequence < 128; sequence++ ) + { + uint32_t payload = htobe32(sequence); + ssize_t amount = send(send_fd, &payload, sizeof(payload), 0); + if ( amount < 0 ) + err(1, "send sequence %u", sequence); + if ( amount != (ssize_t) sizeof(payload) ) + errx(1, "send returned %zi", amount); + } + + struct pollfd pfd = { .fd = send_fd, .events = POLLOUT, .revents = 0 }; + if ( poll(&pfd, 1, 0) != 0 || pfd.revents != 0 ) + errx(1, "full peer unexpectedly writable: revents=%#x", pfd.revents); + + uint32_t payload = htobe32(128); + errno = 0; + if ( send(send_fd, &payload, sizeof(payload), 0) != -1 ) + errx(1, "send to full peer unexpectedly succeeded"); + if ( errno != EAGAIN && errno != EWOULDBLOCK ) + err(1, "send to full peer"); + + uint32_t first = 0; + if ( recv(recv_fd, &first, sizeof(first), MSG_DONTWAIT) != + (ssize_t) sizeof(first) ) + err(1, "recv first"); + if ( be32toh(first) != 0 ) + errx(1, "first sequence was %u", be32toh(first)); + + pfd.revents = 0; + if ( poll(&pfd, 1, 0) != 1 || !(pfd.revents & POLLOUT) ) + errx(1, "drained peer not writable: revents=%#x", pfd.revents); + if ( send(send_fd, &payload, sizeof(payload), 0) != + (ssize_t) sizeof(payload) ) + err(1, "retry send"); + + for ( uint32_t expected = 1; expected <= 128; expected++ ) + { + uint32_t actual = 0; + if ( recv(recv_fd, &actual, sizeof(actual), MSG_DONTWAIT) != + (ssize_t) sizeof(actual) ) + err(1, "recv sequence %u", expected); + if ( be32toh(actual) != expected ) + errx(1, "sequence %u arrived as %u", expected, be32toh(actual)); + } + + if ( close(send_fd) < 0 || close(recv_fd) < 0 ) + err(1, "close"); + if ( unlink(path) < 0 ) + err(1, "unlink after close"); + puts("ok"); + return 0; +} From 6901aacc005e967863394c2e669f6d8e7484e851 Mon Sep 17 00:00:00 2001 From: Brandon Payton Date: Fri, 10 Jul 2026 23:07:18 -0400 Subject: [PATCH 6/6] fix: make Unix datagram backpressure reliable --- crates/kernel/src/syscalls.rs | 642 +++++++++++++++--- crates/kernel/src/wakeup.rs | 30 +- docs/architecture.md | 12 +- docs/posix-status.md | 4 +- host/src/kernel-worker.ts | 221 +++++- host/test/datagram-wakeup.test.ts | 77 +++ host/test/readiness-deadline.test.ts | 107 +++ .../udp/unix-dgram-queue-backpressure.c | 397 ++++++++++- 8 files changed, 1327 insertions(+), 163 deletions(-) create mode 100644 host/test/datagram-wakeup.test.ts create mode 100644 host/test/readiness-deadline.test.ts diff --git a/crates/kernel/src/syscalls.rs b/crates/kernel/src/syscalls.rs index d32111469..fd80c085c 100644 --- a/crates/kernel/src/syscalls.rs +++ b/crates/kernel/src/syscalls.rs @@ -2046,6 +2046,10 @@ pub fn sys_close(proc: &mut Process, host: &mut dyn HostIO, fd: i32) -> Result<( } FileType::Socket => { let sock_idx = (-(host_handle + 1)) as usize; + let unix_dgram_send_state_changed = proc.sockets.get(sock_idx).is_some_and(|sock| { + sock.domain == crate::socket::SocketDomain::Unix + && sock.sock_type == crate::socket::SocketType::Dgram + }); if let Some(sock) = proc.sockets.get(sock_idx) { if let Some(path) = sock.bind_path.as_deref() { let registry = unsafe { @@ -2131,6 +2135,13 @@ pub fn sys_close(proc: &mut Process, host: &mut dyn HostIO, fd: i32) -> Result<( } } proc.sockets.free(sock_idx); + // Closing either endpoint can invalidate a process-local + // AF_UNIX datagram association. Retry blocked sends and + // readiness waits so they observe ECONNREFUSED/EPERM instead + // of waiting until an unrelated timeout. + if unix_dgram_send_state_changed { + crate::wakeup::push_datagram_writable(); + } } FileType::EventFd => { // Free the eventfd state @@ -4205,6 +4216,10 @@ pub fn sys_unlink(proc: &mut Process, host: &mut dyn HostIO, path: &[u8]) -> Res { let registry = unsafe { crate::unix_socket::global_unix_socket_registry() }; if registry.unregister(&resolved) { + // A sendto(path) may be parked on a full AF_UNIX datagram queue. + // Once the name is removed, retry it so it observes the now-stale + // destination instead of sleeping until an unrelated timeout. + crate::wakeup::push_datagram_writable(); match host.host_unlink(&resolved) { Ok(()) | Err(Errno::ENOENT) => return Ok(()), Err(e) => return Err(e), @@ -6540,19 +6555,22 @@ fn udp_socket_accepts_datagram( true } -fn udp_queue_datagram(sock: &mut crate::socket::SocketInfo, datagram: crate::socket::Datagram) { +fn udp_queue_datagram( + sock: &mut crate::socket::SocketInfo, + make_datagram: impl FnOnce() -> crate::socket::Datagram, +) { // This is a fixed internal queue limit; SO_RCVBUF is currently advisory. // UDP is unreliable, so a full receive queue drops the incoming datagram // while preserving the order of datagrams that were already accepted. if sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT { return; } - sock.dgram_queue.push(datagram); + sock.dgram_queue.push(make_datagram()); } fn unix_queue_datagram( sock: &mut crate::socket::SocketInfo, - datagram: crate::socket::Datagram, + make_datagram: impl FnOnce() -> crate::socket::Datagram, ) -> Result<(), Errno> { // AF_UNIX datagrams are reliable. EAGAIN enters the host's ordinary // blocking-write retry path, while O_NONBLOCK or MSG_DONTWAIT exposes it @@ -6560,7 +6578,7 @@ fn unix_queue_datagram( if sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT { return Err(Errno::EAGAIN); } - sock.dgram_queue.push(datagram); + sock.dgram_queue.push(make_datagram()); Ok(()) } @@ -6787,6 +6805,7 @@ fn udp_send_datagram( } else { src_addr }; + let (src_pid, src_uid, src_gid) = (proc.pid, proc.uid, proc.gid); if is_ipv4_multicast_addr(dst_addr) { use wasm_posix_shared::socket::{IP_MULTICAST_IF, IP_MULTICAST_LOOP, IPPROTO_IP}; @@ -6819,21 +6838,6 @@ fn udp_send_datagram( if !loop_enabled { return Ok(buf.len()); } - let datagram = Datagram { - data: buf.to_vec(), - src_addr, - src_addr6: [0; 16], - dst_addr, - dst_addr6: [0; 16], - src_port, - src_sock_idx: Some(sock_idx), - ipv6_tclass: 0, - src_pid: proc.pid, - src_uid: proc.uid, - src_gid: proc.gid, - ancillary_fds: Vec::new(), - }; - let endpoints = crate::socket::udp_lookup(dst_addr, dst_port); for endpoint in endpoints { if endpoint.pid != proc.pid { @@ -6856,7 +6860,20 @@ fn udp_send_datagram( continue; } if let Some(target) = proc.sockets.get_mut(endpoint.sock_idx) { - udp_queue_datagram(target, datagram.clone()); + udp_queue_datagram(target, || Datagram { + data: buf.to_vec(), + src_addr, + src_addr6: [0; 16], + dst_addr, + dst_addr6: [0; 16], + src_port, + src_sock_idx: Some(sock_idx), + ipv6_tclass: 0, + src_pid, + src_uid, + src_gid, + ancillary_fds: Vec::new(), + }); } } return Ok(buf.len()); @@ -6874,21 +6891,6 @@ fn udp_send_datagram( Err(e) => Err(e), }; } - let datagram = Datagram { - data: buf.to_vec(), - src_addr, - src_addr6: [0; 16], - dst_addr, - dst_addr6: [0; 16], - src_port, - src_sock_idx: Some(sock_idx), - ipv6_tclass: 0, - src_pid: proc.pid, - src_uid: proc.uid, - src_gid: proc.gid, - ancillary_fds: Vec::new(), - }; - let mut delivered = false; let endpoints = crate::socket::udp_lookup(dst_addr, dst_port); for endpoint in endpoints { @@ -6904,7 +6906,20 @@ fn udp_send_datagram( continue; } if let Some(target) = proc.sockets.get_mut(endpoint.sock_idx) { - udp_queue_datagram(target, datagram.clone()); + udp_queue_datagram(target, || Datagram { + data: buf.to_vec(), + src_addr, + src_addr6: [0; 16], + dst_addr, + dst_addr6: [0; 16], + src_port, + src_sock_idx: Some(sock_idx), + ipv6_tclass: 0, + src_pid, + src_uid, + src_gid, + ancillary_fds: Vec::new(), + }); delivered = true; break; } @@ -6936,20 +6951,7 @@ fn unix_dgram_send_to_sock( return Err(Errno::EPIPE); } - let datagram = Datagram { - data: buf.to_vec(), - src_addr: [0; 4], - src_addr6: [0; 16], - dst_addr: [0; 4], - dst_addr6: [0; 16], - src_port: 0, - src_sock_idx: Some(src_sock_idx), - ipv6_tclass: 0, - src_pid: proc.pid, - src_uid: proc.uid, - src_gid: proc.gid, - ancillary_fds: Vec::new(), - }; + let (src_pid, src_uid, src_gid) = (proc.pid, proc.uid, proc.gid); let target = proc.sockets.get_mut(dst_sock_idx).ok_or(Errno::ECONNREFUSED)?; if target.domain != crate::socket::SocketDomain::Unix || target.sock_type != crate::socket::SocketType::Dgram @@ -6960,7 +6962,29 @@ fn unix_dgram_send_to_sock( { return Err(Errno::ECONNREFUSED); } - unix_queue_datagram(target, datagram)?; + if !unix_dgram_target_accepts_sender(target, src_sock_idx) { + return Err(Errno::EPERM); + } + // A read-shut AF_UNIX datagram peer cannot accept another reliable + // message. Report the broken association before constructing a payload; + // the send wrapper supplies SIGPIPE unless MSG_NOSIGNAL was requested. + if target.shut_rd { + return Err(Errno::EPIPE); + } + unix_queue_datagram(target, || Datagram { + data: buf.to_vec(), + src_addr: [0; 4], + src_addr6: [0; 16], + dst_addr: [0; 4], + dst_addr6: [0; 16], + src_port: 0, + src_sock_idx: Some(src_sock_idx), + ipv6_tclass: 0, + src_pid, + src_uid, + src_gid, + ancillary_fds: Vec::new(), + })?; Ok(buf.len()) } @@ -7057,6 +7081,23 @@ fn udp6_socket_accepts_datagram( true } +fn unix_dgram_target_accepts_sender( + target: &crate::socket::SocketInfo, + src_sock_idx: usize, +) -> bool { + use crate::socket::SocketState; + + target.state != SocketState::Connected || target.peer_idx == Some(src_sock_idx) +} + +fn unix_dgram_matches_peer( + owner_pid: u32, + peer_idx: usize, + datagram: &crate::socket::Datagram, +) -> bool { + datagram.src_pid == owner_pid && datagram.src_sock_idx == Some(peer_idx) +} + /// Whether a queued datagram is visible through this socket's connected-peer /// filter. Keep recv and poll on the same predicate so readiness cannot claim /// data that the following recv would reject. @@ -7077,12 +7118,9 @@ fn dgram_matches_connected_peer( SocketDomain::Inet6 => { datagram.src_addr6 == sock.peer_addr6 && datagram.src_port == sock.peer_port } - SocketDomain::Unix => { - datagram.src_pid == owner_pid - && sock - .peer_idx - .is_some_and(|peer| datagram.src_sock_idx == Some(peer)) - } + SocketDomain::Unix => sock + .peer_idx + .is_some_and(|peer| unix_dgram_matches_peer(owner_pid, peer, datagram)), } } @@ -7135,20 +7173,7 @@ fn udp6_send_datagram( } else { src_addr }; - let datagram = Datagram { - data: buf.to_vec(), - src_addr: [0; 4], - src_addr6: src_addr, - dst_addr: [0; 4], - dst_addr6: dst_addr, - src_port, - src_sock_idx: Some(sock_idx), - ipv6_tclass: 0, - src_pid: proc.pid, - src_uid: proc.uid, - src_gid: proc.gid, - ancillary_fds: Vec::new(), - }; + let (src_pid, src_uid, src_gid) = (proc.pid, proc.uid, proc.gid); let mut delivered = false; let sock_count = proc.sockets.len(); @@ -7168,7 +7193,20 @@ fn udp6_send_datagram( continue; } if let Some(target) = proc.sockets.get_mut(idx) { - udp_queue_datagram(target, datagram.clone()); + udp_queue_datagram(target, || Datagram { + data: buf.to_vec(), + src_addr: [0; 4], + src_addr6: src_addr, + dst_addr: [0; 4], + dst_addr6: dst_addr, + src_port, + src_sock_idx: Some(sock_idx), + ipv6_tclass: 0, + src_pid, + src_uid, + src_gid, + ancillary_fds: Vec::new(), + }); delivered = true; break; } @@ -7193,20 +7231,6 @@ pub fn inject_udp_datagram_into( ) -> i32 { use crate::socket::Datagram; - let datagram = Datagram { - data: data.to_vec(), - src_addr, - src_addr6: [0; 16], - dst_addr, - dst_addr6: [0; 16], - src_port, - src_sock_idx: None, - ipv6_tclass: 0, - src_pid: 0, - src_uid: 0, - src_gid: 0, - ancillary_fds: Vec::new(), - }; let endpoints = crate::socket::udp_lookup(dst_addr, dst_port); for endpoint in endpoints { if endpoint.pid != proc.pid { @@ -7221,7 +7245,20 @@ pub fn inject_udp_datagram_into( continue; } if let Some(target) = proc.sockets.get_mut(endpoint.sock_idx) { - udp_queue_datagram(target, datagram); + udp_queue_datagram(target, || Datagram { + data: data.to_vec(), + src_addr, + src_addr6: [0; 16], + dst_addr, + dst_addr6: [0; 16], + src_port, + src_sock_idx: None, + ipv6_tclass: 0, + src_pid: 0, + src_uid: 0, + src_gid: 0, + ancillary_fds: Vec::new(), + }); return 0; } } @@ -7347,16 +7384,30 @@ pub fn sys_shutdown( // later close, process exit, fork, or SCM_RIGHTS transfer from dropping or // resurrecting the same pipe reference. Explicit SHUT_RD is a hard receive // refusal; only normal close uses TCP's orderly orphaned-receive state. - let (send_idx, recv_idx, net_handle) = { + let (send_idx, recv_idx, net_handle, datagram_send_state_changed) = { let sock = proc.sockets.get_mut(sock_idx).ok_or(Errno::EBADF)?; + let is_unix_dgram = sock.domain == crate::socket::SocketDomain::Unix + && sock.sock_type == crate::socket::SocketType::Dgram; + let was_shut_rd = sock.shut_rd; + let was_shut_wr = sock.shut_wr; match how { SHUT_RD => { sock.shut_rd = true; - (None, sock.recv_buf_idx.take(), None) + ( + None, + sock.recv_buf_idx.take(), + None, + is_unix_dgram && !was_shut_rd, + ) } SHUT_WR => { sock.shut_wr = true; - (sock.send_buf_idx.take(), None, None) + ( + sock.send_buf_idx.take(), + None, + None, + is_unix_dgram && !was_shut_wr, + ) } SHUT_RDWR => { sock.shut_rd = true; @@ -7365,6 +7416,7 @@ pub fn sys_shutdown( sock.send_buf_idx.take(), sock.recv_buf_idx.take(), sock.host_net_handle.take(), + is_unix_dgram && (!was_shut_rd || !was_shut_wr), ) } _ => return Err(Errno::EINVAL), @@ -7393,6 +7445,12 @@ pub fn sys_shutdown( let _ = host.host_net_close(net_handle); } } + // AF_UNIX datagram writers and readiness waiters are not backed by a + // targetable pipe. A read-side or write-side shutdown changes whether the + // next send blocks or fails, so ask the host to retry them broadly. + if datagram_send_state_changed { + crate::wakeup::push_datagram_writable(); + } Ok(()) } @@ -8866,10 +8924,28 @@ pub fn sys_connect( { return Err(Errno::ECONNREFUSED); } + if !unix_dgram_target_accepts_sender(target, sock_idx) { + return Err(Errno::EPERM); + } - let sock = proc.sockets.get_mut(sock_idx).ok_or(Errno::EBADF)?; - sock.peer_idx = Some(peer_idx); - sock.state = SocketState::Connected; + let owner_pid = proc.pid; + let send_state_changed = { + let sock = proc.sockets.get_mut(sock_idx).ok_or(Errno::EBADF)?; + let association_changed = sock.state != SocketState::Connected + || sock.peer_idx != Some(peer_idx); + sock.peer_idx = Some(peer_idx); + sock.state = SocketState::Connected; + let queued_before = sock.dgram_queue.len(); + sock.dgram_queue + .retain(|datagram| unix_dgram_matches_peer(owner_pid, peer_idx, datagram)); + association_changed || sock.dgram_queue.len() != queued_before + }; + // Connecting changes both this socket's destination and which + // senders it accepts. Even if a selected peer's messages keep + // the queue full, rejected writers must wake to observe EPERM. + if send_state_changed { + crate::wakeup::push_datagram_writable(); + } return Ok(()); } if sock.state == SocketState::Connected { @@ -9099,11 +9175,16 @@ pub fn sys_recvfrom( return Err(Errno::EAGAIN); } let datagram_idx = datagram_idx.unwrap(); - let datagram = if _flags & MSG_PEEK != 0 { + let peek = _flags & MSG_PEEK != 0; + let was_full = sock.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT; + let datagram = if peek { sock.dgram_queue[datagram_idx].clone() } else { sock.dgram_queue.remove(datagram_idx) }; + if !peek && was_full && sock.domain == SocketDomain::Unix { + crate::wakeup::push_datagram_writable(); + } // Copy data to buffer let copy_len = buf.len().min(datagram.data.len()); @@ -9360,7 +9441,15 @@ fn poll_check(proc: &mut Process, host: &mut dyn HostIO, fds: &mut [WasmPollFd]) .peer_idx .and_then(|peer_idx| proc.sockets.get(peer_idx)) .is_some_and(|peer| { - peer.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT + peer.domain == crate::socket::SocketDomain::Unix + && peer.sock_type == SocketType::Dgram + && matches!( + peer.state, + SocketState::Bound | SocketState::Connected + ) + && !peer.shut_rd + && unix_dgram_target_accepts_sender(peer, sock_idx) + && peer.dgram_queue.len() >= UDP_DATAGRAM_QUEUE_LIMIT }); if pollfd.events & POLLOUT != 0 && !sock.shut_wr @@ -9771,6 +9860,7 @@ pub fn sys_unlinkat( { let registry = unsafe { crate::unix_socket::global_unix_socket_registry() }; if registry.unregister(&resolved) { + crate::wakeup::push_datagram_writable(); match host.host_unlink(&resolved) { Ok(()) | Err(Errno::ENOENT) => return Ok(()), Err(e) => return Err(e), @@ -12445,6 +12535,32 @@ mod tests { static UNIX_REGISTRY_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); static THREAD_IDENTITY_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + fn test_unix_addr(path: &[u8]) -> Vec { + let mut addr = vec![0; 2 + path.len() + 1]; + addr[0] = wasm_posix_shared::socket::AF_UNIX as u8; + addr[2..2 + path.len()].copy_from_slice(path); + addr + } + + fn bind_test_unix_dgram( + proc: &mut Process, + host: &mut dyn HostIO, + path: &[u8], + ) -> (i32, Vec) { + use wasm_posix_shared::socket::{AF_UNIX, SOCK_DGRAM}; + + let fd = sys_socket(proc, host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); + let addr = test_unix_addr(path); + sys_bind(proc, host, fd, &addr).unwrap(); + (fd, addr) + } + + fn test_socket_idx(proc: &Process, fd: i32) -> usize { + let entry = proc.fd_table.get(fd).unwrap(); + let ofd = proc.ofd_table.get(entry.ofd_ref.0).unwrap(); + (-(ofd.host_handle + 1)) as usize + } + fn set_test_current_tid(tid: u32) { unsafe { (*crate::process_table::GLOBAL_PROCESS_TABLE.0.get()).set_current_tid(tid); @@ -20132,15 +20248,77 @@ mod tests { use wasm_posix_shared::socket::*; let recv_fd = sys_socket(&mut proc, &mut host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); - let mut addr = [0u8; 64]; - addr[0] = AF_UNIX as u8; - let path = b"/tmp/udg-overflow.sock"; - addr[2..2 + path.len()].copy_from_slice(path); - let addr = &addr[..2 + path.len() + 1]; - sys_bind(&mut proc, &mut host, recv_fd, addr).unwrap(); + let recv_path = b"/tmp/udg-overflow-recv.sock"; + let mut recv_addr = [0u8; 64]; + recv_addr[0] = AF_UNIX as u8; + recv_addr[2..2 + recv_path.len()].copy_from_slice(recv_path); + let recv_addr = &recv_addr[..2 + recv_path.len() + 1]; + sys_bind(&mut proc, &mut host, recv_fd, recv_addr).unwrap(); let send_fd = sys_socket(&mut proc, &mut host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); - sys_connect(&mut proc, &mut host, send_fd, addr).unwrap(); + let send_path = b"/tmp/udg-overflow-send.sock"; + let mut send_addr = [0u8; 64]; + send_addr[0] = AF_UNIX as u8; + send_addr[2..2 + send_path.len()].copy_from_slice(send_path); + let send_addr = &send_addr[..2 + send_path.len() + 1]; + sys_bind(&mut proc, &mut host, send_fd, send_addr).unwrap(); + sys_connect(&mut proc, &mut host, send_fd, recv_addr).unwrap(); + + // A connected receiver admits only its chosen peer. Purge datagrams + // from other senders that arrived before connect, and reject later + // attempts so invisible messages cannot fill the reliable queue. + let attacker_fd = sys_socket(&mut proc, &mut host, AF_UNIX, SOCK_DGRAM, 0).unwrap(); + assert_eq!( + sys_sendto( + &mut proc, + &mut host, + attacker_fd, + b"pre-connect attacker", + 0, + recv_addr, + ) + .unwrap(), + 20, + ); + let recv_idx = { + let entry = proc.fd_table.get(recv_fd).unwrap(); + let ofd = proc.ofd_table.get(entry.ofd_ref.0).unwrap(); + (-(ofd.host_handle + 1)) as usize + }; + assert_eq!( + sys_write( + &mut proc, + &mut host, + send_fd, + b"peer-before-connect", + ) + .unwrap(), + 19, + ); + assert_eq!(proc.sockets.get(recv_idx).unwrap().dgram_queue.len(), 2); + sys_connect(&mut proc, &mut host, recv_fd, send_addr).unwrap(); + assert_eq!(proc.sockets.get(recv_idx).unwrap().dgram_queue.len(), 1); + let mut peer_payload = [0u8; 32]; + let peer_len = sys_read(&mut proc, &mut host, recv_fd, &mut peer_payload).unwrap(); + assert_eq!(&peer_payload[..peer_len], b"peer-before-connect"); + assert!(proc.sockets.get(recv_idx).unwrap().dgram_queue.is_empty()); + assert_eq!( + sys_connect(&mut proc, &mut host, attacker_fd, recv_addr).unwrap_err(), + Errno::EPERM, + ); + assert_eq!( + sys_sendto( + &mut proc, + &mut host, + attacker_fd, + b"post-connect attacker", + 0, + recv_addr, + ) + .unwrap_err(), + Errno::EPERM, + ); + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT { assert_eq!( sys_write( @@ -20153,6 +20331,12 @@ mod tests { 4, ); } + assert_eq!( + unix_queue_datagram(proc.sockets.get_mut(recv_idx).unwrap(), || { + panic!("a full Unix datagram queue must not construct a blocked payload") + }), + Err(Errno::EAGAIN), + ); assert_eq!( sys_write( &mut proc, @@ -20212,9 +20396,253 @@ mod tests { assert_eq!(u32::from_le_bytes(buf), expected as u32); } + sys_close(&mut proc, &mut host, attacker_fd).unwrap(); sys_close(&mut proc, &mut host, send_fd).unwrap(); sys_close(&mut proc, &mut host, recv_fd).unwrap(); - unsafe { crate::unix_socket::global_unix_socket_registry() }.cleanup_process(9052); + sys_unlink(&mut proc, &mut host, send_path).unwrap(); + sys_unlink(&mut proc, &mut host, recv_path).unwrap(); + } + + #[test] + fn test_unix_dgram_connect_releases_rejected_full_queue_sender() { + let _lock = UNIX_REGISTRY_LOCK.lock().unwrap(); + let mut proc = Process::new(9053); + let mut host = MockHostIO::new(); + use wasm_posix_shared::poll::POLLOUT; + + let recv_path = b"/tmp/udg-connect-state-recv.sock"; + let first_path = b"/tmp/udg-connect-state-first.sock"; + let selected_path = b"/tmp/udg-connect-state-selected.sock"; + let (recv_fd, recv_addr) = bind_test_unix_dgram(&mut proc, &mut host, recv_path); + let (first_fd, _) = bind_test_unix_dgram(&mut proc, &mut host, first_path); + let (selected_fd, selected_addr) = + bind_test_unix_dgram(&mut proc, &mut host, selected_path); + sys_connect(&mut proc, &mut host, first_fd, &recv_addr).unwrap(); + sys_connect(&mut proc, &mut host, selected_fd, &recv_addr).unwrap(); + + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT { + sys_send( + &mut proc, + &mut host, + selected_fd, + &(sequence as u32).to_le_bytes(), + 0, + ) + .unwrap(); + } + + let mut first_poll = WasmPollFd { + fd: first_fd, + events: POLLOUT, + revents: 0, + }; + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut first_poll), + 0, + ) + .unwrap(), + 0, + ); + + // The receiver selects the other sender without freeing capacity. + // The first sender must nevertheless become ready so its next send + // can report the now-immediate EPERM instead of remaining parked. + sys_connect(&mut proc, &mut host, recv_fd, &selected_addr).unwrap(); + assert_eq!( + proc.sockets + .get(test_socket_idx(&proc, recv_fd)) + .unwrap() + .dgram_queue + .len(), + UDP_DATAGRAM_QUEUE_LIMIT, + ); + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut first_poll), + 0, + ) + .unwrap(), + 1, + ); + assert_ne!(first_poll.revents & POLLOUT, 0); + assert_eq!( + sys_send(&mut proc, &mut host, first_fd, b"rejected", 0).unwrap_err(), + Errno::EPERM, + ); + + let mut selected_poll = WasmPollFd { + fd: selected_fd, + events: POLLOUT, + revents: 0, + }; + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut selected_poll), + 0, + ) + .unwrap(), + 0, + ); + + for fd in [first_fd, selected_fd, recv_fd] { + sys_close(&mut proc, &mut host, fd).unwrap(); + } + for path in [first_path.as_slice(), selected_path.as_slice(), recv_path.as_slice()] { + sys_unlink(&mut proc, &mut host, path).unwrap(); + } + } + + #[test] + fn test_unix_dgram_read_shutdown_releases_full_sender_to_epipe() { + let _lock = UNIX_REGISTRY_LOCK.lock().unwrap(); + let mut proc = Process::new(9054); + let mut host = MockHostIO::new(); + use wasm_posix_shared::poll::POLLOUT; + use wasm_posix_shared::signal::SIGPIPE; + use wasm_posix_shared::socket::{MSG_NOSIGNAL, SHUT_RD}; + + let recv_path = b"/tmp/udg-shutdown-recv.sock"; + let send_path = b"/tmp/udg-shutdown-send.sock"; + let (recv_fd, recv_addr) = bind_test_unix_dgram(&mut proc, &mut host, recv_path); + let (send_fd, _) = bind_test_unix_dgram(&mut proc, &mut host, send_path); + sys_connect(&mut proc, &mut host, send_fd, &recv_addr).unwrap(); + + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT { + sys_send( + &mut proc, + &mut host, + send_fd, + &(sequence as u32).to_le_bytes(), + 0, + ) + .unwrap(); + } + let recv_idx = test_socket_idx(&proc, recv_fd); + let mut pollfd = WasmPollFd { + fd: send_fd, + events: POLLOUT, + revents: 0, + }; + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 0, + ); + + sys_shutdown(&mut proc, &mut host, recv_fd, SHUT_RD).unwrap(); + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 1, + ); + assert_ne!(pollfd.revents & POLLOUT, 0); + assert_eq!( + sys_send(&mut proc, &mut host, send_fd, b"signal", 0).unwrap_err(), + Errno::EPIPE, + ); + assert!(proc.signals.is_pending(SIGPIPE)); + assert_eq!( + proc.sockets.get(recv_idx).unwrap().dgram_queue.len(), + UDP_DATAGRAM_QUEUE_LIMIT, + ); + proc.signals.clear(SIGPIPE); + assert_eq!( + sys_send( + &mut proc, + &mut host, + send_fd, + b"quiet", + MSG_NOSIGNAL, + ) + .unwrap_err(), + Errno::EPIPE, + ); + assert!(!proc.signals.is_pending(SIGPIPE)); + + for fd in [send_fd, recv_fd] { + sys_close(&mut proc, &mut host, fd).unwrap(); + } + for path in [send_path.as_slice(), recv_path.as_slice()] { + sys_unlink(&mut proc, &mut host, path).unwrap(); + } + } + + #[test] + fn test_unix_dgram_close_releases_full_connected_sender() { + let _lock = UNIX_REGISTRY_LOCK.lock().unwrap(); + let mut proc = Process::new(9055); + let mut host = MockHostIO::new(); + use wasm_posix_shared::poll::POLLOUT; + + let recv_path = b"/tmp/udg-close-recv.sock"; + let send_path = b"/tmp/udg-close-send.sock"; + let (recv_fd, recv_addr) = bind_test_unix_dgram(&mut proc, &mut host, recv_path); + let (send_fd, _) = bind_test_unix_dgram(&mut proc, &mut host, send_path); + sys_connect(&mut proc, &mut host, send_fd, &recv_addr).unwrap(); + for sequence in 0..UDP_DATAGRAM_QUEUE_LIMIT { + sys_send( + &mut proc, + &mut host, + send_fd, + &(sequence as u32).to_le_bytes(), + 0, + ) + .unwrap(); + } + + let mut pollfd = WasmPollFd { + fd: send_fd, + events: POLLOUT, + revents: 0, + }; + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 0, + ); + sys_close(&mut proc, &mut host, recv_fd).unwrap(); + assert_eq!( + sys_poll( + &mut proc, + &mut host, + core::slice::from_mut(&mut pollfd), + 0, + ) + .unwrap(), + 1, + ); + assert_ne!(pollfd.revents & POLLOUT, 0); + assert_eq!( + sys_send(&mut proc, &mut host, send_fd, b"closed", 0).unwrap_err(), + Errno::ECONNREFUSED, + ); + + sys_close(&mut proc, &mut host, send_fd).unwrap(); + for path in [send_path.as_slice(), recv_path.as_slice()] { + sys_unlink(&mut proc, &mut host, path).unwrap(); + } } #[test] @@ -20550,6 +20978,14 @@ mod tests { 4, ); } + let recv_idx = { + let entry = proc.fd_table.get(recv_fd).unwrap(); + let ofd = proc.ofd_table.get(entry.ofd_ref.0).unwrap(); + (-(ofd.host_handle + 1)) as usize + }; + udp_queue_datagram(proc.sockets.get_mut(recv_idx).unwrap(), || { + panic!("a full UDP queue must not construct a dropped payload") + }); // The already-queued datagrams survive in order. for expected in 0..UDP_DATAGRAM_QUEUE_LIMIT { diff --git a/crates/kernel/src/wakeup.rs b/crates/kernel/src/wakeup.rs index b709e3a75..3ddcec3c5 100644 --- a/crates/kernel/src/wakeup.rs +++ b/crates/kernel/src/wakeup.rs @@ -1,8 +1,9 @@ //! Wakeup event buffer for kernel-driven poll/select notification. //! -//! When pipe operations or listener accept queues change readiness state, -//! events are pushed into a global buffer. The host drains this buffer after -//! each syscall to wake only the specific waiters that are affected. +//! When pipe operations, listener accept queues, or AF_UNIX datagram send +//! state changes readiness, events are pushed into a global buffer. The host +//! drains this buffer after each syscall and wakes targeted waiters where an +//! identity is available or performs a bounded broad retry otherwise. use alloc::vec::Vec; use core::cell::UnsafeCell; @@ -16,6 +17,14 @@ pub const WAKE_WRITABLE: u8 = 2; /// Listener accept queue received a pending connection. pub const WAKE_ACCEPT: u8 = 4; +/// AF_UNIX datagram send readiness or its immediate result changed. +/// +/// Datagram writers do not have a pipe index that the host can target. The +/// host therefore retries untargeted blocked sends and issues a broad +/// readiness wake for poll/select/epoll operations when capacity, +/// associations, shutdown, close, or pathname state changes. +pub const WAKE_DATAGRAM_WRITABLE: u8 = 8; + /// A readiness change event. #[derive(Debug, Clone, Copy)] pub struct WakeupEvent { @@ -57,6 +66,12 @@ pub fn push_accept(accept_idx: u32) { push(accept_idx, WAKE_ACCEPT); } +/// Notify the host that AF_UNIX datagram send readiness or its immediate +/// outcome may have changed. +pub fn push_datagram_writable() { + push(0, WAKE_DATAGRAM_WRITABLE); +} + /// Drain all pending wakeup events, writing them to the output buffer. /// Returns the number of events written. /// @@ -101,10 +116,11 @@ mod tests { push(5, WAKE_READABLE); push(10, WAKE_WRITABLE); push_accept(12); + push_datagram_writable(); - let mut buf = [0u8; 20]; + let mut buf = [0u8; 25]; let count = drain(&mut buf, 10); - assert_eq!(count, 3); + assert_eq!(count, 4); // Event 0: pipe_idx=5, WAKE_READABLE assert_eq!(u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]), 5); @@ -118,6 +134,10 @@ mod tests { assert_eq!(u32::from_le_bytes([buf[10], buf[11], buf[12], buf[13]]), 12); assert_eq!(buf[14], WAKE_ACCEPT); + // Event 3: broad datagram-writable readiness change. + assert_eq!(u32::from_le_bytes([buf[15], buf[16], buf[17], buf[18]]), 0); + assert_eq!(buf[19], WAKE_DATAGRAM_WRITABLE); + // Buffer should be empty now let count2 = drain(&mut buf, 10); assert_eq!(count2, 0); diff --git a/docs/architecture.md b/docs/architecture.md index e1bbcefd7..355038995 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -509,10 +509,14 @@ User-visible networking is POSIX-first. Guest programs call normal AF_UNIX, AF_I AF_INET and AF_INET6 receive queues are currently bounded at 128 datagrams per socket. Once that fixed internal queue is full, a newly arriving UDP datagram is dropped and the already-queued datagrams retain their order. `SO_RCVBUF` -is stored and reported but does not size this queue yet. AF_UNIX datagrams use -the same bounded storage but are reliable: a full receive queue makes the send -enter the host's blocking retry path, or returns `EAGAIN` immediately for an -`O_NONBLOCK` or `MSG_DONTWAIT` send, without discarding queued messages. +requests are stored but do not size this queue; `getsockopt` continues to report +the fixed default capacity. AF_UNIX datagrams use the same bounded storage but +are reliable: a full receive queue makes the send enter the host's blocking +retry path, or returns `EAGAIN` immediately for an `O_NONBLOCK` or +`MSG_DONTWAIT` send, without discarding queued messages. Queue-capacity, +association, shutdown, close, and pathname changes wake blocked writers and +writable readiness waiters so they can observe either capacity or the new +immediate error. Loopback addresses are scoped to one Kandelo machine, but not every socket path is machine-wide yet. IPv4 and IPv6 loopback TCP and AF_UNIX streams have explicit cross-process paths. Current in-kernel IPv4/IPv6 loopback datagrams, AF_UNIX datagrams, and IPv4 multicast delivery are confined to the sending process. Forked sockets retain their kernel-local bind reservations and local lookup targets, but host-backed UDP endpoint registrations are not yet shared or transferred between processes. AF_INET6 represents `sockaddr_in6`, supports `::`/`::1`, and models dual-stack wildcard stream-port reservation, but it has no external or virtual-network IPv6 transport and no IPv6 multicast delivery. AF_INET6 datagrams therefore report `IPV6_V6ONLY=1`; disabling it fails until dual-stack datagram routing exists. diff --git a/docs/posix-status.md b/docs/posix-status.md index 9a17faacf..8cf0e25ad 100644 --- a/docs/posix-status.md +++ b/docs/posix-status.md @@ -235,9 +235,9 @@ shortcuts. | `accept()` / `accept4()` | Partial | AF_INET TCP delegates to the active HostIO networking backend; AF_UNIX and AF_INET6 loopback streams return connected sockets from kernel queues. A dual-stack IPv6 listener reports IPv4 peers as IPv4-mapped `sockaddr_in6`. Linux-style accept does not inherit O_NONBLOCK; accept4 applies SOCK_NONBLOCK and SOCK_CLOEXEC explicitly and rejects other flags before consuming a pending connection. Datagram accept rejects as unsupported. | | `connect()` | Partial | AF_UNIX streams support same- and cross-process pathname or abstract-namespace listeners. AF_UNIX datagrams deliver to a registered peer only within the same process; a missing, wrong-type, or cross-process peer returns ECONNREFUSED until machine-wide datagram routing exists. AF_INET TCP is host-backed and works over Node external TCP or the browser local virtual-network backend. AF_INET UDP connect stores the peer, auto-binds an ephemeral local port when needed, filters receives to the connected peer, and supports AF_UNSPEC unconnect. AF_INET6 streams support same- and cross-process `::1`; AF_INET6 datagrams are process-local and report `IPV6_V6ONLY=1` because dual-stack datagram routing is not implemented. Non-loopback IPv6 fails with EADDRNOTAVAIL for streams and ENETUNREACH for datagrams. External raw UDP also returns ENETUNREACH without another HostIO transport. | | `send()` / `recv()` | Partial | Unix domain streams and datagrams, AF_INET/AF_INET6 TCP streams, and connected AF_INET/AF_INET6 UDP preserve their socket-family addressing and datagram boundaries. TCP send/recv works over Node external TCP and the local virtual-network backend. Datagram MSG_PEEK and MSG_DONTWAIT are handled through recvfrom. Normal TCP close drains queued bytes before FIN and EOF; no transport invents a fixed post-FIN write count. A send rejected by a closed/reset stream returns EPIPE and raises SIGPIPE, while direct host/virtual handles may preserve ECONNRESET; accepted pipe-bridged resets currently surface as EOF/EPIPE. MSG_NOSIGNAL suppresses SIGPIPE without changing the errno. | -| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` does not size that fixed queue yet. AF_UNIX uses the same bound but preserves reliable delivery: a full queue blocks a blocking send through host retry and returns EAGAIN for `O_NONBLOCK`/`MSG_DONTWAIT`. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | +| `sendto()` / `recvfrom()` | Partial | AF_INET, AF_INET6, and AF_UNIX datagrams support connected and unconnected send, receive queues, and connected-peer filtering. IPv4/IPv6 return sender addresses; AF_UNIX currently returns only the family. IPv4/IPv6 UDP receive queues hold 128 datagrams and drop a new arrival once full, preserving the accepted queue's order; `SO_RCVBUF` requests do not size that fixed queue, and `getsockopt` reports the fixed default capacity. AF_UNIX uses the same bound but preserves reliable delivery: a full queue blocks a blocking send through host retry and returns EAGAIN for `O_NONBLOCK`/`MSG_DONTWAIT`; capacity, association, shutdown, close, and pathname changes wake blocked sends and writable readiness waits to observe capacity or the new immediate error. In-kernel IPv4/IPv6 loopback, AF_UNIX datagram, and IPv4 multicast delivery currently reaches sockets in the sender's process only; machine-wide cross-process datagram routing remains unimplemented. Fork preserves kernel-local bind reservations and lookup ownership, but it does not yet share or transfer a host-backed UDP registration. The `10.88.*` LocalVirtualNetwork path can route IPv4 datagrams between attached Kandelo machines through HostIO for the process that registered the endpoint. IPv4 multicast supports interface selection, loop suppression, membership, and source filtering only; IPv6 multicast and external raw UDP are not implemented. | | `setsockopt()` / `getsockopt()` | Partial | SOL_SOCKET exposes SO_TYPE, SO_DOMAIN, SO_ERROR, SO_ACCEPTCONN, SO_RCVBUF, and SO_SNDBUF; SO_REUSEADDR affects UDP bind conflicts. SO_LINGER uses `struct linger`; its disabled form is stored, while enabling timed or reset-style linger returns EOPNOTSUPP until every transport supports the close mode. SO_BINDTODEVICE validates `lo`/`eth0`, supports empty-name unbind, and constrains bind/connect/send routing. TCP_CONGESTION uses a string layout and accepts only the modeled `cubic` policy; selecting unimplemented algorithms fails. IPv4 multicast membership/source-filter options drive process-local loopback delivery. IPV6_V6ONLY controls pre-bind stream dual-stack behavior; AF_INET6 datagrams truthfully remain V6-only. Other accepted IPv6 multicast options are stored but do not provide IPv6 multicast transport. | -| `shutdown()` | Partial | SHUT_RD, SHUT_WR, and SHUT_RDWR transitions are idempotent within a process and release each owned pipe/host reference once. UDP write shutdown returns EPIPE on datagram send; read shutdown is EOF-like for recv/poll. Fork-inherited sockets still clone shutdown flags per process instead of sharing one socket-wide shutdown state, and the external host ABI has no half-shutdown operation. | +| `shutdown()` | Partial | SHUT_RD, SHUT_WR, and SHUT_RDWR transitions are idempotent within a process and release each owned pipe/host reference once. UDP write shutdown returns EPIPE on datagram send; read shutdown is EOF-like for recv/poll. Sending to a read-shut AF_UNIX datagram peer returns EPIPE (and SIGPIPE unless MSG_NOSIGNAL is used), and the transition wakes blocked sends/readiness waits. Fork-inherited sockets still clone shutdown flags per process instead of sharing one socket-wide shutdown state, and the external host ABI has no half-shutdown operation. | | `select()` | Partial | Wrapper around poll(). Converts fd_set bitmasks to pollfd array. Timeout supported via polling loop. | | `poll()` | Partial | Checks readiness for regular files, pipes, and sockets. UDP poll reports queued datagrams, connected-peer filtering, EOF-like read shutdown, write-shutdown hangup, and pending socket errors. Timeout supported via polling loop with 1ms sleep intervals. Returns EINTR on pending signals. | | `ppoll()` | Full | Wraps poll() with atomic signal mask swap: save → set → poll → restore. Timespec converted to timeout_ms in glue layer. | diff --git a/host/src/kernel-worker.ts b/host/src/kernel-worker.ts index cd8639838..9e6cf9ad3 100644 --- a/host/src/kernel-worker.ts +++ b/host/src/kernel-worker.ts @@ -446,6 +446,10 @@ interface ChannelInfo { * retry/sleep/fork/exec path. Prevents the poller from re-entering a * channel that is already in flight. Only used when usePolling=true. */ handling?: boolean; + /** Absolute deadline for the current finite poll/select/epoll wait. */ + readinessDeadline?: number; + /** Force the next readiness dispatch to perform a zero-time final check. */ + readinessFinalCheck?: boolean; } /** Info about a registered process. */ @@ -783,6 +787,8 @@ export class CentralizedKernelWorker { needsSignalSafeWake?: boolean; /** Date.now() deadline for finite-timeout poll/ppoll retries, or -1. */ deadline?: number; + /** Generic write-like fallback that has no targetable pipe token. */ + isWriteRetry?: boolean; }>(); /** Pending pselect6/select retries — keyed by channelOffset for per-thread tracking */ private pendingSelectRetries = new Map entry.isWriteRetry, + ); + for (const [key, entry] of matches) { + if (this.pendingPollRetries.get(key) !== entry) continue; + this.pendingPollRetries.delete(key); + if (entry.timer !== null) clearTimeout(entry.timer); + if (this.processes.has(entry.channel.pid)) { + this.retrySyscall(entry.channel); + } + } + } + private anyPendingRetryNeedsSignalSafeWake(): boolean { for (const entry of this.pendingPollRetries.values()) { if (entry.needsSignalSafeWake) return true; @@ -3225,6 +3274,7 @@ export class CentralizedKernelWorker { private scheduleWakeBlockedRetriesDeferred(): void { if (this.pendingPollRetries.size === 0 && this.pendingSelectRetries.size === 0 && this.pendingPipeReaders.size === 0 && this.pendingPipeWriters.size === 0) return; this.postponeSignalSafePollRetries(SIGNAL_SAFE_POLL_WAKE_DELAY_MS); + this.postponeSignalSafeSelectRetries(SIGNAL_SAFE_POLL_WAKE_DELAY_MS); if (this.wakeScheduled) return; this.wakeScheduled = true; setTimeout(() => { @@ -3246,6 +3296,7 @@ export class CentralizedKernelWorker { : delayMs; const retryMs = Math.max(1, Math.min(delayMs, remainingMs)); entry.timer = setTimeout(() => { + if (this.pendingPollRetries.get(key) !== entry) return; this.pendingPollRetries.delete(key); if (this.processes.has(entry.channel.pid)) { this.retrySyscall(entry.channel); @@ -3254,6 +3305,33 @@ export class CentralizedKernelWorker { } } + /** Keep pselect's fallback timer from bypassing the signal-safe wake grace. */ + private postponeSignalSafeSelectRetries(delayMs: number): void { + const now = Date.now(); + for (const [key, entry] of this.pendingSelectRetries) { + if (!entry.needsSignalSafeWake) continue; + if (entry.timer !== null) { + clearTimeout(entry.timer); + clearImmediate(entry.timer); + } + + const remainingMs = entry.deadline > 0 + ? Math.max(1, entry.deadline - now) + : delayMs; + const retryMs = Math.max(1, Math.min(delayMs, remainingMs)); + entry.timer = setTimeout(() => { + if (this.pendingSelectRetries.get(key) !== entry) return; + this.pendingSelectRetries.delete(key); + if (!this.processes.has(entry.channel.pid)) return; + if (entry.syscallNr === SYS_SELECT) { + this.handleSelect(entry.channel, entry.origArgs); + } else { + this.handlePselect6(entry.channel, entry.origArgs); + } + }, retryMs); + } + } + /** * Schedule a microtask to wake all blocked poll/pselect6 retries. * Coalesced via wakeScheduled flag — multiple calls within the same @@ -3375,6 +3453,36 @@ export class CentralizedKernelWorker { } } + /** Reuse one absolute deadline across readiness retries for this syscall. */ + private getReadinessDeadline(channel: ChannelInfo, timeoutMs: number): number { + if (timeoutMs <= 0) return -1; + if (channel.readinessDeadline === undefined) { + channel.readinessDeadline = Date.now() + timeoutMs; + } + return channel.readinessDeadline; + } + + /** Clear readiness deadline and any still-parked retry for a completed call. */ + private clearReadinessWait(channel: ChannelInfo): void { + channel.readinessDeadline = undefined; + channel.readinessFinalCheck = undefined; + + const pollEntry = this.pendingPollRetries.get(channel.channelOffset); + if (pollEntry?.channel === channel) { + if (pollEntry.timer !== null) clearTimeout(pollEntry.timer); + this.pendingPollRetries.delete(channel.channelOffset); + } + + const selectEntry = this.pendingSelectRetries.get(channel.channelOffset); + if (selectEntry?.channel === channel) { + if (selectEntry.timer !== null) { + clearTimeout(selectEntry.timer); + clearImmediate(selectEntry.timer); + } + this.pendingSelectRetries.delete(channel.channelOffset); + } + } + /** * Remove a channel from pending pipe readers (all pipes). * Called when a socket timeout fires to clean up the reader registration. @@ -3663,6 +3771,14 @@ export class CentralizedKernelWorker { this.completeChannel(channel, syscallNr, origArgs, SYSCALL_ARGS[syscallNr], 0, 0); return; } + const deadline = this.getReadinessDeadline(channel, timeoutMs); + if (deadline > 0 && Date.now() >= deadline) { + // Re-enter once with timeout=0. Besides checking readiness at the + // deadline, this lets ppoll restore its temporary signal mask. + channel.readinessFinalCheck = true; + this.retrySyscall(channel); + return; + } // Resolve which pipe/listener readiness tokens the polled fds map to. const { pipeIndices, acceptIndices } = @@ -3674,32 +3790,30 @@ export class CentralizedKernelWorker { const nfds = origArgs[1]; // poll(fds, nfds, ...) / ppoll(fds, nfds, ...) if (timeoutMs > 0 && nfds === 0) { // Pure sleep: no fds to poll, just wait for timeout + const remainingMs = Math.max(deadline - Date.now(), 1); const timer = setTimeout(() => { this.pendingPollRetries.delete(channel.channelOffset); if (this.processes.has(channel.pid)) { - this.completeChannel(channel, syscallNr, origArgs, SYSCALL_ARGS[syscallNr], 0, 0); + channel.readinessFinalCheck = true; + this.retrySyscall(channel); } - }, timeoutMs); + }, remainingMs); this.pendingPollRetries.set(channel.channelOffset, { timer, channel, pipeIndices, acceptIndices, needsSignalSafeWake, - deadline: Date.now() + timeoutMs, + deadline, }); return; } - const deadline = timeoutMs > 0 ? Date.now() + timeoutMs : -1; const retryFn = () => { this.pendingPollRetries.delete(channel.channelOffset); if (!this.processes.has(channel.pid)) return; - // Check deadline for finite timeout - if (deadline > 0 && Date.now() >= deadline) { - this.completeChannel(channel, syscallNr, origArgs, SYSCALL_ARGS[syscallNr], 0, 0); - return; - } + // Always run the kernel once at the deadline. If it still reports + // EAGAIN, the branch above completes the wait with 0. this.retrySyscall(channel); }; // With pipe/listener readiness tokens, rely on targeted wakeups for @@ -3938,7 +4052,12 @@ export class CentralizedKernelWorker { } }; const timer = setTimeout(retryFn, 10); - this.pendingPollRetries.set(channel.channelOffset, { timer, channel, pipeIndices: [] }); + this.pendingPollRetries.set(channel.channelOffset, { + timer, + channel, + pipeIndices: [], + isWriteRetry: WRITE_LIKE_SYSCALLS.has(syscallNr), + }); } /** @@ -4151,6 +4270,10 @@ export class CentralizedKernelWorker { timeoutMs = sec * 1000 + Math.floor(usec / 1000); if (timeoutMs < 0) timeoutMs = 0; } + const finalCheck = channel.readinessFinalCheck === true; + channel.readinessFinalCheck = false; + const kernelTimeoutMs = finalCheck ? 0 : timeoutMs; + const deadline = this.getReadinessDeadline(channel, timeoutMs); // Pure-sleep fast path: select(0, NULL, NULL, NULL, &tv) is `my_sleep`. // The kernel can't tell us anything new — there are no fds to poll — @@ -4159,24 +4282,25 @@ export class CentralizedKernelWorker { // (handleKill -> scheduleWakeBlockedRetries -> wakeAllBlockedRetries // already iterates pendingSelectRetries entries). if (nfds === 0 && readPtr === 0 && writePtr === 0 && exceptPtr === 0) { - if (timeoutMs === 0) { + if (kernelTimeoutMs === 0) { this.completeChannel(channel, SYS_SELECT, origArgs, undefined, 0, 0); return; } const finite = timeoutMs > 0; + const remainingMs = finite ? Math.max(deadline - Date.now(), 1) : -1; const timer = finite ? setTimeout(() => { this.pendingSelectRetries.delete(channel.channelOffset); if (this.processes.has(channel.pid)) { this.completeChannel(channel, SYS_SELECT, origArgs, undefined, 0, 0); } - }, timeoutMs) + }, remainingMs) : (null as any); this.pendingSelectRetries.set(channel.channelOffset, { timer, channel, origArgs, - deadline: finite ? Date.now() + timeoutMs : -1, + deadline, needsSignalSafeWake: false, syscallNr: SYS_SELECT, }); @@ -4213,7 +4337,7 @@ export class CentralizedKernelWorker { kernelView.setBigInt64(CH_ARGS + 1 * CH_ARG_SIZE, BigInt(readPtr !== 0 ? dataStart : 0), true); kernelView.setBigInt64(CH_ARGS + 2 * CH_ARG_SIZE, BigInt(writePtr !== 0 ? dataStart + FD_SET_SIZE : 0), true); kernelView.setBigInt64(CH_ARGS + 3 * CH_ARG_SIZE, BigInt(exceptPtr !== 0 ? dataStart + 2 * FD_SET_SIZE : 0), true); - kernelView.setBigInt64(CH_ARGS + 4 * CH_ARG_SIZE, BigInt(timeoutMs), true); + kernelView.setBigInt64(CH_ARGS + 4 * CH_ARG_SIZE, BigInt(kernelTimeoutMs), true); const handleChannel = this.kernelInstance!.exports.kernel_handle_channel as (offset: KernelPointer, pid: number) => number; @@ -4256,14 +4380,14 @@ export class CentralizedKernelWorker { this.completeChannel(channel, SYS_SELECT, origArgs, undefined, 0, 0); return; } - const deadline = timeoutMs > 0 ? Date.now() + timeoutMs : -1; + if (deadline > 0 && Date.now() >= deadline) { + channel.readinessFinalCheck = true; + this.handleSelect(channel, origArgs); + return; + } const retryFn = () => { this.pendingSelectRetries.delete(channel.channelOffset); if (!this.processes.has(channel.pid)) return; - if (deadline > 0 && Date.now() >= deadline) { - this.completeChannel(channel, SYS_SELECT, origArgs, undefined, 0, 0); - return; - } this.handleSelect(channel, origArgs); }; const finite = timeoutMs > 0; @@ -4318,6 +4442,10 @@ export class CentralizedKernelWorker { const nsec = Number(pv.getBigInt64(8, true)); timeoutMs = sec * 1000 + Math.floor(nsec / 1000000); } + const finalCheck = channel.readinessFinalCheck === true; + channel.readinessFinalCheck = false; + const kernelTimeoutMs = finalCheck ? 0 : timeoutMs; + const deadline = this.getReadinessDeadline(channel, timeoutMs); // Decode sigmask: pselect6 arg6 → pointer to {sigset_t *mask, size_t size} // On wasm32: {u32 mask_ptr, u32 size} = 8 bytes @@ -4355,7 +4483,7 @@ export class CentralizedKernelWorker { kernelView.setBigInt64(CH_ARGS + 1 * CH_ARG_SIZE, BigInt(readPtr !== 0 ? dataStart : 0), true); kernelView.setBigInt64(CH_ARGS + 2 * CH_ARG_SIZE, BigInt(writePtr !== 0 ? dataStart + FD_SET_SIZE : 0), true); kernelView.setBigInt64(CH_ARGS + 3 * CH_ARG_SIZE, BigInt(exceptPtr !== 0 ? dataStart + 2 * FD_SET_SIZE : 0), true); - kernelView.setBigInt64(CH_ARGS + 4 * CH_ARG_SIZE, BigInt(timeoutMs), true); + kernelView.setBigInt64(CH_ARGS + 4 * CH_ARG_SIZE, BigInt(kernelTimeoutMs), true); kernelView.setBigInt64(CH_ARGS + 5 * CH_ARG_SIZE, BigInt(kernelMaskPtr), true); const handleChannel = this.kernelInstance!.exports.kernel_handle_channel as @@ -4402,23 +4530,29 @@ export class CentralizedKernelWorker { this.completeChannel(channel, SYS_PSELECT6, origArgs, undefined, 0, 0); return; } + if (deadline > 0 && Date.now() >= deadline) { + channel.readinessFinalCheck = true; + this.handlePselect6(channel, origArgs); + return; + } - const deadline = timeoutMs > 0 ? Date.now() + timeoutMs : -1; // pselect6 with a non-null sigmask pointer has the same late-signal // race as ppoll. See scheduleWakeBlockedRetriesDeferred. - const needsSignalSafeWake = maskDataPtr !== 0; + const needsSignalSafeWake = kernelMaskPtr !== 0; // nfds=0: pure sleep/sigsuspend-like behavior. // With finite timeout: sleep for that duration. // With infinite timeout: block until signal (wakeAllBlockedRetries). if (nfds === 0) { if (timeoutMs > 0) { + const remainingMs = Math.max(deadline - Date.now(), 1); const timer = setTimeout(() => { this.pendingSelectRetries.delete(channel.channelOffset); if (this.processes.has(channel.pid)) { - this.completeChannel(channel, SYS_PSELECT6, origArgs, undefined, 0, 0); + channel.readinessFinalCheck = true; + this.handlePselect6(channel, origArgs); } - }, timeoutMs); + }, remainingMs); this.pendingSelectRetries.set(channel.channelOffset, { timer, channel, origArgs, deadline, needsSignalSafeWake, syscallNr: SYS_PSELECT6, }); @@ -4437,13 +4571,10 @@ export class CentralizedKernelWorker { const retryFn = () => { this.pendingSelectRetries.delete(channel.channelOffset); if (!this.processes.has(channel.pid)) return; - if (deadline > 0 && Date.now() >= deadline) { - this.completeChannel(channel, SYS_PSELECT6, origArgs, undefined, 0, 0); - return; - } this.handlePselect6(channel, origArgs); }; - const timer = setImmediate(retryFn); + const remainingMs = deadline > 0 ? Math.max(deadline - Date.now(), 1) : 50; + const timer = setTimeout(retryFn, Math.min(remainingMs, 50)); this.pendingSelectRetries.set(channel.channelOffset, { timer, channel, origArgs, deadline, needsSignalSafeWake, syscallNr: SYS_PSELECT6, }); @@ -4591,6 +4722,7 @@ export class CentralizedKernelWorker { const eventsPtr = origArgs[1]; // output pointer in process memory const maxevents = origArgs[2]; const timeoutMs = origArgs[3]; + const deadline = this.getReadinessDeadline(channel, timeoutMs); // origArgs[4] = sigmask ptr (process-space), origArgs[5] = sigset size if (maxevents <= 0) { @@ -4615,6 +4747,11 @@ export class CentralizedKernelWorker { this.relistenChannel(channel); return; } + if (deadline > 0 && Date.now() >= deadline) { + this.completeChannelRaw(channel, 0, 0); + this.relistenChannel(channel); + return; + } // For non-zero timeout with no interests, retry with delay to avoid starvation const retryFn = () => { this.pendingPollRetries.delete(channel.channelOffset); @@ -4622,8 +4759,14 @@ export class CentralizedKernelWorker { this.handleEpollPwait(channel, syscallNr, origArgs); } }; - const timer = setTimeout(retryFn, 10); - this.pendingPollRetries.set(channel.channelOffset, { timer, channel, pipeIndices: [] }); + const retryMs = deadline > 0 ? Math.min(Math.max(deadline - Date.now(), 1), 10) : 10; + const timer = setTimeout(retryFn, retryMs); + this.pendingPollRetries.set(channel.channelOffset, { + timer, + channel, + pipeIndices: [], + deadline, + }); return; } @@ -4736,6 +4879,12 @@ export class CentralizedKernelWorker { this.relistenChannel(channel); return; } + if (deadline > 0 && Date.now() >= deadline) { + // The nonblocking kernel poll above was the final readiness check. + this.completeChannelRaw(channel, 0, 0); + this.relistenChannel(channel); + return; + } // Blocking: retry via setTimeout to avoid starving other processes. // Pipe-based wakeup (via wakeAllBlockedRetries) provides instant wakeup @@ -4748,12 +4897,14 @@ export class CentralizedKernelWorker { this.handleEpollPwait(channel, syscallNr, origArgs); } }; - const timer = setTimeout(retryFn, 10); + const retryMs = deadline > 0 ? Math.min(Math.max(deadline - Date.now(), 1), 10) : 10; + const timer = setTimeout(retryFn, retryMs); this.pendingPollRetries.set(channel.channelOffset, { timer, channel, pipeIndices, acceptIndices, + deadline, }); } diff --git a/host/test/datagram-wakeup.test.ts b/host/test/datagram-wakeup.test.ts new file mode 100644 index 000000000..dbf943d18 --- /dev/null +++ b/host/test/datagram-wakeup.test.ts @@ -0,0 +1,77 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { CentralizedKernelWorker } from "../src/kernel-worker"; + +const WAKE_DATAGRAM_WRITABLE = 8; + +function createSharedMemory(): WebAssembly.Memory { + return new WebAssembly.Memory({ initial: 1, maximum: 1, shared: true }); +} + +function createWorkerHarness(): any { + const memory = createSharedMemory(); + const scratchOffset = 128; + const drain = (outPtr: number): number => { + const bytes = new Uint8Array(memory.buffer, outPtr, 5); + bytes.fill(0); + bytes[4] = WAKE_DATAGRAM_WRITABLE; + return 1; + }; + + return Object.assign(Object.create(CentralizedKernelWorker.prototype), { + kernel: { toKernelPtr: (value: number | bigint) => Number(value) }, + kernelInstance: { exports: { kernel_drain_wakeup_events: drain } }, + kernelMemory: memory, + scratchOffset, + processes: new Map(), + pendingPollRetries: new Map(), + pendingSelectRetries: new Map(), + pendingPipeReaders: new Map(), + pendingPipeWriters: new Map(), + wakeScheduled: false, + }); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("datagram send-state wakeups", () => { + it("retries blocked writes immediately without bypassing signal-safe poll deferral", () => { + vi.useFakeTimers(); + const worker = createWorkerHarness(); + const channel = { pid: 42, channelOffset: 0, memory: createSharedMemory() }; + const pollChannel = { pid: channel.pid, channelOffset: 64, memory: channel.memory }; + worker.processes.set(channel.pid, { channels: [channel, pollChannel] }); + + const fallback = vi.fn(); + const timer = setTimeout(fallback, 1); + worker.pendingPollRetries.set(channel.channelOffset, { + timer, + channel, + pipeIndices: [], + deadline: Date.now() + 1, + isWriteRetry: true, + }); + worker.pendingPollRetries.set(pollChannel.channelOffset, { + timer: null, + channel: pollChannel, + pipeIndices: [], + needsSignalSafeWake: true, + }); + worker.retrySyscall = vi.fn(); + worker.scheduleWakeBlockedRetries = vi.fn(); + worker.scheduleWakeBlockedRetriesDeferred = vi.fn(); + + worker.drainAndProcessWakeupEvents(); + + expect(worker.retrySyscall).toHaveBeenCalledOnce(); + expect(worker.retrySyscall).toHaveBeenCalledWith(channel); + expect(worker.pendingPollRetries.has(channel.channelOffset)).toBe(false); + expect(worker.pendingPollRetries.has(pollChannel.channelOffset)).toBe(true); + expect(worker.scheduleWakeBlockedRetries).not.toHaveBeenCalled(); + expect(worker.scheduleWakeBlockedRetriesDeferred).toHaveBeenCalledOnce(); + + vi.runAllTimers(); + expect(fallback).not.toHaveBeenCalled(); + }); +}); diff --git a/host/test/readiness-deadline.test.ts b/host/test/readiness-deadline.test.ts new file mode 100644 index 000000000..6653a3c9c --- /dev/null +++ b/host/test/readiness-deadline.test.ts @@ -0,0 +1,107 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { ABI_SYSCALLS } from "../src/generated/abi"; +import { CentralizedKernelWorker } from "../src/kernel-worker"; + +function createSharedMemory(): WebAssembly.Memory { + return new WebAssembly.Memory({ initial: 1, maximum: 1, shared: true }); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("finite readiness deadlines", () => { + it("keeps one poll deadline and performs a final readiness retry", () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + + const channel: any = { + pid: 42, + channelOffset: 0, + memory: createSharedMemory(), + }; + const worker: any = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + processes: new Map([[channel.pid, { channels: [channel] }]]), + pendingPollRetries: new Map(), + pendingSelectRetries: new Map(), + pendingPipeReaders: new Map(), + pendingPipeWriters: new Map(), + }); + worker.resolvePollReadinessIndices = () => ({ pipeIndices: [], acceptIndices: [] }); + worker.completeChannel = vi.fn(); + + const args = [0, 1, 120, 0, 0, 0]; + const observedDeadlines: number[] = []; + let finalChecks = 0; + worker.retrySyscall = vi.fn(() => { + observedDeadlines.push(channel.readinessDeadline); + if (channel.readinessFinalCheck) { + // Model the zero-time kernel dispatch returning 0 after its final + // readiness check (and, for ppoll, restoring the temporary mask). + finalChecks++; + channel.readinessFinalCheck = false; + worker.completeChannel(channel, ABI_SYSCALLS.Poll, args, undefined, 0, 0); + return; + } + // Model the kernel's next nonblocking check returning EAGAIN again. + worker.handleBlockingRetry(channel, ABI_SYSCALLS.Poll, args); + }); + + worker.handleBlockingRetry(channel, ABI_SYSCALLS.Poll, args); + expect(channel.readinessDeadline).toBe(1_120); + + vi.advanceTimersByTime(119); + expect(worker.completeChannel).not.toHaveBeenCalled(); + expect(observedDeadlines).toEqual([1_120, 1_120]); + + vi.advanceTimersByTime(1); + expect(observedDeadlines).toEqual([1_120, 1_120, 1_120, 1_120]); + expect(finalChecks).toBe(1); + expect(worker.completeChannel).toHaveBeenCalledOnce(); + expect(worker.completeChannel.mock.calls[0].slice(-2)).toEqual([0, 0]); + }); + + it("postpones a signal-safe pselect fallback until the deferred wake", () => { + vi.useFakeTimers(); + vi.setSystemTime(2_000); + + const channel: any = { + pid: 42, + channelOffset: 64, + memory: createSharedMemory(), + }; + const earlyFallback = vi.fn(); + const entry: any = { + timer: setTimeout(earlyFallback, 1), + channel, + origArgs: [1, 0, 0, 0, 0, 0], + deadline: 2_100, + needsSignalSafeWake: true, + syscallNr: ABI_SYSCALLS.Pselect6, + }; + const worker: any = Object.assign(Object.create(CentralizedKernelWorker.prototype), { + processes: new Map([[channel.pid, { channels: [channel] }]]), + pendingPollRetries: new Map(), + pendingSelectRetries: new Map([[channel.channelOffset, entry]]), + pendingPipeReaders: new Map(), + pendingPipeWriters: new Map(), + wakeScheduled: false, + }); + worker.handlePselect6 = vi.fn(); + worker.wakeAllBlockedRetries = vi.fn(); + + worker.scheduleWakeBlockedRetriesDeferred(); + + expect(entry.deadline).toBe(2_100); + vi.advanceTimersByTime(49); + expect(earlyFallback).not.toHaveBeenCalled(); + expect(worker.handlePselect6).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1); + expect(earlyFallback).not.toHaveBeenCalled(); + expect(worker.handlePselect6).toHaveBeenCalledOnce(); + expect(worker.handlePselect6).toHaveBeenCalledWith(channel, entry.origArgs); + expect(worker.pendingSelectRetries.has(channel.channelOffset)).toBe(false); + expect(worker.wakeAllBlockedRetries).toHaveBeenCalledOnce(); + }); +}); diff --git a/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c b/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c index 42869b983..1d0ec9a00 100644 --- a/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c +++ b/tests/sortix/os-test-local/udp/unix-dgram-queue-backpressure.c @@ -5,35 +5,174 @@ #include "udp.h" +#include #include #include +#include +#include +#include #include +#include + +struct blocking_send_context +{ + int fd; + uint32_t payload; + _Atomic int started; + _Atomic int done; + ssize_t result; + int errnum; +}; + +struct blocking_poll_context +{ + int fd; + _Atomic int started; + _Atomic int done; + int result; + short revents; + int errnum; +}; + +static void* blocking_send_thread(void* context_ptr) +{ + struct blocking_send_context* context = context_ptr; + atomic_store_explicit(&context->started, 1, memory_order_release); + errno = 0; + context->result = send(context->fd, + &context->payload, + sizeof(context->payload), + MSG_NOSIGNAL); + context->errnum = errno; + atomic_store_explicit(&context->done, 1, memory_order_release); + return NULL; +} + +static void* blocking_poll_thread(void* context_ptr) +{ + struct blocking_poll_context* context = context_ptr; + struct pollfd pfd = { .fd = context->fd, .events = POLLOUT, .revents = 0 }; + atomic_store_explicit(&context->started, 1, memory_order_release); + errno = 0; + context->result = poll(&pfd, 1, 1000); + context->errnum = errno; + context->revents = pfd.revents; + atomic_store_explicit(&context->done, 1, memory_order_release); + return NULL; +} + +static void wait_for_thread_start(_Atomic int* started, const char* operation) +{ + for ( int attempt = 0; attempt < 2000; attempt++ ) + { + if ( atomic_load_explicit(started, memory_order_acquire) ) + return; + usleep(1000); + } + errx(1, "%s thread did not start", operation); +} + +static void join_thread(pthread_t thread, const char* operation) +{ + int error = pthread_join(thread, NULL); + if ( error ) + { + errno = error; + err(1, "join %s thread", operation); + } +} int main(void) { - const char* path = "/tmp/kandelo-unix-dgram-overflow.sock"; - if ( unlink(path) < 0 && errno != ENOENT ) - err(1, "unlink before bind"); + const char* recv_path = "/tmp/kandelo-unix-dgram-overflow-recv.sock"; + const char* send_path = "/tmp/kandelo-unix-dgram-overflow-send.sock"; + if ( unlink(recv_path) < 0 && errno != ENOENT ) + err(1, "unlink receiver before bind"); + if ( unlink(send_path) < 0 && errno != ENOENT ) + err(1, "unlink sender before bind"); int recv_fd = socket(AF_UNIX, SOCK_DGRAM, 0); if ( recv_fd < 0 ) err(1, "receiver socket"); - struct sockaddr_un addr; - memset(&addr, 0, sizeof(addr)); - addr.sun_family = AF_UNIX; - strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1); - socklen_t addr_len = + struct sockaddr_un recv_addr; + memset(&recv_addr, 0, sizeof(recv_addr)); + recv_addr.sun_family = AF_UNIX; + strncpy(recv_addr.sun_path, recv_path, sizeof(recv_addr.sun_path) - 1); + socklen_t recv_addr_len = (socklen_t) (offsetof(struct sockaddr_un, sun_path) + - strlen(addr.sun_path) + 1); - if ( bind(recv_fd, (const struct sockaddr*) &addr, addr_len) < 0 ) + strlen(recv_addr.sun_path) + 1); + if ( bind(recv_fd, + (const struct sockaddr*) &recv_addr, + recv_addr_len) < 0 ) err(1, "receiver bind"); int send_fd = socket(AF_UNIX, SOCK_DGRAM, 0); if ( send_fd < 0 ) err(1, "sender socket"); - if ( connect(send_fd, (const struct sockaddr*) &addr, addr_len) < 0 ) + struct sockaddr_un send_addr; + memset(&send_addr, 0, sizeof(send_addr)); + send_addr.sun_family = AF_UNIX; + strncpy(send_addr.sun_path, send_path, sizeof(send_addr.sun_path) - 1); + socklen_t send_addr_len = + (socklen_t) (offsetof(struct sockaddr_un, sun_path) + + strlen(send_addr.sun_path) + 1); + if ( bind(send_fd, + (const struct sockaddr*) &send_addr, + send_addr_len) < 0 ) + err(1, "sender bind"); + if ( connect(send_fd, + (const struct sockaddr*) &recv_addr, + recv_addr_len) < 0 ) err(1, "sender connect"); + + int attacker_fd = socket(AF_UNIX, SOCK_DGRAM, 0); + if ( attacker_fd < 0 ) + err(1, "attacker socket"); + if ( connect(attacker_fd, + (const struct sockaddr*) &recv_addr, + recv_addr_len) < 0 ) + err(1, "attacker pre-connect connect"); + const char attacker_payload[] = "attacker-before-connect"; + if ( send(attacker_fd, + attacker_payload, + sizeof(attacker_payload) - 1, + 0) != (ssize_t) (sizeof(attacker_payload) - 1) ) + err(1, "attacker pre-connect send"); + const char peer_payload[] = "peer-before-connect"; + if ( send(send_fd, peer_payload, sizeof(peer_payload) - 1, 0) != + (ssize_t) (sizeof(peer_payload) - 1) ) + err(1, "peer pre-connect send"); + + if ( connect(recv_fd, + (const struct sockaddr*) &send_addr, + send_addr_len) < 0 ) + err(1, "receiver connect"); + char preconnect_buf[32]; + ssize_t preconnect_amount = + recv(recv_fd, preconnect_buf, sizeof(preconnect_buf), MSG_DONTWAIT); + if ( preconnect_amount != (ssize_t) (sizeof(peer_payload) - 1) || + memcmp(preconnect_buf, peer_payload, sizeof(peer_payload) - 1) != 0 ) + errx(1, "receiver did not preserve only its selected peer's datagram"); + + errno = 0; + if ( connect(attacker_fd, + (const struct sockaddr*) &recv_addr, + recv_addr_len) != -1 ) + errx(1, "attacker connect unexpectedly succeeded"); + if ( errno != EPERM ) + err(1, "attacker connect to connected receiver"); + errno = 0; + if ( sendto(attacker_fd, + attacker_payload, + sizeof(attacker_payload) - 1, + 0, + (const struct sockaddr*) &recv_addr, + recv_addr_len) != -1 ) + errx(1, "attacker send unexpectedly succeeded"); + if ( errno != EPERM ) + err(1, "attacker send to connected receiver"); + int flags = fcntl(send_fd, F_GETFL); if ( flags < 0 || fcntl(send_fd, F_SETFL, flags | O_NONBLOCK) < 0 ) err(1, "sender nonblock"); @@ -47,11 +186,99 @@ int main(void) if ( amount != (ssize_t) sizeof(payload) ) errx(1, "send returned %zi", amount); } + // The receiver selected send_fd while attacker_fd was already connected. + // Even though the selected peer keeps the queue full, the rejected sender + // is writable because its next send now has an immediate EPERM result. + struct pollfd rejected_pfd = { + .fd = attacker_fd, + .events = POLLOUT, + .revents = 0, + }; + if ( poll(&rejected_pfd, 1, 0) != 1 || + !(rejected_pfd.revents & POLLOUT) ) + errx(1, + "rejected full-queue sender remained blocked: revents=%#x", + rejected_pfd.revents); + errno = 0; + if ( send(attacker_fd, + attacker_payload, + sizeof(attacker_payload) - 1, + 0) != -1 ) + errx(1, "rejected connected sender unexpectedly succeeded"); + if ( errno != EPERM ) + err(1, "rejected connected sender"); struct pollfd pfd = { .fd = send_fd, .events = POLLOUT, .revents = 0 }; if ( poll(&pfd, 1, 0) != 0 || pfd.revents != 0 ) errx(1, "full peer unexpectedly writable: revents=%#x", pfd.revents); + // Finite readiness waits must keep one absolute deadline and perform a + // final readiness check instead of restarting their timeout on each retry. + pfd.revents = 0; + if ( poll(&pfd, 1, 120) != 0 || pfd.revents != 0 ) + errx(1, "finite poll reported full peer writable: revents=%#x", pfd.revents); + fd_set writefds; + FD_ZERO(&writefds); + FD_SET(send_fd, &writefds); + struct timeval timeout = { .tv_sec = 0, .tv_usec = 120000 }; + int select_result = select(send_fd + 1, NULL, &writefds, NULL, &timeout); + if ( select_result != 0 ) + errx(1, "finite select reported full peer writable: result=%d", select_result); + sigset_t blocked_mask; + sigset_t old_mask; + sigset_t wait_mask; + sigset_t after_mask; + sigemptyset(&blocked_mask); + sigaddset(&blocked_mask, SIGUSR1); + if ( sigprocmask(SIG_BLOCK, &blocked_mask, &old_mask) < 0 ) + err(1, "block SIGUSR1"); + sigemptyset(&wait_mask); + FD_ZERO(&writefds); + FD_SET(send_fd, &writefds); + struct timespec pselect_timeout = { .tv_sec = 0, .tv_nsec = 120000000 }; + int pselect_result = pselect(send_fd + 1, + NULL, + &writefds, + NULL, + &pselect_timeout, + &wait_mask); + if ( pselect_result != 0 ) + errx(1, + "finite pselect reported full peer writable: result=%d", + pselect_result); + if ( sigprocmask(SIG_BLOCK, NULL, &after_mask) < 0 ) + err(1, "read signal mask after pselect"); + if ( sigismember(&after_mask, SIGUSR1) != 1 ) + errx(1, "pselect did not restore the pre-wait signal mask"); + if ( sigprocmask(SIG_SETMASK, &old_mask, NULL) < 0 ) + err(1, "restore signal mask"); + if ( sigprocmask(SIG_BLOCK, &blocked_mask, &old_mask) < 0 ) + err(1, "block SIGUSR1 before ppoll"); + pfd.revents = 0; + struct timespec ppoll_timeout = { .tv_sec = 0, .tv_nsec = 120000000 }; + int ppoll_result = ppoll(&pfd, 1, &ppoll_timeout, &wait_mask); + if ( ppoll_result != 0 ) + errx(1, + "finite ppoll reported full peer writable: result=%d", + ppoll_result); + if ( sigprocmask(SIG_BLOCK, NULL, &after_mask) < 0 ) + err(1, "read signal mask after ppoll"); + if ( sigismember(&after_mask, SIGUSR1) != 1 ) + errx(1, "ppoll did not restore the pre-wait signal mask"); + if ( sigprocmask(SIG_SETMASK, &old_mask, NULL) < 0 ) + err(1, "restore signal mask after ppoll"); + int epoll_fd = epoll_create1(EPOLL_CLOEXEC); + if ( epoll_fd < 0 ) + err(1, "epoll_create1"); + struct epoll_event interest = { .events = EPOLLOUT, .data.fd = send_fd }; + if ( epoll_ctl(epoll_fd, EPOLL_CTL_ADD, send_fd, &interest) < 0 ) + err(1, "epoll_ctl"); + struct epoll_event event; + if ( epoll_wait(epoll_fd, &event, 1, 120) != 0 ) + errx(1, "finite epoll reported full peer writable"); + if ( close(epoll_fd) < 0 ) + err(1, "close epoll"); + uint32_t payload = htobe32(128); errno = 0; if ( send(send_fd, &payload, sizeof(payload), 0) != -1 ) @@ -83,10 +310,152 @@ int main(void) errx(1, "sequence %u arrived as %u", expected, be32toh(actual)); } - if ( close(send_fd) < 0 || close(recv_fd) < 0 ) + // Exercise a genuinely parked blocking send through a pthread's distinct + // syscall channel. Draining one slot must wake that sender, and the newly + // admitted datagram must remain at the tail of the preserved queue. + if ( fcntl(send_fd, F_SETFL, flags) < 0 ) + err(1, "restore blocking sender"); + struct timeval send_timeout = { .tv_sec = 1, .tv_usec = 0 }; + if ( setsockopt(send_fd, + SOL_SOCKET, + SO_SNDTIMEO, + &send_timeout, + sizeof(send_timeout)) < 0 ) + err(1, "set bounded sender timeout"); + const uint32_t blocking_base = 1000; + for ( uint32_t sequence = 0; sequence < 128; sequence++ ) + { + uint32_t blocking_payload = htobe32(blocking_base + sequence); + if ( send(send_fd, + &blocking_payload, + sizeof(blocking_payload), + 0) != (ssize_t) sizeof(blocking_payload) ) + err(1, "refill before blocking send at %u", sequence); + } + struct blocking_send_context send_context = { + .fd = send_fd, + .payload = htobe32(blocking_base + 128), + .started = 0, + .done = 0, + .result = -1, + .errnum = 0, + }; + pthread_t send_thread; + int thread_error = pthread_create(&send_thread, + NULL, + blocking_send_thread, + &send_context); + if ( thread_error ) + { + errno = thread_error; + err(1, "create blocking send thread"); + } + wait_for_thread_start(&send_context.started, "blocking send"); + usleep(20000); + if ( atomic_load_explicit(&send_context.done, memory_order_acquire) ) + errx(1, "send to full reliable queue did not remain blocked"); + uint32_t blocking_first = 0; + if ( recv(recv_fd, + &blocking_first, + sizeof(blocking_first), + MSG_DONTWAIT) != (ssize_t) sizeof(blocking_first) ) + err(1, "dequeue to wake blocked sender"); + if ( be32toh(blocking_first) != blocking_base ) + errx(1, + "blocking-send first sequence was %u", + be32toh(blocking_first)); + join_thread(send_thread, "blocking send"); + if ( send_context.result != (ssize_t) sizeof(send_context.payload) ) + { + if ( send_context.result < 0 ) + { + errno = send_context.errnum; + err(1, "blocked sender did not resume after dequeue"); + } + errx(1, + "blocked sender returned %zi after dequeue", + send_context.result); + } + for ( uint32_t expected = 1; expected <= 128; expected++ ) + { + uint32_t actual = 0; + if ( recv(recv_fd, &actual, sizeof(actual), MSG_DONTWAIT) != + (ssize_t) sizeof(actual) ) + err(1, "recv blocking-send sequence %u", expected); + if ( be32toh(actual) != blocking_base + expected ) + errx(1, + "blocking-send sequence %u arrived as %u", + blocking_base + expected, + be32toh(actual)); + } + + // A full reliable queue must not strand its sender after the receiver + // shuts down reads. The sender becomes writable only to report EPIPE, and + // MSG_NOSIGNAL must suppress the corresponding SIGPIPE delivery. + for ( uint32_t sequence = 0; sequence < 128; sequence++ ) + { + uint32_t shutdown_payload = htobe32(sequence); + if ( send(send_fd, + &shutdown_payload, + sizeof(shutdown_payload), + 0) != (ssize_t) sizeof(shutdown_payload) ) + err(1, "refill before read shutdown at %u", sequence); + } + pfd.revents = 0; + if ( poll(&pfd, 1, 0) != 0 || pfd.revents != 0 ) + errx(1, "refilled peer unexpectedly writable: revents=%#x", pfd.revents); + struct blocking_poll_context poll_context = { + .fd = send_fd, + .started = 0, + .done = 0, + .result = -1, + .revents = 0, + .errnum = 0, + }; + pthread_t poll_thread; + thread_error = pthread_create(&poll_thread, + NULL, + blocking_poll_thread, + &poll_context); + if ( thread_error ) + { + errno = thread_error; + err(1, "create blocking poll thread"); + } + wait_for_thread_start(&poll_context.started, "blocking poll"); + usleep(20000); + if ( atomic_load_explicit(&poll_context.done, memory_order_acquire) ) + errx(1, "POLLOUT wait on full reliable queue did not remain blocked"); + if ( shutdown(recv_fd, SHUT_RD) < 0 ) + err(1, "receiver SHUT_RD"); + join_thread(poll_thread, "blocking poll"); + if ( poll_context.result != 1 || !(poll_context.revents & POLLOUT) ) + { + if ( poll_context.result < 0 ) + { + errno = poll_context.errnum; + err(1, "read shutdown failed the POLLOUT waiter"); + } + errx(1, + "read shutdown did not wake POLLOUT waiter: result=%d revents=%#x", + poll_context.result, + poll_context.revents); + } + pfd.revents = 0; + if ( poll(&pfd, 1, 0) != 1 || !(pfd.revents & POLLOUT) ) + errx(1, "read-shut peer did not release sender: revents=%#x", pfd.revents); + errno = 0; + if ( send(send_fd, &payload, sizeof(payload), MSG_NOSIGNAL) != -1 ) + errx(1, "send to read-shut peer unexpectedly succeeded"); + if ( errno != EPIPE ) + err(1, "send to read-shut peer"); + + if ( close(attacker_fd) < 0 || close(send_fd) < 0 || close(recv_fd) < 0 ) err(1, "close"); - if ( unlink(path) < 0 ) - err(1, "unlink after close"); + if ( unlink(send_path) < 0 ) + err(1, "unlink sender after close"); + if ( unlink(recv_path) < 0 ) + err(1, "unlink receiver after close"); puts("ok"); return 0; }