devices: join worker threads in drop()

Make sure all devices join any threads they spawn before returning from
the drop() handler after signaling the exit event.

BUG=chromium:992494
TEST=crosvm exits without errors

Change-Id: I6bc91c32a08f568b041765044caa9aff6f7cf4a9
Signed-off-by: Daniel Verkamp <dverkamp@chromium.org>
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/crosvm/+/1802156
Reviewed-by: Stephen Barber <smbarber@chromium.org>
Tested-by: kokoro <noreply+kokoro@google.com>

11 files changed, 161 insertions, 31 deletions

diff --git a/devices/src/usb/xhci/xhci.rs b/devices/src/usb/xhci/xhci.rs
index 47ebe85..bd8e217 100644
--- a/devices/src/usb/xhci/xhci.rs
+++ b/devices/src/usb/xhci/xhci.rs
@@ -14,6 +14,7 @@ use crate::usb::host_backend::host_backend_device_provider::HostBackendDevicePro
 use crate::utils::{Error as UtilsError, EventLoop, FailHandle};
 use std::fmt::{self, Display};
 use std::sync::Arc;
+use std::thread;
 use sync::Mutex;
 use sys_util::{error, EventFd, GuestAddress, GuestMemory};
 
@@ -65,6 +66,8 @@ pub struct Xhci {
     interrupter: Arc<Mutex<Interrupter>>,
     command_ring_controller: Arc<CommandRingController>,
     device_slots: DeviceSlots,
+    event_loop: Arc<EventLoop>,
+    event_loop_join_handle: Option<thread::JoinHandle<()>>,
     // resample handler and device provider only lives on EventLoop to handle corresponding events.
     // By design, event loop only hold weak reference. We need to keep a strong reference here to
     // keep it alive.
@@ -84,7 +87,7 @@ impl Xhci {
         irq_resample_evt: EventFd,
         regs: XhciRegs,
     ) -> Result<Arc<Self>> {
-        let (event_loop, _join_handle) =
+        let (event_loop, join_handle) =
             EventLoop::start("xhci".to_string(), Some(fail_handle.clone()))
                 .map_err(Error::StartEventLoop)?;
         let interrupter = Arc::new(Mutex::new(Interrupter::new(mem.clone(), irq_evt, &regs)));
@@ -122,6 +125,8 @@ impl Xhci {
             command_ring_controller,
             device_slots,
             device_provider,
+            event_loop,
+            event_loop_join_handle: Some(join_handle),
         });
         Self::init_reg_callbacks(&xhci);
         Ok(xhci)
@@ -390,3 +395,12 @@ impl Xhci {
         Ok(())
     }
 }
+
+impl Drop for Xhci {
+    fn drop(&mut self) {
+        self.event_loop.stop();
+        if let Some(join_handle) = self.event_loop_join_handle.take() {
+            let _ = join_handle.join();
+        }
+    }
+}
diff --git a/devices/src/virtio/balloon.rs b/devices/src/virtio/balloon.rs
index 45deffb..8109395 100644
--- a/devices/src/virtio/balloon.rs
+++ b/devices/src/virtio/balloon.rs
@@ -235,6 +235,7 @@ pub struct Balloon {
     config: Arc<BalloonConfig>,
     features: u64,
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
 }
 
 impl Balloon {
@@ -247,6 +248,7 @@ impl Balloon {
                 actual_pages: AtomicUsize::new(0),
             }),
             kill_evt: None,
+            worker_thread: None,
             // TODO(dgreid) - Add stats queue feature.
             features: 1 << VIRTIO_BALLOON_F_MUST_TELL_HOST | 1 << VIRTIO_BALLOON_F_DEFLATE_ON_OOM,
         })
@@ -268,6 +270,10 @@ impl Drop for Balloon {
             // Ignore the result because there is nothing we can do with a failure.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -345,9 +351,15 @@ impl VirtioDevice for Balloon {
                 };
                 worker.run(queue_evts, kill_evt);
             });
-        if let Err(e) = worker_result {
-            error!("failed to spawn virtio_balloon worker: {}", e);
-            return;
+
+        match worker_result {
+            Err(e) => {
+                error!("failed to spawn virtio_balloon worker: {}", e);
+                return;
+            }
+            Ok(join_handle) => {
+                self.worker_thread = Some(join_handle);
+            }
         }
     }
 }
