virtio-devices: net_ctrlq: Port to EpollHelper

Migrate virtio-net and vhost-user-net control queue to EpollHelper so
as to remove code duplication.

Signed-off-by: Sebastien Boeuf <sebastien.boeuf@intel.com>

virtio-devices/src/net.rs

@@ -203,7 +203,7 @@ pub struct Net {
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
     epoll_threads: Option<Vec<thread::JoinHandle<result::Result<(), EpollHelperError>>>>,
-    ctrl_queue_epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    ctrl_queue_epoll_thread: Option<thread::JoinHandle<()>>,
     paused: Arc<AtomicBool>,
     queue_size: Vec<u16>,
     counters: NetCounters,
@@ -426,10 +426,11 @@ impl VirtioDevice for Net {
                 thread::Builder::new()
                     .name("virtio_net".to_string())
                     .spawn(move || {
-                        SeccompFilter::apply(virtio_net_seccomp_filter)
+                        if let Err(e) = SeccompFilter::apply(virtio_net_seccomp_filter) {
-                            .map_err(DeviceError::ApplySeccompFilter)?;
+                            error!("Error applying seccomp filter: {:?}", e);
-
+                        } else if let Err(e) = ctrl_handler.run_ctrl(paused) {
-                        ctrl_handler.run_ctrl(paused)
+                            error!("Error running worker: {:?}", e);
+                        }
                     })
                     .map(|thread| self.ctrl_queue_epoll_thread = Some(thread))
                     .map_err(|e| {

virtio-devices/src/net_util.rs

@@ -2,15 +2,15 @@
 //
 // SPDX-License-Identifier: Apache-2.0 AND BSD-3-Clause
 
-use super::Error as DeviceError;
+use super::{
-use super::{DescriptorChain, DeviceEventT, Queue};
+    DescriptorChain, EpollHelper, EpollHelperError, EpollHelperHandler, Queue,
-use net_util::{register_listener, MacAddr};
+    EPOLL_HELPER_EVENT_LAST,
+};
+use net_util::MacAddr;
 use serde::ser::{Serialize, SerializeStruct, Serializer};
-use std::fs::File;
+use std::os::unix::io::{AsRawFd, RawFd};
-use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
+use std::sync::atomic::AtomicBool;
-use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
-use std::thread;
 use virtio_bindings::bindings::virtio_net::*;
 use vm_memory::{
     ByteValued, Bytes, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryError, GuestMemoryMmap,
@@ -21,16 +21,8 @@ type Result<T> = std::result::Result<T, Error>;
 
 const QUEUE_SIZE: usize = 256;
 
-// The device has been dropped.
+// Event available on the control queue.
-pub const KILL_EVENT: DeviceEventT = 3;
+const CTRL_QUEUE_EVENT: u16 = EPOLL_HELPER_EVENT_LAST + 1;
-// The device should be paused.
-pub const PAUSE_EVENT: DeviceEventT = 4;
-// Number of DeviceEventT events supported by this implementation.
-pub const NET_EVENTS_COUNT: usize = 5;
-// The device has been dropped.
-const CTRL_QUEUE_EVENT: DeviceEventT = 0;
-// Number of DeviceEventT events supported by this implementation.
-const CTRL_EVENT_COUNT: usize = 3;
 
 #[repr(C, packed)]
 #[derive(Copy, Clone, Debug, Default, Deserialize)]
@@ -182,93 +174,40 @@ pub struct NetCtrlEpollHandler {
 }
 
 impl NetCtrlEpollHandler {
-    pub fn run_ctrl(&mut self, paused: Arc<AtomicBool>) -> std::result::Result<(), DeviceError> {
+    pub fn run_ctrl(
-        // Create the epoll file descriptor
+        &mut self,
-        self.epoll_fd = epoll::create(true).map_err(DeviceError::EpollCreateFd)?;
+        paused: Arc<AtomicBool>,
-        // Use 'File' to enforce closing on 'epoll_fd'
+    ) -> std::result::Result<(), EpollHelperError> {
-        let epoll_file = unsafe { File::from_raw_fd(self.epoll_fd) };
+        let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?;
+        helper.add_event(self.ctrl_q.queue_evt.as_raw_fd(), CTRL_QUEUE_EVENT)?;
+        helper.run(paused, self)?;
 
-        register_listener(
+        Ok(())
-            epoll_file.as_raw_fd(),
+    }
-            self.ctrl_q.queue_evt.as_raw_fd(),
+}
-            epoll::Events::EPOLLIN,
-            u64::from(CTRL_QUEUE_EVENT),
-        )
-        .unwrap();
-        register_listener(
-            epoll_file.as_raw_fd(),
-            self.kill_evt.as_raw_fd(),
-            epoll::Events::EPOLLIN,
-            u64::from(KILL_EVENT),
-        )
-        .unwrap();
-        register_listener(
-            epoll_file.as_raw_fd(),
-            self.pause_evt.as_raw_fd(),
-            epoll::Events::EPOLLIN,
-            u64::from(PAUSE_EVENT),
-        )
-        .unwrap();
 
-        let mut events = vec![epoll::Event::new(epoll::Events::empty(), 0); CTRL_EVENT_COUNT];
+impl EpollHelperHandler for NetCtrlEpollHandler {
-
+    fn handle_event(&mut self, _helper: &mut EpollHelper, event: &epoll::Event) -> bool {
-        // Before jumping into the epoll loop, check if the device is expected
+        let ev_type = event.data as u16;
-        // to be in a paused state. This is helpful for the restore code path
+        match ev_type {
-        // as the device thread should not start processing anything before the
+            CTRL_QUEUE_EVENT => {
-        // device has been resumed.
+                let mem = self.mem.memory();
-        while paused.load(Ordering::SeqCst) {
+                if let Err(e) = self.ctrl_q.queue_evt.read() {
-            thread::park();
+                    error!("failed to get ctl queue event: {:?}", e);
-        }
+                    return true;
-
-        'epoll: loop {
-            let num_events = match epoll::wait(epoll_file.as_raw_fd(), -1, &mut events[..]) {
-                Ok(res) => res,
-                Err(e) => {
-                    if e.kind() == std::io::ErrorKind::Interrupted {
-                        continue;
-                    }
-                    return Err(DeviceError::EpollWait(e));
                 }
-            };
+                if let Err(e) = self.ctrl_q.process_cvq(&mem) {
-
+                    error!("failed to process ctrl queue: {:?}", e);
-            for event in events.iter().take(num_events) {
+                    return true;
-                let ev_type = event.data as u16;
-
-                match ev_type {
-                    CTRL_QUEUE_EVENT => {
-                        let mem = self.mem.memory();
-                        if let Err(e) = self.ctrl_q.queue_evt.read() {
-                            error!("failed to get ctl queue event: {:?}", e);
-                        }
-                        if let Err(e) = self.ctrl_q.process_cvq(&mem) {
-                            error!("failed to process ctrl queue: {:?}", e);
-                        }
-                    }
-                    KILL_EVENT => {
-                        break 'epoll;
-                    }
-                    PAUSE_EVENT => {
-                        debug!("PAUSE_EVENT received, pausing vhost-user epoll loop");
-                        // We loop here to handle spurious park() returns.
-                        // Until we have not resumed, the paused boolean will
-                        // be true.
-                        while paused.load(Ordering::SeqCst) {
-                            std::thread::park();
-                        }
-
-                        // Drain pause event after the device has been resumed.
-                        // This ensures the pause event has been seen by each
-                        // and every thread related to this virtio device.
-                        let _ = self.pause_evt.read();
-                    }
-                    _ => {
-                        error!("Unknown event for virtio-net");
-                    }
                 }
             }
+            _ => {
+                error!("Unknown event for virtio-net");
+                return true;
+            }
         }
 
-        Ok(())
+        false
     }
 }
 

virtio-devices/src/vhost_user/net.rs

@@ -4,7 +4,6 @@
 use super::super::net_util::{
     build_net_config_space, CtrlVirtio, NetCtrlEpollHandler, VirtioNetConfig,
 };
-use super::super::Error as CtrlError;
 use super::super::{
     ActivateError, ActivateResult, EpollHelperError, Queue, VirtioDevice, VirtioDeviceType,
 };
@@ -47,7 +46,7 @@ pub struct Net {
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
     epoll_threads: Option<Vec<thread::JoinHandle<result::Result<(), EpollHelperError>>>>,
-    ctrl_queue_epoll_thread: Option<thread::JoinHandle<result::Result<(), CtrlError>>>,
+    ctrl_queue_epoll_thread: Option<thread::JoinHandle<result::Result<(), EpollHelperError>>>,
     paused: Arc<AtomicBool>,
 }