From 39a063a5edb09b414c8cb5494fb65ec857c91366 Mon Sep 17 00:00:00 2001 From: Mike Zeller Date: Tue, 3 Nov 2026 20:12:20 +0000 Subject: [PATCH 1/2] [spr] changes to master this commit is based on [skip ci] --- lib/propolis/src/vsock/poller.rs | 347 +++++++++----------------- lib/propolis/src/vsock/poller_stub.rs | 11 +- lib/propolis/src/vsock/proxy.rs | 30 +-- 3 files changed, 124 insertions(+), 264 deletions(-) diff --git a/lib/propolis/src/vsock/poller.rs b/lib/propolis/src/vsock/poller.rs index 0646d9766..df908d228 100644 --- a/lib/propolis/src/vsock/poller.rs +++ b/lib/propolis/src/vsock/poller.rs @@ -11,7 +11,6 @@ use std::io::Read; use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd, RawFd}; use std::sync::mpsc; use std::sync::Arc; -use std::sync::Barrier; use std::sync::Condvar; use std::sync::Mutex; use std::thread::JoinHandle; @@ -94,7 +93,7 @@ impl PollerState { /// Commands that can be sent to a paused `poller_loop`. enum PausedCmd { /// Continue execution - Resume, + Resume { oneshot: mpsc::SyncSender<()> }, /// Cleanup all connection state and queued packets Reset { oneshot: mpsc::SyncSender<()> }, /// Shutdown the event-loop @@ -132,21 +131,23 @@ impl VsockPollerNotify { } } - pub fn pause(&self) -> std::io::Result<()> { + pub fn pause_evloop(&self) -> std::io::Result<()> { self.port_send(VsockEvent::Pause) } - pub fn resume(&self) { - self.pause_tx.send(PausedCmd::Resume).unwrap(); + pub fn resume_evloop(&self) { + let (tx, rx) = mpsc::sync_channel(0); + self.pause_tx.send(PausedCmd::Resume { oneshot: tx }).unwrap(); + rx.recv().unwrap(); } - pub fn reset(&self) { + pub fn reset_evloop(&self) { let (tx, rx) = mpsc::sync_channel(0); self.pause_tx.send(PausedCmd::Reset { oneshot: tx }).unwrap(); rx.recv().unwrap(); } - pub fn halt(&self) { + pub fn halt_evloop(&self) { self.pause_tx.send(PausedCmd::Halt).unwrap(); } @@ -193,7 +194,6 @@ struct ClosingConn { pub struct VsockPoller { log: Logger, - start_barrier: Arc, /// The guest context id guest_cid: GuestCid, /// Port mappings we are proxying packets to and from @@ -223,7 +223,6 @@ impl VsockPoller { /// the guest VM and host sockets. pub fn new( log: Logger, - start_barrier: Arc, cid: GuestCid, queues: VsockVq, port_mappings: IdHashMap, @@ -257,7 +256,6 @@ impl VsockPoller { Ok(Self { log, - start_barrier, guest_cid: cid, port_mappings, port_fd: Arc::new(unsafe { OwnedFd::from_raw_fd(port_fd) }), @@ -908,37 +906,32 @@ impl VsockPoller { } } - // This is the gerneral flow of the single-threaded processing event-loop: + // This is the general flow of the single-threaded processing event-loop: // - // ┌─────────virtio-socket─event-loop───────┐ - // │ │ - // │ ┌─────────────────────┐ │ - // │ │ start (set_running) ◄────┐ │ - // │ └──────────┬──────────┘ │ │ - // ┌───────────┐ │ │ │ │ - // │vq rx event│ │ │ │ │ - // │vq tx event│ │ ┌──────────▼──────────┐ │ │ - // │fd pollset ├───┼─────────► handle_events() │ │ │ - // │pause event│ │ └──────────┬──────────┘ │ │ - // └───────────┘ │ │ │ │ - // │ │ │ │ - // │ ┌──────────▼──────────┐ │ │ - // │ │ paused (set_stopped)│ │ │ - // │ └──────────┬──────────┘ │ │ - // │ │ │ │ - // │ │ │ │ - // │ ┌──────▼───────┐ │ │ - // │ │ pause_rx │ │ │ - // │ │ │ │ │ - // │ │ - resume────┼───────┘ │ - // │ │ - reset─────┼─►cleanup │ - // │ │ - halt───┐ │ state │ - // │ │ │ │ │ - // │ └───────────┼──┘ │ - // │ │ │ - // └─────────────────────────┼──────────────┘ - // ▼ - // Exit + // ┌─────────────┐ + // │ vq rx event │ + // thread spawn │ vq tx event │ + // │ │ fd pollset │ + // │ │ pause event │ + // │ └──────┬──────┘ + // │ │ + // ┌──────────────┼──virtio-socket─event-loop──────┼─────────────┐ + // │ ▼ │ │ + // │ ┌──────────────────────┐ ┌───────────▼───────────┐ │ + // │┌►│ paused (set_stopped) ◄───pause─┤ handle_events() │ │ + // ││ └───────────┬──────────┘ └───────────▲───────────┘ │ + // ││ │ │ │ + // ││ ┌──────▼───────┐ │ │ + // ││ │ pause_rx │ ┌───────────┴───────────┐ │ + // ││ │ run ──────┼────────────► running (set_running) │ │ + // ││ │ ┌─ reset │ └───────────────────────┘ │ + // ││ │ │ halt ─┐ │ │ + // ││ └─┼────────┤───┘ │ + // ││ ▼ │ │ + // │└─ cleanup state │ │ + // └──────────────────┼──────────────────────────────────────────┘ + // ▼ + // Exit // // The event-loop is executing in a dedicated thread and therefore must // be able to handle triggers from propolis as it manages the device @@ -953,46 +946,34 @@ impl VsockPoller { // every tracked fd and later re-associating them, this allows us to yield // execution until one of the desired next state events is delivered. fn poller_loop(&mut self) { - // Wait until propolis tells us to start the virtio-socket device. - // If we are never told to start that is okay as there will be no need - // to ever transition the device from running->paused->halt. - // - // TODO: If propolis-server is started but the VM is never transitioned - // to running this will leave a useless thread around that will be - // cleanned up when propolis-server exits. This also means if one sends - // a new instance spec to the pre-existing propolis-server we will - // accumulate these "dead" threads, howerver we never do this in the - // actual product. - self.start_barrier.wait(); - loop { - // We are running! - self.state.set_running(); + match self.pause_rx.recv() { + Ok(PausedCmd::Resume { oneshot }) => { + oneshot.send(()).unwrap(); - // Handle events until we are told to pause. - self.handle_events(); + // We are running! + self.state.set_running(); - // Transition to stopped and await for next steps. - self.state.set_stopped(); + // Handle events until we are told to pause. + self.handle_events(); - loop { - match self.pause_rx.recv() { - Ok(PausedCmd::Resume) => break, - Ok(PausedCmd::Reset { oneshot }) => { - self.reset(); - oneshot.send(()).unwrap(); - } - Ok(PausedCmd::Halt) => { - self.reset(); - return; - } - Err(_) => { - error!( + // Transition to stopped and await for next steps. + self.state.set_stopped(); + } + Ok(PausedCmd::Reset { oneshot }) => { + self.reset(); + oneshot.send(()).unwrap(); + } + Ok(PausedCmd::Halt) => { + self.reset(); + return; + } + Err(_) => { + error!( &self.log, "all VsockPoller pause_tx senders have been dropped" ); - return; - } + return; } } } @@ -1070,7 +1051,6 @@ mod test { use std::net::TcpListener; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; - use std::sync::Barrier; use std::time::Duration; use iddqd::IdHashMap; @@ -1235,21 +1215,13 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); harness.add_rx_writable(256); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); let mut hdr = VsockPacketHeader::new(); hdr.set_src_cid(guest_cid) @@ -1276,8 +1248,8 @@ mod test { assert_eq!(resp_hdr.dst_port(), guest_port); assert_eq!(resp_hdr.socket_type(), Some(VsockSocketType::Stream)); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1288,21 +1260,14 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - IdHashMap::new(), - ) - .unwrap(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); harness.add_rx_writable(256); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); let mut hdr = VsockPacketHeader::new(); hdr.set_src_cid(guest_cid) @@ -1328,8 +1293,8 @@ mod test { assert_eq!(resp_hdr.src_port(), 8080); assert_eq!(resp_hdr.dst_port(), 5555); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1344,15 +1309,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -1360,7 +1317,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1410,8 +1367,8 @@ mod test { accepted.read_exact(&mut buf).unwrap(); assert_eq!(&buf, payload); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1426,15 +1383,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide plenty of RX descriptors for RESPONSE + credit updates for _ in 0..16 { @@ -1443,7 +1392,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection let mut req_hdr = VsockPacketHeader::new(); @@ -1537,8 +1486,8 @@ mod test { } assert!(found_credit_update, "expected a CREDIT_UPDATE on RX queue"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1553,15 +1502,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -1569,7 +1510,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1623,8 +1564,8 @@ mod test { Ok(n) => panic!("expected EOF or error, got {n} bytes"), } - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1639,15 +1580,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Pre-populate RX queue with writable descriptors for RESPONSE + data for _ in 0..8 { @@ -1656,7 +1589,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Write REQUEST packet into TX queue let mut req_hdr = VsockPacketHeader::new(); @@ -1724,8 +1657,8 @@ mod test { assert_eq!(resp_hdr.dst_port(), guest_port); assert_eq!(&host_buf, host_payload, "host->guest data mismatch"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1740,22 +1673,14 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide only one RX descriptor, just enough for the RESPONSE. harness.add_rx_writable(4096); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Send REQUEST let mut req_hdr = VsockPacketHeader::new(); @@ -1804,8 +1729,8 @@ mod test { assert_eq!(rw_hdr.dst_port(), guest_port); assert_eq!(&payload, host_data); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -1825,15 +1750,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide initial RX descriptors for RESPONSE + credit updates for _ in 0..8 { @@ -1842,7 +1759,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection // Use a large buf_alloc so host->guest credit doesn't run out @@ -2023,8 +1940,8 @@ mod test { assert_eq!(host_to_guest.len(), total_bytes); assert!(host_to_guest == host_data, "host->guest data mismatch"); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2042,15 +1959,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Provide RX descriptors for RESPONSE + SHUTDOWN for _ in 0..4 { @@ -2059,7 +1968,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish connection let mut req_hdr = VsockPacketHeader::new(); @@ -2116,8 +2025,8 @@ mod test { assert_eq!(hdr.src_port(), vsock_port); assert_eq!(hdr.dst_port(), guest_port); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2132,15 +2041,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); // Pre-populate RX queue with writable descriptors for RESPONSE + data for _ in 0..8 { @@ -2149,7 +2050,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Write REQUEST packet into TX queue let mut req_hdr = VsockPacketHeader::new(); @@ -2230,8 +2131,8 @@ mod test { assert_eq!(rst_hdr.src_port(), vsock_port); assert_eq!(rst_hdr.dst_port(), guest_port); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2249,15 +2150,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..4 { harness.add_rx_writable(4096); @@ -2265,7 +2158,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish a connection let mut req_hdr = VsockPacketHeader::new(); @@ -2290,9 +2183,9 @@ mod test { wait_for_condition(|| harness.rx_used_idx() >= 1, 5000); // Pause then reset. - notify.pause().unwrap(); + notify.pause_evloop().unwrap(); notify.wait_stopped(); - notify.reset(); + notify.reset_evloop(); // NOTE: We don't have a way to actually validate that the reset call // above has not left behind a `GuestAddr` as apart of the `VsockVq`. @@ -2306,8 +2199,8 @@ mod test { Ok(n) => panic!("expected EOF or error, got {n} bytes"), } - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2324,15 +2217,7 @@ mod test { let mut harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - backends, - ) - .unwrap(); + let poller = VsockPoller::new(log, guest_cid, vq, backends).unwrap(); for _ in 0..8 { harness.add_rx_writable(4096); @@ -2340,7 +2225,7 @@ mod test { let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); + notify.resume_evloop(); // Establish a connection let mut req_hdr = VsockPacketHeader::new(); @@ -2365,9 +2250,9 @@ mod test { wait_for_condition(|| harness.rx_used_idx() >= 1, 5000); // Pause and resume — connection state should survive. - notify.pause().unwrap(); + notify.pause_evloop().unwrap(); notify.wait_stopped(); - notify.resume(); + notify.resume_evloop(); // Write data from the host side through the still-open TCP // socket. The poller should deliver it to the guest. @@ -2381,8 +2266,8 @@ mod test { assert_eq!(rw_hdr.op(), Some(VsockPacketOp::ReadWrite)); assert_eq!(&data, host_payload); - notify.pause().unwrap(); - notify.halt(); + notify.pause_evloop().unwrap(); + notify.halt_evloop(); handle.join().unwrap(); } @@ -2393,22 +2278,14 @@ mod test { let harness = VsockTestHarness::new(); let vq = harness.make_vsock_vq(); let log = test_logger(); - let start_barrier = Arc::new(Barrier::new(2)); - let poller = VsockPoller::new( - log, - start_barrier.clone(), - guest_cid, - vq, - IdHashMap::new(), - ) - .unwrap(); + let poller = + VsockPoller::new(log, guest_cid, vq, IdHashMap::new()).unwrap(); let notify = poller.notify_handle(); let handle = poller.run(); - start_barrier.wait(); - - notify.pause().unwrap(); - notify.halt(); + notify.resume_evloop(); + 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 712a4dc40..72c81ddd6 100644 --- a/lib/propolis/src/vsock/poller_stub.rs +++ b/lib/propolis/src/vsock/poller_stub.rs @@ -2,8 +2,6 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -use std::sync::Arc; -use std::sync::Barrier; use std::thread::JoinHandle; use iddqd::IdHashMap; @@ -29,15 +27,15 @@ impl VsockPollerNotify { )); } - pub fn pause(&self) -> std::io::Result<()> { + pub fn pause_evloop(&self) -> std::io::Result<()> { Ok(()) } - pub fn resume(&self) {} + pub fn resume_evloop(&self) {} - pub fn reset(&self) {} + pub fn reset_evloop(&self) {} - pub fn halt(&self) {} + pub fn halt_evloop(&self) {} pub fn wait_stopped(&self) {} } @@ -47,7 +45,6 @@ pub struct VsockPoller; impl VsockPoller { pub fn new( _log: Logger, - _start_barrier: Arc, _cid: GuestCid, _queues: VsockVq, _port_mappings: IdHashMap, diff --git a/lib/propolis/src/vsock/proxy.rs b/lib/propolis/src/vsock/proxy.rs index 01085b157..6ebb71f5b 100644 --- a/lib/propolis/src/vsock/proxy.rs +++ b/lib/propolis/src/vsock/proxy.rs @@ -8,8 +8,6 @@ use std::num::NonZeroUsize; use std::num::Wrapping; use std::os::fd::AsRawFd; use std::os::fd::RawFd; -use std::sync::Arc; -use std::sync::Barrier; use std::thread::JoinHandle; use std::time::Duration; @@ -324,7 +322,6 @@ impl IdHashItem for VsockPortMapping { /// virtio-socket backend that proxies between a guest and a host UDS. pub struct VsockProxy { log: Logger, - start_barrier: Arc, poller: VsockPollerNotify, _evloop_handle: JoinHandle<()>, } @@ -336,23 +333,12 @@ impl VsockProxy { queues: VsockVq, port_mappings: IdHashMap, ) -> Self { - // We use a `Barrier` to gate when the spawned `VsockPoller` thread is - // allowed to transition to the running state and is allowed to start - // using guest memory. Forward progress will be made when the - // virtio-socket device calls `Lifecycle::start`. - let start_barrier = Arc::new(Barrier::new(2)); - let evloop = VsockPoller::new( - log.clone(), - start_barrier.clone(), - cid, - queues, - port_mappings, - ) - .unwrap(); + let evloop = + VsockPoller::new(log.clone(), cid, queues, port_mappings).unwrap(); let poller = evloop.notify_handle(); let jh = evloop.run(); - Self { log, start_barrier, poller, _evloop_handle: jh } + Self { log, poller, _evloop_handle: jh } } /// Notification from the vsock device that one of the queues has had an @@ -365,11 +351,11 @@ impl VsockProxy { // TODO this *should* only be called once given the guarantees propolis // gives us. We might want to start using the `Indicator` type in the // future. - self.start_barrier.wait(); + self.poller.resume_evloop(); } pub fn pause(&self) -> std::io::Result<()> { - self.poller.pause() + self.poller.pause_evloop() } pub fn wait_stopped(&self) { @@ -377,15 +363,15 @@ impl VsockProxy { } pub fn resume(&self) { - self.poller.resume(); + self.poller.resume_evloop(); } pub fn reset(&self) { - self.poller.reset(); + self.poller.reset_evloop(); } pub fn halt(&self) { - self.poller.halt(); + self.poller.halt_evloop(); self.poller.wait_stopped(); } } From 9d0707688625404e8dea1c41a8bf731b3457e397 Mon Sep 17 00:00:00 2001 From: Mike Zeller Date: Thu, 5 Nov 2026 21:51:47 +0000 Subject: [PATCH 2/2] fix rebase --- lib/propolis/src/hw/virtio/queue.rs | 135 +++++++++++++++++++++++----- lib/propolis/src/hw/virtio/vsock.rs | 72 +++++---------- lib/propolis/src/vsock/poller.rs | 4 - 3 files changed, 134 insertions(+), 77 deletions(-) diff --git a/lib/propolis/src/hw/virtio/queue.rs b/lib/propolis/src/hw/virtio/queue.rs index 6031c5fc7..cec1b4481 100644 --- a/lib/propolis/src/hw/virtio/queue.rs +++ b/lib/propolis/src/hw/virtio/queue.rs @@ -88,26 +88,30 @@ pub struct VqAvail { } impl VqAvail { - /// If there's a request ready, pop it off the queue and return the - /// corresponding descriptor and available ring indicies. - fn read_next_avail(&mut self, rsize: u16, mem: &MemCtx) -> Option { + /// Read the entry at the front of the avail ring without consuming it. The + /// same entry will remain visible on subsequent peeks and pops. + fn peek_next_avail(&self, rsize: u16, mem: &MemCtx) -> Option { if !self.valid { return None; } - if let Some(idx) = mem.read::(self.gpa_idx) { - let ndesc = Wrapping(*idx) - self.cur_avail_idx; - if ndesc.0 != 0 && ndesc.0 <= rsize { - let avail_idx = self.cur_avail_idx.0 & (rsize - 1); - self.cur_avail_idx += Wrapping(1); - - fence(Ordering::Acquire); - let addr = self.gpa_ring.offset::(avail_idx as usize); - return mem - .read(addr) - .map(|desc_idx| VqReq { desc_idx: *desc_idx, avail_idx }); - } + let idx = mem.read::(self.gpa_idx)?; + let ndesc = Wrapping(*idx) - self.cur_avail_idx; + if ndesc.0 == 0 || ndesc.0 > rsize { + return None; } - None + let avail_idx = self.cur_avail_idx.0 & (rsize - 1); + + fence(Ordering::Acquire); + let addr = self.gpa_ring.offset::(avail_idx as usize); + mem.read(addr).map(|desc_idx| VqReq { desc_idx: *desc_idx, avail_idx }) + } + + /// If there's a request ready, pop it off the queue and return the + /// corresponding descriptor and available ring indicies. + fn read_next_avail(&mut self, rsize: u16, mem: &MemCtx) -> Option { + let req = self.peek_next_avail(rsize, mem)?; + self.cur_avail_idx += Wrapping(1); + Some(req) } fn read_ring_descr( @@ -494,17 +498,53 @@ impl VirtQueue { assert!(chain.idx.is_none()); let mut avail = self.avail.lock().unwrap(); let req = avail.read_next_avail(self.size(), mem)?; + probes::virtio_vq_pop!(|| ( + self as *const VirtQueue as u64, + req.desc_idx, + req.avail_idx, + )); + self.walk_chain(&avail, req, chain, mem) + } + /// Accumulates the descriptor chain at the front of the avail ring into + /// a `Chain`, returned as an [`AvailChain`], without consuming the queue + /// entry. + /// + /// The queue cursor is untouched: dropping the returned [`AvailChain`] + /// is a no-op and the same chain is returned by the next peek or pop. + /// Consuming it via [`AvailChain::push_used`] advances the cursor and + /// pushes the chain onto the used ring in one step. + pub fn peek_avail(&self, mem: &MemCtx) -> Option> { + let mut avail = self.avail.lock().unwrap(); + let req = avail.peek_next_avail(self.size(), mem)?; + let pos = avail.cur_avail_idx; + let mut chain = Chain::with_capacity(10); + if self.walk_chain(&avail, req, &mut chain, mem).is_none() { + // TODO (#1190) + // If the chain is malformed we bump the index that would have + // happened in `avail.read_next_avail` so that we don't get stuck + // peeking the same malformed chain over and over. What we really + // need to do, is signal the error condition via DEVICE_NEEDS_RESET. + // Until then, we will leak these descriptors. + avail.cur_avail_idx += Wrapping(1); + return None; + } + Some(AvailChain { vq: self, req, cur_avail_idx_snap: pos, chain }) + } + + /// Given the [`VqReq`]'s `desc_idx`, fill the provided [`Chain`]. + fn walk_chain( + &self, + avail: &VqAvail, + req: VqReq, + chain: &mut Chain, + mem: &MemCtx, + ) -> Option<(u16, u32)> { let mut desc = avail.read_ring_descr(req.desc_idx, self.size(), mem)?; let mut flags = DescFlag::from_bits_truncate(desc.flags); let mut count = 0; let mut len = 0; chain.idx = Some(req.desc_idx); - probes::virtio_vq_pop!(|| ( - self as *const VirtQueue as u64, - req.desc_idx, - req.avail_idx, - )); // non-indirect descriptor(s) while !flags.contains(DescFlag::INDIRECT) { @@ -873,6 +913,59 @@ impl VirtQueue { } } +/// The current descriptor chain read from the front of the avail ring, but not +/// yet consumed. +/// +/// Dropping the [`AvailChain`] is a no-op and leaves the data where it is, +/// while calling [`AvailChain::push_used`] will advance the cursor and push the +/// chain onto the used ring. +/// +/// NB: This is here mostly for our virtio-socket implementation so that we do +/// hold onto cached descriptors for the poller, which ensures there is only a +/// single consumer of the guest's virt queues. +pub struct AvailChain<'a> { + vq: &'a VirtQueue, + req: VqReq, + // a snapshot of `cur_avail_idx` when calling [`VirtQueue::peek_avail`], + // used to validate the VirtQueue only has a single consumer. + cur_avail_idx_snap: Wrapping, + chain: Chain, +} + +impl AvailChain<'_> { + pub fn chain(&self) -> &Chain { + &self.chain + } + + pub fn chain_mut(&mut self) -> &mut Chain { + &mut self.chain + } + + /// Consume the chain, and advance the avail cursor. This pushes it onto + /// the used ring. + pub fn push_used(mut self, mem: &MemCtx) { + { + let mut avail = self.vq.avail.lock().unwrap(); + // Our implementation of virtio queues allow for multiple consumers, + // which means racing consumers can step on each other. This should + // only be called when a single consumer is polling a virt queue, + // such as with virtio-socket. We assert here to indicate programmer + // error if we ever find things out of sync. + assert_eq!( + avail.cur_avail_idx, self.cur_avail_idx_snap, + "avail cursor moved since peek" + ); + avail.cur_avail_idx += Wrapping(1); + } + probes::virtio_vq_pop!(|| ( + self.vq as *const VirtQueue as u64, + self.req.desc_idx, + self.req.avail_idx, + )); + self.vq.push_used(&mut self.chain, mem); + } +} + bitflags! { #[derive(Default)] pub struct DescFlag: u16 { diff --git a/lib/propolis/src/hw/virtio/vsock.rs b/lib/propolis/src/hw/virtio/vsock.rs index 4612693f2..7bc595db8 100644 --- a/lib/propolis/src/hw/virtio/vsock.rs +++ b/lib/propolis/src/hw/virtio/vsock.rs @@ -6,15 +6,18 @@ use lazy_static::lazy_static; use slog::Logger; use std::sync::Arc; +use crate::accessors::Guard; use crate::accessors::MemAccessor; use crate::common::*; use crate::hw::pci; use crate::hw::virtio; +use crate::hw::virtio::queue::AvailChain; use crate::hw::virtio::queue::Chain; use crate::hw::virtio::queue::VirtQueue; use crate::hw::virtio::queue::VqSize; use crate::migrate::*; use crate::util::regmap::RegMap; +use crate::vmm::mem::MemAccessed; use crate::vmm::MemCtx; use crate::vsock::packet::VsockPacket; use crate::vsock::packet::VsockPacketError; @@ -40,36 +43,25 @@ pub const VSOCK_EVENT_QUEUE: u16 = 0x2; /// This guarantees we have space to send a packet to the guest before reading /// data from a host socket, preventing data loss if the queue is full. /// -/// The permit holds a mutable reference to `VsockVq`, ensuring only one permit -/// can exist at a time (enforced at compile time). If dropped without calling -/// `write`, the chain is retained in `VsockVq` for reuse. +/// NB: The underlying descriptor chain is not consumed until a call to +/// [`RxPermit::write`] is made. Droopping the [`RxPermit`] is a no-op without a +/// call to write is a no-op. pub struct RxPermit<'a> { - vq: &'a mut VsockVq, + avail: AvailChain<'a>, + mem: Guard<'a, MemAccessed>, } impl RxPermit<'_> { /// Returns the maximum data payload that can fit in this descriptor chain. pub fn available_data_space(&self) -> usize { let header_size = std::mem::size_of::(); - self.vq - .rx_chain - .as_ref() - .expect("has chain") - .remain_write_bytes() - .saturating_sub(header_size) + self.avail.chain().remain_write_bytes().saturating_sub(header_size) } - pub fn write(self, header: &VsockPacketHeader, data: &[u8]) { - // TODO: cannot access memory? - let mem = self.vq.acc_mem.access().expect("mem access for write"); - let queue = - self.vq.queues.get(VSOCK_RX_QUEUE as usize).expect("rx queue"); - - // NB: `RxPermit` should only be created if the owning `VsockVq` - // actually has a `Some(Chain)`. Unfortuantely there doesn't seem to be - // a way to enforce this at compile time. - let mut chain = self.vq.rx_chain.take().expect("has chain"); - chain.write(header, &mem); + pub fn write(mut self, header: &VsockPacketHeader, data: &[u8]) { + let mem = &*self.mem; + let chain = self.avail.chain_mut(); + chain.write(header, mem); if !data.is_empty() { let mut done = 0; @@ -86,15 +78,13 @@ impl RxPermit<'_> { } probes::vsock_pkt_rx!(|| header); - queue.push_used(&mut chain, &mem); + self.avail.push_used(mem); } } pub struct VsockVq { queues: Vec>, acc_mem: MemAccessor, - /// Cached rx chain for permit reuse when dropped without write - rx_chain: Option, } impl VsockVq { @@ -102,13 +92,13 @@ impl VsockVq { queues: Vec>, acc_mem: MemAccessor, ) -> Self { - Self { queues, acc_mem, rx_chain: None } + Self { queues, acc_mem } } /// Try to acquire a permit for sending a packet to the guest. /// - /// Returns `Some(RxPermit)` if a descriptor chain is available, - /// `None` if the rx queue is full. + /// Returns `Some(RxPermit)` if a descriptor chain is available, `None` + /// if the rx queue is full. pub fn try_rx_permit(&mut self) -> Option> { let vq = self.queues.get(VSOCK_RX_QUEUE as usize)?; // See propolis#1110 & propolis#1115 @@ -121,22 +111,9 @@ impl VsockVq { return None; } - // Reuse cached chain or pop a new one - if self.rx_chain.is_none() { - // TODO: cannot access memory? - let mem = self.acc_mem.access().expect("mem access for write"); - let mut chain = Chain::with_capacity(10); - if let Some(_) = vq.pop_avail(&mut chain, &mem) { - self.rx_chain = Some(chain); - } - } - - // We only return a permit iff we know that we are holding onto a valid - // descriptor chain that can be used by the borrowing `RxPermit` - match self.rx_chain { - Some(_) => Some(RxPermit { vq: self }), - None => None, - } + let mem = self.acc_mem.access()?; + let avail = vq.peek_avail(&mem)?; + Some(RxPermit { avail, mem }) } /// Receive all available packets from the TX queue. @@ -161,15 +138,6 @@ impl VsockVq { Some(packet) } - - /// Drop any cached descriptor chain. - /// - /// This MUST be called when reseting the virtio-socket device so - /// that we don't use stale `GuestAddr`s across device resets. - #[cfg(target_os = "illumos")] - pub(crate) fn clear_rx_chain(&mut self) { - self.rx_chain = None; - } } pub struct PciVirtioSock { diff --git a/lib/propolis/src/vsock/poller.rs b/lib/propolis/src/vsock/poller.rs index df908d228..ff70fd734 100644 --- a/lib/propolis/src/vsock/poller.rs +++ b/lib/propolis/src/vsock/poller.rs @@ -984,7 +984,6 @@ impl VsockPoller { self.rx.clear(); self.rx_blocked.clear(); self.quiescing.clear(); - self.queues.clear_rx_chain(); } } @@ -2187,9 +2186,6 @@ mod test { notify.wait_stopped(); notify.reset_evloop(); - // NOTE: We don't have a way to actually validate that the reset call - // above has not left behind a `GuestAddr` as apart of the `VsockVq`. - // The host-side TCP socket should now be closed because the // poller dropped all connections during reset. let mut buf = [0u8; 1];