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

vm-virtio, virtio-devices: Split device implementation from virt queues

Browse Source 
Split the generic virtio code (queues and device type) from the
VirtioDevice trait, transport and device implementations.

This also simplifies the feature handling in vhost_user_backend as the
vm-virtio crate is no longer has any features.

Signed-off-by: Rob Bradford <robert.bradford@intel.com>
This commit is contained in:
Rob Bradford
2020-07-02 13:25:19 +01:00
parent 9a628edfcf
commit 2a6eb31d5b
51 changed files with 301 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1249
virtio-devices/src/vsock/csm/connection.rs Normal file
View File

File diff suppressed because it is too large Load Diff

129
virtio-devices/src/vsock/csm/mod.rs Normal file
View File

@@ -0,0 +1,129 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
/// This module implements our vsock connection state machine. The heavy lifting is done by
/// `connection::VsockConnection`, while this file only defines some constants and helper structs.
///
mod connection;
mod txbuf;
pub use connection::VsockConnection;
pub mod defs {
/// Vsock connection TX buffer capacity.
pub const CONN_TX_BUF_SIZE: u32 = 64 * 1024;
/// When the guest thinks we have less than this amount of free buffer space,
/// we will send them a credit update packet.
pub const CONN_CREDIT_UPDATE_THRESHOLD: u32 = 4 * 1024;
/// Connection request timeout, in millis.
pub const CONN_REQUEST_TIMEOUT_MS: u64 = 2000;
/// Connection graceful shutdown timeout, in millis.
pub const CONN_SHUTDOWN_TIMEOUT_MS: u64 = 2000;
}
#[derive(Debug)]
pub enum Error {
/// Attempted to push data to a full TX buffer.
TxBufFull,
/// An I/O error occurred, when attempting to flush the connection TX buffer.
TxBufFlush(std::io::Error),
/// An I/O error occurred, when attempting to write data to the host-side stream.
StreamWrite(std::io::Error),
}
type Result<T> = std::result::Result<T, Error>;
/// A vsock connection state.
///
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum ConnState {
/// The connection has been initiated by the host end, but is yet to be confirmed by the guest.
LocalInit,
/// The connection has been initiated by the guest, but we are yet to confirm it, by sending
/// a response packet (VSOCK_OP_RESPONSE).
PeerInit,
/// The connection handshake has been performed successfully, and data can now be exchanged.
Established,
/// The host (AF_UNIX) socket was closed.
LocalClosed,
/// A VSOCK_OP_SHUTDOWN packet was received from the guest. The tuple represents the guest R/W
/// indication: (will_not_recv_anymore_data, will_not_send_anymore_data).
PeerClosed(bool, bool),
/// The connection is scheduled to be forcefully terminated as soon as possible.
Killed,
}
/// An RX indication, used by `VsockConnection` to schedule future `recv_pkt()` responses.
/// For instance, after being notified that there is available data to be read from the host stream
/// (via `notify()`), the connection will store a `PendingRx::Rw` to be later inspected by
/// `recv_pkt()`.
///
#[derive(Clone, Copy, PartialEq)]
enum PendingRx {
/// We need to yield a connection request packet (VSOCK_OP_REQUEST).
Request = 0,
/// We need to yield a connection response packet (VSOCK_OP_RESPONSE).
Response = 1,
/// We need to yield a forceful connection termination packet (VSOCK_OP_RST).
Rst = 2,
/// We need to yield a data packet (VSOCK_OP_RW), by reading from the AF_UNIX socket.
Rw = 3,
/// We need to yield a credit update packet (VSOCK_OP_CREDIT_UPDATE).
CreditUpdate = 4,
}
impl PendingRx {
/// Transform the enum value into a bitmask, that can be used for set operations.
///
fn into_mask(self) -> u16 {
1u16 << (self as u16)
}
}
/// A set of RX indications (`PendingRx` items).
///
struct PendingRxSet {
data: u16,
}
impl PendingRxSet {
/// Insert an item into the set.
///
fn insert(&mut self, it: PendingRx) {
self.data |= it.into_mask();
}
/// Remove an item from the set and return:
/// - true, if the item was in the set; or
/// - false, if the item wasn't in the set.
///
fn remove(&mut self, it: PendingRx) -> bool {
let ret = self.contains(it);
self.data &= !it.into_mask();
ret
}
/// Check if an item is present in this set.
///
fn contains(&self, it: PendingRx) -> bool {
self.data & it.into_mask() != 0
}
/// Check if the set is empty.
///
fn is_empty(&self) -> bool {
self.data == 0
}
}
/// Create a set containing only one item.
///
impl From<PendingRx> for PendingRxSet {
fn from(it: PendingRx) -> Self {
Self {
data: it.into_mask(),
}
}
}

283
virtio-devices/src/vsock/csm/txbuf.rs Normal file
View File

