From 737b57cd0f676c41e4a8c2b36e834fb63878d42d Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:00:53 +0000 Subject: [PATCH 1/3] test(cmux-tui): a connection's keys share one kernel queue for their room check (#35413 s17ga; red on macOS) The s17ga typing A/B for #35413 (STATUS 13245, 13254) kept the S1 input lane off: daemon part p50 was +0.016 ms with ON, against an OFF round-to-round spread of 0.009 ms, with both ON rounds above both OFF rounds. The whole difference was the key's write call (daemon.send.write -> written): 0.019 and 0.020 ms at p50 with ON, 0.005 ms with OFF. The inline key write (#35872) checks for room with poll(POLLOUT, 0) before its send. On XNU each poll() opens a kernel event queue, registers a knote on the socket under the socket's lock, and frees both again (sys_generic.c poll_nocancel; the #35872 reviewer's note 3). The test sends 100 keys to one idle connection and counts the kernel event queues that their room checks opened, with a test-only per-thread counter at the room check. It fails on this code: 100 keys open 100 queues, one per key. macOS only (Linux's poll opens no queue). --- .../src/surface/tests/input_port_inline.rs | 39 +++++++++++++++++++ .../src/terminal_host_runtime.rs | 2 + .../shared/attachment.rs | 23 +++++++++++ 3 files changed, 64 insertions(+) diff --git a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs index a33e2de58ee8..67ca1a143b89 100644 --- a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs +++ b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs @@ -55,6 +55,45 @@ fn a_key_to_an_idle_port_is_on_the_host_socket_when_its_request_returns() { ); } +/// #35413 s17ga A/B (STATUS 13254): with the lane on, a key's write call (daemon.send.write -> daemon.send.written) +/// took 0.019-0.020 ms at p50 on cmux21, against 0.005 ms for the old path's write, and the daemon part's p50 was +/// worse with ON beyond the OFF rounds' spread. The inline write's room check was a poll(), which on XNU opens a +/// kernel event queue, registers a knote on the socket and frees both on every call (sys_generic.c poll_nocancel), +/// under the socket's lock each time. A connection's keys share one queue for their room checks. +#[cfg(target_vendor = "apple")] +#[test] +fn a_connections_keys_share_one_kernel_queue_for_their_room_check() { + use crate::terminal_host_runtime::room_check_queues; + const KEYS: usize = 100; + let (_mux, surface, mut host) = hosted_typing_surface("hosted-input-port-inline-queue", true); + let pty = surface.as_pty().expect("a terminal"); + // The writer is held: a key that went to the port instead of inline never reaches the host. + let _resume = pty.input_port.pause_writes_for_test(); + host.set_read_timeout(Some(Duration::from_millis(500))).unwrap(); + let before = room_check_queues::count(); + for key in 0..KEYS { + surface.write_bytes_replying(b"k", InputReply::Accepted).expect("the port took the key"); + // Each key is read before the next one, so the socket has room and the port stays idle. + let frame = crate::terminal_host_protocol::read_frame( + &mut host, + crate::terminal_host_protocol::MAX_FRAME_PAYLOAD, + ) + .ok() + .flatten(); + assert!( + frame + .as_ref() + .is_some_and(|frame| frame.kind == MessageKind::Input && frame.payload == b"k"), + "key {key} was not written inline ({frame:?})" + ); + } + let opened = room_check_queues::count() - before; + assert!( + opened <= 1, + "{KEYS} keys to one connection opened {opened} kernel event queues for their room checks: one per key" + ); +} + /// Another writer holds the connection (a large write of another client): the key does not wait for it. It goes to /// the port, returns at once, and reaches the host once the other writer is done. #[test] diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime.rs index 7185c885cdb9..fa7b06031c2f 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime.rs @@ -656,6 +656,8 @@ pub use shared::attachment::launch::{ pub(crate) use shared::attachment::launch::{ launch_terminal_host_from, launch_terminal_host_seeded, }; +#[cfg(all(test, target_vendor = "apple"))] +pub(crate) use shared::attachment::room_check_queues; #[cfg(any(unix, windows))] pub(crate) use shared::clipboard_read::ClipboardReadSignal; #[cfg(any(unix, windows))] diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs index aaa043c45180..f4afc63bc239 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs @@ -247,6 +247,9 @@ const _: () = assert!( /// (#35872 review). XNU's poll also makes and frees a kqueue on each call, a few us per key. #[cfg(unix)] fn has_room_now(fd: std::os::fd::RawFd) -> bool { + // On XNU a poll() opens a kernel event queue for the call and frees it after (sys_generic.c poll_nocancel). + #[cfg(all(test, target_vendor = "apple"))] + room_check_queues::opened(); let mut ready = libc::pollfd { fd, events: libc::POLLOUT, revents: 0 }; // SAFETY: one valid pollfd for a count of 1; the zero timeout keeps poll from waiting for room. let count = unsafe { libc::poll(&mut ready, 1, 0) }; @@ -255,6 +258,26 @@ fn has_room_now(fd: std::os::fd::RawFd) -> bool { && ready.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) == 0 } +/// Tests: the kernel event queues that this thread's room checks ([`HostInputRoute::try_send_now`]) opened. An inline +/// key runs its room check on its request's own thread, so a test that sends keys counts its own keys' queues. +#[cfg(all(test, target_vendor = "apple"))] +pub(crate) mod room_check_queues { + use std::cell::Cell; + + thread_local! { + static OPENED: Cell = const { Cell::new(0) }; + } + + pub(crate) fn opened() { + OPENED.with(|opened| opened.set(opened.get().saturating_add(1))); + } + + /// The queues this thread's room checks opened so far. + pub(crate) fn count() -> usize { + OPENED.with(Cell::get) + } +} + /// No SIGPIPE from a send to a host that closed its end: the error is returned instead. Elsewhere (Apple targets) std /// sets SO_NOSIGPIPE on the sockets it creates and the Rust runtime ignores SIGPIPE, as for every other write on this /// socket; the send fails with EPIPE. The same gate as `pty_custody`'s sends. From 0fffc797ae45242e7b85bec68c83018de03ae94c Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:13:40 +0000 Subject: [PATCH 2/3] fix(cmux-tui): the inline key write checks room with one kernel queue per connection and sends without a copy (#35413 s17ga) The test is the commit before this one (red on macOS: 100 keys to one connection opened 100 kernel event queues, one per key's room check). s17ga (STATUS 13245, 13254, 13302) kept the S1 input lane off. On one boundary for both arms, ON was worse at p50 beyond the OFF spread. The whole difference was the write call on the key's own request thread: 0.019-0.020 ms with the inline write, against 0.005 ms for the old path's direct write. The inline write did three things the old path did not: 1. poll(POLLOUT, 0) before the send. On XNU each poll() opens a kernel event queue, registers a knote under the socket lock and frees both. 2. Two heap copies of the key (Frame::new(payload.to_vec()), encode_frame). 3. One send of a frame it had just built. Now (terminal_host_runtime/shared/attachment/room_check.rs): - macOS: each host connection keeps one kernel event queue, with the socket's EVFILT_WRITE registered once at its first key. Each key reads it with kevent64(KEVENT_FLAG_IMMEDIATE): one call, no allocation, no sleep. The filter's data is the socket's free room; the key goes inline only when it holds the whole frame. EV_EOF or EV_ERROR is no room, so the key goes to the port and its reconnect path. A failed kqueue() falls back to poll(). The connection's attachment owns the queue and shares it with every route clone; it closes with the attachment, and a reconnect makes a new one. - Linux: no check. MSG_DONTWAIT is honoured, and a Unix stream socket takes a frame this small whole or not at all. - Every Unix target: the frame header is built on the stack (terminal_host_protocol::input_frame_header, pinned to encode_frame's output by the_inline_frame_header_is_the_encoders_header). The header and the key go out in one sendmsg of two iovecs, with no heap copy of the key. The order rules, the size limit and the fallbacks to the port are the same. The room check still runs under the connection's writer guard, so nothing takes the room before the send. --- .../src/surface/tests/input_port_inline.rs | 23 +++ .../src/terminal_host_protocol.rs | 19 ++ .../shared/attachment.rs | 114 ++++++------ .../shared/attachment/connect.rs | 2 + .../shared/attachment/room_check.rs | 163 ++++++++++++++++++ .../src/terminal_host_runtime/unix/tests.rs | 1 + .../unix/tests/attach_protocol.rs | 3 + .../unix/tests/input_records.rs | 3 + .../unix/tests/launch_codec.rs | 1 + 9 files changed, 266 insertions(+), 63 deletions(-) create mode 100644 cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/room_check.rs diff --git a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs index 67ca1a143b89..a390f63440a6 100644 --- a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs +++ b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs @@ -94,6 +94,29 @@ fn a_connections_keys_share_one_kernel_queue_for_their_room_check() { ); } +/// The inline write builds the frame header on the stack (no copy of the key): it must be the header `encode_frame` +/// writes for the same frame, byte for byte, or the host reads a corrupt stream. +#[test] +fn the_inline_frame_header_is_the_encoders_header() { + for payload in [&b"k"[..], &[0u8; 1024][..], &b""[..]] { + let mut frame = Frame::new(MessageKind::Input, payload.to_vec()); + frame.version = 7; + let encoded = crate::terminal_host_protocol::encode_frame(&frame).unwrap(); + let header = crate::terminal_host_protocol::input_frame_header( + 7, + MessageKind::Input, + u32::try_from(payload.len()).unwrap(), + ); + assert_eq!( + &encoded[..header.len()], + &header[..], + "the header of a {}-byte key", + payload.len() + ); + assert_eq!(&encoded[header.len()..], payload); + } +} + /// Another writer holds the connection (a large write of another client): the key does not wait for it. It goes to /// the port, returns at once, and reaches the host once the other writer is done. #[test] diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_protocol.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_protocol.rs index b855b600f8e0..9bbe15409aca 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_protocol.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_protocol.rs @@ -727,6 +727,25 @@ fn encode_header(frame: &Frame, max_payload: usize) -> Result<[u8; HEADER_LEN], Ok(header) } +/// The header of a frame with no flags, request id or sequence (an `Input` key), as [`write_frame`] writes it, +/// built on the stack: the inline key write sends it and the payload in one call, with no copy of the payload. +/// `version` is the connection's (never 0) and `payload_len` at most [`MAX_FRAME_PAYLOAD`]. +pub(crate) fn input_frame_header( + version: u16, + kind: MessageKind, + payload_len: u32, +) -> [u8; HEADER_LEN] { + let [m0, m1, m2, m3] = MAGIC; + let [v0, v1] = version.to_le_bytes(); + let [k0, k1] = (kind as u16).to_le_bytes(); + let [l0, l1, l2, l3] = payload_len.to_le_bytes(); + // magic, version, kind, flags 0, payload length, request id 0, sequence 0 (`encode_header`'s layout). + [ + m0, m1, m2, m3, v0, v1, k0, k1, 0, 0, 0, 0, l0, l1, l2, l3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, + ] +} + pub fn write_frame(writer: &mut impl Write, frame: &Frame) -> Result<(), ProtocolError> { let header = encode_header(frame, MAX_FRAME_PAYLOAD)?; writer.write_all(&header)?; diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs index f4afc63bc239..46380bb73e99 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment.rs @@ -44,7 +44,11 @@ use super::renderer_grant::ControlRequestUnanswered; mod connect; pub(crate) mod launch; mod pending_control; +#[cfg(unix)] +mod room_check; pub(crate) use pending_control::PendingControlResponse; +#[cfg(all(test, target_vendor = "apple"))] +pub(crate) use room_check::room_check_queues; mod terminate; // Only the Unix host calls these until the Windows host lands. #[cfg_attr(not(unix), allow(unused_imports))] @@ -146,6 +150,9 @@ pub struct HostAttachment { /// The terminal's input port order (R-DAEMON-STATE S1.6b): every other frame this attachment writes waits until /// the port wrote the keys it took before, so no frame passes a key. Set by the terminal that owns the attachment. pub(crate) input_order: Option, + /// The room check of this connection's inline keys, shared by every route of the connection. + #[cfg(unix)] + pub(crate) input_room: Arc, } impl std::fmt::Debug for HostAttachment { @@ -165,6 +172,9 @@ impl std::fmt::Debug for HostAttachment { pub(crate) struct HostInputRoute { writer: Arc>, protocol_version: u16, + /// The connection's room check for its inline keys (`room_check.rs`); every clone of the route shares it. + #[cfg(unix)] + room: Arc, } impl HostInputRoute { @@ -191,33 +201,27 @@ impl HostInputRoute { Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(), Err(std::sync::TryLockError::WouldBlock) => return false, }; - // Room first. On macOS a send with MSG_DONTWAIT still waits for room on a stream socket (XNU honours only - // the socket's own non-blocking mode there, and this socket must block: its reader shares it), so a key to a - // host that stopped reading waited here, holding this turn, the writer and the port lock (#35872: the - // stalled-host test). POLLOUT means room for at least the send low-water mark (2048 bytes on macOS, more - // than any inline frame; Linux reports it while under a quarter of the send buffer is used). The writer held - // here is the socket's only writer, so nothing takes that room before the send. - if !has_room_now(writer.as_raw_fd()) { + // Room first, where a send cannot be trusted to return at once (macOS: room_check.rs). The writer held here is + // the socket's only writer, so nothing takes that room before the send. + let Ok(payload_len) = u32::try_from(payload.len()) else { return false }; + let Some(total) = crate::terminal_host_protocol::HEADER_LEN.checked_add(payload.len()) + else { return false; - } - let mut frame = Frame::new(kind, payload.to_vec()); - frame.version = self.protocol_version; - let Ok(bytes) = crate::terminal_host_protocol::encode_frame(&frame) else { return false }; - // MSG_DONTWAIT as well, where it is honoured. With room for the whole frame, a Unix stream socket takes it - // whole: a partial write cannot happen, and if one ever did the stream would be corrupt, so the connection - // is shut down as `send_host_frame` does after a failed write (the port's writer then sends the frame on the - // reconnected host). - // SAFETY: the descriptor is the connection's open socket, kept alive by the writer guard; `bytes` is a live - // buffer of the given length. - let sent = unsafe { - libc::send( - writer.as_raw_fd(), - bytes.as_ptr().cast(), - bytes.len(), - libc::MSG_DONTWAIT | SEND_NO_SIGNAL, - ) }; - if sent == bytes.len() as isize { + if !self.room.has_room(writer.as_raw_fd(), total) { + return false; + } + let header = crate::terminal_host_protocol::input_frame_header( + self.protocol_version, + kind, + payload_len, + ); + // One send of the header (on the stack) and the payload (in place): no heap copy of the key. MSG_DONTWAIT as + // well, where it is honoured. With room for the whole frame, a Unix stream socket takes it whole: a partial + // write cannot happen, and if one ever did the stream would be corrupt, so the connection is shut down as + // `send_host_frame` does after a failed write (the port's writer then sends the frame on the reconnected host). + let sent = send_frame_now(writer.as_raw_fd(), &header, payload); + if usize::try_from(sent).is_ok_and(|sent| sent == total) { return true; } if sent > 0 { @@ -229,53 +233,32 @@ impl HostInputRoute { /// The largest frame payload [`HostInputRoute::try_send_now`] writes on the caller's turn: a person's key (a few /// bytes, a short paste of a word). Larger input goes through the port's writer. With its 32-byte header the frame -/// stays under the 2048-byte send low-water mark that `has_room_now`'s POLLOUT promises on macOS. +/// stays under the 2048-byte send low-water mark that the macOS room check promises (`room_check.rs`). #[cfg(unix)] pub(crate) const INLINE_FRAME_MAX: usize = 1024; -// The room POLLOUT promises on macOS (the default 2048-byte send low-water mark) must hold a whole inline frame, or +// The room the macOS room check promises (the default 2048-byte send low-water mark) must hold a whole inline frame, or // a send after it could still wait or write part of the frame. #[cfg(unix)] const _: () = assert!( INLINE_FRAME_MAX + crate::terminal_host_protocol::HEADER_LEN < 2048, - "an inline frame must fit the room POLLOUT promises" + "an inline frame must fit the room the room check promises" ); -/// Whether the socket `fd` can take a small write now (POLLOUT, no error or hang-up). It never waits for room or for -/// the host. On Linux the zero-timeout poll returns at once. On macOS it returns at once when there is room; with no -/// room, XNU's poll sleeps on its timer, which is already due (about 10 us, then one wake), before it reports none -/// (#35872 review). XNU's poll also makes and frees a kqueue on each call, a few us per key. +/// Sends `header` and `payload` as one message with MSG_DONTWAIT (and no SIGPIPE): the bytes the socket took, or -1. #[cfg(unix)] -fn has_room_now(fd: std::os::fd::RawFd) -> bool { - // On XNU a poll() opens a kernel event queue for the call and frees it after (sys_generic.c poll_nocancel). - #[cfg(all(test, target_vendor = "apple"))] - room_check_queues::opened(); - let mut ready = libc::pollfd { fd, events: libc::POLLOUT, revents: 0 }; - // SAFETY: one valid pollfd for a count of 1; the zero timeout keeps poll from waiting for room. - let count = unsafe { libc::poll(&mut ready, 1, 0) }; - count == 1 - && ready.revents & libc::POLLOUT != 0 - && ready.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) == 0 -} - -/// Tests: the kernel event queues that this thread's room checks ([`HostInputRoute::try_send_now`]) opened. An inline -/// key runs its room check on its request's own thread, so a test that sends keys counts its own keys' queues. -#[cfg(all(test, target_vendor = "apple"))] -pub(crate) mod room_check_queues { - use std::cell::Cell; - - thread_local! { - static OPENED: Cell = const { Cell::new(0) }; - } - - pub(crate) fn opened() { - OPENED.with(|opened| opened.set(opened.get().saturating_add(1))); - } - - /// The queues this thread's room checks opened so far. - pub(crate) fn count() -> usize { - OPENED.with(Cell::get) - } +fn send_frame_now(fd: std::os::fd::RawFd, header: &[u8], payload: &[u8]) -> isize { + let mut parts = [ + libc::iovec { iov_base: header.as_ptr().cast_mut().cast(), iov_len: header.len() }, + libc::iovec { iov_base: payload.as_ptr().cast_mut().cast(), iov_len: payload.len() }, + ]; + // SAFETY: zeroed msghdr is a valid empty message. + let mut message: libc::msghdr = unsafe { std::mem::zeroed() }; + message.msg_iov = parts.as_mut_ptr(); + message.msg_iovlen = 2; + // SAFETY: the descriptor is the connection's open socket, kept alive by the caller's writer guard; both parts are + // live buffers of their given lengths, and sendmsg only reads them. + unsafe { libc::sendmsg(fd, &message, libc::MSG_DONTWAIT | SEND_NO_SIGNAL) } } /// No SIGPIPE from a send to a host that closed its end: the error is returned instead. Elsewhere (Apple targets) std @@ -361,7 +344,12 @@ impl HostAttachment { /// The route the terminal's input port writes this connection's interactive input on. pub(crate) fn input_route(&self) -> HostInputRoute { - HostInputRoute { writer: self.writer.clone(), protocol_version: self.protocol_version } + HostInputRoute { + writer: self.writer.clone(), + protocol_version: self.protocol_version, + #[cfg(unix)] + room: Arc::clone(&self.input_room), + } } pub(crate) fn begin_input_confirmed( diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/connect.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/connect.rs index 1b90d336c2e1..8df4a50b1304 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/connect.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/connect.rs @@ -283,6 +283,8 @@ pub(crate) fn connect_record_at_version( launch_activation_pending, pty_custody: None, input_order: None, + #[cfg(unix)] + input_room: Default::default(), }; attachment.release_viewer_size()?; Ok(attachment) diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/room_check.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/room_check.rs new file mode 100644 index 000000000000..17bd1c487855 --- /dev/null +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/shared/attachment/room_check.rs @@ -0,0 +1,163 @@ +//! The room check of the inline key write (#35872; #35413 s17ga, STATUS 13254, 13302): whether one host connection's +//! socket takes a small frame whole, now, without waiting, on the key's own request thread. +//! +//! - macOS: a send with MSG_DONTWAIT on a blocking stream socket still waits for room (XNU honours only the socket's +//! own non-blocking mode there, and this socket blocks: its reader shares it), so the room is checked first. A +//! zero-timeout poll() did that, but XNU's poll opens a kernel event queue, registers a knote under the socket lock +//! and frees both on every call, and sleeps about 10 us when there is no room. The s17ga A/B measured the lane's +//! write call at 0.019-0.020 ms at p50 against 0.005 ms for the old path's direct write. Each connection now keeps +//! one queue with the socket's write filter registered once, and a key reads it with +//! kevent64(KEVENT_FLAG_IMMEDIATE): one call, no allocation, no sleep. +//! - Linux: MSG_DONTWAIT is honoured, and a Unix stream socket takes a frame this small whole or not at all, so the +//! send needs no check. +//! - Other Unix targets: a zero-timeout poll(). + +use std::os::fd::RawFd; + +/// One host connection's room check. Made with the connection's input route and used only under the connection's +/// writer guard: that writer is the socket's only writer, so the room cannot shrink between the check and the send. +#[derive(Default)] +pub(crate) struct RoomCheck { + /// The connection's kernel event queue, made at its first key; `None` when `kqueue()` failed (poll() instead). + #[cfg(target_vendor = "apple")] + queue: std::sync::OnceLock>, +} + +impl RoomCheck { + /// Whether the socket `fd` takes `len` more bytes now, whole, without waiting. The socket is the same for every + /// call: the route that owns this check also holds the connection's writer, so the descriptor stays open. + #[cfg(target_vendor = "apple")] + pub(crate) fn has_room(&self, fd: RawFd, len: usize) -> bool { + match self.queue.get_or_init(|| WriteQueue::register(fd).ok()) { + Some(queue) => queue.has_room(len), + None => poll_has_room(fd), + } + } + + /// Linux: always. The send's MSG_DONTWAIT refuses a frame that the socket cannot take whole now. + #[cfg(any(target_os = "linux", target_os = "android"))] + pub(crate) fn has_room(&self, _fd: RawFd, _len: usize) -> bool { + true + } + + #[cfg(all( + unix, + not(any(target_vendor = "apple", target_os = "linux", target_os = "android")) + ))] + pub(crate) fn has_room(&self, fd: RawFd, _len: usize) -> bool { + poll_has_room(fd) + } +} + +/// A kernel event queue with one socket's write filter registered. Level-triggered: a read reports the filter while +/// the socket has room for at least its send low-water mark (2048 bytes by default, more than any inline frame). +/// Closing the socket removes the registration (a later read then finds no room, and the key goes to the port); +/// dropping the queue closes it. +#[cfg(target_vendor = "apple")] +struct WriteQueue(std::os::fd::OwnedFd); + +#[cfg(target_vendor = "apple")] +impl WriteQueue { + fn register(socket: RawFd) -> std::io::Result { + use std::os::fd::{AsRawFd as _, FromRawFd as _}; + #[cfg(test)] + room_check_queues::opened(); + // SAFETY: no pointers. Darwin makes kqueues close-on-exec and close-on-fork. + let fd = unsafe { libc::kqueue() }; + if fd < 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: the successful call returned a new descriptor that only this queue owns. + let queue = Self(unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }); + let ident = u64::try_from(socket).map_err(|_| std::io::ErrorKind::InvalidInput)?; + let change = libc::kevent64_s { + ident, + filter: libc::EVFILT_WRITE, + flags: libc::EV_ADD | libc::EV_ENABLE, + fflags: 0, + data: 0, + udata: 0, + ext: [0; 2], + }; + // SAFETY: one initialized change and no output entries: registers the filter and returns without waiting. + let result = unsafe { + libc::kevent64( + queue.0.as_raw_fd(), + &change, + 1, + std::ptr::null_mut(), + 0, + 0, + std::ptr::null(), + ) + }; + if result < 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(queue) + } + + /// Whether the registered socket takes `len` more bytes now. KEVENT_FLAG_IMMEDIATE returns at once, with or + /// without room (a zero-timeout poll sleeps on an already-due timer when there is none). The filter's `data` is + /// the socket's room. EOF (the socket was shut down, or the host closed its end) or an error is no room: the key + /// goes to the port, whose writer meets the same state and handles it (a lost host's reconnect). + fn has_room(&self, len: usize) -> bool { + use std::os::fd::AsRawFd as _; + let mut event = std::mem::MaybeUninit::::uninit(); + // SAFETY: one writable event slot, written only when the call returns 1; the flag keeps the call from waiting. + let ready = unsafe { + libc::kevent64( + self.0.as_raw_fd(), + std::ptr::null(), + 0, + event.as_mut_ptr(), + 1, + libc::KEVENT_FLAG_IMMEDIATE, + std::ptr::null(), + ) + }; + if ready != 1 { + return false; + } + // SAFETY: the call returned 1, so it wrote the one event. + let event = unsafe { event.assume_init() }; + event.filter == libc::EVFILT_WRITE + && event.flags & (libc::EV_EOF | libc::EV_ERROR) == 0 + && usize::try_from(event.data).is_ok_and(|room| room >= len) + } +} + +/// Whether the socket `fd` can take a small write now (POLLOUT, no error or hang-up). The zero timeout keeps poll from +/// waiting for room or for the host; on macOS it is only the fallback when a connection has no event queue. +#[cfg(all(unix, not(any(target_os = "linux", target_os = "android"))))] +fn poll_has_room(fd: RawFd) -> bool { + // On XNU a poll() opens a kernel event queue for the call and frees it after (sys_generic.c poll_nocancel). + #[cfg(all(test, target_vendor = "apple"))] + room_check_queues::opened(); + let mut ready = libc::pollfd { fd, events: libc::POLLOUT, revents: 0 }; + // SAFETY: one valid pollfd for a count of 1; the zero timeout keeps poll from waiting for room. + let count = unsafe { libc::poll(&mut ready, 1, 0) }; + count == 1 + && ready.revents & libc::POLLOUT != 0 + && ready.revents & (libc::POLLERR | libc::POLLHUP | libc::POLLNVAL) == 0 +} + +/// Tests: the kernel event queues that this thread's room checks opened. An inline key runs its room check on its +/// request's own thread, so a test that sends keys counts its own keys' queues. +#[cfg(all(test, target_vendor = "apple"))] +pub(crate) mod room_check_queues { + use std::cell::Cell; + + thread_local! { + static OPENED: Cell = const { Cell::new(0) }; + } + + pub(crate) fn opened() { + OPENED.with(|opened| opened.set(opened.get().saturating_add(1))); + } + + /// The queues this thread's room checks opened so far. + pub(crate) fn count() -> usize { + OPENED.with(Cell::get) + } +} diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests.rs index 8f08bfc57905..b17d95531fca 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests.rs @@ -315,6 +315,7 @@ pub(crate) fn input_ack_surface_fixture() -> (HostAttachment, UnixStream) { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; (attachment, host) } diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/attach_protocol.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/attach_protocol.rs index 77de7eebdccc..20de27297ad8 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/attach_protocol.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/attach_protocol.rs @@ -98,6 +98,7 @@ fn timed_out_cell_pixel_ack_reconciles_when_the_response_arrives_late() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let (release_ack_tx, release_ack_rx) = std::sync::mpsc::channel(); let resolver = { @@ -216,6 +217,7 @@ fn detach_fence_reports_a_delayed_receipt_after_output_as_a_failure() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let (output_queued, output_seen) = sync_channel(1); let (release_ack, ack_release) = sync_channel(1); @@ -381,6 +383,7 @@ fn adoption_quota_reconfiguration_finishes_before_snapshot_use() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let responder = thread::spawn(move || { let request = read_frame(&mut host, MAX_FRAME_PAYLOAD).unwrap().unwrap(); diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/input_records.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/input_records.rs index fd320def63c4..9f662be716f1 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/input_records.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/input_records.rs @@ -40,6 +40,7 @@ fn receipted_input_never_reaches_a_legacy_host_without_ack_support() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let error = match attachment.begin_input_confirmed(b"must-not-send") { @@ -307,6 +308,7 @@ fn terminate_waits_for_the_authoritative_host_receipt() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let responder = thread::spawn(move || { let request = read_frame(&mut host, MAX_FRAME_PAYLOAD).unwrap().unwrap(); @@ -361,6 +363,7 @@ fn clear_history_control_write_failure_after_header_is_ambiguous() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let peer = thread::spawn(move || { let mut header = [0; crate::terminal_host_protocol::HEADER_LEN]; diff --git a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/launch_codec.rs b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/launch_codec.rs index 4288a3382df5..7ccf0353e1e2 100644 --- a/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/launch_codec.rs +++ b/cmux-tui/crates/cmux-tui-core/src/terminal_host_runtime/unix/tests/launch_codec.rs @@ -520,6 +520,7 @@ fn clear_history_ack_preserves_known_not_delivered_failure() { launch_activation_pending: false, pty_custody: None, input_order: None, + input_room: Default::default(), }; let responder = thread::spawn(move || { let request = read_frame(&mut host, MAX_FRAME_PAYLOAD).unwrap().unwrap(); From 058c3a23ab20fd9a1feac7708aa428f51348dfa0 Mon Sep 17 00:00:00 2001 From: lawrencecchen <54008264+lawrencecchen@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:30:36 +0000 Subject: [PATCH 3/3] test(cmux-tui): after a full host socket drains, the next key goes inline again; the header test covers Paste (#35988 review) Review of #35988 (SHIP at 0fffc797ae4), should-fix 1: no test sent a key after the socket was full and the host then read everything. On macOS that is the one state the per-connection room-check queue keeps between keys: if its write filter did not report room again, every later key on the connection would go to the port and nothing would say so. keys_to_a_host_that_stopped_reading_never_block_their_requests now ends by waiting for an idle port, holding the port's writer and sending one more key: it must reach the host inline. The order test uses the same pattern. Note 3: the_inline_frame_header_is_the_encoders_header also covers MessageKind::Paste, which reaches the inline write too. --- .../src/surface/tests/input_port_inline.rs | 46 ++++++++++++------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs index a390f63440a6..bccb5a88fd49 100644 --- a/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs +++ b/cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs @@ -98,22 +98,24 @@ fn a_connections_keys_share_one_kernel_queue_for_their_room_check() { /// writes for the same frame, byte for byte, or the host reads a corrupt stream. #[test] fn the_inline_frame_header_is_the_encoders_header() { - for payload in [&b"k"[..], &[0u8; 1024][..], &b""[..]] { - let mut frame = Frame::new(MessageKind::Input, payload.to_vec()); - frame.version = 7; - let encoded = crate::terminal_host_protocol::encode_frame(&frame).unwrap(); - let header = crate::terminal_host_protocol::input_frame_header( - 7, - MessageKind::Input, - u32::try_from(payload.len()).unwrap(), - ); - assert_eq!( - &encoded[..header.len()], - &header[..], - "the header of a {}-byte key", - payload.len() - ); - assert_eq!(&encoded[header.len()..], payload); + for kind in [MessageKind::Input, MessageKind::Paste] { + for payload in [&b"k"[..], &[0u8; 1024][..], &b""[..]] { + let mut frame = Frame::new(kind, payload.to_vec()); + frame.version = 7; + let encoded = crate::terminal_host_protocol::encode_frame(&frame).unwrap(); + let header = crate::terminal_host_protocol::input_frame_header( + 7, + kind, + u32::try_from(payload.len()).unwrap(), + ); + assert_eq!( + &encoded[..header.len()], + &header[..], + "the header of a {}-byte {kind:?}", + payload.len() + ); + assert_eq!(&encoded[header.len()..], payload); + } } } @@ -205,6 +207,18 @@ fn keys_to_a_host_that_stopped_reading_never_block_their_requests() { "the keys reached the host in another order, more than once, or not at all" ); }); + // The host read everything: the socket has room again, so once the port is idle the next key goes inline (on + // macOS this proves that the connection's room-check queue reports room again after the socket was full). With + // the port's writer held, a key that went to the port instead would miss the read. + let pty = surface.as_pty().expect("a terminal"); + wait_until("the port is idle", || pty.input_port.idle_for_test()); + let _held = pty.input_port.pause_writes_for_test(); + surface.write_bytes_replying(b"next", InputReply::Accepted).expect("the key after the drain"); + assert_eq!( + input_frames(&mut host, Duration::from_millis(300)), + [b"next".to_vec()], + "after the host read everything, the next key did not go inline" + ); } /// Order across the two paths: a key written inline, a write the port's writer holds, a key queued behind it, then