diff --git a/lib/propolis/src/hw/virtio/vsock.rs b/lib/propolis/src/hw/virtio/vsock.rs index 7bc595db8..5f87c908f 100644 --- a/lib/propolis/src/hw/virtio/vsock.rs +++ b/lib/propolis/src/hw/virtio/vsock.rs @@ -82,6 +82,18 @@ impl RxPermit<'_> { } } +#[derive(Debug, thiserror::Error)] +pub enum TransportResetError { + #[error("virt queue not alive")] + NotAlive, + #[error("mem access failed")] + MemAccess, + #[error("event queue has no usable descriptors")] + NoDescriptor, + #[error("descriptor chain buffer too small")] + ChainTooSmall, +} + pub struct VsockVq { queues: Vec>, acc_mem: MemAccessor, @@ -138,6 +150,37 @@ impl VsockVq { Some(packet) } + + /// Send a `VIRTIO_VSOCK_EVENT_TRANSPORT_RESET` to the guest on the + /// event queue. + /// + /// virtio spec 5.10.6.7, states that this must be sent upon guest + /// migration, so that the guest can remove all of its established + /// connections. + pub fn send_transport_reset(&mut self) -> Result<(), TransportResetError> { + let vq = self + .queues + .get(VSOCK_EVENT_QUEUE as usize) + .expect("has event queue"); + + if !vq.is_alive() { + return Err(TransportResetError::NotAlive); + } + + let mem = + self.acc_mem.access().ok_or(TransportResetError::MemAccess)?; + let mut avail = + vq.peek_avail(&mem).ok_or(TransportResetError::NoDescriptor)?; + + let event_id: u32 = VIRTIO_VSOCK_EVENT_TRANSPORT_RESET.to_le(); + if !avail.chain_mut().write(&event_id, &mem) { + avail.push_used(&mem); + return Err(TransportResetError::ChainTooSmall); + } + + avail.push_used(&mem); + Ok(()) + } } pub struct PciVirtioSock { @@ -160,7 +203,7 @@ impl PciVirtioSock { // VSOCK_TX_QUEUE VqSize::new(queue_size), // VSOCK_EVENT_QUEUE - VqSize::new(1), + VqSize::new(16), ]); // One for rx, tx, event @@ -259,9 +302,31 @@ impl Lifecycle for PciVirtioSock { self.backend.halt(); } fn migrate(&'_ self) -> Migrator<'_> { - // TODO (MTZ): - // We need to support migration propolis#1065 - Migrator::NonMigratable + Migrator::Multi(self) + } +} + +impl MigrateMulti for PciVirtioSock { + fn export( + &self, + output: &mut PayloadOutputs, + ctx: &MigrateCtx, + ) -> Result<(), MigrateStateError> { + ::export(self, output, ctx) + } + + fn import( + &self, + offer: &mut PayloadOffers, + ctx: &MigrateCtx, + ) -> Result<(), MigrateStateError> { + ::import(self, offer, ctx)?; + + // Upon import, our vsock poller has no state and the spec indicates + // that we need to tell the guest to reset its established connections. + self.backend.send_transport_reset(); + + Ok(()) } } @@ -289,7 +354,6 @@ mod bits { } } - #[allow(unused)] pub const VIRTIO_VSOCK_EVENT_TRANSPORT_RESET: u32 = 0; } use bits::*; @@ -369,3 +433,238 @@ impl VsockPacket { Ok(packet) } } + +#[cfg(test)] +mod test { + #[cfg(target_os = "illumos")] + mod illumos { + use std::net::TcpListener; + use std::sync::atomic::Ordering; + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use crate::hw::pci::test::Scaffold; + use crate::hw::pci::{BusLocation, Endpoint}; + use crate::hw::virtio::testutil::{TestVirtQueues, VqSize}; + use crate::hw::virtio::vsock::{ + PciVirtioSock, VIRTIO_VSOCK_EVENT_TRANSPORT_RESET, + VSOCK_EVENT_QUEUE, VSOCK_RX_QUEUE, VSOCK_TX_QUEUE, + }; + use crate::hw::virtio::VirtioDevice; + use crate::lifecycle::Lifecycle; + use crate::migrate::{ + MigrateCtx, MigrateMulti, PayloadOffer, PayloadOffers, + PayloadOutputs, + }; + use crate::vsock::packet::{ + VsockPacketHeader, VsockPacketOp, VsockSocketType, + }; + use crate::vsock::proxy::VsockPortMapping; + use crate::vsock::{GuestCid, VSOCK_HOST_CID}; + use zerocopy::{FromBytes, IntoBytes}; + + fn test_logger() -> slog::Logger { + slog::Logger::root(slog::Discard, slog::o!()) + } + + fn test_deadline() -> Instant { + Instant::now() + Duration::from_secs(15) + } + + #[test] + fn migrate_multi_round_trip() { + const SRC_PORT: u32 = 1234; + const DST_PORT: u32 = 3000; + let cid = GuestCid::try_from(716).unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mappings = vec![VsockPortMapping::new( + 3000, + listener.local_addr().unwrap(), + )]; + + let source = + PciVirtioSock::new(256, cid, test_logger(), mappings.clone()); + let target = PciVirtioSock::new(256, cid, test_logger(), mappings); + + let tvqs = TestVirtQueues::new(&[ + VqSize::new(256), + VqSize::new(256), + VqSize::new(16), + ]); + + // Scaffold sets up an orphan acc_mem, swap it for a child of the + // test memory's accessor tree. + let mut scaffold = Scaffold::new(); + scaffold.acc_mem = tvqs.mem_acc().child(Some("pci-bus".into())); + let bus = scaffold.create_bus(); + + // Source vsock device + bus.attach( + BusLocation::new(0, 0).unwrap(), + Arc::clone(&source) as Arc, + None, + ); + // Dest vsock device + bus.attach( + BusLocation::new(1, 0).unwrap(), + Arc::clone(&target) as Arc, + None, + ); + + // Setup the test vsock queues, but don't provide the event buffer. + for qid in [VSOCK_RX_QUEUE, VSOCK_EVENT_QUEUE, VSOCK_TX_QUEUE] { + let layout = tvqs.layout(qid as usize); + let vq = source + .virtio_state + .queues + .get(qid) + .expect("device has the queue"); + vq.map_virtqueue( + layout.desc_base, + layout.avail_base, + layout.used_base, + ); + vq.live.store(true, Ordering::Release); + vq.enabled.store(true, Ordering::Release); + } + let mut ev_writer = tvqs.writer(VSOCK_EVENT_QUEUE as usize, 0); + let mut tx_writer = tvqs.writer(VSOCK_TX_QUEUE as usize, 0x10000); + let mut rx_writer = tvqs.writer(VSOCK_RX_QUEUE as usize, 0x20000); + + let d_rx = rx_writer.add_writable(tvqs.mem_acc(), 256); + rx_writer.publish_avail(tvqs.mem_acc(), d_rx); + + let acc_mem = tvqs.mem_acc().access().expect("test memory"); + let ctx = MigrateCtx { mem: &acc_mem }; + + // Export guest state + let mut outputs = PayloadOutputs::new(); + MigrateMulti::export(source.as_ref(), &mut outputs, &ctx) + .expect("can export PciVirtioSock"); + + // Serialize the export payload into the import payload. + let mut offers = + PayloadOffers::new(outputs.into_iter().map(|output| { + let value = serde_json::to_value(&output.payload) + .expect("serializing payload output"); + PayloadOffer { + kind: output.kind, + version: output.version, + payload: Box::new( + ::erase(value), + ), + } + })); + + // Import guest state + MigrateMulti::import(target.as_ref(), &mut offers, &ctx) + .expect("can import PciVirtioSock"); + + // After import, the guest would be resumed + Lifecycle::resume(target.as_ref()); + + // Attempt to open a connection via REQUEST, which should not + // succeed yet, because our event queue has no buffers in it yet. + let mut hdr = VsockPacketHeader::new(); + hdr.set_src_cid(cid) + .set_dst_cid_raw(VSOCK_HOST_CID) + .set_src_port(SRC_PORT) + .set_dst_port(DST_PORT) + .set_len(0) + .set_socket_type(VsockSocketType::Stream) + .set_op(VsockPacketOp::Request) + .set_buf_alloc(65536) + .set_fwd_cnt(0); + let d_tx = tx_writer.add_readable(tvqs.mem_acc(), hdr.as_bytes()); + tx_writer.publish_avail(tvqs.mem_acc(), d_tx); + let tx_vq = target + .virtio_state + .queues + .get(VSOCK_TX_QUEUE) + .expect("device has a tx queue"); + VirtioDevice::queue_notify(target.as_ref(), tx_vq); + + std::thread::sleep(Duration::from_millis(100)); + assert_eq!( + tx_writer.used_idx(tvqs.mem_acc()), + 0, + "tx packet processed while the transport reset was needed" + ); + + // Pretend the guest updates its event queue with new buffers. A + // well behaved guest should have kept these available from the + // start. + let d_ev = ev_writer.add_writable(tvqs.mem_acc(), 4); + ev_writer.publish_avail(tvqs.mem_acc(), d_ev); + let ev_vq = target + .virtio_state + .queues + .get(VSOCK_EVENT_QUEUE) + .expect("device has an event queue"); + VirtioDevice::queue_notify(target.as_ref(), ev_vq); + + let deadline = test_deadline(); + while ev_writer.used_idx(tvqs.mem_acc()) < 1 { + assert!( + Instant::now() < deadline, + "timed out waiting for transport reset delivery" + ); + std::thread::sleep(Duration::from_millis(10)); + } + + let elem = ev_writer.read_used_elem(tvqs.mem_acc(), 0); + assert_eq!(elem.id, u32::from(d_ev)); + assert_eq!(elem.len, 4, "event is a single le32 id"); + let event = ev_writer.read_desc_data(tvqs.mem_acc(), d_ev, 4); + assert_eq!( + event, + VIRTIO_VSOCK_EVENT_TRANSPORT_RESET.to_le_bytes(), + "delivered event is VIRTIO_VSOCK_EVENT_TRANSPORT_RESET" + ); + + // Once we were able to send a VIRTIO_VSOCK_EVENT_TRANSPORT_RESET to + // the guest we should now see the REQUEST acknowledged. + let deadline = test_deadline(); + while tx_writer.used_idx(tvqs.mem_acc()) < 1 { + assert!( + Instant::now() < deadline, + "timed out waiting for the gated tx packet" + ); + std::thread::sleep(Duration::from_millis(10)); + } + let tx_elem = tx_writer.read_used_elem(tvqs.mem_acc(), 0); + assert_eq!(tx_elem.id, u32::from(d_tx)); + + let deadline = Instant::now() + Duration::from_secs(15); + while rx_writer.used_idx(tvqs.mem_acc()) < 1 { + assert!( + Instant::now() < deadline, + "timed out waiting for the handshake RESPONSE" + ); + std::thread::sleep(Duration::from_millis(10)); + } + let rx_elem = rx_writer.read_used_elem(tvqs.mem_acc(), 0); + assert_eq!(rx_elem.id, u32::from(d_rx)); + let buf = rx_writer.read_desc_data( + tvqs.mem_acc(), + d_rx, + rx_elem.len as usize, + ); + let resp_hdr = VsockPacketHeader::read_from_bytes( + &buf[..std::mem::size_of::()], + ) + .expect("rx buffer holds a packet header"); + assert_eq!(resp_hdr.op(), Some(VsockPacketOp::Response)); + assert_eq!(resp_hdr.src_cid(), VSOCK_HOST_CID); + assert_eq!(resp_hdr.dst_cid(), cid.get()); + assert_eq!(resp_hdr.src_port(), DST_PORT); + assert_eq!(resp_hdr.dst_port(), SRC_PORT); + + // Tear down the source and target vsock devices. + Lifecycle::pause(target.as_ref()); + Lifecycle::halt(source.as_ref()); + Lifecycle::halt(target.as_ref()); + } + } +} diff --git a/lib/propolis/src/vsock/poller.rs b/lib/propolis/src/vsock/poller.rs index ff70fd734..129d7d618 100644 --- a/lib/propolis/src/vsock/poller.rs +++ b/lib/propolis/src/vsock/poller.rs @@ -22,6 +22,7 @@ use nix::poll::PollFlags; use slog::{error, info, warn, Logger}; use crate::hw::virtio::vsock::VsockVq; +use crate::hw::virtio::vsock::VSOCK_EVENT_QUEUE; use crate::hw::virtio::vsock::VSOCK_RX_QUEUE; use crate::hw::virtio::vsock::VSOCK_TX_QUEUE; use crate::vsock::packet::VsockPacket; @@ -46,7 +47,8 @@ const DEFAULT_QUIESCE_TIMEOUT: Duration = Duration::from_secs(2); enum VsockEvent { TxQueue = 0, RxQueue = 1, - Pause = 2, + EventQueue = 2, + Pause = 3, } impl TryFrom for VsockEvent { @@ -56,7 +58,8 @@ impl TryFrom for VsockEvent { match value { 0 => Ok(Self::TxQueue), 1 => Ok(Self::RxQueue), - 2 => Ok(Self::Pause), + 2 => Ok(Self::EventQueue), + 3 => Ok(Self::Pause), unknown => Err(unknown), } } @@ -90,12 +93,22 @@ impl PollerState { } } +/// Whether the guest is needs a `VIRTIO_VSOCK_EVENT_TRANSPORT_RESET` or +/// not. +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +enum TransportResetState { + NotNeeded, + Needed, +} + /// Commands that can be sent to a paused `poller_loop`. enum PausedCmd { /// Continue execution Resume { oneshot: mpsc::SyncSender<()> }, /// Cleanup all connection state and queued packets Reset { oneshot: mpsc::SyncSender<()> }, + /// Tell the guest to drop its established connections + SendTransportReset { oneshot: mpsc::SyncSender<()> }, /// Shutdown the event-loop Halt, } @@ -127,6 +140,7 @@ impl VsockPollerNotify { match id { VSOCK_RX_QUEUE => self.port_send(VsockEvent::RxQueue), VSOCK_TX_QUEUE => self.port_send(VsockEvent::TxQueue), + VSOCK_EVENT_QUEUE => self.port_send(VsockEvent::EventQueue), _ => Ok(()), } } @@ -147,6 +161,14 @@ impl VsockPollerNotify { rx.recv().unwrap(); } + pub fn send_transport_reset(&self) { + let (tx, rx) = mpsc::sync_channel(0); + self.pause_tx + .send(PausedCmd::SendTransportReset { oneshot: tx }) + .unwrap(); + rx.recv().unwrap(); + } + pub fn halt_evloop(&self) { self.pause_tx.send(PausedCmd::Halt).unwrap(); } @@ -214,6 +236,8 @@ pub struct VsockPoller { rx_blocked: Vec, /// Connections waiting to be reaped quiescing: VecDeque, + /// Whether a `VIRTIO_VSOCK_EVENT_TRANSPORT_RESET` is owed to the guest. + transport_reset: TransportResetState, } impl VsockPoller { @@ -267,6 +291,7 @@ impl VsockPoller { rx: Default::default(), rx_blocked: Default::default(), quiescing: Default::default(), + transport_reset: TransportResetState::NotNeeded, }) } @@ -589,12 +614,36 @@ impl VsockPoller { } } + /// Attempt to deliver a transport reset to the guest. If succesful, update + /// the state and process any events we refused to handle before. + fn try_send_transport_reset(&mut self) { + if self.transport_reset != TransportResetState::Needed { + return; + } + match self.queues.send_transport_reset() { + Ok(()) => { + self.transport_reset = TransportResetState::NotNeeded; + self.handle_tx_queue_event(); + self.handle_rx_queue_event(); + } + Err(e) => error!(&self.log, "failed to send transport reset: {e}"), + } + } + /// Handle a user event. Returns `true` if the event loop should pause. fn handle_user_event(&mut self, event: VsockEvent) -> bool { let mut should_pause = false; + let needs_reset = + matches!(self.transport_reset, TransportResetState::Needed); + match event { + // Don't handle any guest->host events until the a transport reset + // has been delivered to the guest. We will handle any missed events + // once we have been successful. + VsockEvent::TxQueue if needs_reset => {} VsockEvent::TxQueue => self.handle_tx_queue_event(), VsockEvent::RxQueue => self.handle_rx_queue_event(), + VsockEvent::EventQueue => self.try_send_transport_reset(), VsockEvent::Pause => should_pause = true, } should_pause @@ -954,6 +1003,9 @@ impl VsockPoller { // We are running! self.state.set_running(); + // If we were migrated we need to send a transport reset. + self.try_send_transport_reset(); + // Handle events until we are told to pause. self.handle_events(); @@ -964,6 +1016,10 @@ impl VsockPoller { self.reset(); oneshot.send(()).unwrap(); } + Ok(PausedCmd::SendTransportReset { oneshot }) => { + self.transport_reset = TransportResetState::Needed; + oneshot.send(()).unwrap(); + } Ok(PausedCmd::Halt) => { self.reset(); return; @@ -984,6 +1040,7 @@ impl VsockPoller { self.rx.clear(); self.rx_blocked.clear(); self.quiescing.clear(); + self.transport_reset = TransportResetState::NotNeeded; } } @@ -1057,7 +1114,9 @@ mod test { use zerocopy::{FromBytes, IntoBytes}; use crate::hw::virtio::testutil::{QueueWriter, TestVirtQueues, VqSize}; - use crate::hw::virtio::vsock::{VsockVq, VSOCK_RX_QUEUE, VSOCK_TX_QUEUE}; + use crate::hw::virtio::vsock::{ + VsockVq, VSOCK_EVENT_QUEUE, VSOCK_RX_QUEUE, VSOCK_TX_QUEUE, + }; use crate::vsock::packet::{ VsockPacketFlags, VsockPacketHeader, VsockPacketOp, VsockSocketType, }; @@ -1095,6 +1154,7 @@ mod test { tvqs: TestVirtQueues, rx_writer: QueueWriter, tx_writer: QueueWriter, + ev_writer: QueueWriter, } impl VsockTestHarness { @@ -1105,12 +1165,14 @@ mod test { VqSize::new(1), // Event ]); - // RX and TX use separate data regions + // RX, TX, and Event use separate data regions let rx_writer = tvqs.writer(VSOCK_RX_QUEUE as usize, 0); let tx_writer = tvqs.writer(VSOCK_TX_QUEUE as usize, PAGE_SIZE * 16); + let ev_writer = + tvqs.writer(VSOCK_EVENT_QUEUE as usize, PAGE_SIZE * 32); - Self { tvqs, rx_writer, tx_writer } + Self { tvqs, rx_writer, tx_writer, ev_writer } } fn make_vsock_vq(&self) -> VsockVq { @@ -1179,6 +1241,31 @@ mod test { self.rx_writer.used_idx(self.tvqs.mem_acc()) } + /// Add a writable event buffer (sized for `virtio_vsock_event`) to + /// the event queue and publish it. + fn add_event_buffer(&mut self) -> u16 { + let d = self.ev_writer.add_writable(self.tvqs.mem_acc(), 4); + self.ev_writer.publish_avail(self.tvqs.mem_acc(), d); + d + } + + fn event_used_idx(&self) -> u16 { + self.ev_writer.used_idx(self.tvqs.mem_acc()) + } + + /// Read a `virtio_vsock_event` id from an event queue used entry. + fn read_event(&self, used_index: u16) -> u32 { + let mem_acc = self.tvqs.mem_acc(); + let elem = self.ev_writer.read_used_elem(mem_acc, used_index); + assert_eq!(elem.len, 4, "event used entry holds one le32 id"); + let buf = self.ev_writer.read_desc_data( + mem_acc, + elem.id as u16, + elem.len as usize, + ); + u32::from_le_bytes(buf.try_into().expect("4 byte event")) + } + fn tx_used_idx(&self) -> u16 { self.tx_writer.used_idx(self.tvqs.mem_acc()) } @@ -2284,4 +2371,63 @@ mod test { notify.halt_evloop(); handle.join().unwrap(); } + + #[test] + fn transport_reset_event_delivered_on_resume() { + let guest_cid = GuestCid::try_from(50).unwrap(); + + let mut harness = VsockTestHarness::new(); + let vq = harness.make_vsock_vq(); + let log = test_logger(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); + + // The "guest" queued an event buffer before the device moved. + harness.add_event_buffer(); + + let notify = poller.notify_handle(); + let handle = poller.run(); + + // Signal while paused (as a migration import would), then resume. + notify.send_transport_reset(); + notify.resume_evloop(); + + wait_for_condition(|| harness.event_used_idx() >= 1, 5000); + assert_eq!(harness.read_event(0), 0 /* TRANSPORT_RESET */); + + notify.pause_evloop().unwrap(); + notify.halt_evloop(); + handle.join().unwrap(); + } + + #[test] + fn transport_reset_event_retried_on_event_queue_kick() { + let guest_cid = GuestCid::try_from(50).unwrap(); + + let mut harness = VsockTestHarness::new(); + let vq = harness.make_vsock_vq(); + let log = test_logger(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); + + let notify = poller.notify_handle(); + let handle = poller.run(); + + // Signal and resume with no event buffer available: the event + // cannot be delivered yet and must stay pending. + notify.send_transport_reset(); + notify.resume_evloop(); + assert_eq!(harness.event_used_idx(), 0); + + // The guest now supplies a buffer and kicks the event queue. + harness.add_event_buffer(); + notify.queue_notify(VSOCK_EVENT_QUEUE).unwrap(); + + wait_for_condition(|| harness.event_used_idx() >= 1, 5000); + assert_eq!(harness.read_event(0), 0 /* TRANSPORT_RESET */); + + notify.pause_evloop().unwrap(); + notify.halt_evloop(); + handle.join().unwrap(); + } } diff --git a/lib/propolis/src/vsock/poller_stub.rs b/lib/propolis/src/vsock/poller_stub.rs index 72c81ddd6..060505314 100644 --- a/lib/propolis/src/vsock/poller_stub.rs +++ b/lib/propolis/src/vsock/poller_stub.rs @@ -37,6 +37,8 @@ impl VsockPollerNotify { pub fn halt_evloop(&self) {} + pub fn send_transport_reset(&self) {} + pub fn wait_stopped(&self) {} } diff --git a/lib/propolis/src/vsock/proxy.rs b/lib/propolis/src/vsock/proxy.rs index 6ebb71f5b..4fb896b37 100644 --- a/lib/propolis/src/vsock/proxy.rs +++ b/lib/propolis/src/vsock/proxy.rs @@ -370,6 +370,10 @@ impl VsockProxy { self.poller.reset_evloop(); } + pub fn send_transport_reset(&self) { + self.poller.send_transport_reset(); + } + pub fn halt(&self) { self.poller.halt_evloop(); self.poller.wait_stopped();