Update worker thread sender on x86

On x86, use the same sender as we would for macOS. Additionally, rather
than using an EventFd to determine when the thread work is done, use a
response sender/receiver like macOS.

Signed-off-by: Jake Correnti <[email protected]>

Author: jakecorrenti

src/devices/src/legacy/ioapic.rs

@@ -1,3 +1,4 @@
+use crossbeam_channel::{unbounded, Sender, Receiver};
 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,
@@ -96,14 +97,15 @@ pub struct IoApic {
     version: u8,
     irq_eoi: [i32; IOAPIC_NUM_PINS],
     irq_routes: Vec<kvm_irq_routing_entry>,
-    irq_sender: crossbeam_channel::Sender<(WorkerMessage, EventFd)>,
-    event_fd: EventFd,
+    irq_sender: Sender<WorkerMessage>,
+    response_sender: Sender<bool>,
+    response_receiver: Receiver<bool>,
 }
 
 impl IoApic {
     pub fn new(
         vm: &VmFd,
-        _irq_sender: crossbeam_channel::Sender<(WorkerMessage, EventFd)>,
+        _irq_sender: Sender<WorkerMessage>,
     ) -> Result<Self, Error> {
         let mut cap = kvm_enable_cap {
             cap: KVM_CAP_SPLIT_IRQCHIP,
@@ -112,6 +114,7 @@ impl IoApic {
         cap.args[0] = 24;
         vm.enable_cap(&cap)?;
 
+        let (response_sender, response_receiver) = unbounded();
         let mut ioapic = Self {
             id: 0,
             ioregsel: 0,
@@ -121,7 +124,8 @@ 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(),
+            response_sender,
+            response_receiver,
         };
 
         (0..IOAPIC_NUM_PINS).for_each(|i| ioapic.add_msi_route(i));
@@ -175,14 +179,6 @@ impl IoApic {
         }
     }
 
-    fn send_irq_worker_message(&self, msg: WorkerMessage) {
-        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;
@@ -249,7 +245,15 @@ impl IoApic {
             }
         }
 
-        self.send_irq_worker_message(WorkerMessage::GsiRoute(self.irq_routes.clone()));
+        self.irq_sender
+            .send(WorkerMessage::GsiRoute(
+                self.response_sender.clone(),
+                self.irq_routes.clone(),
+            ))
+            .unwrap();
+        if !self.response_receiver.recv().unwrap() {
+            error!("failed to update IO APIC routes");
+        }
     }
 
     fn service(&mut self) {
@@ -274,10 +278,37 @@ impl IoApic {
                     }
 
                     if info.trig_mode as u64 == IOAPIC_TRIGGER_EDGE {
-                        self.send_irq_worker_message(WorkerMessage::IrqLine(i as u32, true));
-                        self.send_irq_worker_message(WorkerMessage::IrqLine(i as u32, false));
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                self.response_sender.clone(),
+                                i as u32,
+                                true,
+                            ))
+                            .unwrap();
+                        if !self.response_receiver.recv().unwrap() {
+                            error!("failed to fully service edge triggered interrupt on irq {} with active state", i);
+                        }
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                self.response_sender.clone(),
+                                i as u32,
+                                false,
+                            ))
+                            .unwrap();
+                        if !self.response_receiver.recv().unwrap() {
+                            error!("failed to fully service edge triggered interrupt on irq {} with inactive state", i);
+                        }
                     } else {
-                        self.send_irq_worker_message(WorkerMessage::IrqLine(i as u32, true));
+                        self.irq_sender
+                            .send(WorkerMessage::IrqLine(
+                                self.response_sender.clone(),
+                                i as u32,
+                                true,
+                            ))
+                            .unwrap();
+                        if !self.response_receiver.recv().unwrap() {
+                            error!("failed to service level triggered interrupt on irq {} with active state", i);
+                        }
                     }
                 }
             }

src/utils/src/worker_message.rs

