18602811487

Committed 17 Oct 2025 06:27PM UTC coverage: 17.44% (-0.4%) from 17.856%

Build # 18602811487

Build Type

push

github

Committed by

Lencerf

Commit Message

test(pci): add tests for HostBridge

Signed-off-by: Changyuan Lyu <changyuanl@google.com>

Run Details

1394 of 7993 relevant lines covered (17.44%)

17.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

40.65

/alioth/src/virtio/dev/dev.rs

// 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<()>
    where
        B: ActiveBackend<D>,
    {
        while let Ok(event) = self.event_rx.try_recv() {
            match event {
                WakeEvent::Notify { q_index } => backend.handle_queue(&mut self.dev, q_index)?,
                WakeEvent::Shutdown => {
                    self.state = WorkerState::Shutdown;
                    break;
                }
                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(())
    }

    fn wait_start(&mut self) -> Option<StartParam<S, E>> {
        for wake_event in self.event_rx.iter() {
            match wake_event {
                WakeEvent::Reset => {}
                WakeEvent::Start { param } => {
                    self.state = WorkerState::Running;
                    return Some(param);
                }
                #[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
    }

    pub fn handle_event<B>(&mut self, event: &B::Event, backend: &mut B) -> Result<()>
    where
        B: ActiveBackend<D>,
    {
        if event.token() == TOKEN_WARKER {
            self.handle_wake_events(backend)
        } else {
            backend.handle_event(&mut self.dev, event)
        }
    }
}

impl<D, S, E, B> Worker<D, S, E, B>
where
    D: Virtio,
    S: IrqSender,
    B: Backend<D>,
    E: IoeventFd,
{
    pub fn spawn(
        dev: D,
        mut backend: B,
        event_rx: Receiver<WakeEvent<S, E>>,
        memory: Arc<RamBus>,
        queue_regs: Arc<[QueueReg]>,
    ) -> Result<(JoinHandle<()>, Arc<Notifier>)> {
        let notifier = backend.register_notifier(TOKEN_WARKER)?;
        let worker = Worker {
            context: Context {
                dev,
                event_rx,
                memory,
                queue_regs,
                state: WorkerState::Pending,
            },
            backend,
        };
        let name = worker.context.dev.name();
        let handle = std::thread::Builder::new()
            .name(name.to_owned())
            .spawn(move || worker.do_work())
            .context(error::WorkerThread)?;
        Ok((handle, notifier))
    }

    fn event_loop<'m, Q>(
        &mut self,
        queues: &mut [Option<Queue<'_, 'm, Q>>],
        ram: &'m Ram,
        param: &StartParam<S, E>,
    ) -> Result<()>
    where
        Q: VirtQueue<'m>,
        E: IoeventFd,
    {
        log::debug!(
            "{}: activated with {:x?} {:x?}",
            self.context.dev.name(),
            VirtioFeature::from_bits_retain(param.feature & !D::Feature::all().bits()),
            D::Feature::from_bits_truncate(param.feature)
        );
        self.backend
            .event_loop(ram, &mut self.context, queues, param)
    }

    fn loop_until_reset(&mut self) -> Result<()> {
        let Some(param) = self.context.wait_start() else {
            return Ok(());
        };
        let memory = self.context.memory.clone();
        let ram = memory.lock_layout();
        let feature = param.feature & !VirtioFeature::ACCESS_PLATFORM.bits();
        let queue_regs = self.context.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<Box<_>> = queue_regs.iter().map(new_queue).collect();
            self.event_loop(&mut (queues?), &ram, &param)?;
        } 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<Box<_>> = queue_regs.iter().map(new_queue).collect();
            self.event_loop(&mut (queues?), &ram, &param)?;
        };
        self.backend.reset(&mut self.context.dev)?;
        Ok(())
    }

    fn do_work(mut self) {
        while self.context.state != WorkerState::Shutdown {
            if let Err(e) = self.loop_until_reset() {
                log::error!("worker {}: {e:?}", self.context.dev.name(),);
                return;
            }
        }
        log::debug!("worker {}: done", self.context.dev.name())
    }
}

