feat(Source Device): add async implementation

Author: ShadowApex

src/dbus/interface/source/iio_imu.rs

@@ -21,7 +21,7 @@ impl SourceIioImuInterface {
     pub async fn listen_on_dbus(
         conn: Connection,
         device: UdevDevice,
-    ) -> Result<(), Box<dyn Error>> {
+    ) -> Result<(), Box<dyn Error + Send + Sync>> {
         let iface = SourceIioImuInterface::new(device);
         let Ok(id) = iface.id() else {
             return Ok(());

src/drivers/iio_imu/driver.rs

@@ -1,4 +1,9 @@
-use std::{collections::HashMap, error::Error, time::Duration};
+use std::{
+    collections::HashMap,
+    error::Error,
+    sync::{Arc, Mutex},
+    time::Duration,
+};
 
 use industrial_io::{Channel, ChannelType, Device, Direction};
 
@@ -12,9 +17,9 @@ use super::{
 /// Driver for reading IIO IMU data
 pub struct Driver {
     mount_matrix: MountMatrix,
-    accel: HashMap<String, Channel>,
+    accel: HashMap<String, Arc<Mutex<Channel>>>,
     accel_info: HashMap<String, AxisInfo>,
-    gyro: HashMap<String, Channel>,
+    gyro: HashMap<String, Arc<Mutex<Channel>>>,
     gyro_info: HashMap<String, AxisInfo>,
     pub sample_delay: Duration,
 }
@@ -116,7 +121,7 @@ impl Driver {
             let Some(info) = self.accel_info.get(id) else {
                 continue;
             };
-            let data = channel.attr_read_int("raw")?;
+            let data = channel.lock().unwrap().attr_read_int("raw")?;
 
             // processed_value = (raw + offset) * scale
             let value = (data + info.offset) as f64 * info.scale;
@@ -144,7 +149,7 @@ impl Driver {
             let Some(info) = self.gyro_info.get(id) else {
                 continue;
             };
-            let data = channel.attr_read_int("raw")?;
+            let data = channel.lock().unwrap().attr_read_int("raw")?;
 
             // processed_value = (raw + offset) * scale
             let value = (data + info.offset) as f64 * info.scale;
@@ -194,7 +199,10 @@ impl Driver {
 fn get_channels_with_type(
     device: &Device,
     channel_type: ChannelType,
-) -> (HashMap<String, Channel>, HashMap<String, AxisInfo>) {
+) -> (
+    HashMap<String, Arc<Mutex<Channel>>>,
+    HashMap<String, AxisInfo>,
+) {
     let mut channels = HashMap::new();
     let mut channel_info = HashMap::new();
     device
@@ -276,8 +284,11 @@ fn get_channels_with_type(
                 scales_avail,
             };
             channel_info.insert(id.clone(), info);
-            channels.insert(id, channel);
+            channels.insert(id, Arc::new(Mutex::new(channel)));
         });
 
     (channels, channel_info)
 }
+
+// A mutex is used to access channels
+unsafe impl Send for Driver {}

src/input/composite_device/mod.rs

@@ -22,7 +22,7 @@ use zbus::Connection;
 
 use crate::{
     config::{
-        capability_map::CapabilityMapConfig, path::get_profiles_path, CompositeDeviceConfig,
+        self, capability_map::CapabilityMapConfig, path::get_profiles_path, CompositeDeviceConfig,
         DeviceProfile, ProfileMapping,
     },
     dbus::interface::{
@@ -40,7 +40,7 @@ use crate::{
             evdev::EventDevice, hidraw::HidRawDevice, iio::IioDevice, led::LedDevice, SourceDevice,
         },
     },
-    udev::{device::UdevDevice, hide_device, unhide_device},
+    udev::device::UdevDevice,
 };
 
 use self::{client::CompositeDeviceClient, command::CompositeCommand};
@@ -156,7 +156,7 @@ pub struct CompositeDevice {
 }
 
 impl CompositeDevice {
-    pub fn new(
+    pub async fn new(
         conn: Connection,
         manager: mpsc::Sender<ManagerCommand>,
         config: CompositeDeviceConfig,
@@ -240,7 +240,7 @@ impl CompositeDevice {
             }
         }
 
-        if let Err(e) = device.add_source_device(device_info) {
+        if let Err(e) = device.add_source_device(device_info).await {
             return Err(e.to_string().into());
         }
 
@@ -277,9 +277,6 @@ impl CompositeDevice {
 
         let dbus_path = self.dbus_path.clone();
 
-        // Start all source devices
-        self.run_source_devices().await?;
-
         // Set persist value from config if set, used to determine
         // if CompositeDevice self-closes after all SourceDevices have
         // been removed.
@@ -537,18 +534,6 @@ impl CompositeDevice {
         log::debug!("Stopping target devices");
         self.targets.stop().await;
 
-        // Unhide all source devices
-        for source_path in self.source_device_paths.clone() {
-            if source_path.starts_with("/sys/bus/iio/devices") {
-                log::debug!("Skipping unhiding IIO device: {source_path}");
-                continue;
-            }
-            log::debug!("Un-hiding device: {}", source_path);
-            if let Err(e) = unhide_device(source_path.clone()).await {
-                log::debug!("Unable to unhide device {source_path}: {:?}", e);
-            }
-        }
-
         // Send stop command to all source devices
         for (path, source) in &self.source_devices {
             log::debug!("Stopping source device: {path}");
@@ -590,54 +575,6 @@ impl CompositeDevice {
         self.source_device_paths.clone()
     }
 
-    /// Start and run the source devices that this composite device will
-    /// consume.
-    async fn run_source_devices(&mut self) -> Result<(), Box<dyn Error>> {
-        // Hide the device if specified
-        for source_path in self.source_devices_to_hide.drain(..) {
-            log::debug!("Hiding device: {}", source_path);
-            if let Err(e) = hide_device(source_path.as_str()).await {
-                log::warn!("Failed to hide device '{source_path}': {e:?}");
-            }
-            log::debug!("Finished hiding device: {source_path}");
-        }
-
-        log::debug!("Starting new source devices");
-        // Start listening for events from all source devices
-        let sources = self.source_devices_discovered.drain(..);
-        for source_device in sources {
-            let device_id = source_device.get_id();
-            // If the source device is blocked, don't bother running it
-            if self.source_devices_blocked.contains(&device_id) {
-                log::debug!("Source device '{device_id}' blocked. Skipping running.");
-                continue;
-            }
-
-            let source_tx = source_device.client();
-            self.source_devices.insert(device_id.clone(), source_tx);
-            let tx = self.tx.clone();
-
-            // Add the IIO IMU Dbus interface. We do this here because it needs the source
-            // device transmitter and this is the only place we can refrence it at the moment.
-            let device = source_device.get_device_ref().clone();
-            if let SourceDevice::Iio(_) = source_device {
-                SourceIioImuInterface::listen_on_dbus(self.conn.clone(), device.clone()).await?;
-            }
-
-            self.source_device_tasks.spawn(async move {
-                if let Err(e) = source_device.run().await {
-                    log::error!("Failed running device: {:?}", e);
-                }
-                log::debug!("Source device closed");
-                if let Err(e) = tx.send(CompositeCommand::SourceDeviceStopped(device)).await {
-                    log::error!("Failed to send device stop command: {:?}", e);
-                }
-            });
-        }
-        log::debug!("All source device tasks started");
-        Ok(())
-    }
-
     /// Process a single event from a source device. Events are piped through
     /// a translation layer, then dispatched to the appropriate target device(s)
     async fn process_event(
@@ -1459,10 +1396,9 @@ impl CompositeDevice {
 
     /// Executed whenever a source device is added to this [CompositeDevice].
     async fn on_source_device_added(&mut self, device: UdevDevice) -> Result<(), Box<dyn Error>> {
-        if let Err(e) = self.add_source_device(device) {
+        if let Err(e) = self.add_source_device(device).await {
             return Err(e.to_string().into());
         }
-        self.run_source_devices().await?;
 
         // Signal to DBus that source devices have changed
         self.signal_sources_changed().await;
@@ -1525,72 +1461,52 @@ impl CompositeDevice {
         Ok(())
     }
 
-    /// Creates and adds a source device using the given [SourceDeviceInfo]
-    fn add_source_device(
+    /// Creates, adds, and starts a source device using the given [SourceDeviceInfo]
+    async fn add_source_device(
         &mut self,
         device: UdevDevice,
     ) -> Result<(), Box<dyn Error + Send + Sync>> {
         // Check to see if this source device should be blocked.
-        let mut is_blocked = false;
-        let mut is_blocked_evdev = false;
         let source_config = self.config.get_matching_device(&device);
-        if let Some(source_config) = source_config.as_ref() {
-            if let Some(blocked) = source_config.blocked {
-                is_blocked = blocked;
-            }
-        }
-
-        let subsystem = device.subsystem();
-
-        // Hide the device if specified
-        let should_passthru = source_config
+        let is_blocked = source_config
             .as_ref()
-            .and_then(|c| c.passthrough)
+            .and_then(|conf| conf.blocked)
             .unwrap_or(false);
-        let should_hide = !should_passthru && subsystem.as_str() != "iio";
-        if should_hide {
-            let source_path = device.devnode();
-            self.source_devices_to_hide.push(source_path);
+
+        // Create the source device
+        let client = self.client();
+        let source_device = Self::create_source_device(client, device.clone(), source_config)?;
+        let id = source_device.get_id();
+        let capabilities = source_device.get_capabilities()?;
+        let path = source_device.get_device_path();
+        let source_tx = source_device.client();
+        self.source_devices.insert(id.clone(), source_tx);
+
+        // Add the IIO IMU Dbus interface. We do this here because it needs the source
+        // device transmitter and this is the only place we can refrence it at the moment.
+        if let SourceDevice::Iio(_) = source_device {
+            SourceIioImuInterface::listen_on_dbus(self.conn.clone(), device.clone()).await?;
         }
 
-        let source_device = match subsystem.as_str() {
-            "input" => {
-                log::debug!("Adding EVDEV source device: {:?}", device.name());
-                if is_blocked {
-                    is_blocked_evdev = true;
-                }
-                let device = EventDevice::new(device, self.client(), source_config.clone())?;
-                SourceDevice::Event(device)
-            }
-            "hidraw" => {
-                log::debug!("Adding HIDRAW source device: {:?}", device.name());
-                let device = HidRawDevice::new(device, self.client(), source_config.clone())?;
-                SourceDevice::HidRaw(device)
-            }
-            "iio" => {
-                log::debug!("Adding IIO source device: {:?}", device.name());
-                let device = IioDevice::new(device, self.client(), source_config.clone())?;
-                SourceDevice::Iio(device)
-            }
-            "leds" => {
-                log::debug!("Adding LED source device: {:?}", device.sysname());
-                let device = LedDevice::new(device, self.client(), source_config.clone())?;
-                SourceDevice::Led(device)
-            }
-            _ => {
-                return Err(format!(
-                    "Unspported subsystem: {subsystem}, unable to add source device {}",
-                    device.name()
-                )
-                .into())
+        // Spawn the task and run the source device
+        let composite_device_tx = self.tx.clone();
+        let device_info = device.clone();
+        let device_path = path.clone();
+        tokio::task::spawn(async move {
+            // Run the source device
+            if let Err(e) = source_device.run().await {
+                log::error!("Failed running device: {e:?}");
             }
-        };
+            log::debug!("Source device closed: {device_path}");
+            composite_device_tx
+                .send(CompositeCommand::SourceDeviceStopped(device_info))
+                .await;
+        });
 
         // Get the capabilities of the source device.
         // TODO: When we *remove* a source device, we also need to remove
         // capabilities
         if !is_blocked {
-            let capabilities = source_device.get_capabilities()?;
             for cap in capabilities {
                 if self.translatable_capabilities.contains(&cap) {
                     continue;
@@ -1600,15 +1516,11 @@ impl CompositeDevice {
         }
 
         // Check if this device should be blocked from sending events to target devices.
-        let id = source_device.get_id();
-        if let Some(device_config) = self
-            .config
-            .get_matching_device(source_device.get_device_ref())
-        {
+        if let Some(device_config) = self.config.get_matching_device(&device) {
             if let Some(blocked) = device_config.blocked {
                 // Blocked event devices should still be run so they can be
                 // EVIOGRAB'd
-                if blocked && !is_blocked_evdev {
+                if blocked {
                     self.source_devices_blocked.insert(id.clone());
                 }
             }
@@ -1617,14 +1529,48 @@ impl CompositeDevice {
         // TODO: Based on the capability map in the config, translate
         // the capabilities.
         // Keep track of the source device
-        let device_path = source_device.get_device_path();
-        self.source_devices_discovered.push(source_device);
-        self.source_device_paths.push(device_path);
+        self.source_device_paths.push(path);
         self.source_devices_used.push(id);
 
         Ok(())
     }
 
+    /// Create the source device from the given udev device and configuration
+    fn create_source_device(
+        client: CompositeDeviceClient,
+        device: UdevDevice,
+        config: Option<config::SourceDevice>,
+    ) -> Result<SourceDevice, Box<dyn Error + Send + Sync>> {
+        let subsystem = device.subsystem();
+        match subsystem.as_str() {
+            "input" => {
+                log::debug!("Adding EVDEV source device: {:?}", device.name());
+                let device = EventDevice::new(device, client, config)?;
+                Ok(SourceDevice::Event(device))
+            }
+            "hidraw" => {
+                log::debug!("Adding HIDRAW source device: {:?}", device.name());
+                let device = HidRawDevice::new(device, client, config)?;
+                Ok(SourceDevice::HidRaw(device))
+            }
+            "iio" => {
+                log::debug!("Adding IIO source device: {:?}", device.name());
+                let device = IioDevice::new(device, client, config)?;
+                Ok(SourceDevice::Iio(device))
+            }
+            "leds" => {
+                log::debug!("Adding LED source device: {:?}", device.sysname());
+                let device = LedDevice::new(device, client, config)?;
+                Ok(SourceDevice::Led(device))
+            }
+            _ => Err(format!(
+                "Unspported subsystem: {subsystem}, unable to add source device {}",
+                device.name()
+            )
+            .into()),
+        }
+    }
+
     /// Load the given device profile
     pub fn load_device_profile(
         &mut self,

src/input/manager.rs

@@ -551,7 +551,8 @@ impl Manager {
             device,
             self.next_composite_dbus_path()?,
             capability_map,
-        )?;
+        )
+        .await?;
 
         // Check to see if there's already a CompositeDevice for
         // these source devices.
@@ -809,7 +810,7 @@ impl Manager {
         let composite_path = String::from(device.dbus_path());
         let composite_path_clone = composite_path.clone();
         let tx = self.tx.clone();
-        let task = tokio::spawn(async move {
+        let task = tokio::task::spawn(async move {
             if let Err(e) = device.run().await {
                 log::error!("Error running {composite_path}: {}", e.to_string());
             }