virtio-devices: Custom EpollHelper::run/VirtioCommon:reset for fuzz

It provides fuzzer a reliable way to wait for a sequence of events
to complete for virtio-devices while not using a fixed timeout to
maintain the full speed of fuzzing.

Take virtio-block as an example, the 'queue event' with a valid
available queue setup can trigger a 'completion event'. This is a
meaningful virtio-block code path of processing guest inputs which is
our target for fuzzing virtio devices.

Signed-off-by: Bo Chen <chen.bo@intel.com>

virtio-devices/src/device.rs

@@ -280,6 +280,7 @@ impl VirtioCommon {
         Ok(())
     }
 
+    #[cfg(not(fuzzing))]
     pub fn reset(&mut self) -> Option<Arc<dyn VirtioInterrupt>> {
         // We first must resume the virtio thread if it was paused.
         if self.pause_evt.take().is_some() {
@@ -303,6 +304,20 @@ impl VirtioCommon {
         Some(self.interrupt_cb.take().unwrap())
     }
 
+    #[cfg(fuzzing)]
+    // Wait for the worker thread to finish and return
+    pub fn reset(&mut self) -> Option<Arc<dyn VirtioInterrupt>> {
+        if let Some(mut threads) = self.epoll_threads.take() {
+            for t in threads.drain(..) {
+                if let Err(e) = t.join() {
+                    error!("Error joining thread: {:?}", e);
+                }
+            }
+        }
+
+        None
+    }
+
     pub fn dup_eventfds(&self) -> (EventFd, EventFd) {
         (
             self.kill_evt.as_ref().unwrap().try_clone().unwrap(),

virtio-devices/src/epoll_helper.rs

@@ -148,6 +148,7 @@ impl EpollHelper {
         .map_err(EpollHelperError::Ctl)
     }
 
+    #[cfg(not(fuzzing))]
     pub fn run(
         &mut self,
         paused: Arc<AtomicBool>,
@@ -240,6 +241,76 @@ impl EpollHelper {
             }
         }
     }
+
+    #[cfg(fuzzing)]
+    // Require to have a 'queue_evt' being kicked before calling
+    // and return when no epoll events are active
+    pub fn run(
+        &mut self,
+        paused: Arc<AtomicBool>,
+        paused_sync: Arc<Barrier>,
+        handler: &mut dyn EpollHelperHandler,
+    ) -> std::result::Result<(), EpollHelperError> {
+        const EPOLL_EVENTS_LEN: usize = 100;
+        let mut events = vec![epoll::Event::new(epoll::Events::empty(), 0); EPOLL_EVENTS_LEN];
+
+        loop {
+            let num_events = match epoll::wait(self.epoll_file.as_raw_fd(), 0, &mut events[..]) {
+                Ok(res) => res,
+                Err(e) => {
+                    if e.kind() == std::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(EpollHelperError::Wait(e));
+                }
+            };
+
+            // Return when no epoll events are active
+            if num_events == 0 {
+                return Ok(());
+            }
+
+            for event in events.iter().take(num_events) {
+                let ev_type = event.data as u16;
+
+                match ev_type {
+                    EPOLL_HELPER_EVENT_KILL => {
+                        info!("KILL_EVENT received, stopping epoll loop");
+                        return Ok(());
+                    }
+                    EPOLL_HELPER_EVENT_PAUSE => {
+                        info!("PAUSE_EVENT received, pausing epoll loop");
+
+                        // Acknowledge the pause is effective by using the
+                        // paused_sync barrier.
+                        paused_sync.wait();
+
+                        // We loop here to handle spurious park() returns.
+                        // Until we have not resumed, the paused boolean will
+                        // be true.
+                        while paused.load(Ordering::SeqCst) {
+                            thread::park();
+                        }
+
+                        // Drain pause event after the device has been resumed.
+                        // This ensures the pause event has been seen by each
+                        // thread related to this virtio device.
+                        let _ = self.pause_evt.read();
+                    }
+                    _ => {
+                        handler.handle_event(self, event)?;
+                    }
+                }
+            }
+        }
+    }
 }
 
 impl AsRawFd for EpollHelper {