enhance(worker): send frame header with the sendfile payload (#462)

The serve path wrote the response header from the io_uring ring and then
handed the socket fd to the blocking pool for sendfile. That is two
ring/pool transitions per request: one futex wake to park the ring task
plus a tiny header-only TCP segment ahead of the payload.

Move the header into the same blocking step as sendfile and emit it with
MSG_MORE, so the kernel coalesces it with the first payload chunk. The
ring now hands off once per request instead of twice.

Measured on falcon-zan-ca (8-core pod, 64 KiB ranges, 320 connections
across two client pods on separate nodes). Both arms pre-warmed, live PID
verified per arm, alternating rounds -- the node's throughput is capped by
the client link at ~12 Gbps, so the comparison is CPU per byte served:

  round | before      | after       | delta
  ------|-------------|-------------|-------
    1   | 3.88 Gbps/c | 4.14 Gbps/c | +6.7%
    2   | 4.04 Gbps/c | 4.23 Gbps/c | +4.7%
    3   | 4.02 Gbps/c | 4.21 Gbps/c | +4.7%

Worker CPU falls from 2.98 to 2.84 cores at identical served bandwidth;
4/4 paired rounds positive.

sendfile.rs gains three tests: header-then-payload byte exactness, the
same across a sub-range with a chunk size small enough to force multiple
sendfile calls, and the empty-payload case -- MSG_MORE corks the header,
so a zero-length payload must explicitly flush or the bytes never leave.

Co-authored-by: Kazimierz Szilard <309214913+Kazimierzsier@users.noreply.github.com>

Author: Kazimierzsier

crates/talon-worker/src/lib.rs

@@ -38,7 +38,7 @@ pub use miss::{touched_pages, Admission, InFlightGuard, InFlightLoads, LoadKey};
 pub use observability::{serve_admin, WorkerMetrics, WorkerObservability, WorkerReadiness};
 pub use paged_store::PagedBlockStore;
 pub use runtime::{ServeOutcome, WorkerRuntime};
-pub use sendfile::{send_file_range, DEFAULT_CHUNK};
+pub use sendfile::{send_file_range, send_header_and_file_range, DEFAULT_CHUNK};
 pub use splice::{ingest_put, splice_to_file};
 pub use staging::{Checksum, Stager};
 pub use write_cache::{FlushItem, WriteCache};

crates/talon-worker/src/sendfile.rs

@@ -64,6 +64,69 @@ pub fn send_file_range(
     Ok(sent_total)
 }
 
+/// Send `header` and then `[offset, offset + len)` of `file` to `sock` in a
+/// single blocking step.
+///
+/// This is [`send_file_range`] with the response frame header prepended. Doing
+/// both here rather than writing the header from the ring saves a full
+/// ring-to-blocking-pool round-trip per request — measured as the dominant
+/// per-request cost on the serve path, since the payload copy itself is
+/// already zero-copy.
+///
+/// The header goes out with `MSG_MORE` so the kernel holds it back and
+/// coalesces it with the first `sendfile` chunk into one TCP segment instead
+/// of emitting a tiny header-only packet.
+///
+/// Returns the number of *payload* bytes sent (header bytes are not counted),
+/// so the caller can apply the same short-send check as [`send_file_range`].
+pub fn send_header_and_file_range(
+    sock: &impl AsRawFd,
+    header: &[u8],
+    file: &impl AsRawFd,
+    offset: u64,
+    len: u64,
+    chunk: usize,
+) -> io::Result<u64> {
+    let out_fd: RawFd = sock.as_raw_fd();
+
+    let mut written = 0usize;
+    while written < header.len() {
+        // SAFETY: valid fd; the pointer/length pair stays inside `header`.
+        let n = unsafe {
+            libc::send(
+                out_fd,
+                header[written..].as_ptr() as *const libc::c_void,
+                header.len() - written,
+                libc::MSG_MORE | libc::MSG_NOSIGNAL,
+            )
+        };
+        if n < 0 {
+            let err = io::Error::last_os_error();
+            match err.raw_os_error() {
+                Some(libc::EINTR) => continue,
+                _ => return Err(err),
+            }
+        }
+        if n == 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::WriteZero,
+                "socket accepted no header bytes",
+            ));
+        }
+        written += n as usize;
+    }
+
+    // `MSG_MORE` leaves the header corked; the sendfile below uncorks it by
+    // filling the segment. A zero-length payload would strand it, so flush.
+    if len == 0 {
+        // SAFETY: valid fd; a zero-length send with no MSG_MORE flushes.
+        unsafe { libc::send(out_fd, [].as_ptr(), 0, libc::MSG_NOSIGNAL) };
+        return Ok(0);
+    }
+
+    send_file_range(sock, file, offset, len, chunk)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -138,4 +201,62 @@ mod tests {
         let got = roundtrip(data, 0, 1000, DEFAULT_CHUNK);
         assert_eq!(got, data);
     }