pub trait DevParam {
    type Device;
    fn build(self, name: impl Into<Arc<str>>) -> Result<Self::Device>;
    fn needs_mem_shared_fd(&self) -> bool {
        false
    }
}

1	// Copyright 2024 Google LLC
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// https://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	pub mod balloon;
16	pub mod blk;
17	pub mod entropy;
18	#[path = "fs/fs.rs"]
19	pub mod fs;
20	#[path = "net/net.rs"]
21	pub mod net;
22	#[path = "vsock/vsock.rs"]
23	pub mod vsock;
24
25	use std::fmt::Debug;
26	use std::sync::Arc;
27	use std::sync::atomic::{AtomicU8, AtomicU16, AtomicU32};
28	use std::sync::mpsc::{self, Receiver, Sender};
29	use std::thread::JoinHandle;
30
31	use bitflags::Flags;
32	use snafu::ResultExt;
33
34	use crate::hv::IoeventFd;
35	use crate::mem::emulated::Mmio;
36	use crate::mem::mapped::{Ram, RamBus};
37	use crate::mem::{LayoutChanged, LayoutUpdated, MemRegion};
38	use crate::sync::notifier::Notifier;
39	use crate::virtio::queue::packed::PackedQueue;
40	use crate::virtio::queue::split::SplitQueue;
41	use crate::virtio::queue::{QUEUE_SIZE_MAX, Queue, QueueReg, VirtQueue};
42	#[cfg(target_os = "linux")]
43	use crate::virtio::vu::conn::VuChannel;
44	use crate::virtio::{DeviceId, IrqSender, Result, VirtioFeature, error};
45
46	pub trait Virtio: Debug + Send + Sync + 'static {
47	type Config: Mmio;
48	type Feature: Flags<Bits = u128> + Debug;
49
50	fn name(&self) -> &str;
51	fn id(&self) -> DeviceId;
52	fn num_queues(&self) -> u16;
53	fn config(&self) -> Arc<Self::Config>;
54	fn feature(&self) -> u128;
55	fn spawn_worker<S: IrqSender, E: IoeventFd>(
56	self,
57	event_rx: Receiver<WakeEvent<S, E>>,
58	memory: Arc<RamBus>,
59	queue_regs: Arc<[QueueReg]>,
60	) -> Result<(JoinHandle<()>, Arc<Notifier>)>;
61	fn shared_mem_regions(&self) -> Option<Arc<MemRegion>> {	×
62	None	×
63	}
64	fn ioeventfd_offloaded(&self, _q_index: u16) -> Result<bool> {	×
65	Ok(false)	×
66	}
67	fn mem_update_callback(&self) -> Option<Box<dyn LayoutUpdated>> {	×
68	None	×
69	}
70	fn mem_change_callback(&self) -> Option<Box<dyn LayoutChanged>> {	×
71	None	×
72	}
73	#[cfg(target_os = "linux")]
74	fn set_vu_channel(&mut self, _channel: Arc<VuChannel>) {}
75	}
76
77	#[derive(Debug, Default)]
78	pub struct Register {
79	pub device_feature: [u32; 4],
80	pub driver_feature: [AtomicU32; 4],
81	pub device_feature_sel: AtomicU8,
82	pub driver_feature_sel: AtomicU8,
83	pub queue_sel: AtomicU16,
84	pub status: AtomicU8,
85	}
86
87	const TOKEN_WARKER: u64 = 1 << 63;
88
89	#[derive(Debug, Clone)]
90	pub struct StartParam<S, E>
91	where
92	S: IrqSender,
93	E: IoeventFd,
94	{
95	pub(crate) feature: u128,
96	pub(crate) irq_sender: Arc<S>,
97	pub(crate) ioeventfds: Option<Arc<[E]>>,
98	}
99
100	#[derive(Debug, Clone)]
101	pub enum WakeEvent<S, E>
102	where
103	S: IrqSender,
104	E: IoeventFd,
105	{
106	Notify {
107	q_index: u16,
108	},
109	Shutdown,
110	#[cfg(target_os = "linux")]
111	VuChannel {
112	channel: Arc<VuChannel>,
113	},
114	Start {
115	param: StartParam<S, E>,
116	},
117	Reset,
118	}
119
120	#[derive(Debug, PartialEq, Eq)]
121	pub enum WorkerState {
122	Pending,
123	Running,
124	Shutdown,
125	}
126
127	#[derive(Debug)]
128	pub struct Worker<D, S, E, B>
129	where
130	S: IrqSender,
131	E: IoeventFd,
132	{
133	context: Context<D, S, E>,
134	backend: B,
135	}
136
137	#[derive(Debug)]
138	pub struct VirtioDevice<S, E>
139	where
140	S: IrqSender,
141	E: IoeventFd,
142	{
143	pub name: Arc<str>,
144	pub id: DeviceId,
145	pub device_config: Arc<dyn Mmio>,
146	pub device_feature: u128,
147	pub queue_regs: Arc<[QueueReg]>,
148	pub shared_mem_regions: Option<Arc<MemRegion>>,
149	pub notifier: Arc<Notifier>,
150	pub event_tx: Sender<WakeEvent<S, E>>,
151	worker_handle: Option<JoinHandle<()>>,
152	}
153
154	impl<S, E> VirtioDevice<S, E>
155	where
156	S: IrqSender,
157	E: IoeventFd,
158	{
159	fn shutdown(&mut self) -> Result<(), Box<dyn std::error::Error>> {	×
160	let Some(handle) = self.worker_handle.take() else {	×
161	return Ok(());	×
162	};
163	self.event_tx.send(WakeEvent::Shutdown)?;	×
164	self.notifier.notify()?;	×
165	if let Err(e) = handle.join() {	×
166	log::error!("{}: failed to join worker thread: {e:?}", self.name)	×
167	}
168	Ok(())	×
169	}
170
171	pub fn new<D>(	×
172	name: impl Into<Arc<str>>,
173	dev: D,
174	memory: Arc<RamBus>,
175	restricted_memory: bool,
176	) -> Result<Self>
177	where
178	D: Virtio,
179	{
180	let name = name.into();	×
181	let id = dev.id();	×
182	let device_config = dev.config();	×
183	let mut device_feature = dev.feature();	×
184	if restricted_memory {	×
185	device_feature \|= VirtioFeature::ACCESS_PLATFORM.bits()	×
186	} else {
187	device_feature &= !VirtioFeature::ACCESS_PLATFORM.bits()	×
188	}
189	let num_queues = dev.num_queues();	×
190	let queue_regs = (0..num_queues).map(\|_\| QueueReg {	×
191	size: AtomicU16::new(QUEUE_SIZE_MAX),	×
192	..Default::default()	×
193	});
194	let queue_regs = queue_regs.collect::<Arc<_>>();	×
195
196	let shared_mem_regions = dev.shared_mem_regions();	×
197	let (event_tx, event_rx) = mpsc::channel();	×
198	let (handle, notifier) = dev.spawn_worker(event_rx, memory, queue_regs.clone())?;	×
199	log::debug!(	×
200	"{name}: created with {:x?} {:x?}",	×
201	VirtioFeature::from_bits_retain(device_feature & !D::Feature::all().bits()),	×
202	D::Feature::from_bits_truncate(device_feature)	×
203	);
204	let virtio_dev = VirtioDevice {
205	name,
206	id,
207	device_feature,
208	queue_regs,
209	worker_handle: Some(handle),	×
210	event_tx,
211	notifier,
212	device_config,
213	shared_mem_regions,
214	};
215	Ok(virtio_dev)	×
216	}
217	}
218
219	impl<S, E> Drop for VirtioDevice<S, E>
220	where
221	S: IrqSender,
222	E: IoeventFd,
223	{
224	fn drop(&mut self) {	×
225	if let Err(e) = self.shutdown() {	×
226	log::error!("{}: failed to shutdown: {e}", self.name);	×
227	}
228	}
229	}
230
231	pub trait Backend<D: Virtio>: Send + 'static {
232	fn register_notifier(&mut self, token: u64) -> Result<Arc<Notifier>>;
233	fn reset(&self, dev: &mut D) -> Result<()>;
234	fn event_loop<'m, S, Q, E>(
235	&mut self,
236	memory: &'m Ram,
237	context: &mut Context<D, S, E>,
238	queues: &mut [Option<Queue<'_, 'm, Q>>],
239	param: &StartParam<S, E>,
240	) -> Result<()>
241	where
242	S: IrqSender,
243	Q: VirtQueue<'m>,
244	E: IoeventFd;
245	}
246
247	pub trait BackendEvent {
248	fn token(&self) -> u64;
249	}
250
251	pub trait ActiveBackend<D: Virtio> {
252	type Event: BackendEvent;
253	fn handle_event(&mut self, dev: &mut D, event: &Self::Event) -> Result<()>;
254	fn handle_queue(&mut self, dev: &mut D, index: u16) -> Result<()>;
255	}
256
257	#[derive(Debug)]
258	pub struct Context<D, S, E>
259	where
260	S: IrqSender,
261	E: IoeventFd,
262	{
263	pub dev: D,
264	memory: Arc<RamBus>,
265	event_rx: Receiver<WakeEvent<S, E>>,
266	queue_regs: Arc<[QueueReg]>,
267	pub state: WorkerState,
268	}
269
270	impl<D, S, E> Context<D, S, E>
271	where
272	D: Virtio,
273	S: IrqSender,
274	E: IoeventFd,
275	{
276	fn handle_wake_events<B>(&mut self, backend: &mut B) -> Result<()>	26✔
277	where
278	B: ActiveBackend<D>,
279	{
280	while let Ok(event) = self.event_rx.try_recv() {	72✔
281	match event {	2✔
282	WakeEvent::Notify { q_index } => backend.handle_queue(&mut self.dev, q_index)?,	24✔
283	WakeEvent::Shutdown => {	×
284	self.state = WorkerState::Shutdown;	6✔
285	break;	6✔
286	}
287	WakeEvent::Start { .. } => {	×
288	log::error!("{}: device has already started", self.dev.name())	×
289	}
290	#[cfg(target_os = "linux")]
291	WakeEvent::VuChannel { channel } => self.dev.set_vu_channel(channel),	×
292	WakeEvent::Reset => {	×
293	log::info!("{}: guest requested reset", self.dev.name());	×
294	self.state = WorkerState::Pending;	×
295	break;	×
296	}
297	}
298	}
299	Ok(())	26✔
300	}
301
302	fn wait_start(&mut self) -> Option<StartParam<S, E>> {	6✔
303	for wake_event in self.event_rx.iter() {	16✔
304	match wake_event {	6✔
305	WakeEvent::Reset => {}	×
306	WakeEvent::Start { param } => {	6✔
307	self.state = WorkerState::Running;	6✔
308	return Some(param);	6✔
309	}
310	#[cfg(target_os = "linux")]
311	WakeEvent::VuChannel { channel } => self.dev.set_vu_channel(channel),	×
312	WakeEvent::Shutdown => break,	×
313	WakeEvent::Notify { q_index } => {	×
314	log::error!(	×
315	"{}: driver notified queue {q_index} before device is ready",
316	self.dev.name()	×
317	)
318	}
319	}
320	}
321	self.state = WorkerState::Shutdown;	×
322	None	×
323	}
324
325	pub fn handle_event<B>(&mut self, event: &B::Event, backend: &mut B) -> Result<()>	32✔
326	where
327	B: ActiveBackend<D>,
328	{
329	if event.token() == TOKEN_WARKER {	32✔
330	self.handle_wake_events(backend)	74✔
331	} else {
332	backend.handle_event(&mut self.dev, event)	7✔
333	}
334	}
335	}
336
337	impl<D, S, E, B> Worker<D, S, E, B>
338	where
339	D: Virtio,
340	S: IrqSender,
341	B: Backend<D>,
342	E: IoeventFd,
343	{
344	pub fn spawn(	6✔
345	dev: D,
346	mut backend: B,
347	event_rx: Receiver<WakeEvent<S, E>>,
348	memory: Arc<RamBus>,
349	queue_regs: Arc<[QueueReg]>,
350	) -> Result<(JoinHandle<()>, Arc<Notifier>)> {
351	let notifier = backend.register_notifier(TOKEN_WARKER)?;	16✔
352	let worker = Worker {
353	context: Context {	2✔
354	dev,
355	event_rx,
356	memory,
357	queue_regs,
358	state: WorkerState::Pending,
359	},
360	backend,
361	};
362	let name = worker.context.dev.name();	4✔
363	let handle = std::thread::Builder::new()	10✔
364	.name(name.to_owned())	4✔
365	.spawn(move \|\| worker.do_work())	14✔
366	.context(error::WorkerThread)?;	2✔
367	Ok((handle, notifier))	2✔
368	}
369
370	fn event_loop<'m, Q>(	6✔
371	&mut self,
372	queues: &mut [Option<Queue<'_, 'm, Q>>],
373	ram: &'m Ram,
374	param: &StartParam<S, E>,
375	) -> Result<()>
376	where
377	Q: VirtQueue<'m>,
378	E: IoeventFd,
379	{
380	log::debug!(	8✔
381	"{}: activated with {:x?} {:x?}",	×
382	self.context.dev.name(),	×
383	VirtioFeature::from_bits_retain(param.feature & !D::Feature::all().bits()),	×
384	D::Feature::from_bits_truncate(param.feature)	×
385	);
386	self.backend	6✔
387	.event_loop(ram, &mut self.context, queues, param)	22✔
388	}
389
390	fn loop_until_reset(&mut self) -> Result<()> {	6✔
391	let Some(param) = self.context.wait_start() else {	10✔
392	return Ok(());	×
393	};
394	let memory = self.context.memory.clone();	4✔
395	let ram = memory.lock_layout();	4✔
396	let feature = param.feature & !VirtioFeature::ACCESS_PLATFORM.bits();	4✔
397	let queue_regs = self.context.queue_regs.clone();	2✔
398	let feature = VirtioFeature::from_bits_retain(feature);	4✔
399	let event_idx = feature.contains(VirtioFeature::EVENT_IDX);	2✔
400	if feature.contains(VirtioFeature::RING_PACKED) {	2✔
401	let new_queue = \|reg\| {	×
402	let Some(split_queue) = PackedQueue::new(reg, &ram, event_idx)? else {	×
403	return Ok(None);	×
404	};
405	Ok(Some(Queue::new(split_queue, reg, &ram)))	×
406	};
407	let queues: Result<Box<_>> = queue_regs.iter().map(new_queue).collect();	×
408	self.event_loop(&mut (queues?), &ram, &param)?;	×
409	} else {
410	let new_queue = \|reg\| {	16✔
411	let Some(split_queue) = SplitQueue::new(reg, &ram, event_idx)? else {	42✔
412	return Ok(None);	×
413	};
414	Ok(Some(Queue::new(split_queue, reg, &ram)))	2✔
415	};
416	let queues: Result<Box<_>> = queue_regs.iter().map(new_queue).collect();	4✔
417	self.event_loop(&mut (queues?), &ram, &param)?;	6✔
418	};
419	self.backend.reset(&mut self.context.dev)?;	8✔
420	Ok(())	6✔
421	}
422
423	fn do_work(mut self) {	6✔
424	while self.context.state != WorkerState::Shutdown {	12✔
425	if let Err(e) = self.loop_until_reset() {	8✔
426	log::error!("worker {}: {e:?}", self.context.dev.name(),);	×
427	return;	×
428	}
429	}
430	log::debug!("worker {}: done", self.context.dev.name())	10✔
431	}
432	}
433
434	pub trait DevParam {
435	type Device;
436	fn build(self, name: impl Into<Arc<str>>) -> Result<Self::Device>;
437	fn needs_mem_shared_fd(&self) -> bool {	×
438	false	×
439	}
440	}

google / alioth / 18602811487

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous