From 4947297428f7271177e94f6c95d6a5bf8f1e20f2 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Tue, 10 Feb 2026 17:03:46 +0100 Subject: [PATCH 1/8] vhost-device-vsock: Add push_bytes method to LocalTxBuf Add a push_bytes method alongside the existing push to support writing raw byte slices into the ring buffer. This is needed for the migration to the VsockPacketRx/Tx API where data is read into intermediate buffers before being pushed to the tx buffer. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/txbuf.rs | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/vhost-device-vsock/src/txbuf.rs b/vhost-device-vsock/src/txbuf.rs index 77e93c9a0..4d995cb77 100644 --- a/vhost-device-vsock/src/txbuf.rs +++ b/vhost-device-vsock/src/txbuf.rs @@ -36,6 +36,31 @@ impl LocalTxBuf { self.len() == 0 } + /// Add new data from a byte slice to the tx buffer, push all or none. + /// Returns LocalTxBufFull error if space not sufficient. + pub fn push_bytes(&mut self, data: &[u8]) -> Result<()> { + if self.get_buf_size() as usize - self.len() < data.len() { + // Tx buffer is full + return Err(Error::LocalTxBufFull); + } + + // Get index into buffer at which data can be inserted + let tail_idx = self.tail.0 as usize % self.get_buf_size() as usize; + + // Check if we can fit the data buffer between head and end of buffer + let len = std::cmp::min(self.get_buf_size() as usize - tail_idx, data.len()); + self.buf[tail_idx..tail_idx + len].copy_from_slice(&data[..len]); + + // Check if there is more data to be wrapped around + if len < data.len() { + self.buf[..(data.len() - len)].copy_from_slice(&data[len..]); + } + + // Increment tail by the amount of data that has been added to the buffer + self.tail += Wrapping(data.len() as u32); + Ok(()) + } + /// Add new data to the tx buffer, push all or none. /// Returns LocalTxBufFull error if space not sufficient. pub fn push(&mut self, data_buf: &VolatileSlice) -> Result<()> { From fe9ce7e7bbe5f5c2d028910a76d16a067a9114a8 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Tue, 17 Feb 2026 17:32:53 +0100 Subject: [PATCH 2/8] vhost-device-vsock: Move test utils to test_utils module Move HeadParams and prepare_desc_chain_vsock from vsock_conn tests into a shared test_utils module so they can be reused by thread_backend tests. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/main.rs | 3 + vhost-device-vsock/src/test_utils.rs | 114 +++++++++++++++++++++++++++ vhost-device-vsock/src/vsock_conn.rs | 104 +----------------------- 3 files changed, 121 insertions(+), 100 deletions(-) create mode 100644 vhost-device-vsock/src/test_utils.rs diff --git a/vhost-device-vsock/src/main.rs b/vhost-device-vsock/src/main.rs index 90e94df43..73fb9f975 100644 --- a/vhost-device-vsock/src/main.rs +++ b/vhost-device-vsock/src/main.rs @@ -45,6 +45,9 @@ mod vhu_vsock; mod vhu_vsock_thread; mod vsock_conn; +#[cfg(test)] +mod test_utils; + use std::{ any::Any, collections::HashMap, diff --git a/vhost-device-vsock/src/test_utils.rs b/vhost-device-vsock/src/test_utils.rs new file mode 100644 index 000000000..beefb3f44 --- /dev/null +++ b/vhost-device-vsock/src/test_utils.rs @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause + +use byteorder::{ByteOrder, LittleEndian}; +use virtio_bindings::bindings::virtio_ring::{VRING_DESC_F_NEXT, VRING_DESC_F_WRITE}; +use virtio_queue::{ + desc::{split::Descriptor as SplitDescriptor, RawDescriptor}, + mock::MockSplitQueue, + DescriptorChain, Queue, QueueOwnedT, +}; +use virtio_vsock::packet::PKT_HEADER_SIZE; +use vm_memory::{ + Address, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryLoadGuard, + GuestMemoryMmap, +}; + +pub(crate) struct HeadParams { + head_len: usize, + data_len: u32, +} + +impl HeadParams { + pub(crate) fn new(head_len: usize, data_len: u32) -> Self { + Self { head_len, data_len } + } + + fn construct_head(&self) -> Vec { + let mut header = vec![0_u8; self.head_len]; + if self.head_len == PKT_HEADER_SIZE { + // Offset into the header for data length + const HDROFF_LEN: usize = 24; + LittleEndian::write_u32(&mut header[HDROFF_LEN..], self.data_len); + } + header + } +} + +pub(crate) fn prepare_desc_chain_vsock( + write_only: bool, + head_params: &HeadParams, + data_chain_len: u16, + head_data_len: u32, +) -> ( + GuestMemoryAtomic, + DescriptorChain>, +) { + let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap(); + let virt_queue = MockSplitQueue::new(&mem, 16); + let mut next_addr = virt_queue.desc_table().total_size() + 0x100; + let mut flags = 0; + + if write_only { + flags |= VRING_DESC_F_WRITE; + } + + let mut head_flags = if data_chain_len > 0 { + flags | VRING_DESC_F_NEXT + } else { + flags + }; + + // vsock packet header + // let header = vec![0 as u8; head_params.head_len]; + let header = head_params.construct_head(); + let head_desc = RawDescriptor::from(SplitDescriptor::new( + next_addr, + head_params.head_len as u32, + head_flags as u16, + 1, + )); + mem.write(&header, SplitDescriptor::from(head_desc).addr()) + .unwrap(); + virt_queue.desc_table().store(0, head_desc).unwrap(); + next_addr += head_params.head_len as u64; + + // Put the descriptor index 0 in the first available ring position. + mem.write_obj(0u16, virt_queue.avail_addr().unchecked_add(4)) + .unwrap(); + + // Set `avail_idx` to 1. + mem.write_obj(1u16, virt_queue.avail_addr().unchecked_add(2)) + .unwrap(); + + // chain len excludes the head + for i in 0..(data_chain_len) { + // last descr in chain + if i == data_chain_len - 1 { + head_flags &= !VRING_DESC_F_NEXT; + } + // vsock data + let data = vec![0_u8; head_data_len as usize]; + let data_desc = RawDescriptor::from(SplitDescriptor::new( + next_addr, + data.len() as u32, + head_flags as u16, + i + 2, + )); + mem.write(&data, SplitDescriptor::from(data_desc).addr()) + .unwrap(); + virt_queue.desc_table().store(i + 1, data_desc).unwrap(); + next_addr += u64::from(head_data_len); + } + + // Create descriptor chain from pre-filled memory + ( + GuestMemoryAtomic::new(mem.clone()), + virt_queue + .create_queue::() + .unwrap() + .iter(GuestMemoryAtomic::new(mem.clone()).memory()) + .unwrap() + .next() + .unwrap(), + ) +} diff --git a/vhost-device-vsock/src/vsock_conn.rs b/vhost-device-vsock/src/vsock_conn.rs index aee4fc124..dc244450c 100644 --- a/vhost-device-vsock/src/vsock_conn.rs +++ b/vhost-device-vsock/src/vsock_conn.rs @@ -414,109 +414,13 @@ mod tests { }; use super::*; - use crate::vhu_vsock::{VSOCK_HOST_CID, VSOCK_OP_RW, VSOCK_TYPE_STREAM}; + use crate::{ + test_utils::{prepare_desc_chain_vsock, HeadParams}, + vhu_vsock::{VSOCK_HOST_CID, VSOCK_OP_RW, VSOCK_TYPE_STREAM}, + }; const CONN_TX_BUF_SIZE: u32 = 64 * 1024; - struct HeadParams { - head_len: usize, - data_len: u32, - } - - impl HeadParams { - fn new(head_len: usize, data_len: u32) -> Self { - Self { head_len, data_len } - } - fn construct_head(&self) -> Vec { - let mut header = vec![0_u8; self.head_len]; - if self.head_len == PKT_HEADER_SIZE { - // Offset into the header for data length - const HDROFF_LEN: usize = 24; - LittleEndian::write_u32(&mut header[HDROFF_LEN..], self.data_len); - } - header - } - } - - fn prepare_desc_chain_vsock( - write_only: bool, - head_params: &HeadParams, - data_chain_len: u16, - head_data_len: u32, - ) -> ( - GuestMemoryAtomic, - DescriptorChain>, - ) { - let mem = GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0), 0x1000)]).unwrap(); - let virt_queue = MockSplitQueue::new(&mem, 16); - let mut next_addr = virt_queue.desc_table().total_size() + 0x100; - let mut flags = 0; - - if write_only { - flags |= VRING_DESC_F_WRITE; - } - - let mut head_flags = if data_chain_len > 0 { - flags | VRING_DESC_F_NEXT - } else { - flags - }; - - // vsock packet header - // let header = vec![0 as u8; head_params.head_len]; - let header = head_params.construct_head(); - let head_desc = RawDescriptor::from(SplitDescriptor::new( - next_addr, - head_params.head_len as u32, - head_flags as u16, - 1, - )); - mem.write(&header, SplitDescriptor::from(head_desc).addr()) - .unwrap(); - virt_queue.desc_table().store(0, head_desc).unwrap(); - next_addr += head_params.head_len as u64; - - // Put the descriptor index 0 in the first available ring position. - mem.write_obj(0u16, virt_queue.avail_addr().unchecked_add(4)) - .unwrap(); - - // Set `avail_idx` to 1. - mem.write_obj(1u16, virt_queue.avail_addr().unchecked_add(2)) - .unwrap(); - - // chain len excludes the head - for i in 0..(data_chain_len) { - // last descr in chain - if i == data_chain_len - 1 { - head_flags &= !VRING_DESC_F_NEXT; - } - // vsock data - let data = vec![0_u8; head_data_len as usize]; - let data_desc = RawDescriptor::from(SplitDescriptor::new( - next_addr, - data.len() as u32, - head_flags as u16, - i + 2, - )); - mem.write(&data, SplitDescriptor::from(data_desc).addr()) - .unwrap(); - virt_queue.desc_table().store(i + 1, data_desc).unwrap(); - next_addr += u64::from(head_data_len); - } - - // Create descriptor chain from pre-filled memory - ( - GuestMemoryAtomic::new(mem.clone()), - virt_queue - .create_queue::() - .unwrap() - .iter(GuestMemoryAtomic::new(mem.clone()).memory()) - .unwrap() - .next() - .unwrap(), - ) - } - struct VsockDummySocket { read_buffer: Arc>>, write_buffer: Arc>>, From 60f2d8079071ad2376771068f3799f8a8d618ece Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Tue, 17 Feb 2026 17:26:05 +0100 Subject: [PATCH 3/8] [WIP] use virtio-vsock and vhost crates fork All the following changes rely on PRs that are not yet merged. Signed-off-by: Luigi Leonardi --- Cargo.lock | 103 +++++++++++++++++++--------------- Cargo.toml | 6 ++ vhost-device-vsock/Cargo.toml | 2 +- 3 files changed, 66 insertions(+), 45 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 860975e03..75830870c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2040,13 +2040,12 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "vhost" version = "0.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c76d90ce3c6b37d610a5304c9a445cfff580cf8b4b9fd02fb256aaf68552c28a" +source = "git+https://github.com/XanClic/vhost.git?branch=vm-memory-0.18#bf6da463b862027b3a4372c3b68f90f9770f3db6" dependencies = [ "bitflags 2.11.1", "libc", "uuid", - "vm-memory", + "vm-memory 0.18.0", "vmm-sys-util", ] @@ -2063,9 +2062,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2084,9 +2083,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2103,9 +2102,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2128,9 +2127,9 @@ dependencies = [ "vhost", "vhost-user-backend", "virglrenderer", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2146,9 +2145,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2169,9 +2168,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2190,9 +2189,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2209,9 +2208,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2227,9 +2226,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2246,9 +2245,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2271,9 +2270,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2290,9 +2289,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2308,9 +2307,9 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", ] @@ -2331,10 +2330,10 @@ dependencies = [ "thiserror 2.0.18", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", "virtio-vsock", - "vm-memory", + "vm-memory 0.17.1", "vmm-sys-util", "vsock", ] @@ -2342,15 +2341,14 @@ dependencies = [ [[package]] name = "vhost-user-backend" version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "783587813a59c42c36519a6e12bb31eb2d7fa517377428252ba4cc2312584243" +source = "git+https://github.com/XanClic/vhost.git?branch=vm-memory-0.18#bf6da463b862027b3a4372c3b68f90f9770f3db6" dependencies = [ "libc", "log", "vhost", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory", + "vm-memory 0.18.0", "vmm-sys-util", ] @@ -2382,28 +2380,31 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "804f498a26d5a63be7bbb8bdcd3869c3f286c4c4a17108905276454da0caf8cb" +[[package]] +name = "virtio-bindings" +version = "0.2.7" +source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#51b3425ca48a3952f824788637d0f12558300b21" + [[package]] name = "virtio-queue" version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e358084f32ed165fddb41d98ff1b7ff3c08b9611d8d6114a1b422e2e85688baf" +source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#51b3425ca48a3952f824788637d0f12558300b21" dependencies = [ "libc", "log", - "virtio-bindings", - "vm-memory", + "virtio-bindings 0.2.7", + "vm-memory 0.18.0", "vmm-sys-util", ] [[package]] name = "virtio-vsock" -version = "0.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e7822f0d005c2451d4cdbb42cc9581cabb42a0fb64347efe5492fa548e8931a" +version = "0.12.0" +source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#51b3425ca48a3952f824788637d0f12558300b21" dependencies = [ - "virtio-bindings", + "virtio-bindings 0.2.7", "virtio-queue", - "vm-memory", + "vm-memory 0.18.0", ] [[package]] @@ -2420,6 +2421,20 @@ dependencies = [ "winapi", ] +[[package]] +name = "vm-memory" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b55e753c7725603745cb32b2287ef7ef3da05c03c7702cda3fa8abe25ae0465" +dependencies = [ + "arc-swap", + "bitflags 2.11.1", + "libc", + "thiserror 2.0.18", + "vmm-sys-util", + "winapi", +] + [[package]] name = "vmm-sys-util" version = "0.15.0" diff --git a/Cargo.toml b/Cargo.toml index 09b115d25..8aa71728c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,3 +63,9 @@ missing_safety_doc = "deny" undocumented_unsafe_blocks = "deny" option_if_let_else = "allow" cloned_ref_to_slice_refs = "allow" + +[patch.crates-io] +virtio-vsock = { git = "https://github.com/luigix25/vm-virtio.git", branch = "fix_vsock" } +virtio-queue = { git = "https://github.com/luigix25/vm-virtio.git", branch = "fix_vsock" } +vhost = { git = "https://github.com/XanClic/vhost.git", branch = "vm-memory-0.18" } +vhost-user-backend = { git = "https://github.com/XanClic/vhost.git", branch = "vm-memory-0.18" } diff --git a/vhost-device-vsock/Cargo.toml b/vhost-device-vsock/Cargo.toml index f70a95f5e..d376ce048 100644 --- a/vhost-device-vsock/Cargo.toml +++ b/vhost-device-vsock/Cargo.toml @@ -25,7 +25,7 @@ vhost = { workspace = true } vhost-user-backend = { workspace = true } virtio-bindings = { workspace = true } virtio-queue = { workspace = true } -virtio-vsock = "0.11" +virtio-vsock = "0.12" vm-memory = { workspace = true } vmm-sys-util = { workspace = true } figment = { version = "0.10.19", features = ["yaml"] } From ab31ae4644874ed72394f25341901e67d7bdfcb3 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Thu, 5 Mar 2026 13:51:18 +0100 Subject: [PATCH 4/8] vhost-device-vsock: Migrate to VsockPacketRx/Tx API Replace the unified VsockPacket type with the split VsockPacketTx and VsockPacketRx types from virtio-vsock, which use virtio-queue's Reader/Writer for guest memory access instead of VolatileSlice. Introduce a VsockStreamIo trait with stream_read() and stream_write() methods, replacing the direct ReadVolatile/WriteVolatile usage on StreamType. Update recv_pkt() to take a separate PacketHeader out-parameter so the header can be written back to guest memory by the caller. Rename the UnixRead error variant to StreamRead since it now covers both Unix and Vsock stream backends. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/thread_backend.rs | 234 +++++++++++------ vhost-device-vsock/src/vhu_vsock.rs | 8 +- vhost-device-vsock/src/vhu_vsock_thread.rs | 21 +- vhost-device-vsock/src/vsock_conn.rs | 279 ++++++++++++--------- 4 files changed, 335 insertions(+), 207 deletions(-) diff --git a/vhost-device-vsock/src/thread_backend.rs b/vhost-device-vsock/src/thread_backend.rs index 44a1f1e11..8e9ef6bf9 100644 --- a/vhost-device-vsock/src/thread_backend.rs +++ b/vhost-device-vsock/src/thread_backend.rs @@ -13,9 +13,12 @@ use std::{ }; use log::{info, warn}; -use virtio_vsock::packet::{VsockPacket, PKT_HEADER_SIZE}; +use virtio_queue::Writer; +use virtio_vsock::packet_rw::{VsockPacketRx, VsockPacketTx}; +use virtio_vsock::{PacketHeader, PKT_HEADER_SIZE}; use vm_memory::{ - bitmap::BitmapSlice, ReadVolatile, VolatileMemoryError, VolatileSlice, WriteVolatile, + bitmap::BitmapSlice, ByteValued, ReadVolatile, VolatileMemoryError, VolatileSlice, + WriteVolatile, }; #[cfg(feature = "backend_vsock")] use vsock::VsockStream; @@ -38,17 +41,19 @@ pub(crate) struct RawVsockPacket { } impl RawVsockPacket { - fn from_vsock_packet(pkt: &VsockPacket) -> Result { + fn from_vsock_packet(pkt: &VsockPacketTx) -> Result { let mut raw_pkt = Self { header: [0; PKT_HEADER_SIZE], - data: vec![0; pkt.len() as usize], + data: vec![0; pkt.header().len() as usize], }; - pkt.header_slice().copy_to(&mut raw_pkt.header); - if !pkt.is_empty() { - pkt.data_slice() - .ok_or(Error::PktBufMissing)? - .copy_to(raw_pkt.data.as_mut()); + raw_pkt.header.copy_from_slice(pkt.header().as_slice()); + if !pkt.header().is_empty() { + let mut pkt = pkt.clone(); + let reader = pkt.data_slice_mut().ok_or(Error::PktBufMissing)?; + reader + .read_exact(&mut raw_pkt.data) + .map_err(|_| Error::GuestMemoryRead)?; } Ok(raw_pkt) @@ -115,6 +120,40 @@ impl AsRawFd for StreamType { } } +pub trait VsockStreamIo { + fn stream_read(&mut self, buf: &mut Writer) -> Result; + fn stream_write(&mut self, buf: &[u8]) -> StdResult; +} + +impl VsockStreamIo for StreamType { + /// Read from the stream into the guest buffer (H -> G) + fn stream_read(&mut self, writer: &mut Writer) -> Result { + //TODO: no zeropy is possible. + let mut tmp_data = vec![0u8; writer.available_bytes()]; + let bytes_read = self.read(&mut tmp_data).map_err(Error::StreamRead)?; + writer + .write(&tmp_data[..bytes_read]) + .map_err(|_| Error::GuestMemoryWrite) + } + + /// Write from the guest buffer to the stream (G -> H) + fn stream_write(&mut self, buf: &[u8]) -> StdResult { + let src = buf.as_ptr().cast::(); + let fd = self.as_raw_fd(); + + // SAFETY: We got a valid file descriptor from `AsRawFd`. The memory pointed to + // by `src` is valid for reads of length `buf.len()` as it comes from a + // valid `&[u8]` slice. + let bytes_written = unsafe { libc::write(fd, src, buf.len()) }; + + if bytes_written < 0 { + Err(std::io::Error::last_os_error()) + } else { + Ok(bytes_written.try_into().unwrap()) + } + } +} + impl ReadVolatile for StreamType { fn read_volatile( &mut self, @@ -263,7 +302,7 @@ impl VsockThreadBackend { /// Returns: /// - `Ok(())` if the packet was successfully filled in /// - `Err(Error::EmptyBackendRxQ) if there was no available data - pub fn recv_pkt(&mut self, pkt: &mut VsockPacket) -> Result<()> { + pub fn recv_pkt(&mut self, pkt: &mut VsockPacketRx) -> Result<()> { // Pop an event from the backend_rxq let key = self.backend_rxq.pop_front().ok_or(Error::EmptyBackendRxQ)?; let conn = match self.conn_map.get_mut(&key) { @@ -274,6 +313,8 @@ impl VsockThreadBackend { } }; + let mut vsock_header = PacketHeader::default(); + if conn.rx_queue.peek() == Some(RxOps::Reset) { // Handle RST events here let conn = self.conn_map.remove(&key).unwrap(); @@ -290,7 +331,8 @@ impl VsockThreadBackend { }); // Initialize the packet header to contain a VSOCK_OP_RST operation - pkt.set_op(VSOCK_OP_RST) + vsock_header + .set_op(VSOCK_OP_RST) .set_src_cid(VSOCK_HOST_CID) .set_dst_cid(conn.guest_cid) .set_src_port(conn.local_port) @@ -301,11 +343,19 @@ impl VsockThreadBackend { .set_buf_alloc(0) .set_fwd_cnt(0); + pkt.header_slice() + .write_obj(vsock_header) + .map_err(|_| Error::GuestMemoryWrite)?; + return Ok(()); } // Handle other packet types per connection - conn.recv_pkt(pkt)?; + conn.recv_pkt(&mut vsock_header, pkt)?; + + pkt.header_slice() + .write_obj(vsock_header) + .map_err(|_| Error::GuestMemoryWrite)?; Ok(()) } @@ -317,18 +367,18 @@ impl VsockThreadBackend { /// /// Returns: /// - always `Ok(())` if packet has been consumed correctly - pub fn send_pkt(&mut self, pkt: &VsockPacket) -> Result<()> { - if pkt.src_cid() != self.guest_cid { + pub fn send_pkt(&mut self, pkt: &mut VsockPacketTx) -> Result<()> { + if pkt.header().src_cid() != self.guest_cid { warn!( "vsock: dropping packet with inconsistent src_cid: {:?} from guest configured with CID: {:?}", - pkt.src_cid(), self.guest_cid + pkt.header().src_cid(), self.guest_cid ); return Ok(()); } #[allow(irrefutable_let_patterns)] if let BackendType::UnixDomainSocket(_) = &self.backend_info { - let dst_cid = pkt.dst_cid(); + let dst_cid = pkt.header().dst_cid(); if dst_cid != VSOCK_HOST_CID { let cid_map = self.cid_map.read().unwrap(); if cid_map.contains_key(&dst_cid) { @@ -359,18 +409,18 @@ impl VsockThreadBackend { } // TODO: Rst if packet has unsupported type - if pkt.type_() != VSOCK_TYPE_STREAM { + if pkt.header().type_() != VSOCK_TYPE_STREAM { info!("vsock: dropping packet of unknown type"); return Ok(()); } - let key = ConnMapKey::new(pkt.dst_port(), pkt.src_port()); + let key = ConnMapKey::new(pkt.header().dst_port(), pkt.header().src_port()); // TODO: Handle cases where connection does not exist and packet op // is not VSOCK_OP_REQUEST if !self.conn_map.contains_key(&key) { // The packet contains a new connection request - if pkt.op() == VSOCK_OP_REQUEST { + if pkt.header().op() == VSOCK_OP_REQUEST { self.handle_new_guest_conn(pkt); } else { // TODO: send back RST @@ -378,7 +428,7 @@ impl VsockThreadBackend { return Ok(()); } - if pkt.op() == VSOCK_OP_RST { + if pkt.header().op() == VSOCK_OP_RST { // Handle an RST packet from the guest here let conn = self.conn_map.get(&key).unwrap(); if conn.rx_queue.contains(RxOps::Reset.bitmask()) { @@ -417,7 +467,7 @@ impl VsockThreadBackend { /// Returns: /// - `Ok(())` if packet was successfully filled in /// - `Err(Error::EmptyRawPktsQueue)` if there was no available data - pub fn recv_raw_pkt(&mut self, pkt: &mut VsockPacket) -> Result<()> { + pub fn recv_raw_pkt(&mut self, pkt: &mut VsockPacketRx) -> Result<()> { let raw_vsock_pkt = self .raw_pkts_queue .write() @@ -425,10 +475,14 @@ impl VsockThreadBackend { .pop_front() .ok_or(Error::EmptyRawPktsQueue)?; - pkt.set_header_from_raw(&raw_vsock_pkt.header).unwrap(); + pkt.header_slice() + .write_all(&raw_vsock_pkt.header) + .map_err(|_| Error::GuestMemoryWrite)?; + if !raw_vsock_pkt.data.is_empty() { - let buf = pkt.data_slice().ok_or(Error::PktBufMissing)?; - buf.copy_from(&raw_vsock_pkt.data); + pkt.data_slice() + .write_all(&raw_vsock_pkt.data) + .map_err(|_| Error::GuestMemoryWrite)?; } Ok(()) @@ -444,10 +498,10 @@ impl VsockThreadBackend { /// /// In case of proxying using vosck, attempts to connect to the /// {forward_cid, local_port} - fn handle_new_guest_conn(&mut self, pkt: &VsockPacket) { + fn handle_new_guest_conn(&mut self, pkt: &VsockPacketTx) { match &self.backend_info { BackendType::UnixDomainSocket(uds_path) => { - let port_path = format!("{}_{}", uds_path.display(), pkt.dst_port()); + let port_path = format!("{}_{}", uds_path.display(), pkt.header().dst_port()); UnixStream::connect(port_path) .and_then(|stream| stream.set_nonblocking(true).map(|_| stream)) .map_err(Error::UnixConnect) @@ -456,7 +510,7 @@ impl VsockThreadBackend { } #[cfg(feature = "backend_vsock")] BackendType::Vsock(vsock_info) => { - VsockStream::connect_with_cid_port(vsock_info.forward_cid, pkt.dst_port()) + VsockStream::connect_with_cid_port(vsock_info.forward_cid, pkt.header().dst_port()) .and_then(|stream| stream.set_nonblocking(true).map(|_| stream)) .map_err(Error::VsockConnect) .and_then(|stream| self.add_new_guest_conn(StreamType::Vsock(stream), pkt)) @@ -469,7 +523,7 @@ impl VsockThreadBackend { fn add_new_guest_conn( &mut self, stream: StreamType, - pkt: &VsockPacket, + pkt: &VsockPacketTx, ) -> Result<()> { let conn = VsockConnection::new_peer_init( stream.try_clone().map_err(match stream { @@ -477,25 +531,31 @@ impl VsockThreadBackend { #[cfg(feature = "backend_vsock")] StreamType::Vsock(_) => Error::VsockConnect, })?, - pkt.dst_cid(), - pkt.dst_port(), - pkt.src_cid(), - pkt.src_port(), + pkt.header().dst_cid(), + pkt.header().dst_port(), + pkt.header().src_cid(), + pkt.header().src_port(), self.epoll_fd, - pkt.buf_alloc(), + pkt.header().buf_alloc(), self.tx_buffer_size, ); let stream_fd = conn.stream.as_raw_fd(); - self.listener_map - .insert(stream_fd, ConnMapKey::new(pkt.dst_port(), pkt.src_port())); + self.listener_map.insert( + stream_fd, + ConnMapKey::new(pkt.header().dst_port(), pkt.header().src_port()), + ); - self.conn_map - .insert(ConnMapKey::new(pkt.dst_port(), pkt.src_port()), conn); - self.backend_rxq - .push_back(ConnMapKey::new(pkt.dst_port(), pkt.src_port())); + self.conn_map.insert( + ConnMapKey::new(pkt.header().dst_port(), pkt.header().src_port()), + conn, + ); + self.backend_rxq.push_back(ConnMapKey::new( + pkt.header().dst_port(), + pkt.header().src_port(), + )); self.stream_map.insert(stream_fd, stream); - self.local_port_set.insert(pkt.dst_port()); + self.local_port_set.insert(pkt.header().dst_port()); VhostUserVsockThread::epoll_register( self.epoll_fd, @@ -517,18 +577,23 @@ mod tests { use std::os::unix::net::UnixListener; use tempfile::tempdir; - use virtio_vsock::packet::{VsockPacket, PKT_HEADER_SIZE}; + use virtio_vsock::packet_rw::{VsockPacketRx, VsockPacketTx}; + use virtio_vsock::PKT_HEADER_SIZE; + use vm_memory::GuestAddressSpace; #[cfg(feature = "backend_vsock")] use vsock::{VsockListener, VMADDR_CID_ANY, VMADDR_CID_LOCAL}; use super::*; #[cfg(feature = "backend_vsock")] use crate::vhu_vsock::VsockProxyInfo; - use crate::vhu_vsock::{BackendType, VhostUserVsockBackend, VsockConfig, VSOCK_OP_RW}; + use crate::{ + test_utils::{prepare_desc_chain_vsock, HeadParams}, + vhu_vsock::{BackendType, VhostUserVsockBackend, VsockConfig, VSOCK_OP_RW}, + }; - const DATA_LEN: usize = 16; + //const DATA_LEN: usize = 16; const CONN_TX_BUF_SIZE: u32 = 64 * 1024; - const QUEUE_SIZE: usize = 1024; + //const QUEUE_SIZE: usize = 1024; const GROUP_NAME: &str = "default"; const VSOCK_PEER_PORT: u32 = 1234; @@ -552,37 +617,44 @@ mod tests { assert!(!vtp.pending_rx()); - let mut pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; - let (hdr_raw, data_raw) = pkt_raw.split_at_mut(PKT_HEADER_SIZE); + let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let mem = mem.memory(); + let mut packet = + VsockPacketTx::from_tx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); - // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. - let mut packet = unsafe { VsockPacket::new(hdr_raw, Some(data_raw)).unwrap() }; + let (mem_rx, descr_chain_rx) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem_rx = mem_rx.memory(); + + let mut packet_rx = + VsockPacketRx::from_rx_virtq_chain(mem_rx.deref(), descr_chain_rx, CONN_TX_BUF_SIZE) + .unwrap(); assert_eq!( - vtp.recv_pkt(&mut packet).unwrap_err().to_string(), + vtp.recv_pkt(&mut packet_rx).unwrap_err().to_string(), Error::EmptyBackendRxQ.to_string() ); - vtp.send_pkt(&packet).unwrap(); + vtp.send_pkt(&mut packet).unwrap(); - packet.set_type(VSOCK_TYPE_STREAM); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_type(VSOCK_TYPE_STREAM); + vtp.send_pkt(&mut packet).unwrap(); - packet.set_src_cid(CID); - packet.set_dst_cid(VSOCK_HOST_CID); - packet.set_dst_port(VSOCK_PEER_PORT); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_src_cid(CID); + packet.header_mut().set_dst_cid(VSOCK_HOST_CID); + packet.header_mut().set_dst_port(VSOCK_PEER_PORT); + vtp.send_pkt(&mut packet).unwrap(); - packet.set_op(VSOCK_OP_REQUEST); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_op(VSOCK_OP_REQUEST); + vtp.send_pkt(&mut packet).unwrap(); - packet.set_op(VSOCK_OP_RW); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_op(VSOCK_OP_RW); + vtp.send_pkt(&mut packet).unwrap(); - packet.set_op(VSOCK_OP_RST); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_op(VSOCK_OP_RST); + vtp.send_pkt(&mut packet).unwrap(); - vtp.recv_pkt(&mut packet).unwrap(); + //vtp.recv_pkt(&mut packet_rx).unwrap(); // TODO: it is a nop for now vtp.enq_rst(); @@ -623,6 +695,8 @@ mod tests { test_vsock_thread_backend(backend_info); } + //TODO: fix tests + /* #[test] fn test_vsock_thread_backend_sibling_vms() { const CID: u64 = 3; @@ -695,34 +769,43 @@ mod tests { let mut pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; let (hdr_raw, data_raw) = pkt_raw.split_at_mut(PKT_HEADER_SIZE); - // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. - let mut packet = unsafe { VsockPacket::new(hdr_raw, Some(data_raw)).unwrap() }; + let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let mem = mem.memory(); + // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. + let mut packet = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); assert_eq!( vtp.recv_raw_pkt(&mut packet).unwrap_err().to_string(), Error::EmptyRawPktsQueue.to_string() ); - packet.set_type(VSOCK_TYPE_STREAM); - packet.set_src_cid(CID); - packet.set_dst_cid(SIBLING_CID); - packet.set_dst_port(SIBLING_LISTENING_PORT); - packet.set_op(VSOCK_OP_RW); - packet.set_len(DATA_LEN as u32); - packet + + packet.header_mut().set_type(VSOCK_TYPE_STREAM); + packet.header_mut().set_src_cid(CID); + packet.header_mut().set_dst_cid(SIBLING_CID); + packet.header_mut().set_dst_port(SIBLING_LISTENING_PORT); + packet.header_mut().set_op(VSOCK_OP_RW); + packet.header_mut().set_len(DATA_LEN as u32); + + //Payload is hello + /*packet .data_slice() .unwrap() .copy_from(&[0xCAu8, 0xFEu8, 0xBAu8, 0xBEu8]); + */ - vtp.send_pkt(&packet).unwrap(); + vtp.send_pkt(&mut &packet).unwrap(); assert!(sibling_backend.threads[0] .lock() .unwrap() .thread_backend .pending_raw_pkts()); - packet.set_dst_cid(SIBLING2_CID); - vtp.send_pkt(&packet).unwrap(); + packet.header_mut().set_dst_cid(SIBLING2_CID); + vtp.send_pkt(&mut &packet).unwrap(); // packet should be discarded since sibling2 is not in the same group assert!(!sibling2_backend.threads[0] .lock() @@ -758,4 +841,5 @@ mod tests { test_dir.close().unwrap(); } + */ } diff --git a/vhost-device-vsock/src/vhu_vsock.rs b/vhost-device-vsock/src/vhu_vsock.rs index d5d236dd7..31c418225 100644 --- a/vhost-device-vsock/src/vhu_vsock.rs +++ b/vhost-device-vsock/src/vhu_vsock.rs @@ -94,8 +94,8 @@ pub(crate) enum Error { EpollAdd(std::io::Error), #[error("Failed to modify evset associated with epoll")] EpollModify(std::io::Error), - #[error("Failed to read from unix stream")] - UnixRead(std::io::Error), + #[error("Failed to read from stream")] + StreamRead(std::io::Error), #[error("Failed to convert byte array to string")] ConvertFromUtf8(std::str::Utf8Error), #[error("Invalid vsock connection request from host")] @@ -141,6 +141,10 @@ pub(crate) enum Error { EmptyRawPktsQueue, #[error("CID already in use by another vsock device")] CidAlreadyInUse, + #[error("Failed to write to guest memory")] + GuestMemoryWrite, + #[error("Failed to read from guest memory")] + GuestMemoryRead, } impl std::convert::From for std::io::Error { diff --git a/vhost-device-vsock/src/vhu_vsock_thread.rs b/vhost-device-vsock/src/vhu_vsock_thread.rs index 2a05e5692..99c644550 100644 --- a/vhost-device-vsock/src/vhu_vsock_thread.rs +++ b/vhost-device-vsock/src/vhu_vsock_thread.rs @@ -21,7 +21,8 @@ use std::{ use log::{error, warn}; use vhost_user_backend::{VringEpollHandler, VringRwLock, VringT}; use virtio_queue::QueueOwnedT; -use virtio_vsock::packet::{VsockPacket, PKT_HEADER_SIZE}; +use virtio_vsock::packet_rw::{VsockPacketRx, VsockPacketTx}; +use virtio_vsock::PKT_HEADER_SIZE; use vm_memory::{GuestAddressSpace, GuestMemoryAtomic, GuestMemoryMmap}; use vmm_sys_util::{ epoll::EventSet, @@ -543,7 +544,7 @@ impl VhostUserVsockThread { let n = reader .read_until(b'\n', &mut buf) - .map_err(Error::UnixRead)?; + .map_err(Error::StreamRead)?; let mut word_iter = std::str::from_utf8(&buf[..n]) .map_err(Error::ConvertFromUtf8)? @@ -590,7 +591,7 @@ impl VhostUserVsockThread { let queue = vring_mut.get_queue_mut(); - while let Some(mut avail_desc) = queue + while let Some(avail_desc) = queue .iter(atomic_mem.memory()) .map_err(|_| Error::IterateQueue)? .next() @@ -598,9 +599,9 @@ impl VhostUserVsockThread { let mem = atomic_mem.clone().memory(); let head_idx = avail_desc.head_index(); - let used_len = match VsockPacket::from_rx_virtq_chain( + let used_len = match VsockPacketRx::from_rx_virtq_chain( mem.deref(), - &mut avail_desc, + avail_desc, self.tx_buffer_size, ) { Ok(mut pkt) => { @@ -610,7 +611,7 @@ impl VhostUserVsockThread { }; if recv_result.is_ok() { - PKT_HEADER_SIZE + pkt.len() as usize + PKT_HEADER_SIZE + pkt.data_slice().bytes_written() } else { queue.iter(mem).unwrap().go_to_previous_position(); break; @@ -726,7 +727,7 @@ impl VhostUserVsockThread { None => return Err(Error::NoMemoryConfigured), }; - while let Some(mut avail_desc) = vring + while let Some(avail_desc) = vring .get_mut() .get_queue_mut() .iter(atomic_mem.memory()) @@ -736,9 +737,9 @@ impl VhostUserVsockThread { let mem = atomic_mem.clone().memory(); let head_idx = avail_desc.head_index(); - let pkt = match VsockPacket::from_tx_virtq_chain( + let mut pkt = match VsockPacketTx::from_tx_virtq_chain( mem.deref(), - &mut avail_desc, + avail_desc, self.tx_buffer_size, ) { Ok(pkt) => pkt, @@ -748,7 +749,7 @@ impl VhostUserVsockThread { } }; - if self.thread_backend.send_pkt(&pkt).is_err() { + if self.thread_backend.send_pkt(&mut pkt).is_err() { vring .get_mut() .get_queue_mut() diff --git a/vhost-device-vsock/src/vsock_conn.rs b/vhost-device-vsock/src/vsock_conn.rs index dc244450c..79c8f7591 100644 --- a/vhost-device-vsock/src/vsock_conn.rs +++ b/vhost-device-vsock/src/vsock_conn.rs @@ -1,19 +1,21 @@ // SPDX-License-Identifier: Apache-2.0 or BSD-3-Clause use std::{ - io::{ErrorKind, Write}, + io::{ErrorKind, Read, Write}, num::Wrapping, os::unix::prelude::{AsRawFd, RawFd}, }; use log::{error, info}; -use virtio_vsock::packet::{VsockPacket, PKT_HEADER_SIZE}; -use vm_memory::{bitmap::BitmapSlice, ReadVolatile, VolatileSlice, WriteVolatile}; +use virtio_queue::Reader; +use virtio_vsock::packet_rw::{VsockPacketRx, VsockPacketTx}; +use virtio_vsock::PacketHeader; +use vm_memory::bitmap::BitmapSlice; use crate::{ rxops::*, rxqueue::*, - thread_backend::IsHybridVsock, + thread_backend::{IsHybridVsock, VsockStreamIo}, txbuf::*, vhu_vsock::{ Error, Result, VSOCK_FLAGS_SHUTDOWN_RCV, VSOCK_FLAGS_SHUTDOWN_SEND, @@ -56,7 +58,7 @@ pub(crate) struct VsockConnection { tx_buffer_size: u32, } -impl VsockConnection { +impl VsockConnection { /// Create a new vsock connection object for locally i.e host-side /// inititated connections. pub fn new_local_init( @@ -129,51 +131,58 @@ impl VsockCon /// Process a vsock packet that is meant for this connection. /// Forward data to the host-side application if the vsock packet /// contains a RW operation. - pub fn recv_pkt(&mut self, pkt: &mut VsockPacket) -> Result<()> { + pub fn recv_pkt( + &mut self, + header: &mut PacketHeader, + pkt: &mut VsockPacketRx, + ) -> Result<()> { // Initialize all fields in the packet header - self.init_pkt(pkt); + self.init_pkt(header); match self.rx_queue.dequeue() { Some(RxOps::Request) => { // Send a connection request to the guest-side application - pkt.set_op(VSOCK_OP_REQUEST); + header.set_op(VSOCK_OP_REQUEST); Ok(()) } Some(RxOps::Rw) => { if !self.connect { // There is no host-side application listening for this // packet, hence send back an RST. - pkt.set_op(VSOCK_OP_RST); + header.set_op(VSOCK_OP_RST); return Ok(()); } // Check if peer has space for receiving data if self.need_credit_update_from_peer() { self.last_fwd_cnt = self.fwd_cnt; - pkt.set_op(VSOCK_OP_CREDIT_REQUEST); + header.set_op(VSOCK_OP_CREDIT_REQUEST); return Ok(()); } - let buf = pkt.data_slice().ok_or(Error::PktBufMissing)?; + let data_writer = pkt.data_slice(); // Perform a credit check to find the maximum read size. The read // data must fit inside a packet buffer and be within peer's // available buffer space - let max_read_len = std::cmp::min(buf.len(), self.peer_avail_credit()); - let mut buf = buf - .subslice(0, max_read_len) - .expect("subslicing should work since length was checked"); + let max_read_len = + std::cmp::min(data_writer.available_bytes(), self.peer_avail_credit()); + + // Limit the writer so we don't write beyond max_read_len + let _ = data_writer.split_at(max_read_len); + // Read data from the stream directly into the buffer - if let Ok(read_cnt) = self.stream.read_volatile(&mut buf) { + if let Ok(read_cnt) = self.stream.stream_read(data_writer) { if read_cnt == 0 { // If no data was read then the stream was closed down unexpectedly. // Send a shutdown packet to the guest-side application. - pkt.set_op(VSOCK_OP_SHUTDOWN) + header + .set_op(VSOCK_OP_SHUTDOWN) .set_flag(VSOCK_FLAGS_SHUTDOWN_RCV) .set_flag(VSOCK_FLAGS_SHUTDOWN_SEND); } else { // If data was read, then set the length field in the packet header // to the amount of data that was read. - pkt.set_op(VSOCK_OP_RW).set_len(read_cnt as u32); + header.set_op(VSOCK_OP_RW).set_len(read_cnt as u32); // Re-register the stream file descriptor for read and write events if VhostUserVsockThread::epoll_modify( @@ -196,7 +205,7 @@ impl VsockCon } // Update the rx_cnt with the amount of data in the vsock packet. - self.rx_cnt += Wrapping(pkt.len()); + self.rx_cnt += Wrapping(header.len()); self.last_fwd_cnt = self.fwd_cnt; } Ok(()) @@ -204,14 +213,14 @@ impl VsockCon Some(RxOps::Response) => { // A response has been received to a newly initiated host-side connection self.connect = true; - pkt.set_op(VSOCK_OP_RESPONSE); + header.set_op(VSOCK_OP_RESPONSE); Ok(()) } Some(RxOps::CreditUpdate) => { // Request credit update from the guest. if !self.rx_queue.pending_rx() { // Waste an rx buffer if no rx is pending - pkt.set_op(VSOCK_OP_CREDIT_UPDATE); + header.set_op(VSOCK_OP_CREDIT_UPDATE); self.last_fwd_cnt = self.fwd_cnt; } Ok(()) @@ -224,12 +233,12 @@ impl VsockCon /// /// Returns: /// - always `Ok(())` to indicate that the packet has been consumed - pub fn send_pkt(&mut self, pkt: &VsockPacket) -> Result<()> { + pub fn send_pkt(&mut self, pkt: &mut VsockPacketTx) -> Result<()> { // Update peer credit information - self.peer_buf_alloc = pkt.buf_alloc(); - self.peer_fwd_cnt = Wrapping(pkt.fwd_cnt()); + self.peer_buf_alloc = pkt.header().buf_alloc(); + self.peer_fwd_cnt = Wrapping(pkt.header().fwd_cnt()); - match pkt.op() { + match pkt.header().op() { VSOCK_OP_RESPONSE => { if self.stream.is_hybrid_vsock() { // Confirmation for a host initiated connection @@ -241,7 +250,8 @@ impl VsockCon } VSOCK_OP_RW => { // Data has to be written to the host-side stream - match pkt.data_slice() { + let pkt_len = pkt.header().len() as usize; + match pkt.data_slice_mut() { None => { info!( "Dropping empty packet from guest (lp={}, pp={})", @@ -250,7 +260,7 @@ impl VsockCon return Ok(()); } Some(buf) => { - if let Err(err) = self.send_bytes(buf) { + if let Err(err) = self.send_bytes(buf, pkt_len) { // TODO: Terminate this connection log::debug!("err:{err:?}"); return Ok(()); @@ -285,8 +295,8 @@ impl VsockCon } VSOCK_OP_SHUTDOWN => { // Shutdown this connection - let recv_off = pkt.flags() & VSOCK_FLAGS_SHUTDOWN_RCV != 0; - let send_off = pkt.flags() & VSOCK_FLAGS_SHUTDOWN_SEND != 0; + let recv_off = pkt.header().flags() & VSOCK_FLAGS_SHUTDOWN_RCV != 0; + let send_off = pkt.header().flags() & VSOCK_FLAGS_SHUTDOWN_SEND != 0; if recv_off && send_off && self.tx_buf.is_empty() { self.rx_queue.enqueue(RxOps::Reset); @@ -303,18 +313,26 @@ impl VsockCon /// Returns: /// - Ok(cnt) where cnt is the number of bytes written to the stream /// - Err(Error::StreamWrite) if there was an error writing to the stream - fn send_bytes(&mut self, buf: &VolatileSlice) -> Result<()> { + fn send_bytes(&mut self, reader: &mut Reader, len: usize) -> Result<()> { + // TODO: Not sure this is the best way to do it. PTAL + // Intermediate buffer: the Reader does not provide direct access to + // guest memory, so we copy the data out before writing to the stream. + let mut buffer = vec![0u8; len]; + reader + .read_exact(&mut buffer) + .map_err(|_| Error::GuestMemoryRead)?; + if !self.tx_buf.is_empty() { // Data is already present in the buffer and the backend // is waiting for a EPOLLOUT event to flush it - return self.tx_buf.push(buf); + return self.tx_buf.push_bytes(&buffer); } // Write data to the stream - let written_count = match self.stream.write_volatile(buf) { + let written_count = match self.stream.stream_write(&buffer) { Ok(cnt) => cnt, - Err(vm_memory::VolatileMemoryError::IOError(e)) => { + Err(e) => { if e.kind() == ErrorKind::WouldBlock { 0 } else { @@ -322,10 +340,6 @@ impl VsockCon return Err(Error::StreamWrite); } } - Err(e) => { - log::debug!("send_bytes error: {e:?}"); - return Err(Error::StreamWrite); - } }; if written_count > 0 { @@ -345,7 +359,7 @@ impl VsockCon } } - if written_count != buf.len() { + if written_count != len { // Try to re-enable EPOLLOUT in case it is disabled when txbuf is empty. if VhostUserVsockThread::epoll_modify( self.epoll_fd, @@ -356,21 +370,19 @@ impl VsockCon { error!("Failed to re-enable EPOLLOUT"); } - return self.tx_buf.push(&buf.offset(written_count).unwrap()); + return self.tx_buf.push_bytes(&buffer[written_count..]); } Ok(()) } /// Initialize all header fields in the vsock packet. - fn init_pkt<'a, 'b, B: BitmapSlice>( - &self, - pkt: &'a mut VsockPacket<'b, B>, - ) -> &'a mut VsockPacket<'b, B> { + fn init_pkt<'a>(&self, header: &'a mut PacketHeader) -> &'a mut PacketHeader { // Zero out the packet header - pkt.set_header_from_raw(&[0u8; PKT_HEADER_SIZE]).unwrap(); + *header = PacketHeader::default(); - pkt.set_src_cid(self.local_cid) + header + .set_src_cid(self.local_cid) .set_dst_cid(self.guest_cid) .set_src_port(self.local_port) .set_dst_port(self.peer_port) @@ -401,17 +413,9 @@ mod tests { sync::{Arc, Mutex}, }; - use byteorder::{ByteOrder, LittleEndian}; - use virtio_bindings::bindings::virtio_ring::{VRING_DESC_F_NEXT, VRING_DESC_F_WRITE}; - use virtio_queue::{ - desc::{split::Descriptor as SplitDescriptor, RawDescriptor}, - mock::MockSplitQueue, - DescriptorChain, Queue, QueueOwnedT, - }; - use vm_memory::{ - Address, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryLoadGuard, - GuestMemoryMmap, - }; + use virtio_queue::Writer; + use virtio_vsock::packet::PKT_HEADER_SIZE; + use vm_memory::{GuestAddressSpace, ReadVolatile, VolatileSlice}; use super::*; use crate::{ @@ -472,22 +476,32 @@ mod tests { } } - impl WriteVolatile for VsockDummySocket { - fn write_volatile( - &mut self, - buf: &VolatileSlice, - ) -> std::result::Result { + impl VsockStreamIo for VsockDummySocket { + fn stream_write(&mut self, buf: &[u8]) -> std::io::Result { // VecDequeue has no fancy unsafe tricks that vm-memory can abstract. // One could do fairly efficient stuff using the moving From imp... // But this is just for tests, so lets clone, convert to Vec, append, convert // back and replace. let mut write_buffer = self.write_buffer.lock().unwrap(); let mut vec = Vec::from(write_buffer.clone()); - let n = vec.write_volatile(buf)?; + let n = vec.write(buf).unwrap(); *write_buffer = VecDeque::from(vec); Ok(n) } + + fn stream_read(&mut self, writer: &mut Writer) -> Result { + // Similar to the std's Read impl, we only read on the head. Since + // we drain the head, successive reads will cover the rest of the + // queue. + let mut read_buffer = self.read_buffer.lock().unwrap(); + let (head, _) = read_buffer.as_slices(); + let n = writer.write(head).unwrap(); + + read_buffer.drain(..n); + + Ok(n) + } } impl Read for VsockDummySocket { @@ -620,23 +634,18 @@ mod tests { CONN_TX_BUF_SIZE, ); - // write only descriptor chain - let (mem, mut descr_chain) = prepare_desc_chain_vsock(true, &head_params, 2, 10); - let mem = mem.memory(); - let mut pkt = - VsockPacket::from_rx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE) - .unwrap(); - + let mut header = PacketHeader::default(); // initialize a vsock packet for the guest - conn_local.init_pkt(&mut pkt); - - assert_eq!(pkt.src_cid(), VSOCK_HOST_CID); - assert_eq!(pkt.dst_cid(), 3); - assert_eq!(pkt.src_port(), 5000); - assert_eq!(pkt.dst_port(), 5001); - assert_eq!(pkt.type_(), VSOCK_TYPE_STREAM); - assert_eq!(pkt.buf_alloc(), CONN_TX_BUF_SIZE); - assert_eq!(pkt.fwd_cnt(), 0); + conn_local.init_pkt(&mut header); + //pkt.header_slice().write_obj::(header).expect("TODO: write message"); + + assert_eq!(header.src_cid(), VSOCK_HOST_CID); + assert_eq!(header.dst_cid(), 3); + assert_eq!(header.src_port(), 5000); + assert_eq!(header.dst_port(), 5001); + assert_eq!(header.type_(), VSOCK_TYPE_STREAM); + assert_eq!(header.buf_alloc(), CONN_TX_BUF_SIZE); + assert_eq!(header.fwd_cnt(), 0); } #[test] @@ -655,84 +664,115 @@ mod tests { CONN_TX_BUF_SIZE, ); - // write only descriptor chain - let (mem, mut descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); - let mut pkt = - VsockPacket::from_rx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE) - .unwrap(); + let mut packet_header = PacketHeader::default(); // VSOCK_OP_REQUEST: new local conn request + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.rx_queue.enqueue(RxOps::Request); - let op_req = conn_local.recv_pkt(&mut pkt); + let op_req = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_req.unwrap(); assert!(!conn_local.rx_queue.pending_rx()); - assert_eq!(pkt.op(), VSOCK_OP_REQUEST); + assert_eq!(packet_header.op(), VSOCK_OP_REQUEST); // VSOCK_OP_RST: reset if connection not established + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.rx_queue.enqueue(RxOps::Rw); - let op_rst = conn_local.recv_pkt(&mut pkt); + let op_rst = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_rst.unwrap(); assert!(!conn_local.rx_queue.pending_rx()); - assert_eq!(pkt.op(), VSOCK_OP_RST); + assert_eq!(packet_header.op(), VSOCK_OP_RST); // VSOCK_OP_CREDIT_UPDATE: need credit update from peer/guest + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.connect = true; conn_local.rx_queue.enqueue(RxOps::Rw); conn_local.fwd_cnt = Wrapping(1024); - let op_credit_update = conn_local.recv_pkt(&mut pkt); + let op_credit_update = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_credit_update.unwrap(); assert!(!conn_local.rx_queue.pending_rx()); - assert_eq!(pkt.op(), VSOCK_OP_CREDIT_REQUEST); + assert_eq!(packet_header.op(), VSOCK_OP_CREDIT_REQUEST); assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024)); // VSOCK_OP_SHUTDOWN: zero data read from stream/file + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.peer_buf_alloc = 65536; conn_local.rx_queue.enqueue(RxOps::Rw); - let op_zero_read_shutdown = conn_local.recv_pkt(&mut pkt); + let op_zero_read_shutdown = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_zero_read_shutdown.unwrap(); assert!(!conn_local.rx_queue.pending_rx()); assert_eq!(conn_local.rx_cnt, Wrapping(0)); assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024)); - assert_eq!(pkt.op(), VSOCK_OP_SHUTDOWN); + assert_eq!(packet_header.op(), VSOCK_OP_SHUTDOWN); assert_eq!( - pkt.flags(), + packet_header.flags(), VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND ); // VSOCK_OP_RW: finite data read from stream/file + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); let payload = b"hello"; host_socket.write_all(payload).unwrap(); conn_local.rx_queue.enqueue(RxOps::Rw); - let op_zero_read = conn_local.recv_pkt(&mut pkt); + let op_zero_read = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_zero_read.unwrap(); - assert_eq!(pkt.op(), VSOCK_OP_RW); + assert_eq!(packet_header.op(), VSOCK_OP_RW); assert!(!conn_local.rx_queue.pending_rx()); assert_eq!(conn_local.rx_cnt, Wrapping(payload.len() as u32)); assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024)); - assert_eq!(pkt.len(), 5); - let buf = &mut [0u8; 5]; - pkt.data_slice().unwrap().read_slice(buf, 0).unwrap(); - assert_eq!(buf, b"hello"); + assert_eq!(packet_header.len(), 5); // VSOCK_OP_RESPONSE: response from a locally initiated connection + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.rx_queue.enqueue(RxOps::Response); - let op_response = conn_local.recv_pkt(&mut pkt); + let op_response = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_response.unwrap(); assert!(!conn_local.rx_queue.pending_rx()); - assert_eq!(pkt.op(), VSOCK_OP_RESPONSE); + assert_eq!(packet_header.op(), VSOCK_OP_RESPONSE); assert!(conn_local.connect); // VSOCK_OP_CREDIT_UPDATE: guest needs credit update + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); conn_local.rx_queue.enqueue(RxOps::CreditUpdate); - let op_credit_update = conn_local.recv_pkt(&mut pkt); + let op_credit_update = conn_local.recv_pkt(&mut packet_header, &mut pkt); assert!(!conn_local.rx_queue.pending_rx()); op_credit_update.unwrap(); - assert_eq!(pkt.op(), VSOCK_OP_CREDIT_UPDATE); + assert_eq!(packet_header.op(), VSOCK_OP_CREDIT_UPDATE); assert_eq!(conn_local.last_fwd_cnt, Wrapping(1024)); // non-existent request - let op_error = conn_local.recv_pkt(&mut pkt); + packet_header = PacketHeader::default(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let mem = mem.memory(); + let mut pkt = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + let op_error = conn_local.recv_pkt(&mut packet_header, &mut pkt); assert!(op_error.is_err()); } @@ -754,25 +794,24 @@ mod tests { ); // write only descriptor chain - let (mem, mut descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); let mem = mem.memory(); let mut pkt = - VsockPacket::from_tx_virtq_chain(mem.deref(), &mut descr_chain, CONN_TX_BUF_SIZE) - .unwrap(); + VsockPacketTx::from_tx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); // peer credit information - pkt.set_buf_alloc(65536).set_fwd_cnt(1024); + pkt.header_mut().set_buf_alloc(65536).set_fwd_cnt(1024); // check if peer credit information is updated currently - let credit_check = conn_local.send_pkt(&pkt); + let credit_check = conn_local.send_pkt(&mut pkt); credit_check.unwrap(); assert_eq!(conn_local.peer_buf_alloc, 65536); assert_eq!(conn_local.peer_fwd_cnt, Wrapping(1024)); // VSOCK_OP_RESPONSE - pkt.set_op(VSOCK_OP_RESPONSE); + pkt.header_mut().set_op(VSOCK_OP_RESPONSE); assert_eq!(conn_local.peer_port, 5001); - let peer_response = conn_local.send_pkt(&pkt); + let peer_response = conn_local.send_pkt(&mut pkt); peer_response.unwrap(); assert!(conn_local.connect); let mut resp_buf = vec![0; 8]; @@ -780,25 +819,25 @@ mod tests { assert_eq!(&resp_buf, b"OK 5001\n"); // VSOCK_OP_RW - pkt.set_op(VSOCK_OP_RW); - let buf = b"hello"; - pkt.data_slice().unwrap().write_slice(buf, 0).unwrap(); - let rw_response = conn_local.send_pkt(&pkt); + pkt.header_mut().set_op(VSOCK_OP_RW); + // TODO: Data is already written in the descriptor by prepare_desc_chain_vsock + let rw_response = conn_local.send_pkt(&mut pkt); rw_response.unwrap(); let mut resp_buf = vec![0; 5]; host_socket.read_exact(&mut resp_buf).unwrap(); assert_eq!(resp_buf, b"hello"); // VSOCK_OP_CREDIT_REQUEST - pkt.set_op(VSOCK_OP_CREDIT_REQUEST); - let credit_response = conn_local.send_pkt(&pkt); + pkt.header_mut().set_op(VSOCK_OP_CREDIT_REQUEST); + let credit_response = conn_local.send_pkt(&mut pkt); credit_response.unwrap(); assert_eq!(conn_local.rx_queue.peek().unwrap(), RxOps::CreditUpdate); // VSOCK_OP_SHUTDOWN - pkt.set_op(VSOCK_OP_SHUTDOWN); - pkt.set_flags(VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND); - let shutdown_response = conn_local.send_pkt(&pkt); + pkt.header_mut().set_op(VSOCK_OP_SHUTDOWN); + pkt.header_mut() + .set_flags(VSOCK_FLAGS_SHUTDOWN_RCV | VSOCK_FLAGS_SHUTDOWN_SEND); + let shutdown_response = conn_local.send_pkt(&mut pkt); shutdown_response.unwrap(); assert!(conn_local.rx_queue.contains(RxOps::Reset.bitmask())); } From c4ea3a84d19219d1f0793d277c8bff5698d08110 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Thu, 26 Mar 2026 12:39:43 +0100 Subject: [PATCH 5/8] vhost-device-vsock: Remove ReadVolatile/WriteVolatile impls Remove the ReadVolatile and WriteVolatile trait implementations for StreamType and the push method from LocalTxBuf. These are no longer needed after the migration to VsockPacketRx/Tx which uses Reader/Writer instead of VolatileSlice. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/thread_backend.rs | 71 +----------------------- vhost-device-vsock/src/txbuf.rs | 60 ++++---------------- vhost-device-vsock/src/vsock_conn.rs | 19 +------ 3 files changed, 12 insertions(+), 138 deletions(-) diff --git a/vhost-device-vsock/src/thread_backend.rs b/vhost-device-vsock/src/thread_backend.rs index 8e9ef6bf9..deecf44f1 100644 --- a/vhost-device-vsock/src/thread_backend.rs +++ b/vhost-device-vsock/src/thread_backend.rs @@ -16,10 +16,7 @@ use log::{info, warn}; use virtio_queue::Writer; use virtio_vsock::packet_rw::{VsockPacketRx, VsockPacketTx}; use virtio_vsock::{PacketHeader, PKT_HEADER_SIZE}; -use vm_memory::{ - bitmap::BitmapSlice, ByteValued, ReadVolatile, VolatileMemoryError, VolatileSlice, - WriteVolatile, -}; +use vm_memory::{bitmap::BitmapSlice, ByteValued}; #[cfg(feature = "backend_vsock")] use vsock::VsockStream; @@ -154,72 +151,6 @@ impl VsockStreamIo for StreamType { } } -impl ReadVolatile for StreamType { - fn read_volatile( - &mut self, - buf: &mut VolatileSlice<'_, B>, - ) -> StdResult { - match self { - StreamType::Unix(stream) => stream.read_volatile(buf), - // Copied from vm_memory crate's ReadVolatile implementation for UnixStream - #[cfg(feature = "backend_vsock")] - StreamType::Vsock(stream) => { - let fd = stream.as_raw_fd(); - let guard = buf.ptr_guard_mut(); - - let dst = guard.as_ptr().cast::(); - - // SAFETY: We got a valid file descriptor from `AsRawFd`. The memory pointed to - // by `dst` is valid for writes of length `buf.len() by the - // invariants upheld by the constructor of `VolatileSlice`. - let bytes_read = unsafe { libc::read(fd, dst, buf.len()) }; - - if bytes_read < 0 { - // We don't know if a partial read might have happened, so mark everything as - // dirty - buf.bitmap().mark_dirty(0, buf.len()); - - Err(VolatileMemoryError::IOError(std::io::Error::last_os_error())) - } else { - let bytes_read = bytes_read.try_into().unwrap(); - buf.bitmap().mark_dirty(0, bytes_read); - Ok(bytes_read) - } - } - } - } -} - -impl WriteVolatile for StreamType { - fn write_volatile( - &mut self, - buf: &VolatileSlice<'_, B>, - ) -> StdResult { - match self { - StreamType::Unix(stream) => stream.write_volatile(buf), - // Copied from vm_memory crate's WriteVolatile implementation for UnixStream - #[cfg(feature = "backend_vsock")] - StreamType::Vsock(stream) => { - let fd = stream.as_raw_fd(); - let guard = buf.ptr_guard(); - - let src = guard.as_ptr().cast::(); - - // SAFETY: We got a valid file descriptor from `AsRawFd`. The memory pointed to - // by `src` is valid for reads of length `buf.len() by the - // invariants upheld by the constructor of `VolatileSlice`. - let bytes_written = unsafe { libc::write(fd, src, buf.len()) }; - - if bytes_written < 0 { - Err(VolatileMemoryError::IOError(std::io::Error::last_os_error())) - } else { - Ok(bytes_written.try_into().unwrap()) - } - } - } - } -} - pub(crate) trait IsHybridVsock { fn is_hybrid_vsock(&self) -> bool; } diff --git a/vhost-device-vsock/src/txbuf.rs b/vhost-device-vsock/src/txbuf.rs index 4d995cb77..e327be98b 100644 --- a/vhost-device-vsock/src/txbuf.rs +++ b/vhost-device-vsock/src/txbuf.rs @@ -2,8 +2,6 @@ use std::{io::Write, num::Wrapping}; -use vm_memory::{bitmap::BitmapSlice, Bytes, VolatileSlice}; - use crate::vhu_vsock::{Error, Result}; #[derive(Debug)] @@ -61,36 +59,6 @@ impl LocalTxBuf { Ok(()) } - /// Add new data to the tx buffer, push all or none. - /// Returns LocalTxBufFull error if space not sufficient. - pub fn push(&mut self, data_buf: &VolatileSlice) -> Result<()> { - if self.get_buf_size() as usize - self.len() < data_buf.len() { - // Tx buffer is full - return Err(Error::LocalTxBufFull); - } - - // Get index into buffer at which data can be inserted - let tail_idx = self.tail.0 as usize % self.get_buf_size() as usize; - - // Check if we can fit the data buffer between head and end of buffer - let len = std::cmp::min(self.get_buf_size() as usize - tail_idx, data_buf.len()); - let txbuf = &mut self.buf[tail_idx..tail_idx + len]; - data_buf.copy_to(txbuf); - - // Check if there is more data to be wrapped around - if len < data_buf.len() { - let remain_txbuf = &mut self.buf[..(data_buf.len() - len)]; - data_buf - .read_slice(remain_txbuf, len) - .expect("shouldn't fail because remain_txbuf's len is data_buf.len() - len"); - } - - // Increment tail by the amount of data that has been added to the buffer - self.tail += Wrapping(data_buf.len() as u32); - - Ok(()) - } - /// Flush buf data to stream. pub fn flush_to(&mut self, stream: &mut S) -> Result { if self.is_empty() { @@ -166,33 +134,29 @@ mod tests { #[test] fn test_txbuf_push() { let mut loc_tx_buf = LocalTxBuf::new(CONN_TX_BUF_SIZE); - let mut buf = [0; CONN_TX_BUF_SIZE as usize]; - // SAFETY: Safe as the buffer is guaranteed to be valid here. - let data = unsafe { VolatileSlice::new(buf.as_mut_ptr(), buf.len()) }; + let buf = [0; CONN_TX_BUF_SIZE as usize]; // push data into empty tx buffer - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); res_push.unwrap(); assert_eq!(loc_tx_buf.head, Wrapping(0)); assert_eq!(loc_tx_buf.tail, Wrapping(CONN_TX_BUF_SIZE)); // push data into full tx buffer - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); assert!(res_push.is_err()); // head and tail wrap at full loc_tx_buf.head = Wrapping(CONN_TX_BUF_SIZE); - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); res_push.unwrap(); assert_eq!(loc_tx_buf.tail, Wrapping(CONN_TX_BUF_SIZE * 2)); // only tail wraps at full - let mut buf = vec![1, 1, 3, 3]; - // SAFETY: Safe as the buffer is guaranteed to be valid here. - let data = unsafe { VolatileSlice::new(buf.as_mut_ptr(), buf.len()) }; + let buf = vec![1, 1, 3, 3]; loc_tx_buf.head = Wrapping(2); loc_tx_buf.tail = Wrapping(CONN_TX_BUF_SIZE - 2); - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); res_push.unwrap(); assert_eq!(loc_tx_buf.head, Wrapping(2)); assert_eq!(loc_tx_buf.tail, Wrapping(CONN_TX_BUF_SIZE + 2)); @@ -208,12 +172,10 @@ mod tests { let mut loc_tx_buf = LocalTxBuf::new(CONN_TX_BUF_SIZE); // data to be flushed - let mut buf = vec![1; CONN_TX_BUF_SIZE as usize]; - // SAFETY: Safe as the buffer is guaranteed to be valid here. - let data = unsafe { VolatileSlice::new(buf.as_mut_ptr(), buf.len()) }; + let buf = vec![1; CONN_TX_BUF_SIZE as usize]; // target to which data is flushed - let mut cmp_vec = Vec::with_capacity(data.len()); + let mut cmp_vec = Vec::with_capacity(buf.len()); // flush no data let res_flush = loc_tx_buf.flush_to(&mut cmp_vec); @@ -221,7 +183,7 @@ mod tests { assert_eq!(res_flush.unwrap(), 0); // flush data of CONN_TX_BUF_SIZE amount - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); res_push.unwrap(); let res_flush = loc_tx_buf.flush_to(&mut cmp_vec); if let Ok(n) = res_flush { @@ -234,12 +196,10 @@ mod tests { // wrapping head flush let mut buf = vec![0; (CONN_TX_BUF_SIZE / 2) as usize]; buf.append(&mut vec![1; (CONN_TX_BUF_SIZE / 2) as usize]); - // SAFETY: Safe as the buffer is guaranteed to be valid here. - let data = unsafe { VolatileSlice::new(buf.as_mut_ptr(), buf.len()) }; loc_tx_buf.head = Wrapping(0); loc_tx_buf.tail = Wrapping(0); - let res_push = loc_tx_buf.push(&data); + let res_push = loc_tx_buf.push_bytes(&buf); res_push.unwrap(); cmp_vec.clear(); loc_tx_buf.head = Wrapping(CONN_TX_BUF_SIZE / 2); diff --git a/vhost-device-vsock/src/vsock_conn.rs b/vhost-device-vsock/src/vsock_conn.rs index 79c8f7591..24b89e6c2 100644 --- a/vhost-device-vsock/src/vsock_conn.rs +++ b/vhost-device-vsock/src/vsock_conn.rs @@ -415,7 +415,7 @@ mod tests { use virtio_queue::Writer; use virtio_vsock::packet::PKT_HEADER_SIZE; - use vm_memory::{GuestAddressSpace, ReadVolatile, VolatileSlice}; + use vm_memory::GuestAddressSpace; use super::*; use crate::{ @@ -510,23 +510,6 @@ mod tests { } } - impl ReadVolatile for VsockDummySocket { - fn read_volatile( - &mut self, - buf: &mut VolatileSlice, - ) -> std::result::Result { - // Similar to the std's Read impl, we only read on the head. Since - // we drain the head, successive reads will cover the rest of the - // queue. - let mut read_buffer = self.read_buffer.lock().unwrap(); - let (head, _) = read_buffer.as_slices(); - let n = ReadVolatile::read_volatile(&mut &head[..], buf)?; - read_buffer.drain(..n); - - Ok(n) - } - } - impl AsRawFd for VsockDummySocket { fn as_raw_fd(&self) -> RawFd { -1 From 5ec2b121ae09efe35762eb112a3c2eaa03dbb123 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Thu, 26 Mar 2026 14:41:12 +0100 Subject: [PATCH 6/8] vhost-device-vsock: Pass data to `prepare_desc_chain_vsock` Change the test helper to accept a data slice instead of a length, this allows to write a real payload into the descriptor chain memory. This enables some additional tests for send_pkt/recv_pkt. Remove the HeadParams wrapper in favor of passing head_len directly since the data length is now derived from the slice. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/test_utils.rs | 43 +++++---------- vhost-device-vsock/src/thread_backend.rs | 13 ++--- vhost-device-vsock/src/vsock_conn.rs | 70 ++++++++++++------------ 3 files changed, 53 insertions(+), 73 deletions(-) diff --git a/vhost-device-vsock/src/test_utils.rs b/vhost-device-vsock/src/test_utils.rs index beefb3f44..31ae57f2f 100644 --- a/vhost-device-vsock/src/test_utils.rs +++ b/vhost-device-vsock/src/test_utils.rs @@ -13,32 +13,11 @@ use vm_memory::{ GuestMemoryMmap, }; -pub(crate) struct HeadParams { - head_len: usize, - data_len: u32, -} - -impl HeadParams { - pub(crate) fn new(head_len: usize, data_len: u32) -> Self { - Self { head_len, data_len } - } - - fn construct_head(&self) -> Vec { - let mut header = vec![0_u8; self.head_len]; - if self.head_len == PKT_HEADER_SIZE { - // Offset into the header for data length - const HDROFF_LEN: usize = 24; - LittleEndian::write_u32(&mut header[HDROFF_LEN..], self.data_len); - } - header - } -} - pub(crate) fn prepare_desc_chain_vsock( write_only: bool, - head_params: &HeadParams, + head_len: usize, data_chain_len: u16, - head_data_len: u32, + data: &[u8], ) -> ( GuestMemoryAtomic, DescriptorChain>, @@ -59,18 +38,23 @@ pub(crate) fn prepare_desc_chain_vsock( }; // vsock packet header - // let header = vec![0 as u8; head_params.head_len]; - let header = head_params.construct_head(); + let mut header = vec![0_u8; head_len]; + if head_len == PKT_HEADER_SIZE { + // Offset into the header for data length + const HDROFF_LEN: usize = 24; + LittleEndian::write_u32(&mut header[HDROFF_LEN..], data.len() as u32); + } + let head_desc = RawDescriptor::from(SplitDescriptor::new( next_addr, - head_params.head_len as u32, + head_len as u32, head_flags as u16, 1, )); mem.write(&header, SplitDescriptor::from(head_desc).addr()) .unwrap(); virt_queue.desc_table().store(0, head_desc).unwrap(); - next_addr += head_params.head_len as u64; + next_addr += head_len as u64; // Put the descriptor index 0 in the first available ring position. mem.write_obj(0u16, virt_queue.avail_addr().unchecked_add(4)) @@ -87,17 +71,16 @@ pub(crate) fn prepare_desc_chain_vsock( head_flags &= !VRING_DESC_F_NEXT; } // vsock data - let data = vec![0_u8; head_data_len as usize]; let data_desc = RawDescriptor::from(SplitDescriptor::new( next_addr, data.len() as u32, head_flags as u16, i + 2, )); - mem.write(&data, SplitDescriptor::from(data_desc).addr()) + mem.write(data, SplitDescriptor::from(data_desc).addr()) .unwrap(); virt_queue.desc_table().store(i + 1, data_desc).unwrap(); - next_addr += u64::from(head_data_len); + next_addr += data.len() as u64; } // Create descriptor chain from pre-filled memory diff --git a/vhost-device-vsock/src/thread_backend.rs b/vhost-device-vsock/src/thread_backend.rs index deecf44f1..7ef5ab3e5 100644 --- a/vhost-device-vsock/src/thread_backend.rs +++ b/vhost-device-vsock/src/thread_backend.rs @@ -518,8 +518,8 @@ mod tests { #[cfg(feature = "backend_vsock")] use crate::vhu_vsock::VsockProxyInfo; use crate::{ - test_utils::{prepare_desc_chain_vsock, HeadParams}, - vhu_vsock::{BackendType, VhostUserVsockBackend, VsockConfig, VSOCK_OP_RW}, + test_utils::prepare_desc_chain_vsock, + vhu_vsock::{BackendType, VSOCK_OP_RW}, }; //const DATA_LEN: usize = 16; @@ -548,13 +548,13 @@ mod tests { assert!(!vtp.pending_rx()); - let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); - let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, PKT_HEADER_SIZE, 1, b"hello"); let mem = mem.memory(); let mut packet = VsockPacketTx::from_tx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); - let (mem_rx, descr_chain_rx) = prepare_desc_chain_vsock(true, &head_params, 1, 5); + let (mem_rx, descr_chain_rx) = + prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); let mem_rx = mem_rx.memory(); let mut packet_rx = @@ -700,8 +700,7 @@ mod tests { let mut pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; let (hdr_raw, data_raw) = pkt_raw.split_at_mut(PKT_HEADER_SIZE); - let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); - let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, PKT_HEADER_SIZE, 1, b"hello"); let mem = mem.memory(); // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. diff --git a/vhost-device-vsock/src/vsock_conn.rs b/vhost-device-vsock/src/vsock_conn.rs index 24b89e6c2..c105298ac 100644 --- a/vhost-device-vsock/src/vsock_conn.rs +++ b/vhost-device-vsock/src/vsock_conn.rs @@ -419,7 +419,7 @@ mod tests { use super::*; use crate::{ - test_utils::{prepare_desc_chain_vsock, HeadParams}, + test_utils::prepare_desc_chain_vsock, vhu_vsock::{VSOCK_HOST_CID, VSOCK_OP_RW, VSOCK_TYPE_STREAM}, }; @@ -603,9 +603,6 @@ mod tests { #[test] fn test_vsock_conn_init_pkt() { - // parameters for packet head construction - let head_params = HeadParams::new(PKT_HEADER_SIZE, 10); - let dummy_file = VsockDummySocket::new(); let conn_local = VsockConnection::new_local_init( dummy_file, @@ -633,9 +630,6 @@ mod tests { #[test] fn test_vsock_conn_recv_pkt() { - // parameters for packet head construction - let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); - let (mut host_socket, backend_socket) = VsockDummySocket::pair(); let mut conn_local = VsockConnection::new_local_init( backend_socket, @@ -650,10 +644,11 @@ mod tests { let mut packet_header = PacketHeader::default(); // VSOCK_OP_REQUEST: new local conn request - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.rx_queue.enqueue(RxOps::Request); let op_req = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_req.unwrap(); @@ -662,10 +657,11 @@ mod tests { // VSOCK_OP_RST: reset if connection not established packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.rx_queue.enqueue(RxOps::Rw); let op_rst = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_rst.unwrap(); @@ -674,10 +670,11 @@ mod tests { // VSOCK_OP_CREDIT_UPDATE: need credit update from peer/guest packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.connect = true; conn_local.rx_queue.enqueue(RxOps::Rw); conn_local.fwd_cnt = Wrapping(1024); @@ -689,10 +686,11 @@ mod tests { // VSOCK_OP_SHUTDOWN: zero data read from stream/file packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.peer_buf_alloc = 65536; conn_local.rx_queue.enqueue(RxOps::Rw); let op_zero_read_shutdown = conn_local.recv_pkt(&mut packet_header, &mut pkt); @@ -708,10 +706,11 @@ mod tests { // VSOCK_OP_RW: finite data read from stream/file packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); let payload = b"hello"; host_socket.write_all(payload).unwrap(); conn_local.rx_queue.enqueue(RxOps::Rw); @@ -725,10 +724,11 @@ mod tests { // VSOCK_OP_RESPONSE: response from a locally initiated connection packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.rx_queue.enqueue(RxOps::Response); let op_response = conn_local.recv_pkt(&mut packet_header, &mut pkt); op_response.unwrap(); @@ -738,10 +738,11 @@ mod tests { // VSOCK_OP_CREDIT_UPDATE: guest needs credit update packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); conn_local.rx_queue.enqueue(RxOps::CreditUpdate); let op_credit_update = conn_local.recv_pkt(&mut packet_header, &mut pkt); assert!(!conn_local.rx_queue.pending_rx()); @@ -751,19 +752,17 @@ mod tests { // non-existent request packet_header = PacketHeader::default(); - let (mem, descr_chain) = prepare_desc_chain_vsock(true, &head_params, 1, 5); - let mem = mem.memory(); + let (mem, descr_chain) = prepare_desc_chain_vsock(true, PKT_HEADER_SIZE, 1, &[0u8; 5]); + let mem_ref = mem.memory(); let mut pkt = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); + VsockPacketRx::from_rx_virtq_chain(mem_ref.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); let op_error = conn_local.recv_pkt(&mut packet_header, &mut pkt); assert!(op_error.is_err()); } #[test] fn test_vsock_conn_send_pkt() { - // parameters for packet head construction - let head_params = HeadParams::new(PKT_HEADER_SIZE, 5); - // new locally inititated connection let (mut host_socket, backend_socket) = VsockDummySocket::pair(); let mut conn_local = VsockConnection::new_local_init( @@ -777,7 +776,7 @@ mod tests { ); // write only descriptor chain - let (mem, descr_chain) = prepare_desc_chain_vsock(false, &head_params, 1, 5); + let (mem, descr_chain) = prepare_desc_chain_vsock(false, PKT_HEADER_SIZE, 1, b"hello"); let mem = mem.memory(); let mut pkt = VsockPacketTx::from_tx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE).unwrap(); @@ -803,7 +802,6 @@ mod tests { // VSOCK_OP_RW pkt.header_mut().set_op(VSOCK_OP_RW); - // TODO: Data is already written in the descriptor by prepare_desc_chain_vsock let rw_response = conn_local.send_pkt(&mut pkt); rw_response.unwrap(); let mut resp_buf = vec![0; 5]; From eaad4de33b90a1e925c8c53e863d48ff5e8e1909 Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Fri, 17 Apr 2026 16:26:17 +0200 Subject: [PATCH 7/8] [wip] Bump vm-memory from 0.17.1 to 0.18 Signed-off-by: Luigi Leonardi --- Cargo.lock | 50 +++++++------------ Cargo.toml | 2 +- staging/Cargo.lock | 24 +++++---- staging/Cargo.toml | 6 ++- .../src/vhu_video_thread.rs | 2 +- vhost-device-gpu/src/backend/gfxstream.rs | 2 +- vhost-device-gpu/src/backend/virgl.rs | 2 +- vhost-device-gpu/src/testutils.rs | 2 +- 8 files changed, 41 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 75830870c..4054f3a16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2045,7 +2045,7 @@ dependencies = [ "bitflags 2.11.1", "libc", "uuid", - "vm-memory 0.18.0", + "vm-memory", "vmm-sys-util", ] @@ -2064,7 +2064,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2085,7 +2085,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2104,7 +2104,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2129,7 +2129,7 @@ dependencies = [ "virglrenderer", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2147,7 +2147,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2170,7 +2170,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2191,7 +2191,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2210,7 +2210,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2228,7 +2228,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2247,7 +2247,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2272,7 +2272,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2291,7 +2291,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2309,7 +2309,7 @@ dependencies = [ "vhost-user-backend", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", ] @@ -2333,7 +2333,7 @@ dependencies = [ "virtio-bindings 0.2.6", "virtio-queue", "virtio-vsock", - "vm-memory 0.17.1", + "vm-memory", "vmm-sys-util", "vsock", ] @@ -2348,7 +2348,7 @@ dependencies = [ "vhost", "virtio-bindings 0.2.6", "virtio-queue", - "vm-memory 0.18.0", + "vm-memory", "vmm-sys-util", ] @@ -2393,7 +2393,7 @@ dependencies = [ "libc", "log", "virtio-bindings 0.2.7", - "vm-memory 0.18.0", + "vm-memory", "vmm-sys-util", ] @@ -2404,21 +2404,7 @@ source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#51b3425 dependencies = [ "virtio-bindings 0.2.7", "virtio-queue", - "vm-memory 0.18.0", -] - -[[package]] -name = "vm-memory" -version = "0.17.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f39348a049689cabd3377cdd9182bf526ec76a6f823b79903896452e9d7a7380" -dependencies = [ - "arc-swap", - "bitflags 2.11.1", - "libc", - "thiserror 2.0.18", - "vmm-sys-util", - "winapi", + "vm-memory", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 8aa71728c..bdbb94177 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ vhost = { version = "0.15", features = ["vhost-user-backend"] } vhost-user-backend = "0.21" virtio-bindings = "0.2.5" virtio-queue = "0.17" -vm-memory = "0.17.1" +vm-memory = "0.18" vmm-sys-util = "0.15" [workspace.lints.rust] diff --git a/staging/Cargo.lock b/staging/Cargo.lock index f0a357a94..9bed2163f 100644 --- a/staging/Cargo.lock +++ b/staging/Cargo.lock @@ -852,8 +852,7 @@ dependencies = [ [[package]] name = "vhost" version = "0.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c76d90ce3c6b37d610a5304c9a445cfff580cf8b4b9fd02fb256aaf68552c28a" +source = "git+https://github.com/XanClic/vhost.git?branch=vm-memory-0.18#bf6da463b862027b3a4372c3b68f90f9770f3db6" dependencies = [ "bitflags 2.11.1", "libc", @@ -881,7 +880,7 @@ dependencies = [ "v4l2r", "vhost", "vhost-user-backend", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", "vm-memory", "vmm-sys-util", @@ -890,13 +889,12 @@ dependencies = [ [[package]] name = "vhost-user-backend" version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "783587813a59c42c36519a6e12bb31eb2d7fa517377428252ba4cc2312584243" +source = "git+https://github.com/XanClic/vhost.git?branch=vm-memory-0.18#bf6da463b862027b3a4372c3b68f90f9770f3db6" dependencies = [ "libc", "log", "vhost", - "virtio-bindings", + "virtio-bindings 0.2.6", "virtio-queue", "vm-memory", "vmm-sys-util", @@ -908,24 +906,28 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "804f498a26d5a63be7bbb8bdcd3869c3f286c4c4a17108905276454da0caf8cb" +[[package]] +name = "virtio-bindings" +version = "0.2.7" +source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#79927ec7ca8d4fd530b1f2b54ed7a31cd590d503" + [[package]] name = "virtio-queue" version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e358084f32ed165fddb41d98ff1b7ff3c08b9611d8d6114a1b422e2e85688baf" +source = "git+https://github.com/luigix25/vm-virtio.git?branch=fix_vsock#79927ec7ca8d4fd530b1f2b54ed7a31cd590d503" dependencies = [ "libc", "log", - "virtio-bindings", + "virtio-bindings 0.2.7", "vm-memory", "vmm-sys-util", ] [[package]] name = "vm-memory" -version = "0.17.1" +version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f39348a049689cabd3377cdd9182bf526ec76a6f823b79903896452e9d7a7380" +checksum = "9b55e753c7725603745cb32b2287ef7ef3da05c03c7702cda3fa8abe25ae0465" dependencies = [ "arc-swap", "bitflags 2.11.1", diff --git a/staging/Cargo.toml b/staging/Cargo.toml index 20f8216db..76378948f 100644 --- a/staging/Cargo.toml +++ b/staging/Cargo.toml @@ -10,5 +10,9 @@ vhost = { version = "0.15", features = ["vhost-user-backend"] } vhost-user-backend = "0.21" virtio-bindings = "0.2.5" virtio-queue = "0.17" -vm-memory = "0.17.1" +vm-memory = "0.18" vmm-sys-util = "0.15" +[patch.crates-io] +virtio-queue = { git = "https://github.com/luigix25/vm-virtio.git", branch = "fix_vsock" } +vhost = { git = "https://github.com/XanClic/vhost.git", branch = "vm-memory-0.18" } +vhost-user-backend = { git = "https://github.com/XanClic/vhost.git", branch = "vm-memory-0.18" } diff --git a/staging/vhost-device-video/src/vhu_video_thread.rs b/staging/vhost-device-video/src/vhu_video_thread.rs index cae2263d5..8742de96b 100644 --- a/staging/vhost-device-video/src/vhu_video_thread.rs +++ b/staging/vhost-device-video/src/vhu_video_thread.rs @@ -13,7 +13,7 @@ use log::{debug, warn}; use vhost_user_backend::{VringEpollHandler, VringRwLock, VringT}; use virtio_queue::{desc::split::Descriptor as SplitDescriptor, QueueOwnedT}; use vm_memory::{ - ByteValued, Bytes, GuestAddress, GuestAddressSpace, GuestMemory, GuestMemoryAtomic, + ByteValued, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryBackend, GuestMemoryMmap, }; use vmm_sys_util::epoll::EventSet; diff --git a/vhost-device-gpu/src/backend/gfxstream.rs b/vhost-device-gpu/src/backend/gfxstream.rs index 310826237..438df0b03 100644 --- a/vhost-device-gpu/src/backend/gfxstream.rs +++ b/vhost-device-gpu/src/backend/gfxstream.rs @@ -25,7 +25,7 @@ use vhost::vhost_user::{ GpuBackend, }; use vhost_user_backend::{VringRwLock, VringT}; -use vm_memory::{GuestAddress, GuestMemory, GuestMemoryMmap, VolatileSlice}; +use vm_memory::{GuestAddress, GuestMemoryBackend, GuestMemoryMmap, VolatileSlice}; use vmm_sys_util::eventfd::EventFd; use crate::{ diff --git a/vhost-device-gpu/src/backend/virgl.rs b/vhost-device-gpu/src/backend/virgl.rs index 06324fa78..24e8d2545 100644 --- a/vhost-device-gpu/src/backend/virgl.rs +++ b/vhost-device-gpu/src/backend/virgl.rs @@ -28,7 +28,7 @@ use virglrenderer::{ FenceHandler, Iovec, VirglRenderer, VirglRendererFlags, VirglResource, VIRGL_HANDLE_TYPE_MEM_DMABUF, }; -use vm_memory::{GuestAddress, GuestMemory, GuestMemoryMmap, VolatileSlice}; +use vm_memory::{GuestAddress, GuestMemoryBackend, GuestMemoryMmap, VolatileSlice}; use vmm_sys_util::eventfd::EventFd; use crate::{ diff --git a/vhost-device-gpu/src/testutils.rs b/vhost-device-gpu/src/testutils.rs index c1f94ae1d..a6d8e7aae 100644 --- a/vhost-device-gpu/src/testutils.rs +++ b/vhost-device-gpu/src/testutils.rs @@ -21,7 +21,7 @@ use virtio_queue::{ Queue, QueueT, }; use vm_memory::{ - Bytes, GuestAddress, GuestAddressSpace, GuestMemory, GuestMemoryAtomic, GuestMemoryMmap, + Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryBackend, GuestMemoryMmap, }; use vmm_sys_util::eventfd::EventFd; From 20f64c32ebdc7df611993f708c8bdef559e31c3f Mon Sep 17 00:00:00 2001 From: Luigi Leonardi Date: Mon, 20 Apr 2026 13:03:09 +0200 Subject: [PATCH 8/8] [wip] vhost-device-vsock: fix cargo fmt problems Apply cargo fmt suggestions. Not doing this in the "main" commit to make review a little bit easier and this code will be changes anyway. Signed-off-by: Luigi Leonardi --- vhost-device-vsock/src/thread_backend.rs | 288 +++++++++++------------ 1 file changed, 144 insertions(+), 144 deletions(-) diff --git a/vhost-device-vsock/src/thread_backend.rs b/vhost-device-vsock/src/thread_backend.rs index 7ef5ab3e5..79a0abbef 100644 --- a/vhost-device-vsock/src/thread_backend.rs +++ b/vhost-device-vsock/src/thread_backend.rs @@ -628,148 +628,148 @@ mod tests { //TODO: fix tests /* - #[test] - fn test_vsock_thread_backend_sibling_vms() { - const CID: u64 = 3; - const SIBLING_CID: u64 = 4; - const SIBLING2_CID: u64 = 5; - const SIBLING_LISTENING_PORT: u32 = 1234; - - let test_dir = tempdir().expect("Could not create a temp test directory."); - - let vsock_socket_path = test_dir.path().join("test_vsock_thread_backend.vsock"); - let sibling_vhost_socket_path = test_dir - .path() - .join("test_vsock_thread_backend_sibling.socket"); - let sibling_vsock_socket_path = test_dir - .path() - .join("test_vsock_thread_backend_sibling.vsock"); - let sibling2_vhost_socket_path = test_dir - .path() - .join("test_vsock_thread_backend_sibling2.socket"); - let sibling2_vsock_socket_path = test_dir - .path() - .join("test_vsock_thread_backend_sibling2.vsock"); - - let cid_map: Arc> = Arc::new(RwLock::new(HashMap::new())); - - let sibling_config = VsockConfig::new( - SIBLING_CID, - sibling_vhost_socket_path, - BackendType::UnixDomainSocket(sibling_vsock_socket_path), - CONN_TX_BUF_SIZE, - QUEUE_SIZE, - vec!["group1", "group2", "group3"] - .into_iter() - .map(String::from) - .collect(), - ); - - let sibling2_config = VsockConfig::new( - SIBLING2_CID, - sibling2_vhost_socket_path, - BackendType::UnixDomainSocket(sibling2_vsock_socket_path), - CONN_TX_BUF_SIZE, - QUEUE_SIZE, - vec!["group1"].into_iter().map(String::from).collect(), - ); - - let sibling_backend = - Arc::new(VhostUserVsockBackend::new(sibling_config, cid_map.clone()).unwrap()); - let sibling2_backend = - Arc::new(VhostUserVsockBackend::new(sibling2_config, cid_map.clone()).unwrap()); - - let epoll_fd = epoll::create(false).unwrap(); - - let groups_set: HashSet = vec!["groupA", "groupB", "group3"] - .into_iter() - .map(String::from) - .collect(); - - let mut vtp = VsockThreadBackend::new( - BackendType::UnixDomainSocket(vsock_socket_path), - epoll_fd, - CID, - CONN_TX_BUF_SIZE, - Arc::new(RwLock::new(groups_set)), - cid_map, - ); - - assert!(!vtp.pending_raw_pkts()); - - let mut pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; - let (hdr_raw, data_raw) = pkt_raw.split_at_mut(PKT_HEADER_SIZE); - - let (mem, descr_chain) = prepare_desc_chain_vsock(false, PKT_HEADER_SIZE, 1, b"hello"); - let mem = mem.memory(); - - // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. - let mut packet = - VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE) - .unwrap(); - assert_eq!( - vtp.recv_raw_pkt(&mut packet).unwrap_err().to_string(), - Error::EmptyRawPktsQueue.to_string() - ); - - - packet.header_mut().set_type(VSOCK_TYPE_STREAM); - packet.header_mut().set_src_cid(CID); - packet.header_mut().set_dst_cid(SIBLING_CID); - packet.header_mut().set_dst_port(SIBLING_LISTENING_PORT); - packet.header_mut().set_op(VSOCK_OP_RW); - packet.header_mut().set_len(DATA_LEN as u32); - - //Payload is hello - /*packet - .data_slice() - .unwrap() - .copy_from(&[0xCAu8, 0xFEu8, 0xBAu8, 0xBEu8]); - */ - - vtp.send_pkt(&mut &packet).unwrap(); - assert!(sibling_backend.threads[0] - .lock() - .unwrap() - .thread_backend - .pending_raw_pkts()); - - packet.header_mut().set_dst_cid(SIBLING2_CID); - vtp.send_pkt(&mut &packet).unwrap(); - // packet should be discarded since sibling2 is not in the same group - assert!(!sibling2_backend.threads[0] - .lock() - .unwrap() - .thread_backend - .pending_raw_pkts()); - - let mut recvd_pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; - let (recvd_hdr_raw, recvd_data_raw) = recvd_pkt_raw.split_at_mut(PKT_HEADER_SIZE); - - let mut recvd_packet = - // SAFETY: Safe as recvd_hdr_raw and recvd_data_raw are guaranteed to be valid. - unsafe { VsockPacket::new(recvd_hdr_raw, Some(recvd_data_raw)).unwrap() }; - - sibling_backend.threads[0] - .lock() - .unwrap() - .thread_backend - .recv_raw_pkt(&mut recvd_packet) - .unwrap(); - - assert_eq!(recvd_packet.type_(), VSOCK_TYPE_STREAM); - assert_eq!(recvd_packet.src_cid(), CID); - assert_eq!(recvd_packet.dst_cid(), SIBLING_CID); - assert_eq!(recvd_packet.dst_port(), SIBLING_LISTENING_PORT); - assert_eq!(recvd_packet.op(), VSOCK_OP_RW); - assert_eq!(recvd_packet.len(), DATA_LEN as u32); - - assert_eq!(recvd_data_raw[0], 0xCAu8); - assert_eq!(recvd_data_raw[1], 0xFEu8); - assert_eq!(recvd_data_raw[2], 0xBAu8); - assert_eq!(recvd_data_raw[3], 0xBEu8); - - test_dir.close().unwrap(); - } - */ + #[test] + fn test_vsock_thread_backend_sibling_vms() { + const CID: u64 = 3; + const SIBLING_CID: u64 = 4; + const SIBLING2_CID: u64 = 5; + const SIBLING_LISTENING_PORT: u32 = 1234; + + let test_dir = tempdir().expect("Could not create a temp test directory."); + + let vsock_socket_path = test_dir.path().join("test_vsock_thread_backend.vsock"); + let sibling_vhost_socket_path = test_dir + .path() + .join("test_vsock_thread_backend_sibling.socket"); + let sibling_vsock_socket_path = test_dir + .path() + .join("test_vsock_thread_backend_sibling.vsock"); + let sibling2_vhost_socket_path = test_dir + .path() + .join("test_vsock_thread_backend_sibling2.socket"); + let sibling2_vsock_socket_path = test_dir + .path() + .join("test_vsock_thread_backend_sibling2.vsock"); + + let cid_map: Arc> = Arc::new(RwLock::new(HashMap::new())); + + let sibling_config = VsockConfig::new( + SIBLING_CID, + sibling_vhost_socket_path, + BackendType::UnixDomainSocket(sibling_vsock_socket_path), + CONN_TX_BUF_SIZE, + QUEUE_SIZE, + vec!["group1", "group2", "group3"] + .into_iter() + .map(String::from) + .collect(), + ); + + let sibling2_config = VsockConfig::new( + SIBLING2_CID, + sibling2_vhost_socket_path, + BackendType::UnixDomainSocket(sibling2_vsock_socket_path), + CONN_TX_BUF_SIZE, + QUEUE_SIZE, + vec!["group1"].into_iter().map(String::from).collect(), + ); + + let sibling_backend = + Arc::new(VhostUserVsockBackend::new(sibling_config, cid_map.clone()).unwrap()); + let sibling2_backend = + Arc::new(VhostUserVsockBackend::new(sibling2_config, cid_map.clone()).unwrap()); + + let epoll_fd = epoll::create(false).unwrap(); + + let groups_set: HashSet = vec!["groupA", "groupB", "group3"] + .into_iter() + .map(String::from) + .collect(); + + let mut vtp = VsockThreadBackend::new( + BackendType::UnixDomainSocket(vsock_socket_path), + epoll_fd, + CID, + CONN_TX_BUF_SIZE, + Arc::new(RwLock::new(groups_set)), + cid_map, + ); + + assert!(!vtp.pending_raw_pkts()); + + let mut pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; + let (hdr_raw, data_raw) = pkt_raw.split_at_mut(PKT_HEADER_SIZE); + + let (mem, descr_chain) = prepare_desc_chain_vsock(false, PKT_HEADER_SIZE, 1, b"hello"); + let mem = mem.memory(); + + // SAFETY: Safe as hdr_raw and data_raw are guaranteed to be valid. + let mut packet = + VsockPacketRx::from_rx_virtq_chain(mem.deref(), descr_chain, CONN_TX_BUF_SIZE) + .unwrap(); + assert_eq!( + vtp.recv_raw_pkt(&mut packet).unwrap_err().to_string(), + Error::EmptyRawPktsQueue.to_string() + ); + + + packet.header_mut().set_type(VSOCK_TYPE_STREAM); + packet.header_mut().set_src_cid(CID); + packet.header_mut().set_dst_cid(SIBLING_CID); + packet.header_mut().set_dst_port(SIBLING_LISTENING_PORT); + packet.header_mut().set_op(VSOCK_OP_RW); + packet.header_mut().set_len(DATA_LEN as u32); + + //Payload is hello + /*packet + .data_slice() + .unwrap() + .copy_from(&[0xCAu8, 0xFEu8, 0xBAu8, 0xBEu8]); + */ + + vtp.send_pkt(&mut &packet).unwrap(); + assert!(sibling_backend.threads[0] + .lock() + .unwrap() + .thread_backend + .pending_raw_pkts()); + + packet.header_mut().set_dst_cid(SIBLING2_CID); + vtp.send_pkt(&mut &packet).unwrap(); + // packet should be discarded since sibling2 is not in the same group + assert!(!sibling2_backend.threads[0] + .lock() + .unwrap() + .thread_backend + .pending_raw_pkts()); + + let mut recvd_pkt_raw = [0u8; PKT_HEADER_SIZE + DATA_LEN]; + let (recvd_hdr_raw, recvd_data_raw) = recvd_pkt_raw.split_at_mut(PKT_HEADER_SIZE); + + let mut recvd_packet = + // SAFETY: Safe as recvd_hdr_raw and recvd_data_raw are guaranteed to be valid. + unsafe { VsockPacket::new(recvd_hdr_raw, Some(recvd_data_raw)).unwrap() }; + + sibling_backend.threads[0] + .lock() + .unwrap() + .thread_backend + .recv_raw_pkt(&mut recvd_packet) + .unwrap(); + + assert_eq!(recvd_packet.type_(), VSOCK_TYPE_STREAM); + assert_eq!(recvd_packet.src_cid(), CID); + assert_eq!(recvd_packet.dst_cid(), SIBLING_CID); + assert_eq!(recvd_packet.dst_port(), SIBLING_LISTENING_PORT); + assert_eq!(recvd_packet.op(), VSOCK_OP_RW); + assert_eq!(recvd_packet.len(), DATA_LEN as u32); + + assert_eq!(recvd_data_raw[0], 0xCAu8); + assert_eq!(recvd_data_raw[1], 0xFEu8); + assert_eq!(recvd_data_raw[2], 0xBAu8); + assert_eq!(recvd_data_raw[3], 0xBEu8); + + test_dir.close().unwrap(); + } + */ }