power-policy/type-c: Refactor to use deferred channel

Refactor to use deferred channel IPC and increase default timeout, add
timeout to power policy command execution.

Author: RobertZ2011

embedded-service/src/power/policy/action/policy.rs

@@ -1,8 +1,13 @@
 //! Policy state machine
+use embassy_time::{with_timeout, Duration};
+
 use super::*;
 use crate::power::policy::{device, Error, PowerCapability};
 use crate::{error, info};
 
+/// Default timeout for device commands to prevent the policy from getting stuck
+const DEFAULT_TIMEOUT: Duration = Duration::from_millis(5000);
+
 /// Policy state machine control
 pub struct Policy<'a, S: Kind> {
     device: &'a device::Device,
@@ -42,23 +47,32 @@ impl<'a, S: Kind> Policy<'a, S> {
         }
     }
 
-    async fn disconnect_internal(&self) -> Result<(), Error> {
+    /// Common disconnect function used by multiple states
+    async fn disconnect_internal_no_timeout(&self) -> Result<(), Error> {
         info!("Device {} got disconnect request", self.device.id().0);
         self.device
-            .execute_device_request(device::RequestData::Disconnect)
+            .execute_device_command(device::CommandData::Disconnect)
             .await?
             .complete_or_err()?;
         self.device.set_state(device::State::Idle).await;
         self.device.exit_recovery().await;
         Ok(())
     }
 
-    /// Connect this device as a provider
-    async fn connect_provider_internal(&self, capability: PowerCapability) -> Result<(), Error> {
+    /// Common disconnect function used by multiple states
+    async fn disconnect_internal(&self) -> Result<(), Error> {
+        match with_timeout(DEFAULT_TIMEOUT, self.disconnect_internal_no_timeout()).await {
+            Ok(r) => r,
+            Err(_) => Err(Error::Timeout),
+        }
+    }
+
+    /// Common connect provider function used by multiple states
+    async fn connect_provider_internal_no_timeout(&self, capability: PowerCapability) -> Result<(), Error> {
         info!("Device {} connecting provider", self.device.id().0);
 
         self.device
-            .execute_device_request(device::RequestData::ConnectProvider(capability))
+            .execute_device_command(device::CommandData::ConnectProvider(capability))
             .await?
             .complete_or_err()?;
 
@@ -68,18 +82,29 @@ impl<'a, S: Kind> Policy<'a, S> {
 
         Ok(())
     }
+
+    /// Common connect provider function used by multiple states
+    async fn connect_provider_internal(&self, capability: PowerCapability) -> Result<(), Error> {
+        match with_timeout(DEFAULT_TIMEOUT, self.connect_provider_internal_no_timeout(capability)).await {
+            Ok(r) => r,
+            Err(_) => Err(Error::Timeout),
+        }
+    }
 }
 
 // The policy can do nothing when no device is attached
 impl Policy<'_, Detached> {}
 
 impl<'a> Policy<'a, Idle> {
     /// Connect this device as a consumer
-    pub async fn connect_consumer(self, capability: PowerCapability) -> Result<Policy<'a, ConnectedConsumer>, Error> {
+    pub async fn connect_consumer_no_timeout(
+        self,
+        capability: PowerCapability,
+    ) -> Result<Policy<'a, ConnectedConsumer>, Error> {
         info!("Device {} connecting consumer", self.device.id().0);
 
         self.device
-            .execute_device_request(device::RequestData::ConnectConsumer(capability))
+            .execute_device_command(device::CommandData::ConnectConsumer(capability))
             .await?
             .complete_or_err()?;
 
@@ -89,6 +114,23 @@ impl<'a> Policy<'a, Idle> {
         Ok(Policy::new(self.device))
     }
 
+    /// Connect this device as a consumer
+    pub async fn connect_consumer(self, capability: PowerCapability) -> Result<Policy<'a, ConnectedConsumer>, Error> {
+        match with_timeout(DEFAULT_TIMEOUT, self.connect_consumer_no_timeout(capability)).await {
+            Ok(r) => r,
+            Err(_) => Err(Error::Timeout),
+        }
+    }
+
+    /// Connect this device as a provider
+    pub async fn connect_provider_no_timeout(
+        self,
+        capability: PowerCapability,
+    ) -> Result<Policy<'a, ConnectedProvider>, Error> {
+        self.connect_provider_internal_no_timeout(capability).await?;
+        Ok(Policy::new(self.device))
+    }
+
     /// Connect this device as a provider
     pub async fn connect_provider(self, capability: PowerCapability) -> Result<Policy<'a, ConnectedProvider>, Error> {
         self.connect_provider_internal(capability).await?;
@@ -97,6 +139,12 @@ impl<'a> Policy<'a, Idle> {
 }
 
 impl<'a> Policy<'a, ConnectedConsumer> {
+    /// Disconnect this device
+    pub async fn disconnect_no_timeout(self) -> Result<Policy<'a, Idle>, Error> {
+        self.disconnect_internal_no_timeout().await?;
+        Ok(Policy::new(self.device))
+    }
+
     /// Disconnect this device
     pub async fn disconnect(self) -> Result<Policy<'a, Idle>, Error> {
         self.disconnect_internal().await?;
@@ -106,18 +154,36 @@ impl<'a> Policy<'a, ConnectedConsumer> {
 
 impl<'a> Policy<'a, ConnectedProvider> {
     /// Disconnect this device
-    pub async fn disconnect(self) -> Result<Policy<'a, Idle>, Error> {
-        if let Err(e) = self.disconnect_internal().await {
+    pub async fn disconnect_no_timeout(self) -> Result<Policy<'a, Idle>, Error> {
+        if let Err(e) = self.disconnect_internal_no_timeout().await {
             error!("Error disconnecting device {}: {:?}", self.device.id().0, e);
             self.device.enter_recovery().await;
             return Err(e);
         }
         Ok(Policy::new(self.device))
     }
 
+    /// Disconnect this device
+    pub async fn disconnect(self) -> Result<Policy<'a, Idle>, Error> {
+        match with_timeout(DEFAULT_TIMEOUT, self.disconnect_no_timeout()).await {
+            Ok(r) => r,
+            Err(_) => Err(Error::Timeout),
+        }
+    }
+
     /// Connect this device as a provider
-    pub async fn connect_provider(&self, capability: PowerCapability) -> Result<(), Error> {
-        self.connect_provider_internal(capability).await
+    pub async fn connect_provider_no_timeout(
+        &self,
+        capability: PowerCapability,
+    ) -> Result<Policy<'a, ConnectedProvider>, Error> {
+        self.connect_provider_internal_no_timeout(capability).await?;
+        Ok(Policy::new(self.device))
+    }
+
+    /// Connect this device as a provider
+    pub async fn connect_provider(&self, capability: PowerCapability) -> Result<Policy<'a, ConnectedProvider>, Error> {
+        self.connect_provider_internal(capability).await?;
+        Ok(Policy::new(self.device))
     }
 
     /// Get the provider power capability of this device

embedded-service/src/power/policy/device.rs

@@ -2,11 +2,11 @@
 use core::ops::DerefMut;
 
 use embassy_sync::blocking_mutex::raw::NoopRawMutex;
-use embassy_sync::channel::Channel;
 use embassy_sync::mutex::Mutex;
 
 use super::{action, DeviceId, Error, PowerCapability};
 use crate::intrusive_list;
+use crate::ipc::deferred;
 
 /// Most basic device states
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -63,7 +63,7 @@ struct InternalState {
 /// Data for a device request
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
-pub enum RequestData {
+pub enum CommandData {
     /// Start consuming on this device
     ConnectConsumer(PowerCapability),
     /// Start providinig on this device
@@ -75,11 +75,11 @@ pub enum RequestData {
 /// Request from power policy service to a device
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
-pub struct Request {
+pub struct Command {
     /// Target device
     pub id: DeviceId,
     /// Request data
-    pub data: RequestData,
+    pub data: CommandData,
 }
 
 /// Data for a device response
@@ -110,9 +110,6 @@ pub struct Response {
     pub data: ResponseData,
 }
 
-/// Channel size for device requests
-pub const DEVICE_CHANNEL_SIZE: usize = 1;
-
 /// Device struct
 pub struct Device {
     /// Intrusive list node
@@ -121,10 +118,8 @@ pub struct Device {
     id: DeviceId,
     /// Current state of the device
     state: Mutex<NoopRawMutex, InternalState>,
-    /// Channel for requests to the device
-    request: Channel<NoopRawMutex, RequestData, DEVICE_CHANNEL_SIZE>,
-    /// Channel for responses from the device
-    response: Channel<NoopRawMutex, InternalResponseData, DEVICE_CHANNEL_SIZE>,
+    /// Command channel
+    command: deferred::Channel<NoopRawMutex, CommandData, InternalResponseData>,
 }
 
 impl Device {
@@ -138,8 +133,7 @@ impl Device {
                 consumer_capability: None,
                 in_recovery: false,
             }),
-            request: Channel::new(),
-            response: Channel::new(),
+            command: deferred::Channel::new(),
         }
     }
 
@@ -191,20 +185,14 @@ impl Device {
         self.state.lock().await.in_recovery = false;
     }
 
-    /// Sends a request to this device and returns a response
-    pub(super) async fn execute_device_request(&self, request: RequestData) -> Result<ResponseData, Error> {
-        self.request.send(request).await;
-        self.response.receive().await
-    }
-
-    /// Wait for a request
-    pub async fn wait_request(&self) -> RequestData {
-        self.request.receive().await
+    /// Execute a command on the device
+    pub(super) async fn execute_device_command(&self, command: CommandData) -> Result<ResponseData, Error> {
+        self.command.execute(command).await
     }
 
-    /// Send a response
-    pub async fn send_response(&self, response: InternalResponseData) {
-        self.response.send(response).await;
+    /// Create a handler for the command channel
+    pub async fn receive(&self) -> deferred::Request<'_, NoopRawMutex, CommandData, InternalResponseData> {
+        self.command.receive().await
     }
 
     /// Internal function to set device state

embedded-service/src/power/policy/mod.rs

@@ -20,6 +20,8 @@ pub enum Error {
     InvalidState(device::StateKind, device::StateKind),
     /// Invalid response
     InvalidResponse,
+    /// Timeout
+    Timeout,
     /// Bus error
     Bus,
     /// Generic failure