Reactor panic safety

khonsulabs · Aug 10, 2024 · 51f5283 · 51f5283
1 parent 9805e00
commit 51f5283
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 7 deletions.
diff --git a/muse-reactor/src/lib.rs b/muse-reactor/src/lib.rs
@@ -72,6 +72,7 @@ use std::fmt::Debug;
 use std::future::Future;
 use std::marker::PhantomData;
 use std::num::NonZeroUsize;
+use std::panic::{self, AssertUnwindSafe};
 use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
@@ -516,30 +517,30 @@ where
                 let pinned_future = Pin::new(&mut future);
 
                 let mut context = Context::from_waker(&task.waker);
-                match pinned_future.poll(&mut context) {
-                    Poll::Ready(Ok(result)) => {
+                match panic::catch_unwind(AssertUnwindSafe(|| pinned_future.poll(&mut context))) {
+                    Ok(Poll::Ready(Ok(result))) => {
                         drop(future);
                         let result = root_result(Ok(result), &mut vm_context);
                         drop(vm_context);
                         tasks.complete_running_task(result, &mut self.budgets);
                     }
-                    Poll::Ready(Err(ExecutionError::Exception(err))) => {
+                    Ok(Poll::Ready(Err(ExecutionError::Exception(err)))) => {
                         drop(future);
                         let result = root_result(Err(err), &mut vm_context);
                         drop(vm_context);
                         tasks.complete_running_task(result, &mut self.budgets);
                     }
-                    Poll::Ready(Err(ExecutionError::Waiting)) | Poll::Pending => {
+                    Ok(Poll::Ready(Err(ExecutionError::Waiting)) | Poll::Pending) => {
                         task.executing = false;
                         tasks.executing.pop_front();
                     }
-                    Poll::Ready(Err(ExecutionError::Timeout)) => {
+                    Ok(Poll::Ready(Err(ExecutionError::Timeout))) => {
                         // Task is still executing, but took longer than its
                         // time slice. Keep it in queue for the next iteration
                         // of the loop.
                         tasks.executing.rotate_left(1);
                     }
-                    Poll::Ready(Err(ExecutionError::NoBudget)) => {
+                    Ok(Poll::Ready(Err(ExecutionError::NoBudget))) => {
                         if let Some(budget) = task
                             .budget_pool
                             .and_then(|pool_id| self.budgets.get_mut(&pool_id))
@@ -578,6 +579,20 @@ where
                             tasks.complete_running_task(result, &mut self.budgets);
                         }
                     }
+                    Err(mut panic) => {
+                        drop(future);
+                        let result = root_result(
+                            Err(Value::from(SymbolRef::from("panic"))),
+                            &mut vm_context,
+                        );
+                        drop(vm_context);
+                        tasks.complete_running_task(result, &mut self.budgets);
+                        while let Err(new_panic) =
+                            panic::catch_unwind(AssertUnwindSafe(move || drop(panic)))
+                        {
+                            panic = new_panic;
+                        }
+                    }
                 }
             }
 

diff --git a/muse-reactor/src/tests.rs b/muse-reactor/src/tests.rs
@@ -3,7 +3,8 @@ use std::time::{Duration, Instant};
 
 use muse_lang::compiler::syntax::Ranged;
 use muse_lang::compiler::{self};
-use muse_lang::runtime::value::{Primitive, RootedValue};
+use muse_lang::runtime::symbol::Symbol;
+use muse_lang::runtime::value::{Primitive, RootedValue, RustFunction, Value};
 use muse_lang::vm::Vm;
 use refuse::CollectionGuard;
 use tracing_subscriber::filter::LevelFilter;
@@ -177,3 +178,26 @@ fn pool_cancellation() {
         other => unreachable!("unexpected result: {other:?}"),
     }
 }
+
+#[test]
+fn task_panic() {
+    let reactor = Reactor::build()
+        .new_vm(
+            |guard: &mut CollectionGuard<'_>, _reactor: &ReactorHandle| {
+                let vm = Vm::new(&guard);
+                vm.declare(
+                    "panics",
+                    Value::dynamic(RustFunction::new(|_vm, _arity| panic!()), &guard),
+                    guard,
+                )?;
+                Ok(vm)
+            },
+        )
+        .finish();
+    let task = reactor.spawn_source("panics()").unwrap();
+    let error = task.join().unwrap_err();
+    match error {
+        TaskError::Exception(exc) if exc == RootedValue::from(Symbol::from("panic")) => {}
+        other => unreachable!("Unexpected result: {other:?}"),
+    }
+}