diff --git a/alioth/src/cpu/cpu_amd64/sev.rs b/alioth/src/cpu/cpu_amd64/sev.rs index ed88cafb..c15f97a6 100644 --- a/alioth/src/cpu/cpu_amd64/sev.rs +++ b/alioth/src/cpu/cpu_amd64/sev.rs @@ -57,7 +57,7 @@ where fn update_snp_desc(&self, desc: &SevMetadataDesc) -> Result<()> { let mut cpuid_table = SnpCpuidInfo::new_zeroed(); let ram_bus = self.ctx.board.memory.ram_bus(); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let page_type = match desc.type_ { SevDescType::SNP_DESC_MEM => SnpPageType::UNMEASURED, SevDescType::SNP_SECRETS => SnpPageType::SECRETS, diff --git a/alioth/src/mem/addressable.rs b/alioth/src/mem/addressable.rs index b77142f7..cd7314d1 100644 --- a/alioth/src/mem/addressable.rs +++ b/alioth/src/mem/addressable.rs @@ -20,7 +20,7 @@ pub trait SlotBackend { fn size(&self) -> u64; } -#[derive(Debug)] +#[derive(Debug, Clone)] struct Slot where B: SlotBackend, @@ -52,7 +52,7 @@ where } } -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct Addressable where B: SlotBackend, diff --git a/alioth/src/mem/mapped.rs b/alioth/src/mem/mapped.rs index c6fea98b..ab94837d 100644 --- a/alioth/src/mem/mapped.rs +++ b/alioth/src/mem/mapped.rs @@ -24,6 +24,7 @@ use std::os::fd::FromRawFd; use std::os::fd::{AsFd, AsRawFd, BorrowedFd}; use std::ptr::{NonNull, null_mut}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; #[cfg(target_os = "linux")] use libc::{MADV_HUGEPAGE, MFD_CLOEXEC}; @@ -31,7 +32,7 @@ use libc::{ MAP_ANONYMOUS, MAP_FAILED, MAP_PRIVATE, MAP_SHARED, MS_ASYNC, PROT_READ, PROT_WRITE, c_void, madvise, mmap, msync, munmap, }; -use parking_lot::{RwLock, RwLockReadGuard}; +use parking_lot::RwLock; use snafu::ResultExt; use zerocopy::{FromBytes, Immutable, IntoBytes}; @@ -180,14 +181,26 @@ impl ArcMemPages { } } -#[derive(Debug)] +/// A snapshot of the guest RAM layout. +/// +/// A `Ram` does not change once a [`RamBus`] publishes it. Its slots hold +/// reference-counted [`ArcMemPages`], so a snapshot keeps the host mappings +/// it refers to alive, even after they are removed from the bus. +#[derive(Debug, Clone, Default)] pub struct Ram { inner: Addressable, + generation: u64, } -#[derive(Debug)] +/// The guest RAM layout, updated in the read-copy-update (RCU) way. +/// +/// A reader takes a snapshot with [`RamBus::load`] and keeps it as long as +/// it needs, while a writer publishes a new layout with [`RamBus::update`] +/// without waiting for readers. +#[derive(Debug, Default)] pub struct RamBus { - ram: RwLock, + ram: RwLock>, + generation: AtomicU64, } struct Iter<'m> { @@ -397,66 +410,83 @@ impl Ram { } Ok(()) } -} -impl Default for RamBus { - fn default() -> Self { - Self::new() + pub(crate) fn add(&mut self, gpa: u64, user_mem: ArcMemPages) -> Result<()> { + self.inner.add(gpa, user_mem)?; + Ok(()) + } + + pub(crate) fn remove(&mut self, gpa: u64) -> Result { + self.inner.remove(gpa) + } + + /// Returns the generation of the layout, see [`RamBus::generation`]. + pub fn generation(&self) -> u64 { + self.generation } } impl RamBus { - pub fn lock_layout(&self) -> RwLockReadGuard<'_, Ram> { - self.ram.read() + pub fn new() -> Self { + Self::default() } - pub fn new() -> Self { - Self { - ram: RwLock::new(Ram { - inner: Addressable::default(), - }), - } + /// Returns a snapshot of the current layout. + /// + /// The snapshot stays valid for as long as the caller holds it, but does + /// not see later updates. + pub fn load(&self) -> Arc { + self.ram.read().clone() } - pub(crate) fn add(&self, gpa: u64, user_mem: ArcMemPages) -> Result<(), Error> { - let mut ram = self.ram.write(); - ram.inner.add(gpa, user_mem)?; - Ok(()) + /// Returns the generation of the current layout. + /// + /// Every update increments it, so a reader can tell whether its snapshot + /// is the latest one without taking the lock. + pub fn generation(&self) -> u64 { + self.generation.load(Ordering::Acquire) } - pub(crate) fn remove(&self, gpa: u64) -> Result { - let mut ram = self.ram.write(); - ram.inner.remove(gpa) + /// Copies the current layout, changes the copy with `f` and publishes it. + /// + /// If `f` fails, nothing is published, so a change of several slots is + /// either fully visible or not visible at all. + pub(crate) fn update(&self, f: impl FnOnce(&mut Ram) -> Result) -> Result { + let mut current = self.ram.write(); + let mut ram = Ram::clone(¤t); + let ret = f(&mut ram)?; + ram.generation = current.generation + 1; + // A reader that sees the new generation before the new layout is + // published waits for it in load(). + self.generation.store(ram.generation, Ordering::Release); + *current = Arc::new(ram); + Ok(ret) } pub fn read(&self, gpa: u64, buf: &mut [u8]) -> Result<()> { - let ram = self.ram.read(); - ram.read(gpa, buf) + self.load().read(gpa, buf) } pub fn write(&self, gpa: u64, buf: &[u8]) -> Result<()> { - let ram = self.ram.read(); - ram.write(gpa, buf) + self.load().write(gpa, buf) } pub fn read_t(&self, gpa: u64) -> Result where T: FromBytes + IntoBytes, { - let ram = self.ram.read(); - ram.read_t(gpa) + self.load().read_t(gpa) } pub fn write_t(&self, gpa: u64, val: &T) -> Result<(), Error> where T: IntoBytes + Immutable, { - let ram = self.ram.read(); - ram.write_t(gpa, val) + self.load().write_t(gpa, val) } pub fn read_range(&self, gpa: u64, len: u64, dst: &mut impl Write) -> Result<()> { - let ram = self.ram.read(); + let ram = self.load(); for r in ram.slice_iter(gpa, len) { dst.write_all(r?).context(error::Write)?; } @@ -464,7 +494,7 @@ impl RamBus { } pub fn write_range(&self, gpa: u64, len: u64, mut src: impl Read) -> Result<()> { - let ram = self.ram.read(); + let ram = self.load(); for r in ram.slice_iter_mut(gpa, len) { src.read_exact(r?).context(error::Read)?; } diff --git a/alioth/src/mem/mapped_test.rs b/alioth/src/mem/mapped_test.rs index 5bc7f2b5..87186bcf 100644 --- a/alioth/src/mem/mapped_test.rs +++ b/alioth/src/mem/mapped_test.rs @@ -38,11 +38,11 @@ fn test_ram_bus_read() { let mem2 = ArcMemPages::from_anonymous(PAGE_SIZE as usize, Some(prot), None).unwrap(); if mem1.addr > mem2.addr { - bus.add(0x0, mem1).unwrap(); - bus.add(PAGE_SIZE, mem2).unwrap(); + bus.update(|ram| ram.add(0x0, mem1)).unwrap(); + bus.update(|ram| ram.add(PAGE_SIZE, mem2)).unwrap(); } else { - bus.add(0x0, mem2).unwrap(); - bus.add(PAGE_SIZE, mem1).unwrap(); + bus.update(|ram| ram.add(0x0, mem2)).unwrap(); + bus.update(|ram| ram.add(PAGE_SIZE, mem1)).unwrap(); } let data = MyStruct { @@ -75,19 +75,49 @@ fn test_ram_bus_read() { let guest_iov = [(0, 16), (PAGE_SIZE - 16, 32), (2 * PAGE_SIZE - 16, 16)]; - let locked_bus = bus.lock_layout(); - let mut iov = locked_bus.translate_iov_mut(&guest_iov).unwrap(); + let ram = bus.load(); + let mut iov = ram.translate_iov_mut(&guest_iov).unwrap(); assert_eq!(iov.len(), 4); assert_matches!((&*data).read_vectored(&mut iov), Ok(64)); drop(iov); - let iov = locked_bus.translate_iov(&guest_iov).unwrap(); + let iov = ram.translate_iov(&guest_iov).unwrap(); assert_eq!(iov.len(), 4); let mut buf_read = Vec::new(); assert_matches!(buf_read.write_vectored(&iov), Ok(64)); assert_eq!(buf_read, data); drop(iov); - drop(locked_bus); - bus.remove(0x0).unwrap(); + drop(ram); + bus.update(|ram| ram.remove(0x0)).unwrap(); +} + +#[test] +fn test_ram_bus_snapshot() { + let bus = RamBus::new(); + let empty = bus.load(); + assert_eq!((bus.generation(), empty.generation()), (0, 0)); + + // Holding a snapshot does not block updates. + let mem = ArcMemPages::from_anonymous(PAGE_SIZE as usize, None, None).unwrap(); + bus.update(|ram| ram.add(0x0, mem.clone())).unwrap(); + assert_matches!(empty.read_t::(0x0), Err(_)); + let snapshot = bus.load(); + assert_eq!((bus.generation(), snapshot.generation()), (1, 1)); + snapshot.write_t(0x0, &0xdeadbeef_u32).unwrap(); + + // A failed update publishes nothing. + let ret = bus.update(|ram| { + ram.add(PAGE_SIZE, mem.clone())?; + ram.add(0x0, mem.clone()) + }); + assert_matches!(ret, Err(_)); + assert_matches!(bus.read_t::(PAGE_SIZE), Err(_)); + assert_eq!(bus.generation(), 1); + + // A removed slot stays accessible through older snapshots. + bus.update(|ram| ram.remove(0x0)).unwrap(); + assert_eq!((bus.generation(), bus.load().generation()), (2, 2)); + assert_matches!(bus.read_t::(0x0), Err(_)); + assert_matches!(snapshot.read_t::(0x0), Ok(0xdeadbeef)); } #[test] @@ -97,9 +127,9 @@ fn test_ram_get_array_ptr() { // Two adjacent guest pages, backed by separate host mappings. for gpa in [0, PAGE_SIZE] { let pages = ArcMemPages::from_anonymous(PAGE_SIZE as usize, Some(prot), None).unwrap(); - bus.add(gpa, pages).unwrap(); + bus.update(|ram| ram.add(gpa, pages)).unwrap(); } - let ram = bus.lock_layout(); + let ram = bus.load(); let ptr = ram.get_array_ptr::(PAGE_SIZE - 16, 2).unwrap(); assert_eq!(ptr, ram.get_ptr::(PAGE_SIZE - 16).unwrap()); diff --git a/alioth/src/mem/mem.rs b/alioth/src/mem/mem.rs index 3add4685..b86f3747 100644 --- a/alioth/src/mem/mem.rs +++ b/alioth/src/mem/mem.rs @@ -266,6 +266,19 @@ struct LayoutCallbacks { updated: Vec>, } +/// Iterates over the RAM-backed ranges of `region` placed at `addr`. +fn ram_ranges(addr: u64, region: &MemRegion) -> impl Iterator { + let ranges = region.ranges.iter().scan(addr, |gpa, range| { + let start = *gpa; + *gpa += range.size(); + Some((start, range)) + }); + ranges.filter_map(|(gpa, range)| match range { + MemRange::Ram(pages) | MemRange::DevMem { pages, .. } => Some((gpa, pages)), + MemRange::Emulated(_) | MemRange::Span(_) => None, + }) +} + // lock order: region -> callbacks -> bus #[derive(Debug, Default)] pub struct Memory { @@ -308,7 +321,7 @@ impl Memory { pub fn register_update_callback(&self, callback: Box) -> Result<()> { let _regions = self.regions.lock(); let mut callbacks = self.callbacks.lock(); - let ram = self.ram_bus.lock_layout(); + let ram = self.ram_bus.load(); callback.ram_updated(&ram)?; callbacks.updated.push(callback); Ok(()) @@ -332,15 +345,20 @@ impl Memory { region.validate()?; let mut regions = self.regions.lock(); regions.add(addr, region.clone())?; - let mut offset = 0; let callbacks = self.callbacks.lock(); - let mut ram_updated = false; + let ram_ranges: Vec<_> = ram_ranges(addr, ®ion).collect(); + let ram_updated = !ram_ranges.is_empty(); + if ram_updated { + self.ram_bus.update(|ram| { + for (gpa, pages) in ram_ranges { + ram.add(gpa, pages.clone())?; + } + Ok(()) + })?; + } + let mut offset = 0; for range in ®ion.ranges { let gpa = addr + offset; - if let MemRange::Ram(pages) | MemRange::DevMem { pages, .. } = range { - self.ram_bus.add(gpa, pages.clone())?; - ram_updated = true; - } match range { MemRange::Emulated(r) => { let mut mmio_bus = self.mmio_bus.write(); @@ -361,7 +379,7 @@ impl Memory { offset += range.size(); } if ram_updated { - let ram = self.ram_bus.lock_layout(); + let ram = self.ram_bus.load(); for update_callback in &callbacks.updated { update_callback.ram_updated(&ram)?; } @@ -374,15 +392,20 @@ impl Memory { } fn unmap_region(&self, addr: u64, region: &MemRegion) -> Result<()> { - let mut offset = 0; let callbacks = self.callbacks.lock(); - let mut ram_updated = false; + let ram_ranges: Vec<_> = ram_ranges(addr, region).collect(); + let ram_updated = !ram_ranges.is_empty(); + if ram_updated { + self.ram_bus.update(|ram| { + for (gpa, _) in ram_ranges { + ram.remove(gpa)?; + } + Ok(()) + })?; + } + let mut offset = 0; for range in ®ion.ranges { let gpa = addr + offset; - if let MemRange::Ram(_) | MemRange::DevMem { .. } = range { - self.ram_bus.remove(gpa)?; - ram_updated = true; - } match range { MemRange::Emulated(_) => { let mut mmio_bus = self.mmio_bus.write(); @@ -403,7 +426,7 @@ impl Memory { offset += range.size(); } if ram_updated { - let ram = self.ram_bus.lock_layout(); + let ram = self.ram_bus.load(); for callback in &callbacks.updated { callback.ram_updated(&ram)?; } diff --git a/alioth/src/virtio/dev/balloon.rs b/alioth/src/virtio/dev/balloon.rs index a16976bc..909d95f0 100644 --- a/alioth/src/virtio/dev/balloon.rs +++ b/alioth/src/virtio/dev/balloon.rs @@ -216,13 +216,9 @@ impl Virtio for Balloon { } impl VirtioMio for Balloon { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, _active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let feature = BalloonFeature::from_bits_retain(feature); @@ -243,13 +239,9 @@ impl VirtioMio for Balloon { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { @@ -273,7 +265,7 @@ impl VirtioMio for Balloon { }; queue.handle_desc(index, active_mio.irq_sender, |chain| { match ballon_q { - BalloonQueue::Inflate => self.inflate(&chain.readable, active_mio.mem), + BalloonQueue::Inflate => self.inflate(&chain.readable, chain.ram()), BalloonQueue::Deflate => { log::info!("{}: VQ_DEFLATE available", self.name); } @@ -285,13 +277,13 @@ impl VirtioMio for Balloon { }) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, _event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) diff --git a/alioth/src/virtio/dev/blk.rs b/alioth/src/virtio/dev/blk.rs index 634082c2..949301c6 100644 --- a/alioth/src/virtio/dev/blk.rs +++ b/alioth/src/virtio/dev/blk.rs @@ -314,37 +314,33 @@ impl Virtio for Block { impl VirtioMio for Block { fn reset(&mut self, _registry: &Registry) {} - fn activate<'m, Q, S>( + fn activate( &mut self, _feature: u128, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, _event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { @@ -406,14 +402,10 @@ impl VirtioMio for Block { #[cfg(target_os = "linux")] impl VirtioIoUring for Block { - fn activate<'m, Q, S>( - &mut self, - _feature: u128, - _ring: &mut ActiveIoUring<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, _feature: u128, _ring: &mut ActiveIoUring<'_, Q, S>) -> Result<()> where S: IrqSender, - Q: VirtQueue<'m>, + Q: VirtQueue, { Ok(()) } diff --git a/alioth/src/virtio/dev/dev.rs b/alioth/src/virtio/dev/dev.rs index df8e0d9b..37fd6c2f 100644 --- a/alioth/src/virtio/dev/dev.rs +++ b/alioth/src/virtio/dev/dev.rs @@ -23,8 +23,8 @@ pub mod net; pub mod vsock; use std::fmt::Debug; -use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU16, AtomicU32, Ordering}; +use std::sync::{Arc, Weak}; use std::thread::JoinHandle; use std::time::{Duration, Instant}; @@ -35,11 +35,11 @@ use snafu::ResultExt; use crate::mem::emulated::Mmio; use crate::mem::mapped::{Ram, RamBus}; -use crate::mem::{LayoutChanged, LayoutUpdated, MemRegion}; +use crate::mem::{self, LayoutChanged, LayoutUpdated, MemRegion}; use crate::sync::notifier::Notifier; use crate::virtio::queue::packed::PackedQueue; use crate::virtio::queue::split::SplitQueue; -use crate::virtio::queue::{QUEUE_SIZE_MAX, Queue, QueueReg, VirtQueue}; +use crate::virtio::queue::{self, QUEUE_SIZE_MAX, Queue, QueueReg, VirtQueue}; #[cfg(target_os = "linux")] use crate::virtio::vu::conn::VuChannel; use crate::virtio::{DeviceId, IrqSender, Result, VirtioFeature, error}; @@ -292,6 +292,16 @@ where true } + /// Wakes up the worker without an event. + /// + /// The worker moves its queues to the latest guest memory layout before + /// it handles any event, so a kick makes an idle worker drop old layouts. + pub fn kick(&self) { + if let Err(e) = self.notifier.notify() { + log::error!("{}: failed to kick worker: {e}", self.name); + } + } + /// Asks the worker to reset the device. /// /// Returns the sequence number of this reset, which @@ -342,6 +352,29 @@ where } } +/// Kicks a worker after each guest memory update, so that an idle worker +/// does not keep the old layout, and the memory removed from it, alive. +#[derive(Debug)] +struct KickWorker +where + S: IrqSender, +{ + ctl: Weak>, +} + +impl LayoutUpdated for KickWorker +where + S: IrqSender, +{ + fn ram_updated(&self, _: &Ram) -> mem::Result<()> { + // The kick neither waits for the worker nor fails the update. + if let Some(ctl) = self.ctl.upgrade() { + ctl.kick(); + } + Ok(()) + } +} + #[derive(Debug)] pub struct VirtioDevice where @@ -424,6 +457,14 @@ where }; Ok(virtio_dev) } + + /// Returns a callback that kicks the worker after each guest memory + /// update. + pub fn mem_update_kick(&self) -> Box { + Box::new(KickWorker { + ctl: Arc::downgrade(&self.ctl), + }) + } } impl Drop for VirtioDevice @@ -438,16 +479,15 @@ where pub trait Backend: Send + 'static { fn register_notifier(&mut self, token: u64) -> Result>; fn reset(&self, dev: &mut D) -> Result<()>; - fn event_loop<'m, S, Q>( + fn event_loop( &mut self, - memory: &'m Ram, context: &mut Context, - queues: &mut [Option>], + queues: &mut [Option>], param: &StartParam, ) -> Result<()> where S: IrqSender, - Q: VirtQueue<'m>; + Q: VirtQueue; } pub trait BackendEvent { @@ -456,6 +496,9 @@ pub trait BackendEvent { pub trait ActiveBackend { type Event: BackendEvent; + type Q: VirtQueue; + /// Returns the queues of the event loop. + fn queues(&mut self) -> &mut [Option>]; fn handle_event(&mut self, dev: &mut D, event: &Self::Event) -> Result<()>; fn handle_queue(&mut self, dev: &mut D, index: u16) -> Result<()>; } @@ -471,6 +514,9 @@ where /// The reset being handled, acknowledged once the worker no longer uses /// the queues. reset_seq: Option, + /// The guest memory generation the queues were last moved to, or `None` + /// at the start of an event loop. + ram_generation: Option, } impl Context @@ -522,6 +568,7 @@ where WakeEvent::Reset { seq } => self.ack_reset(seq), WakeEvent::Start { param } => { self.state = WorkerState::Running; + self.ram_generation = None; return Some(param); } #[cfg(target_os = "linux")] @@ -543,6 +590,17 @@ where where B: ActiveBackend, { + // Between events, the worker accesses guest memory only through the + // queues, so it moves them to the latest layout here. A layout + // published while they move is caught at the next event. + let generation = self.param.memory.generation(); + if self.ram_generation != Some(generation) { + if queue::sync_queues(backend.queues())? { + let name = self.dev.name(); + log::debug!("{name}: moved to guest memory generation {generation}"); + } + self.ram_generation = Some(generation); + } if event.token() == TOKEN_WARKER { self.handle_wake_events(backend) } else { @@ -569,6 +627,7 @@ where param, state: WorkerState::Pending, reset_seq: None, + ram_generation: None, }, backend, }; @@ -580,15 +639,14 @@ where Ok((handle, notifier)) } - fn event_loop<'m, Q>( - &mut self, - queues: &mut [Option>], - ram: &'m Ram, - param: &StartParam, - ) -> Result<()> + fn event_loop(&mut self, param: &StartParam, event_idx: bool) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, { + let ram_bus = &self.context.param.memory; + let new_queue = |reg| Queue::::new(reg, ram_bus.clone(), event_idx); + let queue_regs = self.context.param.queue_regs.iter(); + let mut queues = queue_regs.map(new_queue).collect::>>()?; log::debug!( "{}: activated with {:x?}, {:x?}", self.context.dev.name(), @@ -597,7 +655,7 @@ where ); let ret = self .backend - .event_loop(ram, &mut self.context, queues, param); + .event_loop(&mut self.context, &mut queues, param); // Clean up even if the event loop failed, since the device might // have been partially activated. let reset = self.backend.reset(&mut self.context.dev); @@ -605,30 +663,13 @@ where } fn run(&mut self, param: &StartParam) -> Result<()> { - let memory = self.context.param.memory.clone(); - let ram = memory.lock_layout(); let feature = param.feature & !VirtioFeature::ACCESS_PLATFORM.bits(); - let queue_regs = self.context.param.queue_regs.clone(); let feature = VirtioFeature::from_bits_retain(feature); let event_idx = feature.contains(VirtioFeature::EVENT_IDX); if feature.contains(VirtioFeature::RING_PACKED) { - let new_queue = |reg| { - let Some(split_queue) = PackedQueue::new(reg, &ram, event_idx)? else { - return Ok(None); - }; - Ok(Some(Queue::new(split_queue, reg, &ram))) - }; - let queues: Result> = queue_regs.iter().map(new_queue).collect(); - self.event_loop(&mut (queues?), &ram, param) + self.event_loop::(param, event_idx) } else { - let new_queue = |reg| { - let Some(split_queue) = SplitQueue::new(reg, &ram, event_idx)? else { - return Ok(None); - }; - Ok(Some(Queue::new(split_queue, reg, &ram))) - }; - let queues: Result> = queue_regs.iter().map(new_queue).collect(); - self.event_loop(&mut (queues?), &ram, param) + self.event_loop::(param, event_idx) } } @@ -656,7 +697,8 @@ where if let Err(e) = self.loop_until_reset() { log::error!("worker {}: {e:?}", self.context.dev.name()); } - // The memory layout lock is released by now. + // The worker has dropped the queues and its guest memory + // snapshot by now. if let Some(seq) = self.context.reset_seq.take() { self.context.ack_reset(seq); } diff --git a/alioth/src/virtio/dev/dev_test.rs b/alioth/src/virtio/dev/dev_test.rs index fa2a3fff..d9c2a3a9 100644 --- a/alioth/src/virtio/dev/dev_test.rs +++ b/alioth/src/virtio/dev/dev_test.rs @@ -22,13 +22,17 @@ use assert_matches::assert_matches; use mio::Registry; use mio::event::Event; +use crate::mem::mapped::{ArcMemPages, RamBus}; +use crate::mem::{MemRegion, MemRegionType, Memory}; use crate::sync::notifier::Notifier; use crate::virtio::dev::entropy::{EntropyConfig, EntropyFeature}; use crate::virtio::dev::{ - Register, StartParam, Virtio, WakeEvent, WorkerCtl, WorkerParam, WorkerReport, + ActiveBackend, BackendEvent, Context, KickWorker, Register, StartParam, Virtio, WakeEvent, + WorkerCtl, WorkerParam, WorkerReport, WorkerState, }; -use crate::virtio::queue::VirtQueue; -use crate::virtio::tests::{fixture_queues, fixture_ram_bus}; +use crate::virtio::queue::split::SplitQueue; +use crate::virtio::queue::{Queue, VirtQueue}; +use crate::virtio::tests::{MEM_SIZE, fixture_queues, fixture_ram_bus}; use crate::virtio::worker::mio::{ActiveMio, Mio, VirtioMio}; use crate::virtio::{DeviceId, Error, IrqSender, Result, VirtioFeature, error}; @@ -59,6 +63,8 @@ fn test_virtio_register_get_driver_feature() { struct FakeDev { /// Makes activation fail, like a backend rejecting the driver setup. fail_activate: Arc, + /// Counts successful activations. + activations: Arc, /// Counts the calls to [`VirtioMio::reset`]. resets: Arc, } @@ -96,40 +102,41 @@ impl Virtio for FakeDev { } impl VirtioMio for FakeDev { - fn activate<'m, Q, S>( + fn activate( &mut self, _feature: u128, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { if self.fail_activate.load(Ordering::Acquire) { return error::InvalidBuffer.fail(); } + self.activations.fetch_add(1, Ordering::AcqRel); Ok(()) } - fn handle_queue<'m, Q, S>( + fn handle_queue( &mut self, _index: u16, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) } - fn handle_event<'m, Q, S>( + fn handle_event( &mut self, _event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) @@ -169,32 +176,42 @@ impl IrqSender for TestIrqSender { } struct TestWorker { - ctl: WorkerCtl, + ctl: Arc>, handle: JoinHandle<()>, irq_sender: Arc, + ram_bus: Arc, fail_activate: Arc, + activations: Arc, resets: Arc, } impl TestWorker { fn spawn() -> Self { + Self::spawn_on(Arc::new(fixture_ram_bus())) + } + + /// Spawns a worker whose queue is in the fixture memory of `ram_bus`. + fn spawn_on(ram_bus: Arc) -> Self { let dev = FakeDev::default(); let fail_activate = dev.fail_activate.clone(); + let activations = dev.activations.clone(); let resets = dev.resets.clone(); let (event_tx, event_rx) = flume::unbounded(); let (report_tx, report_rx) = flume::unbounded(); let param = WorkerParam { event_rx, report_tx, - memory: Arc::new(fixture_ram_bus()), + memory: ram_bus.clone(), queue_regs: Arc::from(fixture_queues(1)), }; let (handle, notifier) = dev.spawn_worker(param).unwrap(); TestWorker { - ctl: WorkerCtl::new("fake".into(), event_tx, notifier, report_rx), + ctl: Arc::new(WorkerCtl::new("fake".into(), event_tx, notifier, report_rx)), handle, irq_sender: Arc::default(), + ram_bus, fail_activate, + activations, resets, } } @@ -222,6 +239,15 @@ impl TestWorker { self.irq_sender.config_irqs.load(Ordering::Acquire) } + /// Waits until the device has been activated `count` times. + fn wait_activations(&self, count: u32) { + let deadline = Instant::now() + TIMEOUT; + while self.activations.load(Ordering::Acquire) < count { + assert!(Instant::now() < deadline, "device not activated"); + thread::sleep(Duration::from_millis(1)); + } + } + /// Waits until the worker reports a failure. fn wait_failed(&self) { let deadline = Instant::now() + TIMEOUT; @@ -287,6 +313,144 @@ fn test_worker_survives_activation_failure() { worker.shutdown(); } +#[test] +fn test_worker_fails_without_rings() { + let worker = TestWorker::spawn(); + worker.start(); + worker.wait_activations(1); + + // The driver takes away the memory of the rings while the device runs. + // Before the next event, the worker tries to move the queue to the new + // layout, and fails. + let pages = worker.ram_bus.update(|ram| ram.remove(0)).unwrap(); + assert!(worker.ctl.wake(WakeEvent::Notify { q_index: 0 })); + worker.wait_failed(); + assert_eq!(worker.resets(), 1); + + // A reset brings the device back into service. + worker.reset(); + assert!(!worker.ctl.failed()); + assert_eq!(worker.config_irqs(), 1); + + worker.ram_bus.update(|ram| ram.add(0, pages)).unwrap(); + worker.start(); + worker.wait_activations(2); + worker.reset(); + assert_eq!(worker.resets(), 2); + assert!(!worker.ctl.failed()); + worker.shutdown(); +} + +#[test] +fn test_worker_kicked_after_mem_update() { + let memory = Memory::new(); + let add_ram = |gpa, size| { + let pages = ArcMemPages::from_anonymous(size, None, None).unwrap(); + let region = MemRegion::with_ram(pages, MemRegionType::Ram); + memory.add_region(gpa, Arc::new(region)).unwrap(); + }; + add_ram(0, MEM_SIZE); + let worker = TestWorker::spawn_on(memory.ram_bus()); + let kick = KickWorker { + ctl: Arc::downgrade(&worker.ctl), + }; + memory.register_update_callback(Box::new(kick)).unwrap(); + worker.start(); + worker.wait_activations(1); + + // The worker is idle, but drops the layout it started with once memory + // is plugged. + let old_ram = Arc::downgrade(&memory.ram_bus().load()); + add_ram(4 << 20, 1 << 20); + let deadline = Instant::now() + TIMEOUT; + while old_ram.strong_count() > 0 { + assert!(Instant::now() < deadline, "old layout still in use"); + thread::sleep(Duration::from_millis(1)); + } + + // Memory updates do not fail once the worker is gone. + worker.shutdown(); + assert_matches!(memory.remove_region(4 << 20), Ok(_)); +} + +/// An event that does not come from the worker notifier. +struct DeviceEvent; + +impl BackendEvent for DeviceEvent { + fn token(&self) -> u64 { + 0 + } +} + +/// Counts how often the worker takes the queues to move them to the +/// latest guest memory layout. +#[derive(Debug, Default)] +struct SyncCounter { + syncs: u32, +} + +impl ActiveBackend for SyncCounter { + type Event = DeviceEvent; + type Q = SplitQueue; + + fn queues(&mut self) -> &mut [Option>] { + self.syncs += 1; + &mut [] + } + + fn handle_event(&mut self, _: &mut FakeDev, _: &DeviceEvent) -> Result<()> { + Ok(()) + } + + fn handle_queue(&mut self, _: &mut FakeDev, _: u16) -> Result<()> { + Ok(()) + } +} + +#[test] +fn test_handle_event_syncs_ram_once_per_layout() { + let (event_tx, event_rx) = flume::unbounded(); + let ram_bus = Arc::new(fixture_ram_bus()); + let mut context = Context { + dev: FakeDev::default(), + param: WorkerParam { + event_rx, + report_tx: flume::unbounded().0, + memory: ram_bus.clone(), + queue_regs: Arc::new([]), + }, + state: WorkerState::Running, + reset_seq: None, + ram_generation: None, + }; + let mut backend = SyncCounter::default(); + + // The first event of an event loop syncs, later ones do not while the + // layout stays the same. + for _ in 0..2 { + context.handle_event(&DeviceEvent, &mut backend).unwrap(); + } + assert_eq!(backend.syncs, 1); + + // A layout change is picked up by the next event, and only by it. + ram_bus.update(|ram| ram.remove(0)).unwrap(); + for _ in 0..2 { + context.handle_event(&DeviceEvent, &mut backend).unwrap(); + } + assert_eq!(backend.syncs, 2); + + // A new event loop starts with a sync, even if the layout is unchanged. + let param = StartParam { + feature: 0, + irq_sender: Arc::new(TestIrqSender::default()), + notifiers: None, + }; + event_tx.send(WakeEvent::Start { param }).unwrap(); + assert!(context.wait_start().is_some()); + context.handle_event(&DeviceEvent, &mut backend).unwrap(); + assert_eq!(backend.syncs, 3); +} + #[test] fn test_worker_exit_completes_reset() { let worker = TestWorker::spawn(); diff --git a/alioth/src/virtio/dev/entropy.rs b/alioth/src/virtio/dev/entropy.rs index 70b16728..97949592 100644 --- a/alioth/src/virtio/dev/entropy.rs +++ b/alioth/src/virtio/dev/entropy.rs @@ -110,25 +110,21 @@ impl Virtio for Entropy { } impl VirtioMio for Entropy { - fn activate<'m, Q, S>( + fn activate( &mut self, _feature: u128, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { @@ -138,13 +134,13 @@ impl VirtioMio for Entropy { queue.handle_desc(index, active_mio.irq_sender, copy_from_reader(&self.source)) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, _event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) diff --git a/alioth/src/virtio/dev/entropy_test.rs b/alioth/src/virtio/dev/entropy_test.rs index 5f809c8d..b84b60f7 100644 --- a/alioth/src/virtio/dev/entropy_test.rs +++ b/alioth/src/virtio/dev/entropy_test.rs @@ -13,25 +13,27 @@ // limitations under the License. use std::ffi::CString; -use std::fs::OpenOptions; +use std::fs::{File, OpenOptions}; use std::io::Write; use std::os::unix::fs::OpenOptionsExt; use std::sync::Arc; +use std::thread::{self, JoinHandle}; use std::time::Duration; use assert_matches::assert_matches; -use flume::TryRecvError; +use flume::{Receiver, Sender, TryRecvError}; use tempfile::TempDir; use crate::ffi; use crate::mem::emulated::{Action, Mmio}; +use crate::mem::mapped::{ArcMemPages, RamBus}; use crate::sync::notifier::Notifier; use crate::virtio::dev::entropy::{EntropyConfig, EntropySpec}; use crate::virtio::dev::{DevSpec, StartParam, Virtio, WakeEvent, WorkerParam}; use crate::virtio::queue::QueueReg; use crate::virtio::queue::split::SplitQueue; use crate::virtio::queue::tests::GuestQueue; -use crate::virtio::tests::{DATA_ADDR, FakeIrqSender, fixture_queues, fixture_ram_bus}; +use crate::virtio::tests::{DATA_ADDR, FakeIrqSender, MEM_SIZE, fixture_queues, fixture_ram_bus}; use crate::virtio::{DeviceId, FEATURE_BUILT_IN, VirtioFeature}; #[test] @@ -46,7 +48,7 @@ fn entry_config_test() { #[test] fn entropy_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(1)); let mut guest_q = GuestQueue::new( @@ -114,6 +116,17 @@ fn entropy_test() { assert_eq!(used0.id, id0); assert_eq!(used0.len, s0.len() as u32); + // The guest memory layout can change while the device is running. + let pages = ArcMemPages::from_anonymous(4 << 10, None, None).unwrap(); + let (done_tx, done_rx) = flume::bounded(1); + let bus = ram_bus.clone(); + thread::spawn(move || { + let ret = bus.update(|ram| ram.add(MEM_SIZE as u64, pages)); + done_tx.send(ret).unwrap(); + }); + let ret = done_rx.recv_timeout(Duration::from_secs(5)); + assert_matches!(ret, Ok(Ok(()))); + writer.write_all(s1.as_bytes()).unwrap(); writer.flush().unwrap(); let id1 = guest_q.add_desc(&[], &[(buf1_addr, 4 << 10)]); @@ -135,3 +148,119 @@ fn entropy_test() { assert_eq!(String::from_utf8_lossy(buf.as_slice()), s); } } + +/// A running entropy worker with a guest queue, fed through a FIFO. +struct EntropyWorker { + guest_q: GuestQueue, + writer: File, + tx: Sender>, + irq_rx: Receiver, + notifier: Arc, + handle: JoinHandle<()>, + _temp_dir: TempDir, +} + +impl EntropyWorker { + /// Starts a worker whose queue is in the fixture memory of `ram_bus`. + fn start(ram_bus: &Arc) -> Self { + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(1)); + let ram = ram_bus.load(); + let guest_q = GuestQueue::new( + SplitQueue::new(®s[0], &ram, false).unwrap().unwrap(), + ®s[0], + ); + + let temp_dir = TempDir::new().unwrap(); + let pipe_path = temp_dir.path().join("urandom"); + let pipe_path_c = CString::new(pipe_path.as_os_str().as_encoded_bytes()).unwrap(); + ffi!(unsafe { libc::mkfifo(pipe_path_c.as_ptr(), 0o600) }).unwrap(); + let param = EntropySpec { + source: Some(pipe_path.clone().into()), + }; + let dev = param.build("entropy").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev + .spawn_worker(WorkerParam { + event_rx: rx, + report_tx: flume::unbounded().0, + memory: ram_bus.clone(), + queue_regs: regs, + }) + .unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender: Arc::new(FakeIrqSender { q_tx: irq_tx }), + notifiers: None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + notifier.notify().unwrap(); + let writer = OpenOptions::new() + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(&pipe_path) + .unwrap(); + EntropyWorker { + guest_q, + writer, + tx, + irq_rx, + notifier, + handle, + _temp_dir: temp_dir, + } + } + + /// Asks the device to fill the buffer at `gpa` with `s`. + fn request(&mut self, s: &str, gpa: u64) { + self.writer.write_all(s.as_bytes()).unwrap(); + self.writer.flush().unwrap(); + let id = self.guest_q.add_desc(&[], &[(gpa, 4 << 10)]); + self.tx.send(WakeEvent::Notify { q_index: 0 }).unwrap(); + self.notifier.notify().unwrap(); + let irq = self.irq_rx.recv_timeout(Duration::from_secs(1)); + assert_eq!(irq, Ok(0), "no response for a buffer at {gpa:#x}"); + let used = self.guest_q.get_used().unwrap(); + assert_eq!(used.id, id); + assert_eq!(used.len, s.len() as u32); + } + + fn shutdown(self) { + self.tx.send(WakeEvent::Shutdown).unwrap(); + self.notifier.notify().unwrap(); + self.handle.join().unwrap(); + } +} + +const HOTPLUG_GPA: u64 = 4 << 20; +const HOTPLUG_SIZE: usize = 1 << 20; + +#[test] +fn entropy_memory_hotplug_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let mut worker = EntropyWorker::start(&ram_bus); + let s0 = "before hot-plug"; + worker.request(s0, DATA_ADDR); + + // The driver uses memory plugged while the device runs right away. + let pages = ArcMemPages::from_anonymous(HOTPLUG_SIZE, None, None).unwrap(); + ram_bus.update(|ram| ram.add(HOTPLUG_GPA, pages)).unwrap(); + let s1 = "right after hot-plug"; + worker.request(s1, HOTPLUG_GPA); + let mut buf = vec![0u8; s1.len()]; + ram_bus.read(HOTPLUG_GPA, &mut buf).unwrap(); + assert_eq!(buf, s1.as_bytes()); + + // So is unplugging memory. + ram_bus.update(|ram| ram.remove(HOTPLUG_GPA)).unwrap(); + let s2 = "after hot-unplug"; + worker.request(s2, DATA_ADDR + (4 << 10)); + + worker.shutdown(); + for (s, gpa) in [(s0, DATA_ADDR), (s2, DATA_ADDR + (4 << 10))] { + let mut buf = vec![0u8; s.len()]; + ram_bus.read(gpa, &mut buf).unwrap(); + assert_eq!(buf, s.as_bytes()); + } +} diff --git a/alioth/src/virtio/dev/fs/fs.rs b/alioth/src/virtio/dev/fs/fs.rs index 5f2ab511..4922bca6 100644 --- a/alioth/src/virtio/dev/fs/fs.rs +++ b/alioth/src/virtio/dev/fs/fs.rs @@ -338,38 +338,30 @@ impl VirtioMio for Fs where F: Fuse + Debug + Send + Sync + 'static, { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, _active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.driver_feature = FsFeature::from_bits_retain(feature); Ok(()) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, _event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { unreachable!() } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { diff --git a/alioth/src/virtio/dev/fs/vu.rs b/alioth/src/virtio/dev/fs/vu.rs index 22dc47d2..0f8b3cce 100644 --- a/alioth/src/virtio/dev/fs/vu.rs +++ b/alioth/src/virtio/dev/fs/vu.rs @@ -174,13 +174,9 @@ impl Virtio for VuFs { } impl VirtioMio for VuFs { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.frontend.activate(feature, active_mio)?; @@ -195,13 +191,13 @@ impl VirtioMio for VuFs { Ok(()) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, event: &Event, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let q_index = event.token().0; @@ -236,13 +232,9 @@ impl VirtioMio for VuFs { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.frontend.handle_queue(index, active_mio) diff --git a/alioth/src/virtio/dev/net/tap.rs b/alioth/src/virtio/dev/net/tap.rs index b0042193..2cecc044 100644 --- a/alioth/src/virtio/dev/net/tap.rs +++ b/alioth/src/virtio/dev/net/tap.rs @@ -246,13 +246,9 @@ impl VirtioMio for Net { let _ = registry.deregister(&mut SourceFd(&self.tap_sockets[0].as_raw_fd())); } - fn activate<'m, Q, S>( - &mut self, - feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.driver_feature = NetFeature::from_bits_retain(feature); @@ -266,13 +262,13 @@ impl VirtioMio for Net { Ok(()) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, event: &Event, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let token = event.token().0; @@ -304,13 +300,9 @@ impl VirtioMio for Net { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { @@ -338,14 +330,10 @@ impl VirtioMio for Net { } impl VirtioIoUring for Net { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - _ring: &mut ActiveIoUring<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, _ring: &mut ActiveIoUring<'_, Q, S>) -> Result<()> where S: IrqSender, - Q: VirtQueue<'m>, + Q: VirtQueue, { self.driver_feature = NetFeature::from_bits_retain(feature); let socket = &mut self.tap_sockets[0]; diff --git a/alioth/src/virtio/dev/net/vmnet.rs b/alioth/src/virtio/dev/net/vmnet.rs index d83300f7..5c712dfd 100644 --- a/alioth/src/virtio/dev/net/vmnet.rs +++ b/alioth/src/virtio/dev/net/vmnet.rs @@ -285,13 +285,9 @@ impl VirtioMio for Net { let _ = registry.deregister(&mut self.rx_notifier); } - fn activate<'m, Q, S>( - &mut self, - _feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, _feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let registry = active_mio.poll.registry(); @@ -342,13 +338,13 @@ impl VirtioMio for Net { Ok(()) } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, event: &Event, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let token = event.token().0; @@ -365,13 +361,9 @@ impl VirtioMio for Net { Ok(()) } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let Some(Some(queue)) = active_mio.queues.get_mut(index as usize) else { diff --git a/alioth/src/virtio/dev/vsock/uds_vsock.rs b/alioth/src/virtio/dev/vsock/uds_vsock.rs index 9481eeb1..c82005bd 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock.rs @@ -157,16 +157,16 @@ impl UdsVsock { Ok(()) } - fn handle_conn_request<'m, Q, S>( + fn handle_conn_request( &mut self, token: Token, mut pending: PendingConn, registry: &Registry, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { // The socket is non-blocking, so a request line can arrive in pieces. @@ -234,14 +234,14 @@ impl UdsVsock { Ok(()) } - fn respond_rst<'m, Q, S>( + fn respond_rst( &self, hdr: &VsockHeader, irq_sender: &S, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let resp = VsockHeader { @@ -256,14 +256,9 @@ impl UdsVsock { self.respond(&resp, irq_sender, rx_q) } - fn respond<'m, Q, S>( - &self, - hdr: &VsockHeader, - irq_sender: &S, - rx_q: &mut Queue<'_, 'm, Q>, - ) -> Result<()> + fn respond(&self, hdr: &VsockHeader, irq_sender: &S, rx_q: &mut Queue) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let mut hdr_buf = hdr.as_bytes(); @@ -285,15 +280,15 @@ impl UdsVsock { Ok(()) } - fn handle_tx_response<'m, Q, S>( + fn handle_tx_response( &mut self, hdr: &VsockHeader, registry: &Registry, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let host_port = hdr.dst_port; @@ -378,15 +373,15 @@ impl UdsVsock { Ok(()) } - fn handle_tx_shutdown<'m, Q, S>( + fn handle_tx_shutdown( &mut self, hdr: &VsockHeader, registry: &Registry, irq_sender: &S, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let host_port = hdr.dst_port; @@ -423,15 +418,15 @@ impl UdsVsock { Ok(()) } - fn handle_tx_request<'m, Q, S>( + fn handle_tx_request( &mut self, hdr: &VsockHeader, registry: &Registry, irq_sender: &S, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let host_port = hdr.dst_port; @@ -489,15 +484,15 @@ impl UdsVsock { Ok(()) } - fn handle_tx_desc<'m, Q, S>( + fn handle_tx_desc( &mut self, desc: &mut DescChain, registry: &Registry, irq_sender: &S, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let name = &*self.name; @@ -544,9 +539,9 @@ impl UdsVsock { } } - fn handle_tx<'m, Q, S>(&mut self, active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>) -> Result<()> + fn handle_tx(&mut self, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let [Some(rx_q), Some(tx_q), ..] = active_mio.queues else { @@ -561,15 +556,15 @@ impl UdsVsock { }) } - fn transfer_rx_data<'m, Q, S>( + fn transfer_rx_data( &mut self, host_port: u32, guest_port: u32, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { fn copy_to_rx( @@ -666,16 +661,16 @@ impl UdsVsock { Ok(()) } - fn process_rx_data<'m, Q, S>( + fn process_rx_data( &mut self, host_port: u32, guest_port: u32, registry: &Registry, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.transfer_rx_data(host_port, guest_port, rx_q, irq_sender)?; @@ -699,14 +694,14 @@ impl UdsVsock { Ok(()) } - fn flush_rx_data<'m, Q, S>( + fn flush_rx_data( &mut self, registry: &Registry, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { if self.connections.is_empty() { @@ -730,17 +725,17 @@ impl UdsVsock { Ok(()) } - fn transfer_tx_data<'m, Q, S>( + fn transfer_tx_data( &mut self, hdr: &VsockHeader, body: &[u8], buffers: &[IoSlice], registry: &Registry, - rx_q: &mut Queue<'_, 'm, Q>, + rx_q: &mut Queue, irq_sender: &S, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { fn copy_to_conn( @@ -907,13 +902,9 @@ impl Virtio for UdsVsock { } impl VirtioMio for UdsVsock { - fn activate<'m, Q, S>( - &mut self, - _feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, _feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { active_mio.poll.registry().register( @@ -924,13 +915,13 @@ impl VirtioMio for UdsVsock { Ok(()) } - fn handle_event<'m, Q, S>( + fn handle_event( &mut self, event: &Event, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let token = event.token(); @@ -953,13 +944,9 @@ impl VirtioMio for UdsVsock { } } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let index = VsockVirtq::from(index); diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs index d6a904f9..4b82f026 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs @@ -95,8 +95,8 @@ fn guest_hdr(op: VsockOp, guest_port: u32, host_port: u32) -> VsockHeader { struct VsockTest<'m> { ram: &'m Ram, sock_path: PathBuf, - rx_q: GuestQueue<'m, SplitQueue<'m>>, - tx_q: GuestQueue<'m, SplitQueue<'m>>, + rx_q: GuestQueue, + tx_q: GuestQueue, tx: Sender>, irq_rx: Receiver, notifier: Arc, @@ -106,7 +106,7 @@ struct VsockTest<'m> { impl<'m> VsockTest<'m> { /// Starts a device worker listening on a socket in a temporary directory. - fn new(ram_bus: &Arc, ram: &'m Ram) -> Self { + fn new(ram_bus: &Arc, ram: &'m Arc) -> Self { let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; @@ -296,7 +296,7 @@ impl<'m> VsockTest<'m> { #[test] fn vsock_conn_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); // 0. Setup connection @@ -387,7 +387,7 @@ fn vsock_conn_test() { #[test] fn vsock_host_close_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); // Establish a host-initiated connection @@ -415,7 +415,7 @@ fn vsock_host_close_test() { #[test] fn vsock_host_close_no_desc_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); // Establish a host-initiated connection @@ -458,7 +458,7 @@ fn vsock_host_close_no_desc_test() { #[test] fn vsock_partial_conn_request_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); let mut h2g_stream = t.connect(); @@ -487,7 +487,7 @@ fn vsock_partial_conn_request_test() { #[test] fn vsock_simultaneous_conn_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); let buf_addrs = [RX_ADDR, RX_ADDR + 2048]; @@ -523,7 +523,7 @@ fn vsock_simultaneous_conn_test() { #[test] fn vsock_conn_request_eof_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); // A client that connects and disconnects without saying anything. @@ -547,7 +547,7 @@ fn vsock_conn_request_eof_test() { #[test] fn vsock_conn_request_close_test() { let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut t = VsockTest::new(&ram_bus, &ram); // A client that sends a complete request and hangs up before the guest diff --git a/alioth/src/virtio/dev/vsock/vhost_vsock.rs b/alioth/src/virtio/dev/vsock/vhost_vsock.rs index 49c60397..0e497fff 100644 --- a/alioth/src/virtio/dev/vsock/vhost_vsock.rs +++ b/alioth/src/virtio/dev/vsock/vhost_vsock.rs @@ -15,7 +15,6 @@ use std::os::fd::{AsFd, AsRawFd, FromRawFd, OwnedFd}; use std::path::Path; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::thread::JoinHandle; use libc::{EFD_CLOEXEC, EFD_NONBLOCK, eventfd}; @@ -142,13 +141,9 @@ impl Virtio for VhostVsock { } impl VirtioMio for VhostVsock { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { self.vhost_dev.set_features(&(feature as u64))?; @@ -163,7 +158,7 @@ impl VirtioMio for VhostVsock { let Some(queue) = queue else { continue; }; - let reg = queue.reg(); + let addrs = queue.addrs(); let index = index as u32; active_mio.irq_sender.queue_irqfd(index as _, |fd| { self.vhost_dev.set_virtq_call(&VirtqFile { @@ -175,17 +170,17 @@ impl VirtioMio for VhostVsock { self.vhost_dev.set_virtq_num(&VirtqState { index, - val: reg.size.load(Ordering::Acquire) as _, + val: addrs.size as _, })?; self.vhost_dev .set_virtq_base(&VirtqState { index, val: 0 })?; - let mem = active_mio.mem; + let mem = queue.ram(); let virtq_addr = VirtqAddr { index, flags: 0, - desc_hva: mem.translate(reg.desc.load(Ordering::Acquire))? as _, - used_hva: mem.translate(reg.device.load(Ordering::Acquire))? as _, - avail_hva: mem.translate(reg.driver.load(Ordering::Acquire))? as _, + desc_hva: mem.translate(addrs.desc)? as _, + used_hva: mem.translate(addrs.device)? as _, + avail_hva: mem.translate(addrs.driver)? as _, log_guest_addr: 0, }; self.vhost_dev.set_virtq_addr(&virtq_addr)?; @@ -231,13 +226,13 @@ impl VirtioMio for VhostVsock { } } - fn handle_event<'a, 'm, Q, S>( + fn handle_event<'a, Q, S>( &mut self, event: &Event, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let q_index = event.token(); @@ -249,13 +244,13 @@ impl VirtioMio for VhostVsock { Ok(()) } - fn handle_queue<'m, Q, S>( + fn handle_queue( &mut self, index: u16, - _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + _active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { match index { diff --git a/alioth/src/virtio/queue/packed.rs b/alioth/src/virtio/queue/packed.rs index 67a0fb80..3d0f053a 100644 --- a/alioth/src/virtio/queue/packed.rs +++ b/alioth/src/virtio/queue/packed.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::marker::PhantomData; +use std::sync::Arc; use std::sync::atomic::{AtomicU16, Ordering}; use bitfield::bitfield; @@ -20,7 +20,7 @@ use zerocopy::{FromBytes, IntoBytes}; use crate::consts; use crate::mem::mapped::Ram; -use crate::virtio::queue::{DescChain, DescFlag, QueueReg, VirtQueue}; +use crate::virtio::queue::{DescChain, DescFlag, QueueReg, RingAddrs, VirtQueue}; use crate::virtio::{Result, error}; /// The maximum size of a packed virtqueue (VirtIO spec 1.4, Sec 2.8). @@ -90,43 +90,24 @@ struct DescEvent { } #[derive(Debug)] -pub struct PackedQueue<'m> { +pub struct PackedQueue { size: u16, desc: *mut Desc, enable_event_idx: bool, notification: *mut DescEvent, interrupt: *mut DescEvent, - _phantom: PhantomData<&'m ()>, + /// The snapshot the pointers above point into. + ram: Arc, } -impl<'m> PackedQueue<'m> { - pub fn new(reg: &QueueReg, ram: &'m Ram, event_idx: bool) -> Result>> { - if !reg.enabled.load(Ordering::Acquire) { +impl PackedQueue { + /// Creates the queue from its registers, or returns `None` if the queue + /// is disabled. + pub fn new(reg: &QueueReg, ram: &Arc, event_idx: bool) -> Result> { + let Some(addrs) = RingAddrs::load(reg) else { return Ok(None); - } - let size = reg.size.load(Ordering::Acquire); - let desc = reg.desc.load(Ordering::Acquire); - let device = reg.device.load(Ordering::Acquire); - let driver = reg.driver.load(Ordering::Acquire); - // The registers may be cleared by a concurrent reset, which disables - // the queue first. - if !reg.enabled.load(Ordering::Acquire) { - return Ok(None); - } - if size == 0 || size > PACKED_QUEUE_SIZE_MAX { - return error::InvalidQueueSize { size }.fail(); - } - let notification: *mut DescEvent = ram.get_ptr(device)?; - Ok(Some(PackedQueue { - size, - // The queue accesses descriptors up to the queue size, so the - // whole ring must be in guest memory. - desc: ram.get_array_ptr(desc, size as u64)?, - enable_event_idx: event_idx, - notification, - interrupt: ram.get_ptr(driver)?, - _phantom: PhantomData, - })) + }; + ::new(&addrs, ram, event_idx).map(Some) } fn flag_is_avail(&self, flag: DescFlag, wrap_counter: bool) -> bool { @@ -143,18 +124,40 @@ impl<'m> PackedQueue<'m> { } } -impl<'m> VirtQueue<'m> for PackedQueue<'m> { +impl VirtQueue for PackedQueue { type Index = WrappedIndex; const INIT_INDEX: WrappedIndex = WrappedIndex::INIT; + fn new(addrs: &RingAddrs, ram: &Arc, event_idx: bool) -> Result { + let size = addrs.size; + if size == 0 || size > PACKED_QUEUE_SIZE_MAX { + return error::InvalidQueueSize { size }.fail(); + } + let notification: *mut DescEvent = ram.get_ptr(addrs.device)?; + Ok(PackedQueue { + size, + // The queue accesses descriptors up to the queue size, so the + // whole ring must be in guest memory. + desc: ram.get_array_ptr(addrs.desc, size as u64)?, + enable_event_idx: event_idx, + notification, + interrupt: ram.get_ptr(addrs.driver)?, + ram: ram.clone(), + }) + } + + fn ram(&self) -> &Arc { + &self.ram + } + fn desc_avail(&self, index: WrappedIndex) -> bool { let flag = unsafe { &(*self.desc.offset(index.offset() as isize)).flag }.load(Ordering::Acquire); self.flag_is_avail(DescFlag::from_bits_retain(flag), index.wrap_counter()) } - fn get_avail(&self, index: Self::Index, ram: &'m Ram) -> Result>> { + fn get_avail<'m>(&self, index: Self::Index, ram: &'m Ram) -> Result>> { if !self.desc_avail(index) { return Ok(None); } @@ -210,6 +213,7 @@ impl<'m> VirtQueue<'m> for PackedQueue<'m> { delta, readable: ram.translate_iov(&readable)?, writable: ram.translate_iov_mut(&writeable)?, + ram, })) } diff --git a/alioth/src/virtio/queue/packed_test.rs b/alioth/src/virtio/queue/packed_test.rs index 8f15a3d3..4bedda36 100644 --- a/alioth/src/virtio/queue/packed_test.rs +++ b/alioth/src/virtio/queue/packed_test.rs @@ -63,14 +63,14 @@ fn index_wrapping_sub( ); } -impl<'m> PackedQueue<'m> { +impl PackedQueue { fn flag_is_used(&self, flag: DescFlag, wrap_counter: bool) -> bool { flag.contains(DescFlag::AVAIL) == wrap_counter && flag.contains(DescFlag::USED) == wrap_counter } } -impl<'m> VirtQueueGuest<'m> for PackedQueue<'m> { +impl VirtQueueGuest for PackedQueue { fn add_desc( &mut self, index: WrappedIndex, @@ -143,7 +143,7 @@ impl<'m> VirtQueueGuest<'m> for PackedQueue<'m> { fn disabled_queue() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.enabled.store(false, Ordering::Relaxed); let split_queue = PackedQueue::new(reg, &ram, false); @@ -158,7 +158,7 @@ fn disabled_queue() { fn queue_size(#[case] size: u16, #[case] valid: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.size.store(size, Ordering::Relaxed); @@ -177,7 +177,7 @@ fn queue_size(#[case] size: u16, #[case] valid: bool) { fn ring_at_end_of_memory(#[case] entries: u16, #[case] valid: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; // Place the ring so that only its first `entries` descriptors are in @@ -201,7 +201,7 @@ fn ring_at_end_of_memory(#[case] entries: u16, #[case] valid: bool) { fn chain_longer_than_ring() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.size.store(4, Ordering::Relaxed); let q = PackedQueue::new(reg, &ram, false).unwrap().unwrap(); @@ -249,7 +249,7 @@ fn write_indirect_table(ram: &Ram, num: u16) { fn unexpected_indirect_desc(#[case] flag_0: DescFlag, #[case] flag_1: DescFlag) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let q = PackedQueue::new(&queues[0], &ram, false).unwrap().unwrap(); write_indirect_table(&ram, 1); @@ -275,7 +275,7 @@ fn unexpected_indirect_desc(#[case] flag_0: DescFlag, #[case] flag_1: DescFlag) fn indirect_table_len(#[case] num: u32, #[case] valid: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.size.store(4, Ordering::Relaxed); let q = PackedQueue::new(reg, &ram, false).unwrap().unwrap(); @@ -295,7 +295,7 @@ fn indirect_table_len(#[case] num: u32, #[case] valid: bool) { fn enabled_queue() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; let q = PackedQueue::new(reg, &ram, false).unwrap().unwrap(); let mut guest_q = GuestQueue::new(PackedQueue::new(reg, &ram, false).unwrap().unwrap(), reg); @@ -356,7 +356,7 @@ fn enabled_queue() { fn enable_notification() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; let q = PackedQueue::new(reg, &ram, false).unwrap().unwrap(); @@ -396,7 +396,7 @@ fn is_interrupt_enabled( ) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; let q = PackedQueue::new(reg, &ram, enable_event_idx) .unwrap() diff --git a/alioth/src/virtio/queue/queue.rs b/alioth/src/virtio/queue/queue.rs index c2b278be..e59bd1aa 100644 --- a/alioth/src/virtio/queue/queue.rs +++ b/alioth/src/virtio/queue/queue.rs @@ -18,10 +18,11 @@ pub mod split; use std::collections::HashMap; use std::fmt::Debug; use std::io::{ErrorKind, IoSlice, IoSliceMut, Read, Write}; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering, fence}; use crate::bitflags; -use crate::mem::mapped::Ram; +use crate::mem::mapped::{Ram, RamBus}; use crate::virtio::{IrqSender, Result, error}; pub const QUEUE_SIZE_MAX: u16 = 256; @@ -45,25 +46,96 @@ pub struct QueueReg { pub enabled: AtomicBool, } +/// The ring addresses of a queue, copied from its [`QueueReg`]. +/// +/// Transports do not let the driver change the registers of an enabled +/// queue, so the copy stays valid until the device is reset. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RingAddrs { + pub size: u16, + pub desc: u64, + pub driver: u64, + pub device: u64, +} + +impl RingAddrs { + /// Reads the ring addresses, or returns `None` if the queue is disabled. + pub fn load(reg: &QueueReg) -> Option { + if !reg.enabled.load(Ordering::Acquire) { + return None; + } + let addrs = RingAddrs { + size: reg.size.load(Ordering::Acquire), + desc: reg.desc.load(Ordering::Acquire), + driver: reg.driver.load(Ordering::Acquire), + device: reg.device.load(Ordering::Acquire), + }; + // The registers may be cleared by a concurrent reset, which disables + // the queue first. + if !reg.enabled.load(Ordering::Acquire) { + return None; + } + Some(addrs) + } +} + #[derive(Debug)] pub struct DescChain<'m> { id: u16, delta: u16, pub readable: Vec>, pub writable: Vec>, + ram: &'m Ram, } -impl DescChain<'_> { +impl<'m> DescChain<'m> { pub fn id(&self) -> u16 { self.id } + + /// Returns the guest memory snapshot the buffers were translated with. + /// + /// A device accessing guest memory at addresses found in the buffers + /// should use it. + pub fn ram(&self) -> &Ram { + self.ram + } + + /// Erases the lifetime of the guest memory the chain points to. + /// + /// # Safety + /// + /// The caller must keep the snapshot the chain was translated with alive + /// until the returned chain is dropped. + /// + /// Device code must only see the returned chain through a callback that + /// takes `&mut DescChain` of any lifetime, like the `op` of + /// [`Queue::handle_deferred`], so that it cannot keep the buffers. + unsafe fn into_static(self) -> DescChain<'static> { + unsafe { std::mem::transmute::, DescChain<'static>>(self) } + } } -pub trait VirtQueue<'m> { +/// A descriptor chain the device has not completed yet. +/// +/// It holds the snapshot the chain was translated with, so that the buffers +/// stay valid until the device completes the chain. +struct DeferredChain { + chain: DescChain<'static>, + // Dropped after `chain`, since fields are dropped in declaration order. + _ram: Arc, +} + +pub trait VirtQueue: Sized { type Index: Clone + Copy; const INIT_INDEX: Self::Index; + /// Creates the queue on the rings at `addrs` in the snapshot `ram`, + /// which the queue holds on to. + fn new(addrs: &RingAddrs, ram: &Arc, event_idx: bool) -> Result; + /// Returns the guest memory snapshot the rings are in. + fn ram(&self) -> &Arc; fn desc_avail(&self, index: Self::Index) -> bool; - fn get_avail(&self, index: Self::Index, ram: &'m Ram) -> Result>>; + fn get_avail<'m>(&self, index: Self::Index, ram: &'m Ram) -> Result>>; fn set_used(&self, index: Self::Index, id: u16, len: u32); fn enable_notification(&self, enabled: bool); fn interrupt_enabled(&self, index: Self::Index, delta: u16) -> bool; @@ -77,35 +149,79 @@ pub enum Status { Break, } -pub struct Queue<'r, 'm, Q> +/// The device side of a virtqueue. +/// +/// The queue can move to a newer guest memory snapshot, see +/// [`Queue::sync_ram`]. +pub struct Queue where - Q: VirtQueue<'m>, + Q: VirtQueue, { q: Q, avail: Q::Index, used: Q::Index, - reg: &'r QueueReg, - ram: &'m Ram, - deferred: HashMap>, + addrs: RingAddrs, + event_idx: bool, + deferred: HashMap, + ram_bus: Arc, } -impl<'r, 'm, Q> Queue<'r, 'm, Q> +impl Queue where - Q: VirtQueue<'m>, + Q: VirtQueue, { - pub fn new(q: Q, reg: &'r QueueReg, ram: &'m Ram) -> Self { - Self { + /// Creates the queue from its registers on the latest snapshot of + /// `ram_bus`, or returns `None` if the queue is disabled. + pub fn new(reg: &QueueReg, ram_bus: Arc, event_idx: bool) -> Result> { + let Some(addrs) = RingAddrs::load(reg) else { + return Ok(None); + }; + let ram = ram_bus.load(); + let q = Q::new(&addrs, &ram, event_idx)?; + Ok(Some(Queue { q, avail: Q::INIT_INDEX, used: Q::INIT_INDEX, - reg, - ram, + addrs, + event_idx, deferred: HashMap::new(), + ram_bus, + })) + } + + /// Returns the ring addresses. + pub fn addrs(&self) -> &RingAddrs { + &self.addrs + } + + /// Returns the guest memory snapshot the queue works on. + pub fn ram(&self) -> &Ram { + self.q.ram() + } + + /// Moves the queue to the latest snapshot of its [`RamBus`], keeping the + /// ring indices and the deferred chains. Returns `true` if it moved. + /// + /// It fails if the rings are not accessible in the latest snapshot, and + /// the queue stays where it is. + pub fn sync_ram(&mut self) -> Result { + if self.q.ram().generation() == self.ram_bus.generation() { + return Ok(false); } + let ram = self.ram_bus.load(); + self.q = Q::new(&self.addrs, &ram, self.event_idx)?; + Ok(true) } - pub fn reg(&self) -> &QueueReg { - self.reg + /// Forgets the chains the device has not completed, and leaks the + /// snapshots they hold. + /// + /// This is the last resort for buffers that might still be accessed, + /// e.g. by the kernel, but cannot be waited for. + pub fn leak_deferred(&mut self) { + for (_, deferred) in self.deferred.drain() { + std::mem::forget(deferred); + } } pub fn desc_avail(&self) -> bool { @@ -124,12 +240,12 @@ where irq_sender: &impl IrqSender, mut op: impl FnMut(&mut DescChain) -> Result, ) -> Result<()> { - let Some(mut chain) = self.deferred.remove(&id) else { + let Some(mut deferred) = self.deferred.remove(&id) else { return error::InvalidDescriptor { id }.fail(); }; - let len = op(&mut chain)?; - let delta = chain.delta; - self.push_used(chain, len); + let len = op(&mut deferred.chain)?; + let delta = deferred.chain.delta; + self.push_used(deferred.chain, len); fence(Ordering::SeqCst); if self.q.interrupt_enabled(self.used, delta) { irq_sender.queue_irq(q_index); @@ -145,12 +261,42 @@ where ) -> Result<()> { let mut send_irq = false; let mut ret = Ok(()); + let mut ram = self.q.ram().clone(); 'out: loop { if !self.q.desc_avail(self.avail) { break; } self.q.enable_notification(false); - while let Some(mut chain) = self.q.get_avail(self.avail, self.ram)? { + loop { + let mut chain = match self.q.get_avail(self.avail, &ram) { + Ok(Some(chain)) => chain, + Ok(None) => break, + Err(e) => { + // The chain may be in memory published after the + // queue moved to its snapshot, e.g. while the worker + // was handling an earlier event. Catch up and retry. + match self.sync_ram() { + Ok(true) => { + ram = self.q.ram().clone(); + continue; + } + Ok(false) => ret = Err(e), + Err(sync_err) => ret = Err(sync_err), + } + // Chains completed so far still need the interrupt. + self.q.enable_notification(true); + break 'out; + } + }; + // A driver cannot make a chain available again before the + // device has used it. Replacing the deferred chain would + // release its snapshot while the device may still access the + // buffers. + if self.deferred.contains_key(&chain.id) { + ret = error::InvalidDescriptor { id: chain.id }.fail(); + self.q.enable_notification(true); + break 'out; + } let delta = chain.delta; match op(&mut chain) { Err(e) => { @@ -167,7 +313,12 @@ where } } Ok(Status::Deferred) => { - self.deferred.insert(chain.id, chain); + let id = chain.id; + // SAFETY: the deferred chain holds the snapshot the + // chain was translated with. + let chain = unsafe { chain.into_static() }; + let _ram = ram.clone(); + self.deferred.insert(id, DeferredChain { chain, _ram }); } } self.avail = self.q.index_add(self.avail, delta); @@ -182,6 +333,19 @@ where } } +/// Moves `queues` to the latest guest memory snapshot, see +/// [`Queue::sync_ram`]. Returns `true` if any queue moved. +pub fn sync_queues(queues: &mut [Option>]) -> Result +where + Q: VirtQueue, +{ + let mut moved = false; + for queue in queues.iter_mut().flatten() { + moved |= queue.sync_ram()?; + } + Ok(moved) +} + pub fn copy_from_reader(mut reader: impl Read) -> impl FnMut(&mut DescChain) -> Result { move |chain| { let ret = reader.read_vectored(&mut chain.writable); diff --git a/alioth/src/virtio/queue/queue_test.rs b/alioth/src/virtio/queue/queue_test.rs index d14d0793..32ad561a 100644 --- a/alioth/src/virtio/queue/queue_test.rs +++ b/alioth/src/virtio/queue/queue_test.rs @@ -15,27 +15,34 @@ use std::collections::HashMap; use std::io::{ErrorKind, IoSlice, IoSliceMut, Read, Write}; use std::ptr::eq as ptr_eq; +use std::sync::Arc; use std::sync::atomic::Ordering; use assert_matches::assert_matches; use flume::TryRecvError; use rstest::rstest; +use crate::mem::mapped::{ArcMemPages, Ram}; use crate::virtio::Error; use crate::virtio::queue::packed::PackedQueue; use crate::virtio::queue::split::SplitQueue; use crate::virtio::queue::{ - DescChain, Queue, QueueReg, Status, VirtQueue, copy_from_reader, copy_to_writer, + DescChain, Queue, QueueReg, RingAddrs, Status, VirtQueue, copy_from_reader, copy_to_writer, }; use crate::virtio::tests::{DATA_ADDR, FakeIrqSender, fixture_queues, fixture_ram_bus}; +/// Guest memory plugged by tests, beyond the fixture memory. +const EXTRA_GPA: u64 = 4 << 20; +const EXTRA_SIZE: usize = 1 << 20; + +#[derive(Debug)] pub struct UsedDesc { pub id: u16, pub delta: u16, pub len: u32, } -pub trait VirtQueueGuest<'m>: VirtQueue<'m> { +pub trait VirtQueueGuest: VirtQueue { fn add_desc( &mut self, index: Self::Index, @@ -52,9 +59,9 @@ pub trait VirtQueueGuest<'m>: VirtQueue<'m> { fn enable_interrupt(&mut self, enabled: bool, index: Self::Index); } -pub struct GuestQueue<'m, Q> +pub struct GuestQueue where - Q: VirtQueueGuest<'m>, + Q: VirtQueueGuest, { q: Q, avail: Q::Index, @@ -64,9 +71,9 @@ where next_id: u16, } -impl<'m, Q> GuestQueue<'m, Q> +impl GuestQueue where - Q: VirtQueueGuest<'m>, + Q: VirtQueueGuest, { pub fn new(q: Q, reg: &QueueReg) -> Self { let size = reg.size.load(Ordering::Acquire); @@ -81,9 +88,9 @@ where } } -impl<'m, Q> GuestQueue<'m, Q> +impl GuestQueue where - Q: VirtQueueGuest<'m>, + Q: VirtQueueGuest, { pub fn add_desc(&mut self, readable: &[(u64, u32)], writable: &[(u64, u32)]) -> u16 { let mut ids = vec![]; @@ -182,17 +189,16 @@ impl<'a> Read for Reader<'a> { #[test] fn test_copy_from_reader() { - let ram_bus = fixture_ram_bus(); + let ram_bus = Arc::new(fixture_ram_bus()); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; - let mut host_q = Queue::new( - SplitQueue::new(reg, &ram, false).unwrap().unwrap(), - reg, - &ram, - ); + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); - assert!(ptr_eq(host_q.reg(), reg)); + assert_eq!(host_q.addrs(), &RingAddrs::load(reg).unwrap()); + assert!(ptr_eq(host_q.ram(), &*ram)); let (irq_tx, irq_rx) = flume::unbounded(); let irq_sender = FakeIrqSender { q_tx: irq_tx }; @@ -341,15 +347,13 @@ impl<'a> Write for Writer<'a> { #[test] fn test_copy_to_writer() { - let ram_bus = fixture_ram_bus(); + let ram_bus = Arc::new(fixture_ram_bus()); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; - let mut host_q = Queue::new( - SplitQueue::new(reg, &ram, false).unwrap().unwrap(), - reg, - &ram, - ); + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); let (irq_tx, irq_rx) = flume::unbounded(); let irq_sender = FakeIrqSender { q_tx: irq_tx }; @@ -446,12 +450,14 @@ fn test_written_bytes() { let str_0 = "Hello, World!"; let str_1 = "Goodbye, World!"; + let ram = Ram::default(); let mut buf = vec![0u8; str_0.len()]; let mut chain = DescChain { id: 0, delta: 1, readable: vec![], writable: vec![IoSliceMut::new(buf.as_mut_slice())], + ram: &ram, }; let reader = str_0.as_bytes(); assert_matches!( @@ -466,6 +472,7 @@ fn test_written_bytes() { delta: 1, readable: vec![IoSlice::new(str_1.as_bytes())], writable: vec![], + ram: &ram, }; assert_matches!( copy_to_writer(&mut buf)(&mut chain), @@ -474,12 +481,9 @@ fn test_written_bytes() { assert_eq!(buf.as_slice(), str_1.as_bytes()); } -fn check_handle_deferred<'m, Q>( - mut host_q: Queue<'_, 'm, Q>, - mut guest_q: GuestQueue<'m, Q>, - ram: &'m crate::mem::mapped::Ram, -) where - Q: VirtQueueGuest<'m>, +fn check_handle_deferred(mut host_q: Queue, mut guest_q: GuestQueue, ram: &Ram) +where + Q: VirtQueueGuest, { let (irq_tx, irq_rx) = flume::unbounded(); let irq_sender = FakeIrqSender { q_tx: irq_tx }; @@ -560,28 +564,245 @@ fn check_handle_deferred<'m, Q>( #[case(true, false)] #[case(true, true)] fn test_handle_deferred(#[case] packed: bool, #[case] event_idx: bool) { - let ram_bus = fixture_ram_bus(); + let ram_bus = Arc::new(fixture_ram_bus()); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; if packed { - let host_q = Queue::new( - PackedQueue::new(reg, &ram, event_idx).unwrap().unwrap(), - reg, - &ram, - ); + let host_q = Queue::::new(reg, ram_bus.clone(), event_idx) + .unwrap() + .unwrap(); let guest_q = GuestQueue::new( PackedQueue::new(reg, &ram, event_idx).unwrap().unwrap(), reg, ); check_handle_deferred(host_q, guest_q, &ram); } else { - let host_q = Queue::new( - SplitQueue::new(reg, &ram, event_idx).unwrap().unwrap(), - reg, - &ram, - ); + let host_q = Queue::::new(reg, ram_bus.clone(), event_idx) + .unwrap() + .unwrap(); let guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, event_idx).unwrap().unwrap(), reg); check_handle_deferred(host_q, guest_q, &ram); } } + +#[test] +fn test_handle_desc_chain_in_flight() { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let ram = ram_bus.load(); + let reg = &queues[0]; + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + + let id = guest_q.add_desc(&[], &[(DATA_ADDR, 16)]); + host_q + .handle_desc(0, &irq_sender, |_| Ok(Status::Deferred)) + .unwrap(); + + // A buggy driver makes the chain available again while it is in flight. + let delta = guest_q + .q + .add_desc(guest_q.avail, &[id], &[], &[(DATA_ADDR, 16)]); + guest_q.avail = guest_q.q.index_add(guest_q.avail, delta); + let mut seen = false; + let ret = host_q.handle_desc(0, &irq_sender, |_| { + seen = true; + Ok(Status::Deferred) + }); + assert_matches!(ret, Err(Error::InvalidDescriptor { id: i, .. }) if i == id); + + // The device never sees the chain, and the one in flight is kept. + assert!(!seen); + host_q + .handle_deferred(id, 0, &irq_sender, |chain| { + Ok(chain.writable[0].len() as u32) + }) + .unwrap(); + assert_eq!(irq_rx.try_recv(), Ok(0)); + assert_matches!(guest_q.get_used(), Some(UsedDesc { id: i, len: 16, .. }) if i == id); +} + +#[rstest] +#[case(false)] +#[case(true)] +fn test_leak_deferred(#[case] leak: bool) { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let ram = ram_bus.load(); + let reg = &queues[0]; + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); + let (irq_tx, _irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + + // The chain stays in flight, e.g. as an io_uring request. + guest_q.add_desc(&[], &[(DATA_ADDR, 16)]); + host_q + .handle_desc(0, &irq_sender, |_| Ok(Status::Deferred)) + .unwrap(); + assert_eq!(host_q.deferred.len(), 1); + + let snapshot = Arc::downgrade(&ram); + drop(ram); + drop(guest_q); + if leak { + host_q.leak_deferred(); + assert!(host_q.deferred.is_empty()); + } + drop(host_q); + drop(ram_bus); + // Only a leaked chain keeps its snapshot after the queues and the bus + // are gone. + assert_eq!(snapshot.upgrade().is_some(), leak); +} + +#[test] +fn test_sync_ram() { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let reg = &queues[0]; + let fixture_ram = ram_bus.load(); + let mut guest_q = GuestQueue::new( + SplitQueue::new(reg, &fixture_ram, false).unwrap().unwrap(), + reg, + ); + let extra = ArcMemPages::from_anonymous(EXTRA_SIZE, None, None).unwrap(); + ram_bus.update(|ram| ram.add(EXTRA_GPA, extra)).unwrap(); + let ram = ram_bus.load(); + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + + // A chain in the extra memory stays in flight. + let str_0 = "in flight across a layout change"; + ram.write(EXTRA_GPA, str_0.as_bytes()).unwrap(); + let id_0 = guest_q.add_desc(&[(EXTRA_GPA, str_0.len() as u32)], &[]); + host_q + .handle_desc(0, &irq_sender, |_| Ok(Status::Deferred)) + .unwrap(); + + // The extra memory is unplugged, and the queue moves to the new layout. + ram_bus.update(|ram| ram.remove(EXTRA_GPA)).unwrap(); + let old_ram = Arc::downgrade(&ram); + drop(ram); + assert_matches!(host_q.sync_ram(), Ok(true)); + let new_ram = ram_bus.load(); + assert!(ptr_eq(host_q.ram(), &*new_ram)); + assert_matches!(host_q.sync_ram(), Ok(false)); + // The deferred chain still holds the old snapshot. + assert!(old_ram.upgrade().is_some()); + + // The queue continues at the same ring index. + let str_1 = "after the layout change"; + new_ram.write(DATA_ADDR, str_1.as_bytes()).unwrap(); + let id_1 = guest_q.add_desc(&[(DATA_ADDR, str_1.len() as u32)], &[]); + let mut ids = vec![]; + host_q + .handle_desc(0, &irq_sender, |chain| { + ids.push(chain.id()); + assert_eq!(&*chain.readable[0], str_1.as_bytes()); + Ok(Status::Done { len: 0 }) + }) + .unwrap(); + assert_eq!(ids, [id_1]); + assert_eq!(irq_rx.try_recv(), Ok(0)); + assert_matches!(guest_q.get_used(), Some(UsedDesc { id, .. }) if id == id_1); + + // The buffer of the chain in flight is still accessible, and completing + // the chain releases the old snapshot. + host_q + .handle_deferred(id_0, 0, &irq_sender, |chain| { + assert_eq!(&*chain.readable[0], str_0.as_bytes()); + Ok(0) + }) + .unwrap(); + assert_matches!(guest_q.get_used(), Some(UsedDesc { id, .. }) if id == id_0); + assert!(old_ram.upgrade().is_none()); +} + +#[test] +fn test_sync_ram_without_rings() { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let ram = ram_bus.load(); + let reg = &queues[0]; + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + + // The driver takes away the memory of the rings while the queue runs. + ram_bus.update(|ram| ram.remove(0)).unwrap(); + assert_matches!(host_q.sync_ram(), Err(Error::Memory { .. })); + assert!(ptr_eq(host_q.ram(), &*ram)); +} + +#[test] +fn test_handle_desc_catches_up_with_ram_bus() { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let ram = ram_bus.load(); + let reg = &queues[0]; + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + + // The driver posts a buffer in memory plugged after the queue moved to + // its snapshot, e.g. while the worker is handling an earlier event. + let extra = ArcMemPages::from_anonymous(EXTRA_SIZE, None, None).unwrap(); + ram_bus.update(|ram| ram.add(EXTRA_GPA, extra)).unwrap(); + let s = "in plugged memory"; + ram_bus.write(EXTRA_GPA, s.as_bytes()).unwrap(); + let id = guest_q.add_desc(&[(EXTRA_GPA, s.len() as u32)], &[]); + assert!(ptr_eq(host_q.ram(), &*ram)); + + // The queue moves to the latest layout and retries. + host_q + .handle_desc(0, &irq_sender, |chain| { + assert_eq!(&*chain.readable[0], s.as_bytes()); + assert_eq!(chain.ram().generation(), ram_bus.generation()); + Ok(Status::Done { len: 0 }) + }) + .unwrap(); + assert_eq!(host_q.ram().generation(), ram_bus.generation()); + assert_eq!(irq_rx.try_recv(), Ok(0)); + assert_matches!(guest_q.get_used(), Some(UsedDesc { id: used, .. }) if used == id); +} + +#[test] +fn test_handle_desc_irq_before_invalid_chain() { + let ram_bus = Arc::new(fixture_ram_bus()); + let queues = fixture_queues(1); + let ram = ram_bus.load(); + let reg = &queues[0]; + let mut host_q = Queue::::new(reg, ram_bus.clone(), false) + .unwrap() + .unwrap(); + let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + + // A valid chain followed by one outside guest memory. + let s = "valid"; + ram.write(DATA_ADDR, s.as_bytes()).unwrap(); + let id = guest_q.add_desc(&[(DATA_ADDR, s.len() as u32)], &[]); + guest_q.add_desc(&[(1 << 40, 16)], &[]); + + let ret = host_q.handle_desc(0, &irq_sender, |_| Ok(Status::Done { len: 0 })); + assert_matches!(ret, Err(Error::Memory { .. })); + // The driver is still told about the chain completed before the error. + assert_eq!(irq_rx.try_recv(), Ok(0)); + assert_matches!(guest_q.get_used(), Some(UsedDesc { id: used, .. }) if used == id); + assert!(guest_q.notification_enabled()); +} diff --git a/alioth/src/virtio/queue/split.rs b/alioth/src/virtio/queue/split.rs index 11a8931d..2676d347 100644 --- a/alioth/src/virtio/queue/split.rs +++ b/alioth/src/virtio/queue/split.rs @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::marker::PhantomData; use std::mem::size_of; +use std::sync::Arc; use std::sync::atomic::{AtomicU16, Ordering, fence}; use alioth_macros::Layout; @@ -21,7 +21,7 @@ use zerocopy::{FromBytes, Immutable, IntoBytes}; use crate::bitflags; use crate::mem::mapped::Ram; -use crate::virtio::queue::{DescChain, DescFlag, QueueReg, VirtQueue}; +use crate::virtio::queue::{DescChain, DescFlag, QueueReg, RingAddrs, VirtQueue}; use crate::virtio::{Result, error}; #[repr(C, align(16))] @@ -67,7 +67,7 @@ pub struct UsedElem { } #[derive(Debug)] -pub struct SplitQueue<'m> { +pub struct SplitQueue { size: u16, avail_hdr: *mut AvailHeader, avail_ring: *mut u16, @@ -76,10 +76,11 @@ pub struct SplitQueue<'m> { used_ring: *mut UsedElem, avail_event: Option<*mut AtomicU16>, desc: *mut Desc, - _phantom: PhantomData<&'m ()>, + /// The snapshot the pointers above point into. + ram: Arc, } -impl SplitQueue<'_> { +impl SplitQueue { pub fn avail_index(&self) -> u16 { unsafe { &*self.avail_hdr }.idx.load(Ordering::Acquire) } @@ -132,23 +133,29 @@ impl SplitQueue<'_> { } } -impl<'m> SplitQueue<'m> { - pub fn new(reg: &QueueReg, ram: &'m Ram, event_idx: bool) -> Result>> { - if !reg.enabled.load(Ordering::Acquire) { +impl SplitQueue { + /// Creates the queue from its registers, or returns `None` if the queue + /// is disabled. + pub fn new(reg: &QueueReg, ram: &Arc, event_idx: bool) -> Result> { + let Some(addrs) = RingAddrs::load(reg) else { return Ok(None); - } - let size = reg.size.load(Ordering::Acquire) as u64; - let used = reg.device.load(Ordering::Acquire); - let avail = reg.driver.load(Ordering::Acquire); - let desc = reg.desc.load(Ordering::Acquire); - // The registers may be cleared by a concurrent reset, which disables - // the queue first. - if !reg.enabled.load(Ordering::Acquire) { - return Ok(None); - } + }; + ::new(&addrs, ram, event_idx).map(Some) + } +} + +impl VirtQueue for SplitQueue { + type Index = u16; + + const INIT_INDEX: u16 = 0; + + fn new(addrs: &RingAddrs, ram: &Arc, event_idx: bool) -> Result { + let size = addrs.size as u64; if size == 0 || !size.is_power_of_two() { - return error::InvalidQueueSize { size: size as u16 }.fail(); + return error::InvalidQueueSize { size: addrs.size }.fail(); } + let used = addrs.device; + let avail = addrs.driver; let mut avail_event = None; let mut used_event = None; if event_idx { @@ -164,31 +171,29 @@ impl<'m> SplitQueue<'m> { let used_ring_gpa = used + size_of::() as u64; // The queue accesses entries up to the queue size through the ring // pointers, so the whole rings must be in guest memory. - Ok(Some(SplitQueue { - size: size as u16, + Ok(SplitQueue { + size: addrs.size, avail_hdr: ram.get_ptr(avail)?, avail_ring: ram.get_array_ptr(avail_ring_gpa, size)?, used_event, used_hdr, used_ring: ram.get_array_ptr(used_ring_gpa, size)?, avail_event, - desc: ram.get_array_ptr(desc, size)?, - _phantom: PhantomData, - })) + desc: ram.get_array_ptr(addrs.desc, size)?, + ram: ram.clone(), + }) } -} -impl<'m> VirtQueue<'m> for SplitQueue<'m> { - type Index = u16; - - const INIT_INDEX: u16 = 0; + fn ram(&self) -> &Arc { + &self.ram + } fn desc_avail(&self, index: u16) -> bool { let avail_index = self.avail_index(); index < avail_index || index - avail_index >= !(self.size - 1) } - fn get_avail(&self, index: Self::Index, ram: &'m Ram) -> Result>> { + fn get_avail<'m>(&self, index: Self::Index, ram: &'m Ram) -> Result>> { if !self.desc_avail(index) { return Ok(None); } @@ -243,6 +248,7 @@ impl<'m> VirtQueue<'m> for SplitQueue<'m> { delta: 1, readable, writable, + ram, })) } diff --git a/alioth/src/virtio/queue/split_test.rs b/alioth/src/virtio/queue/split_test.rs index ce181246..8c319f78 100644 --- a/alioth/src/virtio/queue/split_test.rs +++ b/alioth/src/virtio/queue/split_test.rs @@ -27,7 +27,7 @@ use crate::virtio::queue::split::{AvailHeader, Desc, DescFlag, SplitQueue, UsedE use crate::virtio::queue::tests::{GuestQueue, UsedDesc, VirtQueueGuest}; use crate::virtio::tests::{DATA_ADDR, MEM_SIZE, QUEUE_SIZE, fixture_queues, fixture_ram_bus}; -impl<'m> VirtQueueGuest<'m> for SplitQueue<'m> { +impl VirtQueueGuest for SplitQueue { fn add_desc( &mut self, index: u16, @@ -104,7 +104,7 @@ impl<'m> VirtQueueGuest<'m> for SplitQueue<'m> { fn disabled_queue() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.enabled.store(false, Ordering::Relaxed); let split_queue = SplitQueue::new(reg, &ram, false); @@ -115,7 +115,7 @@ fn disabled_queue() { fn invalid_queue_size() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.size.store(0, Ordering::Relaxed); @@ -147,7 +147,7 @@ enum Ring { fn ring_at_end_of_memory(#[case] ring: Ring, #[case] entries: u16, #[case] valid: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; // Place the ring so that only its first `entries` entries are in guest @@ -188,7 +188,7 @@ fn ring_at_end_of_memory(#[case] ring: Ring, #[case] entries: u16, #[case] valid fn enabled_queue() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; let q = SplitQueue::new(reg, &ram, false).unwrap().unwrap(); let mut guest_q = GuestQueue::new(SplitQueue::new(reg, &ram, false).unwrap().unwrap(), reg); @@ -243,7 +243,7 @@ fn enabled_queue() { fn event_idx_enabled() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; let q = SplitQueue::new(reg, &ram, true).unwrap().unwrap(); unsafe { &*q.used_event.unwrap() }.store(1, Ordering::Release); @@ -283,7 +283,7 @@ fn add_chain(q: &SplitQueue, descs: &[Desc]) { fn unexpected_indirect_desc(#[case] flag: DescFlag, #[case] table_flag: DescFlag) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let q = SplitQueue::new(&queues[0], &ram, false).unwrap().unwrap(); let table = [desc(DATA_ADDR, 1, table_flag, 0)]; @@ -319,7 +319,7 @@ fn linked_descs(n: u16) -> Vec { fn indirect_chain() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let q = SplitQueue::new(&queues[0], &ram, false).unwrap().unwrap(); ram.write(DATA_ADDR, b"abcdefg").unwrap(); @@ -360,7 +360,7 @@ fn indirect_chain() { fn chain_len(#[case] direct: u16, #[case] indirect: u16, #[case] valid: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let reg = &queues[0]; reg.size.store(4, Ordering::Relaxed); let q = SplitQueue::new(reg, &ram, false).unwrap().unwrap(); @@ -391,7 +391,7 @@ fn chain_len(#[case] direct: u16, #[case] indirect: u16, #[case] valid: bool) { fn chain_with_loop(#[case] indirect: bool) { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let q = SplitQueue::new(&queues[0], &ram, false).unwrap().unwrap(); // Descriptors 0 and 1 link to each other, so the chain never ends. @@ -419,7 +419,7 @@ fn chain_with_loop(#[case] indirect: bool) { fn desc_out_of_indirect_table() { let ram_bus = fixture_ram_bus(); let queues = fixture_queues(1); - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let q = SplitQueue::new(&queues[0], &ram, false).unwrap().unwrap(); // The indirect table has only descriptor 0, which links to descriptor 1. diff --git a/alioth/src/virtio/virtio_test.rs b/alioth/src/virtio/virtio_test.rs index fd2a34fa..f1d46c20 100644 --- a/alioth/src/virtio/virtio_test.rs +++ b/alioth/src/virtio/virtio_test.rs @@ -30,7 +30,7 @@ const QUEUE_START: u64 = 1 << 20; pub fn fixture_ram_bus() -> RamBus { let host_pages = ArcMemPages::from_anonymous(MEM_SIZE, None, None).unwrap(); let ram_bus = RamBus::new(); - ram_bus.add(0, host_pages).unwrap(); + ram_bus.update(|ram| ram.add(0, host_pages)).unwrap(); ram_bus } diff --git a/alioth/src/virtio/vu/backend.rs b/alioth/src/virtio/vu/backend.rs index 76cdaba2..b33175e6 100644 --- a/alioth/src/virtio/vu/backend.rs +++ b/alioth/src/virtio/vu/backend.rs @@ -67,8 +67,6 @@ pub enum Error { Convert { hva: u64 }, #[snafu(display("invalid message {req:?} with payload size {size}"))] InvalidMsg { req: VuFrontMsg, size: u32 }, - #[snafu(display("Cannot change memory layout at runtime"))] - ChangeMemoryLayout, #[snafu(display("Failed to send backend request channel to device"))] SendChannel, } @@ -194,6 +192,14 @@ impl VuBackend { Ok(()) } + /// Kicks a running worker, so that it moves to the latest memory layout + /// and drops the old one. + fn kick_worker(&self) { + if self.running { + self.dev.ctl.kick(); + } + } + fn convert_frontend_hva(&self, hva: u64) -> Result { for r in &self.init.regions { if hva >= r.hva && hva < r.hva + r.size { @@ -410,9 +416,6 @@ impl VuBackend { return error::MissingFd { req: msg.request }.fail(); }; let region = &single.region; - if self.running { - return error::ChangeMemoryLayout.fail(); - } log::debug!("{name}: add mem: {region:x?}, fd: {}", fd.as_raw_fd()); let user_mem = ArcMemPages::from_file( File::from(fd), @@ -420,20 +423,19 @@ impl VuBackend { region.size as usize, libc::PROT_READ | libc::PROT_WRITE, )?; - self.memory.add(region.gpa, user_mem)?; + self.memory.update(|ram| ram.add(region.gpa, user_mem))?; self.init.regions.push(single.region); + self.kick_worker(); } (VuFrontMsg::REM_MEM_REG, 40) => { let single: MemorySingleRegion = self.session.recv_payload()?; let region = &single.region; - if self.running { - return error::ChangeMemoryLayout.fail(); - } for (index, r) in self.init.regions.iter().enumerate() { if r.gpa == region.gpa && r.hva == region.hva && r.size == region.size { log::info!("{name}: remove mem: {r:x?}"); self.init.regions.remove(index); - let _ = self.memory.remove(region.gpa); + let _ = self.memory.update(|ram| ram.remove(region.gpa)); + self.kick_worker(); break; } } diff --git a/alioth/src/virtio/vu/backend_test.rs b/alioth/src/virtio/vu/backend_test.rs index 212c7165..7f774a4e 100644 --- a/alioth/src/virtio/vu/backend_test.rs +++ b/alioth/src/virtio/vu/backend_test.rs @@ -20,6 +20,7 @@ use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; use flume::{Receiver, Sender}; +use mio::{Events, Interest, Poll, Token}; use crate::mem; use crate::mem::emulated::{Action, Mmio}; @@ -28,7 +29,7 @@ use crate::sync::notifier::Notifier; use crate::virtio::dev::{VirtioDevice, WakeEvent, WorkerCtl, WorkerReport}; use crate::virtio::queue::QueueReg; use crate::virtio::vu::backend::{VuBackend, VuIrqSender}; -use crate::virtio::vu::bindings::VirtqState; +use crate::virtio::vu::bindings::{MemoryRegion, MemorySingleRegion, VirtqState}; use crate::virtio::vu::conn::VuSession; use crate::virtio::{DevStatus, DeviceId}; @@ -56,12 +57,25 @@ type TestWakeReceiver = Receiver>; /// /// The thread joins the worker after the frontend disconnects. fn spawn_backend(worker: W) -> (VuSession, JoinHandle<()>) +where + W: FnOnce(TestWakeReceiver, Sender) + Send + 'static, +{ + let notifier = Arc::new(Notifier::new().unwrap()); + spawn_backend_on(Arc::default(), notifier, worker) +} + +/// Like [`spawn_backend`], with the guest memory of the backend and the +/// notifier of the worker. +fn spawn_backend_on( + memory: Arc, + notifier: Arc, + worker: W, +) -> (VuSession, JoinHandle<()>) where W: FnOnce(TestWakeReceiver, Sender) + Send + 'static, { let (event_tx, event_rx) = flume::unbounded(); let (report_tx, report_rx) = flume::unbounded(); - let notifier = Arc::new(Notifier::new().unwrap()); let dev = VirtioDevice { name: "vu-backend".into(), id: DeviceId::NET, @@ -80,7 +94,7 @@ where let worker = thread::spawn(move || worker(event_rx, report_tx)); let (frontend_conn, backend_conn) = UnixStream::pair().unwrap(); - let mut backend = VuBackend::new(backend_conn, dev, Arc::new(RamBus::new())).unwrap(); + let mut backend = VuBackend::new(backend_conn, dev, memory).unwrap(); let backend_thread = thread::spawn(move || { backend.run().unwrap(); // Dropping the backend disconnects the worker. @@ -173,3 +187,58 @@ fn test_vu_backend_needs_reset_after_worker_failure() { drop(frontend); backend_thread.join().unwrap(); } + +#[test] +fn test_vu_backend_changes_memory_while_running() { + let memory = Arc::new(RamBus::new()); + let mut poll = Poll::new().unwrap(); + let mut notifier = Notifier::new().unwrap(); + let registry = poll.registry(); + registry + .register(&mut notifier, Token(0), Interest::READABLE) + .unwrap(); + // A fake worker that does nothing but complete resets. + let worker = |event_rx: TestWakeReceiver, report_tx: Sender| { + for event in event_rx.iter() { + match event { + WakeEvent::Start { .. } => {} + WakeEvent::Reset { seq } => { + report_tx.send(WorkerReport::ResetDone { seq }).unwrap() + } + event => panic!("unexpected event: {event:?}"), + } + } + }; + let (frontend, backend_thread) = spawn_backend_on(memory.clone(), Arc::new(notifier), worker); + start(&frontend); + + let gpa = 0x2_0000; + let file = tempfile::tempfile().unwrap(); + file.set_len(0x1000).unwrap(); + let region = MemorySingleRegion { + _padding: 0, + region: MemoryRegion { + gpa, + size: 0x1000, + hva: 0x5_0000, + mmap_offset: 0, + }, + }; + // The backend changes the layout while the worker runs, and kicks the + // worker after each change. + let mut events = Events::with_capacity(1); + frontend.add_mem_region(®ion, file.as_fd()).unwrap(); + assert!(memory.load().translate(gpa).is_ok()); + poll.poll(&mut events, Some(Duration::from_secs(5))) + .unwrap(); + assert!(!events.is_empty()); + + frontend.remove_mem_region(®ion).unwrap(); + assert!(memory.load().translate(gpa).is_err()); + poll.poll(&mut events, Some(Duration::from_secs(5))) + .unwrap(); + assert!(!events.is_empty()); + + drop(frontend); + backend_thread.join().unwrap(); +} diff --git a/alioth/src/virtio/vu/frontend.rs b/alioth/src/virtio/vu/frontend.rs index 67018fed..efc00340 100644 --- a/alioth/src/virtio/vu/frontend.rs +++ b/alioth/src/virtio/vu/frontend.rs @@ -15,7 +15,6 @@ use std::os::fd::{AsFd, AsRawFd, BorrowedFd, FromRawFd, OwnedFd}; use std::path::Path; use std::sync::Arc; -use std::sync::atomic::Ordering; use std::thread::JoinHandle; use mio::event::Event; @@ -296,13 +295,9 @@ impl Virtio for VuFrontend { } impl VirtioMio for VuFrontend { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { let name = &*self.name; @@ -322,7 +317,7 @@ impl VirtioMio for VuFrontend { log::trace!("{name}: queue-{index} is disabled"); continue; }; - let reg = queue.reg(); + let addrs = queue.addrs(); let _ = active_mio.irq_sender.queue_irqfd(index as _, |fd| { self.session.set_virtq_call(&(index as u64), fd)?; @@ -332,7 +327,7 @@ impl VirtioMio for VuFrontend { let virtq_num = VirtqState { index: index as _, - val: reg.size.load(Ordering::Acquire) as _, + val: addrs.size as _, }; self.session.set_virtq_num(&virtq_num)?; log::trace!("{name}: queue-{index}: set size: {}", virtq_num.val); @@ -344,13 +339,13 @@ impl VirtioMio for VuFrontend { self.session.set_virtq_base(&virtq_base)?; log::trace!("{name}: queue-{index}: set base: {}", virtq_base.val); - let mem = active_mio.mem; + let mem = queue.ram(); let virtq_addr = VirtqAddr { index: index as _, flags: 0, - desc_hva: mem.translate(reg.desc.load(Ordering::Acquire) as _)? as _, - used_hva: mem.translate(reg.device.load(Ordering::Acquire) as _)? as _, - avail_hva: mem.translate(reg.driver.load(Ordering::Acquire) as _)? as _, + desc_hva: mem.translate(addrs.desc)? as _, + used_hva: mem.translate(addrs.device)? as _, + avail_hva: mem.translate(addrs.driver)? as _, log_guest_addr: 0, }; self.session.set_virtq_addr(&virtq_addr)?; @@ -380,25 +375,17 @@ impl VirtioMio for VuFrontend { Ok(()) } - fn handle_event<'a, 'm, Q, S>( - &mut self, - _: &Event, - _: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_event<'a, Q, S>(&mut self, _: &Event, _: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { unreachable!() } - fn handle_queue<'m, Q, S>( - &mut self, - index: u16, - _: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn handle_queue(&mut self, index: u16, _: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { unreachable!( diff --git a/alioth/src/virtio/worker/io_uring.rs b/alioth/src/virtio/worker/io_uring.rs index 435081f1..32d043c4 100644 --- a/alioth/src/virtio/worker/io_uring.rs +++ b/alioth/src/virtio/worker/io_uring.rs @@ -22,7 +22,6 @@ use io_uring::cqueue::Entry as Cqe; use io_uring::squeue::Entry as Sqe; use io_uring::{SubmissionQueue, opcode, types}; -use crate::mem::mapped::Ram; use crate::sync::notifier::Notifier; use crate::virtio::dev::{ ActiveBackend, Backend, BackendEvent, Context, StartParam, Virtio, Worker, WorkerParam, @@ -37,13 +36,9 @@ pub enum BufferAction { } pub trait VirtioIoUring: Virtio { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - ring: &mut ActiveIoUring<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, ring: &mut ActiveIoUring<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender; fn handle_desc(&mut self, q_index: u16, chain: &mut DescChain) -> Result; @@ -111,16 +106,15 @@ where Ok(()) } - fn event_loop<'m, S, Q>( + fn event_loop( &mut self, - memory: &'m Ram, context: &mut Context, - queues: &mut [Option>], + queues: &mut [Option>], param: &StartParam, ) -> Result<()> where S: IrqSender, - Q: VirtQueue<'m>, + Q: VirtQueue, { let submit_counts = iter::repeat_n(0, queues.len()).collect(); // One entry for the worker notifier, and a reserve for each queue @@ -131,7 +125,6 @@ where shared_count: RING_SIZE - 1 - num_queues * QUEUE_RESERVE_SIZE, irq_sender: &*param.irq_sender, notifiers: param.notifiers.as_deref().unwrap_or(&[]), - mem: memory, queues, submit_counts, }; @@ -155,15 +148,14 @@ where } } -pub struct ActiveIoUring<'a, 'r, 'm, Q, S> +pub struct ActiveIoUring<'a, Q, S> where - Q: VirtQueue<'m>, + Q: VirtQueue, { ring: io_uring::IoUring, - pub queues: &'a mut [Option>], + pub queues: &'a mut [Option>], pub irq_sender: &'a S, pub notifiers: &'a [Notifier], - pub mem: &'m Ram, shared_count: u16, submit_counts: Box<[u16]>, } @@ -178,9 +170,9 @@ fn submit_queue_notifier(index: u16, notifier: &Notifier, sq: &mut SubmissionQue Ok(()) } -impl<'m, Q, S> ActiveIoUring<'_, '_, 'm, Q, S> +impl ActiveIoUring<'_, Q, S> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { fn run(&mut self, context: &mut Context) -> Result<()> @@ -207,8 +199,8 @@ where /// Requests like reading from a tap device may never complete by /// themselves, and closing the ring does not wait for the kernel to /// finish them. Without draining, the kernel could still write into - /// guest memory after the worker releases the memory layout, i.e. - /// after the memory may be unmapped and reused. + /// guest memory after the worker drops the chains and the snapshots they + /// hold, i.e. after the memory may be unmapped and reused. /// /// The results are discarded. The device is being reset or shut down, /// so the guest does not expect the chains back. @@ -216,8 +208,9 @@ where /// Some requests cannot be cancelled, e.g. block I/O on a hung NFS /// mount. Waiting is bounded by [`DRAIN_TIMEOUT`] so that a reset or a /// VM shutdown does not hang forever. If requests are still in flight - /// after that, or draining fails, guest memory is leaked instead: it - /// stays mapped even after it is removed from the guest. + /// after that, or draining fails, the chains and their snapshots are + /// leaked instead: the memory stays mapped even after it is removed from + /// the guest. fn drain(&mut self) -> Result<()> { let ret = self.try_drain(); match &ret { @@ -230,8 +223,8 @@ where } } if !matches!(ret, Ok(0)) { - for (_, pages) in self.mem.iter() { - std::mem::forget(pages.clone()); + for queue in self.queues.iter_mut().flatten() { + queue.leak_deferred(); } } self.submit_counts.fill(0); @@ -307,13 +300,18 @@ where } } -impl<'m, D, Q, S> ActiveBackend for ActiveIoUring<'_, '_, 'm, Q, S> +impl ActiveBackend for ActiveIoUring<'_, Q, S> where D: VirtioIoUring, - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { type Event = Cqe; + type Q = Q; + + fn queues(&mut self) -> &mut [Option>] { + self.queues + } fn handle_event(&mut self, dev: &mut D, event: &Self::Event) -> Result<()> { let token = event.user_data(); diff --git a/alioth/src/virtio/worker/io_uring_test.rs b/alioth/src/virtio/worker/io_uring_test.rs index 694626e5..9ec43bea 100644 --- a/alioth/src/virtio/worker/io_uring_test.rs +++ b/alioth/src/virtio/worker/io_uring_test.rs @@ -26,12 +26,13 @@ use io_uring::opcode; use io_uring::types::Fd; use crate::ffi; +use crate::mem::mapped::ArcMemPages; use crate::sync::notifier::Notifier; use crate::virtio::dev::entropy::{EntropyConfig, EntropyFeature}; -use crate::virtio::dev::{StartParam, Virtio, WakeEvent, WorkerParam}; +use crate::virtio::dev::{ActiveBackend, StartParam, Virtio, WakeEvent, WorkerParam}; use crate::virtio::queue::split::SplitQueue; use crate::virtio::queue::tests::GuestQueue; -use crate::virtio::queue::{DescChain, Queue, QueueReg, VirtQueue}; +use crate::virtio::queue::{self, DescChain, Queue, QueueReg, VirtQueue}; use crate::virtio::tests::{DATA_ADDR, FakeIrqSender, fixture_queues, fixture_ram_bus}; use crate::virtio::worker::io_uring::{ ActiveIoUring, BufferAction, IoUring, RING_SIZE, VirtioIoUring, @@ -80,13 +81,9 @@ impl Virtio for PipeReader { } impl VirtioIoUring for PipeReader { - fn activate<'m, Q, S>( - &mut self, - _feature: u128, - _ring: &mut ActiveIoUring<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, _feature: u128, _ring: &mut ActiveIoUring<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { Ok(()) @@ -154,7 +151,7 @@ fn io_uring_drain_test() { // The pipe is empty, so the read stays in flight. { - let ram = ram_bus.lock_layout(); + let ram = ram_bus.load(); let mut guest_q = GuestQueue::new( SplitQueue::new(®s[0], &ram, false).unwrap().unwrap(), ®s[0], @@ -191,8 +188,8 @@ fn io_uring_drain_test() { #[test] fn io_uring_drain_cancels_pending_reads() { - let ram_bus = fixture_ram_bus(); - let ram = ram_bus.lock_layout(); + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.load(); let regs = fixture_queues(1); let mut guest_q = GuestQueue::new( SplitQueue::new(®s[0], &ram, false).unwrap().unwrap(), @@ -213,14 +210,12 @@ fn io_uring_drain_cancels_pending_reads() { let (irq_tx, irq_rx) = flume::unbounded(); let irq_sender = FakeIrqSender { q_tx: irq_tx }; - let split_q = SplitQueue::new(®s[0], &ram, false).unwrap().unwrap(); - let mut queues = [Some(Queue::new(split_q, ®s[0], &ram))]; + let mut queues = [Queue::::new(®s[0], ram_bus.clone(), false).unwrap()]; let mut active = ActiveIoUring { ring: io_uring::IoUring::new(RING_SIZE as u32).unwrap(), queues: &mut queues, irq_sender: &irq_sender, notifiers: &[], - mem: &ram, shared_count: RING_SIZE - 1, submit_counts: Box::new([0]), }; @@ -245,3 +240,67 @@ fn io_uring_drain_cancels_pending_reads() { assert!(irq_rx.try_recv().is_err()); drop(active); } + +#[test] +fn io_uring_request_in_flight_across_layout_change() { + const EXTRA_GPA: u64 = 4 << 20; + + let ram_bus = Arc::new(fixture_ram_bus()); + let regs = fixture_queues(1); + let mut guest_q = GuestQueue::new( + SplitQueue::new(®s[0], &ram_bus.load(), false) + .unwrap() + .unwrap(), + ®s[0], + ); + let pages = ArcMemPages::from_anonymous(1 << 20, None, None).unwrap(); + ram_bus.update(|ram| ram.add(EXTRA_GPA, pages)).unwrap(); + let ram = ram_bus.load(); + let id = guest_q.add_desc(&[], &[(EXTRA_GPA, 4 << 10)]); + + let (reader, mut writer) = pipe(); + let (submitted_tx, submitted_rx) = flume::unbounded(); + let mut dev = PipeReader { + reader, + config: Arc::new(EntropyConfig), + submitted: submitted_tx, + completed: Arc::default(), + }; + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = FakeIrqSender { q_tx: irq_tx }; + let mut queues = [Queue::::new(®s[0], ram_bus.clone(), false).unwrap()]; + let mut active = ActiveIoUring { + ring: io_uring::IoUring::new(RING_SIZE as u32).unwrap(), + queues: &mut queues, + irq_sender: &irq_sender, + notifiers: &[], + shared_count: RING_SIZE - 1, + submit_counts: Box::new([0]), + }; + active.submit_buffers(&mut dev, 0).unwrap(); + active.ring.submit().unwrap(); + assert_eq!(submitted_rx.try_recv(), Ok(id)); + + // The memory of the buffer is unplugged while the read is in flight, + // and the queue moves to the new layout. The chain keeps the old one. + let old_ram = Arc::downgrade(&ram); + drop(ram); + ram_bus.update(|ram| ram.remove(EXTRA_GPA)).unwrap(); + let queues = ActiveBackend::::queues(&mut active); + assert!(queue::sync_queues(queues).unwrap()); + let queue = active.queues[0].as_ref().unwrap(); + assert_eq!(queue.ram().generation(), ram_bus.generation()); + assert!(old_ram.upgrade().is_some()); + + // The kernel still writes into the buffer, and completing the chain + // releases the old layout. + let s = b"written after unplug"; + writer.write_all(s).unwrap(); + active.ring.submit_and_wait(1).unwrap(); + let cqe = active.ring.completion().next().unwrap(); + active.handle_event(&mut dev, &cqe).unwrap(); + assert_eq!(irq_rx.try_recv(), Ok(0)); + let used = guest_q.get_used().unwrap(); + assert_eq!((used.id, used.len), (id, s.len() as u32)); + assert!(old_ram.upgrade().is_none()); +} diff --git a/alioth/src/virtio/worker/mio.rs b/alioth/src/virtio/worker/mio.rs index b2a57abf..e95baa60 100644 --- a/alioth/src/virtio/worker/mio.rs +++ b/alioth/src/virtio/worker/mio.rs @@ -23,7 +23,6 @@ use mio::unix::SourceFd; use mio::{Events, Interest, Poll, Registry, Token}; use snafu::ResultExt; -use crate::mem::mapped::Ram; use crate::sync::notifier::Notifier; use crate::virtio::dev::{ ActiveBackend, Backend, BackendEvent, Context, StartParam, Virtio, Worker, WorkerParam, @@ -33,31 +32,27 @@ use crate::virtio::queue::{Queue, VirtQueue}; use crate::virtio::{IrqSender, Result, error}; pub trait VirtioMio: Virtio { - fn activate<'m, Q, S>( - &mut self, - feature: u128, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, - ) -> Result<()> + fn activate(&mut self, feature: u128, active_mio: &mut ActiveMio<'_, Q, S>) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender; - fn handle_queue<'m, Q, S>( + fn handle_queue( &mut self, index: u16, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender; - fn handle_event<'m, Q, S>( + fn handle_event( &mut self, event: &Event, - active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>, + active_mio: &mut ActiveMio<'_, Q, S>, ) -> Result<()> where - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender; fn reset(&mut self, registry: &Registry); @@ -106,16 +101,15 @@ where Ok(()) } - fn event_loop<'m, S, Q>( + fn event_loop( &mut self, - memory: &'m Ram, context: &mut Context, - queues: &mut [Option>], + queues: &mut [Option>], param: &StartParam, ) -> Result<()> where S: IrqSender, - Q: VirtQueue<'m>, + Q: VirtQueue, { let mut events = Events::with_capacity(128); let mut active_mio = ActiveMio { @@ -123,7 +117,6 @@ where irq_sender: &*param.irq_sender, notifiers: param.notifiers.as_deref().unwrap_or(&[]), poll: &mut self.poll, - mem: memory, }; context.dev.activate(param.feature, &mut active_mio)?; // Only KVM offers ioeventfds, see [`crate::hv::NotifierRegistry`]. @@ -168,24 +161,28 @@ where } } -pub struct ActiveMio<'a, 'r, 'm, Q, S> +pub struct ActiveMio<'a, Q, S> where - Q: VirtQueue<'m>, + Q: VirtQueue, { - pub queues: &'a mut [Option>], + pub queues: &'a mut [Option>], pub irq_sender: &'a S, pub notifiers: &'a [Notifier], pub poll: &'a mut Poll, - pub mem: &'m Ram, } -impl<'m, D, Q, S> ActiveBackend for ActiveMio<'_, '_, 'm, Q, S> +impl ActiveBackend for ActiveMio<'_, Q, S> where D: VirtioMio, - Q: VirtQueue<'m>, + Q: VirtQueue, S: IrqSender, { type Event = Event; + type Q = Q; + + fn queues(&mut self) -> &mut [Option>] { + self.queues + } fn handle_event(&mut self, dev: &mut D, event: &Self::Event) -> Result<()> { let token = event.token().0 as u64; diff --git a/alioth/src/vm/vm.rs b/alioth/src/vm/vm.rs index a393ac88..0663fb57 100644 --- a/alioth/src/vm/vm.rs +++ b/alioth/src/vm/vm.rs @@ -316,6 +316,8 @@ where self.ctx.board.memory.ram_bus(), self.ctx.board.spec.coco.is_some(), )?; + let kick = virtio_dev.mem_update_kick(); + self.ctx.board.memory.register_update_callback(kick)?; let msi_sender = self.ctx.board.vm.create_msi_sender( #[cfg(target_arch = "aarch64")] u32::from(bdf.0),