vhost_user_backend/
lib.rs

1// Copyright 2019 Intel Corporation. All Rights Reserved.
2// Copyright 2019-2021 Alibaba Cloud Computing. All rights reserved.
3//
4// SPDX-License-Identifier: Apache-2.0
5
6//! A simple framework to run a vhost-user backend service.
7
8#[macro_use]
9extern crate log;
10
11use std::fmt::{Display, Formatter};
12use std::net::Shutdown;
13use std::os::unix::net::UnixStream;
14use std::path::Path;
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::{Arc, Mutex};
17use std::thread;
18
19use vhost::vhost_user::{BackendListener, BackendReqHandler, Error as VhostUserError, Listener};
20use vm_memory::mmap::NewBitmap;
21use vm_memory::{GuestMemoryAtomic, GuestMemoryMmap};
22
23use self::handler::VhostUserHandler;
24
25mod backend;
26pub use self::backend::{VhostUserBackend, VhostUserBackendMut};
27
28mod event_loop;
29pub use self::event_loop::VringEpollHandler;
30
31mod handler;
32pub use self::handler::VhostUserHandlerError;
33
34pub mod bitmap;
35use crate::bitmap::BitmapReplace;
36
37mod vring;
38pub use self::vring::{
39    VringMutex, VringRwLock, VringState, VringStateGuard, VringStateMutGuard, VringT,
40};
41
42// Due to the way `xen` handles memory mappings we can not combine it with
43// `postcopy` feature which relies on persistent memory mappings. Thus we
44// disallow enabling both features at the same time.
45#[cfg(all(
46    not(RUSTDOC_disable_feature_compat_errors),
47    not(doc),
48    feature = "postcopy",
49    feature = "xen"
50))]
51compile_error!("Both `postcopy` and `xen` features can not be enabled at the same time.");
52
53/// An alias for `GuestMemoryAtomic<GuestMemoryMmap<B>>` to simplify code.
54type GM<B> = GuestMemoryAtomic<GuestMemoryMmap<B>>;
55
56#[derive(Debug)]
57/// Errors related to vhost-user daemon.
58pub enum Error {
59    /// Failed to create a new vhost-user handler.
60    NewVhostUserHandler(VhostUserHandlerError),
61    /// Failed creating vhost-user backend listener.
62    CreateBackendListener(VhostUserError),
63    /// Failed creating vhost-user backend handler.
64    CreateBackendReqHandler(VhostUserError),
65    /// Failed creating listener socket
66    CreateVhostUserListener(VhostUserError),
67    /// Failed starting daemon thread.
68    StartDaemon(std::io::Error),
69    /// Failed waiting for daemon thread.
70    WaitDaemon(std::boxed::Box<dyn std::any::Any + std::marker::Send>),
71    /// Failed handling a vhost-user request.
72    HandleRequest(VhostUserError),
73}
74
75impl Display for Error {
76    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
77        match self {
78            Error::NewVhostUserHandler(e) => write!(f, "cannot create vhost user handler: {e}"),
79            Error::CreateBackendListener(e) => write!(f, "cannot create backend listener: {e}"),
80            Error::CreateBackendReqHandler(e) => {
81                write!(f, "cannot create backend req handler: {e}")
82            }
83            Error::CreateVhostUserListener(e) => {
84                write!(f, "cannot create vhost-user listener: {e}")
85            }
86            Error::StartDaemon(e) => write!(f, "failed to start daemon: {e}"),
87            Error::WaitDaemon(_e) => write!(f, "failed to wait for daemon exit"),
88            Error::HandleRequest(e) => write!(f, "failed to handle request: {e}"),
89        }
90    }
91}
92
93/// Result of vhost-user daemon operations.
94pub type Result<T> = std::result::Result<T, Error>;
95
96struct ConnectionState {
97    conn: UnixStream,
98    shutdown_requested: AtomicBool,
99}
100
101/// Thread-safe handle to request shutdown of a running [`VhostUserDaemon`].
102///
103/// Safe to call multiple times or after the connection has already closed.
104#[derive(Clone)]
105pub struct ShutdownHandle {
106    state: Arc<ConnectionState>,
107}
108
109impl ShutdownHandle {
110    /// Request the daemon to shut down.
111    pub fn shutdown(&self) {
112        self.state.shutdown_requested.store(true, Ordering::Release);
113        let _ = self.state.conn.shutdown(Shutdown::Both);
114    }
115}
116
117/// Implement a simple framework to run a vhost-user service daemon.
118///
119/// This structure is the public API the backend is allowed to interact with in order to run
120/// a fully functional vhost-user daemon.
121pub struct VhostUserDaemon<T: VhostUserBackend> {
122    name: String,
123    handler: Arc<Mutex<VhostUserHandler<T>>>,
124    main_thread: Option<thread::JoinHandle<Result<()>>>,
125    conn_state: Option<Arc<ConnectionState>>,
126}
127
128impl<T> VhostUserDaemon<T>
129where
130    T: VhostUserBackend + Clone + 'static,
131    T::Bitmap: BitmapReplace + NewBitmap + Clone + Send + Sync,
132    T::Vring: Clone + Send + Sync,
133{
134    /// Create the daemon instance, providing the backend implementation of `VhostUserBackend`.
135    ///
136    /// Under the hood, this will start a dedicated thread responsible for listening onto
137    /// registered event. Those events can be vring events or custom events from the backend,
138    /// but they get to be registered later during the sequence.
139    pub fn new(
140        name: String,
141        backend: T,
142        atomic_mem: GuestMemoryAtomic<GuestMemoryMmap<T::Bitmap>>,
143    ) -> Result<Self> {
144        let handler = Arc::new(Mutex::new(
145            VhostUserHandler::new(backend, atomic_mem).map_err(Error::NewVhostUserHandler)?,
146        ));
147
148        Ok(VhostUserDaemon {
149            name,
150            handler,
151            main_thread: None,
152            conn_state: None,
153        })
154    }
155
156    fn reset_connection_state(&mut self) {
157        self.conn_state = None;
158    }
159
160    /// Run a dedicated thread handling all requests coming through the socket.
161    /// This runs in an infinite loop that should be terminating once the other
162    /// end of the socket (the VMM) hangs up or [`request_shutdown`](Self::request_shutdown)
163    /// is called.
164    ///
165    /// This function is the common code for starting a new daemon, no matter if
166    /// it acts as a client or a server.
167    fn start_daemon(
168        &mut self,
169        mut handler: BackendReqHandler<Mutex<VhostUserHandler<T>>>,
170    ) -> Result<()> {
171        let state = Arc::new(ConnectionState {
172            conn: handler.try_clone_connection().map_err(Error::StartDaemon)?,
173            shutdown_requested: AtomicBool::new(false),
174        });
175
176        let thread_state = state.clone();
177        let handle = thread::Builder::new()
178            .name(self.name.clone())
179            .spawn(move || {
180                let result = loop {
181                    if let Err(e) = handler.handle_request().map_err(Error::HandleRequest) {
182                        break Err(e);
183                    }
184                };
185                let _ = thread_state.conn.shutdown(Shutdown::Both);
186                result
187            })
188            .map_err(Error::StartDaemon)?;
189
190        self.conn_state = Some(state);
191        self.main_thread = Some(handle);
192
193        Ok(())
194    }
195
196    /// Connect to the vhost-user socket and run a dedicated thread handling
197    /// all requests coming through this socket. This runs in an infinite loop
198    /// that should be terminating once the other end of the socket (the VMM)
199    /// hangs up.
200    pub fn start_client(&mut self, socket_path: &str) -> Result<()> {
201        let backend_handler = BackendReqHandler::connect(socket_path, self.handler.clone())
202            .map_err(Error::CreateBackendReqHandler)?;
203        self.start_daemon(backend_handler)
204    }
205
206    /// Listen to the vhost-user socket and run a dedicated thread handling all requests coming
207    /// through this socket.
208    ///
209    /// This runs in an infinite loop that should be terminating once the other end of the socket
210    /// (the VMM) disconnects.
211    ///
212    /// *Note:* A convenience function [VhostUserDaemon::serve] exists that
213    /// may be a better option than this for simple use-cases.
214    pub fn start(&mut self, listener: &mut Listener) -> Result<()> {
215        let mut backend_listener = BackendListener::new(listener, self.handler.clone())
216            .map_err(Error::CreateBackendListener)?;
217        let backend_handler = self.accept(&mut backend_listener)?;
218        self.start_daemon(backend_handler)
219    }
220
221    fn accept(
222        &self,
223        backend_listener: &mut BackendListener<Mutex<VhostUserHandler<T>>>,
224    ) -> Result<BackendReqHandler<Mutex<VhostUserHandler<T>>>> {
225        loop {
226            match backend_listener.accept() {
227                Err(e) => return Err(Error::CreateBackendListener(e)),
228                Ok(Some(v)) => return Ok(v),
229                Ok(None) => continue,
230            }
231        }
232    }
233
234    /// Wait for the thread handling the vhost-user socket connection to terminate.
235    ///
236    /// *Note:* A convenience function [VhostUserDaemon::serve] exists that
237    /// may be a better option than this for simple use-cases.
238    pub fn wait(&mut self) -> Result<()> {
239        let Some(handle) = self.main_thread.take() else {
240            self.reset_connection_state();
241            return Ok(());
242        };
243
244        let shutdown_requested = || {
245            self.conn_state
246                .as_ref()
247                .is_some_and(|s| s.shutdown_requested.load(Ordering::Acquire))
248        };
249
250        let result = match handle.join().map_err(Error::WaitDaemon)? {
251            Ok(()) => Ok(()),
252            Err(Error::HandleRequest(VhostUserError::SocketBroken(_))) => Ok(()),
253            Err(Error::HandleRequest(_)) if shutdown_requested() => Ok(()),
254            Err(error) => Err(error),
255        };
256
257        self.reset_connection_state();
258
259        result
260    }
261
262    /// Returns a handle to shut down the daemon, or `None` if not connected.
263    pub fn shutdown_handle(&self) -> Option<ShutdownHandle> {
264        self.conn_state
265            .as_ref()
266            .map(|s| ShutdownHandle { state: s.clone() })
267    }
268
269    /// Shut down the connection to unblock the daemon thread.
270    pub fn request_shutdown(&self) {
271        if let Some(handle) = self.shutdown_handle() {
272            handle.shutdown();
273        }
274    }
275
276    /// Bind to socket, handle a single connection and shutdown
277    ///
278    /// This is a convenience function that provides an easy way to handle the
279    /// following actions without needing to call the low-level functions:
280    /// - Create a listener
281    /// - Start listening
282    /// - Handle a single event
283    /// - Send the exit event to all handler threads
284    ///
285    /// Internal `Err` results that indicate a device disconnect will be treated
286    /// as success and `Ok(())` will be returned in those cases.
287    ///
288    /// *Note:* See [VhostUserDaemon::start] and [VhostUserDaemon::wait] if you
289    /// need more flexibility.
290    pub fn serve<P: AsRef<Path>>(&mut self, socket: P) -> Result<()> {
291        let mut listener = Listener::new(socket, true).map_err(Error::CreateVhostUserListener)?;
292
293        self.start(&mut listener)?;
294        let result = self.wait();
295
296        // Regardless of the result, we want to signal worker threads to exit
297        self.handler.lock().unwrap().send_exit_event();
298
299        // For this convenience function we are not treating certain "expected"
300        // outcomes as error. Disconnects and partial messages can be usual
301        // behaviour seen from quitting guests.
302        match &result {
303            Err(e) => match e {
304                Error::HandleRequest(VhostUserError::Disconnected) => Ok(()),
305                Error::HandleRequest(VhostUserError::PartialMessage) => Ok(()),
306                _ => result,
307            },
308            _ => result,
309        }
310    }
311
312    /// Retrieve the vring epoll handler.
313    ///
314    /// This is necessary to perform further actions like registering and unregistering some extra
315    /// event file descriptors.
316    pub fn get_epoll_handlers(&self) -> Vec<Arc<VringEpollHandler<T>>> {
317        // Do not expect poisoned lock.
318        self.handler.lock().unwrap().get_epoll_handlers()
319    }
320}
321
322impl<T: VhostUserBackend> Drop for VhostUserDaemon<T> {
323    fn drop(&mut self) {
324        if let Some(state) = self.conn_state.take() {
325            let _ = state.conn.shutdown(Shutdown::Both);
326        }
327    }
328}
329
330#[cfg(test)]
331mod tests {
332    use super::backend::tests::MockVhostBackend;
333    use super::*;
334    use libc::EAGAIN;
335    use std::os::unix::net::{UnixListener, UnixStream};
336    use std::sync::Barrier;
337    use std::time::Duration;
338    use vm_memory::{GuestAddress, GuestMemoryAtomic, GuestMemoryMmap};
339
340    #[test]
341    fn test_new_daemon() {
342        let mem = GuestMemoryAtomic::new(
343            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0x100000), 0x10000)]).unwrap(),
344        );
345        let backend = Arc::new(Mutex::new(MockVhostBackend::new()));
346        let mut daemon = VhostUserDaemon::new("test".to_owned(), backend, mem).unwrap();
347
348        let handlers = daemon.get_epoll_handlers();
349        assert_eq!(handlers.len(), 2);
350
351        let barrier = Arc::new(Barrier::new(2));
352        let tmpdir = tempfile::tempdir().unwrap();
353        let path = tmpdir.path().join("socket");
354
355        thread::scope(|s| {
356            s.spawn(|| {
357                barrier.wait();
358                let socket = UnixStream::connect(&path).unwrap();
359                barrier.wait();
360                drop(socket)
361            });
362
363            let mut listener = Listener::new(&path, false).unwrap();
364            barrier.wait();
365            daemon.start(&mut listener).unwrap();
366            barrier.wait();
367            // Above process generates a `HandleRequest(PartialMessage)` error.
368            daemon.wait().unwrap_err();
369            daemon.wait().unwrap();
370        });
371    }
372
373    #[test]
374    fn test_new_daemon_client() {
375        let mem = GuestMemoryAtomic::new(
376            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0x100000), 0x10000)]).unwrap(),
377        );
378        let backend = Arc::new(Mutex::new(MockVhostBackend::new()));
379        let mut daemon = VhostUserDaemon::new("test".to_owned(), backend, mem).unwrap();
380
381        let handlers = daemon.get_epoll_handlers();
382        assert_eq!(handlers.len(), 2);
383
384        let barrier = Arc::new(Barrier::new(2));
385        let tmpdir = tempfile::tempdir().unwrap();
386        let path = tmpdir.path().join("socket");
387
388        thread::scope(|s| {
389            s.spawn(|| {
390                let listener = UnixListener::bind(&path).unwrap();
391                barrier.wait();
392                let (stream, _) = listener.accept().unwrap();
393                barrier.wait();
394                drop(stream)
395            });
396
397            barrier.wait();
398            daemon
399                .start_client(path.as_path().to_str().unwrap())
400                .unwrap();
401            barrier.wait();
402            // Above process generates a `HandleRequest(PartialMessage)` error.
403            daemon.wait().unwrap_err();
404            daemon.wait().unwrap();
405        });
406    }
407
408    #[test]
409    fn test_daemon_serve() {
410        let mem = GuestMemoryAtomic::new(
411            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0x100000), 0x10000)]).unwrap(),
412        );
413        let backend = Arc::new(Mutex::new(MockVhostBackend::new()));
414        let mut daemon = VhostUserDaemon::new("test".to_owned(), backend.clone(), mem).unwrap();
415        let tmpdir = tempfile::tempdir().unwrap();
416        let socket_path = tmpdir.path().join("socket");
417
418        thread::scope(|s| {
419            s.spawn(|| {
420                let _ = daemon.serve(&socket_path);
421            });
422
423            // We have no way to wait for when the server becomes available...
424            // So we will have to spin!
425            while !socket_path.exists() {
426                thread::sleep(Duration::from_millis(10));
427            }
428
429            // Check that no exit events got triggered yet
430            for thread_id in 0..backend.queues_per_thread().len() {
431                let fd = backend.exit_event(thread_id).unwrap();
432                // Reading from exit fd should fail since nothing was written yet
433                assert_eq!(
434                    fd.0.consume().unwrap_err().raw_os_error().unwrap(),
435                    EAGAIN,
436                    "exit event should not have been raised yet!"
437                );
438            }
439
440            let socket = UnixStream::connect(&socket_path).unwrap();
441            // disconnect immediately again
442            drop(socket);
443        });
444
445        // Check that exit events got triggered
446        let backend = backend.lock().unwrap();
447        for thread_id in 0..backend.queues_per_thread().len() {
448            let fd = backend.exit_event(thread_id).unwrap();
449            assert!(fd.0.consume().is_ok(), "No exit event was raised!");
450        }
451    }
452
453    #[test]
454    fn test_shutdown_while_connected() {
455        let mem = GuestMemoryAtomic::new(
456            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0x100000), 0x10000)]).unwrap(),
457        );
458        let backend = Arc::new(Mutex::new(MockVhostBackend::new()));
459        let mut daemon = VhostUserDaemon::new("test".to_owned(), backend, mem).unwrap();
460
461        let barrier = Arc::new(Barrier::new(2));
462        let tmpdir = tempfile::tempdir().unwrap();
463        let path = tmpdir.path().join("socket");
464
465        thread::scope(|s| {
466            let barrier_client = barrier.clone();
467            let path_client = path.clone();
468            let _client = s.spawn(move || {
469                barrier_client.wait();
470                let stream = UnixStream::connect(&path_client).unwrap();
471                barrier_client.wait();
472                drop(stream);
473            });
474
475            let mut listener = Listener::new(&path, false).unwrap();
476            barrier.wait();
477            daemon.start(&mut listener).unwrap();
478
479            let handle = daemon.shutdown_handle().expect("daemon started");
480            handle.shutdown();
481
482            daemon.wait().unwrap();
483            assert!(daemon.shutdown_handle().is_none());
484            barrier.wait();
485
486            let barrier = Arc::new(Barrier::new(2));
487            let barrier_client = barrier.clone();
488            let path_client = path.clone();
489            let _client = s.spawn(move || {
490                barrier_client.wait();
491                let stream = UnixStream::connect(&path_client).unwrap();
492                barrier_client.wait();
493                drop(stream);
494            });
495
496            barrier.wait();
497            daemon.start(&mut listener).unwrap();
498            barrier.wait();
499            daemon.wait().unwrap_err();
500            assert!(daemon.shutdown_handle().is_none());
501        });
502    }
503
504    #[test]
505    fn test_double_shutdown() {
506        let mem = GuestMemoryAtomic::new(
507            GuestMemoryMmap::<()>::from_ranges(&[(GuestAddress(0x100000), 0x10000)]).unwrap(),
508        );
509        let backend = Arc::new(Mutex::new(MockVhostBackend::new()));
510        let mut daemon = VhostUserDaemon::new("test".to_owned(), backend, mem).unwrap();
511
512        let barrier = Arc::new(Barrier::new(2));
513        let tmpdir = tempfile::tempdir().unwrap();
514        let path = tmpdir.path().join("socket");
515
516        thread::scope(|s| {
517            let barrier_client = barrier.clone();
518            let path_client = path.clone();
519            let _client = s.spawn(move || {
520                barrier_client.wait();
521                let stream = UnixStream::connect(&path_client).unwrap();
522                barrier_client.wait();
523                drop(stream);
524            });
525
526            let mut listener = Listener::new(&path, false).unwrap();
527            barrier.wait();
528            daemon.start(&mut listener).unwrap();
529
530            let handle = daemon.shutdown_handle().expect("daemon started");
531            handle.shutdown();
532            handle.shutdown();
533
534            daemon.wait().unwrap();
535            barrier.wait();
536        });
537    }
538
539    #[test]
540    fn test_shutdown_handle_is_send_sync() {
541        fn assert_send_sync<T: Send + Sync>() {}
542        assert_send_sync::<ShutdownHandle>();
543    }
544}
vhost_user_backend/lib.rs

vhost_user_backend/
lib.rs
