consomme: deliver TCP window payload without an extra copy

vm/devices/net/net_backend/src/lib.rs

@@ -285,6 +285,31 @@ pub trait BufferAccess {
         self.write_data(id, data);
         self.write_header(id, metadata);
     }
+
+    /// Writes the packet header and a payload composed of multiple
+    /// discontiguous segments, in order, as a single logical packet.
+    ///
+    /// This allows callers to hand off a frame whose bytes are not contiguous
+    /// in memory (for example, an Ethernet/IP/TCP header followed by payload
+    /// that wraps a ring buffer) without first linearizing it into a scratch
+    /// buffer.
+    ///
+    /// The default implementation copies the segments into a temporary
+    /// contiguous buffer and forwards to [`BufferAccess::write_packet`].
+    /// Backends that write directly into guest memory should override this to
+    /// write each segment at its running offset and avoid the copy.
+    fn write_packet_segments(&mut self, id: RxId, metadata: &RxMetadata, segments: &[&[u8]]) {
+        if let [segment] = segments {
+            self.write_packet(id, metadata, segment);
+            return;
+        }
+        let total = segments.iter().map(|s| s.len()).sum();
+        let mut data = Vec::with_capacity(total);
+        for segment in segments {
+            data.extend_from_slice(segment);
+        }
+        self.write_packet(id, metadata, &data);
+    }
 }
 
 pub const ETHERNET_HEADER_LEN: u32 = 14;

vm/devices/net/net_consomme/consomme/src/lib.rs

