perf(protocol): put a frame's header and payload on the wire in one write (#713)

`write_frame` sent the length, the kind byte and the payload as three
separate `write_all` calls. The pane socket is a loopback `TcpStream` with
`TCP_NODELAY` set, so each of those was its own segment and its own wakeup
on the far side: the client woke from `read()` about twice per frame just
to reassemble a header it had already been told the shape of. The daemon
frames every ConPTY read, so under output that is tens of thousands of
extra syscalls a second on each end.

Write the 5-byte header and the payload as one `write_vectored` instead.
The bytes on the wire are identical, and a writer with no native vectored
write still terminates on the loop's fallback — it just costs what it
used to.

Measured on a Ryzen 9 9950X (32 threads, integrated Radeon), release
build, one pane in a 2182x1361 window, a generator emitting 20000
80-column lines/s (1.64 MB/s), three 25-second samples each. CPU is % of
one core; "reads/frame" is `TTY7_TRACE`'s client socket reads divided by
its frame count.

| | tty7-app (GUI) | daemon | conpty host | total | reads/frame |
|---|---|---|---|---|---|
| before | 22.0 | 3.8 | 2.3 | 28.1 | 1.90 |
| after | 17.7 | 3.1 | 2.5 | 23.3 | 0.93 |

Socket reads per frame halve, and CPU at a fixed output rate falls 17%.
Under an unbounded flood the pipeline instead gets faster — 35.7 MB/s
before, 47.0 MB/s after — so that case is not rate-matched and is not
claimed as a CPU win.

This does not on its own explain #713: at the rate that issue describes
(a held Return, ~30 lines/s) tty7 costs about 11% of one core here either
way. It is a real cost on the output path regardless.

crates/tty7-core/src/daemon/protocol.rs

@@ -5,6 +5,9 @@ use serde::{Deserialize, Serialize};
 
 pub const MAX_FRAME: usize = 64 * 1024 * 1024;
 
+/// A frame is a little-endian `u32` length, a one-byte kind, then the payload.
+const HEADER: usize = 5;
+
 pub const PROTOCOL_VERSION: u32 = 6;
 
 pub const FEATURE_PANE_OWNER: &str = "pane-owner";
@@ -952,9 +955,33 @@ pub fn write_frame<W: Write>(w: &mut W, kind: u8, payload: &[u8]) -> io::Result<
             "frame payload exceeds MAX_FRAME",
         ));
     }
-    w.write_all(&(len as u32).to_le_bytes())?;
-    w.write_all(&[kind])?;
-    w.write_all(payload)?;
+    // One write, not three. The pane socket is a loopback `TcpStream` with
+    // `TCP_NODELAY` set, so three `write_all`s put the length, the kind and the
+    // payload on the wire as three separate segments, and the reader on the far
+    // side wakes from `read()` three times for one frame. Under a PTY flood the
+    // daemon frames every ConPTY read, so that is two extra syscalls on each
+    // side per frame, tens of thousands a second (issue #713).
+    let mut header = [0u8; HEADER];
+    header[..4].copy_from_slice(&(len as u32).to_le_bytes());
+    header[4] = kind;
+    let mut bufs = [io::IoSlice::new(&header), io::IoSlice::new(payload)];
+    let mut rest: &mut [io::IoSlice<'_>] = &mut bufs;
+    while !rest.is_empty() {
+        match w.write_vectored(rest) {
+            Ok(0) => {
+                return Err(io::Error::new(
+                    io::ErrorKind::WriteZero,
+                    "the frame could not be written in full",
+                ));
+            }
+            // A writer that does not implement `write_vectored` natively falls
+            // back to writing the first non-empty slice, so this loop still
+            // terminates — it just costs the two writes it used to cost.
+            Ok(n) => io::IoSlice::advance_slices(&mut rest, n),
+            Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
+            Err(e) => return Err(e),
+        }
+    }
     Ok(())
 }
 
@@ -984,7 +1011,6 @@ pub fn is_error_kind(kind: u8) -> bool {
 }
 
 pub fn take_frame(buf: &mut Vec<u8>) -> io::Result<Option<(u8, Vec<u8>)>> {
-    const HEADER: usize = 5;
     if buf.len() < HEADER {
         return Ok(None);
     }
@@ -2046,6 +2072,57 @@ mod tests {
         assert!(buf.is_empty());
     }
 
+    /// The pane socket has `TCP_NODELAY` set, so a write is a segment and a
+    /// segment is a wakeup on the far side. A frame must therefore cost one
+    /// write, not one for the length, one for the kind and one for the payload
+    /// (issue #713) — at flood rates that difference is tens of thousands of
+    /// syscalls a second on each end.
+    #[test]
+    fn a_frame_is_one_write_on_a_vectored_writer() {
+        #[derive(Default)]
+        struct Counting {
+            writes: usize,
+            bytes: Vec<u8>,
+        }
+        impl Write for Counting {
+            fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+                self.writes += 1;
+                self.bytes.extend_from_slice(buf);
+                Ok(buf.len())
+            }
+            fn write_vectored(&mut self, bufs: &[io::IoSlice<'_>]) -> io::Result<usize> {
+                self.writes += 1;
+                let mut n = 0;
+                for b in bufs {
+                    self.bytes.extend_from_slice(b);
+                    n += b.len();
+                }
+                Ok(n)
+            }
+            fn flush(&mut self) -> io::Result<()> {
+                Ok(())
+            }
+        }
+
+        let mut w = Counting::default();
+        write_frame(&mut w, kind::OUTPUT, b"a chunk of pty output").expect("write the frame");
+        assert_eq!(w.writes, 1, "one frame must cost one write");
+
+        // An empty payload is a frame too — the header still has to land, and
+        // the empty second slice must not spin the loop.
+        let mut empty = Counting::default();
+        write_frame(&mut empty, kind::DETACH, &[]).expect("write the empty frame");
+        assert_eq!(empty.writes, 1);
+
+        // Whatever the write count, the bytes on the wire are unchanged: a
+        // `read_frame` over them gives back exactly what went in.
+        let mut cursor = io::Cursor::new(w.bytes);
+        assert_eq!(
+            read_frame(&mut cursor).expect("read it back"),
+            (kind::OUTPUT, b"a chunk of pty output".to_vec())
+        );
+    }
+
     #[test]
     fn from_frame_rejects_unknown_kind() {
         assert!(ClientMsg::from_frame(99, vec![]).is_err());