diff --git a/core/src/runtime/combinator.rs b/core/src/runtime/combinator.rs index 5140b85..5d09bc6 100644 --- a/core/src/runtime/combinator.rs +++ b/core/src/runtime/combinator.rs @@ -4,7 +4,11 @@ use std::time::{Duration, Instant}; use super::fiber::{Fiber, FiberState}; use super::runtime::Runtime; -pub fn all(fibers: Vec>) -> Vec { +/// `None` if any fiber in `fibers` was cancelled (`Fiber::wait`'s own +/// `None` case) -- `collect::>>()` short-circuits on the +/// first `None` it sees, same as `all` short-circuiting on a `Result::Err` +/// would. +pub fn all(fibers: Vec>) -> Option> { fibers.into_iter().map(|f| f.wait()).collect() } @@ -15,7 +19,7 @@ pub fn race(_rt: &Runtime, fibers: Vec>) -> Option(_rt: &Runtime, fiber: Fiber, dur: Duration) let deadline = Instant::now() + dur; loop { if fiber.state() == FiberState::Completed { - return Some(fiber.wait()); + return fiber.wait(); } if Instant::now() >= deadline { return None; @@ -46,15 +50,15 @@ mod tests { let f2 = rt.spawn(|| 2); let f3 = rt.spawn(|| 3); let results = all(vec![f1, f2, f3]); - assert_eq!(results, vec![1, 2, 3]); + assert_eq!(results, Some(vec![1, 2, 3])); rt.shutdown(); } #[test] fn test_all_empty() { let rt = Runtime::new(2); - let results: Vec = all(vec![]); - assert!(results.is_empty()); + let results: Option> = all(vec![]); + assert_eq!(results, Some(vec![])); rt.shutdown(); } @@ -65,7 +69,25 @@ mod tests { let f2 = rt.spawn(|| 20); let f3 = rt.spawn(|| 30); let results = all(vec![f1, f2, f3]); - assert_eq!(results, vec![10, 20, 30]); + assert_eq!(results, Some(vec![10, 20, 30])); + rt.shutdown(); + } + + #[test] + fn test_all_none_if_any_cancelled() { + let rt = Runtime::new(1); + // Saturate the sole worker so the second fiber stays Pending long + // enough to cancel (same technique as scheduler::tests:: + // test_scheduler_cancel_fiber). + let (tx, rx) = std::sync::mpsc::channel::<()>(); + let blocker = rt.spawn(move || { + rx.recv().unwrap(); + }); + let f1 = rt.spawn(|| 1); + f1.cancel(); + tx.send(()).unwrap(); + blocker.wait(); + assert_eq!(all(vec![f1]), None); rt.shutdown(); } diff --git a/core/src/runtime/fiber.rs b/core/src/runtime/fiber.rs index 856a570..e9302a0 100644 --- a/core/src/runtime/fiber.rs +++ b/core/src/runtime/fiber.rs @@ -78,7 +78,13 @@ impl Fiber { ) } - pub fn wait(self) -> T { + /// Blocks until the fiber reaches a terminal state, returning its result + /// -- or `None` if it was `Cancelled` (a cancelled fiber's task body + /// never runs, so `set_result` is never called for it; `take().unwrap()` + /// here used to panic on a cancelled fiber, which `try_start()`'s fix to + /// `Scheduler::spawn` made a reliably-hit case rather than a racy one -- + /// see `test_fiber_wait_after_cancel_returns_none`). + pub fn wait(self) -> Option { let mut state = self.inner.state.lock().unwrap(); loop { match *state { @@ -90,7 +96,7 @@ impl Fiber { } } } - self.inner.result.lock().unwrap().take().unwrap() + self.inner.result.lock().unwrap().take() } } @@ -128,7 +134,17 @@ mod tests { fiber.set_result(42); fiber.set_state(FiberState::Completed); assert_eq!(fiber.state(), FiberState::Completed); - assert_eq!(fiber.wait(), 42); + assert_eq!(fiber.wait(), Some(42)); + } + + #[test] + fn test_fiber_wait_after_cancel_returns_none() { + // A cancelled fiber never has set_result() called for it (its task + // body never runs) -- wait() must report that as None, not panic. + let fiber = Fiber::::new(); + fiber.cancel(); + assert_eq!(fiber.state(), FiberState::Cancelled); + assert_eq!(fiber.wait(), None); } #[test] diff --git a/core/src/runtime/runtime.rs b/core/src/runtime/runtime.rs index 408abe4..2934626 100644 --- a/core/src/runtime/runtime.rs +++ b/core/src/runtime/runtime.rs @@ -121,8 +121,8 @@ mod tests { let rt = Runtime::new(2); let f1 = rt.spawn(|| 10); let f2 = rt.spawn(|| 20); - assert_eq!(f1.wait(), 10); - assert_eq!(f2.wait(), 20); + assert_eq!(f1.wait(), Some(10)); + assert_eq!(f2.wait(), Some(20)); rt.shutdown(); } @@ -150,7 +150,7 @@ mod tests { for i in 0..50 { fibers.push(rt.spawn(move || i * 2)); } - let mut results: Vec = fibers.into_iter().map(|f| f.wait()).collect(); + let mut results: Vec = fibers.into_iter().map(|f| f.wait().unwrap()).collect(); results.sort(); let expected: Vec = (0..50).map(|i| i * 2).collect(); assert_eq!(results, expected); @@ -161,7 +161,7 @@ mod tests { fn test_runtime_shutdown() { let rt = Runtime::new(2); let fiber = rt.spawn(|| 7); - assert_eq!(fiber.wait(), 7); + assert_eq!(fiber.wait(), Some(7)); rt.shutdown(); } @@ -169,7 +169,7 @@ mod tests { fn test_runtime_drop_shutdown() { let rt = Runtime::new(2); let fiber = rt.spawn(|| 7); - assert_eq!(fiber.wait(), 7); + assert_eq!(fiber.wait(), Some(7)); drop(rt); } diff --git a/core/src/runtime/scheduler.rs b/core/src/runtime/scheduler.rs index 7d42c6a..b1319ec 100644 --- a/core/src/runtime/scheduler.rs +++ b/core/src/runtime/scheduler.rs @@ -178,8 +178,13 @@ impl Scheduler { F: FnOnce() -> T + Send + 'static, T: Send + 'static, { + // The fiber never leaves this function, so nothing else can cancel it + // -- wait() returning None here would mean the scheduler itself is + // broken, not a real outcome callers need to handle. let fiber = self.spawn(f); - fiber.wait() + fiber + .wait() + .expect("spawn_and_await's own fiber was cancelled by something other than its caller") } pub fn shutdown(mut self) { @@ -232,8 +237,8 @@ mod tests { let sched = Scheduler::with_workers(2); let fiber1 = sched.spawn(|| 10); let fiber2 = sched.spawn(|| 20); - assert_eq!(fiber1.wait(), 10); - assert_eq!(fiber2.wait(), 20); + assert_eq!(fiber1.wait(), Some(10)); + assert_eq!(fiber2.wait(), Some(20)); } #[test] @@ -261,7 +266,7 @@ mod tests { for i in 0..100 { fibers.push(sched.spawn(move || i * 2)); } - let mut results: Vec = fibers.into_iter().map(|f| f.wait()).collect(); + let mut results: Vec = fibers.into_iter().map(|f| f.wait().unwrap()).collect(); results.sort(); let expected: Vec = (0..100).map(|i| i * 2).collect(); assert_eq!(results, expected); @@ -318,7 +323,7 @@ mod tests { fn test_scheduler_shutdown() { let sched = Scheduler::with_workers(2); let fiber = sched.spawn(|| 7); - assert_eq!(fiber.wait(), 7); + assert_eq!(fiber.wait(), Some(7)); sched.shutdown(); } @@ -326,7 +331,7 @@ mod tests { fn test_scheduler_drop_shutdown() { let sched = Scheduler::with_workers(2); let fiber = sched.spawn(|| 7); - assert_eq!(fiber.wait(), 7); + assert_eq!(fiber.wait(), Some(7)); drop(sched); }