@@ -517,13 +517,33 @@ pub trait Client {
     /// packet contains an IPv4 header, TCP header, and/or UDP header with a
     /// valid checksum.
     ///
-    /// TODO:
-    ///
-    /// 1. support >MTU sized packets (RSC/LRO/GRO)
-    /// 2. allow discontiguous data to eliminate the extra copy from the TCP
-    ///    window.
+    /// TODO: support >MTU sized packets (RSC/LRO/GRO).
     fn recv(&mut self, data: &[u8], checksum: &ChecksumState);
 
+    /// Transmits a packet whose bytes are provided as multiple discontiguous
+    /// segments, delivered in order as a single logical packet.
+    ///
+    /// This lets callers avoid linearizing a frame into a scratch buffer when
+    /// its payload is not contiguous (for example, a header segment followed by
+    /// TCP window bytes that wrap a ring buffer). The `checksum` argument has
+    /// the same meaning as for [`Client::recv`].
+    ///
+    /// The default implementation copies the segments into a contiguous buffer
+    /// and forwards to [`Client::recv`]. Clients that can write discontiguous
+    /// data directly to the guest should override this to eliminate the copy.
+    fn recv_segments(&mut self, segments: &[&[u8]], checksum: &ChecksumState) {
+        if let [segment] = segments {
+            self.recv(segment, checksum);
+            return;
+        }
+        let total = segments.iter().map(|s| s.len()).sum();
+        let mut data = Vec::with_capacity(total);
+        for segment in segments {
+            data.extend_from_slice(segment);
+        }
+        self.recv(&data, checksum);
+    }
+
     /// Specifies the maximum size for the next call to `recv`.
     ///
     /// This is the MTU including the Ethernet frame header. This must be at

vm/devices/net/net_consomme/consomme/src/tcp.rs

@@ -675,6 +675,7 @@ struct Sender<'a, T> {
 
 impl<T: Client> Sender<'_, T> {
     fn send_packet(&mut self, tcp: &TcpRepr<'_>, payload: Option<ring::View<'_>>) {
+        let payload_len = payload.as_ref().map_or(0, |p| p.len());
         let buffer = &mut self.state.buffer;
         let mut eth_packet = EthernetFrame::new_unchecked(&mut buffer[..]);
         eth_packet.set_dst_addr(self.state.params.client_mac);
@@ -683,7 +684,7 @@ impl<T: Client> Sender<'_, T> {
             self.ft.dst.ip().into(),
             self.ft.src.ip().into(),
             IpProtocol::Tcp,
-            tcp.header_len() + payload.as_ref().map_or(0, |p| p.len()),
+            tcp.header_len() + payload_len,
             64,
         );
         // Set the ethernet type based on IP version
@@ -714,26 +715,66 @@ impl<T: Client> Sender<'_, T> {
         let dst_ip_addr: IpAddress = self.ft.dst.ip().into();
         let src_ip_addr: IpAddress = self.ft.src.ip().into();
         let mut tcp_packet = TcpPacket::new_unchecked(tcp_payload_buf);
-        // Checksum is filled by `fill_checksum` below, after the payload is copied in.
+        // Checksums are computed explicitly below, so skip smoltcp's pass.
         tcp.emit(
             &mut tcp_packet,
             &dst_ip_addr,
             &src_ip_addr,
             &ChecksumCapabilities::ignored(),
         );
 
-        // Copy payload into TCP packet
-        if let Some(payload) = &payload {
-            payload.copy_to_slice(tcp_packet.payload_mut());
-        }
-        tcp_packet.fill_checksum(&dst_ip_addr, &src_ip_addr);
-        let n = ETHERNET_HEADER_LEN + ip_total_len;
         let checksum_state = match self.ft.dst {
             SocketAddr::V4(_) => ChecksumState::TCP4,
             SocketAddr::V6(_) => ChecksumState::TCP6,
         };
+        let n = ETHERNET_HEADER_LEN + ip_total_len;
+
+        let payload = match payload {
+            Some(p) if p.len() != 0 => p,
+            _ => {
+                tcp_packet.fill_checksum(&dst_ip_addr, &src_ip_addr);
+                let buffer = &self.state.buffer;
+                self.client.recv(&buffer[..n], &checksum_state);
+                return;
+            }
+        };
 
-        self.client.recv(&buffer[..n], &checksum_state);
+        // Zero-copy payload path: leave the TCP window bytes in place and hand
+        // the header and (up to two) payload slices to the client as
+        // discontiguous segments, avoiding a copy of the payload into the
+        // header buffer. The TCP checksum must be recomputed across those
+        // segments because the payload was never linearized here.
+        let (a, b) = payload.as_slices();
+        let tcp_header_len = tcp_packet.header_len() as usize;
+        tcp_packet.set_checksum(0);
+        // The TCP header length is a multiple of 4, so the header and `a` are
+        // 16-bit aligned; `b` shifts by a byte only when `a` has odd length, in
+        // which case its partial checksum is byte-swapped.
+        let checksum = !checksum::combine(&[
+            checksum::pseudo_header(
+                &src_ip_addr,
+                &dst_ip_addr,
+                IpProtocol::Tcp,
+                (tcp_header_len + payload_len) as u32,
+            ),
+            checksum::data(&tcp_packet.as_ref()[..tcp_header_len]),
+            checksum::data(a),
+            if a.len() % 2 == 0 {
+                checksum::data(b)
+            } else {
+                checksum::data(b).swap_bytes()
+            },
+        ]);
+        tcp_packet.set_checksum(checksum);
+
+        let header_len = n - payload_len;
+        let buffer = &self.state.buffer;
+        let header = &buffer[..header_len];
+        if b.is_empty() {
+            self.client.recv_segments(&[header, a], &checksum_state);
+        } else {
+            self.client.recv_segments(&[header, a, b], &checksum_state);
+        }
     }
 
     fn rst(&mut self, seq: TcpSeqNumber, ack: Option<TcpSeqNumber>) {
@@ -1848,6 +1889,78 @@ fn trace_tcp_packet(ft: &FourTuple, tcp: &TcpRepr<'_>, payload_len: usize, label
     );
 }
 