@@ -0,0 +1,283 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
use std::io::Write;
use std::mem;
use std::num::Wrapping;
use super::defs;
use super::{Error, Result};
/// 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.
///
pub struct TxBuf {
/// The actual u8 buffer - only allocated after the first push.
data: Option<Box<[u8; Self::SIZE]>>,
/// Ring-buffer head offset - where new data is pushed to.
head: Wrapping<u32>,
/// Ring-buffer tail offset - where data is flushed from.
tail: Wrapping<u32>,
}
impl TxBuf {
/// Total buffer size, in bytes.
///
const SIZE: usize = defs::CONN_TX_BUF_SIZE as usize;
/// Ring-buffer constructor.
///
pub fn new() -> Self {
Self {
data: None,
head: Wrapping(0),
tail: Wrapping(0),
}
}
/// Get the used length of this buffer - number of bytes that have been pushed in, but not
/// yet flushed out.
///
pub fn len(&self) -> usize {
(self.head - self.tail).0 as usize
}
/// Push a byte slice onto 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.
///
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 {
return Err(Error::TxBufFull);
}
// We're using a closure here to return the boxed slice, instead of a value (i.e.
// `get_or_insert_with()` instead of `get_or_insert()`), because we only want the box
// created when `self.data` is None. If we were to use `get_or_insert(box)`, the box
// argument would always get evaluated (which implies a heap allocation), even though
// it would later be discarded (when `self.data.is_some()`). Apparently, clippy fails
// to see this, and insists on issuing some warning.
#[allow(clippy::redundant_closure)]
let data = self.data.get_or_insert_with(||
// Using uninitialized memory here is quite safe, since we never read from any
// area of the buffer before writing to it. First we push, then we flush only
// what had been prviously pushed.
Box::new(unsafe {mem::MaybeUninit::<[u8; Self::SIZE]>::uninit().assume_init()}));
// 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
// 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]);
// If the slice 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..]);
}
// Either way, we've just pushed exactly `src.len()` bytes, so that's the amount by
// which the (wrapping) buffer head needs to move forward.
self.head += Wrapping(src.len() as u32);
Ok(())
}
/// Flush the contents of the ring-buffer to a writable stream.
///
/// Return the number of bytes that have been transferred out of the ring-buffer and into
/// the writable stream.
///
pub fn flush_to<W>(&mut self, sink: &mut W) -> Result<usize>
where
W: Write,
{
// Nothing to do, if this buffer holds no data.
if self.is_empty() {
return Ok(0);
}
// Buffer tail, as an offset into the buffer data slice.
let tail_ofs = self.tail.0 as usize % Self::SIZE;
// Flushing the buffer can take either one or two writes:
// - one write, if the tail doesn't need to wrap around to reach the head; or
// - two writes, if the tail would wrap around: tail to slice end, then slice end to
// head.
// First write length: the lesser of tail to slice end, or tail to head.
let len_to_write = std::cmp::min(Self::SIZE - tail_ofs, self.len());
// It's safe to unwrap here, since we've already checked if the buffer was empty.
let data = self.data.as_ref().unwrap();
// Issue the first write and absorb any `WouldBlock` error (we can just try again
// later).
let written = sink
.write(&data[tail_ofs..(tail_ofs + len_to_write)])
.map_err(Error::TxBufFlush)?;
// Move the buffer tail ahead by the amount (of bytes) we were able to flush out.
self.tail += Wrapping(written as u32);
// If we weren't able to flush out as much as we tried, there's no point in attempting
// our second write.
if written < len_to_write {
return Ok(written);
}
// Attempt our second write. This will return immediately if a second write isn't
// needed, since checking for an empty buffer is the first thing we do in this
// function.
//
// Interesting corner case: if we've already written some data in the first pass,
// and then the second write fails, we will consider the flush action a success
// and return the number of bytes written in the first pass.
Ok(written + self.flush_to(sink).unwrap_or(0))
}
/// Check if the buffer holds any data that hasn't yet been flushed out.
///
pub fn is_empty(&self) -> bool {
self.len() == 0
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Error as IoError;
use std::io::Result as IoResult;
use std::io::{ErrorKind, Write};
struct TestSink {
data: Vec<u8>,
err: Option<IoError>,
capacity: usize,
}
impl TestSink {
const DEFAULT_CAPACITY: usize = 2 * TxBuf::SIZE;
fn new() -> Self {
Self {
data: Vec::with_capacity(Self::DEFAULT_CAPACITY),
err: None,
capacity: Self::DEFAULT_CAPACITY,
}
}
}
impl Write for TestSink {
fn write(&mut self, src: &[u8]) -> IoResult<usize> {
if self.err.is_some() {
return Err(self.err.take().unwrap());
}
let len_to_push = std::cmp::min(self.capacity - self.data.len(), src.len());
self.data.extend_from_slice(&src[..len_to_push]);
Ok(len_to_push)
}
fn flush(&mut self) -> IoResult<()> {
Ok(())
}
}
impl TestSink {
fn clear(&mut self) {
self.data = Vec::with_capacity(self.capacity);
self.err = None;
}
fn set_err(&mut self, err: IoError) {
self.err = Some(err);
}
fn set_capacity(&mut self, capacity: usize) {
self.capacity = capacity;
if self.data.len() > self.capacity {
self.data.resize(self.capacity, 0);
}
}
}
#[test]
fn test_push_nowrap() {
let mut txbuf = TxBuf::new();
let mut sink = TestSink::new();
assert!(txbuf.is_empty());
assert!(txbuf.data.is_none());
txbuf.push(&[1, 2, 3, 4]).unwrap();
txbuf.push(&[5, 6, 7, 8]).unwrap();
txbuf.flush_to(&mut sink).unwrap();
assert_eq!(sink.data, [1, 2, 3, 4, 5, 6, 7, 8]);
}
#[test]
fn test_push_wrap() {
let mut txbuf = TxBuf::new();
let mut sink = TestSink::new();
let mut tmp: Vec<u8> = Vec::new();
tmp.resize(TxBuf::SIZE - 2, 0);
txbuf.push(tmp.as_slice()).unwrap();
txbuf.flush_to(&mut sink).unwrap();
sink.clear();
txbuf.push(&[1, 2, 3, 4]).unwrap();
assert_eq!(txbuf.flush_to(&mut sink).unwrap(), 4);
assert_eq!(sink.data, [1, 2, 3, 4]);
}
#[test]
fn test_push_error() {
let mut txbuf = TxBuf::new();
let mut tmp = Vec::with_capacity(TxBuf::SIZE);
tmp.resize(TxBuf::SIZE - 1, 0);
txbuf.push(tmp.as_slice()).unwrap();
match txbuf.push(&[1, 2]) {
Err(Error::TxBufFull) => (),
other => panic!("Unexpected result: {:?}", other),
}
}
#[test]
fn test_incomplete_flush() {
let mut txbuf = TxBuf::new();
let mut sink = TestSink::new();
sink.set_capacity(2);
txbuf.push(&[1, 2, 3, 4]).unwrap();
assert_eq!(txbuf.flush_to(&mut sink).unwrap(), 2);
assert_eq!(txbuf.len(), 2);
assert_eq!(sink.data, [1, 2]);
sink.set_capacity(4);
assert_eq!(txbuf.flush_to(&mut sink).unwrap(), 2);
assert!(txbuf.is_empty());
assert_eq!(sink.data, [1, 2, 3, 4]);
}
#[test]
fn test_flush_error() {
const EACCESS: i32 = 13;
let mut txbuf = TxBuf::new();
let mut sink = TestSink::new();
txbuf.push(&[1, 2, 3, 4]).unwrap();
let io_err = IoError::from_raw_os_error(EACCESS);
sink.set_err(io_err);
match txbuf.flush_to(&mut sink) {
Err(Error::TxBufFlush(ref err)) if err.kind() == ErrorKind::PermissionDenied => (),
other => panic!("Unexpected result: {:?}", other),
}
}
}

1018
virtio-devices/src/vsock/device.rs Normal file
View File

File diff suppressed because it is too large Load Diff

361
virtio-devices/src/vsock/mod.rs Normal file
View File

