Merge pull request #14 from xgroleau/feat/drop-bomb-cancellation

feat: Added drop bomb to panic in case of request cancellation

ector/examples/cancel_panic.rs

@@ -0,0 +1,52 @@
+//! Example where cancellation will cause a panic
+
+#![macro_use]
+#![feature(type_alias_impl_trait)]
+#![feature(async_fn_in_trait)]
+#![allow(incomplete_features)]
+
+use {
+    ector::*,
+    embassy_time::{Duration, Timer},
+    futures::future::{join, select},
+};
+
+async fn test(addr: DynamicAddress<Request<&'static str, &'static str>>) {
+    let req = core::pin::pin!(addr.request("Hello"));
+    let timeout = Timer::after(Duration::from_secs(1));
+    select(req, timeout).await;
+    println!("Timeout");
+}
+
+#[embassy_executor::main]
+async fn main(_s: embassy_executor::Spawner) {
+    // Example of request response
+    static SERVER: ActorContext<Server> = ActorContext::new();
+
+    let address = SERVER.dyn_address();
+    let server = SERVER.mount(Server);
+    let test = test(address);
+    join(server, test).await;
+}
+
+pub struct Server;
+
+impl Actor for Server {
+    type Message = Request<&'static str, &'static str>;
+
+    async fn on_mount<M>(
+        &mut self,
+        _: DynamicAddress<Request<&'static str, &'static str>>,
+        mut inbox: M,
+    ) -> !
+    where
+        M: Inbox<Self::Message>,
+    {
+        println!("Server started!");
+
+        loop {
+            // We don't reply, since we cancel
+            inbox.next().await;
+        }
+    }
+}

ector/src/actor.rs

@@ -7,6 +7,8 @@ use embassy_sync::{
     channel::{Channel, DynamicSender, Receiver, TrySendError},
 };
 
+use crate::drop::DropBomb;
+
 /// Trait that each actor must implement. An actor defines a message type
 /// that it acts on, and an implementation of `on_mount` which is invoked
 /// when the actor is started.
@@ -78,13 +80,17 @@ impl<'a, M, R> ActorRequest<M, R> for DynamicSender<'a, Request<M, R>> {
     async fn request(&self, message: M) -> R {
         let channel: Channel<NoopRawMutex, R, 1> = Channel::new();
         let sender: DynamicSender<'_, R> = channel.sender().into();
+        let bomb = DropBomb::new();
 
         // We guarantee that channel lives until we've been notified on it, at which
         // point its out of reach for the replier.
         let reply_to = unsafe { core::mem::transmute(&sender) };
         let message = Request::new(message, reply_to);
         self.notify(message).await;
-        channel.receive().await
+        let res = channel.receive().await;
+
+        bomb.defuse();
+        res
     }
 }
 
@@ -111,13 +117,17 @@ where
     async fn request(&self, message: M) -> R {
         let channel: Channel<MUT, R, 1> = Channel::new();
         let sender: DynamicSender<'_, R> = channel.sender().into();
+        let bomb = DropBomb::new();
 
         // We guarantee that channel lives until we've been notified on it, at which
         // point its out of reach for the replier.
         let reply_to = unsafe { core::mem::transmute(&sender) };
         let message = Request::new(message, reply_to);
         self.notify(message).await;
-        channel.receive().await
+        let res = channel.receive().await;
+
+        bomb.defuse();
+        res
     }
 }
 
@@ -128,6 +138,9 @@ where
 pub type DynamicAddress<M> = DynamicSender<'static, M>;
 
 /// Type alias over a [DynamicAddress] using a [Request] as message
+///
+/// Safety: You should not cancel a request, it will cause the requesting thread to panic,
+/// and cause UB if the Actor is still alive after the panic (i.e. in a different thread)
 pub type DynamicRequestAddress<M, R> = DynamicSender<'static, Request<M, R>>;
 
 /// A handle to another actor for dispatching messages.
@@ -137,6 +150,9 @@ pub type DynamicRequestAddress<M, R> = DynamicSender<'static, Request<M, R>>;
 pub type Address<M, MUT, const N: usize = 1> = Sender<'static, MUT, M, N>;
 
 /// Type alias over a [Address] using a [Request] as message
+///
+/// Safety: You should not cancel a request, it will cause the requesting thread to panic,
+/// and cause UB if the Actor is still alive after the panic (i.e. in a different thread)
 pub type RequestAddress<M, R, MUT, const N: usize = 1> = Sender<'static, MUT, Request<M, R>, N>;
 
 pub struct Request<M, R>

ector/src/drop.rs

@@ -0,0 +1,25 @@
+//! Original reference https://github.com/embassy-rs/embassy/blob/e1ed492577d2151b4cc8ef995536dd045d2db087/embassy-hal-internal/src/drop.rs#L32
+
+/// Panics if it is improperly disposed of.
+///
+/// This is to forbid cancelling a future/request.
+///
+/// To properly dispose, call the [defuse](Self::defuse) method before this object is dropped.
+#[must_use = "to delay the drop bomb invokation to the end of the scope"]
+pub struct DropBomb(());
+impl DropBomb {
+    pub fn new() -> Self {
+        Self(())
+    }
+
+    /// Defuses the bomb, rendering it safe to drop.
+    pub fn defuse(self) {
+        core::mem::forget(self)
+    }
+}
+
+impl Drop for DropBomb {
+    fn drop(&mut self) {
+        panic!("Dropped before the request completed. You  cannot cancel an ongoing request")
+    }
+}

ector/src/lib.rs

@@ -8,6 +8,7 @@
 pub(crate) mod fmt;
 
 mod actor;
+mod drop;
 pub use {actor::*, ector_macros::*};
 
 // Reexport mutex types