+// RFC 1071 primitives vendored from smoltcp (0BSD); replace with a `use` of
+// smoltcp::wire::checksum once smoltcp-rs/smoltcp#1172 exposes it.
+mod checksum {
+    use smoltcp::wire::IpAddress;
+    use smoltcp::wire::IpProtocol;
+
+    const fn propagate_carries(word: u32) -> u16 {
+        let sum = (word >> 16) + (word & 0xffff);
+        ((sum >> 16) as u16) + (sum as u16)
+    }
+
+    pub fn data(mut data: &[u8]) -> u16 {
+        let mut accum: u32 = 0;
+        const CHUNK_SIZE: usize = 32;
+        while data.len() >= CHUNK_SIZE {
+            let mut d = &data[..CHUNK_SIZE];
+            while d.len() >= 2 {
+                accum = accum.wrapping_add(u16::from_be_bytes([d[0], d[1]]) as u32);
+                d = &d[2..];
+            }
+            data = &data[CHUNK_SIZE..];
+        }
+        while data.len() >= 2 {
+            accum = accum.wrapping_add(u16::from_be_bytes([data[0], data[1]]) as u32);
+            data = &data[2..];
+        }
+        if let Some(&value) = data.first() {
+            accum = accum.wrapping_add((value as u32) << 8);
+        }
+        propagate_carries(accum)
+    }
+
+    pub fn combine(checksums: &[u16]) -> u16 {
+        let mut accum: u32 = 0;
+        for &word in checksums {
+            accum = accum.wrapping_add(word as u32);
+        }
+        propagate_carries(accum)
+    }
+
+    pub fn pseudo_header(
+        src_addr: &IpAddress,
+        dst_addr: &IpAddress,
+        next_header: IpProtocol,
+        length: u32,
+    ) -> u16 {
+        match (src_addr, dst_addr) {
+            (IpAddress::Ipv4(src_addr), IpAddress::Ipv4(dst_addr)) => {
+                let mut proto_len = [0u8; 4];
+                proto_len[1] = next_header.into();
+                proto_len[2..4].copy_from_slice(&(length as u16).to_be_bytes());
+                combine(&[
+                    data(&src_addr.octets()),
+                    data(&dst_addr.octets()),
+                    data(&proto_len),
+                ])
+            }
+            (IpAddress::Ipv6(src_addr), IpAddress::Ipv6(dst_addr)) => {
+                let mut len_proto = [0u8; 8];
+                len_proto[0..4].copy_from_slice(&length.to_be_bytes());
+                len_proto[7] = next_header.into();
+                combine(&[
+                    data(&src_addr.octets()),
+                    data(&dst_addr.octets()),
+                    data(&len_proto),
+                ])
+            }
+            _ => unreachable!("mismatched IP address families"),
+        }
+    }
+}
+
 fn take_socket_error(socket: &PolledSocket<Socket>) -> io::Error {
     match socket.get().take_error() {
         Ok(Some(err)) => err,

vm/devices/net/net_consomme/consomme/src/tcp/ring.rs

@@ -165,16 +165,6 @@ impl<'a> View<'a> {
             (a, b)
         }
     }
-
-    /// Copies the view contents into `buf`.
-    ///
-    /// # Panics
-    /// Panics if `buf` is smaller than the view length.
-    pub fn copy_to_slice(&self, buf: &mut [u8]) {
-        let (a, b) = self.as_slices();
-        buf[..a.len()].copy_from_slice(a);
-        buf[a.len()..a.len() + b.len()].copy_from_slice(b);
-    }
 }
 
 #[cfg(test)]

vm/devices/net/net_consomme/consomme/src/tcp/tests.rs

@@ -1878,3 +1878,75 @@ async fn test_tcp_zero_window_reopen_sends_update(driver: DefaultDriver) {
          host drains the backlog; got {window_update:?}"
     );
 }