@@ -0,0 +1,361 @@
// Copyright 2019 Intel Corporation. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
// Portions Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
// Portions Copyright 2017 The Chromium OS Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the THIRD-PARTY file.
mod csm;
mod device;
mod packet;
mod unix;
pub use self::device::Vsock;
pub use self::unix::VsockUnixBackend;
pub use self::unix::VsockUnixError;
pub use packet::VsockPacket;
use std::os::unix::io::RawFd;
mod defs {
/// Max vsock packet data/buffer size.
pub const MAX_PKT_BUF_SIZE: usize = 64 * 1024;
pub mod uapi {
/// Vsock packet operation IDs.
/// Defined in `/include/uapi/linux/virtio_vsock.h`.
///
/// Connection request.
pub const VSOCK_OP_REQUEST: u16 = 1;
/// Connection response.
pub const VSOCK_OP_RESPONSE: u16 = 2;
/// Connection reset.
pub const VSOCK_OP_RST: u16 = 3;
/// Connection clean shutdown.
pub const VSOCK_OP_SHUTDOWN: u16 = 4;
/// Connection data (read/write).
pub const VSOCK_OP_RW: u16 = 5;
/// Flow control credit update.
pub const VSOCK_OP_CREDIT_UPDATE: u16 = 6;
/// Flow control credit update request.
pub const VSOCK_OP_CREDIT_REQUEST: u16 = 7;
/// Vsock packet flags.
/// Defined in `/include/uapi/linux/virtio_vsock.h`.
///
/// Valid with a VSOCK_OP_SHUTDOWN packet: the packet sender will receive no more data.
pub const VSOCK_FLAGS_SHUTDOWN_RCV: u32 = 1;
/// Valid with a VSOCK_OP_SHUTDOWN packet: the packet sender will send no more data.
pub const VSOCK_FLAGS_SHUTDOWN_SEND: u32 = 2;
/// Vsock packet type.
/// Defined in `/include/uapi/linux/virtio_vsock.h`.
///
/// Stream / connection-oriented packet (the only currently valid type).
pub const VSOCK_TYPE_STREAM: u16 = 1;
pub const VSOCK_HOST_CID: u64 = 2;
}
}
#[derive(Debug)]
pub enum VsockError {
/// The vsock data/buffer virtio descriptor length is smaller than expected.
BufDescTooSmall,
/// The vsock data/buffer virtio descriptor is expected, but missing.
BufDescMissing,
/// Chained GuestMemory error.
GuestMemory,
/// Bounds check failed on guest memory pointer.
GuestMemoryBounds,
/// The vsock header descriptor length is too small.
HdrDescTooSmall(u32),
/// The vsock header `len` field holds an invalid value.
InvalidPktLen(u32),
/// A data fetch was attempted when no data was available.
NoData,
/// A data buffer was expected for the provided packet, but it is missing.
PktBufMissing,
/// Encountered an unexpected write-only virtio descriptor.
UnreadableDescriptor,
/// Encountered an unexpected read-only virtio descriptor.
UnwritableDescriptor,
}
type Result<T> = std::result::Result<T, VsockError>;
#[derive(Debug)]
pub enum VsockEpollHandlerError {
/// The vsock data/buffer virtio descriptor length is smaller than expected.
BufDescTooSmall,
/// The vsock data/buffer virtio descriptor is expected, but missing.
BufDescMissing,
/// Chained GuestMemory error.
GuestMemory,
/// Bounds check failed on guest memory pointer.
GuestMemoryBounds,
/// The vsock header descriptor length is too small.
HdrDescTooSmall(u32),
/// The vsock header `len` field holds an invalid value.
InvalidPktLen(u32),
/// A data fetch was attempted when no data was available.
NoData,
/// A data buffer was expected for the provided packet, but it is missing.
PktBufMissing,
/// Encountered an unexpected write-only virtio descriptor.
UnreadableDescriptor,
/// Encountered an unexpected read-only virtio descriptor.
UnwritableDescriptor,
}
/// A passive, event-driven object, that needs to be notified whenever an epoll-able event occurs.
/// An event-polling control loop will use `get_polled_fd()` and `get_polled_evset()` to query
/// the listener for the file descriptor and the set of events it's interested in. When such an
/// event occurs, the control loop will route the event to the listener via `notify()`.
///
pub trait VsockEpollListener {
/// Get the file descriptor the listener needs polled.
fn get_polled_fd(&self) -> RawFd;
/// Get the set of events for which the listener wants to be notified.
fn get_polled_evset(&self) -> epoll::Events;
/// Notify the listener that one ore more events have occurred.
fn notify(&mut self, evset: epoll::Events);
}
/// Any channel that handles vsock packet traffic: sending and receiving packets. Since we're
/// implementing the device model here, our responsibility is to always process the sending of
/// packets (i.e. the TX queue). So, any locally generated data, addressed to the driver (e.g.
/// a connection response or RST), will have to be queued, until we get to processing the RX queue.
///
/// Note: `recv_pkt()` and `send_pkt()` are named analogous to `Read::read()` and `Write::write()`,
/// respectively. I.e.
/// - `recv_pkt(&mut pkt)` will read data from the channel, and place it into `pkt`; and
/// - `send_pkt(&pkt)` will fetch data from `pkt`, and place it into the channel.
pub trait VsockChannel {
/// Read/receive an incoming packet from the channel.
fn recv_pkt(&mut self, pkt: &mut VsockPacket) -> Result<()>;
/// Write/send a packet through the channel.
fn send_pkt(&mut self, pkt: &VsockPacket) -> Result<()>;
/// Checks whether there is pending incoming data inside the channel, meaning that a subsequent
/// call to `recv_pkt()` won't fail.
fn has_pending_rx(&self) -> bool;
}
/// The vsock backend, which is basically an epoll-event-driven vsock channel, that needs to be
/// sendable through a mpsc channel (the latter due to how `vmm::EpollContext` works).
/// Currently, the only implementation we have is `crate::virtio::unix::muxer::VsockMuxer`, which
/// translates guest-side vsock connections to host-side Unix domain socket connections.
pub trait VsockBackend: VsockChannel + VsockEpollListener + Send {}
#[cfg(test)]
mod tests {
use super::device::{VsockEpollHandler, RX_QUEUE_EVENT, TX_QUEUE_EVENT};
use super::packet::VSOCK_PKT_HDR_SIZE;
use super::*;
use crate::device::{VirtioInterrupt, VirtioInterruptType};
use libc::EFD_NONBLOCK;
use std::os::unix::io::AsRawFd;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, RwLock};
use vm_memory::{GuestAddress, GuestMemoryAtomic, GuestMemoryMmap};
use vm_virtio::queue::testing::VirtQueue as GuestQ;
use vm_virtio::queue::Queue;
use vm_virtio::queue::{VIRTQ_DESC_F_NEXT, VIRTQ_DESC_F_WRITE};
use vmm_sys_util::eventfd::EventFd;
pub struct NoopVirtioInterrupt {}
impl VirtioInterrupt for NoopVirtioInterrupt {
fn trigger(
&self,
_int_type: &VirtioInterruptType,
_queue: Option<&Queue>,
) -> std::result::Result<(), std::io::Error> {
Ok(())
}
}
pub struct TestBackend {
pub evfd: EventFd,
pub rx_err: Option<VsockError>,
pub tx_err: Option<VsockError>,
pub pending_rx: bool,
pub rx_ok_cnt: usize,
pub tx_ok_cnt: usize,
pub evset: Option<epoll::Events>,
}
impl TestBackend {
pub fn new() -> Self {
Self {
evfd: EventFd::new(EFD_NONBLOCK).unwrap(),
rx_err: None,
tx_err: None,
pending_rx: false,
rx_ok_cnt: 0,
tx_ok_cnt: 0,
evset: None,
}
}
pub fn set_rx_err(&mut self, err: Option<VsockError>) {
self.rx_err = err;
}
pub fn set_tx_err(&mut self, err: Option<VsockError>) {
self.tx_err = err;
}
pub fn set_pending_rx(&mut self, prx: bool) {
self.pending_rx = prx;
}
}
impl VsockChannel for TestBackend {
fn recv_pkt(&mut self, _pkt: &mut VsockPacket) -> Result<()> {
match self.rx_err.take() {
None => {
self.rx_ok_cnt += 1;
Ok(())
}
Some(e) => Err(e),
}
}
fn send_pkt(&mut self, _pkt: &VsockPacket) -> Result<()> {
match self.tx_err.take() {
None => {
self.tx_ok_cnt += 1;
Ok(())
}
Some(e) => Err(e),
}
}
fn has_pending_rx(&self) -> bool {
self.pending_rx
}
}
impl VsockEpollListener for TestBackend {
fn get_polled_fd(&self) -> RawFd {
self.evfd.as_raw_fd()
}
fn get_polled_evset(&self) -> epoll::Events {
epoll::Events::EPOLLIN
}
fn notify(&mut self, evset: epoll::Events) {
self.evset = Some(evset);
}
}
impl VsockBackend for TestBackend {}
pub struct TestContext {
pub cid: u64,
pub mem: GuestMemoryMmap,
pub mem_size: usize,
pub device: Vsock<TestBackend>,
}
impl TestContext {
pub fn new() -> Self {
const CID: u64 = 52;
const MEM_SIZE: usize = 1024 * 1024 * 128;
Self {
cid: CID,
mem: GuestMemoryMmap::from_ranges(&[(GuestAddress(0), MEM_SIZE)]).unwrap(),
mem_size: MEM_SIZE,
device: Vsock::new(
String::from("vsock"),
CID,
PathBuf::from("/test/sock"),
TestBackend::new(),
false,
)
.unwrap(),
}
}
pub fn create_epoll_handler_context(&self) -> EpollHandlerContext {
const QSIZE: u16 = 2;
let guest_rxvq = GuestQ::new(GuestAddress(0x0010_0000), &self.mem, QSIZE as u16);
let guest_txvq = GuestQ::new(GuestAddress(0x0020_0000), &self.mem, QSIZE as u16);
let guest_evvq = GuestQ::new(GuestAddress(0x0030_0000), &self.mem, QSIZE as u16);
let rxvq = guest_rxvq.create_queue();
let txvq = guest_txvq.create_queue();
let evvq = guest_evvq.create_queue();
// Set up one available descriptor in the RX queue.
guest_rxvq.dtable[0].set(
0x0040_0000,
VSOCK_PKT_HDR_SIZE as u32,
VIRTQ_DESC_F_WRITE | VIRTQ_DESC_F_NEXT,
1,
);
guest_rxvq.dtable[1].set(0x0040_1000, 4096, VIRTQ_DESC_F_WRITE, 0);
guest_rxvq.avail.ring[0].set(0);
guest_rxvq.avail.idx.set(1);
// Set up one available descriptor in the TX queue.
guest_txvq.dtable[0].set(0x0050_0000, VSOCK_PKT_HDR_SIZE as u32, VIRTQ_DESC_F_NEXT, 1);
guest_txvq.dtable[1].set(0x0050_1000, 4096, 0, 0);
guest_txvq.avail.ring[0].set(0);
guest_txvq.avail.idx.set(1);
let queues = vec![rxvq, txvq, evvq];
let queue_evts = vec![
EventFd::new(EFD_NONBLOCK).unwrap(),
EventFd::new(EFD_NONBLOCK).unwrap(),
EventFd::new(EFD_NONBLOCK).unwrap(),
];
let interrupt_cb = Arc::new(NoopVirtioInterrupt {});
EpollHandlerContext {
guest_rxvq,
guest_txvq,
guest_evvq,
handler: VsockEpollHandler {
mem: GuestMemoryAtomic::new(self.mem.clone()),
queues,
queue_evts,
kill_evt: EventFd::new(EFD_NONBLOCK).unwrap(),
pause_evt: EventFd::new(EFD_NONBLOCK).unwrap(),
interrupt_cb,
backend: Arc::new(RwLock::new(TestBackend::new())),
},
}
}
}
pub struct EpollHandlerContext<'a> {
pub handler: VsockEpollHandler<TestBackend>,
pub guest_rxvq: GuestQ<'a>,
pub guest_txvq: GuestQ<'a>,
pub guest_evvq: GuestQ<'a>,
}
impl<'a> EpollHandlerContext<'a> {
pub fn signal_txq_event(&mut self) {
self.handler.queue_evts[1].write(1).unwrap();
self.handler
.handle_event(
TX_QUEUE_EVENT,
epoll::Events::EPOLLIN,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
}
pub fn signal_rxq_event(&mut self) {
self.handler.queue_evts[0].write(1).unwrap();
self.handler
.handle_event(
RX_QUEUE_EVENT,
epoll::Events::EPOLLIN,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
}
}
}

654
virtio-devices/src/vsock/packet.rs Normal file
View File

@@ -0,0 +1,654 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
/// `VsockPacket` provides a thin wrapper over the buffers exchanged via virtio queues.
/// There are two components to a vsock packet, each using its own descriptor in a
/// virtio queue:
/// - the packet header; and
/// - the packet data/buffer.
/// There is a 1:1 relation between descriptor chains and packets: the first (chain head) holds
/// the header, and an optional second descriptor holds the data. The second descriptor is only
/// present for data packets (VSOCK_OP_RW).
///
/// `VsockPacket` wraps these two buffers and provides direct access to the data stored
/// in guest memory. This is done to avoid unnecessarily copying data from guest memory
/// to temporary buffers, before passing it on to the vsock backend.
///
use byteorder::{ByteOrder, LittleEndian};
use super::super::DescriptorChain;
use super::defs;
use super::{Result, VsockError};
use vfio_ioctls::get_host_address_range;
// The vsock packet header is defined by the C struct:
//
// ```C
// struct virtio_vsock_hdr {
// le64 src_cid;
// le64 dst_cid;
// le32 src_port;
// le32 dst_port;
// le32 len;
// le16 type;
// le16 op;
// le32 flags;
// le32 buf_alloc;
// le32 fwd_cnt;
// };
// ```
//
// This structed will occupy the buffer pointed to by the head descriptor. We'll be accessing it
// as a byte slice. To that end, we define below the offsets for each field struct, as well as the
// packed struct size, as a bunch of `usize` consts.
// Note that these offsets are only used privately by the `VsockPacket` struct, the public interface
// consisting of getter and setter methods, for each struct field, that will also handle the correct
// endianess.
/// The vsock packet header struct size (when packed).
pub const VSOCK_PKT_HDR_SIZE: usize = 44;
// Source CID.
const HDROFF_SRC_CID: usize = 0;
// Destination CID.
const HDROFF_DST_CID: usize = 8;
// Source port.
const HDROFF_SRC_PORT: usize = 16;
// Destination port.
const HDROFF_DST_PORT: usize = 20;
// Data length (in bytes) - may be 0, if there is no data buffer.
const HDROFF_LEN: usize = 24;
// Socket type. Currently, only connection-oriented streams are defined by the vsock protocol.
const HDROFF_TYPE: usize = 28;
// Operation ID - one of the VSOCK_OP_* values; e.g.
// - VSOCK_OP_RW: a data packet;
// - VSOCK_OP_REQUEST: connection request;
// - VSOCK_OP_RST: forcefull connection termination;
// etc (see `super::defs::uapi` for the full list).
const HDROFF_OP: usize = 30;
// Additional options (flags) associated with the current operation (`op`).
// Currently, only used with shutdown requests (VSOCK_OP_SHUTDOWN).
const HDROFF_FLAGS: usize = 32;
// Size (in bytes) of the packet sender receive buffer (for the connection to which this packet
// belongs).
const HDROFF_BUF_ALLOC: usize = 36;
// Number of bytes the sender has received and consumed (for the connection to which this packet
// belongs). For instance, for our Unix backend, this counter would be the total number of bytes
// we have successfully written to a backing Unix socket.
const HDROFF_FWD_CNT: usize = 40;
/// The vsock packet, implemented as a wrapper over a virtq descriptor chain:
/// - the chain head, holding the packet header; and
/// - (an optional) data/buffer descriptor, only present for data packets (VSOCK_OP_RW).
///
pub struct VsockPacket {
hdr: *mut u8,
buf: Option<*mut u8>,
buf_size: usize,
}
impl VsockPacket {
/// Create the packet wrapper from a TX virtq chain head.
///
/// The chain head is expected to hold valid packet header data. A following packet buffer
/// descriptor can optionally end the chain. Bounds and pointer checks are performed when
/// creating the wrapper.
///
pub fn from_tx_virtq_head(head: &DescriptorChain) -> Result<Self> {
// All buffers in the TX queue must be readable.
//
if head.is_write_only() {
return Err(VsockError::UnreadableDescriptor);
}
// The packet header should fit inside the head descriptor.
if head.len < VSOCK_PKT_HDR_SIZE as u32 {
return Err(VsockError::HdrDescTooSmall(head.len));
}
let mut pkt = Self {
hdr: get_host_address_range(head.mem, head.addr, VSOCK_PKT_HDR_SIZE)
.ok_or_else(|| VsockError::GuestMemory)? as *mut u8,
buf: None,
buf_size: 0,
};
// No point looking for a data/buffer descriptor, if the packet is zero-lengthed.
if pkt.is_empty() {
return Ok(pkt);
}
// Reject weirdly-sized packets.
//
if pkt.len() > defs::MAX_PKT_BUF_SIZE as u32 {
return Err(VsockError::InvalidPktLen(pkt.len()));
}
// If the packet header showed a non-zero length, there should be a data descriptor here.
let buf_desc = head.next_descriptor().ok_or(VsockError::BufDescMissing)?;
// TX data should be read-only.
if buf_desc.is_write_only() {
return Err(VsockError::UnreadableDescriptor);
}
// The data buffer should be large enough to fit the size of the data, as described by
// the header descriptor.
if buf_desc.len < pkt.len() {
return Err(VsockError::BufDescTooSmall);
}
pkt.buf_size = buf_desc.len as usize;
pkt.buf = Some(
get_host_address_range(buf_desc.mem, buf_desc.addr, pkt.buf_size)
.ok_or_else(|| VsockError::GuestMemory)? as *mut u8,
);
Ok(pkt)
}
/// Create the packet wrapper from an RX virtq chain head.
///
/// There must be two descriptors in the chain, both writable: a header descriptor and a data
/// descriptor. Bounds and pointer checks are performed when creating the wrapper.
///
pub fn from_rx_virtq_head(head: &DescriptorChain) -> Result<Self> {
// All RX buffers must be writable.
//
if !head.is_write_only() {
return Err(VsockError::UnwritableDescriptor);
}
// The packet header should fit inside the head descriptor.
if head.len < VSOCK_PKT_HDR_SIZE as u32 {
return Err(VsockError::HdrDescTooSmall(head.len));
}
// All RX descriptor chains should have a header and a data descriptor.
if !head.has_next() {
return Err(VsockError::BufDescMissing);
}
let buf_desc = head.next_descriptor().ok_or(VsockError::BufDescMissing)?;
let buf_size = buf_desc.len as usize;
Ok(Self {
hdr: get_host_address_range(head.mem, head.addr, VSOCK_PKT_HDR_SIZE)
.ok_or_else(|| VsockError::GuestMemory)? as *mut u8,
buf: Some(
get_host_address_range(buf_desc.mem, buf_desc.addr, buf_size)
.ok_or_else(|| VsockError::GuestMemory)? as *mut u8,
),
buf_size,
})
}
/// Provides in-place, byte-slice, access to the vsock packet header.
///
pub fn hdr(&self) -> &[u8] {
// This is safe since bound checks have already been performed when creating the packet
// from the virtq descriptor.
unsafe { std::slice::from_raw_parts(self.hdr as *const u8, VSOCK_PKT_HDR_SIZE) }
}
/// Provides in-place, byte-slice, mutable access to the vsock packet header.
///
pub fn hdr_mut(&mut self) -> &mut [u8] {
// This is safe since bound checks have already been performed when creating the packet
// from the virtq descriptor.
unsafe { std::slice::from_raw_parts_mut(self.hdr, VSOCK_PKT_HDR_SIZE) }
}
/// 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]> {
self.buf.map(|ptr| {
// This is safe since bound checks have already been performed when creating the packet
// from the virtq descriptor.
unsafe { std::slice::from_raw_parts(ptr as *const u8, self.buf_size) }
})
}
/// 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]> {
self.buf.map(|ptr| {
// This is safe since bound checks have already been performed when creating the packet
// from the virtq descriptor.
unsafe { std::slice::from_raw_parts_mut(ptr, self.buf_size) }
})
}
pub fn src_cid(&self) -> u64 {
LittleEndian::read_u64(&self.hdr()[HDROFF_SRC_CID..])
}
pub fn set_src_cid(&mut self, cid: u64) -> &mut Self {
LittleEndian::write_u64(&mut self.hdr_mut()[HDROFF_SRC_CID..], cid);
self
}
pub fn dst_cid(&self) -> u64 {
LittleEndian::read_u64(&self.hdr()[HDROFF_DST_CID..])
}
pub fn set_dst_cid(&mut self, cid: u64) -> &mut Self {
LittleEndian::write_u64(&mut self.hdr_mut()[HDROFF_DST_CID..], cid);
self
}
pub fn src_port(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_SRC_PORT..])
}
pub fn set_src_port(&mut self, port: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_SRC_PORT..], port);
self
}
pub fn dst_port(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_DST_PORT..])
}
pub fn set_dst_port(&mut self, port: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_DST_PORT..], port);
self
}
pub fn len(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_LEN..])
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn set_len(&mut self, len: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_LEN..], len);
self
}
pub fn type_(&self) -> u16 {
LittleEndian::read_u16(&self.hdr()[HDROFF_TYPE..])
}
pub fn set_type(&mut self, type_: u16) -> &mut Self {
LittleEndian::write_u16(&mut self.hdr_mut()[HDROFF_TYPE..], type_);
self
}
pub fn op(&self) -> u16 {
LittleEndian::read_u16(&self.hdr()[HDROFF_OP..])
}
pub fn set_op(&mut self, op: u16) -> &mut Self {
LittleEndian::write_u16(&mut self.hdr_mut()[HDROFF_OP..], op);
self
}
pub fn flags(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_FLAGS..])
}
pub fn set_flags(&mut self, flags: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_FLAGS..], flags);
self
}
pub fn set_flag(&mut self, flag: u32) -> &mut Self {
self.set_flags(self.flags() | flag);
self
}
pub fn buf_alloc(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_BUF_ALLOC..])
}
pub fn set_buf_alloc(&mut self, buf_alloc: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_BUF_ALLOC..], buf_alloc);
self
}
pub fn fwd_cnt(&self) -> u32 {
LittleEndian::read_u32(&self.hdr()[HDROFF_FWD_CNT..])
}
pub fn set_fwd_cnt(&mut self, fwd_cnt: u32) -> &mut Self {
LittleEndian::write_u32(&mut self.hdr_mut()[HDROFF_FWD_CNT..], fwd_cnt);
self
}
}
#[cfg(test)]
mod tests {
use super::super::tests::TestContext;
use super::*;
use crate::vsock::defs::MAX_PKT_BUF_SIZE;
use vm_memory::{GuestAddress, GuestMemoryMmap};
use vm_virtio::queue::testing::VirtqDesc as GuestQDesc;
use vm_virtio::queue::VIRTQ_DESC_F_WRITE;
macro_rules! create_context {
($test_ctx:ident, $handler_ctx:ident) => {
let $test_ctx = TestContext::new();
let mut $handler_ctx = $test_ctx.create_epoll_handler_context();
// For TX packets, hdr.len should be set to a valid value.
set_pkt_len(1024, &$handler_ctx.guest_txvq.dtable[0], &$test_ctx.mem);
};
}
macro_rules! expect_asm_error {
(tx, $test_ctx:expr, $handler_ctx:expr, $err:pat) => {
expect_asm_error!($test_ctx, $handler_ctx, $err, from_tx_virtq_head, 1);
};
(rx, $test_ctx:expr, $handler_ctx:expr, $err:pat) => {
expect_asm_error!($test_ctx, $handler_ctx, $err, from_rx_virtq_head, 0);
};
($test_ctx:expr, $handler_ctx:expr, $err:pat, $ctor:ident, $vq:expr) => {
match VsockPacket::$ctor(
&$handler_ctx.handler.queues[$vq]
.iter(&$test_ctx.mem)
.next()
.unwrap(),
) {
Err($err) => (),
Ok(_) => panic!("Packet assembly should've failed!"),
Err(other) => panic!("Packet assembly failed with: {:?}", other),
}
};
}
fn set_pkt_len(len: u32, guest_desc: &GuestQDesc, mem: &GuestMemoryMmap) {
let hdr_gpa = guest_desc.addr.get();
let hdr_ptr = get_host_address_range(mem, GuestAddress(hdr_gpa), VSOCK_PKT_HDR_SIZE)
.unwrap() as *mut u8;
let len_ptr = unsafe { hdr_ptr.add(HDROFF_LEN) };
LittleEndian::write_u32(unsafe { std::slice::from_raw_parts_mut(len_ptr, 4) }, len);
}
#[test]
#[allow(clippy::cognitive_complexity)]
fn test_tx_packet_assembly() {
// Test case: successful TX packet assembly.
{
create_context!(test_ctx, handler_ctx);
let pkt = VsockPacket::from_tx_virtq_head(
&handler_ctx.handler.queues[1]
.iter(&test_ctx.mem)
.next()
.unwrap(),
)
.unwrap();
assert_eq!(pkt.hdr().len(), VSOCK_PKT_HDR_SIZE);
assert_eq!(
pkt.buf().unwrap().len(),
handler_ctx.guest_txvq.dtable[1].len.get() as usize
);
}
// Test case: error on write-only hdr descriptor.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_txvq.dtable[0]
.flags
.set(VIRTQ_DESC_F_WRITE);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::UnreadableDescriptor);
}
// Test case: header descriptor has insufficient space to hold the packet header.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_txvq.dtable[0]
.len
.set(VSOCK_PKT_HDR_SIZE as u32 - 1);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::HdrDescTooSmall(_));
}
// Test case: zero-length TX packet.
{
create_context!(test_ctx, handler_ctx);
set_pkt_len(0, &handler_ctx.guest_txvq.dtable[0], &test_ctx.mem);
let mut pkt = VsockPacket::from_tx_virtq_head(
&handler_ctx.handler.queues[1]
.iter(&test_ctx.mem)
.next()
.unwrap(),
)
.unwrap();
assert!(pkt.buf().is_none());
assert!(pkt.buf_mut().is_none());
}
// Test case: TX packet has more data than we can handle.
{
create_context!(test_ctx, handler_ctx);
set_pkt_len(
MAX_PKT_BUF_SIZE as u32 + 1,
&handler_ctx.guest_txvq.dtable[0],
&test_ctx.mem,
);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::InvalidPktLen(_));
}
// Test case:
// - packet header advertises some data length; and
// - the data descriptor is missing.
{
create_context!(test_ctx, handler_ctx);
set_pkt_len(1024, &handler_ctx.guest_txvq.dtable[0], &test_ctx.mem);
handler_ctx.guest_txvq.dtable[0].flags.set(0);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::BufDescMissing);
}
// Test case: error on write-only buf descriptor.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_txvq.dtable[1]
.flags
.set(VIRTQ_DESC_F_WRITE);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::UnreadableDescriptor);
}
// Test case: the buffer descriptor cannot fit all the data advertised by the the
// packet header `len` field.
{
create_context!(test_ctx, handler_ctx);
set_pkt_len(8 * 1024, &handler_ctx.guest_txvq.dtable[0], &test_ctx.mem);
handler_ctx.guest_txvq.dtable[1].len.set(4 * 1024);
expect_asm_error!(tx, test_ctx, handler_ctx, VsockError::BufDescTooSmall);
}
}
#[test]
fn test_rx_packet_assembly() {
// Test case: successful RX packet assembly.
{
create_context!(test_ctx, handler_ctx);
let pkt = VsockPacket::from_rx_virtq_head(
&handler_ctx.handler.queues[0]
.iter(&test_ctx.mem)
.next()
.unwrap(),
)
.unwrap();
assert_eq!(pkt.hdr().len(), VSOCK_PKT_HDR_SIZE);
assert_eq!(
pkt.buf().unwrap().len(),
handler_ctx.guest_rxvq.dtable[1].len.get() as usize
);
}
// Test case: read-only RX packet header.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_rxvq.dtable[0].flags.set(0);
expect_asm_error!(rx, test_ctx, handler_ctx, VsockError::UnwritableDescriptor);
}
// Test case: RX descriptor head cannot fit the entire packet header.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_rxvq.dtable[0]
.len
.set(VSOCK_PKT_HDR_SIZE as u32 - 1);
expect_asm_error!(rx, test_ctx, handler_ctx, VsockError::HdrDescTooSmall(_));
}
// Test case: RX descriptor chain is missing the packet buffer descriptor.
{
create_context!(test_ctx, handler_ctx);
handler_ctx.guest_rxvq.dtable[0]
.flags
.set(VIRTQ_DESC_F_WRITE);
expect_asm_error!(rx, test_ctx, handler_ctx, VsockError::BufDescMissing);
}
}
#[test]
#[allow(clippy::cognitive_complexity)]
fn test_packet_hdr_accessors() {
const SRC_CID: u64 = 1;
const DST_CID: u64 = 2;
const SRC_PORT: u32 = 3;
const DST_PORT: u32 = 4;
const LEN: u32 = 5;
const TYPE: u16 = 6;
const OP: u16 = 7;
const FLAGS: u32 = 8;
const BUF_ALLOC: u32 = 9;
const FWD_CNT: u32 = 10;
create_context!(test_ctx, handler_ctx);
let mut pkt = VsockPacket::from_rx_virtq_head(
&handler_ctx.handler.queues[0]
.iter(&test_ctx.mem)
.next()
.unwrap(),
)
.unwrap();
// Test field accessors.
pkt.set_src_cid(SRC_CID)
.set_dst_cid(DST_CID)
.set_src_port(SRC_PORT)
.set_dst_port(DST_PORT)
.set_len(LEN)
.set_type(TYPE)
.set_op(OP)
.set_flags(FLAGS)
.set_buf_alloc(BUF_ALLOC)
.set_fwd_cnt(FWD_CNT);
assert_eq!(pkt.src_cid(), SRC_CID);
assert_eq!(pkt.dst_cid(), DST_CID);
assert_eq!(pkt.src_port(), SRC_PORT);
assert_eq!(pkt.dst_port(), DST_PORT);
assert_eq!(pkt.len(), LEN);
assert_eq!(pkt.type_(), TYPE);
assert_eq!(pkt.op(), OP);
assert_eq!(pkt.flags(), FLAGS);
assert_eq!(pkt.buf_alloc(), BUF_ALLOC);
assert_eq!(pkt.fwd_cnt(), FWD_CNT);
// Test individual flag setting.
let flags = pkt.flags() | 0b1000;
pkt.set_flag(0b1000);
assert_eq!(pkt.flags(), flags);
// Test packet header as-slice access.
//
assert_eq!(pkt.hdr().len(), VSOCK_PKT_HDR_SIZE);
assert_eq!(
SRC_CID,
LittleEndian::read_u64(&pkt.hdr()[HDROFF_SRC_CID..])
);
assert_eq!(
DST_CID,
LittleEndian::read_u64(&pkt.hdr()[HDROFF_DST_CID..])
);
assert_eq!(
SRC_PORT,
LittleEndian::read_u32(&pkt.hdr()[HDROFF_SRC_PORT..])
);
assert_eq!(
DST_PORT,
LittleEndian::read_u32(&pkt.hdr()[HDROFF_DST_PORT..])
);
assert_eq!(LEN, LittleEndian::read_u32(&pkt.hdr()[HDROFF_LEN..]));
assert_eq!(TYPE, LittleEndian::read_u16(&pkt.hdr()[HDROFF_TYPE..]));
assert_eq!(OP, LittleEndian::read_u16(&pkt.hdr()[HDROFF_OP..]));
assert_eq!(FLAGS, LittleEndian::read_u32(&pkt.hdr()[HDROFF_FLAGS..]));
assert_eq!(
BUF_ALLOC,
LittleEndian::read_u32(&pkt.hdr()[HDROFF_BUF_ALLOC..])
);
assert_eq!(
FWD_CNT,
LittleEndian::read_u32(&pkt.hdr()[HDROFF_FWD_CNT..])
);
assert_eq!(pkt.hdr_mut().len(), VSOCK_PKT_HDR_SIZE);
for b in pkt.hdr_mut() {
*b = 0;
}
assert_eq!(pkt.src_cid(), 0);
assert_eq!(pkt.dst_cid(), 0);
assert_eq!(pkt.src_port(), 0);
assert_eq!(pkt.dst_port(), 0);
assert_eq!(pkt.len(), 0);
assert_eq!(pkt.type_(), 0);
assert_eq!(pkt.op(), 0);
assert_eq!(pkt.flags(), 0);
assert_eq!(pkt.buf_alloc(), 0);
assert_eq!(pkt.fwd_cnt(), 0);
}
#[test]
fn test_packet_buf() {
create_context!(test_ctx, handler_ctx);
let mut pkt = VsockPacket::from_rx_virtq_head(
&handler_ctx.handler.queues[0]
.iter(&test_ctx.mem)
.next()
.unwrap(),
)
.unwrap();
assert_eq!(
pkt.buf().unwrap().len(),
handler_ctx.guest_rxvq.dtable[1].len.get() as usize
);
assert_eq!(
pkt.buf_mut().unwrap().len(),
handler_ctx.guest_rxvq.dtable[1].len.get() as usize
);
for i in 0..pkt.buf().unwrap().len() {
pkt.buf_mut().unwrap()[i] = (i % 0x100) as u8;
assert_eq!(pkt.buf().unwrap()[i], (i % 0x100) as u8);
}
}
}

