vm-virtio: Make all virtio devices potentially multi-threaded

Although only the block and net virtio devices can actually be multi
threaded (for now), handling them as special cases makes the code more
complex.

Signed-off-by: Samuel Ortiz <sameo@linux.intel.com>

vm-virtio/src/block.rs

@@ -775,7 +775,7 @@ pub struct Block<T: DiskFile> {
     config_space: Vec<u8>,
     queue_evt: Option<EventFd>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     pause_evt: Option<EventFd>,
     paused: Arc<AtomicBool>,
 }
@@ -975,15 +975,18 @@ impl<T: 'static + DiskFile + Send> VirtioDevice for Block<T> {
             };
 
             let paused = self.paused.clone();
+            let mut epoll_threads = Vec::new();
             thread::Builder::new()
                 .name("virtio_blk".to_string())
                 .spawn(move || handler.run(queue_evt, paused))
-                .map(|thread| self.epoll_thread = Some(thread))
+                .map(|thread| epoll_threads.push(thread))
                 .map_err(|e| {
                     error!("failed to clone the virtio-blk epoll thread: {}", e);
                     ActivateError::BadActivate
                 })?;
 
+            self.epoll_thread = Some(epoll_threads);
+
             return Ok(());
         }
         Err(ActivateError::BadActivate)

vm-virtio/src/console.rs

@@ -351,7 +351,7 @@ pub struct Console {
     out: Arc<Mutex<Box<dyn io::Write + Send + Sync + 'static>>>,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -544,15 +544,18 @@ impl VirtioDevice for Console {
         };
 
         let paused = self.paused.clone();
+        let mut epoll_threads = Vec::new();
         thread::Builder::new()
             .name("virtio_console".to_string())
             .spawn(move || handler.run(paused))
-            .map(|thread| self.epoll_thread = Some(thread))
+            .map(|thread| epoll_threads.push(thread))
             .map_err(|e| {
                 error!("failed to clone the virtio-console epoll thread: {}", e);
                 ActivateError::BadActivate
             })?;
 
+        self.epoll_thread = Some(epoll_threads);
+
         Ok(())
     }
 

vm-virtio/src/device.rs

@@ -137,7 +137,9 @@ macro_rules! virtio_pausable_inner {
             );
             self.paused.store(false, Ordering::SeqCst);
             if let Some(epoll_thread) = &self.epoll_thread {
-                epoll_thread.thread().unpark();
+                for i in 0..epoll_thread.len() {
+                    epoll_thread[i].thread().unpark();
+                }
             }
 
             Ok(())

vm-virtio/src/iommu.rs

@@ -765,7 +765,7 @@ pub struct Iommu {
     ext_mapping: BTreeMap<u32, Arc<dyn ExternalDmaMapping>>,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -938,15 +938,18 @@ impl VirtioDevice for Iommu {
         };
 
         let paused = self.paused.clone();
+        let mut epoll_threads = Vec::new();
         thread::Builder::new()
             .name("virtio_iommu".to_string())
             .spawn(move || handler.run(paused))
-            .map(|thread| self.epoll_thread = Some(thread))
+            .map(|thread| epoll_threads.push(thread))
             .map_err(|e| {
                 error!("failed to clone the virtio-iommu epoll thread: {}", e);
                 ActivateError::BadActivate
             })?;
 
+        self.epoll_thread = Some(epoll_threads);
+
         Ok(())
     }
 

vm-virtio/src/pmem.rs

@@ -318,7 +318,7 @@ pub struct Pmem {
     config: VirtioPmemConfig,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -484,15 +484,18 @@ impl VirtioDevice for Pmem {
             };
 
             let paused = self.paused.clone();
+            let mut epoll_threads = Vec::new();
             thread::Builder::new()
                 .name("virtio_pmem".to_string())
                 .spawn(move || handler.run(paused))
-                .map(|thread| self.epoll_thread = Some(thread))
+                .map(|thread| epoll_threads.push(thread))
                 .map_err(|e| {
                     error!("failed to clone virtio-pmem epoll thread: {}", e);
                     ActivateError::BadActivate
                 })?;
 
+            self.epoll_thread = Some(epoll_threads);
+
             return Ok(());
         }
         Err(ActivateError::BadActivate)

