diff --git a/vm/devices/net/net_backend/src/lib.rs b/vm/devices/net/net_backend/src/lib.rs index d3f1502b55..f99e7cffad 100644 --- a/vm/devices/net/net_backend/src/lib.rs +++ b/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; diff --git a/vm/devices/net/net_consomme/consomme/src/lib.rs b/vm/devices/net/net_consomme/consomme/src/lib.rs index bf501dd39b..eae26156bf 100644 --- a/vm/devices/net/net_consomme/consomme/src/lib.rs +++ b/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 diff --git a/vm/devices/net/net_consomme/consomme/src/tcp.rs b/vm/devices/net/net_consomme/consomme/src/tcp.rs index 67265a5d93..76a664baef 100644 --- a/vm/devices/net/net_consomme/consomme/src/tcp.rs +++ b/vm/devices/net/net_consomme/consomme/src/tcp.rs @@ -675,6 +675,7 @@ struct Sender<'a, T> { impl Sender<'_, T> { fn send_packet(&mut self, tcp: &TcpRepr<'_>, payload: Option>) { + 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 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,7 +715,7 @@ impl 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, @@ -722,18 +723,58 @@ impl Sender<'_, T> { &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) { @@ -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) -> io::Error { match socket.get().take_error() { Ok(Some(err)) => err, diff --git a/vm/devices/net/net_consomme/consomme/src/tcp/ring.rs b/vm/devices/net/net_consomme/consomme/src/tcp/ring.rs index 3a2446d536..141daafde5 100644 --- a/vm/devices/net/net_consomme/consomme/src/tcp/ring.rs +++ b/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)] diff --git a/vm/devices/net/net_consomme/consomme/src/tcp/tests.rs b/vm/devices/net/net_consomme/consomme/src/tcp/tests.rs index b45e990d51..fb76e1acb6 100644 --- a/vm/devices/net/net_consomme/consomme/src/tcp/tests.rs +++ b/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 = (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}" + ); + } + } + } +} diff --git a/vm/devices/net/net_consomme/src/lib.rs b/vm/devices/net/net_consomme/src/lib.rs index 2899f3189d..ea7d37b74b 100644 --- a/vm/devices/net/net_consomme/src/lib.rs +++ b/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); } } diff --git a/vm/devices/net/netvsp/src/buffers.rs b/vm/devices/net/netvsp/src/buffers.rs index e5a7bf4e11..c29ecdb69c 100644 --- a/vm/devices/net/netvsp/src/buffers.rs +++ b/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)] diff --git a/vm/devices/virtio/virtio_net/src/buffers.rs b/vm/devices/virtio/virtio_net/src/buffers.rs index 56e475461d..bead7fab9c 100644 --- a/vm/devices/virtio/virtio_net/src/buffers.rs +++ b/vm/devices/virtio/virtio_net/src/buffers.rs @@ -135,6 +135,24 @@ impl BufferAccess for VirtioWorkPool { } } + fn write_packet_segments(&mut self, id: RxId, metadata: &RxMetadata, segments: &[&[u8]]) { + let packet = self.rx_packets[id.0 as usize] + .as_mut() + .expect("invalid buffer index"); + let mut offset = header_size() as u64; + for segment in segments { + if let Err(err) = packet.work.write_at_offset(offset, &self.mem, segment) { + tracelimit::warn_ratelimited!( + len = segment.len(), + error = &err as &dyn std::error::Error, + "rx memory write failure" + ); + } + offset += segment.len() as u64; + } + self.write_header(id, metadata); + } + fn push_guest_addresses(&self, id: RxId, buf: &mut Vec) { let packet = self.rx_packets[id.0 as usize] .as_ref()