diff --git a/AGENTS.md b/AGENTS.md index 0aaf02d7..f18d628e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,7 +35,8 @@ Declare restricted visibility at the module boundary and use `pub` for items in - Allow normal executor `Waker::clone` inside short state critical sections. Custom clone panic recovery and reentrancy are not general guarantees; do not require them in reviews unless an explicit local contract does. - Reuse borrowed-waker registration and avoid redundant clones. -- Keep wake callbacks and replaced or cancelled waker destruction outside primitive locks. If a batch wake panics, attempt the remaining wakes and propagate the first panic. +- Keep wake callbacks and replaced or cancelled waker destruction outside primitive locks. +- Executor waker operations, including cloning, waking, and dropping, are expected not to panic; do not add custom panic recovery. Decision: [#257](https://github.com/apache/asyncband/pull/257). diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ab5099d..a59b9983 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,10 @@ All notable changes to this project will be documented in this file. * Add bounded MPMC `reserve` and `try_reserve` methods returning a borrowed `Permit`, so callers can wait for capacity before constructing a value; sends and reservations receive capacity in wait-queue order, and unused permits release it. +### Notable changes + +* Executor waker operations, including cloning, waking, and dropping, are expected not to panic; recovery from panicking waker callbacks is no longer supported. + ## v0.7.3 (2026-09-29) ### New features diff --git a/asyncband/src/completion/mod.rs b/asyncband/src/completion/mod.rs index 00b03526..2055ec1c 100644 --- a/asyncband/src/completion/mod.rs +++ b/asyncband/src/completion/mod.rs @@ -121,11 +121,6 @@ impl Completer { /// /// Returns `value` if all observers were already dropped. A successful completion does not /// guarantee that an observer will remain alive long enough to read the value. - /// - /// # Panics - /// - /// Panics if notifying a waiting observer panics. The value remains committed, and notification - /// is still attempted for every other pending observer before the panic resumes. pub fn complete(mut self, value: T) -> Result<(), T> { let Some(shared) = self.shared.upgrade() else { return Err(value); diff --git a/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index af72eaec..3bbe5ca2 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -102,11 +102,6 @@ impl AutoResetEvent { /// A stored signal is available to a future wait. Further sets coalesce while it remains /// unassigned. Creating a wait future does not register it; registration happens when it is /// first polled without a stored signal. - /// - /// # Panics - /// - /// Panics if waking a selected task panics. Its signal remains assigned and can still be - /// consumed by polling that wait or passed on by dropping it. pub fn set(&self) { let waker = self.state.lock().signal(); if let Some(waker) = waker { diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs index 47722aa2..4e0e2ce9 100644 --- a/asyncband/src/event/manual_reset.rs +++ b/asyncband/src/event/manual_reset.rs @@ -97,11 +97,6 @@ impl ManualResetEvent { /// /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is /// already set has no effect. - /// - /// # Panics - /// - /// Panics if waking a selected task panics. The event remains set, and waking is still - /// attempted for every other selected task before the panic resumes. pub fn set(&self) { let mut wakers = WakerBatch::new(); { diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 0b831c48..76d3ce80 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -15,8 +15,6 @@ // specific language governing permissions and limitations // under the License. -use std::panic; -use std::panic::AssertUnwindSafe; use std::task::Waker; /// Retains the current task waker, returning any replaced registration for unlocked destruction. @@ -35,32 +33,12 @@ pub fn register_waker(slot: &mut Option, waker: &Waker) -> Option } } -/// Wakes every waker while preserving the first panic. -/// -/// If a wake callback panics, the remaining callbacks are still attempted during unwinding. Any -/// later panic is suppressed so the first panic can continue to the caller. +/// Wakes every waker. #[inline] // A no-feature or blocking-only build has no primitive that fans notifications out. #[allow(dead_code)] -pub(crate) fn wake_all(mut wakers: impl Iterator) { - struct WakeRemaining<'a, I: Iterator> { - wakers: &'a mut I, - } - - impl> Drop for WakeRemaining<'_, I> { - fn drop(&mut self) { - // This iterator is empty after normal completion. During unwinding, attempt every - // callback left after the one that panicked without replacing the original panic. - for waker in self.wakers.by_ref() { - let _ = panic::catch_unwind(AssertUnwindSafe(|| waker.wake())); - } - } - } - - let remaining = WakeRemaining { - wakers: &mut wakers, - }; - for waker in remaining.wakers.by_ref() { +pub(crate) fn wake_all(wakers: impl Iterator) { + for waker in wakers { waker.wake(); } } diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 6c7cc4ae..51c7baed 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -162,88 +162,62 @@ impl Semaphore { pub fn notify_all(&self) { let mut waiters = self.waiters.lock(); let mut wakers = WakerBatch::new(); - loop { - match waiters.unlink_first_waiter(|node| { - node.permits = 0; - true - }) { - None => break, - Some((id, waiter)) => { - let remove_now = waiter.waker.is_none(); - if let Some(waker) = waiter.waker.take() { - wakers.push(waker); - } - if remove_now { - waiters.remove_unlinked_waiter(id); - } - } + while let Some((id, waiter)) = waiters.unlink_first_waiter(|node| { + node.permits = 0; + true + }) { + if let Some(waker) = waiter.waker.take() { + wakers.push(waker); + } else { + waiters.remove_unlinked_waiter(id); } } drop(waiters); wake_all(&mut wakers); } - fn insert_permits_with_lock( - &self, + fn insert_permits_with_lock<'a>( + &'a self, mut rem: usize, - waiters: MutexGuard<'_, WaitList>, + mut waiters: MutexGuard<'a, WaitList>, ) { let mut batch = WakerBatch::new(); - let mut lock = Some(waiters); - - // One iterator covers the entire release. If a callback panics, `wake_all` keeps pulling - // batches during unwinding, so the remaining permits are still distributed and notified. - wake_all(std::iter::from_fn(|| { - loop { - if let Some(waker) = batch.next() { - return Some(waker); - } - if rem == 0 { - return None; - } - let mut waiters = lock.take().unwrap_or_else(|| self.waiters.lock()); - while !batch.will_spill() { - match waiters.unlink_first_waiter(|node| { - if node.permits <= rem { - rem -= node.permits; - node.permits = 0; - true - } else { - node.permits -= rem; - rem = 0; - false - } - }) { - None => break, - Some((id, waiter)) => { - let remove_now = waiter.waker.is_none(); - if let Some(waker) = waiter.waker.take() { - batch.push(waker); - } - if remove_now { - waiters.remove_unlinked_waiter(id); - } - } - } + loop { + while rem > 0 && !batch.will_spill() { + let Some((id, waiter)) = waiters.unlink_first_waiter(|node| { + let acquired = node.permits.min(rem); + node.permits -= acquired; + rem -= acquired; + node.permits == 0 + }) else { + break; + }; + if let Some(waker) = waiter.waker.take() { + batch.push(waker); + } else { + waiters.remove_unlinked_waiter(id); } + } - if rem > 0 && waiters.is_empty() { - // Retire the remainder before the overflow check so unwinding cannot retry it. - let added = std::mem::take(&mut rem); - // The lock serializes additions; concurrent operations can only remove permits. - let current = self.permits.load(Ordering::Relaxed); - assert!( - current.checked_add(added).is_some(), - "number of added permits ({added}) would overflow usize::MAX (prev: {current})" - ); - self.permits.fetch_add(added, Ordering::Release); - } + if rem > 0 && waiters.is_empty() { + // The lock serializes additions; concurrent operations can only remove permits. + let current = self.permits.load(Ordering::Relaxed); + assert!( + current.checked_add(rem).is_some(), + "number of added permits ({rem}) would overflow usize::MAX (prev: {current})" + ); + self.permits.fetch_add(rem, Ordering::Release); + rem = 0; + } - // Neither wake callbacks nor destruction of the taken waker run under this lock. - drop(waiters); + drop(waiters); + wake_all(&mut batch); + if rem == 0 { + return; } - })); + waiters = self.waiters.lock(); + } } } @@ -411,8 +385,6 @@ fn acquired_or_enqueue( #[cfg(test)] mod tests { - use std::panic; - use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -442,87 +414,35 @@ mod tests { } #[test] - fn release_distributes_permits_to_all_waiters() { - const WAITER_COUNT: usize = 35; + fn release_distributes_permits_across_multiple_batches() { + const WAITER_COUNT: usize = WakerBatch::STACK_SIZE * 2 + 1; let semaphore = Semaphore::new(0); - let counter = Arc::new(WakeCounter(AtomicUsize::new(0))); - let waker = Waker::from(counter.clone()); + let counters = (0..WAITER_COUNT) + .map(|_| Arc::new(WakeCounter(AtomicUsize::new(0)))) + .collect::>(); + let wakers = counters + .iter() + .map(|counter| Waker::from(counter.clone())) + .collect::>(); let mut acquires = (0..WAITER_COUNT) .map(|_| semaphore.poll_acquire(1)) .collect::>(); - for acquire in &mut acquires { - assert!(acquire.poll_once(&waker).is_pending()); + for (acquire, waker) in acquires.iter_mut().zip(&wakers) { + assert!(acquire.poll_once(waker).is_pending()); } assert_eq!(semaphore.waiters.lock().occupied_len(), WAITER_COUNT); - semaphore.release(WAITER_COUNT); - assert_eq!(counter.0.load(Ordering::Relaxed), WAITER_COUNT); - - for acquire in &mut acquires { - assert!(acquire.poll_once(&waker).is_ready()); - } - assert_eq!(semaphore.waiters.lock().occupied_len(), 0); - } - - #[test] - fn panicking_wakes_preserve_permits_and_the_first_panic() { - const WAITER_COUNT: usize = 65; - - struct TrackedWake { - count: AtomicUsize, - panic_message: Option<&'static str>, - } - - impl Wake for TrackedWake { - fn wake(self: Arc) { - self.count.fetch_add(1, Ordering::Relaxed); - if let Some(message) = self.panic_message { - panic::panic_any(message); - } - } + semaphore.release(WAITER_COUNT + 2); + for counter in counters { + assert_eq!(counter.0.load(Ordering::Relaxed), 1); } + assert_eq!(semaphore.available_permits(), 2); - let semaphore = Semaphore::new(0); - let trackers = (0..WAITER_COUNT) - .map(|index| { - Arc::new(TrackedWake { - count: AtomicUsize::new(0), - panic_message: if index == 0 { - Some("first wake panic") - } else if index == WAITER_COUNT / 2 { - Some("later wake panic") - } else { - None - }, - }) - }) - .collect::>(); - let mut acquires = trackers - .iter() - .map(|tracker| { - let mut acquire = semaphore.poll_acquire(1); - assert!( - acquire - .poll_once(&Waker::from(tracker.clone())) - .is_pending() - ); - acquire - }) - .collect::>(); - - let payload = panic::catch_unwind(AssertUnwindSafe(|| semaphore.release(WAITER_COUNT + 2))) - .expect_err("the original wake panic must reach the caller"); - assert_eq!(payload.downcast_ref::<&str>(), Some(&"first wake panic")); - - for tracker in trackers { - assert_eq!(tracker.count.load(Ordering::Relaxed), 1); - } for acquire in &mut acquires { assert!(acquire.poll_once(Waker::noop()).is_ready()); } - assert_eq!(semaphore.available_permits(), 2); assert_eq!(semaphore.waiters.lock().occupied_len(), 0); } } diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 375ebe9a..f2003d85 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -93,6 +93,9 @@ //! remain with the caller. Await Asyncband futures inside any executor that polls standard Rust //! futures, and compose those runtime services around them. //! +//! Executor waker operations, including cloning, waking, and dropping, are expected not to panic. +//! Recovery from panicking waker operations is not part of the public API contract. +//! //! # Async first, blocking by adaptation //! //! Async and synchronous primitives have different optimization constraints. Asyncband designs its diff --git a/asyncband/src/mpmc/queue.rs b/asyncband/src/mpmc/queue.rs index 19b5e3a7..327a411c 100644 --- a/asyncband/src/mpmc/queue.rs +++ b/asyncband/src/mpmc/queue.rs @@ -30,7 +30,6 @@ use super::TrySendError; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; -use crate::internal::wake_all; pub struct Shared { state: Mutex>, @@ -140,12 +139,11 @@ impl WaitList { return Some(mem::replace(waker, current.clone())); } } - let waker = current.clone(); if let Some(notified) = id.take() { // The notification already took this node's waker, so nothing is retired. remove_waiter(self, notified); } - *id = Some(self.push_back(RecvWaiter::Waiting(waker))); + *id = Some(self.push_back(RecvWaiter::Waiting(current.clone()))); None } } @@ -221,7 +219,9 @@ impl Shared { // notification and reclamation happen without holding the queue lock. mem::replace(&mut state.recv_waiters, WaitList::new()) }; - wake_all(std::iter::from_fn(|| waiters.notify_one())); + while let Some(waker) = waiters.notify_one() { + waker.wake(); + } } pub fn clone_receiver(&self) { @@ -240,9 +240,10 @@ impl Shared { mem::replace(&mut state.send_waiters, WaitList::new()), ) }; - // Release blocked senders before destroying buffered values. Local ownership still drops - // the values if a wake callback unwinds. - wake_all(std::iter::from_fn(|| waiters.take_waiting_waker())); + // Notify blocked senders before destroying buffered values, whose destructors may panic. + while let Some(waker) = waiters.take_waiting_waker() { + waker.wake(); + } drop(discarded); } @@ -337,7 +338,7 @@ impl Permit<'_, T> { return Err(SendError::new(value)); } let waker = state.push(value); - // The queued value now owns this capacity, including if waking a receiver panics. + // The queued value now owns this capacity. mem::forget(self); waker }; @@ -454,7 +455,6 @@ impl Recv<'_, T> { .take() .map(|id| remove_waiter(&mut state.recv_waiters, id)); drop(state); - // Deliver the notification before running waker destructors, which may panic. let result = outcome.map(|(value, waker)| { if let Some(waker) = waker { waker.wake(); diff --git a/asyncband/src/mpsc/bounded/receiver.rs b/asyncband/src/mpsc/bounded/receiver.rs index 11f6f7f3..c09ffee5 100644 --- a/asyncband/src/mpsc/bounded/receiver.rs +++ b/asyncband/src/mpsc/bounded/receiver.rs @@ -26,7 +26,6 @@ use super::State; use crate::internal::mutex::Mutex; use crate::internal::register_waker; use crate::internal::waitlist::WaitList; -use crate::internal::wake_all; use crate::mpsc::RecvError; use crate::mpsc::TryRecvError; @@ -55,15 +54,12 @@ impl Drop for BoundedReceiver { let waiters = mem::replace(&mut state.send_waiters, WaitList::new()); (queue, recv_waker, waiters) }; - // Local ownership also drains the queue if a wake or waker destructor unwinds. - wake_all(std::iter::from_fn(|| { - loop { - let (_, waiter) = waiters.unlink_first_waiter(|_| true)?; - if let Some(waker) = waiter.waker.take() { - return Some(waker); - } + // Notify blocked senders before destroying buffered values, whose destructors may panic. + while let Some((_, waiter)) = waiters.unlink_first_waiter(|_| true) { + if let Some(waker) = waiter.waker.take() { + waker.wake(); } - })); + } drop(recv_waker); drop(queue); } diff --git a/asyncband/src/mpsc/bounded/sender.rs b/asyncband/src/mpsc/bounded/sender.rs index 2f286f04..2781ff6c 100644 --- a/asyncband/src/mpsc/bounded/sender.rs +++ b/asyncband/src/mpsc/bounded/sender.rs @@ -209,7 +209,7 @@ impl Permit<'_, T> { return Err(SendError::new(value)); } state.queue.push_back(value); - // The queued message now owns capacity, even if the wake callback panics. + // The queued message now owns capacity. mem::forget(self); let wake = state.recv_waker.take(); drop(state); diff --git a/asyncband/src/phaser/mod.rs b/asyncband/src/phaser/mod.rs index 72bdea1a..9b00e7c5 100644 --- a/asyncband/src/phaser/mod.rs +++ b/asyncband/src/phaser/mod.rs @@ -240,11 +240,6 @@ impl Phaser { /// /// This operation is idempotent and affects every handle and participant. Existing participants /// may still deregister or be dropped; their removal no longer advances the phase. - /// - /// # Panics - /// - /// If a waker panics, the phaser remains closed and notification is attempted for the other - /// waiters before the panic resumes. pub fn close(&self) { let wakers = { let mut state = self.state.lock(); @@ -437,8 +432,6 @@ impl PhaserParticipant { /// Returns the phase in which arrival was recorded, or [`Closed`] without recording one. /// Repeated calls within one phase count only once. After advancement, an explicit new call /// arrives in the new phase and replaces any previous pending observation. - /// - /// Arrival and the pending observation remain committed if notifying a waker panics. pub fn arrive(&mut self) -> Result { let mut wakers = WakerBatch::new(); let phase = { diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index 67d0a171..876742cd 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -152,9 +152,7 @@ impl Semaphore { /// /// # Panics /// - /// Panics if adding the permits would overflow the total permit count, or if notifying a waiter - /// panics. Added permits remain available, and notification is still attempted for every other - /// eligible waiter before the panic resumes. + /// Panics if adding the permits would overflow the total permit count. /// /// # Examples /// diff --git a/asyncband/src/spmc/queue.rs b/asyncband/src/spmc/queue.rs index c8be6bc1..82b77187 100644 --- a/asyncband/src/spmc/queue.rs +++ b/asyncband/src/spmc/queue.rs @@ -29,7 +29,6 @@ use super::TrySendError; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; -use crate::internal::wake_all; pub struct Shared { state: Mutex>, @@ -109,12 +108,11 @@ impl WaitList { return Some(mem::replace(waker, current.clone())); } } - let waker = current.clone(); if let Some(notified) = id.take() { // The notification already took this node's waker, so nothing is retired. self.remove_waiter(notified); } - *id = Some(self.push_back(Waiter::Waiting(waker))); + *id = Some(self.push_back(Waiter::Waiting(current.clone()))); None } } @@ -149,7 +147,9 @@ impl Shared { // notification and reclamation happen without holding the queue lock. mem::replace(&mut state.recv_waiters, WaitList::new()) }; - wake_all(std::iter::from_fn(|| waiters.notify_one())); + while let Some(waker) = waiters.notify_one() { + waker.wake(); + } } pub fn clone_receiver(&self) { @@ -168,9 +168,10 @@ impl Shared { mem::replace(&mut state.send_waiters, WaitList::new()), ) }; - // Release blocked senders before destroying buffered values. Local ownership still drops - // the values if a wake callback unwinds. - wake_all(std::iter::from_fn(|| waiters.notify_one())); + // Notify blocked senders before destroying buffered values, whose destructors may panic. + while let Some(waker) = waiters.notify_one() { + waker.wake(); + } drop(discarded); } @@ -258,7 +259,6 @@ impl Send<'_, T> { .take() .map(|id| state.send_waiters.remove_waiter(id)); drop(state); - // Deliver the notification before running waker destructors, which may panic. let result = outcome .map(|waker| { if let Some(waker) = waker { @@ -324,7 +324,6 @@ impl Recv<'_, T> { .take() .map(|id| state.recv_waiters.remove_waiter(id)); drop(state); - // Deliver the notification before running waker destructors, which may panic. let result = outcome.map(|(value, waker)| { if let Some(waker) = waker { waker.wake(); diff --git a/tests-integration/src/lib.rs b/tests-integration/src/lib.rs index 00f2045e..58e9fc89 100644 --- a/tests-integration/src/lib.rs +++ b/tests-integration/src/lib.rs @@ -79,14 +79,6 @@ impl Wake for WakeCounter { } } -pub struct PanicWake; - -impl Wake for PanicWake { - fn wake(self: Arc) { - panic!("wake failed"); - } -} - pub fn waker_on_wake(callback: impl FnOnce() + Send + 'static) -> Waker { struct OnWake(Mutex>>); diff --git a/tests-integration/tests/broadcast_test/bounded.rs b/tests-integration/tests/broadcast_test/bounded.rs index 5acfd17b..a278c837 100644 --- a/tests-integration/tests/broadcast_test/bounded.rs +++ b/tests-integration/tests/broadcast_test/bounded.rs @@ -19,7 +19,6 @@ use std::future::Future; use std::sync::Arc; use std::sync::Barrier; use std::task::Context; -use std::task::Wake; use std::task::Waker; use std::thread; @@ -28,7 +27,6 @@ use asyncband::broadcast::mpmc::*; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_once; -use tests_integration::waker_on_wake; /// A payload whose destructor re-enters the channel it was sent through. struct Reentrant { @@ -541,12 +539,8 @@ fn bounded_parked_recv_wakes_when_the_last_sender_drops() { ); } -// --------------------------------------------------------------------------------------------- -// Panic safety -// --------------------------------------------------------------------------------------------- - #[test] -fn panicking_wake_does_not_strand_senders_after_a_large_reclaim() { +fn large_reclaim_notifies_every_blocked_sender() { let (tx, mut fast) = bounded(40); let slow = tx.subscribe(); for value in 0..40 { @@ -559,14 +553,7 @@ fn panicking_wake_does_not_strand_senders_after_a_large_reclaim() { .collect::>(); let wakers = trackers .iter() - .enumerate() - .map(|(index, tracker)| { - let tracker = tracker.clone(); - waker_on_wake(move || { - tracker.wake(); - assert_ne!(index, 0, "first sender wake panics"); - }) - }) + .map(|tracker| Waker::from(tracker.clone())) .collect::>(); let mut sends = (40..80) .map(|value| Box::pin(tx.send(value))) @@ -579,8 +566,7 @@ fn panicking_wake_does_not_strand_senders_after_a_large_reclaim() { ); } - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(slow))); - assert!(result.is_err()); + drop(slow); assert_eq!(tx.retained_message_count(), 0); for tracker in trackers { assert_eq!(tracker.count(), 1); @@ -595,6 +581,10 @@ fn panicking_wake_does_not_strand_senders_after_a_large_reclaim() { } } +// --------------------------------------------------------------------------------------------- +// Panic safety +// --------------------------------------------------------------------------------------------- + #[test] fn bounded_panicking_clone_leaves_the_channel_consistent() { let (tx, mut rx1) = bounded(4); diff --git a/tests-integration/tests/completion_test.rs b/tests-integration/tests/completion_test.rs index 8200e23b..4ea0afc1 100644 --- a/tests-integration/tests/completion_test.rs +++ b/tests-integration/tests/completion_test.rs @@ -24,7 +24,6 @@ use std::thread; use asyncband::blocking::FutureExt; use asyncband::completion; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_with; @@ -190,28 +189,6 @@ fn cancelling_after_abandonment_does_not_retain_the_waker() { assert!(FutureExt::block_on(completion.wait()).is_err()); } -#[test] -fn completion_attempts_every_waker_after_one_panics() { - let (completer, first) = completion::new(); - let second = first.clone(); - let panicking = Waker::from(Arc::new(PanicWake)); - let tracker = Arc::new(WakeCounter::default()); - let tracked = Waker::from(tracker.clone()); - let mut first_wait = Box::pin(first.wait()); - let mut second_wait = Box::pin(second.wait()); - - assert!(poll_with(first_wait.as_mut(), &panicking).is_pending()); - assert!(poll_with(second_wait.as_mut(), &tracked).is_pending()); - - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| completer.complete(11))); - assert!(result.is_err()); - assert_eq!(tracker.count(), 1); - assert_eq!( - poll_with(second_wait.as_mut(), &tracked), - Poll::Ready(Ok(&11)) - ); -} - #[test] fn wake_callbacks_run_outside_the_completion_lock() { assert_completes_without_deadlock(|| { diff --git a/tests-integration/tests/event_test/auto_reset.rs b/tests-integration/tests/event_test/auto_reset.rs index dc0e66b8..aee08760 100644 --- a/tests-integration/tests/event_test/auto_reset.rs +++ b/tests-integration/tests/event_test/auto_reset.rs @@ -16,8 +16,6 @@ // under the License. use std::future::Future; -use std::panic; -use std::panic::AssertUnwindSafe; use std::pin::pin; use std::sync::Arc; use std::sync::Barrier; @@ -30,7 +28,6 @@ use std::thread; use asyncband::blocking::FutureExt; use asyncband::event::AutoResetEvent; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_once; @@ -313,26 +310,6 @@ fn wake_and_waker_destruction_happen_outside_the_event_lock() { }); } -#[test] -fn a_panicking_wake_leaves_its_signal_available_for_cancellation_handoff() { - let event = AutoResetEvent::new(); - let waker = Waker::from(Arc::new(PanicWake)); - let mut selected = Box::pin(event.wait()); - let mut next = Box::pin(event.wait()); - assert!( - selected - .as_mut() - .poll(&mut Context::from_waker(&waker)) - .is_pending() - ); - assert!(panic::catch_unwind(AssertUnwindSafe(|| event.set())).is_err()); - assert!(!event.try_wait()); - assert!(poll_once(next.as_mut()).is_pending()); - drop(selected); - assert!(poll_once(next.as_mut()).is_ready()); - assert!(!event.try_wait()); -} - #[test] fn concurrent_sets_and_waits_publish_state_without_losing_signals() { assert_completes_without_deadlock(|| { diff --git a/tests-integration/tests/event_test/manual_reset.rs b/tests-integration/tests/event_test/manual_reset.rs index f234a7cf..db46cb7f 100644 --- a/tests-integration/tests/event_test/manual_reset.rs +++ b/tests-integration/tests/event_test/manual_reset.rs @@ -16,8 +16,6 @@ // under the License. use std::future::Future; -use std::panic; -use std::panic::AssertUnwindSafe; use std::pin::Pin; use std::pin::pin; use std::sync::Arc; @@ -29,7 +27,6 @@ use std::task::Wake; use std::task::Waker; use asyncband::event::ManualResetEvent; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_once; @@ -115,52 +112,6 @@ fn cancelling_a_committed_waiter_leaves_the_others_committed() { assert!(!event.is_set()); } -// `set` releases every waiter registered for the current unset period, so one broken waker must -// not strand the waiters queued behind it. -#[test] -fn a_panicking_waker_still_releases_the_remaining_waiters() { - let event = ManualResetEvent::new(); - let tracker = Arc::new(WakeCounter::default()); - let tracking_waker = Waker::from(tracker.clone()); - let panicking_waker = Waker::from(Arc::new(PanicWake)); - - let mut first = Box::pin(event.wait()); - let mut exploding = Box::pin(event.wait()); - let mut last = Box::pin(event.wait()); - assert!( - first - .as_mut() - .poll(&mut Context::from_waker(&tracking_waker)) - .is_pending() - ); - assert!( - exploding - .as_mut() - .poll(&mut Context::from_waker(&panicking_waker)) - .is_pending() - ); - assert!( - last.as_mut() - .poll(&mut Context::from_waker(&tracking_waker)) - .is_pending() - ); - - let panicked = panic::catch_unwind(AssertUnwindSafe(|| event.set())); - - assert!(panicked.is_err(), "the waker panic must reach the caller"); - assert_eq!( - tracker.count(), - 2, - "the waiter queued behind the panicking waker was never woken" - ); - assert!(event.is_set()); - assert!( - last.as_mut() - .poll(&mut Context::from_waker(&tracking_waker)) - .is_ready() - ); -} - // A waker that re-enters the event it belongs to, both when woken and when dropped. The internal // lock is non-reentrant, so waking or dropping this waker inside the critical section deadlocks. struct ReentrantWaker(Arc); diff --git a/tests-integration/tests/mpmc_test/callbacks.rs b/tests-integration/tests/mpmc_test/callbacks.rs index cf851391..2cf537de 100644 --- a/tests-integration/tests/mpmc_test/callbacks.rs +++ b/tests-integration/tests/mpmc_test/callbacks.rs @@ -15,18 +15,10 @@ // specific language governing permissions and limitations // under the License. -use std::panic::AssertUnwindSafe; -use std::panic::catch_unwind; -use std::sync::Arc; -use std::task::Waker; - use asyncband::mpmc; -use asyncband::mpmc::RecvError; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::expect_ready; -use tests_integration::poll_once; use tests_integration::poll_with; use tests_integration::waker_on_drop; @@ -68,47 +60,3 @@ fn replacing_a_receive_waker_allows_its_destructor_to_send() { drop(sender); }); } - -#[test] -fn last_sender_attempts_every_wake_after_one_panics() { - let (sender, receiver) = mpmc::unbounded::(); - let competing = receiver.clone(); - let mut panicking = Box::pin(receiver.recv()); - let mut waiting = Box::pin(competing.recv()); - let panic_waker = Waker::from(Arc::new(PanicWake)); - let (waker, wakes) = WakeCounter::new(); - assert!(poll_with(panicking.as_mut(), &panic_waker).is_pending()); - assert!(poll_with(waiting.as_mut(), &waker).is_pending()); - - assert!(catch_unwind(AssertUnwindSafe(|| drop(sender))).is_err()); - assert_eq!(wakes.count(), 1); - assert_eq!( - expect_ready(poll_with(waiting.as_mut(), &waker)), - Err(RecvError::Disconnected) - ); - assert_eq!( - expect_ready(poll_once(panicking.as_mut())), - Err(RecvError::Disconnected) - ); -} - -#[test] -fn last_receiver_releases_senders_and_values_after_a_wake_panics() { - let buffered = Arc::new(()); - let released = Arc::downgrade(&buffered); - let (sender, receiver) = mpmc::bounded(1); - sender.try_send(buffered).unwrap(); - let competing = sender.clone(); - let mut panicking = Box::pin(sender.send(Arc::new(()))); - let mut waiting = Box::pin(competing.send(Arc::new(()))); - let panic_waker = Waker::from(Arc::new(PanicWake)); - let (waker, wakes) = WakeCounter::new(); - assert!(poll_with(panicking.as_mut(), &panic_waker).is_pending()); - assert!(poll_with(waiting.as_mut(), &waker).is_pending()); - - assert!(catch_unwind(AssertUnwindSafe(|| drop(receiver))).is_err()); - assert_eq!(wakes.count(), 1); - assert!(released.upgrade().is_none()); - assert!(expect_ready(poll_with(waiting.as_mut(), &waker)).is_err()); - assert!(expect_ready(poll_once(panicking.as_mut())).is_err()); -} diff --git a/tests-integration/tests/mpsc_test/callbacks.rs b/tests-integration/tests/mpsc_test/callbacks.rs index d3174f68..f8b5d733 100644 --- a/tests-integration/tests/mpsc_test/callbacks.rs +++ b/tests-integration/tests/mpsc_test/callbacks.rs @@ -16,7 +16,6 @@ // under the License. use std::sync::Arc; -use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Context; @@ -230,7 +229,7 @@ fn unbounded_disconnect_drops_partial_and_queued_batches_outside_lock() { #[cfg(panic = "unwind")] #[test] -fn bounded_disconnect_finishes_cleanup_when_a_callback_panics() { +fn bounded_disconnect_finishes_cleanup_when_a_value_destructor_panics() { struct Value { id: usize, drops: Arc<[AtomicUsize; 5]>, @@ -243,68 +242,48 @@ fn bounded_disconnect_finishes_cleanup_when_a_callback_panics() { assert!(!self.panic_on_drop, "payload destructor panicked"); } } - struct Notify { - woken: AtomicBool, - panic_on_wake: bool, - } - impl Wake for Notify { - fn wake(self: Arc) { - self.woken.store(true, Ordering::Relaxed); - assert!(!self.panic_on_wake, "wake callback panicked"); - } - } - for panic_on_wake in [false, true] { - let (tx, rx) = mpsc::bounded(3); - let drops = Arc::new(std::array::from_fn(|_| AtomicUsize::new(0))); - for id in 0..3 { - assert!( - tx.try_send(Value { - id, - drops: drops.clone(), - panic_on_drop: id == 0 && !panic_on_wake, - _sender: Some(tx.clone()), - }) - .is_ok() - ); - } - let notify = [false, true].map(|second| { - Arc::new(Notify { - woken: AtomicBool::new(false), - panic_on_wake: !second && panic_on_wake, - }) - }); - let wakers = notify.each_ref().map(|notify| Waker::from(notify.clone())); - let mut sends = (3..5) - .map(|id| { - Box::pin(tx.send(Value { - id, - drops: drops.clone(), - panic_on_drop: false, - _sender: None, - })) - }) - .collect::>(); - for (send, waker) in sends.iter_mut().zip(&wakers) { - assert!(poll_with(send.as_mut(), waker).is_pending()); - } - assert!(std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(rx))).is_err()); + let (tx, rx) = mpsc::bounded(3); + let drops = Arc::new(std::array::from_fn(|_| AtomicUsize::new(0))); + for id in 0..3 { assert!( - notify - .iter() - .all(|notify| notify.woken.load(Ordering::Relaxed)) + tx.try_send(Value { + id, + drops: drops.clone(), + panic_on_drop: id == 0, + _sender: Some(tx.clone()), + }) + .is_ok() ); - for count in &drops[..3] { - assert_eq!(count.load(Ordering::Relaxed), 1); - } - for (id, send) in (3..5).zip(&mut sends) { - assert_eq!(drops[id].load(Ordering::Relaxed), 0); - let error = match expect_ready(poll_once(send.as_mut())) { - Err(error) => error, - Ok(()) => panic!("the receiver is disconnected"), - }; - assert_eq!(error.into_inner().id, id); - assert_eq!(drops[id].load(Ordering::Relaxed), 1); - } + } + let notifications = [WakeCounter::new(), WakeCounter::new()]; + let mut sends = (3..5) + .map(|id| { + Box::pin(tx.send(Value { + id, + drops: drops.clone(), + panic_on_drop: false, + _sender: None, + })) + }) + .collect::>(); + for (send, (waker, _)) in sends.iter_mut().zip(¬ifications) { + assert!(poll_with(send.as_mut(), waker).is_pending()); + } + assert!(std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(rx))).is_err()); + for (_, wakes) in ¬ifications { + assert_eq!(wakes.count(), 1); + } + for count in &drops[..3] { + assert_eq!(count.load(Ordering::Relaxed), 1); + } + for (id, send) in (3..5).zip(&mut sends) { + assert_eq!(drops[id].load(Ordering::Relaxed), 0); + let error = match expect_ready(poll_once(send.as_mut())) { + Err(error) => error, + Ok(()) => panic!("the receiver is disconnected"), + }; + assert_eq!(error.into_inner().id, id); + assert_eq!(drops[id].load(Ordering::Relaxed), 1); } } diff --git a/tests-integration/tests/mpsc_test/reservation.rs b/tests-integration/tests/mpsc_test/reservation.rs index b16f7efe..99751d4a 100644 --- a/tests-integration/tests/mpsc_test/reservation.rs +++ b/tests-integration/tests/mpsc_test/reservation.rs @@ -17,15 +17,10 @@ use std::cell::Cell; use std::mem; -use std::panic::AssertUnwindSafe; -use std::panic::catch_unwind; -use std::sync::Arc; -use std::task::Waker; use asyncband::mpsc; use asyncband::mpsc::TryRecvError; use asyncband::mpsc::TrySendError; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::expect_ready; use tests_integration::poll_once; @@ -153,21 +148,6 @@ fn a_permit_can_publish_send_only_payloads_from_another_thread() { assert_eq!(rx.try_recv().unwrap().get(), 42); } -#[test] -fn a_panicking_publication_wake_cannot_return_capacity_twice() { - let (tx, mut rx) = mpsc::bounded(1); - let permit = tx.try_reserve().unwrap(); - let waker = Waker::from(Arc::new(PanicWake)); - let mut receive = Box::pin(rx.recv()); - assert!(poll_with(receive.as_mut(), &waker).is_pending()); - assert!(catch_unwind(AssertUnwindSafe(|| permit.send(1))).is_err()); - assert_eq!(tx.try_send(2), Err(TrySendError::Full(2))); - assert_eq!(expect_ready(poll_once(receive.as_mut())), Ok(1)); - drop(receive); - tx.try_send(2).unwrap(); - assert_eq!(rx.try_recv(), Ok(2)); -} - #[test] fn concurrent_cancellation_preserves_capacity_and_message_order() { const PRODUCERS: usize = 3; diff --git a/tests-integration/tests/phaser_test.rs b/tests-integration/tests/phaser_test.rs index 615672c3..28c225a1 100644 --- a/tests-integration/tests/phaser_test.rs +++ b/tests-integration/tests/phaser_test.rs @@ -16,13 +16,10 @@ // under the License. use std::panic; -use std::sync::Arc; use std::task::Poll; -use std::task::Waker; use asyncband::blocking::FutureExt; use asyncband::phaser::Phaser; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::poll_once; use tests_integration::poll_with; @@ -395,37 +392,6 @@ fn cancelling_a_woken_waiter_does_not_unregister_a_next_phase_waiter() { )); } -#[test] -fn panicking_waker_does_not_lose_a_pending_phase() { - let phaser = Phaser::new(); - let phase0 = phaser.phase(); - let mut first = phaser.register_one().unwrap(); - let mut second = phaser.register_one().unwrap(); - let panic_waker = Waker::from(Arc::new(PanicWake)); - let mut observer = Box::pin(phaser.wait(phase0)); - - assert_eq!(poll_with(observer.as_mut(), &panic_waker), Poll::Pending); - assert_eq!(first.arrive().unwrap(), phase0); - - let (polling_waker, _) = WakeCounter::new(); - let mut wait = Box::pin(second.wait()); - let result = panic::catch_unwind(panic::AssertUnwindSafe(|| { - poll_with(wait.as_mut(), &polling_waker) - })); - - assert!(result.is_err()); - drop(wait); - drop(observer); - - let phase1 = phaser.phase(); - assert_ne!(phase1, phase0); - assert_eq!(phaser.arrived_parties(), 0); - - let mut retry = Box::pin(second.wait()); - assert_eq!(poll_once(retry.as_mut()), Poll::Ready(Ok(phase1))); - assert_eq!(phaser.arrived_parties(), 0); -} - #[test] fn a_late_waiter_for_a_completed_phase_is_immediately_ready() { let phaser = Phaser::new(); @@ -564,24 +530,6 @@ fn completed_arrival_remains_successful_after_close_but_cannot_start_another_rou assert_eq!(phaser.phase(), completed); } -#[test] -fn close_survives_a_panicking_waker_and_notifies_other_waiters() { - let phaser = Phaser::new(); - let observed = phaser.phase(); - let panic_waker = Waker::from(Arc::new(PanicWake)); - let (count_waker, counter) = WakeCounter::new(); - let mut first = Box::pin(phaser.wait(observed)); - let mut second = Box::pin(phaser.wait(observed)); - assert!(poll_with(first.as_mut(), &panic_waker).is_pending()); - assert!(poll_with(second.as_mut(), &count_waker).is_pending()); - - assert!(panic::catch_unwind(|| phaser.close()).is_err()); - assert!(phaser.is_closed()); - assert_eq!(counter.count(), 1); - assert!(matches!(poll_once(first.as_mut()), Poll::Ready(Err(_)))); - assert!(matches!(poll_once(second.as_mut()), Poll::Ready(Err(_)))); -} - #[test] fn replacing_a_waiter_waker_can_close_the_phaser_from_its_destructor() { let phaser = Phaser::new(); diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index eeebe23b..02431f3c 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -23,7 +23,6 @@ use std::task::Wake; use std::task::Waker; use asyncband::semaphore::Semaphore; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::expect_ready; use tests_integration::poll_with; @@ -165,46 +164,6 @@ fn wake_then_drop() { assert_eq!(s.available_permits(), 2); } -#[test] -fn release_attempts_every_waker_after_one_panics() { - let semaphore = Semaphore::new(0); - let mut panicking = pin!(semaphore.acquire(1)); - let mut tracked = pin!(semaphore.acquire(1)); - let panic_waker = Waker::from(Arc::new(PanicWake)); - let wake_count = Arc::new(WakeCounter::default()); - let tracked_waker = Waker::from(wake_count.clone()); - - assert!( - panicking - .as_mut() - .poll(&mut Context::from_waker(&panic_waker)) - .is_pending() - ); - assert!( - tracked - .as_mut() - .poll(&mut Context::from_waker(&tracked_waker)) - .is_pending() - ); - - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| semaphore.release(2))); - - assert!(result.is_err()); - assert_eq!(wake_count.count(), 1); - assert!( - panicking - .as_mut() - .poll(&mut Context::from_waker(Waker::noop())) - .is_ready() - ); - assert!( - tracked - .as_mut() - .poll(&mut Context::from_waker(Waker::noop())) - .is_ready() - ); -} - #[test] fn cancellation_restores_permits_before_dropping_waker() { struct AssertPermitsOnDrop { diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs index 2636e77f..d4e5d2e3 100644 --- a/tests-integration/tests/watch_test.rs +++ b/tests-integration/tests/watch_test.rs @@ -23,7 +23,6 @@ use std::task::Waker; use asyncband::blocking::FutureExt; use asyncband::watch; -use tests_integration::PanicWake; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_with; @@ -300,28 +299,6 @@ fn one_update_wakes_every_waiting_receiver_once() { assert_eq!(second.get(), 1); } -#[test] -fn panicking_waker_does_not_skip_other_waiters() { - let (tx, mut first) = watch::channel(0); - let mut second = first.clone(); - let panicking = Waker::from(Arc::new(PanicWake)); - let tracker = Arc::new(WakeCounter::default()); - let tracked = Waker::from(tracker.clone()); - let mut first_changed = Box::pin(first.changed()); - let mut second_changed = Box::pin(second.changed()); - - assert!(poll_with(first_changed.as_mut(), &panicking).is_pending()); - assert!(poll_with(second_changed.as_mut(), &tracked).is_pending()); - - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| tx.send(1))); - assert!(result.is_err()); - assert_eq!(tracker.count(), 1); - assert_eq!( - poll_with(second_changed.as_mut(), &tracked), - Poll::Ready(Ok(())) - ); -} - #[test] fn only_the_last_sender_drop_wakes_a_waiter() { let (tx, mut rx) = watch::channel(());