devices: Use Event for PCIe hotplug completion

This CL changes the notification method for hotplug completion to Event.
This is a pre-requisite for queueing the hotplug events in an event
driven model.

BUG=b:331529292
TEST=./tools/dev_container ./tools/presubmit

Change-Id: I44a44b388f99bee0fb7312cc4001785318f8fa59
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/5465234
Commit-Queue: Ningyuan Wang <ningyuan@google.com>
Reviewed-by: Keiichi Watanabe <keiichiw@chromium.org>
Reviewed-by: Daniel Verkamp <dverkamp@chromium.org>

5 files changed, 51 insertions, 46 deletions

diff --git a/devices/src/bus.rs b/devices/src/bus.rs
index 44a2c5f41..dd309a882 100644
--- a/devices/src/bus.rs
+++ b/devices/src/bus.rs
@@ -12,13 +12,13 @@ use std::collections::BTreeMap;
 use std::collections::BTreeSet;
 use std::fmt;
 use std::result;
-use std::sync::mpsc;
 use std::sync::Arc;
 
 use anyhow::anyhow;
 use anyhow::Context;
 use base::debug;
 use base::error;
+use base::Event;
 use base::SharedMemory;
 use remain::sorted;
 use serde::Deserialize;
@@ -211,15 +211,15 @@ pub enum HotPlugKey {
 /// Trait for devices that notify hotplug event into guest
 pub trait HotPlugBus: Send {
     /// Request hot plug event. Returns error if the request is not sent. Upon success, optionally
-    /// returns a notification receiver, where a notification is sent when the guest OS completes
-    /// the request (by sending PCI_EXP_SLTCTL_CCIE). Returns None if no such mechanism is provided.
+    /// returns an event, which is triggerred once when the guest OS completes the request (by
+    /// sending PCI_EXP_SLTCTL_CCIE). Returns None if no such mechanism is provided.
     /// * 'addr' - the guest pci address for hotplug in device
-    fn hot_plug(&mut self, addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>>;
+    fn hot_plug(&mut self, addr: PciAddress) -> anyhow::Result<Option<Event>>;
     /// Request hot unplug event. Returns error if the request is not sent. Upon success, optionally
-    /// returns a notification receiver, where a notification is sent when the guest OS completes
-    /// the request (by sending PCI_EXP_SLTCTL_CCIE). Returns None if no such mechanism is provided.
+    /// returns an event, which is triggerred once when the guest OS completes the request (by
+    /// sending PCI_EXP_SLTCTL_CCIE). Returns None if no such mechanism is provided.
     /// * 'addr' - the guest pci address for hotplug out device
-    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>>;
+    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<Event>>;
     /// Check whether the hotplug bus is available to add the new device
     ///
     /// - 'None': hotplug bus isn't match with host pci device
diff --git a/devices/src/pci/pcie/pcie_port.rs b/devices/src/pci/pcie/pcie_port.rs
index 8804f135a..ae13685c9 100644
--- a/devices/src/pci/pcie/pcie_port.rs
+++ b/devices/src/pci/pcie/pcie_port.rs
@@ -3,10 +3,10 @@
 // found in the LICENSE file.
 
 use std::str::FromStr;
-use std::sync::mpsc;
 use std::sync::Arc;
 
 use base::warn;
+use base::Event;
 use resources::Alloc;
 use resources::SystemAllocator;
 use sync::Mutex;
@@ -341,7 +341,7 @@ impl PciePort {
     }
 
     /// Sets a sender for notifying guest report command complete. Returns sender replaced.
-    pub fn set_cc_sender(&mut self, cc_sender: mpsc::Sender<()>) -> Option<mpsc::Sender<()>> {
+    pub fn set_cc_sender(&mut self, cc_sender: Event) -> Option<Event> {
         self.pcie_config.lock().cc_sender.replace(cc_sender)
     }
 
@@ -400,7 +400,7 @@ pub struct PcieConfig {
     root_cap: Arc<Mutex<PcieRootCap>>,
     port_type: PcieDevicePortType,
 
-    cc_sender: Option<mpsc::Sender<()>>,
+    cc_sender: Option<Event>,
     hp_interrupt_pending: bool,
     removed_downstream_valid: bool,
 
@@ -491,7 +491,7 @@ impl PcieConfig {
                 if old_control != value {
                     // send Command completed events
                     if let Some(sender) = self.cc_sender.take() {
-                        if let Err(e) = sender.send(()) {
+                        if let Err(e) = sender.signal() {
                             warn!("Failed to notify command complete for slot event: {:#}", &e);
                         }
                     }
diff --git a/devices/src/pci/pcie/pcie_rp.rs b/devices/src/pci/pcie/pcie_rp.rs
index 756b00b26..c7c599e10 100644
--- a/devices/src/pci/pcie/pcie_rp.rs
+++ b/devices/src/pci/pcie/pcie_rp.rs
@@ -3,11 +3,11 @@
 // found in the LICENSE file.
 
 use std::collections::BTreeMap;
-use std::sync::mpsc;
 
 use anyhow::bail;
 use anyhow::Context;
 use anyhow::Result;
+use base::Event;
 use vm_control::GpeNotify;
 use vm_control::PmeNotify;
 
@@ -88,7 +88,7 @@ impl PciePortVariant for PcieRootPort {
 }
 
 impl HotPlugBus for PcieRootPort {
-    fn hot_plug(&mut self, addr: PciAddress) -> Result<Option<mpsc::Receiver<()>>> {
+    fn hot_plug(&mut self, addr: PciAddress) -> Result<Option<Event>> {
         if self.pcie_port.is_cc_pending() {
             bail!("Hot plug fail: previous slot event is pending.");
         }
@@ -96,7 +96,8 @@ impl HotPlugBus for PcieRootPort {
             .get(&addr)
             .context("No downstream devices.")?;
 
-        let (cc_sender, cc_recvr) = mpsc::channel();
+        let cc_sender = Event::new()?;
+        let cc_recvr = cc_sender.try_clone()?;
         self.pcie_port.set_cc_sender(cc_sender);
         self.pcie_port
             .set_slot_status(PCIE_SLTSTA_PDS | PCIE_SLTSTA_ABP);
@@ -104,7 +105,7 @@ impl HotPlugBus for PcieRootPort {
         Ok(Some(cc_recvr))
     }
 
-    fn hot_unplug(&mut self, addr: PciAddress) -> Result<Option<mpsc::Receiver<()>>> {
+    fn hot_unplug(&mut self, addr: PciAddress) -> Result<Option<Event>> {
         if self.pcie_port.is_cc_pending() {
             bail!("Hot unplug fail: previous slot event is pending.");
         }
@@ -123,7 +124,8 @@ impl HotPlugBus for PcieRootPort {
             self.removed_downstream.push(*guest_pci_addr);
         }
 
-        let (cc_sender, cc_recvr) = mpsc::channel();
+        let cc_sender = Event::new()?;
+        let cc_recvr = cc_sender.try_clone()?;
         self.pcie_port.set_cc_sender(cc_sender);
         self.pcie_port.set_slot_status(PCIE_SLTSTA_ABP);
         self.pcie_port.trigger_hp_or_pme_interrupt();
diff --git a/devices/src/pci/pcie/pcie_switch.rs b/devices/src/pci/pcie/pcie_switch.rs
index c3090cba9..2ecbb0f25 100644
--- a/devices/src/pci/pcie/pcie_switch.rs
+++ b/devices/src/pci/pcie/pcie_switch.rs
@@ -4,10 +4,10 @@
 
 use std::collections::BTreeMap;
 use std::str::FromStr;
-use std::sync::mpsc;
 
 use anyhow::bail;
 use base::error;
+use base::Event;
 
 use crate::bus::HotPlugBus;
 use crate::bus::HotPlugKey;
@@ -85,12 +85,12 @@ impl PciePortVariant for PcieUpstreamPort {
 // hotplug out.
 impl HotPlugBus for PcieUpstreamPort {
     // Do nothing. We are not a real hotplug bus.
-    fn hot_plug(&mut self, _addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>> {
+    fn hot_plug(&mut self, _addr: PciAddress) -> anyhow::Result<Option<Event>> {
         bail!("hot plug not supported on upstream port.")
     }
 
     // Just remove the downstream device.
-    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>> {
+    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<Event>> {
         self.downstream_devices.remove(&addr);
         Ok(None)
     }
@@ -213,7 +213,7 @@ impl PciePortVariant for PcieDownstreamPort {
 }
 
 impl HotPlugBus for PcieDownstreamPort {
-    fn hot_plug(&mut self, addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>> {
+    fn hot_plug(&mut self, addr: PciAddress) -> anyhow::Result<Option<Event>> {
         if !self.pcie_port.hotplug_implemented() {
             bail!("hotplug not implemented.");
         }
@@ -226,7 +226,7 @@ impl HotPlugBus for PcieDownstreamPort {
         Ok(None)
     }
 
-    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<mpsc::Receiver<()>>> {
+    fn hot_unplug(&mut self, addr: PciAddress) -> anyhow::Result<Option<Event>> {
         if !self.pcie_port.hotplug_implemented() {
             bail!("hotplug not implemented.");
         }
diff --git a/src/crosvm/sys/linux/pci_hotplug_manager.rs b/src/crosvm/sys/linux/pci_hotplug_manager.rs
index c912563fc..6bd03e785 100644
--- a/src/crosvm/sys/linux/pci_hotplug_manager.rs
+++ b/src/crosvm/sys/linux/pci_hotplug_manager.rs
@@ -19,6 +19,8 @@ use anyhow::Error;
 use arch::RunnableLinuxVm;
 use arch::VcpuArch;
 use arch::VmArch;
+use base::Event;
+use base::EventWaitResult;
 use devices::HotPlugBus;
 use devices::HotPlugKey;
 use devices::IrqEventSource;
@@ -66,7 +68,7 @@ enum HotPlugAvailability {
     /// available now
     Now,
     /// available after notification is received.
-    After(mpsc::Receiver<()>),
+    After(Event),
 }
 
 /// PortStub contains a hotplug port and set of hotplug devices on it.
@@ -91,14 +93,11 @@ impl PortStub {
     }
 
     /// Sends hotplug signal after notification on the port.
-    fn send_hotplug_signal_after_notification(
-        &mut self,
-        notf_recvr: mpsc::Receiver<()>,
-    ) -> Result<()> {
+    fn send_hotplug_signal_after_notification(&mut self, notf_recvr: Event) -> Result<()> {
         let base_pci_address = PciAddress::new(0, self.downstream_bus.into(), 0, 0)?;
         let weak_port = Arc::downgrade(&self.port);
         thread::spawn(move || {
-            if let Err(e) = notf_recvr.recv_timeout(PCI_SLOT_TIMEOUT) {
+            if let Err(e) = notf_recvr.wait_timeout(PCI_SLOT_TIMEOUT) {
                 error!(
                     "failed to receive hot unplug command complete notification: {:#}",
                     &e
@@ -324,7 +323,7 @@ impl PciHotPlugManager {
     }
 
     /// Sends eject signal on the port, and removes downstream devices on the port.
-    fn remove_device_from_port(&self, bus: u8) -> Result<mpsc::Receiver<()>> {
+    fn remove_device_from_port(&self, bus: u8) -> Result<Event> {
         let port_stub = self
             .occupied_ports
             .get(&bus)
@@ -343,7 +342,7 @@ struct PortPool {
     /// map of empty ports that are available
     ports_available: BTreeMap<PciAddress, PortStub>,
     /// map of ports that will soon be available
-    ports_pending: BTreeMap<PciAddress, (mpsc::Receiver<()>, PortStub)>,
+    ports_pending: BTreeMap<PciAddress, (Event, PortStub)>,
 }
 
 impl PortPool {
@@ -357,19 +356,17 @@ impl PortPool {
 
     /// Insert a port_stub that is available immediately.
     fn insert(&mut self, port_stub: PortStub) -> Result<()> {
-        self.update_port_availability();
+        self.update_port_availability()
+            .context("Update port availability")?;
         let pci_addr = port_stub.pci_address;
         self.ports_available.insert(pci_addr, port_stub);
         Ok(())
     }
 
     /// Insert a port_stub that will be available after notification received. Returns immediately.
-    fn insert_after_notification(
-        &mut self,
-        port_stub: PortStub,
-        cc_recvr: mpsc::Receiver<()>,
-    ) -> Result<()> {
-        self.update_port_availability();
+    fn insert_after_notification(&mut self, port_stub: PortStub, cc_recvr: Event) -> Result<()> {
+        self.update_port_availability()
+            .context("Update port availability")?;
         self.ports_pending
             .insert(port_stub.pci_address, (cc_recvr, port_stub));
         Ok(())
@@ -378,7 +375,8 @@ impl PortPool {
     /// Pop the first available port in the order of PCI enumeration. If no port is available now,
     /// pop the first in the order of PCI enumeration.
     fn pop_first(&mut self) -> Result<(PortStub, HotPlugAvailability)> {
-        self.update_port_availability();
+        self.update_port_availability()
+            .context("Update port availability")?;
         if let Some((_, port_stub)) = self.ports_available.pop_first() {
             return Ok((port_stub, HotPlugAvailability::Now));
         }
@@ -391,17 +389,21 @@ impl PortPool {
     }
 
     /// Move pending ports to available ports if notification is received.
-    fn update_port_availability(&mut self) {
+    fn update_port_availability(&mut self) -> Result<()> {
         let mut new_ports_pending = BTreeMap::new();
         while let Some((pci_addr, (cc_recvr, port_stub))) = self.ports_pending.pop_first() {
-            if cc_recvr.try_recv().is_ok() {
-                let pci_addr = port_stub.pci_address;
-                self.ports_available.insert(pci_addr, port_stub);
-            } else {
-                new_ports_pending.insert(pci_addr, (cc_recvr, port_stub));
+            match cc_recvr.wait_timeout(Duration::ZERO)? {
+                EventWaitResult::Signaled => {
+                    let pci_addr = port_stub.pci_address;
+                    self.ports_available.insert(pci_addr, port_stub);
+                }
+                EventWaitResult::TimedOut => {
+                    new_ports_pending.insert(pci_addr, (cc_recvr, port_stub));
+                }
             }
         }
         self.ports_pending = new_ports_pending;
+        Ok(())
     }
 }
 
@@ -425,7 +427,7 @@ mod tests {
     fn port_pool_pop_before_completion() {
         let mut port_pool = PortPool::new();
         let mock_port = new_mock_port_stub(PciAddress::new(0, 1, 0, 0).unwrap());
-        let (_cc_sender, cc_recvr) = mpsc::channel();
+        let cc_recvr = Event::new().unwrap();
         port_pool
             .insert_after_notification(mock_port, cc_recvr)
             .unwrap();
@@ -437,11 +439,12 @@ mod tests {
     fn port_pool_pop_after_completion() {
         let mut port_pool = PortPool::new();
         let mock_port = new_mock_port_stub(PciAddress::new(0, 1, 0, 0).unwrap());
-        let (cc_sender, cc_recvr) = mpsc::channel();
+        let cc_recvr = Event::new().unwrap();
+        let cc_sender = cc_recvr.try_clone().unwrap();
         port_pool
             .insert_after_notification(mock_port, cc_recvr)
             .unwrap();
-        cc_sender.send(()).unwrap();
+        cc_sender.signal().unwrap();
         let (_port_stub, port_availability) = port_pool.pop_first().unwrap();
         assert!(matches!(port_availability, HotPlugAvailability::Now));
     }