alioth/src/virtio/dev/dev.rs - f5bf11a

dev.rs48.37%

// Copyright 2024 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
pub mod balloon;
pub mod blk;
pub mod entropy;
#[path = "fs/fs.rs"]
pub mod fs;
#[path = "net/net.rs"]
pub mod net;
#[path = "vsock/vsock.rs"]
pub mod vsock;
 
use std::fmt::Debug;
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU16, AtomicU32};
use std::sync::mpsc::{self, Receiver, Sender};
use std::thread::JoinHandle;
 
use bitflags::Flags;
use snafu::ResultExt;
 
use crate::hv::IoeventFd;
use crate::mem::emulated::Mmio;
use crate::mem::mapped::{Ram, RamBus};
use crate::mem::{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};
#[cfg(target_os = "linux")]
use crate::virtio::vu::conn::VuChannel;
use crate::virtio::{DeviceId, IrqSender, Result, VirtioFeature, error};
 
pub trait Virtio: Debug + Send + Sync + 'static {
    type Config: Mmio;
    type Feature: Flags<Bits = u128> + Debug;
 
    fn name(&self) -> &str;
    fn id(&self) -> DeviceId;
    fn num_queues(&self) -> u16;
    fn config(&self) -> Arc<Self::Config>;
    fn feature(&self) -> u128;
    fn spawn_worker<S: IrqSender, E: IoeventFd>(
        self,
        event_rx: Receiver<WakeEvent<S, E>>,
        memory: Arc<RamBus>,
        queue_regs: Arc<[QueueReg]>,
    ) -> Result<(JoinHandle<()>, Arc<Notifier>)>;
    fn shared_mem_regions(&self) -> Option<Arc<MemRegion>> {
        None
    }
    fn ioeventfd_offloaded(&self, _q_index: u16) -> Result<bool> {
        Ok(false)
    }
    fn mem_update_callback(&self) -> Option<Box<dyn LayoutUpdated>> {
        None
    }
    fn mem_change_callback(&self) -> Option<Box<dyn LayoutChanged>> {
        None
    }
    #[cfg(target_os = "linux")]
    fn set_vu_channel(&mut self, _channel: Arc<VuChannel>) {}
}
 
#[derive(Debug, Default)]
pub struct Register {
    pub device_feature: [u32; 4],
    pub driver_feature: [AtomicU32; 4],
    pub device_feature_sel: AtomicU8,
    pub driver_feature_sel: AtomicU8,
    pub queue_sel: AtomicU16,
    pub status: AtomicU8,
}
 
const TOKEN_WARKER: u64 = 1 << 63;
 
#[derive(Debug, Clone)]
pub struct StartParam<S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    pub(crate) feature: u128,
    pub(crate) irq_sender: Arc<S>,
    pub(crate) ioeventfds: Option<Arc<[E]>>,
}
 
#[derive(Debug, Clone)]
pub enum WakeEvent<S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    Notify {
        q_index: u16,
    },
    Shutdown,
    #[cfg(target_os = "linux")]
    VuChannel {
        channel: Arc<VuChannel>,
    },
    Start {
        param: StartParam<S, E>,
    },
    Reset,
}
 
#[derive(Debug, PartialEq, Eq)]
pub enum WorkerState {
    Pending,
    Running,
    Shutdown,
}
 
#[derive(Debug)]
pub struct Worker<D, S, E, B>
where
    S: IrqSender,
    E: IoeventFd,
{
    context: Context<D, S, E>,
    backend: B,
}
 
#[derive(Debug)]
pub struct VirtioDevice<S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    pub name: Arc<str>,
    pub id: DeviceId,
    pub device_config: Arc<dyn Mmio>,
    pub device_feature: u128,
    pub queue_regs: Arc<[QueueReg]>,
    pub shared_mem_regions: Option<Arc<MemRegion>>,
    pub notifier: Arc<Notifier>,
    pub event_tx: Sender<WakeEvent<S, E>>,
    worker_handle: Option<JoinHandle<()>>,
}
 