@@ -3,9 +3,9 @@ use crossbeam_channel::Sender;
 #[derive(Debug)]
 pub enum WorkerMessage {
     #[cfg(target_arch = "x86_64")]
-    GsiRoute(Vec<kvm_bindings::kvm_irq_routing_entry>),
+    GsiRoute(Sender<bool>, Vec<kvm_bindings::kvm_irq_routing_entry>),
     #[cfg(target_arch = "x86_64")]
-    IrqLine(u32, bool),
+    IrqLine(Sender<bool>, u32, bool),
     #[cfg(target_os = "macos")]
     GpuAddMapping(Sender<bool>, u64, u64, u64),
     #[cfg(target_os = "macos")]

src/vmm/src/builder.rs

@@ -508,8 +508,7 @@ pub fn build_microvm(
     vm_resources: &super::resources::VmResources,
     event_manager: &mut EventManager,
     _shutdown_efd: Option<EventFd>,
-    #[cfg(target_os = "macos")] _sender: Sender<WorkerMessage>,
-    #[cfg(target_arch = "x86_64")] _sender: Sender<(WorkerMessage, EventFd)>,
+    _sender: Sender<WorkerMessage>,
 ) -> std::result::Result<Arc<Mutex<Vmm>>, StartMicrovmError> {
     let payload = choose_payload(vm_resources)?;
 

src/vmm/src/lib.rs

@@ -383,37 +383,27 @@ impl Vmm {
 
     pub fn start_worker_thread(
         vmm: Arc<Mutex<Vmm>>,
-        #[cfg(target_os = "macos")] receiver: Receiver<WorkerMessage>,
-        #[cfg(not(target_os = "macos"))] receiver: Receiver<(WorkerMessage, EventFd)>,
+        receiver: Receiver<WorkerMessage>,
     ) -> io::Result<()> {
         std::thread::Builder::new()
             .name("vmm worker".into())
             .spawn(move || loop {
                 match receiver.recv() {
                     Err(e) => error!("error receiving message from vmm worker thread: {:?}", e),
-                    #[cfg(target_os = "macos")]
                     Ok(message) => vmm.lock().unwrap().match_worker_message(message),
-                    #[cfg(target_os = "linux")]
-                    Ok((message, evt_fd)) => {
-                        vmm.lock().unwrap().match_worker_message(message, evt_fd)
-                    }
                 }
             })?;
         Ok(())
     }
 
-    fn match_worker_message(
-        &self,
-        msg: WorkerMessage,
-        #[cfg(target_os = "linux")] evt_fd: EventFd,
-    ) {
+    fn match_worker_message(&self, msg: WorkerMessage) {
         match msg {
             #[cfg(target_os = "macos")]
             WorkerMessage::GpuAddMapping(s, h, g, l) => self.add_mapping(s, h, g, l),
             #[cfg(target_os = "macos")]
             WorkerMessage::GpuRemoveMapping(s, g, l) => self.remove_mapping(s, g, l),
             #[cfg(target_arch = "x86_64")]
-            WorkerMessage::GsiRoute(entries) => {
+            WorkerMessage::GsiRoute(sender, entries) => {
                 let mut irq_routing = utils::sized_vec::vec_with_array_field::<
                     kvm_bindings::kvm_irq_routing,
                     kvm_bindings::kvm_irq_routing_entry,
@@ -427,14 +417,15 @@ impl Vmm {
                     entries_slice.copy_from_slice(&entries);
                 }
 
-                self.vm.fd().set_gsi_routing(&irq_routing[0]).unwrap();
-
-                evt_fd.write(1).unwrap();
+                sender
+                    .send(self.vm.fd().set_gsi_routing(&irq_routing[0]).is_ok())
+                    .unwrap();
             }
             #[cfg(target_arch = "x86_64")]
-            WorkerMessage::IrqLine(irq, active) => {
-                self.vm.fd().set_irq_line(irq, active).unwrap();
-                evt_fd.write(1).unwrap();
+            WorkerMessage::IrqLine(sender, irq, active) => {
+                sender
+                    .send(self.vm.fd().set_irq_line(irq, active).is_ok())
+                    .unwrap();
             }
         }
     }