diff --git a/devices/src/virtio/block.rs b/devices/src/virtio/block.rs
index 3662bd9..5939652 100644
--- a/devices/src/virtio/block.rs
+++ b/devices/src/virtio/block.rs
@@ -470,6 +470,7 @@ impl Worker {
 /// Virtio device for exposing block level read/write operations on a host file.
 pub struct Block {
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     disk_image: Option<Box<dyn DiskFile>>,
     disk_size: Arc<Mutex<u64>>,
     avail_features: u64,
@@ -522,6 +523,7 @@ impl Block {
 
         Ok(Block {
             kill_evt: None,
+            worker_thread: None,
             disk_image: Some(disk_image),
             disk_size: Arc::new(Mutex::new(disk_size)),
             avail_features,
@@ -699,6 +701,10 @@ impl Drop for Block {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -780,9 +786,14 @@ impl VirtioDevice for Block {
                             worker.run(queue_evts.remove(0), kill_evt, control_socket);
                         });
 
-                if let Err(e) = worker_result {
-                    error!("failed to spawn virtio_blk worker: {}", e);
-                    return;
+                match worker_result {
+                    Err(e) => {
+                        error!("failed to spawn virtio_blk worker: {}", e);
+                        return;
+                    }
+                    Ok(join_handle) => {
+                        self.worker_thread = Some(join_handle);
+                    }
                 }
             }
         }
diff --git a/devices/src/virtio/gpu/mod.rs b/devices/src/virtio/gpu/mod.rs
index 49787fb..48ee6cb 100644
--- a/devices/src/virtio/gpu/mod.rs
+++ b/devices/src/virtio/gpu/mod.rs
@@ -678,6 +678,7 @@ pub struct Gpu {
     gpu_device_socket: Option<VmMemoryControlRequestSocket>,
     resource_bridges: Vec<ResourceResponseSocket>,
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     num_scanouts: NonZeroU8,
     display_backends: Vec<DisplayBackend>,
 }
@@ -696,6 +697,7 @@ impl Gpu {
             gpu_device_socket,
             resource_bridges,
             kill_evt: None,
+            worker_thread: None,
             num_scanouts,
             display_backends,
         }
@@ -721,6 +723,10 @@ impl Drop for Gpu {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -837,9 +843,14 @@ impl VirtioDevice for Gpu {
                         .run()
                     });
 
-            if let Err(e) = worker_result {
-                error!("failed to spawn virtio_gpu worker: {}", e);
-                return;
+            match worker_result {
+                Err(e) => {
+                    error!("failed to spawn virtio_gpu worker: {}", e);
+                    return;
+                }
+                Ok(join_handle) => {
+                    self.worker_thread = Some(join_handle);
+                }
             }
         }
     }
diff --git a/devices/src/virtio/input/mod.rs b/devices/src/virtio/input/mod.rs
index 29152a2..5cc68e0 100644
--- a/devices/src/virtio/input/mod.rs
+++ b/devices/src/virtio/input/mod.rs
@@ -610,6 +610,7 @@ impl<T: EventSource> Worker<T> {
 
 pub struct Input<T: EventSource> {
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     config: VirtioInputConfig,
     source: Option<T>,
 }
@@ -620,6 +621,10 @@ impl<T: EventSource> Drop for Input<T> {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -695,9 +700,14 @@ where
                     worker.run(event_queue_evt_fd, status_queue_evt_fd, kill_evt);
                 });
 
-            if let Err(e) = worker_result {
-                error!("failed to spawn virtio_input worker: {}", e);
-                return;
+            match worker_result {
+                Err(e) => {
+                    error!("failed to spawn virtio_input worker: {}", e);
+                    return;
+                }
+                Ok(join_handle) => {
+                    self.worker_thread = Some(join_handle);
+                }
             }
         } else {
             error!("tried to activate device without a source for events");
@@ -713,6 +723,7 @@ where
 {
     Ok(Input {
         kill_evt: None,
+        worker_thread: None,
         config: VirtioInputConfig::from_evdev(&source)?,
         source: Some(EvdevEventSource::new(source)),
     })
@@ -729,6 +740,7 @@ where
 {
     Ok(Input {
         kill_evt: None,
+        worker_thread: None,
         config: defaults::new_single_touch_config(width, height),
         source: Some(SocketEventSource::new(source)),
     })
@@ -742,6 +754,7 @@ where
 {
     Ok(Input {
         kill_evt: None,
+        worker_thread: None,
         config: defaults::new_trackpad_config(width, height),
         source: Some(SocketEventSource::new(source)),
     })
@@ -754,6 +767,7 @@ where
 {
     Ok(Input {
         kill_evt: None,
+        worker_thread: None,
         config: defaults::new_mouse_config(),
         source: Some(SocketEventSource::new(source)),
     })
@@ -766,6 +780,7 @@ where
 {
     Ok(Input {
         kill_evt: None,
+        worker_thread: None,
         config: defaults::new_keyboard_config(),
         source: Some(SocketEventSource::new(source)),
     })
diff --git a/devices/src/virtio/net.rs b/devices/src/virtio/net.rs
index 4875178..2a4784c 100644
--- a/devices/src/virtio/net.rs
+++ b/devices/src/virtio/net.rs
@@ -270,6 +270,7 @@ where
 pub struct Net<T: TapT> {
     workers_kill_evt: Option<EventFd>,
     kill_evt: EventFd,
+    worker_thread: Option<thread::JoinHandle<()>>,
     tap: Option<T>,
     avail_features: u64,
     acked_features: u64,
@@ -317,6 +318,7 @@ where
         Ok(Net {
             workers_kill_evt: Some(kill_evt.try_clone().map_err(NetError::CloneKillEventFd)?),
             kill_evt,
+            worker_thread: None,
             tap: Some(tap),
             avail_features,
             acked_features: 0u64,
@@ -376,6 +378,10 @@ where
             // Ignore the result because there is nothing we can do about it.
             let _ = self.kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -468,9 +474,14 @@ where
                             }
                         });
 
-                if let Err(e) = worker_result {
-                    error!("failed to spawn virtio_net worker: {}", e);
-                    return;
+                match worker_result {
+                    Err(e) => {
+                        error!("failed to spawn virtio_net worker: {}", e);
+                        return;
+                    }
+                    Ok(join_handle) => {
+                        self.worker_thread = Some(join_handle);
+                    }
                 }
             }
         }
diff --git a/devices/src/virtio/pmem.rs b/devices/src/virtio/pmem.rs
index b9e42cc..26ff0bf 100644
--- a/devices/src/virtio/pmem.rs
+++ b/devices/src/virtio/pmem.rs
@@ -246,6 +246,7 @@ impl Worker {
 
 pub struct Pmem {
     kill_event: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     disk_image: Option<File>,
     mapping_address: GuestAddress,
     mapping_size: u64,
@@ -259,6 +260,7 @@ impl Pmem {
     ) -> SysResult<Pmem> {
         Ok(Pmem {
             kill_event: None,
+            worker_thread: None,
             disk_image: Some(disk_image),
             mapping_address,
             mapping_size,
@@ -272,6 +274,10 @@ impl Drop for Pmem {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -344,9 +350,15 @@ impl VirtioDevice for Pmem {
                     };
                     worker.run(queue_event, kill_event);
                 });
-            if let Err(e) = worker_result {
-                error!("failed to spawn virtio_pmem worker: {}", e);
-                return;
+
+            match worker_result {
+                Err(e) => {
+                    error!("failed to spawn virtio_pmem worker: {}", e);
+                    return;
+                }
+                Ok(join_handle) => {
+                    self.worker_thread = Some(join_handle);
+                }
             }
         }
     }
diff --git a/devices/src/virtio/rng.rs b/devices/src/virtio/rng.rs
index 21d5c28..1ad683a 100644
--- a/devices/src/virtio/rng.rs
+++ b/devices/src/virtio/rng.rs
@@ -135,6 +135,7 @@ impl Worker {
 /// Virtio device for exposing entropy to the guest OS through virtio.
 pub struct Rng {
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     random_file: Option<File>,
 }
 
@@ -144,6 +145,7 @@ impl Rng {
         let random_file = File::open("/dev/urandom").map_err(RngError::AccessingRandomDev)?;
         Ok(Rng {
             kill_evt: None,
+            worker_thread: None,
             random_file: Some(random_file),
         })
     }
@@ -155,6 +157,10 @@ impl Drop for Rng {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -217,9 +223,14 @@ impl VirtioDevice for Rng {
                         worker.run(queue_evts.remove(0), kill_evt);
                     });
 
-            if let Err(e) = worker_result {
-                error!("failed to spawn virtio_rng worker: {}", e);
-                return;
+            match worker_result {
+                Err(e) => {
+                    error!("failed to spawn virtio_rng worker: {}", e);
+                    return;
+                }
+                Ok(join_handle) => {
+                    self.worker_thread = Some(join_handle);
+                }
             }
         }
     }
diff --git a/devices/src/virtio/tpm.rs b/devices/src/virtio/tpm.rs
index e8c50f2..e7088ed 100644
--- a/devices/src/virtio/tpm.rs
+++ b/devices/src/virtio/tpm.rs
@@ -190,6 +190,7 @@ impl Worker {
 pub struct Tpm {
     storage: PathBuf,
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
 }
 
 impl Tpm {
@@ -197,6 +198,7 @@ impl Tpm {
         Tpm {
             storage,
             kill_evt: None,
+            worker_thread: None,
         }
     }
 }
@@ -206,6 +208,10 @@ impl Drop for Tpm {
         if let Some(kill_evt) = self.kill_evt.take() {
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -271,9 +277,14 @@ impl VirtioDevice for Tpm {
             .name("virtio_tpm".to_string())
             .spawn(|| worker.run());
 
-        if let Err(e) = worker_result {
-            error!("vtpm failed to spawn virtio_tpm worker: {}", e);
-            return;
+        match worker_result {
+            Err(e) => {
+                error!("vtpm failed to spawn virtio_tpm worker: {}", e);
+                return;
+            }
+            Ok(join_handle) => {
+                self.worker_thread = Some(join_handle);
+            }
         }
     }
 }
diff --git a/devices/src/virtio/vhost/net.rs b/devices/src/virtio/vhost/net.rs
index a7eaf18..3f66b27 100644
--- a/devices/src/virtio/vhost/net.rs
+++ b/devices/src/virtio/vhost/net.rs
@@ -27,6 +27,7 @@ const QUEUE_SIZES: &[u16] = &[QUEUE_SIZE; NUM_QUEUES];
 pub struct Net<T: TapT, U: VhostNetT<T>> {
     workers_kill_evt: Option<EventFd>,
     kill_evt: EventFd,
+    worker_thread: Option<thread::JoinHandle<()>>,
     tap: Option<T>,
     vhost_net_handle: Option<U>,
     vhost_interrupt: Option<EventFd>,
@@ -84,6 +85,7 @@ where
         Ok(Net {
             workers_kill_evt: Some(kill_evt.try_clone().map_err(Error::CloneKillEventFd)?),
             kill_evt,
+            worker_thread: None,
             tap: Some(tap),
             vhost_net_handle: Some(vhost_net_handle),
             vhost_interrupt: Some(EventFd::new().map_err(Error::VhostIrqCreate)?),
@@ -104,6 +106,10 @@ where
             // Ignore the result because there is nothing we can do about it.
             let _ = self.kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -206,9 +212,14 @@ where
                                 }
                             });
 
-                        if let Err(e) = worker_result {
-                            error!("failed to spawn vhost_net worker: {}", e);
-                            return;
+                        match worker_result {
+                            Err(e) => {
+                                error!("failed to spawn vhost_net worker: {}", e);
+                                return;
+                            }
+                            Ok(join_handle) => {
+                                self.worker_thread = Some(join_handle);
+                            }
                         }
                     }
                 }
diff --git a/devices/src/virtio/wl.rs b/devices/src/virtio/wl.rs
index e6d4f40..a4988cb 100644
--- a/devices/src/virtio/wl.rs
+++ b/devices/src/virtio/wl.rs
@@ -1669,6 +1669,7 @@ impl Worker {
 
 pub struct Wl {
     kill_evt: Option<EventFd>,
+    worker_thread: Option<thread::JoinHandle<()>>,
     wayland_path: PathBuf,
     vm_socket: Option<VmMemoryControlRequestSocket>,
     resource_bridge: Option<ResourceRequestSocket>,
@@ -1683,6 +1684,7 @@ impl Wl {
     ) -> Result<Wl> {
         Ok(Wl {
             kill_evt: None,
+            worker_thread: None,
             wayland_path: wayland_path.as_ref().to_owned(),
             vm_socket: Some(vm_socket),
             resource_bridge,
@@ -1697,6 +1699,10 @@ impl Drop for Wl {
             // Ignore the result because there is nothing we can do about it.
             let _ = kill_evt.write(1);
         }
+
+        if let Some(worker_thread) = self.worker_thread.take() {
+            let _ = worker_thread.join();
+        }
     }
 }
 
@@ -1777,9 +1783,14 @@ impl VirtioDevice for Wl {
                         .run(queue_evts, kill_evt);
                     });
 
-            if let Err(e) = worker_result {
-                error!("failed to spawn virtio_wl worker: {}", e);
-                return;
+            match worker_result {
+                Err(e) => {
+                    error!("failed to spawn virtio_wl worker: {}", e);
+                    return;
+                }
+                Ok(join_handle) => {
+                    self.worker_thread = Some(join_handle);
+                }
             }
         }
     }