impl<S, E> VirtioDevice<S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    fn shutdown(&mut self) -> Result<(), Box<dyn std::error::Error>> {
        let Some(handle) = self.worker_handle.take() else {
            return Ok(());
        };
        self.event_tx.send(WakeEvent::Shutdown)?;
        self.notifier.notify()?;
        if let Err(e) = handle.join() {
            log::error!("{}: failed to join worker thread: {e:?}", self.name)
        }
        Ok(())
    }
 
    pub fn new<D>(
        name: impl Into<Arc<str>>,
        dev: D,
        memory: Arc<RamBus>,
        restricted_memory: bool,
    ) -> Result<Self>
    where
        D: Virtio,
    {
        let name = name.into();
        let id = dev.id();
        let device_config = dev.config();
        let mut device_feature = dev.feature();
        if restricted_memory {
            device_feature |= VirtioFeature::ACCESS_PLATFORM.bits()
        } else {
            device_feature &= !VirtioFeature::ACCESS_PLATFORM.bits()
        }
        let num_queues = dev.num_queues();
        let queue_regs = (0..num_queues).map(|_| QueueReg {
            size: AtomicU16::new(QUEUE_SIZE_MAX),
            ..Default::default()
        });
        let queue_regs = queue_regs.collect::<Arc<_>>();
 
        let shared_mem_regions = dev.shared_mem_regions();
        let (event_tx, event_rx) = mpsc::channel();
        let (handle, notifier) = dev.spawn_worker(event_rx, memory, queue_regs.clone())?;
        log::debug!(
            "{name}: created with {:x?}, {:x?}",
            VirtioFeature::from_bits_retain(device_feature & !D::Feature::all().bits()),
            D::Feature::from_bits_truncate(device_feature)
        );
        let virtio_dev = VirtioDevice {
            name,
            id,
            device_feature,
            queue_regs,
            worker_handle: Some(handle),
            event_tx,
            notifier,
            device_config,
            shared_mem_regions,
        };
        Ok(virtio_dev)
    }
}
 
impl<S, E> Drop for VirtioDevice<S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    fn drop(&mut self) {
        if let Err(e) = self.shutdown() {
            log::error!("{}: failed to shutdown: {e}", self.name);
        }
    }
}
 
pub trait Backend<D: Virtio>: Send + 'static {
    fn register_notifier(&mut self, token: u64) -> Result<Arc<Notifier>>;
    fn reset(&self, dev: &mut D) -> Result<()>;
    fn event_loop<'m, S, Q, E>(
        &mut self,
        memory: &'m Ram,
        context: &mut Context<D, S, E>,
        queues: &mut [Option<Queue<'_, 'm, Q>>],
        param: &StartParam<S, E>,
    ) -> Result<()>
    where
        S: IrqSender,
        Q: VirtQueue<'m>,
        E: IoeventFd;
}
 
pub trait BackendEvent {
    fn token(&self) -> u64;
}
 
pub trait ActiveBackend<D: Virtio> {
    type Event: BackendEvent;
    fn handle_event(&mut self, dev: &mut D, event: &Self::Event) -> Result<()>;
    fn handle_queue(&mut self, dev: &mut D, index: u16) -> Result<()>;
}
 
#[derive(Debug)]
pub struct Context<D, S, E>
where
    S: IrqSender,
    E: IoeventFd,
{
    pub dev: D,
    memory: Arc<RamBus>,
    event_rx: Receiver<WakeEvent<S, E>>,
    queue_regs: Arc<[QueueReg]>,
    pub state: WorkerState,
}
 
impl<D, S, E> Context<D, S, E>
where
    D: Virtio,
    S: IrqSender,
    E: IoeventFd,
{
    fn handle_wake_events<B>(&mut self, backend: &mut B) -> Result<()>29x
    where29x
        B: ActiveBackend<D>,29x
    {
        while let Ok(event) = self.event_rx.try_recv() {59x
            match event {36x
                WakeEvent::Notify { q_index } => backend.handle_queue(&mut self.dev, q_index)?,30x
                WakeEvent::Shutdown => {
                    self.state = WorkerState::Shutdown;6x
                    break;6x
                }
                WakeEvent::Start { .. } => {
                    log::error!("{}: device has already started", self.dev.name())
                }
                #[cfg(target_os = "linux")]
                WakeEvent::VuChannel { channel } => self.dev.set_vu_channel(channel),
                WakeEvent::Reset => {
                    log::info!("{}: guest requested reset", self.dev.name());
                    self.state = WorkerState::Pending;
                    break;
                }
            }
        }
        Ok(())29x
    }29x
 
    fn wait_start(&mut self) -> Option<StartParam<S, E>> {6x
        for wake_event in self.event_rx.iter() {6x
            match wake_event {6x
                WakeEvent::Reset => {}
                WakeEvent::Start { param } => {6x
                    self.state = WorkerState::Running;6x
                    return Some(param);6x
                }
                #[cfg(target_os = "linux")]
                WakeEvent::VuChannel { channel } => self.dev.set_vu_channel(channel),
                WakeEvent::Shutdown => break,
                WakeEvent::Notify { q_index } => {
                    log::error!(
                        "{}: driver notified queue {q_index} before device is ready",
                        self.dev.name()
                    )
                }
            }
        }
        self.state = WorkerState::Shutdown;
        None
    }6x
 
    pub fn handle_event<B>(&mut self, event: &B::Event, backend: &mut B) -> Result<()>38x
    where38x
        B: ActiveBackend<D>,38x
    {
        if event.token() == TOKEN_WARKER {38x
            self.handle_wake_events(backend)29x
        } else {
            backend.handle_event(&mut self.dev, event)9x
        }
    }38x
}
 