56
virtio-devices/src/vsock/unix/mod.rs Normal file
View File

@@ -0,0 +1,56 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
/// This module implements the Unix Domain Sockets backend for vsock - a mediator between
/// guest-side AF_VSOCK sockets and host-side AF_UNIX sockets. The heavy lifting is performed by
/// `muxer::VsockMuxer`, a connection multiplexer that uses `super::csm::VsockConnection` for
/// handling vsock connection states.
/// Check out `muxer.rs` for a more detailed explanation of the inner workings of this backend.
///
mod muxer;
mod muxer_killq;
mod muxer_rxq;
pub use muxer::VsockMuxer as VsockUnixBackend;
pub use Error as VsockUnixError;
mod defs {
/// Maximum number of established connections that we can handle.
pub const MAX_CONNECTIONS: usize = 1023;
/// Size of the muxer RX packet queue.
pub const MUXER_RXQ_SIZE: usize = 256;
/// Size of the muxer connection kill queue.
pub const MUXER_KILLQ_SIZE: usize = 128;
}
#[derive(Debug)]
pub enum Error {
/// Error converting from UTF-8
ConvertFromUTF8(std::str::Utf8Error),
/// Error registering a new epoll-listening FD.
EpollAdd(std::io::Error),
/// Error creating an epoll FD.
EpollFdCreate(std::io::Error),
/// The host made an invalid vsock port connection request.
InvalidPortRequest,
/// Error parsing integer.
ParseInteger(std::num::ParseIntError),
/// Error reading stream port.
ReadStreamPort(Box<Error>),
/// Error accepting a new connection from the host-side Unix socket.
UnixAccept(std::io::Error),
/// Error binding to the host-side Unix socket.
UnixBind(std::io::Error),
/// Error connecting to a host-side Unix socket.
UnixConnect(std::io::Error),
/// Error reading from host-side Unix socket.
UnixRead(std::io::Error),
/// Muxer connection limit reached.
TooManyConnections,
}
type Result<T> = std::result::Result<T, Error>;
type MuxerConnection = super::csm::VsockConnection<std::os::unix::net::UnixStream>;

