Replace Eventfd with EventNotifier/EventConsumer

Eventfd is Linux-specific. To support more platforms, we replace it with
the EventNotifier/EventConsumer abstractions.
EventSender and EventReceiver are wrappers that encapsulate eventfd functionality
Use pipefd to replace eventfd in the test.

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

Author: uran0sH

vhost-user-backend/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Added
 ### Changed
+- [[#308](https://github.com/rust-vmm/vhost/pull/308)] Replace Eventfd with EventNotifier/EventConsumer.
+
 ### Deprecated
 ### Fixed
 

vhost-user-backend/src/backend.rs

@@ -30,7 +30,7 @@ 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::eventfd::EventFd;
+use vmm_sys_util::event::{EventConsumer, EventNotifier};
 
 use vhost::vhost_user::GpuBackend;
 
@@ -132,7 +132,8 @@ pub trait VhostUserBackend: Send + Sync {
     ///
     /// The returned `EventFd` will be monitored for IO events. When the
     /// returned EventFd is written to, the worker thread will exit.
-    fn exit_event(&self, _thread_index: usize) -> Option<EventFd> {
+    /// TODO: Refine this API to return only EventNotifier.
+    fn exit_event(&self, _thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
         None
     }
 
@@ -275,7 +276,7 @@ pub trait VhostUserBackendMut: Send + Sync {
     /// If an (`EventFd`, `token`) pair is returned, the returned `EventFd` will be monitored for IO
     /// events by using epoll with the specified `token`. When the returned EventFd is written to,
     /// the worker thread will exit.
-    fn exit_event(&self, _thread_index: usize) -> Option<EventFd> {
+    fn exit_event(&self, _thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
         None
     }
 
@@ -382,7 +383,7 @@ impl<T: VhostUserBackend> VhostUserBackend for Arc<T> {
         self.deref().queues_per_thread()
     }
 
-    fn exit_event(&self, thread_index: usize) -> Option<EventFd> {
+    fn exit_event(&self, thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
         self.deref().exit_event(thread_index)
     }
 
@@ -471,7 +472,7 @@ impl<T: VhostUserBackendMut> VhostUserBackend for Mutex<T> {
         self.lock().unwrap().queues_per_thread()
     }
 
-    fn exit_event(&self, thread_index: usize) -> Option<EventFd> {
+    fn exit_event(&self, thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
         self.lock().unwrap().exit_event(thread_index)
     }
 
@@ -563,7 +564,7 @@ impl<T: VhostUserBackendMut> VhostUserBackend for RwLock<T> {
         self.read().unwrap().queues_per_thread()
     }
 
-    fn exit_event(&self, thread_index: usize) -> Option<EventFd> {
+    fn exit_event(&self, thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
         self.read().unwrap().exit_event(thread_index)
     }
 
@@ -600,15 +601,18 @@ pub mod tests {
     use super::*;
     use crate::VringRwLock;
     use libc::EFD_NONBLOCK;
-    use std::sync::Mutex;
+    use std::{
+        os::fd::{AsRawFd, FromRawFd, IntoRawFd},
+        sync::Mutex,
+    };
     use uuid::Uuid;
     use vm_memory::{GuestAddress, GuestMemoryAtomic, GuestMemoryMmap};
 
     pub struct MockVhostBackend {
         events: u64,
         event_idx: bool,
         acked_features: u64,
-        exit_event_fds: Vec<EventFd>,
+        exit_event_fds: Vec<(EventConsumer, EventNotifier)>,
     }
 
     impl MockVhostBackend {
@@ -624,7 +628,21 @@ pub mod tests {
             // order to allow tests maximum flexibility in checking whether
             // signals arrived or not.
             backend.exit_event_fds = (0..backend.queues_per_thread().len())
-                .map(|_| EventFd::new(EFD_NONBLOCK).unwrap())
+                .map(|_| unsafe {
+                    let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
+                    let flags = libc::fcntl(reader.as_raw_fd(), libc::F_GETFL);
+                    let ret = libc::fcntl(reader.as_raw_fd(), libc::F_SETFL, flags | EFD_NONBLOCK);
+                    if ret < 0 {
+                        panic!(
+                            "Failed to set pipe to non-blocking: {}",
+                            std::io::Error::last_os_error()
+                        );
+                    }
+                    (
+                        EventConsumer::from_raw_fd(reader.into_raw_fd()),
+                        EventNotifier::from_raw_fd(writer.into_raw_fd()),
+                    )
+                })
                 .collect();
 
             backend
@@ -695,13 +713,13 @@ pub mod tests {
             vec![1, 1]
         }
 
-        fn exit_event(&self, thread_index: usize) -> Option<EventFd> {
-            Some(
-                self.exit_event_fds
-                    .get(thread_index)?
-                    .try_clone()
-                    .expect("Could not clone exit eventfd"),
-            )
+        fn exit_event(&self, thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
+            self.exit_event_fds.get(thread_index).map(|(s, r)| {
+                (
+                    s.try_clone().expect("Failed to clone EventConsumer"),
+                    r.try_clone().expect("Failed to clone EventNotifier"),
+                )
+            })
         }
 
         fn handle_event(

vhost-user-backend/src/event_loop.rs

@@ -9,7 +9,7 @@ use std::marker::PhantomData;
 use std::os::unix::io::{AsRawFd, RawFd};
 
 use vmm_sys_util::epoll::{ControlOperation, Epoll, EpollEvent, EventSet};
-use vmm_sys_util::eventfd::EventFd;
+use vmm_sys_util::event::EventNotifier;
 
 use super::backend::VhostUserBackend;
 use super::vring::VringT;
@@ -61,15 +61,15 @@ pub struct VringEpollHandler<T: VhostUserBackend> {
     backend: T,
     vrings: Vec<T::Vring>,
     thread_id: usize,
-    exit_event_fd: Option<EventFd>,
+    exit_event_fd: Option<EventNotifier>,
     phantom: PhantomData<T::Bitmap>,
 }
 
 impl<T: VhostUserBackend> VringEpollHandler<T> {
     /// Send `exit event` to break the event loop.
     pub fn send_exit_event(&self) {
         if let Some(eventfd) = self.exit_event_fd.as_ref() {
-            let _ = eventfd.write(1);
+            let _ = eventfd.notify();
         }
     }
 }
@@ -87,12 +87,12 @@ where
         let epoll = Epoll::new().map_err(VringEpollError::EpollCreateFd)?;
         let exit_event_fd = backend.exit_event(thread_id);
 
-        if let Some(exit_event_fd) = &exit_event_fd {
+        if let Some((consumer, _)) = &exit_event_fd {
             let id = backend.num_queues();
             epoll
                 .ctl(
                     ControlOperation::Add,
-                    exit_event_fd.as_raw_fd(),
+                    consumer.as_raw_fd(),
                     EpollEvent::new(EventSet::IN, id as u64),
                 )
                 .map_err(VringEpollError::RegisterExitEvent)?;
@@ -103,7 +103,7 @@ where
             backend,
             vrings,
             thread_id,
-            exit_event_fd,
+            exit_event_fd: exit_event_fd.map(|(_, notifier)| notifier),
             phantom: PhantomData,
         })
     }

vhost-user-backend/src/lib.rs

@@ -350,7 +350,7 @@ mod tests {
                 let fd = backend.exit_event(thread_id).unwrap();
                 // Reading from exit fd should fail since nothing was written yet
                 assert_eq!(
-                    fd.read().unwrap_err().raw_os_error().unwrap(),
+                    fd.0.consume().unwrap_err().raw_os_error().unwrap(),
                     EAGAIN,
                     "exit event should not have been raised yet!"
                 );
@@ -365,7 +365,7 @@ mod tests {
         let backend = backend.lock().unwrap();
         for thread_id in 0..backend.queues_per_thread().len() {
             let fd = backend.exit_event(thread_id).unwrap();
-            assert!(fd.read().is_ok(), "No exit event was raised!");
+            assert!(fd.0.consume().is_ok(), "No exit event was raised!");
         }
     }
 }

vhost-user-backend/src/vring.rs

@@ -15,7 +15,8 @@ use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuar
 
 use virtio_queue::{Error as VirtQueError, Queue, QueueT};
 use vm_memory::{GuestAddress, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryMmap};
-use vmm_sys_util::eventfd::EventFd;
+use vmm_sys_util::event::{EventConsumer, EventNotifier};
+// use vmm_sys_util::eventfd::EventFd;
 
 /// Trait for objects returned by `VringT::get_ref()`.
 pub trait VringStateGuard<'a, M: GuestAddressSpace> {
@@ -109,9 +110,9 @@ pub trait VringT<M: GuestAddressSpace>:
 /// object for single-threaded context.
 pub struct VringState<M: GuestAddressSpace = GuestMemoryAtomic<GuestMemoryMmap>> {
     queue: Queue,
-    kick: Option<EventFd>,
-    call: Option<EventFd>,
-    err: Option<EventFd>,
+    kick: Option<EventConsumer>,
+    call: Option<EventNotifier>,
+    err: Option<EventConsumer>,
     enabled: bool,
     mem: M,
 }
@@ -148,7 +149,7 @@ impl<M: GuestAddressSpace> VringState<M> {
     /// Notify the vhost-user frontend that used descriptors have been put into the used queue.
     pub fn signal_used_queue(&self) -> io::Result<()> {
         if let Some(call) = self.call.as_ref() {
-            call.write(1)
+            call.notify()
         } else {
             Ok(())
         }
@@ -227,7 +228,7 @@ impl<M: GuestAddressSpace> VringState<M> {
     }
 
     /// Get the `EventFd` for kick.
-    pub fn get_kick(&self) -> &Option<EventFd> {
+    pub fn get_kick(&self) -> &Option<EventConsumer> {
         &self.kick
     }
 
@@ -237,13 +238,13 @@ impl<M: GuestAddressSpace> VringState<M> {
         // EventFd requires that it has sole ownership of its fd. So does File, so this is safe.
         // Ideally, we'd have a generic way to refer to a uniquely-owned fd, such as that proposed
         // by Rust RFC #3128.
-        self.kick = file.map(|f| unsafe { EventFd::from_raw_fd(f.into_raw_fd()) });
+        self.kick = file.map(|f| unsafe { EventConsumer::from_raw_fd(f.into_raw_fd()) });
     }
 
     /// Read event from the kick `EventFd`.
     fn read_kick(&self) -> io::Result<bool> {
         if let Some(kick) = &self.kick {
-            kick.read()?;
+            kick.consume()?;
         }
 
         Ok(self.enabled)
@@ -252,18 +253,18 @@ impl<M: GuestAddressSpace> VringState<M> {
     /// Set `EventFd` for call.
     fn set_call(&mut self, file: Option<File>) {
         // SAFETY: see comment in set_kick()
-        self.call = file.map(|f| unsafe { EventFd::from_raw_fd(f.into_raw_fd()) });
+        self.call = file.map(|f| unsafe { EventNotifier::from_raw_fd(f.into_raw_fd()) });
     }
 
     /// Get the `EventFd` for call.
-    pub fn get_call(&self) -> &Option<EventFd> {
+    pub fn get_call(&self) -> &Option<EventNotifier> {
         &self.call
     }
 
     /// Set `EventFd` for err.
     fn set_err(&mut self, file: Option<File>) {
         // SAFETY: see comment in set_kick()
-        self.err = file.map(|f| unsafe { EventFd::from_raw_fd(f.into_raw_fd()) });
+        self.err = file.map(|f| unsafe { EventConsumer::from_raw_fd(f.into_raw_fd()) });
     }
 }
 

vhost-user-backend/tests/vhost-user-server.rs

@@ -1,5 +1,6 @@
 use std::fs::File;
 use std::io::Result;
+use std::os::fd::{FromRawFd, IntoRawFd};
 use std::os::unix::io::AsRawFd;
 use std::os::unix::net::UnixStream;
 use std::path::Path;
@@ -18,6 +19,7 @@ use vm_memory::{
     FileOffset, GuestAddress, GuestAddressSpace, GuestMemory, GuestMemoryAtomic, GuestMemoryMmap,
 };
 use vmm_sys_util::epoll::EventSet;
+use vmm_sys_util::event::{EventConsumer, EventNotifier};
 use vmm_sys_util::eventfd::EventFd;
 
 struct MockVhostBackend {
@@ -105,10 +107,15 @@ impl VhostUserBackendMut for MockVhostBackend {
         vec![1, 1]
     }
 
-    fn exit_event(&self, _thread_index: usize) -> Option<EventFd> {
-        let event_fd = EventFd::new(0).unwrap();
+    fn exit_event(&self, _thread_index: usize) -> Option<(EventConsumer, EventNotifier)> {
+        let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
 
-        Some(event_fd)
+        Some(unsafe {
+            (
+                EventConsumer::from_raw_fd(reader.into_raw_fd()),
+                EventNotifier::from_raw_fd(writer.into_raw_fd()),
+            )
+        })
     }
 
     fn handle_event(