Forget TaskHandles to keep them running

src/executor.rs

@@ -12,18 +12,20 @@ pub trait Task: Future<Output = ()> + Send {}
 impl<T: Future<Output = ()> + Send> Task for T {}
 
 pub trait TaskExecutor {
-    fn spawn(&self, future: Pin<Box<dyn Task>>) -> Result<Box<dyn TaskExec>, Box<dyn Error>>;
+    fn spawn(&self, future: Pin<Box<dyn Task>>) -> Result<Box<dyn TaskExec<()>>, Box<dyn Error>>;
 }
-pub trait TaskExec: Future<Output = Result<(), Box<dyn Error>>> + Unpin + Send + Sync {
-    fn abort(self);
-    fn forget(self);
+pub trait TaskExec<T: Send>:
+    Future<Output = Result<T, Box<dyn Error>>> + Unpin + Send + Sync
+{
+    fn abort(self: Box<Self>);
+    fn forget(self: Box<Self>);
 }
 pub struct TaskHandle<T: Send> {
-    handle: Box<dyn TaskExec>,
+    handle: Box<dyn TaskExec<()>>,
     recv: Receiver<T>,
 }
 impl<T: Send> TaskHandle<T> {
-    pub fn new(handle: Box<dyn TaskExec>, recv: Receiver<T>) -> Self {
+    pub fn new(handle: Box<dyn TaskExec<()>>, recv: Receiver<T>) -> Self {
         Self { handle, recv }
     }
 }
@@ -39,6 +41,22 @@ impl<T: Send> Future for TaskHandle<T> {
         Poll::Pending
     }
 }
+impl<T: Send> TaskHandle<T> {
+    pub fn abort(self) {
+        self.handle.abort()
+    }
+    pub fn forget(self) {
+        self.handle.forget()
+    }
+}
+impl<T: Send> TaskExec<T> for TaskHandle<T> {
+    fn abort(self: Box<Self>) {
+        self.handle.abort()
+    }
+    fn forget(self: Box<Self>) {
+        self.handle.forget()
+    }
+}
 
 pub use executor_impl::*;
 #[cfg(feature = "tokio_executor")]
@@ -49,7 +67,10 @@ mod executor_impl {
     use super::*;
     pub struct TokioExecutor(pub tokio::runtime::Handle);
     impl TaskExecutor for TokioExecutor {
-        fn spawn(&self, future: Pin<Box<dyn Task>>) -> Result<Box<dyn TaskExec>, Box<dyn Error>> {
+        fn spawn(
+            &self,
+            future: Pin<Box<dyn Task>>,
+        ) -> Result<Box<dyn TaskExec<()>>, Box<dyn Error>> {
             Ok(Box::new(TokioJoinHandle(self.0.spawn(future))))
         }
     }
@@ -61,11 +82,11 @@ mod executor_impl {
                 .map_err(|e| Box::new(e) as Box<dyn Error + 'static>)
         }
     }
-    impl TaskExec for TokioJoinHandle {
-        fn abort(self) {
+    impl TaskExec<()> for TokioJoinHandle {
+        fn abort(self: Box<Self>) {
             self.0.abort();
         }
-        fn forget(self) {
+        fn forget(self: Box<Self>) {
             drop(self);
         }
     }
@@ -88,10 +109,13 @@ mod executor_impl {
     }
     pub struct FuturesExecutor(pub futures::executor::ThreadPool);
     impl TaskExecutor for FuturesExecutor {
-        fn spawn(&self, future: Pin<Box<dyn Task>>) -> Result<Box<dyn TaskExec>, Box<dyn Error>> {
+        fn spawn(
+            &self,
+            future: Pin<Box<dyn Task>>,
+        ) -> Result<Box<dyn TaskExec<()>>, Box<dyn Error>> {
             self.0
                 .spawn_with_handle(future)
-                .map(|h| Box::new(FuturesJoinHandle(h)) as Box<dyn TaskExec>)
+                .map(|h| Box::new(FuturesJoinHandle(h)) as Box<dyn TaskExec<()>>)
                 .map_err(|e| Box::new(e) as Box<dyn Error>)
         }
     }
@@ -102,12 +126,12 @@ mod executor_impl {
             Future::poll(Pin::new(&mut self.0), cx).map(|_| Ok(()) as Result<(), Box<dyn Error>>)
         }
     }
-    impl TaskExec for FuturesJoinHandle {
-        fn abort(self) {
-            drop(self.0)
+    impl TaskExec<()> for FuturesJoinHandle {
+        fn abort(self: Box<Self>) {
+            drop(self)
         }
-        fn forget(self) {
-            self.0.forget();
+        fn forget(self: Box<Self>) {
+            self.0.forget()
         }
     }
 }

src/kernel.rs

@@ -98,7 +98,7 @@ where
         }
     };
 
-    sys.run(f).unwrap();
+    sys.run(f).unwrap().forget();
     Ok(kr)
 }
 

src/kernel/kernel_ref.rs

@@ -36,10 +36,11 @@ impl KernelRef {
 
     fn send(&self, msg: KernelMsg, sys: &ActorSystem) {
         let mut tx = self.tx.clone();
-        let res = sys.run(async move {
+        sys.run(async move {
             drop(tx.send(msg).await);
-        });
-        res.unwrap();
+        })
+        .unwrap()
+        .forget();
     }
 }