+
+/// Verifies that the zero-copy TCP checksum computed over the header and the
+/// two payload slices — combined with the pseudo-header — matches smoltcp's
+/// contiguous checksum, for every payload split point (which exercises the
+/// odd-length `a` boundary that byte-swaps `b`).
+#[test]
+fn tcp_checksum_matches_smoltcp() {
+    use smoltcp::wire::IpAddress;
+    use smoltcp::wire::IpProtocol;
+    use smoltcp::wire::Ipv6Address;
+    use smoltcp::wire::TcpControl;
+    use smoltcp::wire::TcpPacket;
+    use smoltcp::wire::TcpRepr;
+    use smoltcp::wire::TcpSeqNumber;
+
+    let v4 = (IpAddress::v4(93, 184, 216, 34), IpAddress::v4(10, 0, 0, 2));
+    let v6 = (
+        IpAddress::Ipv6(Ipv6Address::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 0x1)),
+        IpAddress::Ipv6(Ipv6Address::new(0xfe80, 0, 0, 0, 0, 0, 0, 0x2)),
+    );
+
+    for (src, dst) in [v4, v6] {
+        for payload_len in [0usize, 1, 2, 3, 4, 5, 15, 16, 17, 63, 1460] {
+            let payload: Vec<u8> = (0..payload_len).map(|i| (i * 7 + 3) as u8).collect();
+            let repr = TcpRepr {
+                src_port: 443,
+                dst_port: 51000,
+                control: TcpControl::None,
+                seq_number: TcpSeqNumber(0x1234_5678),
+                ack_number: Some(TcpSeqNumber(0x7654_4321)),
+                window_len: 65535,
+                window_scale: None,
+                max_seg_size: None,
+                sack_permitted: false,
+                sack_ranges: [None, None, None],
+                timestamp: None,
+                payload: &payload,
+            };
+            let mut buf = vec![0u8; repr.buffer_len()];
+            let mut pkt = TcpPacket::new_unchecked(&mut buf);
+            repr.emit(&mut pkt, &src, &dst, &Default::default());
+            pkt.set_checksum(0);
+            pkt.fill_checksum(&src, &dst);
+            let reference = pkt.checksum();
+
+            let header_len = pkt.header_len() as usize;
+            pkt.set_checksum(0);
+            for split in 0..=payload_len {
+                let (a, b) = payload.split_at(split);
+                let got = !checksum::combine(&[
+                    checksum::pseudo_header(
+                        &src,
+                        &dst,
+                        IpProtocol::Tcp,
+                        (header_len + payload_len) as u32,
+                    ),
+                    checksum::data(&pkt.as_ref()[..header_len]),
+                    checksum::data(a),
+                    if a.len() % 2 == 0 {
+                        checksum::data(b)
+                    } else {
+                        checksum::data(b).swap_bytes()
+                    },
+                ]);
+                assert_eq!(
+                    got, reference,
+                    "checksum mismatch: payload_len={payload_len} split={split}"
+                );
+            }
+        }
+    }
+}

vm/devices/net/net_consomme/src/lib.rs

@@ -799,6 +799,10 @@ impl consomme::Client for Client<'_> {
     }
 
     fn recv(&mut self, data: &[u8], checksum: &ChecksumState) {
+        self.recv_segments(&[data], checksum);
+    }
+
+    fn recv_segments(&mut self, segments: &[&[u8]], checksum: &ChecksumState) {
         let Some(rx_id) = self.state.rx_avail.pop_front() else {
             // This should be rare, only affecting unbuffered protocols. TCP and
             // UDP are buffered and they won't indicate packets unless rx_mtu()
@@ -807,12 +811,13 @@ impl consomme::Client for Client<'_> {
             return;
         };
         let max = self.pool.capacity(rx_id) as usize;
-        if data.len() <= max {
-            self.pool.write_packet(
+        let len: usize = segments.iter().map(|s| s.len()).sum();
+        if len <= max {
+            self.pool.write_packet_segments(
                 rx_id,
                 &RxMetadata {
                     offset: 0,
-                    len: data.len(),
+                    len,
                     ip_checksum: if checksum.ipv4 {
                         RxChecksumState::Good
                     } else {
@@ -832,11 +837,11 @@ impl consomme::Client for Client<'_> {
                     },
                     vlan: None,
                 },
-                data,
+                segments,
             );
             self.state.rx_ready.push_back(rx_id);
         } else {
-            tracing::warn!(len = data.len(), max, "dropping rx packet: too large");
+            tracing::warn!(len, max, "dropping rx packet: too large");
             self.state.rx_avail.push_front(rx_id);
         }
     }

vm/devices/net/netvsp/src/buffers.rs

@@ -191,6 +191,15 @@ impl BufferAccess for BufferPool {
         self.buffers.write_at(self.offset(id) + RX_HEADER_LEN, data);
     }
 
+    fn write_packet_segments(&mut self, id: RxId, metadata: &RxMetadata, segments: &[&[u8]]) {
+        let mut offset = self.offset(id) + RX_HEADER_LEN;
+        for segment in segments {
+            self.buffers.write_at(offset, segment);
+            offset += segment.len() as u32;
+        }
+        self.write_header(id, metadata);
+    }
+
     fn write_header(&mut self, id: RxId, metadata: &RxMetadata) {
         #[repr(C)]
         #[derive(zerocopy::IntoBytes, Immutable, KnownLayout, Debug)]