diff --git a/virtio-devices/src/vsock/csm/connection.rs b/virtio-devices/src/vsock/csm/connection.rs index fe89365e3..2264887e6 100644 --- a/virtio-devices/src/vsock/csm/connection.rs +++ b/virtio-devices/src/vsock/csm/connection.rs @@ -86,17 +86,25 @@ use std::os::unix::io::{AsRawFd, RawFd}; use std::time::{Duration, Instant}; use log::{debug, error, info, warn}; +use vm_memory::{ReadVolatile, WriteVolatile}; use super::super::defs::uapi; use super::super::packet::VsockPacket; use super::super::{Result as VsockResult, VsockChannel, VsockEpollListener, VsockError}; -use super::txbuf::TxBuf; +use super::txbuf::{TxBuf, TxBufSource}; use super::{ConnState, Error, PendingRx, PendingRxSet, Result, defs}; +impl TxBufSource for VsockPacket { + fn copy_to_tx_buf(&self, offset: usize, dst: &mut [u8]) -> Result<()> { + self.copy_buf_to_slice(offset, dst) + .map_err(|_| Error::PktBufRead) + } +} + /// A self-managing connection object, that handles communication between a guest-side AF_VSOCK /// socket and a host-side `Read + Write + AsRawFd` stream. /// -pub struct VsockConnection { +pub struct VsockConnection { /// The current connection state. state: ConnState, /// The local CID. Most of the time this will be the constant `2` (the vsock host CID). @@ -133,7 +141,7 @@ pub struct VsockConnection { impl VsockChannel for VsockConnection where - S: Read + Write + AsRawFd, + S: Read + ReadVolatile + Write + WriteVolatile + AsRawFd, { /// Fill in a vsock packet, to be delivered to our peer (the guest driver). /// @@ -205,14 +213,14 @@ where return Ok(()); } - let buf = pkt.buf_mut().ok_or(VsockError::PktBufMissing)?; + let buf_capacity = pkt.buf_capacity().ok_or(VsockError::PktBufMissing)?; // The maximum amount of data we can read in is limited by both the RX buffer size and // the peer available buffer space. - let max_len = std::cmp::min(buf.len(), self.peer_avail_credit()); + let max_len = std::cmp::min(buf_capacity, self.peer_avail_credit()); // Read data from the stream straight to the RX buffer, for maximum throughput. - match self.stream.read(&mut buf[..max_len]) { + match pkt.read_volatile_from(&mut self.stream, max_len) { Ok(read_cnt) => { if read_cnt == 0 { // A 0-length read means the host stream was closed down. In that case, @@ -291,7 +299,7 @@ where ConnState::Established | ConnState::PeerClosed(_, false) if pkt.op() == uapi::VSOCK_OP_RW => { - if pkt.buf().is_none() { + if !pkt.has_buf() { info!( "vsock: dropping empty data packet from guest (lp={}, pp={}", self.local_port, self.peer_port @@ -299,9 +307,7 @@ where return Ok(()); } - // Unwrapping here is safe, since we just checked `pkt.buf()` above. - let buf_slice = &pkt.buf().unwrap()[..(pkt.len() as usize)]; - if let Err(err) = self.send_bytes(buf_slice) { + if let Err(err) = self.send_pkt_bytes(pkt) { // If we can't write to the host stream, that's an unrecoverable error, so // we'll terminate this connection. warn!( @@ -396,7 +402,7 @@ where impl VsockEpollListener for VsockConnection where - S: Read + Write + AsRawFd, + S: Read + ReadVolatile + Write + WriteVolatile + AsRawFd, { /// Get the file descriptor that this connection wants polled. /// @@ -481,7 +487,7 @@ where impl VsockConnection where - S: Read + Write + AsRawFd, + S: Read + ReadVolatile + Write + WriteVolatile + AsRawFd, { /// Create a new guest-initiated connection object. /// @@ -589,22 +595,24 @@ where self.stream.write(buf).map_err(Error::StreamWrite) } - /// Send some raw data (a byte-slice) to the host stream. + /// Send packet data to the host stream. /// - /// Raw data can either be sent straight to the host stream, or to our TX buffer, if the + /// Packet data can either be sent straight to the host stream, or to our TX buffer, if the /// former fails. /// - fn send_bytes(&mut self, buf: &[u8]) -> Result<()> { + fn send_pkt_bytes(&mut self, pkt: &VsockPacket) -> Result<()> { + let len = pkt.len() as usize; + // If there is data in the TX buffer, that means we're already registered for EPOLLOUT // events on the underlying stream. Therefore, there's no point in attempting a write // at this point. `self.notify()` will get called when EPOLLOUT arrives, and it will // attempt to drain the TX buffer then. if !self.tx_buf.is_empty() { - return self.tx_buf.push(buf); + return self.buffer_pkt_bytes(pkt, 0, len); } // The TX buffer is empty, so we can try to write straight to the host stream. - let written = match self.stream.write(buf) { + let written = match pkt.write_volatile_to(&mut self.stream, 0, len) { Ok(cnt) => cnt, Err(e) => { // Absorb any would-block errors, since we can always try again later. @@ -620,15 +628,19 @@ where // Move the "forwarded bytes" counter ahead by how much we were able to send out. self.fwd_cnt += Wrapping(written as u32); - // If we couldn't write the whole slice, we'll need to push the remaining data to our + // If we couldn't write the whole packet, we'll need to push the remaining data to our // buffer. - if written < buf.len() { - self.tx_buf.push(&buf[written..])?; + if written < len { + self.buffer_pkt_bytes(pkt, written, len - written)?; } Ok(()) } + fn buffer_pkt_bytes(&mut self, pkt: &VsockPacket, offset: usize, len: usize) -> Result<()> { + self.tx_buf.push_from(pkt, offset, len) + } + /// Check if the credit information the peer has last received from us is outdated. /// fn peer_needs_credit_update(&self) -> bool { @@ -677,6 +689,8 @@ mod unit_tests { use libc::EFD_NONBLOCK; use virtio_queue::QueueOwnedT; + use vm_memory::bitmap::BitmapSlice; + use vm_memory::{VolatileMemoryError, VolatileSlice}; use vmm_sys_util::eventfd::EventFd; use super::super::super::unit_tests::TestContext; @@ -763,9 +777,32 @@ mod unit_tests { } } + impl ReadVolatile for TestStream { + fn read_volatile( + &mut self, + data: &mut VolatileSlice, + ) -> std::result::Result { + let mut buf = vec![0u8; data.len()]; + let len = self.read(&mut buf).map_err(VolatileMemoryError::IOError)?; + data.copy_from(&buf[..len]); + Ok(len) + } + } + + impl WriteVolatile for TestStream { + fn write_volatile( + &mut self, + data: &VolatileSlice, + ) -> std::result::Result { + let mut buf = vec![0u8; data.len()]; + data.copy_to(&mut buf); + self.write(&buf).map_err(VolatileMemoryError::IOError) + } + } + impl VsockConnection where - S: Read + Write + AsRawFd, + S: Read + ReadVolatile + Write + WriteVolatile + AsRawFd, { /// Get the fwd_cnt value from the connection. pub(crate) fn fwd_cnt(&self) -> Wrapping { @@ -894,11 +931,17 @@ mod unit_tests { } fn init_data_pkt(&mut self, data: &[u8]) -> &VsockPacket { - assert!(data.len() <= self.pkt.buf().unwrap().len()); + assert!(data.len() <= self.pkt.buf_capacity().unwrap()); self.init_pkt(uapi::VSOCK_OP_RW, data.len() as u32); - self.pkt.buf_mut().unwrap()[..data.len()].copy_from_slice(data); + self.pkt.copy_buf_from_slice(0, data).unwrap(); &self.pkt } + + fn pkt_data(&self) -> Vec { + let mut data = vec![0u8; self.pkt.len() as usize]; + self.pkt.copy_buf_to_slice(0, &mut data).unwrap(); + data + } } #[test] @@ -964,7 +1007,7 @@ mod unit_tests { ctx.recv(); assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RW); assert_eq!(ctx.pkt.len() as usize, data.len()); - assert_eq!(ctx.pkt.buf().unwrap()[..ctx.pkt.len() as usize], *data); + assert_eq!(ctx.pkt_data().as_slice(), data); // There's no more data in the stream, so `recv_pkt` should yield `VsockError::NoData`. match ctx.conn.recv_pkt(&mut ctx.pkt) { @@ -1034,7 +1077,7 @@ mod unit_tests { ctx.notify_epollin(); ctx.recv(); assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RW); - assert_eq!(&ctx.pkt.buf().unwrap()[..ctx.pkt.len() as usize], data); + assert_eq!(ctx.pkt_data().as_slice(), data); ctx.init_data_pkt(data); ctx.send(); @@ -1234,7 +1277,7 @@ mod unit_tests { ctx.set_stream(stream); // Fill up the TX buffer. - let data = vec![0u8; ctx.pkt.buf().unwrap().len()]; + let data = vec![0u8; ctx.pkt.buf_capacity().unwrap()]; ctx.init_data_pkt(data.as_slice()); for _i in 0..(csm_defs::CONN_TX_BUF_SIZE / data.len() as u32) { ctx.send(); diff --git a/virtio-devices/src/vsock/csm/mod.rs b/virtio-devices/src/vsock/csm/mod.rs index 69de7073d..590ba0a7b 100644 --- a/virtio-devices/src/vsock/csm/mod.rs +++ b/virtio-devices/src/vsock/csm/mod.rs @@ -37,6 +37,9 @@ pub enum Error { /// An I/O error occurred, when attempting to write data to the host-side stream. #[error("Error writing to host side stream")] StreamWrite(#[source] std::io::Error), + /// An I/O error occurred, when reading packet data. + #[error("Error reading packet buffer")] + PktBufRead, } type Result = std::result::Result; diff --git a/virtio-devices/src/vsock/csm/txbuf.rs b/virtio-devices/src/vsock/csm/txbuf.rs index 38906fc83..d5720c620 100644 --- a/virtio-devices/src/vsock/csm/txbuf.rs +++ b/virtio-devices/src/vsock/csm/txbuf.rs @@ -7,6 +7,19 @@ use std::num::Wrapping; use super::{Error, Result, defs}; +pub(super) trait TxBufSource { + fn copy_to_tx_buf(&self, offset: usize, dst: &mut [u8]) -> Result<()>; +} + +impl TxBufSource for [u8] { + fn copy_to_tx_buf(&self, offset: usize, dst: &mut [u8]) -> Result<()> { + let end = offset.checked_add(dst.len()).ok_or(Error::PktBufRead)?; + let src = self.get(offset..end).ok_or(Error::PktBufRead)?; + dst.copy_from_slice(src); + Ok(()) + } +} + /// A simple ring-buffer implementation, used by vsock connections to buffer TX (guest -> host) /// data. Memory for this buffer is allocated lazily, since buffering will only be needed when /// the host can't read fast enough. @@ -42,17 +55,24 @@ impl TxBuf { (self.head - self.tail).0 as usize } - /// Push a byte slice onto the ring-buffer. + /// Push data from a copy source into the ring-buffer. /// - /// Either the entire source slice will be pushed to the ring-buffer, or none of it, if - /// there isn't enough room, in which case `Err(Error::TxBufFull)` is returned. + /// Either the entire length will be pushed to the ring-buffer, or none of it, if there + /// isn't enough room, in which case `Err(Error::TxBufFull)` is returned. /// - pub fn push(&mut self, src: &[u8]) -> Result<()> { - // Error out if there's no room to push the entire slice. - if self.len() + src.len() > Self::SIZE { + pub(super) fn push_from(&mut self, src: &S, offset: usize, len: usize) -> Result<()> + where + S: TxBufSource + ?Sized, + { + // Error out if there's no room to push the entire length. + if self.len() + len > Self::SIZE { return Err(Error::TxBufFull); } + if len == 0 { + return Ok(()); + } + let data = self .data .get_or_insert_with(|| vec![0u8; Self::SIZE].into_boxed_slice()); @@ -60,23 +80,24 @@ impl TxBuf { // Buffer head, as an offset into the data slice. let head_ofs = self.head.0 as usize % Self::SIZE; - // Pushing a slice to this buffer can take either one or two slice copies: - one copy, - // if the slice fits between `head_ofs` and `Self::SIZE`; or - two copies, if the + // Pushing to this buffer can take either one or two copies: - one copy, if the data + // fits between `head_ofs` and `Self::SIZE`; or - two copies, if the // ring-buffer head wraps around. // First copy length: we can only go from the head offset up to the total buffer size. - let len = std::cmp::min(Self::SIZE - head_ofs, src.len()); - data[head_ofs..(head_ofs + len)].copy_from_slice(&src[..len]); + let first_len = std::cmp::min(Self::SIZE - head_ofs, len); + src.copy_to_tx_buf(offset, &mut data[head_ofs..(head_ofs + first_len)])?; - // If the slice didn't fit, the buffer head will wrap around, and pushing continues + // If the data didn't fit, the buffer head will wrap around, and pushing continues // from the start of the buffer (`&self.data[0]`). - if len < src.len() { - data[..(src.len() - len)].copy_from_slice(&src[len..]); + if first_len < len { + let offset = offset.checked_add(first_len).ok_or(Error::PktBufRead)?; + src.copy_to_tx_buf(offset, &mut data[..(len - first_len)])?; } - // Either way, we've just pushed exactly `src.len()` bytes, so that's the amount by + // Either way, we've just pushed exactly `len` bytes, so that's the amount by // which the (wrapping) buffer head needs to move forward. - self.head += Wrapping(src.len() as u32); + self.head += Wrapping(len as u32); Ok(()) } @@ -139,6 +160,17 @@ impl TxBuf { pub fn is_empty(&self) -> bool { self.len() == 0 } + + /// Push a byte slice onto the ring-buffer. + /// + /// A thin convenience wrapper around `push_from`, used only by the unit tests to push a + /// plain slice without having to spell out the offset and length. Production code pushes + /// directly from a packet buffer via `push_from`. + /// + #[cfg(test)] + pub fn push(&mut self, src: &[u8]) -> Result<()> { + self.push_from(src, 0, src.len()) + } } #[cfg(test)] @@ -221,6 +253,22 @@ mod unit_tests { assert_eq!(sink.data, [1, 2, 3, 4]); } + #[test] + fn test_push_from_wrap() { + let mut txbuf = TxBuf::new(); + let mut sink = TestSink::new(); + let tmp: Vec = vec![0; TxBuf::SIZE - 2]; + txbuf.push(tmp.as_slice()).unwrap(); + txbuf.flush_to(&mut sink).unwrap(); + sink.clear(); + + let src = [1, 2, 3, 4]; + txbuf.push_from(&src[..], 0, src.len()).unwrap(); + + assert_eq!(txbuf.flush_to(&mut sink).unwrap(), 4); + assert_eq!(sink.data, src); + } + #[test] fn test_push_error() { let mut txbuf = TxBuf::new(); diff --git a/virtio-devices/src/vsock/packet.rs b/virtio-devices/src/vsock/packet.rs index 2f5246a34..df6a1e7b5 100644 --- a/virtio-devices/src/vsock/packet.rs +++ b/virtio-devices/src/vsock/packet.rs @@ -27,7 +27,7 @@ use vm_virtio::AccessPlatform; use vm_virtio::checked_descriptor::DescriptorChainExt; use super::{Result, VsockError, defs}; -use crate::{GuestMemoryMmap, get_host_address_range}; +use crate::GuestMemoryMmap; // The vsock packet header is defined by the C struct: // @@ -530,46 +530,6 @@ impl VsockPacket { .copy_to_slice(offset, dst) } - /// Provides in-place, byte-slice access to the vsock packet data buffer. - /// - /// Note: control packets (e.g. connection request or reset) have no data buffer associated. - /// For those packets, this method will return `None`. - /// Also note: calling `len()` on the returned slice will yield the buffer size, which may be - /// (and often is) larger than the length of the packet data. The packet data length - /// is stored in the packet header, and accessible via `VsockPacket::len()`. - pub fn buf(&self) -> Option<&[u8]> { - match self.buf.as_ref()? { - PacketBuffer::Owned(owned) => Some(owned), - PacketBuffer::Guest { mem, addr, len } => { - let ptr = get_host_address_range(mem, *addr, *len)?; - - // SAFETY: bound checks have already been performed when creating the packet - // from the virtq descriptor. - Some(unsafe { std::slice::from_raw_parts(ptr.cast(), *len) }) - } - } - } - - /// Provides in-place, byte-slice, mutable access to the vsock packet data buffer. - /// - /// Note: control packets (e.g. connection request or reset) have no data buffer associated. - /// For those packets, this method will return `None`. - /// Also note: calling `len()` on the returned slice will yield the buffer size, which may be - /// (and often is) larger than the length of the packet data. The packet data length - /// is stored in the packet header, and accessible via `VsockPacket::len()`. - pub fn buf_mut(&mut self) -> Option<&mut [u8]> { - match self.buf.as_mut()? { - PacketBuffer::Owned(owned) => Some(owned), - PacketBuffer::Guest { mem, addr, len } => { - let ptr = get_host_address_range(mem, *addr, *len)?; - - // SAFETY: bound checks have already been performed when creating the packet - // from the virtq descriptor. - Some(unsafe { std::slice::from_raw_parts_mut(ptr, *len) }) - } - } - } - #[cfg(test)] pub fn copy_buf_from_slice(&mut self, offset: usize, src: &[u8]) -> Result<()> { self.buf diff --git a/virtio-devices/src/vsock/unix/muxer.rs b/virtio-devices/src/vsock/unix/muxer.rs index 499c68eac..f60a46d7c 100644 --- a/virtio-devices/src/vsock/unix/muxer.rs +++ b/virtio-devices/src/vsock/unix/muxer.rs @@ -969,13 +969,19 @@ mod unit_tests { peer_port: u32, data: &[u8], ) -> &mut VsockPacket { - assert!(data.len() <= self.pkt.buf().unwrap().len()); + assert!(data.len() <= self.pkt.buf_capacity().unwrap()); self.init_pkt(local_port, peer_port, uapi::VSOCK_OP_RW) .set_len(data.len() as u32); - self.pkt.buf_mut().unwrap()[..data.len()].copy_from_slice(data); + self.pkt.copy_buf_from_slice(0, data).unwrap(); &mut self.pkt } + fn pkt_data(&self) -> Vec { + let mut data = vec![0u8; self.pkt.len() as usize]; + self.pkt.copy_buf_to_slice(0, &mut data).unwrap(); + data + } + fn send(&mut self) { self.muxer.send_pkt(&self.pkt).unwrap(); } @@ -1204,7 +1210,7 @@ mod unit_tests { assert!(ctx.muxer.has_pending_rx()); ctx.recv(); assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RW); - assert_eq!(ctx.pkt.buf().unwrap()[..data.len()], data); + assert_eq!(ctx.pkt_data().as_slice(), data); assert_eq!(ctx.pkt.src_port(), LOCAL_PORT); assert_eq!(ctx.pkt.dst_port(), PEER_PORT); @@ -1236,7 +1242,7 @@ mod unit_tests { assert_eq!(ctx.pkt.op(), uapi::VSOCK_OP_RW); assert_eq!(ctx.pkt.src_port(), local_port); assert_eq!(ctx.pkt.dst_port(), peer_port); - assert_eq!(ctx.pkt.buf().unwrap()[..data.len()], data); + assert_eq!(ctx.pkt_data().as_slice(), data); } #[test]