Consolidate worker threads

src/devices/src/legacy/ioapic.rs

@@ -1,10 +1,12 @@
+use crossbeam_channel::unbounded;
 use kvm_bindings::{
     kvm_enable_cap, kvm_irq_routing, kvm_irq_routing_entry, kvm_irq_routing_entry__bindgen_ty_1,
     kvm_irq_routing_msi, KVM_CAP_SPLIT_IRQCHIP, KVM_IRQ_ROUTING_MSI,
 };
 use kvm_ioctls::{Error, VmFd};
 
 use utils::eventfd::EventFd;
+use utils::worker_message::WorkerMessage;
 
 use crate::bus::BusDevice;
 use crate::legacy::irqchip::IrqChipT;
@@ -67,12 +69,6 @@ const IOAPIC_TRIGGER_EDGE: u64 = 0;
 /// 7:0 Interrupt Vector (INTVEC) (RW)
 type RedirectionTableEntry = u64;
 
-#[derive(Debug)]
-pub enum IrqWorkerMessage {
-    GsiRoute(Vec<kvm_irq_routing_entry>),
-    IrqLine(u32, bool),
-}
-
 #[derive(Debug, Default)]
 pub struct IoApicEntryInfo {
     masked: u8,
@@ -101,14 +97,13 @@ pub struct IoApic {
     version: u8,
     irq_eoi: [i32; IOAPIC_NUM_PINS],
     irq_routes: Vec<kvm_irq_routing_entry>,
-    irq_sender: crossbeam_channel::Sender<(IrqWorkerMessage, EventFd)>,
-    event_fd: EventFd,
+    irq_sender: crossbeam_channel::Sender<WorkerMessage>,
 }
 
 impl IoApic {
     pub fn new(
         vm: &VmFd,
-        _irq_sender: crossbeam_channel::Sender<(IrqWorkerMessage, EventFd)>,
+        _irq_sender: crossbeam_channel::Sender<WorkerMessage>,
     ) -> Result<Self, Error> {
         let mut cap = kvm_enable_cap {
             cap: KVM_CAP_SPLIT_IRQCHIP,
@@ -126,7 +121,6 @@ impl IoApic {
             irq_eoi: [0; IOAPIC_NUM_PINS],
             irq_routes: Vec::with_capacity(IOAPIC_NUM_PINS),
             irq_sender: _irq_sender,
-            event_fd: EventFd::new(libc::EFD_SEMAPHORE).unwrap(),
         };
 
         (0..IOAPIC_NUM_PINS).for_each(|i| ioapic.add_msi_route(i));
@@ -180,14 +174,6 @@ impl IoApic {
         }
     }
 
-    fn send_irq_worker_message(&self, msg: IrqWorkerMessage) {
-        self.irq_sender
-            .send((msg, self.event_fd.try_clone().unwrap()))
-            .unwrap();
-
-        self.event_fd.read().unwrap();
-    }
-
     fn parse_entry(&self, entry: &RedirectionTableEntry) -> IoApicEntryInfo {
         let vector = (entry & IOAPIC_VECTOR_MASK) as u8;
         let dest_idx = ((entry >> IOAPIC_LVT_DEST_IDX_SHIFT) & 0xffff) as u16;
@@ -254,7 +240,16 @@ impl IoApic {
             }
         }
 
-        self.send_irq_worker_message(IrqWorkerMessage::GsiRoute(self.irq_routes.clone()));
+        let (response_sender, response_receiver) = unbounded();
+        self.irq_sender
+            .send(WorkerMessage::GsiRoute(
+                response_sender.clone(),
+                self.irq_routes.clone(),
+            ))
+            .unwrap();
+        if !response_receiver.recv().unwrap() {
+            error!("unable to set GSI Routes for IO APIC");
+        }
     }
 
     fn service(&mut self) {
@@ -278,11 +273,49 @@ impl IoApic {
                         continue;
                     }
 
+                    let (response_sender, response_receiver) = unbounded();
                     if info.trig_mode as u64 == IOAPIC_TRIGGER_EDGE {
-                        self.send_irq_worker_message(IrqWorkerMessage::IrqLine(i as u32, true));
-                        self.send_irq_worker_message(IrqWorkerMessage::IrqLine(i as u32, false));
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                response_sender.clone(),
+                                i as u32,
+                                true,
+                            ))
+                            .unwrap();
+                        if !response_receiver.recv().unwrap() {
+                            error!(
+                                "unable to set IRQ LINE for IRQ {} with active set to {}",
+                                i, true
+                            );
+                        }
+
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                response_sender.clone(),
+                                i as u32,
+                                false,
+                            ))
+                            .unwrap();
+                        if !response_receiver.recv().unwrap() {
+                            error!(
+                                "unable to set IRQ LINE for IRQ {} with active set to {}",
+                                i, false
+                            );
+                        }
                     } else {
-                        self.send_irq_worker_message(IrqWorkerMessage::IrqLine(i as u32, true));
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                response_sender.clone(),
+                                i as u32,
+                                true,
+                            ))
+                            .unwrap();
+                        if !response_receiver.recv().unwrap() {
+                            error!(
+                                "unable to set IRQ LINE for IRQ {} with active set to {}",
+                                i, true
+                            );
+                        }
                     }
                 }
             }