impl<D, S, E, B> Worker<D, S, E, B>
where
    D: Virtio,
    S: IrqSender,
    B: Backend<D>,
    E: IoeventFd,
{
    pub fn spawn(6x
        dev: D,6x
        mut backend: B,6x
        event_rx: Receiver<WakeEvent<S, E>>,6x
        memory: Arc<RamBus>,6x
        queue_regs: Arc<[QueueReg]>,6x
    ) -> Result<(JoinHandle<()>, Arc<Notifier>)> {6x
        let notifier = backend.register_notifier(TOKEN_WARKER)?;6x
        let worker = Worker {6x
            context: Context {6x
                dev,6x
                event_rx,6x
                memory,6x
                queue_regs,6x
                state: WorkerState::Pending,6x
            },6x
            backend,6x
        };6x
        let name = worker.context.dev.name();6x
        let handle = std::thread::Builder::new()6x
            .name(name.to_owned())6x
            .spawn(move || worker.do_work())6x
            .context(error::WorkerThread)?;6x
        Ok((handle, notifier))6x
    }6x
 
    fn event_loop<'m, Q>(6x
        &mut self,6x
        queues: &mut [Option<Queue<'_, 'm, Q>>],6x
        ram: &'m Ram,6x
        param: &StartParam<S, E>,6x
    ) -> Result<()>6x
    where6x
        Q: VirtQueue<'m>,6x
        E: IoeventFd,6x
    {
        log::debug!(6x
            "{}: activated with {:x?}, {:x?}",
            self.context.dev.name(),6x
            VirtioFeature::from_bits_retain(param.feature & !D::Feature::all().bits()),6x
            D::Feature::from_bits_truncate(param.feature)6x
        );
        self.backend6x
            .event_loop(ram, &mut self.context, queues, param)6x
    }6x
 
    fn loop_until_reset(&mut self) -> Result<()> {6x
        let Some(param) = self.context.wait_start() else {6x
            return Ok(());
        };
        let memory = self.context.memory.clone();6x
        let ram = memory.lock_layout();6x
        let feature = param.feature & !VirtioFeature::ACCESS_PLATFORM.bits();6x
        let queue_regs = self.context.queue_regs.clone();6x
        let feature = VirtioFeature::from_bits_retain(feature);6x
        let event_idx = feature.contains(VirtioFeature::EVENT_IDX);6x
        if feature.contains(VirtioFeature::RING_PACKED) {6x
            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<Box<_>> = queue_regs.iter().map(new_queue).collect();
            self.event_loop(&mut (queues?), &ram, &param)?;
        } else {
            let new_queue = |reg| {12x
                let Some(split_queue) = SplitQueue::new(reg, &ram, event_idx)? else {12x
                    return Ok(None);
                };
                Ok(Some(Queue::new(split_queue, reg, &ram)))12x
            };12x
            let queues: Result<Box<_>> = queue_regs.iter().map(new_queue).collect();6x
            self.event_loop(&mut (queues?), &ram, &param)?;6x
        };
        self.backend.reset(&mut self.context.dev)?;6x
        Ok(())6x
    }6x
 
    fn do_work(mut self) {6x
        while self.context.state != WorkerState::Shutdown {12x
            if let Err(e) = self.loop_until_reset() {6x
                log::error!("worker {}: {e:?}", self.context.dev.name(),);
                return;
            }6x
        }
        log::debug!("worker {}: done", self.context.dev.name())6x
    }6x
}
 
pub trait DevParam {
    type Device;
    fn build(self, name: impl Into<Arc<str>>) -> Result<Self::Device>;
    fn needs_mem_shared_fd(&self) -> bool {
        false
    }
}
 