perf(cmux-tui): the inline key write checks room with one kernel queue per connection and sends without a copy (#35413 s17ga)

cmux-tui/crates/cmux-tui-core/src/surface/tests/input_port_inline.rs

@@ -55,6 +55,70 @@ 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"
+    );
+}
+
+/// 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 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);
+        }
+    }
+}
+
 /// 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]
@@ -143,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

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)?;

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))]

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<crate::surface::input_port::PortOrder>,
+    /// The room check of this connection's inline keys, shared by every route of the connection.
+    #[cfg(unix)]
+    pub(crate) input_room: Arc<room_check::RoomCheck>,
 }
 
 impl std::fmt::Debug for HostAttachment {
@@ -165,6 +172,9 @@ impl std::fmt::Debug for HostAttachment {
 pub(crate) struct HostInputRoute {
     writer: Arc<Mutex<HostStream>>,
     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<room_check::RoomCheck>,
 }
 
 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,30 +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 {
-    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
+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
@@ -338,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(

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)