src/devices/src/legacy/mod.rs

@@ -42,7 +42,7 @@ pub use self::hvfgicv3::HvfGicV3;
 pub use self::i8042::Error as I8042DeviceError;
 pub use self::i8042::I8042Device;
 #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
-pub use self::ioapic::{IoApic, IrqWorkerMessage};
+pub use self::ioapic::IoApic;
 pub use self::irqchip::{IrqChip, IrqChipDevice, IrqChipT};
 #[cfg(all(target_os = "linux", target_arch = "aarch64"))]
 pub use self::kvmgicv3::KvmGicV3;

src/devices/src/virtio/fs/device.rs

@@ -6,9 +6,9 @@ use std::sync::atomic::{AtomicI32, AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::thread::JoinHandle;
 
-#[cfg(target_os = "macos")]
-use hvf::MemoryMapping;
 use utils::eventfd::{EventFd, EFD_NONBLOCK};
+#[cfg(target_os = "macos")]
+use utils::worker_message::WorkerMessage;
 use virtio_bindings::{virtio_config::VIRTIO_F_VERSION_1, virtio_ring::VIRTIO_RING_F_EVENT_IDX};
 use vm_memory::{ByteValued, GuestMemoryMmap};
 
@@ -56,7 +56,7 @@ pub struct Fs {
     worker_stopfd: EventFd,
     exit_code: Arc<AtomicI32>,
     #[cfg(target_os = "macos")]
-    map_sender: Option<Sender<MemoryMapping>>,
+    map_sender: Option<Sender<WorkerMessage>>,
 }
 
 impl Fs {
@@ -135,7 +135,7 @@ impl Fs {
     }
 
     #[cfg(target_os = "macos")]
-    pub fn set_map_sender(&mut self, map_sender: Sender<MemoryMapping>) {
+    pub fn set_map_sender(&mut self, map_sender: Sender<WorkerMessage>) {
         self.map_sender = Some(map_sender);
     }
 }

src/devices/src/virtio/fs/filesystem.rs

@@ -5,7 +5,7 @@
 #[cfg(target_os = "macos")]
 use crossbeam_channel::Sender;
 #[cfg(target_os = "macos")]
-use hvf::MemoryMapping;
+use utils::worker_message::WorkerMessage;
 
 use std::collections::BTreeMap;
 use std::convert::TryInto;
@@ -1134,7 +1134,7 @@ pub trait FileSystem {
         moffset: u64,
         host_shm_base: u64,
         shm_size: u64,
-        #[cfg(target_os = "macos")] map_sender: &Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: &Option<Sender<WorkerMessage>>,
     ) -> io::Result<()> {
         Err(io::Error::from_raw_os_error(libc::ENOSYS))
     }
@@ -1145,7 +1145,7 @@ pub trait FileSystem {
         requests: Vec<RemovemappingOne>,
         host_shm_base: u64,
         shm_size: u64,
-        #[cfg(target_os = "macos")] map_sender: &Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: &Option<Sender<WorkerMessage>>,
     ) -> io::Result<()> {
         Err(io::Error::from_raw_os_error(libc::ENOSYS))
     }

src/devices/src/virtio/fs/macos/passthrough.rs

@@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex, RwLock};
 use std::time::Duration;
 
 use crossbeam_channel::{unbounded, Sender};
-use hvf::MemoryMapping;
+use utils::worker_message::WorkerMessage;
 
 use crate::virtio::fs::filesystem::SecContext;
 
@@ -1994,7 +1994,7 @@ impl FileSystem for PassthroughFs {
         moffset: u64,
         guest_shm_base: u64,
         shm_size: u64,
-        map_sender: &Option<Sender<MemoryMapping>>,
+        map_sender: &Option<Sender<WorkerMessage>>,
     ) -> io::Result<()> {
         if map_sender.is_none() {
             return Err(linux_error(io::Error::from_raw_os_error(libc::ENOSYS)));
@@ -2043,7 +2043,7 @@ impl FileSystem for PassthroughFs {
         let sender = map_sender.as_ref().unwrap();
         let (reply_sender, reply_receiver) = unbounded();
         sender
-            .send(MemoryMapping::AddMapping(
+            .send(WorkerMessage::GpuAddMapping(
                 reply_sender,
                 host_addr as u64,
                 guest_addr,
@@ -2070,7 +2070,7 @@ impl FileSystem for PassthroughFs {
         requests: Vec<fuse::RemovemappingOne>,
         guest_shm_base: u64,
         shm_size: u64,
-        map_sender: &Option<Sender<MemoryMapping>>,
+        map_sender: &Option<Sender<WorkerMessage>>,
     ) -> io::Result<()> {
         if map_sender.is_none() {
             return Err(linux_error(io::Error::from_raw_os_error(libc::ENOSYS)));
@@ -2093,7 +2093,7 @@ impl FileSystem for PassthroughFs {
             let sender = map_sender.as_ref().unwrap();
             let (reply_sender, reply_receiver) = unbounded();
             sender
-                .send(MemoryMapping::RemoveMapping(
+                .send(WorkerMessage::GpuRemoveMapping(
                     reply_sender,
                     guest_addr,
                     req.len,

src/devices/src/virtio/fs/server.rs

@@ -5,7 +5,7 @@
 #[cfg(target_os = "macos")]
 use crossbeam_channel::Sender;
 #[cfg(target_os = "macos")]
-use hvf::MemoryMapping;
+use utils::worker_message::WorkerMessage;
 
 use std::convert::TryInto;
 use std::ffi::{CStr, CString};
@@ -85,7 +85,7 @@ impl<F: FileSystem + Sync> Server<F> {
         w: Writer,
         shm_region: &Option<VirtioShmRegion>,
         exit_code: &Arc<AtomicI32>,
-        #[cfg(target_os = "macos")] map_sender: &Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: &Option<Sender<WorkerMessage>>,
     ) -> Result<usize> {
         let in_header: InHeader = r.read_obj().map_err(Error::DecodeMessage)?;
 
@@ -1341,7 +1341,7 @@ impl<F: FileSystem + Sync> Server<F> {
         w: Writer,
         host_shm_base: u64,
         shm_size: u64,
-        #[cfg(target_os = "macos")] map_sender: &Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: &Option<Sender<WorkerMessage>>,
     ) -> Result<usize> {
         let SetupmappingIn {
             fh,
@@ -1376,7 +1376,7 @@ impl<F: FileSystem + Sync> Server<F> {
         w: Writer,
         host_shm_base: u64,
         shm_size: u64,
-        #[cfg(target_os = "macos")] map_sender: &Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: &Option<Sender<WorkerMessage>>,
     ) -> Result<usize> {
         let RemovemappingIn { count } = r.read_obj().map_err(Error::DecodeMessage)?;
 

src/devices/src/virtio/fs/worker.rs

@@ -1,7 +1,7 @@
 #[cfg(target_os = "macos")]
 use crossbeam_channel::Sender;
 #[cfg(target_os = "macos")]
-use hvf::MemoryMapping;
+use utils::worker_message::WorkerMessage;
 
 use std::os::fd::AsRawFd;
 use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
@@ -34,7 +34,7 @@ pub struct FsWorker {
     stop_fd: EventFd,
     exit_code: Arc<AtomicI32>,
     #[cfg(target_os = "macos")]
-    map_sender: Option<Sender<MemoryMapping>>,
+    map_sender: Option<Sender<WorkerMessage>>,
 }
 
 impl FsWorker {
@@ -51,7 +51,7 @@ impl FsWorker {
         passthrough_cfg: passthrough::Config,
         stop_fd: EventFd,
         exit_code: Arc<AtomicI32>,
-        #[cfg(target_os = "macos")] map_sender: Option<Sender<MemoryMapping>>,
+        #[cfg(target_os = "macos")] map_sender: Option<Sender<WorkerMessage>>,
     ) -> Self {
         Self {
             queues,

src/devices/src/virtio/gpu/device.rs

@@ -18,7 +18,7 @@ use super::worker::Worker;
 use crate::legacy::IrqChip;
 use crate::Error as DeviceError;
 #[cfg(target_os = "macos")]
-use hvf::MemoryMapping;
+use utils::worker_message::WorkerMessage;
 
 // Control queue.
 pub(crate) const CTL_INDEX: usize = 0;
@@ -49,15 +49,15 @@ pub struct Gpu {
     pub(crate) sender: Option<Sender<u64>>,
     virgl_flags: u32,
     #[cfg(target_os = "macos")]
-    map_sender: Sender<MemoryMapping>,
+    map_sender: Sender<WorkerMessage>,
     export_table: Option<ExportTable>,
 }
 
 impl Gpu {
     pub(crate) fn with_queues(
         queues: Vec<VirtQueue>,
         virgl_flags: u32,
-        #[cfg(target_os = "macos")] map_sender: Sender<MemoryMapping>,
+        #[cfg(target_os = "macos")] map_sender: Sender<WorkerMessage>,
     ) -> super::Result<Gpu> {
         let mut queue_events = Vec::new();
         for _ in 0..queues.len() {
@@ -92,7 +92,7 @@ impl Gpu {
 
     pub fn new(
         virgl_flags: u32,
-        #[cfg(target_os = "macos")] map_sender: Sender<MemoryMapping>,
+        #[cfg(target_os = "macos")] map_sender: Sender<WorkerMessage>,
     ) -> super::Result<Gpu> {
         let queues: Vec<VirtQueue> = defs::QUEUE_SIZES
             .iter()