vm-virtio/src/rng.rs

@@ -184,7 +184,7 @@ pub struct Rng {
     acked_features: u64,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -335,15 +335,18 @@ impl VirtioDevice for Rng {
             };
 
             let paused = self.paused.clone();
+            let mut epoll_threads = Vec::new();
             thread::Builder::new()
                 .name("virtio_rng".to_string())
                 .spawn(move || handler.run(paused))
-                .map(|thread| self.epoll_thread = Some(thread))
+                .map(|thread| epoll_threads.push(thread))
                 .map_err(|e| {
                     error!("failed to clone the virtio-rng epoll thread: {}", e);
                     ActivateError::BadActivate
                 })?;
 
+            self.epoll_thread = Some(epoll_threads);
+
             return Ok(());
         }
         Err(ActivateError::BadActivate)

vm-virtio/src/vhost_user/blk.rs

@@ -47,7 +47,7 @@ pub struct Blk {
     queue_sizes: Vec<u16>,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -274,15 +274,18 @@ impl VirtioDevice for Blk {
         });
 
         let paused = self.paused.clone();
+        let mut epoll_threads = Vec::new();
         thread::Builder::new()
             .name("vhost_user_blk".to_string())
             .spawn(move || handler.run(paused))
-            .map(|thread| self.epoll_thread = Some(thread))
+            .map(|thread| epoll_threads.push(thread))
             .map_err(|e| {
                 error!("failed to clone virtio epoll thread: {}", e);
                 ActivateError::BadActivate
             })?;
 
+        self.epoll_thread = Some(epoll_threads);
+
         Ok(())
     }
 

vm-virtio/src/vhost_user/fs.rs

@@ -162,7 +162,7 @@ pub struct Fs {
     slave_req_support: bool,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -420,15 +420,18 @@ impl VirtioDevice for Fs {
         });
 
         let paused = self.paused.clone();
+        let mut epoll_threads = Vec::new();
         thread::Builder::new()
             .name("virtio_fs".to_string())
             .spawn(move || handler.run(paused))
-            .map(|thread| self.epoll_thread = Some(thread))
+            .map(|thread| epoll_threads.push(thread))
             .map_err(|e| {
                 error!("failed to clone queue EventFd: {}", e);
                 ActivateError::BadActivate
             })?;
 
+        self.epoll_thread = Some(epoll_threads);
+
         Ok(())
     }
 

vm-virtio/src/vsock/device.rs

@@ -383,7 +383,7 @@ pub struct Vsock<B: VsockBackend> {
     acked_features: u64,
     queue_evts: Option<Vec<EventFd>>,
     interrupt_cb: Option<Arc<dyn VirtioInterrupt>>,
-    epoll_thread: Option<thread::JoinHandle<result::Result<(), DeviceError>>>,
+    epoll_thread: Option<Vec<thread::JoinHandle<result::Result<(), DeviceError>>>>,
     paused: Arc<AtomicBool>,
 }
 
@@ -554,15 +554,18 @@ where
         };
 
         let paused = self.paused.clone();
+        let mut epoll_threads = Vec::new();
         thread::Builder::new()
             .name("virtio_vsock".to_string())
             .spawn(move || handler.run(paused))
-            .map(|thread| self.epoll_thread = Some(thread))
+            .map(|thread| epoll_threads.push(thread))
             .map_err(|e| {
                 error!("failed to clone the vsock epoll thread: {}", e);
                 ActivateError::BadActivate
             })?;
 
+        self.epoll_thread = Some(epoll_threads);
+
         Ok(())
     }