kata-containers/cloud-hypervisor
1
0
Fork 0
mirror of https://github.com/cloud-hypervisor/cloud-hypervisor.git synced 2026-08-05 02:19:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

virtio-devices: vsock: use volatile packet I/O

Browse Source 
Remove the need for unsafely materializing slices from guest memory
pointers which is, by definition, undefined behavior.

Achieved by introducing a TxBufSource trait that is implemented for both
types of sources (Guest Memory or local copy) and by using the volatile
read/write primities for moving data from a readable or writable to
guest memory.

Assisted-by: Codex:GPT-5
Signed-off-by: Dylan Reid <dgreid@fb.com>
This commit is contained in:
Dylan Reid
2026-05-29 11:52:25 -07:00
committed by Rob Bradford
parent 44f0360d28
commit c6ecc13d08
5 changed files with 146 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
virtio-devices/src/vsock/csm/connection.rs
View File

@@ -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<S: Read + Write + AsRawFd> {
pub struct VsockConnection<S: Read + ReadVolatile + Write + WriteVolatile + AsRawFd> {
/// 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<S: Read + Write + AsRawFd> {
impl<S> VsockChannel for VsockConnection<S>
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<S> VsockEpollListener for VsockConnection<S>
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<S> VsockConnection<S>
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<B: BitmapSlice>(
&mut self,
data: &mut VolatileSlice<B>,
) -> std::result::Result<usize, VolatileMemoryError> {
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<B: BitmapSlice>(
&mut self,
data: &VolatileSlice<B>,
) -> std::result::Result<usize, VolatileMemoryError> {
let mut buf = vec![0u8; data.len()];
data.copy_to(&mut buf);
self.write(&buf).map_err(VolatileMemoryError::IOError)
}
}
impl<S> VsockConnection<S>
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<u32> {
@@ -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<u8> {
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();

3
virtio-devices/src/vsock/csm/mod.rs
View File

@@ -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<T> = std::result::Result<T, Error>;

78
virtio-devices/src/vsock/csm/txbuf.rs
View File

@@ -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<S>(&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<u8> = 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();

42
virtio-devices/src/vsock/packet.rs
View File

@@ -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

14
virtio-devices/src/vsock/unix/muxer.rs
View File

@@ -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<u8> {
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]