devices/virtio/vhost_user_frontend/
mod.rs1mod error;
8mod handler;
9mod sys;
10mod worker;
11
12use std::cell::RefCell;
13use std::collections::BTreeMap;
14use std::io::Read;
15use std::io::Write;
16
17use anyhow::bail;
18use anyhow::Context;
19use base::error;
20use base::trace;
21use base::AsRawDescriptor;
22#[cfg(windows)]
23use base::CloseNotifier;
24use base::Event;
25use base::RawDescriptor;
26use base::ReadNotifier;
27use base::SafeDescriptor;
28use base::SendTube;
29use base::WorkerThread;
30use snapshot::AnySnapshot;
31use vm_memory::GuestMemory;
32use vmm_vhost::message::VhostUserConfigFlags;
33use vmm_vhost::message::VhostUserMigrationPhase;
34use vmm_vhost::message::VhostUserProtocolFeatures;
35use vmm_vhost::message::VhostUserTransferDirection;
36use vmm_vhost::BackendClient;
37use vmm_vhost::VhostUserMemoryRegionInfo;
38use vmm_vhost::VringConfigData;
39use vmm_vhost::VHOST_USER_F_PROTOCOL_FEATURES;
40
41use crate::virtio::device_constants::VIRTIO_DEVICE_TYPE_SPECIFIC_FEATURES_MASK;
42use crate::virtio::vhost_user_frontend::error::Error;
43use crate::virtio::vhost_user_frontend::error::Result;
44use crate::virtio::vhost_user_frontend::handler::BackendReqHandler;
45use crate::virtio::vhost_user_frontend::handler::BackendReqHandlerImpl;
46use crate::virtio::vhost_user_frontend::sys::create_backend_req_handler;
47use crate::virtio::vhost_user_frontend::worker::Worker;
48use crate::virtio::DeviceType;
49use crate::virtio::Interrupt;
50use crate::virtio::Queue;
51use crate::virtio::SharedMemoryMapper;
52use crate::virtio::SharedMemoryRegion;
53use crate::virtio::VirtioDevice;
54use crate::PciAddress;
55
56pub struct VhostUserFrontend {
57 device_type: DeviceType,
58 worker_thread: Option<WorkerThread<(Option<BackendReqHandler>, SendTube)>>,
59
60 backend_client: BackendClient,
61 avail_features: u64,
62 acked_features: u64,
63 last_acked_features: u64,
65 protocol_features: VhostUserProtocolFeatures,
66 backend_req_handler: Option<BackendReqHandler>,
70 shmem_region: RefCell<Option<Option<SharedMemoryRegion>>>,
72
73 queue_sizes: Vec<u16>,
74 expose_shmem_descriptors_with_viommu: bool,
75 pci_address: Option<PciAddress>,
76 vm_evt_wrtube: SendTube,
77
78 sent_queues: Option<BTreeMap<usize, Queue>>,
82}
83
84fn power_of_two_le(val: u16) -> Option<u16> {
86 if val == 0 {
87 None
88 } else if val.is_power_of_two() {
89 Some(val)
90 } else {
91 val.checked_next_power_of_two()
92 .map(|next_pow_two| next_pow_two / 2)
93 }
94}
95
96impl VhostUserFrontend {
97 pub fn new(
108 device_type: DeviceType,
109 mut base_features: u64,
110 connection: vmm_vhost::Connection,
111 vm_evt_wrtube: SendTube,
112 max_queue_size: Option<u16>,
113 pci_address: Option<PciAddress>,
114 is_remote_backend: bool,
115 ) -> Result<VhostUserFrontend> {
116 if base_features & (1 << virtio_sys::virtio_config::VIRTIO_F_RING_PACKED) != 0 {
120 base_features &= !(1 << virtio_sys::virtio_config::VIRTIO_F_RING_PACKED);
121 base::warn!(
122 "VIRTIO_F_RING_PACKED requested, but not yet supported by vhost-user frontend. \
123 Automatically disabled."
124 );
125 }
126
127 #[cfg(windows)]
128 let backend_pid = connection.target_pid();
129
130 let mut backend_client = BackendClient::new(connection);
131
132 backend_client.set_owner().map_err(Error::SetOwner)?;
133
134 let allow_features = VIRTIO_DEVICE_TYPE_SPECIFIC_FEATURES_MASK
135 | base_features
136 | 1 << VHOST_USER_F_PROTOCOL_FEATURES;
137 let avail_features =
138 allow_features & backend_client.get_features().map_err(Error::GetFeatures)?;
139 let mut acked_features = 0;
140
141 let allow_protocol_features = VhostUserProtocolFeatures::CONFIG
142 | VhostUserProtocolFeatures::MQ
143 | VhostUserProtocolFeatures::BACKEND_REQ
144 | VhostUserProtocolFeatures::DEVICE_STATE
145 | VhostUserProtocolFeatures::SHMEM
146 | VhostUserProtocolFeatures::REPLY_ACK;
152
153 let mut protocol_features = VhostUserProtocolFeatures::empty();
154 if avail_features & 1 << VHOST_USER_F_PROTOCOL_FEATURES != 0 {
155 acked_features |= 1 << VHOST_USER_F_PROTOCOL_FEATURES;
163
164 let avail_protocol_features = backend_client
165 .get_protocol_features()
166 .map_err(Error::GetProtocolFeatures)?;
167 protocol_features = allow_protocol_features & avail_protocol_features;
168 backend_client
169 .set_protocol_features(protocol_features)
170 .map_err(Error::SetProtocolFeatures)?;
171 }
172
173 let backend_req_handler =
175 if protocol_features.contains(VhostUserProtocolFeatures::BACKEND_REQ) {
176 let (mut handler, tx_fd) = create_backend_req_handler(
177 BackendReqHandlerImpl::new(is_remote_backend),
178 #[cfg(windows)]
179 backend_pid,
180 )?;
181 handler.set_reply_ack_flag(
182 protocol_features.contains(VhostUserProtocolFeatures::REPLY_ACK),
183 );
184 backend_client
185 .set_backend_req_fd(&tx_fd)
186 .map_err(Error::SetDeviceRequestChannel)?;
187 Some(handler)
188 } else {
189 None
190 };
191
192 let num_queues = if protocol_features.contains(VhostUserProtocolFeatures::MQ) {
196 trace!("backend supports VHOST_USER_PROTOCOL_F_MQ");
197 let num_queues = backend_client.get_queue_num().map_err(Error::GetQueueNum)?;
198 trace!("VHOST_USER_GET_QUEUE_NUM returned {num_queues}");
199 num_queues as usize
200 } else {
201 trace!("backend does not support VHOST_USER_PROTOCOL_F_MQ");
202 device_type.min_queues()
203 };
204
205 let max_queue_size = max_queue_size
207 .and_then(power_of_two_le)
208 .unwrap_or(Queue::MAX_SIZE);
209
210 trace!(
211 "vhost-user {device_type} frontend with {num_queues} queues x {max_queue_size} entries\
212 {}",
213 if let Some(pci_address) = pci_address {
214 format!(" pci-address {pci_address}")
215 } else {
216 "".to_string()
217 }
218 );
219
220 let queue_sizes = vec![max_queue_size; num_queues];
221
222 Ok(VhostUserFrontend {
223 device_type,
224 worker_thread: None,
225 backend_client,
226 avail_features,
227 acked_features,
228 last_acked_features: acked_features,
229 protocol_features,
230 backend_req_handler,
231 shmem_region: RefCell::new(None),
232 queue_sizes,
233 expose_shmem_descriptors_with_viommu: device_type == DeviceType::Gpu,
234 pci_address,
235 vm_evt_wrtube,
236 sent_queues: None,
237 })
238 }
239
240 fn set_mem_table(&mut self, mem: &GuestMemory) -> Result<()> {
241 let regions: Vec<_> = mem
242 .regions()
243 .map(|region| VhostUserMemoryRegionInfo {
244 guest_phys_addr: region.guest_addr.0,
245 memory_size: region.size as u64,
246 userspace_addr: region.host_addr as u64,
247 mmap_offset: region.shm_offset,
248 mmap_handle: region.shm.as_raw_descriptor(),
249 })
250 .collect();
251
252 self.backend_client
253 .set_mem_table(regions.as_slice())
254 .map_err(Error::SetMemTable)?;
255
256 Ok(())
257 }
258
259 fn activate_vring(
261 &mut self,
262 mem: &GuestMemory,
263 queue_index: usize,
264 queue: &Queue,
265 irqfd: &Event,
266 ) -> Result<()> {
267 self.backend_client
268 .set_vring_num(queue_index, queue.size())
269 .map_err(Error::SetVringNum)?;
270
271 let config_data = VringConfigData {
272 queue_size: queue.size(),
273 flags: 0u32,
274 desc_table_addr: mem
275 .get_host_address(queue.desc_table())
276 .map_err(Error::GetHostAddress)? as u64,
277 used_ring_addr: mem
278 .get_host_address(queue.used_ring())
279 .map_err(Error::GetHostAddress)? as u64,
280 avail_ring_addr: mem
281 .get_host_address(queue.avail_ring())
282 .map_err(Error::GetHostAddress)? as u64,
283 log_addr: None,
284 };
285 self.backend_client
286 .set_vring_addr(queue_index, &config_data)
287 .map_err(Error::SetVringAddr)?;
288
289 self.backend_client
290 .set_vring_base(queue_index, queue.next_avail_to_process())
291 .map_err(Error::SetVringBase)?;
292
293 self.backend_client
294 .set_vring_call(queue_index, irqfd)
295 .map_err(Error::SetVringCall)?;
296 self.backend_client
297 .set_vring_kick(queue_index, queue.event())
298 .map_err(Error::SetVringKick)?;
299
300 if self.acked_features & 1 << VHOST_USER_F_PROTOCOL_FEATURES != 0 {
303 self.backend_client
304 .set_vring_enable(queue_index, true)
305 .map_err(Error::SetVringEnable)?;
306 }
307
308 Ok(())
309 }
310
311 fn deactivate_vring(&self, queue_index: usize) -> Result<u16> {
313 if self.acked_features & 1 << VHOST_USER_F_PROTOCOL_FEATURES != 0 {
314 self.backend_client
315 .set_vring_enable(queue_index, false)
316 .map_err(Error::SetVringEnable)?;
317 }
318
319 let vring_base = self
320 .backend_client
321 .get_vring_base(queue_index)
322 .map_err(Error::GetVringBase)?;
323
324 vring_base
325 .try_into()
326 .map_err(|_| Error::VringBaseTooBig(vring_base))
327 }
328
329 fn start_worker(&mut self, interrupt: Interrupt, non_msix_evt: Event) {
332 assert!(
333 self.worker_thread.is_none(),
334 "BUG: attempted to start worker twice"
335 );
336
337 let label = self.debug_label();
338
339 let mut backend_req_handler = self.backend_req_handler.take();
340 if let Some(handler) = &mut backend_req_handler {
341 handler.frontend_mut().set_interrupt(interrupt.clone());
343 }
344
345 let backend_client_read_notifier =
346 SafeDescriptor::try_from(self.backend_client.get_read_notifier())
347 .expect("failed to get backend read notifier");
348 #[cfg(windows)]
349 let backend_client_close_notifier =
350 SafeDescriptor::try_from(self.backend_client.get_close_notifier())
351 .expect("failed to get backend close notifier");
352
353 let vm_evt_wrtube = self
354 .vm_evt_wrtube
355 .try_clone()
356 .expect("failed to clone vm_evt_wrtube");
357
358 self.worker_thread = Some(WorkerThread::start(label.clone(), move |kill_evt| {
359 let mut worker = Worker {
360 kill_evt,
361 non_msix_evt,
362 backend_req_handler,
363 backend_client_read_notifier,
364 #[cfg(windows)]
365 backend_client_close_notifier,
366 };
367 if let Err(e) = worker
368 .run(interrupt)
369 .with_context(|| format!("{label}: vhost_user_frontend worker failed"))
370 {
371 error!("vhost-user worker thread exited with an error: {:#}", e);
372
373 if let Err(e) = vm_evt_wrtube.send(&base::VmEventType::DeviceCrashed) {
374 error!("failed to send crash event: {}", e);
375 }
376 }
377 (worker.backend_req_handler, vm_evt_wrtube)
378 }));
379 }
380}
381
382impl VirtioDevice for VhostUserFrontend {
383 fn debug_label(&self) -> String {
385 format!("vu-{}", self.device_type())
386 }
387
388 fn keep_rds(&self) -> Vec<RawDescriptor> {
389 Vec::new()
390 }
391
392 fn device_type(&self) -> DeviceType {
393 self.device_type
394 }
395
396 fn queue_max_sizes(&self) -> &[u16] {
397 &self.queue_sizes
398 }
399
400 fn features(&self) -> u64 {
401 self.avail_features
402 }
403
404 fn ack_features(&mut self, features: u64) {
405 self.acked_features |= features & self.avail_features;
406 }
407
408 fn read_config(&self, offset: u64, data: &mut [u8]) {
409 let Ok(offset) = offset.try_into() else {
410 error!("failed to read config: invalid config offset is given: {offset}");
411 return;
412 };
413 let Ok(data_len) = data.len().try_into() else {
414 error!(
415 "failed to read config: invalid config length is given: {}",
416 data.len()
417 );
418 return;
419 };
420 let (_, config) = match self.backend_client.get_config(
421 offset,
422 data_len,
423 VhostUserConfigFlags::WRITABLE,
424 data,
425 ) {
426 Ok(x) => x,
427 Err(e) => {
428 error!("failed to read config: {}", Error::GetConfig(e));
429 return;
430 }
431 };
432 data.copy_from_slice(&config);
433 }
434
435 fn write_config(&mut self, offset: u64, data: &[u8]) {
436 let Ok(offset) = offset.try_into() else {
437 error!("failed to write config: invalid config offset is given: {offset}");
438 return;
439 };
440 if let Err(e) = self
441 .backend_client
442 .set_config(offset, VhostUserConfigFlags::empty(), data)
443 .map_err(Error::SetConfig)
444 {
445 error!("failed to write config: {}", e);
446 }
447 }
448
449 fn activate(
450 &mut self,
451 mem: GuestMemory,
452 interrupt: Interrupt,
453 queues: BTreeMap<usize, Queue>,
454 ) -> anyhow::Result<()> {
455 if self.last_acked_features != self.acked_features {
456 self.backend_client
457 .set_features(self.acked_features)
458 .map_err(Error::SetFeatures)?;
459 self.last_acked_features = self.acked_features;
460 }
461
462 self.set_mem_table(&mem)?;
463
464 let msix_config_opt = interrupt
465 .get_msix_config()
466 .as_ref()
467 .ok_or(Error::MsixConfigUnavailable)?;
468 let msix_config = msix_config_opt.lock();
469
470 let non_msix_evt = Event::new().map_err(Error::CreateEvent)?;
471 for (&queue_index, queue) in queues.iter() {
472 let irqfd = msix_config
473 .get_irqfd(queue.vector() as usize)
474 .unwrap_or(&non_msix_evt);
475 self.activate_vring(&mem, queue_index, queue, irqfd)?;
476 }
477
478 self.sent_queues = Some(queues);
479
480 drop(msix_config);
481
482 self.start_worker(interrupt, non_msix_evt);
483 Ok(())
484 }
485
486 fn reset(&mut self) -> anyhow::Result<()> {
487 if let Some(sent_queues) = self.sent_queues.take() {
491 for queue_index in sent_queues.into_keys() {
492 let _vring_base = self
493 .deactivate_vring(queue_index)
494 .context("deactivate_vring failed during reset")?;
495 }
496 }
497
498 if let Some(w) = self.worker_thread.take() {
499 let (backend_req_handler, vm_evt_wrtube) = w.stop();
500 self.backend_req_handler = backend_req_handler;
501 self.vm_evt_wrtube = vm_evt_wrtube;
502 }
503
504 Ok(())
505 }
506
507 fn pci_address(&self) -> Option<PciAddress> {
508 self.pci_address
509 }
510
511 fn get_shared_memory_region(&self) -> Option<SharedMemoryRegion> {
512 if !self
513 .protocol_features
514 .contains(VhostUserProtocolFeatures::SHMEM)
515 {
516 return None;
517 }
518 if let Some(r) = self.shmem_region.borrow().as_ref() {
519 return *r;
520 }
521 let regions = match self
522 .backend_client
523 .get_shmem_config()
524 .map_err(Error::ShmemRegions)
525 {
526 Ok(x) => x,
527 Err(e) => {
528 error!("Failed to get shared memory config {}", e);
529 return None;
530 }
531 };
532 let region = match regions.len() {
533 0 => None,
534 1 => Some(regions[0]),
535 n => {
536 error!(
537 "Failed to get shared memory region {}",
538 Error::TooManyShmemRegions(n)
539 );
540 return None;
541 }
542 };
543 *self.shmem_region.borrow_mut() = Some(region);
544 region
545 }
546
547 fn set_shared_memory_mapper(&mut self, mapper: Box<dyn SharedMemoryMapper>) {
548 let Some(backend_req_handler) = self.backend_req_handler.as_mut() else {
551 error!(
552 "Error setting shared memory mapper {}",
553 Error::ProtocolFeatureNotNegoiated(VhostUserProtocolFeatures::BACKEND_REQ)
554 );
555 return;
556 };
557
558 let shmid = self
560 .shmem_region
561 .borrow()
562 .flatten()
563 .expect("missing shmid")
564 .id;
565
566 backend_req_handler
567 .frontend_mut()
568 .set_shared_mapper_state(mapper, shmid);
569 }
570
571 fn expose_shmem_descriptors_with_viommu(&self) -> bool {
572 self.expose_shmem_descriptors_with_viommu
573 }
574
575 fn virtio_sleep(&mut self) -> anyhow::Result<Option<BTreeMap<usize, Queue>>> {
576 let Some(mut queues) = self.sent_queues.take() else {
577 return Ok(None);
578 };
579
580 for (&queue_index, queue) in queues.iter_mut() {
581 let vring_base = self
582 .deactivate_vring(queue_index)
583 .context("deactivate_vring failed during sleep")?;
584 queue.vhost_user_reclaim(vring_base);
585 }
586
587 if let Some(w) = self.worker_thread.take() {
588 let (backend_req_handler, vm_evt_wrtube) = w.stop();
589 self.backend_req_handler = backend_req_handler;
590 self.vm_evt_wrtube = vm_evt_wrtube;
591 }
592
593 Ok(Some(queues))
594 }
595
596 fn virtio_wake(
597 &mut self,
598 queues_state: Option<(GuestMemory, Interrupt, BTreeMap<usize, Queue>)>,
599 ) -> anyhow::Result<()> {
600 if let Some((mem, interrupt, queues)) = queues_state {
601 self.activate(mem, interrupt, queues)?;
602 }
603 Ok(())
604 }
605
606 fn virtio_snapshot(&mut self) -> anyhow::Result<AnySnapshot> {
607 if !self
608 .protocol_features
609 .contains(VhostUserProtocolFeatures::DEVICE_STATE)
610 {
611 bail!("snapshot requires VHOST_USER_PROTOCOL_F_DEVICE_STATE");
612 }
613 let (mut r, w) = new_pipe_pair()?;
616 let backend_r = self
617 .backend_client
618 .set_device_state_fd(
619 VhostUserTransferDirection::Save,
620 VhostUserMigrationPhase::Stopped,
621 &w,
622 )
623 .context("failed to negotiate device state fd")?;
624 std::mem::drop(w);
627 let mut snapshot_bytes = Vec::new();
629 if let Some(mut backend_r) = backend_r {
630 backend_r.read_to_end(&mut snapshot_bytes)
631 } else {
632 r.read_to_end(&mut snapshot_bytes)
633 }
634 .context("failed to read device state")?;
635 self.backend_client
637 .check_device_state()
638 .context("failed to transfer device state")?;
639 Ok(AnySnapshot::to_any(VhostUserDeviceState {
640 acked_features: self.acked_features,
641 backend_state: snapshot_bytes,
642 })
643 .map_err(Error::SliceToSerdeValue)?)
644 }
645
646 fn virtio_restore(&mut self, data: AnySnapshot) -> anyhow::Result<()> {
647 if !self
648 .protocol_features
649 .contains(VhostUserProtocolFeatures::DEVICE_STATE)
650 {
651 bail!("restore requires VHOST_USER_PROTOCOL_F_DEVICE_STATE");
652 }
653
654 let device_state: VhostUserDeviceState =
655 AnySnapshot::from_any(data).map_err(Error::SerdeValueToSlice)?;
656
657 let missing_features = !self.avail_features & device_state.acked_features;
659 if missing_features != 0 {
660 bail!("The destination backend doesn't support all features acknowledged by the source, missing: {}", missing_features);
661 }
662 self.acked_features = device_state.acked_features;
663 if self.last_acked_features != self.acked_features {
664 self.backend_client
665 .set_features(self.acked_features)
666 .map_err(Error::SetFeatures)?;
667 self.last_acked_features = self.acked_features;
668 }
669
670 let (r, w) = new_pipe_pair()?;
673 let backend_w = self
674 .backend_client
675 .set_device_state_fd(
676 VhostUserTransferDirection::Load,
677 VhostUserMigrationPhase::Stopped,
678 &r,
679 )
680 .context("failed to negotiate device state fd")?;
681 {
683 let backend_w = backend_w;
687 let mut w = w;
688 if let Some(mut backend_w) = backend_w {
689 backend_w.write_all(device_state.backend_state.as_slice())
690 } else {
691 w.write_all(device_state.backend_state.as_slice())
692 }
693 .context("failed to write device state")?;
694 }
695 self.backend_client
697 .check_device_state()
698 .context("failed to transfer device state")?;
699 Ok(())
700 }
701}
702
703#[derive(serde::Serialize, serde::Deserialize, Debug)]
704struct VhostUserDeviceState {
705 acked_features: u64,
706 backend_state: Vec<u8>,
707}
708
709#[cfg(unix)]
710fn new_pipe_pair() -> anyhow::Result<(impl AsRawDescriptor + Read, impl AsRawDescriptor + Write)> {
711 base::pipe().context("failed to create pipe")
712}
713
714#[cfg(windows)]
715fn new_pipe_pair() -> anyhow::Result<(impl AsRawDescriptor + Read, impl AsRawDescriptor + Write)> {
716 base::named_pipes::pair(
717 &base::named_pipes::FramingMode::Byte,
718 &base::named_pipes::BlockingMode::Wait,
719 0,
720 )
721 .context("failed to create named pipes")
722}