1418
virtio-devices/src/vsock/unix/muxer.rs Normal file
View File

File diff suppressed because it is too large Load Diff

140
virtio-devices/src/vsock/unix/muxer_killq.rs Normal file
View File

@@ -0,0 +1,140 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
/// `MuxerKillQ` implements a helper object that `VsockMuxer` can use for scheduling forced
/// connection termination. I.e. after one peer issues a clean shutdown request
/// (VSOCK_OP_SHUTDOWN), the concerned connection is queued for termination (VSOCK_OP_RST) in
/// the near future (herein implemented via an expiring timer).
///
/// Whenever the muxer needs to schedule a connection for termination, it pushes it (or rather
/// an identifier - the connection key) to this queue. A subsequent pop() operation will
/// succeed if and only if the first connection in the queue is ready to be terminated (i.e.
/// its kill timer expired).
///
/// Without using this queue, the muxer would have to walk its entire connection pool
/// (hashmap), whenever it needs to check for expired kill timers. With this queue, both
/// scheduling and termination are performed in constant time. However, since we don't want to
/// waste space on a kill queue that's as big as the connection hashmap itself, it is possible
/// that this queue may become full at times. We call this kill queue "synchronized" if we are
/// certain that all connections that are awaiting termination are present in the queue. This
/// means a simple constant-time pop() operation is enough to check whether any connections
/// need to be terminated. When the kill queue becomes full, though, pushing fails, so
/// connections that should be terminated are left out. The queue is not synchronized anymore.
/// When that happens, the muxer will first drain the queue, and then replace it with a new
/// queue, created by walking the connection pool, looking for connections that will be
/// expiring in the future.
///
use std::collections::{HashMap, VecDeque};
use std::time::Instant;
use super::defs;
use super::muxer::ConnMapKey;
use super::MuxerConnection;
/// A kill queue item, holding the connection key and the scheduled time for termination.
///
#[derive(Clone, Copy)]
struct MuxerKillQItem {
key: ConnMapKey,
kill_time: Instant,
}
/// The connection kill queue: a FIFO structure, storing the connections that are scheduled for
/// termination.
///
pub struct MuxerKillQ {
/// The kill queue contents.
q: VecDeque<MuxerKillQItem>,
/// The kill queue sync status:
/// - when true, all connections that are awaiting termination are guaranteed to be in this
/// queue;
/// - when false, some connections may have been left out.
///
synced: bool,
}
impl MuxerKillQ {
const SIZE: usize = defs::MUXER_KILLQ_SIZE;
/// Trivial kill queue constructor.
///
pub fn new() -> Self {
Self {
q: VecDeque::with_capacity(Self::SIZE),
synced: true,
}
}
/// Create a kill queue by walking the connection pool, looking for connections that are
/// set to expire at some point in the future.
/// Note: if more than `Self::SIZE` connections are found, the queue will be created in an
/// out-of-sync state, and will be discarded after it is emptied.
///
pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, MuxerConnection>) -> Self {
let mut q_buf: Vec<MuxerKillQItem> = Vec::with_capacity(Self::SIZE);
let mut synced = true;
for (key, conn) in conn_map.iter() {
if !conn.will_expire() {
continue;
}
if q_buf.len() >= Self::SIZE {
synced = false;
break;
}
q_buf.push(MuxerKillQItem {
key: *key,
kill_time: conn.expiry().unwrap(),
});
}
q_buf.sort_unstable_by_key(|it| it.kill_time);
Self {
q: q_buf.into(),
synced,
}
}
/// Push a connection key to the queue, scheduling it for termination at
/// `CONN_SHUTDOWN_TIMEOUT_MS` from now (the push time).
///
pub fn push(&mut self, key: ConnMapKey, kill_time: Instant) {
if !self.is_synced() || self.is_full() {
self.synced = false;
return;
}
self.q.push_back(MuxerKillQItem { key, kill_time });
}
/// Attempt to pop an expired connection from the kill queue.
///
/// This will succeed and return a connection key, only if the connection at the front of
/// the queue has expired. Otherwise, `None` is returned.
///
pub fn pop(&mut self) -> Option<ConnMapKey> {
if let Some(item) = self.q.front() {
if Instant::now() > item.kill_time {
return Some(self.q.pop_front().unwrap().key);
}
}
None
}
/// Check if the kill queue is synchronized with the connection pool.
///
pub fn is_synced(&self) -> bool {
self.synced
}
/// Check if the kill queue is empty, obviously.
///
pub fn is_empty(&self) -> bool {
self.q.len() == 0
}
/// Check if the kill queue is full.
///
pub fn is_full(&self) -> bool {
self.q.len() == Self::SIZE
}
}