+
+    /// Same as [`roundtrip`] but through the header-coalescing entry point.
+    fn roundtrip_with_header(
+        header: &[u8],
+        data: &[u8],
+        offset: u64,
+        len: u64,
+        chunk: usize,
+    ) -> Vec<u8> {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let addr = listener.local_addr().unwrap();
+        let file = temp_file_with(data);
+        let header = header.to_vec();
+
+        let server = std::thread::spawn(move || {
+            let (mut conn, _) = listener.accept().unwrap();
+            let sent =
+                send_header_and_file_range(&conn, &header, &file, offset, len, chunk).unwrap();
+            conn.flush().unwrap();
+            sent
+        });
+
+        let mut client = TcpStream::connect(addr).unwrap();
+        let mut got = Vec::new();
+        client.read_to_end(&mut got).unwrap();
+        let payload_sent = server.join().unwrap();
+        // The return value counts payload only, never the header.
+        assert_eq!(payload_sent as usize, got.len() - HDR.len());
+        got
+    }
+
+    const HDR: &[u8] = b"\x01\x02\x03\x04HEADERBYTES!";
+
+    #[test]
+    fn header_precedes_payload_byte_exactly() {
+        let data: Vec<u8> = (0..8192u32).map(|i| (i % 256) as u8).collect();
+        let got = roundtrip_with_header(HDR, &data, 0, data.len() as u64, DEFAULT_CHUNK);
+        assert_eq!(&got[..HDR.len()], HDR);
+        assert_eq!(&got[HDR.len()..], &data[..]);
+    }
+
+    #[test]
+    fn header_precedes_sub_range_and_survives_chunking() {
+        let data: Vec<u8> = (0..10_000u32).map(|i| (i * 13 % 251) as u8).collect();
+        let (off, len) = (777u64, 3333u64);
+        // A tiny chunk forces many sendfile calls after the corked header.
+        let got = roundtrip_with_header(HDR, &data, off, len, 64);
+        assert_eq!(&got[..HDR.len()], HDR);
+        assert_eq!(&got[HDR.len()..], &data[off as usize..(off + len) as usize]);
+    }
+
+    /// `MSG_MORE` corks the header; with no payload to uncork it the bytes
+    /// would sit in the kernel forever. The zero-length flush must release it.
+    #[test]
+    fn header_is_flushed_when_payload_is_empty() {
+        let got = roundtrip_with_header(HDR, b"anything", 0, 0, DEFAULT_CHUNK);
+        assert_eq!(got, HDR);
+    }
 }

crates/talon-worker/src/uring_conn.rs

@@ -64,7 +64,7 @@ use talon_transport::uring::{read_frame, write_all};
 
 use crate::observability::WorkerObservability;
 use crate::runtime::{ServeOutcome, WorkerRuntime};
-use crate::{send_file_range, DEFAULT_CHUNK};
+use crate::{send_header_and_file_range, DEFAULT_CHUNK};
 
 /// Serve one accepted data-plane connection until EOF or a fatal error.
 pub async fn handle_conn(
@@ -167,8 +167,7 @@ pub async fn handle_conn(
             Ok(ServeOutcome::Sendfile(handle)) => {
                 let len = handle.len;
                 let hdr = data::response_header_ok(h.request_id, len as u32).to_vec();
-                write_all(&mut stream, hdr).await?;
-                match sendfile_payload(&stream, handle).await {
+                match sendfile_payload(&stream, hdr, handle).await {
                     Ok(()) => observability
                         .metrics()
                         .record_request_success(len, request_started.elapsed()),
@@ -220,21 +219,31 @@ fn rejoin(header: &FrameHeader, payload: &[u8]) -> Vec<u8> {
 }
 
 /// Stream a resident block to the client with `sendfile(2)` from the blocking
-/// pool.
+/// pool, writing the response frame header in the same blocking step.
 ///
 /// The ring keeps ownership of the socket throughout: only the raw fd crosses
 /// to the blocking thread, so there is no `into_std`/`from_std` round-trip and
 /// no non-blocking-mode toggling. `sendfile` must not run on the ring — it is
 /// blocking, and a slow client would stall every connection this ring owns.
-async fn sendfile_payload(stream: &TcpStream, handle: BlockHandle) -> anyhow::Result<()> {
+///
+/// The header travels with the payload rather than being written from the ring
+/// first: that removes one ring-to-pool hand-off (and its futex wake) per
+/// request, and `MSG_MORE` lets the kernel put the header and the first
+/// payload chunk in one segment.
+async fn sendfile_payload(
+    stream: &TcpStream,
+    header: Vec<u8>,
+    handle: BlockHandle,
+) -> anyhow::Result<()> {
     let sock_fd = stream.as_raw_fd();
     let len = handle.len;
     let sent = monoio::spawn_blocking(move || {
         // SAFETY-adjacent note: the fd outlives this call because `stream` is
         // borrowed for the duration of the await, and the connection task is the
         // only owner.
-        send_file_range(
+        send_header_and_file_range(
             &FdRef(sock_fd),
+            &header,
             &handle.fd,
             handle.offset,
             handle.len,