Use mio to replace Epoll

Epoll is linux-specific. So we use mio, which is a cross-platform
event notification, to replace Epoll.

Signed-off-by: Wenyu Huang <[email protected]>

Author: uran0sH

vhost-user-backend/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Added
 ### Changed
 - [[#308](https://github.com/rust-vmm/vhost/pull/308)] Replace Eventfd with EventNotifier/EventConsumer.
+- [[316]](https://github.com/rust-vmm/vhost/pull/316) Use mio to replace Epoll. Expose event_loop::EventSet.
 
 ### Deprecated
 ### Fixed

vhost-user-backend/Cargo.toml

@@ -16,6 +16,7 @@ postcopy = ["vhost/postcopy", "userfaultfd"]
 libc = "0.2.39"
 log = "0.4.17"
 userfaultfd = { version = "0.9.0", optional = true }
+mio = { version = "1.0.4", features = ["os-poll", "os-ext"] }
 vhost = { path = "../vhost", version = "0.14.0", features = ["vhost-user-backend"] }
 virtio-bindings = { workspace = true }
 virtio-queue = { workspace = true }

vhost-user-backend/src/backend.rs

@@ -29,14 +29,15 @@ use vhost::vhost_user::message::{
 };
 use vhost::vhost_user::Backend;
 use vm_memory::bitmap::Bitmap;
-use vmm_sys_util::epoll::EventSet;
 use vmm_sys_util::event::{EventConsumer, EventNotifier};
 
 use vhost::vhost_user::GpuBackend;
 
 use super::vring::VringT;
 use super::GM;
 
+use crate::EventSet;
+
 /// Trait with interior mutability for vhost user backend servers to implement concrete services.
 ///
 /// To support multi-threading and asynchronous IO, we enforce `Send + Sync` bound.
@@ -798,7 +799,7 @@ pub mod tests {
 
         let vring = VringRwLock::new(mem, 0x1000).unwrap();
         backend
-            .handle_event(0x1, EventSet::IN, &[vring], 0)
+            .handle_event(0x1, EventSet::Readable, &[vring], 0)
             .unwrap();
 
         backend.reset_device();

vhost-user-backend/src/event_loop.rs

@@ -3,53 +3,90 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
+use std::collections::HashSet;
 use std::fmt::{Display, Formatter};
 use std::io::{self, Result};
 use std::marker::PhantomData;
 use std::os::fd::IntoRawFd;
 use std::os::unix::io::{AsRawFd, RawFd};
+use std::sync::Mutex;
 
-use vmm_sys_util::epoll::{ControlOperation, Epoll, EpollEvent, EventSet};
+use mio::event::Event;
+use mio::unix::SourceFd;
+use mio::{Events, Interest, Poll, Registry, Token};
 use vmm_sys_util::event::EventNotifier;
 
 use super::backend::VhostUserBackend;
 use super::vring::VringT;
 
 /// Errors related to vring epoll event handling.
 #[derive(Debug)]
-pub enum VringEpollError {
+pub enum VringPollError {
     /// Failed to create epoll file descriptor.
-    EpollCreateFd(io::Error),
+    PollerCreate(io::Error),
     /// Failed while waiting for events.
-    EpollWait(io::Error),
+    PollerWait(io::Error),
     /// Could not register exit event
     RegisterExitEvent(io::Error),
     /// Failed to read the event from kick EventFd.
     HandleEventReadKick(io::Error),
     /// Failed to handle the event from the backend.
     HandleEventBackendHandling(io::Error),
+    /// Failed to clone registry.
+    RegistryClone(io::Error),
 }
 
-impl Display for VringEpollError {
+impl Display for VringPollError {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {
-            VringEpollError::EpollCreateFd(e) => write!(f, "cannot create epoll fd: {e}"),
-            VringEpollError::EpollWait(e) => write!(f, "failed to wait for epoll event: {e}"),
-            VringEpollError::RegisterExitEvent(e) => write!(f, "cannot register exit event: {e}"),
-            VringEpollError::HandleEventReadKick(e) => {
+            VringPollError::PollerCreate(e) => write!(f, "cannot create poller: {e}"),
+            VringPollError::PollerWait(e) => write!(f, "failed to wait for poller event: {e}"),
+            VringPollError::RegisterExitEvent(e) => write!(f, "cannot register exit event: {e}"),
+            VringPollError::HandleEventReadKick(e) => {
                 write!(f, "cannot read vring kick event: {e}")
             }
-            VringEpollError::HandleEventBackendHandling(e) => {
-                write!(f, "failed to handle epoll event: {e}")
+            VringPollError::HandleEventBackendHandling(e) => {
+                write!(f, "failed to handle poll event: {e}")
             }
+            VringPollError::RegistryClone(e) => write!(f, "cannot clone poller's registry: {e}"),
         }
     }
 }
 
-impl std::error::Error for VringEpollError {}
+impl std::error::Error for VringPollError {}
 
 /// Result of vring epoll operations.
-pub type VringEpollResult<T> = std::result::Result<T, VringEpollError>;
+pub type VringEpollResult<T> = std::result::Result<T, VringPollError>;
+
+#[derive(Debug, Clone, Copy)]
+pub enum EventSet {
+    Readable,
+    Writable,
+    All,
+}
+
+impl EventSet {
+    fn to_interest(self) -> Interest {
+        match self {
+            EventSet::Readable => Interest::READABLE,
+            EventSet::Writable => Interest::WRITABLE,
+            EventSet::All => Interest::READABLE | Interest::WRITABLE,
+        }
+    }
+}
+
+fn event_to_event_set(evt: &Event) -> Option<EventSet> {
+    if evt.is_readable() && evt.is_writable() {
+        return Some(EventSet::All);
+    }
+    if evt.is_readable() {
+        return Some(EventSet::Readable);
+    }
+    if evt.is_writable() {
+        return Some(EventSet::Writable);
+    }
+    None
+}
 
 /// Epoll event handler to manage and process epoll events for registered file descriptor.
 ///
@@ -58,7 +95,11 @@ pub type VringEpollResult<T> = std::result::Result<T, VringEpollError>;
 /// - remove registered file descriptors from the epoll fd
 /// - run the event loop to handle pending events on the epoll fd
 pub struct VringEpollHandler<T: VhostUserBackend> {
-    epoll: Epoll,
+    poller: Mutex<Poll>,
+    registry: Registry,
+    // Record the registered fd.
+    // Because in mio, consecutive calls to register is unspecified behavior.
+    fd_set: Mutex<HashSet<RawFd>>,
     backend: T,
     vrings: Vec<T::Vring>,
     thread_id: usize,
@@ -85,25 +126,35 @@ where
         vrings: Vec<T::Vring>,
         thread_id: usize,
     ) -> VringEpollResult<Self> {
-        let epoll = Epoll::new().map_err(VringEpollError::EpollCreateFd)?;
+        let poller = Poll::new().map_err(VringPollError::PollerCreate)?;
         let exit_event_fd = backend.exit_event(thread_id);
+        let fd_set = Mutex::new(HashSet::new());
 
+        let registry = poller
+            .registry()
+            .try_clone()
+            .map_err(VringPollError::RegistryClone)?;
         let exit_event_fd = if let Some((consumer, notifier)) = exit_event_fd {
             let id = backend.num_queues();
-            epoll
-                .ctl(
-                    ControlOperation::Add,
-                    consumer.into_raw_fd(),
-                    EpollEvent::new(EventSet::IN, id as u64),
+
+            registry
+                .register(
+                    &mut SourceFd(&consumer.as_raw_fd()),
+                    Token(id),
+                    Interest::READABLE,
                 )
-                .map_err(VringEpollError::RegisterExitEvent)?;
+                .map_err(VringPollError::RegisterExitEvent)?;
+
+            fd_set.lock().unwrap().insert(consumer.into_raw_fd());
             Some(notifier)
         } else {
             None
         };
 
         Ok(VringEpollHandler {
-            epoll,
+            poller: Mutex::new(poller),
+            registry,
+            fd_set,
             backend,
             vrings,
             thread_id,
@@ -129,29 +180,37 @@ where
     ///
     /// If the event is triggered after this function has been called, the event will be silently
     /// dropped.
-    pub fn unregister_listener(&self, fd: RawFd, ev_type: EventSet, data: usize) -> Result<()> {
+    pub fn unregister_listener(&self, fd: RawFd, data: usize) -> Result<()> {
         // `data` range [0...num_queues] is reserved for queues and exit event.
         if data <= self.backend.num_queues() {
             Err(io::Error::from_raw_os_error(libc::EINVAL))
         } else {
-            self.unregister_event(fd, ev_type, data)
+            self.unregister_event(fd)
         }
     }
 
     pub(crate) fn register_event(&self, fd: RawFd, ev_type: EventSet, data: usize) -> Result<()> {
-        self.epoll.ctl(
-            ControlOperation::Add,
-            fd,
-            EpollEvent::new(ev_type, data as u64),
-        )
+        let mut fd_set = self.fd_set.lock().unwrap();
+        if fd_set.contains(&fd) {
+            return Err(io::Error::from_raw_os_error(libc::EEXIST));
+        }
+        self.registry
+            .register(&mut SourceFd(&fd), Token(data), ev_type.to_interest())
+            .map_err(std::io::Error::other)?;
+        fd_set.insert(fd);
+        Ok(())
     }
 
-    pub(crate) fn unregister_event(&self, fd: RawFd, ev_type: EventSet, data: usize) -> Result<()> {
-        self.epoll.ctl(
-            ControlOperation::Delete,
-            fd,
-            EpollEvent::new(ev_type, data as u64),
-        )
+    pub(crate) fn unregister_event(&self, fd: RawFd) -> Result<()> {
+        let mut fd_set = self.fd_set.lock().unwrap();
+        if !fd_set.contains(&fd) {
+            return Err(io::Error::from_raw_os_error(libc::ENOENT));
+        }
+        self.registry
+            .deregister(&mut SourceFd(&fd))
+            .map_err(|e| std::io::Error::other(format!("Failed to deregister fd {fd}: {e}")))?;
+        fd_set.remove(&fd);
+        Ok(())
     }
 
     /// Run the event poll loop to handle all pending events on registered fds.
@@ -160,41 +219,22 @@ where
     /// associated with the backend.
     pub(crate) fn run(&self) -> VringEpollResult<()> {
         const EPOLL_EVENTS_LEN: usize = 100;
-        let mut events = vec![EpollEvent::new(EventSet::empty(), 0); EPOLL_EVENTS_LEN];
-
-        'epoll: loop {
-            let num_events = match self.epoll.wait(-1, &mut events[..]) {
-                Ok(res) => res,
-                Err(e) => {
-                    if e.kind() == io::ErrorKind::Interrupted {
-                        // It's well defined from the epoll_wait() syscall
-                        // documentation that the epoll loop can be interrupted
-                        // before any of the requested events occurred or the
-                        // timeout expired. In both those cases, epoll_wait()
-                        // returns an error of type EINTR, but this should not
-                        // be considered as a regular error. Instead it is more
-                        // appropriate to retry, by calling into epoll_wait().
-                        continue;
-                    }
-                    return Err(VringEpollError::EpollWait(e));
-                }
-            };
-
-            for event in events.iter().take(num_events) {
-                let evset = match EventSet::from_bits(event.events) {
-                    Some(evset) => evset,
-                    None => {
-                        let evbits = event.events;
-                        println!("epoll: ignoring unknown event set: 0x{evbits:x}");
-                        continue;
-                    }
-                };
 
-                let ev_type = event.data();
+        let mut events = Events::with_capacity(EPOLL_EVENTS_LEN);
+        'poll: loop {
+            self.poller
+                .lock()
+                .unwrap()
+                .poll(&mut events, None)
+                .map_err(VringPollError::PollerWait)?;
 
-                // handle_event() returns true if an event is received from the exit event fd.
-                if self.handle_event(ev_type as usize, evset)? {
-                    break 'epoll;
+            for event in &events {
+                let token = event.token();
+
+                if let Some(evt_set) = event_to_event_set(event) {
+                    if self.handle_event(token.0, evt_set)? {
+                        break 'poll;
+                    }
                 }
             }
         }
@@ -211,7 +251,7 @@ where
             let vring = &self.vrings[device_event];
             let enabled = vring
                 .read_kick()
-                .map_err(VringEpollError::HandleEventReadKick)?;
+                .map_err(VringPollError::HandleEventReadKick)?;
 
             // If the vring is not enabled, it should not be processed.
             if !enabled {
@@ -221,15 +261,15 @@ where
 
         self.backend
             .handle_event(device_event, evset, &self.vrings, self.thread_id)
-            .map_err(VringEpollError::HandleEventBackendHandling)?;
+            .map_err(VringPollError::HandleEventBackendHandling)?;
 
         Ok(false)
     }
 }
 
 impl<T: VhostUserBackend> AsRawFd for VringEpollHandler<T> {
     fn as_raw_fd(&self) -> RawFd {
-        self.epoll.as_raw_fd()
+        self.poller.lock().unwrap().as_raw_fd()
     }
 }
 
@@ -254,29 +294,32 @@ mod tests {
 
         let (consumer, _notifier) = new_event_consumer_and_notifier(EventFlag::empty()).unwrap();
         handler
-            .register_listener(consumer.as_raw_fd(), EventSet::IN, 3)
+            .register_listener(consumer.as_raw_fd(), EventSet::Readable, 3)
             .unwrap();
         // Register an already registered fd.
         handler
-            .register_listener(consumer.as_raw_fd(), EventSet::IN, 3)
+            .register_listener(consumer.as_raw_fd(), EventSet::Readable, 3)
             .unwrap_err();
         // Register an invalid data.
         handler
-            .register_listener(consumer.as_raw_fd(), EventSet::IN, 1)
+            .register_listener(consumer.as_raw_fd(), EventSet::Readable, 1)
             .unwrap_err();
 
         handler
-            .unregister_listener(consumer.as_raw_fd(), EventSet::IN, 3)
+            .unregister_listener(consumer.as_raw_fd(), 3)
             .unwrap();
         // unregister an already unregistered fd.
         handler
-            .unregister_listener(consumer.as_raw_fd(), EventSet::IN, 3)
+            .unregister_listener(consumer.as_raw_fd(), 3)
             .unwrap_err();
         // unregister an invalid data.
         handler
-            .unregister_listener(consumer.as_raw_fd(), EventSet::IN, 1)
+            .unregister_listener(consumer.as_raw_fd(), 1)
             .unwrap_err();
         // Check we retrieve the correct file descriptor
-        assert_eq!(handler.as_raw_fd(), handler.epoll.as_raw_fd());
+        assert_eq!(
+            handler.as_raw_fd(),
+            handler.poller.lock().unwrap().as_raw_fd()
+        );
     }
 }