145
virtio-devices/src/vsock/unix/muxer_rxq.rs Normal file
View File

@@ -0,0 +1,145 @@
// Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
//
/// `MuxerRxQ` implements a helper object that `VsockMuxer` can use for queuing RX (host -> guest)
/// packets (or rather instructions on how to build said packets).
///
/// Under ideal operation, every connection, that has pending RX data, will be present in the muxer
/// RX queue. However, since the RX queue is smaller than the connection pool, it may, under some
/// conditions, become full, meaning that it can no longer account for all the connections that can
/// yield RX data. When that happens, we say that it is no longer "synchronized" (i.e. with the
/// connection pool). A desynchronized RX queue still holds valid data, and the muxer will
/// continue to pop packets from it. However, when a desynchronized queue is drained, additional
/// data may still be available, so the muxer will have to perform a more costly walk of the entire
/// connection pool to find it. This walk is performed here, as part of building an RX queue from
/// the connection pool. When an out-of-sync is drained, the muxer will discard it, and attempt to
/// rebuild a synced one.
///
use std::collections::{HashMap, VecDeque};
use super::super::VsockChannel;
use super::defs;
use super::muxer::{ConnMapKey, MuxerRx};
use super::MuxerConnection;
/// The muxer RX queue.
///
pub struct MuxerRxQ {
/// The RX queue data.
q: VecDeque<MuxerRx>,
/// The RX queue sync status.
synced: bool,
}
impl MuxerRxQ {
const SIZE: usize = defs::MUXER_RXQ_SIZE;
/// Trivial RX queue constructor.
///
pub fn new() -> Self {
Self {
q: VecDeque::with_capacity(Self::SIZE),
synced: true,
}
}
/// Attempt to build an RX queue, that is synchronized to the connection pool.
/// Note: the resulting queue may still be desynchronized, if there are too many connections
/// that have pending RX data. In that case, the muxer will first drain this queue, and
/// then try again to build a synchronized one.
///
pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, MuxerConnection>) -> Self {
let mut q = VecDeque::new();
let mut synced = true;
for (key, conn) in conn_map.iter() {
if !conn.has_pending_rx() {
continue;
}
if q.len() >= Self::SIZE {
synced = false;
break;
}
q.push_back(MuxerRx::ConnRx(*key));
}
Self { q, synced }
}
/// Push a new RX item to the queue.
///
/// A push will fail when:
/// - trying to push a connection key onto an out-of-sync, or full queue; or
/// - trying to push an RST onto a queue already full of RSTs.
/// RSTs take precedence over connections, because connections can always be queried for
/// pending RX data later. Aside from this queue, there is no other storage for RSTs, so
/// failing to push one means that we have to drop the packet.
///
/// Returns:
/// - `true` if the new item has been successfully queued; or
/// - `false` if there was no room left in the queue.
///
pub fn push(&mut self, rx: MuxerRx) -> bool {
// Pushing to a non-full, synchronized queue will always succeed.
if self.is_synced() && !self.is_full() {
self.q.push_back(rx);
return true;
}
match rx {
MuxerRx::RstPkt { .. } => {
// If we just failed to push an RST packet, we'll look through the queue, trying to
// find a connection key that we could evict. This way, the queue does lose sync,
// but we don't drop any packets.
for qi in self.q.iter_mut().rev() {
if let MuxerRx::ConnRx(_) = qi {
*qi = rx;
self.synced = false;
return true;
}
}
}
MuxerRx::ConnRx(_) => {
self.synced = false;
}
};
false
}
/// Peek into the front of the queue.
///
pub fn peek(&self) -> Option<MuxerRx> {
self.q.front().copied()
}
/// Pop an RX item from the front of the queue.
///
pub fn pop(&mut self) -> Option<MuxerRx> {
self.q.pop_front()
}
/// Check if the RX queue is synchronized with the connection pool.
///
pub fn is_synced(&self) -> bool {
self.synced
}
/// Get the total number of items in the queue.
///
pub fn len(&self) -> usize {
self.q.len()
}
/// Check if the queue is empty.
///
pub fn is_empty(&self) -> bool {
self.len() == 0
}
/// Check if the queue is full.
///
pub fn is_full(&self) -> bool {
self.len() == Self::SIZE
}
}