diff --git a/AGENTS.md b/AGENTS.md index 3029eabb..cefe6005 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,9 +31,11 @@ Use the shared [license-audit skill](.agents/skills/license-audit/SKILL.md) for Declare restricted visibility at the module boundary and use `pub` for items in that module's API. -Use imports for referenced symbols. Keep a short module qualifier or use an explicit alias when a bare name would obscure its origin or conflict with another symbol, such as `io::Error`. +Use imports for referenced symbols and preserve their original names. Keep enough module or crate qualification to explain the operation, identify its source or execution context, and distinguish same-named symbols, such as `io::Error`, `pool::bounded`, `iter::once`, or `tokio::time::sleep`. Meaningful qualification is useful even without a name conflict. Avoid `use ... as ...` unless there is a compelling reason; aliases make symbols harder to discover. -Start ordinary intra-crate imports at `crate`; avoid explicit `super::...` imports and qualified paths. Reserve `use super::*` for test modules. +Start ordinary intra-crate imports at `crate`; avoid explicit `super::...` paths. Reserve `use super::*` for test modules. + +Keep `pub use self::...` when re-exporting a module's own children as its public API. ## Waker Contract diff --git a/asyncband/src/blocking/executor.rs b/asyncband/src/blocking/executor.rs index 0a9c1e2a..8bebf8fc 100644 --- a/asyncband/src/blocking/executor.rs +++ b/asyncband/src/blocking/executor.rs @@ -16,7 +16,7 @@ use std::task::Waker; use std::time::Duration; use std::time::Instant; -use super::parker::Parker; +use crate::blocking::parker::Parker; pub(super) fn block_on(future: F) -> F::Output { let mut future = pin!(future.into_future()); diff --git a/asyncband/src/blocking/mod.rs b/asyncband/src/blocking/mod.rs index 979714ca..95cd671d 100644 --- a/asyncband/src/blocking/mod.rs +++ b/asyncband/src/blocking/mod.rs @@ -26,7 +26,7 @@ //! ``` //! use std::time::Duration; //! -//! use asyncband::blocking::FutureExt as _; +//! use asyncband::blocking::FutureExt; //! //! assert_eq!(async { 42 }.block_on(), 42); //! assert_eq!(async { 42 }.wait_timeout(Duration::ZERO), Some(42)); @@ -67,7 +67,7 @@ use std::time::Duration; /// ``` /// use std::time::Duration; /// -/// use asyncband::blocking::FutureExt as _; +/// use asyncband::blocking::FutureExt; /// /// assert_eq!(async { 42 }.block_on(), 42); /// assert_eq!(async { 42 }.wait_timeout(Duration::ZERO), Some(42)); diff --git a/asyncband/src/broadcast/mpmc/bounded/mod.rs b/asyncband/src/broadcast/mpmc/bounded/mod.rs index 0bfbc8b5..43735ed1 100644 --- a/asyncband/src/broadcast/mpmc/bounded/mod.rs +++ b/asyncband/src/broadcast/mpmc/bounded/mod.rs @@ -109,12 +109,12 @@ use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; -use super::common; -use super::common::Backlog; -use super::common::Inner; -use super::error::RecvError; -use super::error::TryRecvError; -use super::error::TrySendError; +use crate::broadcast::mpmc::common; +use crate::broadcast::mpmc::common::Backlog; +use crate::broadcast::mpmc::common::Inner; +use crate::broadcast::mpmc::error::RecvError; +use crate::broadcast::mpmc::error::TryRecvError; +use crate::broadcast::mpmc::error::TrySendError; use crate::internal::arena::SlotId; use crate::internal::mutex::Mutex; use crate::internal::semaphore::Acquire; diff --git a/asyncband/src/broadcast/mpmc/common.rs b/asyncband/src/broadcast/mpmc/common.rs index 74a04a9a..f1a847b0 100644 --- a/asyncband/src/broadcast/mpmc/common.rs +++ b/asyncband/src/broadcast/mpmc/common.rs @@ -32,8 +32,8 @@ use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; -use super::error::RecvError; -use super::error::TryRecvError; +use crate::broadcast::mpmc::error::RecvError; +use crate::broadcast::mpmc::error::TryRecvError; use crate::internal::arena::Arena; use crate::internal::arena::SlotId; use crate::internal::mutex::Mutex; @@ -531,9 +531,7 @@ pub fn take_msg(msg: Arc, reclaimed: Reclaimed) -> T { #[cfg(test)] mod tests { - use std::sync::Arc; - - use super::Backlog; + use super::*; /// Drives every cursor-count mutation point — subscribing, publishing, receiving, and dropping /// — and checks the accounting after each step. diff --git a/asyncband/src/broadcast/mpmc/error.rs b/asyncband/src/broadcast/mpmc/error.rs index f1cbce93..28f21726 100644 --- a/asyncband/src/broadcast/mpmc/error.rs +++ b/asyncband/src/broadcast/mpmc/error.rs @@ -16,6 +16,7 @@ // under the License. use std::any::type_name; +use std::error::Error; use std::fmt; /// Error returned by [`BoundedSender::try_send`]. @@ -65,7 +66,7 @@ impl fmt::Debug for TrySendError { } } -impl std::error::Error for TrySendError {} +impl Error for TrySendError {} /// Error returned by `recv`. #[derive(Debug, Clone, PartialEq, Eq)] @@ -80,7 +81,7 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} /// Error returned by `try_recv`. #[derive(Debug, Clone, PartialEq, Eq)] @@ -100,4 +101,4 @@ impl fmt::Display for TryRecvError { } } -impl std::error::Error for TryRecvError {} +impl Error for TryRecvError {} diff --git a/asyncband/src/broadcast/mpmc/unbounded/mod.rs b/asyncband/src/broadcast/mpmc/unbounded/mod.rs index de2eab06..044796d5 100644 --- a/asyncband/src/broadcast/mpmc/unbounded/mod.rs +++ b/asyncband/src/broadcast/mpmc/unbounded/mod.rs @@ -52,7 +52,7 @@ //! assert_eq!(late.try_recv(), Err(TryRecvError::Empty)); //! ``` //! -//! [`bounded`]: super::bounded +//! [`bounded`]: crate::broadcast::mpmc::bounded use std::fmt; use std::future::Future; @@ -63,11 +63,11 @@ use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; -use super::common; -use super::common::Backlog; -use super::common::Inner; -use super::error::RecvError; -use super::error::TryRecvError; +use crate::broadcast::mpmc::common; +use crate::broadcast::mpmc::common::Backlog; +use crate::broadcast::mpmc::common::Inner; +use crate::broadcast::mpmc::error::RecvError; +use crate::broadcast::mpmc::error::TryRecvError; use crate::internal::arena::SlotId; use crate::internal::mutex::Mutex; use crate::internal::wakerset::WakerToken; diff --git a/asyncband/src/completion/mod.rs b/asyncband/src/completion/mod.rs index 95ce3c15..923e80be 100644 --- a/asyncband/src/completion/mod.rs +++ b/asyncband/src/completion/mod.rs @@ -48,6 +48,7 @@ //! # } //! ``` +use std::error::Error; use std::fmt; use std::future::Future; use std::mem; @@ -91,7 +92,7 @@ impl fmt::Display for Abandoned { } } -impl std::error::Error for Abandoned {} +impl Error for Abandoned {} /// The capability that completes a [`Completion`] with one value. /// diff --git a/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index 3bbe5ca2..9e6107f5 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -37,8 +37,8 @@ use crate::internal::waitlist::WaiterId; /// An unassigned signal can be cleared with [`reset`](Self::reset). /// /// Waiting consumes a signal without returning it on completion. Unlike a -/// [`ManualResetEvent`](super::ManualResetEvent), this event does not release all observers of a -/// condition. Unlike a semaphore, it does not count unused signals or return a permit guard. +/// [`ManualResetEvent`](crate::event::ManualResetEvent), this event does not release all observers +/// of a condition. Unlike a semaphore, it does not count unused signals or return a permit guard. /// /// # Usage /// diff --git a/asyncband/src/internal/arena.rs b/asyncband/src/internal/arena.rs index 68a6b4e1..2c33b5fc 100644 --- a/asyncband/src/internal/arena.rs +++ b/asyncband/src/internal/arena.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use std::fmt; use std::mem; use std::num::NonZeroUsize; @@ -41,8 +42,8 @@ impl SlotId { } } -impl std::fmt::Debug for SlotId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl fmt::Debug for SlotId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_tuple("SlotId").field(&self.index()).finish() } } diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 1a539a2b..13b26424 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -385,8 +385,6 @@ fn acquired_or_enqueue( #[cfg(test)] mod tests { use std::sync::Arc; - use std::sync::atomic::AtomicUsize; - use std::sync::atomic::Ordering; use std::task::Wake; use super::*; diff --git a/asyncband/src/internal/waker_batch.rs b/asyncband/src/internal/waker_batch.rs index ff9c0e81..af27db78 100644 --- a/asyncband/src/internal/waker_batch.rs +++ b/asyncband/src/internal/waker_batch.rs @@ -107,9 +107,8 @@ mod tests { use std::sync::Arc; use std::sync::Mutex; use std::task::Wake; - use std::task::Waker; - use super::WakerBatch; + use super::*; const STACK_SIZE: usize = WakerBatch::STACK_SIZE; diff --git a/asyncband/src/internal/wakerset.rs b/asyncband/src/internal/wakerset.rs index cbd3ff0e..58f19aeb 100644 --- a/asyncband/src/internal/wakerset.rs +++ b/asyncband/src/internal/wakerset.rs @@ -111,7 +111,7 @@ impl WakerSet { #[cfg(test)] mod tests { - use super::WakerToken; + use super::*; #[test] fn waker_token_preserves_the_option_niche() { diff --git a/asyncband/src/mpmc/bounded.rs b/asyncband/src/mpmc/bounded.rs index 31dc1d9b..84f19e23 100644 --- a/asyncband/src/mpmc/bounded.rs +++ b/asyncband/src/mpmc/bounded.rs @@ -18,12 +18,12 @@ use std::fmt; use std::sync::Arc; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; -pub use super::queue::Permit; -use super::queue::Shared; +use crate::mpmc::RecvError; +use crate::mpmc::SendError; +use crate::mpmc::TryRecvError; +use crate::mpmc::TrySendError; +use crate::mpmc::queue::Permit; +use crate::mpmc::queue::Shared; /// Creates a bounded multi-producer, multi-consumer queue. /// @@ -108,7 +108,9 @@ impl BoundedSender { /// ``` /// # #[tokio::main] /// # async fn main() { - /// let (sender, receiver) = asyncband::mpmc::bounded(1); + /// use asyncband::mpmc; + /// + /// let (sender, receiver) = mpmc::bounded(1); /// let permit = sender.reserve().await.unwrap(); /// let message = String::from("constructed after capacity became available"); /// permit.send(message).unwrap(); diff --git a/asyncband/src/mpmc/error.rs b/asyncband/src/mpmc/error.rs index 200f013e..06bfcf89 100644 --- a/asyncband/src/mpmc/error.rs +++ b/asyncband/src/mpmc/error.rs @@ -16,6 +16,7 @@ // under the License. use std::any::type_name; +use std::error::Error; use std::fmt; /// An error returned when trying to send on a disconnected queue. @@ -52,7 +53,7 @@ impl fmt::Debug for SendError { } } -impl std::error::Error for SendError {} +impl Error for SendError {} /// Error returned by [`BoundedSender::try_send`](crate::mpmc::BoundedSender::try_send) or /// [`BoundedSender::try_reserve`](crate::mpmc::BoundedSender::try_reserve). @@ -101,7 +102,7 @@ impl fmt::Debug for TrySendError { } } -impl std::error::Error for TrySendError {} +impl Error for TrySendError {} /// Error returned when receiving a value fails. #[derive(Debug, Clone, PartialEq, Eq)] @@ -116,7 +117,7 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} /// Error returned by a non-blocking attempt to receive a value. #[derive(Debug, Clone, PartialEq, Eq)] @@ -136,4 +137,4 @@ impl fmt::Display for TryRecvError { } } -impl std::error::Error for TryRecvError {} +impl Error for TryRecvError {} diff --git a/asyncband/src/mpmc/mod.rs b/asyncband/src/mpmc/mod.rs index ebb03a0c..ba484949 100644 --- a/asyncband/src/mpmc/mod.rs +++ b/asyncband/src/mpmc/mod.rs @@ -29,12 +29,12 @@ mod unbounded; pub use self::bounded::BoundedReceiver; pub use self::bounded::BoundedSender; -pub use self::bounded::Permit; pub use self::bounded::bounded; pub use self::error::RecvError; pub use self::error::SendError; pub use self::error::TryRecvError; pub use self::error::TrySendError; +pub use self::queue::Permit; pub use self::unbounded::UnboundedReceiver; pub use self::unbounded::UnboundedSender; pub use self::unbounded::unbounded; diff --git a/asyncband/src/mpmc/queue.rs b/asyncband/src/mpmc/queue.rs index 211bab46..02bb9d7a 100644 --- a/asyncband/src/mpmc/queue.rs +++ b/asyncband/src/mpmc/queue.rs @@ -23,13 +23,13 @@ use std::task::Context; use std::task::Poll; use std::task::Waker; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; +use crate::mpmc::RecvError; +use crate::mpmc::SendError; +use crate::mpmc::TryRecvError; +use crate::mpmc::TrySendError; pub struct Shared { state: Mutex>, diff --git a/asyncband/src/mpmc/unbounded.rs b/asyncband/src/mpmc/unbounded.rs index 20bf3591..1a9e6bc8 100644 --- a/asyncband/src/mpmc/unbounded.rs +++ b/asyncband/src/mpmc/unbounded.rs @@ -18,11 +18,11 @@ use std::fmt; use std::sync::Arc; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; -use super::queue::Shared; +use crate::mpmc::RecvError; +use crate::mpmc::SendError; +use crate::mpmc::TryRecvError; +use crate::mpmc::TrySendError; +use crate::mpmc::queue::Shared; /// Creates an unbounded multi-producer, multi-consumer queue. /// diff --git a/asyncband/src/mpsc/bounded/receiver.rs b/asyncband/src/mpsc/bounded/receiver.rs index bd4f6110..2d37f898 100644 --- a/asyncband/src/mpsc/bounded/receiver.rs +++ b/asyncband/src/mpsc/bounded/receiver.rs @@ -22,12 +22,12 @@ use std::sync::Arc; use std::task::Context; use std::task::Poll; -use super::State; use crate::internal::mutex::Mutex; use crate::internal::register_waker; use crate::internal::waitlist::WaitList; use crate::mpsc::RecvError; use crate::mpsc::TryRecvError; +use crate::mpsc::bounded::State; /// The receiving endpoint of a bounded mpsc channel. /// diff --git a/asyncband/src/mpsc/bounded/sender.rs b/asyncband/src/mpsc/bounded/sender.rs index 586ec9c1..f90f99bd 100644 --- a/asyncband/src/mpsc/bounded/sender.rs +++ b/asyncband/src/mpsc/bounded/sender.rs @@ -22,13 +22,13 @@ use std::sync::Arc; use std::task::Context; use std::task::Poll; -use super::State; -use super::Waiter; use crate::internal::mutex::Mutex; use crate::internal::register_waker; use crate::internal::waitlist::WaiterId; use crate::mpsc::SendError; use crate::mpsc::TrySendError; +use crate::mpsc::bounded::State; +use crate::mpsc::bounded::Waiter; /// The sending endpoint of a bounded mpsc channel. /// @@ -117,7 +117,9 @@ impl BoundedSender { /// ``` /// # #[tokio::main] /// # async fn main() { - /// let (tx, mut rx) = asyncband::mpsc::bounded(1); + /// use asyncband::mpsc; + /// + /// let (tx, mut rx) = mpsc::bounded(1); /// let permit = tx.reserve().await.unwrap(); /// let message = String::from("constructed after capacity became available"); /// permit.send(message).unwrap(); diff --git a/asyncband/src/mpsc/error.rs b/asyncband/src/mpsc/error.rs index 9737e2f2..e55d801a 100644 --- a/asyncband/src/mpsc/error.rs +++ b/asyncband/src/mpsc/error.rs @@ -16,6 +16,7 @@ // under the License. use std::any::type_name; +use std::error::Error; use std::fmt; /// An attempt to send a message or reserve capacity failed because the receiver has been dropped. @@ -63,7 +64,7 @@ impl fmt::Debug for SendError { } } -impl std::error::Error for SendError {} +impl Error for SendError {} /// An attempt to send or reserve capacity failed. /// @@ -114,7 +115,7 @@ impl fmt::Debug for TrySendError { } } -impl std::error::Error for TrySendError {} +impl Error for TrySendError {} /// An attempt to receive a value failed because the channel is disconnected. #[derive(Debug, Clone, PartialEq, Eq)] @@ -129,7 +130,7 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} /// A non-blocking attempt to receive a value failed. #[derive(Debug, Clone, PartialEq, Eq)] @@ -149,4 +150,4 @@ impl fmt::Display for TryRecvError { } } -impl std::error::Error for TryRecvError {} +impl Error for TryRecvError {} diff --git a/asyncband/src/mpsc/unbounded/buffer.rs b/asyncband/src/mpsc/unbounded/buffer.rs index 371bb2a4..b4c8be21 100644 --- a/asyncband/src/mpsc/unbounded/buffer.rs +++ b/asyncband/src/mpsc/unbounded/buffer.rs @@ -98,11 +98,9 @@ pub fn pop_batch(batch: &mut VecDeque) -> T { #[cfg(test)] mod tests { - use std::collections::VecDeque; + use std::iter; - use super::Buffer; - use super::SEGMENT_BYTES; - use super::pop_batch; + use super::*; fn allocated_bytes(buffer: &Buffer, batch: &VecDeque) -> usize { let slots = batch.capacity() @@ -149,7 +147,7 @@ mod tests { for _ in 0..2 { for value in 0..messages { buffer.push([value as u8; SIZE]); - for segment in std::iter::once(&buffer.writable).chain(&buffer.sealed) { + for segment in iter::once(&buffer.writable).chain(&buffer.sealed) { assert!(segment.capacity() * SIZE <= SEGMENT_BYTES); } } diff --git a/asyncband/src/mpsc/unbounded/mod.rs b/asyncband/src/mpsc/unbounded/mod.rs index 4955c27e..086f1807 100644 --- a/asyncband/src/mpsc/unbounded/mod.rs +++ b/asyncband/src/mpsc/unbounded/mod.rs @@ -21,8 +21,8 @@ use std::sync::Arc; use std::task::Waker; -use self::buffer::Buffer; use crate::internal::mutex::Mutex; +use crate::mpsc::unbounded::buffer::Buffer; mod buffer; mod receiver; diff --git a/asyncband/src/mpsc/unbounded/receiver.rs b/asyncband/src/mpsc/unbounded/receiver.rs index 6874b4c2..6e7009de 100644 --- a/asyncband/src/mpsc/unbounded/receiver.rs +++ b/asyncband/src/mpsc/unbounded/receiver.rs @@ -23,13 +23,13 @@ use std::sync::Arc; use std::task::Context; use std::task::Poll; -use super::State; -use super::buffer::Buffer; -use super::buffer::pop_batch; use crate::internal::mutex::Mutex; use crate::internal::register_waker; use crate::mpsc::RecvError; use crate::mpsc::TryRecvError; +use crate::mpsc::unbounded::State; +use crate::mpsc::unbounded::buffer::Buffer; +use crate::mpsc::unbounded::buffer::pop_batch; /// The receiving endpoint of an unbounded mpsc channel. /// diff --git a/asyncband/src/mpsc/unbounded/sender.rs b/asyncband/src/mpsc/unbounded/sender.rs index a90589c5..ab353031 100644 --- a/asyncband/src/mpsc/unbounded/sender.rs +++ b/asyncband/src/mpsc/unbounded/sender.rs @@ -18,9 +18,9 @@ use std::fmt; use std::sync::Arc; -use super::State; use crate::internal::mutex::Mutex; use crate::mpsc::SendError; +use crate::mpsc::unbounded::State; /// The sending endpoint of an unbounded mpsc channel. /// diff --git a/asyncband/src/mutex/mod.rs b/asyncband/src/mutex/mod.rs index ffed4559..5469d492 100644 --- a/asyncband/src/mutex/mod.rs +++ b/asyncband/src/mutex/mod.rs @@ -69,6 +69,7 @@ use std::marker::PhantomData; use std::mem::ManuallyDrop; use std::ops::Deref; use std::ops::DerefMut; +use std::ptr; use std::ptr::NonNull; use std::sync::Arc; @@ -372,17 +373,19 @@ impl<'a, T: ?Sized> MutexGuard<'a, T> { /// ``` /// # #[tokio::main] /// # async fn main() { + /// use std::collections::HashMap; + /// /// use asyncband::mutex::Mutex; /// use asyncband::mutex::MutexGuard; /// /// #[derive(Debug)] /// struct Database { - /// users: std::collections::HashMap, + /// users: HashMap, /// admin_user_id: Option, /// } /// /// let mut db = Database { - /// users: std::collections::HashMap::new(), + /// users: HashMap::new(), /// admin_user_id: Some(1), /// }; /// db.users.insert(1, "admin@example.com".to_owned()); @@ -516,7 +519,7 @@ impl OwnedMutexGuard { // SAFETY: The guard is wrapped in `ManuallyDrop` and will not be dropped, // so the `Arc` can be moved out to transfer lock ownership to the new guard. - let lock = unsafe { std::ptr::read(&guard.lock) }; + let lock = unsafe { ptr::read(&guard.lock) }; OwnedMappedMutexGuard { lock, @@ -563,7 +566,7 @@ impl OwnedMutexGuard { // SAFETY: The guard is wrapped in `ManuallyDrop` and will not be dropped, // so the `Arc` can be moved out to transfer lock ownership to the new guard. - let lock = unsafe { std::ptr::read(&guard.lock) }; + let lock = unsafe { ptr::read(&guard.lock) }; Ok(OwnedMappedMutexGuard { lock, @@ -920,7 +923,7 @@ impl OwnedMappedMutexGuard { // SAFETY: The guard is wrapped in `ManuallyDrop` and will not be dropped, // so the `Arc` can be moved out to transfer lock ownership to the new guard. - let lock = unsafe { std::ptr::read(&orig.lock) }; + let lock = unsafe { ptr::read(&orig.lock) }; OwnedMappedMutexGuard { lock, @@ -989,7 +992,7 @@ impl OwnedMappedMutexGuard { // SAFETY: The guard is wrapped in `ManuallyDrop` and will not be dropped, // so the `Arc` can be moved out to transfer lock ownership to the new guard. - let lock = unsafe { std::ptr::read(&orig.lock) }; + let lock = unsafe { ptr::read(&orig.lock) }; Ok(OwnedMappedMutexGuard { lock, diff --git a/asyncband/src/once/lazy_cell/mod.rs b/asyncband/src/once/lazy_cell/mod.rs index 25c49528..a0418765 100644 --- a/asyncband/src/once/lazy_cell/mod.rs +++ b/asyncband/src/once/lazy_cell/mod.rs @@ -19,11 +19,14 @@ use std::fmt; use std::future::Future; use std::future::Pending; use std::future::Ready; +use std::future::poll_fn; +use std::future::ready; use std::panic::RefUnwindSafe; use std::panic::UnwindSafe; use std::pin::Pin; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; +use std::thread; use crate::internal::value_cell::ValueCell; use crate::mutex::Mutex; @@ -131,10 +134,12 @@ impl LazyCell { /// ``` /// # #[tokio::main] /// # async fn main() { + /// use std::pin::pin; + /// /// use asyncband::once::LazyCell; /// /// let lazy = LazyCell::new(async || 92); - /// let lazy = std::pin::pin!(lazy); + /// let lazy = pin!(lazy); /// assert_eq!(*LazyCell::force_pin(lazy.as_ref()).await, 92); /// # } /// ``` @@ -372,7 +377,7 @@ where *unsafe { state.as_mut().get_unchecked_mut() } = State::Attempt(future); } - let value = std::future::poll_fn(|cx| { + let value = poll_fn(|cx| { let _poison = PoisonOnPanic(poisoned); // SAFETY: The state remains pinned while this future is polled, and the attempt is never // moved after it is installed. @@ -401,7 +406,7 @@ where { fn default() -> Self { fn initialize() -> Ready { - std::future::ready(T::default()) + ready(T::default()) } Self::new(initialize::) @@ -464,7 +469,7 @@ struct PoisonOnPanic<'a>(&'a AtomicBool); impl Drop for PoisonOnPanic<'_> { fn drop(&mut self) { - if std::thread::panicking() { + if thread::panicking() { self.0.store(true, Ordering::Relaxed); } } diff --git a/asyncband/src/once/once_map/tests.rs b/asyncband/src/once/once_map/tests.rs index fe25d1cf..d89ddacc 100644 --- a/asyncband/src/once/once_map/tests.rs +++ b/asyncband/src/once/once_map/tests.rs @@ -16,17 +16,16 @@ // under the License. use std::cell::Cell; +use std::future::pending; use std::future::poll_fn; use std::hash::BuildHasherDefault; use std::hash::Hasher; use std::panic::AssertUnwindSafe; use std::panic::catch_unwind; use std::pin::pin; -use std::sync::Arc; use std::task::Poll; -use super::Lookup; -use super::OnceMap; +use super::*; use crate::test_support::poll_once; // These tests stay next to the implementation because they inspect private state. @@ -78,7 +77,7 @@ fn cancelled_compute_removes_empty_entry() { let map = OnceMap::<&str, i32>::new(); { - let mut compute = pin!(map.compute("key", async || std::future::pending::().await)); + let mut compute = pin!(map.compute("key", async || pending::().await)); assert!(poll_once(compute.as_mut()).is_pending()); assert_eq!(map.len(), 1); } @@ -90,13 +89,11 @@ fn cancelled_compute_removes_empty_entry() { fn pending_computation_does_not_block_another_key() { let map = OnceMap::new(); { - let mut pending = std::pin::pin!( - map.compute("pending", async || { std::future::pending::().await }) - ); + let mut pending = pin!(map.compute("pending", async || { pending::().await })); assert!(poll_once(pending.as_mut()).is_pending()); - let mut ready = std::pin::pin!(map.compute("ready", async || 1)); - assert_eq!(poll_once(ready.as_mut()), std::task::Poll::Ready(1)); + let mut ready = pin!(map.compute("ready", async || 1)); + assert_eq!(poll_once(ready.as_mut()), Poll::Ready(1)); } assert_eq!(map.len(), 1); diff --git a/asyncband/src/oneshot/receiver.rs b/asyncband/src/oneshot/receiver.rs index 9f0228b3..abe7279d 100644 --- a/asyncband/src/oneshot/receiver.rs +++ b/asyncband/src/oneshot/receiver.rs @@ -22,6 +22,7 @@ // https://github.com/faern/oneshot/blob/25274e995ee0a702b3e9e1ac81e577f8c3ce0892/src/lib.rs // https://github.com/faern/oneshot/blob/25274e995ee0a702b3e9e1ac81e577f8c3ce0892/src/errors.rs +use std::error::Error; use std::fmt; use std::mem::ManuallyDrop; use std::pin::Pin; @@ -420,7 +421,7 @@ impl fmt::Display for TryRecvError { } } -impl std::error::Error for TryRecvError {} +impl Error for TryRecvError {} /// An error returned when awaiting the message via [`Receiver`]. /// @@ -440,4 +441,4 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} diff --git a/asyncband/src/oneshot/sender.rs b/asyncband/src/oneshot/sender.rs index 738c4dda..74d004bf 100644 --- a/asyncband/src/oneshot/sender.rs +++ b/asyncband/src/oneshot/sender.rs @@ -23,6 +23,7 @@ // https://github.com/faern/oneshot/blob/25274e995ee0a702b3e9e1ac81e577f8c3ce0892/src/errors.rs use std::any::type_name; +use std::error::Error; use std::fmt; use std::mem::ManuallyDrop; use std::ptr::NonNull; @@ -275,7 +276,7 @@ impl fmt::Debug for SendError { } } -impl std::error::Error for SendError {} +impl Error for SendError {} #[cfg(test)] impl Sender { diff --git a/asyncband/src/oneshot/tests.rs b/asyncband/src/oneshot/tests.rs index 4e293261..d6a9da67 100644 --- a/asyncband/src/oneshot/tests.rs +++ b/asyncband/src/oneshot/tests.rs @@ -26,17 +26,11 @@ use std::future::Future; use std::future::IntoFuture; use std::mem; use std::pin::Pin; -use std::sync::atomic::Ordering; use std::sync::mpsc; use std::task::Context; -use std::task::Poll; use std::time::Duration; -use crate::oneshot; -use crate::oneshot::Channel; -use crate::oneshot::MESSAGE; -use crate::oneshot::RECEIVING; -use crate::oneshot::drop_message_and_deallocate_channel; +use super::*; use crate::oneshot::tests::support::DropProbe; use crate::oneshot::tests::support::WakerProbe; use crate::oneshot::tests::support::spawn_named; @@ -45,7 +39,7 @@ use crate::oneshot::tests::support::spawn_named; #[test] fn poll_returns_while_sender_owns_waker() { - let (sender, receiver) = oneshot::channel(); + let (sender, receiver) = channel(); let channel_ptr = sender.channel_ptr(); mem::forget(sender); let mut receiver = receiver.into_future(); @@ -109,7 +103,7 @@ fn poll_returns_while_sender_owns_waker() { #[test] fn drop_transfers_cleanup_while_sender_owns_waker() { - let (sender, receiver) = oneshot::channel(); + let (sender, receiver) = channel(); let channel_ptr = sender.channel_ptr(); mem::forget(sender); let mut receiver = receiver.into_future(); @@ -177,6 +171,7 @@ mod support { use std::sync::atomic::Ordering; use std::task::Wake; use std::task::Waker; + use std::thread; pub(super) struct DropProbe { drop_count: Arc, @@ -234,12 +229,12 @@ mod support { } } - pub(super) fn spawn_named(name: &str, f: F) -> std::thread::JoinHandle + pub(super) fn spawn_named(name: &str, f: F) -> thread::JoinHandle where F: FnOnce() -> T + Send + 'static, T: Send + 'static, { - std::thread::Builder::new() + thread::Builder::new() .name(name.to_owned()) .spawn(f) .unwrap() diff --git a/asyncband/src/phaser/mod.rs b/asyncband/src/phaser/mod.rs index b96cba5f..60c0f40e 100644 --- a/asyncband/src/phaser/mod.rs +++ b/asyncband/src/phaser/mod.rs @@ -121,6 +121,7 @@ //! from the same phaser to `wait`; it tests for a different phase, not a target number //! or numeric threshold. An observation must not be retained across a full counter cycle. +use std::error::Error; use std::fmt; use std::future::Future; use std::iter::FusedIterator; @@ -150,7 +151,7 @@ impl fmt::Display for Closed { } } -impl std::error::Error for Closed {} +impl Error for Closed {} /// A shared handle to a reusable phase barrier with dynamic participants. /// diff --git a/asyncband/src/phaser/tests.rs b/asyncband/src/phaser/tests.rs index 48cffa89..b85b7fcc 100644 --- a/asyncband/src/phaser/tests.rs +++ b/asyncband/src/phaser/tests.rs @@ -15,9 +15,7 @@ // specific language governing permissions and limitations // under the License. -use std::task::Poll; - -use super::Phaser; +use super::*; use crate::test_support::poll_once; #[test] diff --git a/asyncband/src/pool/bounded.rs b/asyncband/src/pool/bounded.rs index 9b3e2fd5..1a5f9de6 100644 --- a/asyncband/src/pool/bounded.rs +++ b/asyncband/src/pool/bounded.rs @@ -73,6 +73,7 @@ //! # } //! ``` +use std::fmt; use std::ops::Deref; use std::ops::DerefMut; use std::sync::Arc; @@ -164,12 +165,12 @@ pub struct Pool { slots: Mutex>, } -impl std::fmt::Debug for Pool +impl fmt::Debug for Pool where M: ManageObject, - M::Object: std::fmt::Debug, + M::Object: fmt::Debug, { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Pool") .field("slots", &self.slots) .field("config", &self.config) @@ -476,12 +477,12 @@ pub struct Object { pool: Weak>, } -impl std::fmt::Debug for Object +impl fmt::Debug for Object where M: ManageObject, - M::Object: std::fmt::Debug, + M::Object: fmt::Debug, { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Object") .field("state", &self.state) .field("permit", &self.permit) diff --git a/asyncband/src/pool/mod.rs b/asyncband/src/pool/mod.rs index 8895e941..c4068fa2 100644 --- a/asyncband/src/pool/mod.rs +++ b/asyncband/src/pool/mod.rs @@ -180,13 +180,15 @@ //! For example, create and recycle behavior can be expressed directly by [`ManageObject`]: //! //! ``` +//! use std::convert::Infallible; +//! //! use asyncband::pool::ManageObject; //! use asyncband::pool::ObjectStatus; //! //! struct Manager; //! impl ManageObject for Manager { //! type Object = i32; -//! type Error = std::convert::Infallible; +//! type Error = Infallible; //! //! async fn create(&self) -> Result { //! let o = 42; diff --git a/asyncband/src/pool/unbounded.rs b/asyncband/src/pool/unbounded.rs index 6f7337ce..e4478433 100644 --- a/asyncband/src/pool/unbounded.rs +++ b/asyncband/src/pool/unbounded.rs @@ -88,7 +88,10 @@ //! # } //! ``` +use std::error::Error; +use std::fmt; use std::future::Future; +use std::future::ready; use std::marker::PhantomData; use std::ops::Deref; use std::ops::DerefMut; @@ -180,8 +183,8 @@ impl Clone for NeverManageObject { } } -impl std::fmt::Debug for NeverManageObject { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl fmt::Debug for NeverManageObject { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("NeverManageObject") } } @@ -197,26 +200,26 @@ impl Default for NeverManageObject { /// The error returned by [`NeverManageObject::create`]. pub struct PoolIsEmpty(()); -impl std::fmt::Debug for PoolIsEmpty { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl fmt::Debug for PoolIsEmpty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("unbounded pool is empty") } } -impl std::fmt::Display for PoolIsEmpty { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { +impl fmt::Display for PoolIsEmpty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str("unbounded pool is empty") } } -impl std::error::Error for PoolIsEmpty {} +impl Error for PoolIsEmpty {} impl ManageObject for NeverManageObject { type Object = T; type Error = PoolIsEmpty; fn create(&self) -> impl Future> + Send { - std::future::ready(Err(PoolIsEmpty(()))) + ready(Err(PoolIsEmpty(()))) } fn is_recyclable( @@ -224,7 +227,7 @@ impl ManageObject for NeverManageObject { _: &mut Self::Object, _: &ObjectStatus, ) -> impl Future> + Send { - std::future::ready(Ok(())) + ready(Ok(())) } } @@ -239,12 +242,12 @@ pub struct Pool = NeverManageObject> { slots: Mutex>, } -impl std::fmt::Debug for Pool +impl fmt::Debug for Pool where - T: std::fmt::Debug, + T: fmt::Debug, M: ManageObject, { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Pool") .field("slots", &self.slots) .field("config", &self.config) @@ -492,12 +495,12 @@ pub struct Object = NeverManageObject> { pool: Weak>, } -impl std::fmt::Debug for Object +impl fmt::Debug for Object where - T: std::fmt::Debug, + T: fmt::Debug, M: ManageObject, { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Object") .field("state", &self.state) .finish() diff --git a/asyncband/src/rwlock/access.rs b/asyncband/src/rwlock/access.rs index fd9637e4..1e92587b 100644 --- a/asyncband/src/rwlock/access.rs +++ b/asyncband/src/rwlock/access.rs @@ -20,6 +20,7 @@ //! Guards hold a token instead of implementing `Drop`, so projecting or downgrading a guard moves //! the token into the new guard. +use std::mem; use std::sync::Arc; use crate::internal::semaphore::Semaphore; @@ -72,7 +73,7 @@ impl<'a, T: ?Sized> ReadAccess<&'a RwLock> { let lock = self.owner; // The new token takes over releasing the permit; forgetting this one avoids releasing it // during the transfer. Its borrowed owner needs no cleanup. - std::mem::forget(self); + mem::forget(self); ReadAccess::new(&lock.s) } } @@ -119,7 +120,7 @@ impl<'a, T: ?Sized> WriteAccess<&'a RwLock> { let permits_acquired = self.permits_acquired; // The new token takes over releasing the permits; forgetting this one avoids releasing them // during the transfer. Its borrowed owner needs no cleanup. - std::mem::forget(self); + mem::forget(self); WriteAccess::new(&lock.s, permits_acquired) } } diff --git a/asyncband/src/rwlock/owned_read_guard.rs b/asyncband/src/rwlock/owned_read_guard.rs index 7424c70e..8eb4a62e 100644 --- a/asyncband/src/rwlock/owned_read_guard.rs +++ b/asyncband/src/rwlock/owned_read_guard.rs @@ -24,6 +24,7 @@ use std::fmt; use std::ops::Deref; +use std::ptr::NonNull; use std::sync::Arc; use crate::rwlock::OwnedMappedRwLockReadGuard; @@ -101,7 +102,7 @@ impl OwnedRwLockReadGuard { F: FnOnce(&T) -> &U, U: ?Sized, { - let d = std::ptr::NonNull::from(f(&*orig)); + let d = NonNull::from(f(&*orig)); OwnedMappedRwLockReadGuard::new(d, orig.access) } @@ -117,7 +118,7 @@ impl OwnedRwLockReadGuard { { match f(&*orig) { Some(d) => { - let d = std::ptr::NonNull::from(d); + let d = NonNull::from(d); Ok(OwnedMappedRwLockReadGuard::new(d, orig.access)) } None => Err(orig), diff --git a/asyncband/src/semaphore/mod.rs b/asyncband/src/semaphore/mod.rs index 876742cd..93340e31 100644 --- a/asyncband/src/semaphore/mod.rs +++ b/asyncband/src/semaphore/mod.rs @@ -51,6 +51,7 @@ //! # } //! ``` +use std::ptr; use std::sync::Arc; use crate::internal::semaphore; @@ -399,7 +400,7 @@ impl SemaphorePermit<'_> { #[track_caller] pub fn merge(&mut self, mut other: Self) { assert!( - std::ptr::eq(self.sem, other.sem), + ptr::eq(self.sem, other.sem), "merging permits from different semaphore instances" ); self.permits = self diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs index f5307500..8dcaab22 100644 --- a/asyncband/src/shutdown/mod.rs +++ b/asyncband/src/shutdown/mod.rs @@ -38,7 +38,9 @@ //! ``` //! # #[tokio::main] //! # async fn main() { -//! let (shutdown, guard) = asyncband::shutdown::new(); +//! use asyncband::shutdown; +//! +//! let (shutdown, guard) = shutdown::new(); //! let mut tasks = vec![]; //! //! for _ in 0..3 { @@ -101,7 +103,11 @@ pub fn new() -> (Shutdown, ShutdownGuard) { /// ``` /// # #[tokio::main] /// # async fn main() { -/// let (shutdown, guard) = asyncband::shutdown::new(); +/// use std::future::pending; +/// +/// use asyncband::shutdown; +/// +/// let (shutdown, guard) = shutdown::new(); /// let worker = tokio::spawn(async move { /// guard.shutdown_requested().await; /// }); @@ -110,7 +116,7 @@ pub fn new() -> (Shutdown, ShutdownGuard) { /// /// tokio::select! { /// _ = shutdown => {} -/// _ = std::future::pending::<()>() => {} +/// _ = pending::<()>() => {} /// } /// /// worker.await.unwrap(); diff --git a/asyncband/src/singleflight/tests.rs b/asyncband/src/singleflight/tests.rs index 123bfa43..d97ad5b6 100644 --- a/asyncband/src/singleflight/tests.rs +++ b/asyncband/src/singleflight/tests.rs @@ -16,13 +16,14 @@ // under the License. use std::cell::Cell; +use std::future::pending; use std::future::poll_fn; use std::panic::AssertUnwindSafe; use std::panic::catch_unwind; use std::pin::pin; use std::task::Poll; -use super::Group; +use super::*; use crate::test_support::poll_once; // These tests stay next to the implementation because they inspect private state. @@ -54,7 +55,7 @@ fn cancelled_work_removes_empty_entry() { let group = Group::<&str, &str>::new(); { - let mut work = pin!(group.work("key", || async { std::future::pending::<&str>().await })); + let mut work = pin!(group.work("key", || async { pending::<&str>().await })); assert!(poll_once(work.as_mut()).is_pending()); assert_eq!(entry_count(&group), 1); } diff --git a/asyncband/src/spmc/bounded.rs b/asyncband/src/spmc/bounded.rs index b4d7e529..fad902a3 100644 --- a/asyncband/src/spmc/bounded.rs +++ b/asyncband/src/spmc/bounded.rs @@ -18,11 +18,11 @@ use std::fmt; use std::sync::Arc; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; -use super::queue::Shared; +use crate::spmc::RecvError; +use crate::spmc::SendError; +use crate::spmc::TryRecvError; +use crate::spmc::TrySendError; +use crate::spmc::queue::Shared; /// Creates a bounded single-producer, multi-consumer queue. /// diff --git a/asyncband/src/spmc/error.rs b/asyncband/src/spmc/error.rs index fe63ad2a..09a680dc 100644 --- a/asyncband/src/spmc/error.rs +++ b/asyncband/src/spmc/error.rs @@ -16,6 +16,7 @@ // under the License. use std::any::type_name; +use std::error::Error; use std::fmt; /// An error returned when trying to send on a disconnected queue. @@ -52,7 +53,7 @@ impl fmt::Debug for SendError { } } -impl std::error::Error for SendError {} +impl Error for SendError {} /// Error returned by [`BoundedSender::try_send`](crate::spmc::BoundedSender::try_send). #[derive(Clone, PartialEq, Eq)] @@ -100,7 +101,7 @@ impl fmt::Debug for TrySendError { } } -impl std::error::Error for TrySendError {} +impl Error for TrySendError {} /// Error returned when receiving a value fails. #[derive(Debug, Clone, PartialEq, Eq)] @@ -115,7 +116,7 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} /// Error returned by a non-blocking attempt to receive a value. #[derive(Debug, Clone, PartialEq, Eq)] @@ -135,4 +136,4 @@ impl fmt::Display for TryRecvError { } } -impl std::error::Error for TryRecvError {} +impl Error for TryRecvError {} diff --git a/asyncband/src/spmc/mod.rs b/asyncband/src/spmc/mod.rs index dd79fbe9..eed5ace4 100644 --- a/asyncband/src/spmc/mod.rs +++ b/asyncband/src/spmc/mod.rs @@ -52,25 +52,33 @@ //! Neither sender supports cloning: //! //! ```compile_fail,E0599 -//! let (sender, _receiver) = asyncband::spmc::bounded::(1); +//! use asyncband::spmc; +//! +//! let (sender, _receiver) = spmc::bounded::(1); //! let second_producer = sender.clone(); //! ``` //! //! ```compile_fail,E0599 -//! let (sender, _receiver) = asyncband::spmc::unbounded::(); +//! use asyncband::spmc; +//! +//! let (sender, _receiver) = spmc::unbounded::(); //! let second_producer = sender.clone(); //! ``` //! //! Sending through a shared reference is rejected: //! //! ```compile_fail,E0596 -//! fn send(sender: &asyncband::spmc::BoundedSender) { +//! use asyncband::spmc::BoundedSender; +//! +//! fn send(sender: &BoundedSender) { //! let _ = sender.try_send(1); //! } //! ``` //! //! ```compile_fail,E0596 -//! fn send(sender: &asyncband::spmc::UnboundedSender) { +//! use asyncband::spmc::UnboundedSender; +//! +//! fn send(sender: &UnboundedSender) { //! let _ = sender.send(1); //! } //! ``` @@ -78,7 +86,9 @@ //! A bounded `send` future retains the exclusive borrow until completion or cancellation: //! //! ```compile_fail,E0499 -//! let (mut sender, _receiver) = asyncband::spmc::bounded(1); +//! use asyncband::spmc; +//! +//! let (mut sender, _receiver) = spmc::bounded(1); //! let pending = sender.send(1); //! let _ = sender.try_send(2); //! drop(pending); diff --git a/asyncband/src/spmc/queue.rs b/asyncband/src/spmc/queue.rs index ad47bddf..9c8ce74d 100644 --- a/asyncband/src/spmc/queue.rs +++ b/asyncband/src/spmc/queue.rs @@ -22,13 +22,13 @@ use std::task::Context; use std::task::Poll; use std::task::Waker; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; +use crate::spmc::RecvError; +use crate::spmc::SendError; +use crate::spmc::TryRecvError; +use crate::spmc::TrySendError; pub struct Shared { state: Mutex>, diff --git a/asyncband/src/spmc/unbounded.rs b/asyncband/src/spmc/unbounded.rs index b2b4851e..b5bb0b4b 100644 --- a/asyncband/src/spmc/unbounded.rs +++ b/asyncband/src/spmc/unbounded.rs @@ -18,11 +18,11 @@ use std::fmt; use std::sync::Arc; -use super::RecvError; -use super::SendError; -use super::TryRecvError; -use super::TrySendError; -use super::queue::Shared; +use crate::spmc::RecvError; +use crate::spmc::SendError; +use crate::spmc::TryRecvError; +use crate::spmc::TrySendError; +use crate::spmc::queue::Shared; /// Creates an unbounded single-producer, multi-consumer queue. /// diff --git a/asyncband/src/watch/error.rs b/asyncband/src/watch/error.rs index 017d6820..7f7121f7 100644 --- a/asyncband/src/watch/error.rs +++ b/asyncband/src/watch/error.rs @@ -16,6 +16,7 @@ // under the License. use std::any::type_name; +use std::error::Error; use std::fmt; /// An error returned when sending on a watch channel without any receivers. @@ -52,7 +53,7 @@ impl fmt::Debug for SendError { } } -impl std::error::Error for SendError {} +impl Error for SendError {} /// An error returned after all senders have been dropped and no unseen value remains. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -67,4 +68,4 @@ impl fmt::Display for RecvError { } } -impl std::error::Error for RecvError {} +impl Error for RecvError {} diff --git a/asyncband/src/watch/tests.rs b/asyncband/src/watch/tests.rs index 6df8bb6f..27bfaf78 100644 --- a/asyncband/src/watch/tests.rs +++ b/asyncband/src/watch/tests.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::thread; + use super::*; #[test] @@ -42,7 +44,7 @@ fn concurrent_senders_commit_every_version() { let workers = (0..SENDERS) .map(|sender_index| { let tx = tx.clone(); - std::thread::spawn(move || { + thread::spawn(move || { for offset in 0..SENDS_PER_THREAD { tx.send(sender_index * SENDS_PER_THREAD + offset).unwrap(); } diff --git a/benchmarks/benches/ecosystem/broadcast/mpmc/bounded.rs b/benchmarks/benches/ecosystem/broadcast/mpmc/bounded.rs index 88dba15c..262b4567 100644 --- a/benchmarks/benches/ecosystem/broadcast/mpmc/bounded.rs +++ b/benchmarks/benches/ecosystem/broadcast/mpmc/bounded.rs @@ -17,21 +17,23 @@ // Tokio broadcast overwrites at capacity, so it cannot participate in lossless backpressure cases. +use benchmarks::channels::runtime; use benchmarks::support::bench_context; use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; +use tokio::runtime::Builder; -use super::adapters::AsyncBroadcast; -use super::adapters::Asyncband; -use super::adapters::BoundedBroadcastMpmc; -use super::support::BACKPRESSURE_SHAPE; -use super::support::BATCH_MESSAGES; -use super::support::BOUNDED_SHAPES; -use super::support::BoundedConcurrent; -use super::support::BoundedShape; -use super::support::BoundedTasks; -use super::support::ROUND_TRIP_CAPACITY; +use crate::broadcast::mpmc::adapters::AsyncBroadcast; +use crate::broadcast::mpmc::adapters::Asyncband; +use crate::broadcast::mpmc::adapters::BoundedBroadcastMpmc; +use crate::broadcast::mpmc::support::BACKPRESSURE_SHAPE; +use crate::broadcast::mpmc::support::BATCH_MESSAGES; +use crate::broadcast::mpmc::support::BOUNDED_SHAPES; +use crate::broadcast::mpmc::support::BoundedConcurrent; +use crate::broadcast::mpmc::support::BoundedShape; +use crate::broadcast::mpmc::support::BoundedTasks; +use crate::broadcast::mpmc::support::ROUND_TRIP_CAPACITY; // Send-then-receive pairing keeps at most one message retained, so these never reach capacity. #[divan::bench(types = [Asyncband, AsyncBroadcast], sample_size = 512)] @@ -79,7 +81,7 @@ fn thread_batch(bencher: Bencher, shape: BoundedShape) counter = ItemsCount::new(BATCH_MESSAGES), )] fn task_batch(bencher: Bencher, shape: BoundedShape) { - let runtime = tokio::runtime::Builder::new_multi_thread() + let runtime = Builder::new_multi_thread() .worker_threads(4) .build() .unwrap(); @@ -91,7 +93,7 @@ fn task_batch(bencher: Bencher, shape: BoundedShape) { #[divan::bench(types = [Asyncband, AsyncBroadcast], sample_count = 10, sample_size = 1, counter = ItemsCount::new(BATCH_MESSAGES))] fn capacity_one_task_handoff(bencher: Bencher) { - let runtime = benchmarks::channels::runtime(4); + let runtime = runtime(4); bencher .with_inputs(|| BoundedTasks::new::(&runtime, BACKPRESSURE_SHAPE)) .bench_local_refs(|tasks| tasks.run(&runtime)); diff --git a/benchmarks/benches/ecosystem/broadcast/mpmc/no_backpressure.rs b/benchmarks/benches/ecosystem/broadcast/mpmc/no_backpressure.rs index c5a40722..1073fa4d 100644 --- a/benchmarks/benches/ecosystem/broadcast/mpmc/no_backpressure.rs +++ b/benchmarks/benches/ecosystem/broadcast/mpmc/no_backpressure.rs @@ -22,16 +22,16 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::adapters::AsyncBroadcast; -use super::adapters::Asyncband; -use super::adapters::BroadcastMpmc; -use super::adapters::Tokio; -use super::support::BATCH_MESSAGES; -use super::support::ConcurrentSend; -use super::support::Fanout; -use super::support::PRODUCER_COUNTS; -use super::support::RECEIVER_COUNTS; -use super::support::ROUND_TRIP_CAPACITY; +use crate::broadcast::mpmc::adapters::AsyncBroadcast; +use crate::broadcast::mpmc::adapters::Asyncband; +use crate::broadcast::mpmc::adapters::BroadcastMpmc; +use crate::broadcast::mpmc::adapters::Tokio; +use crate::broadcast::mpmc::support::BATCH_MESSAGES; +use crate::broadcast::mpmc::support::ConcurrentSend; +use crate::broadcast::mpmc::support::Fanout; +use crate::broadcast::mpmc::support::PRODUCER_COUNTS; +use crate::broadcast::mpmc::support::RECEIVER_COUNTS; +use crate::broadcast::mpmc::support::ROUND_TRIP_CAPACITY; #[divan::bench(types = [Asyncband, Tokio, AsyncBroadcast])] fn try_send_receive(bencher: Bencher) { diff --git a/benchmarks/benches/ecosystem/broadcast/mpmc/support.rs b/benchmarks/benches/ecosystem/broadcast/mpmc/support.rs index 1d07fd9e..17f1133e 100644 --- a/benchmarks/benches/ecosystem/broadcast/mpmc/support.rs +++ b/benchmarks/benches/ecosystem/broadcast/mpmc/support.rs @@ -26,8 +26,8 @@ use divan::black_box; use tokio::runtime::Runtime; use tokio::task::JoinSet; -use super::adapters::BoundedBroadcastMpmc; -use super::adapters::BroadcastMpmc; +use crate::broadcast::mpmc::adapters::BoundedBroadcastMpmc; +use crate::broadcast::mpmc::adapters::BroadcastMpmc; pub const BATCH_MESSAGES: usize = 4096; pub const PRODUCER_COUNTS: &[usize] = &[1, 2, 4, 8]; diff --git a/benchmarks/benches/ecosystem/mpsc/bounded.rs b/benchmarks/benches/ecosystem/mpsc/bounded.rs index 4c28cd1b..8d324f3f 100644 --- a/benchmarks/benches/ecosystem/mpsc/bounded.rs +++ b/benchmarks/benches/ecosystem/mpsc/bounded.rs @@ -20,20 +20,20 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::adapters::AsyncChannel; -use super::adapters::Asyncband; -use super::adapters::BoundedMpsc; -use super::adapters::Flume; -use super::adapters::Tokio; -use super::support::BATCH_MESSAGES; -use super::support::BOUNDED_CAPACITY; -use super::support::Bounded; -use super::support::PARALLEL_TASK_SHAPES; -use super::support::PRODUCER_COUNTS; -use super::support::RepeatedBatch; -use super::support::RepeatedTasks; -use super::support::TASK_SHAPES; -use super::support::TaskShape; +use crate::mpsc::adapters::AsyncChannel; +use crate::mpsc::adapters::Asyncband; +use crate::mpsc::adapters::BoundedMpsc; +use crate::mpsc::adapters::Flume; +use crate::mpsc::adapters::Tokio; +use crate::mpsc::support::BATCH_MESSAGES; +use crate::mpsc::support::BOUNDED_CAPACITY; +use crate::mpsc::support::Bounded; +use crate::mpsc::support::PARALLEL_TASK_SHAPES; +use crate::mpsc::support::PRODUCER_COUNTS; +use crate::mpsc::support::RepeatedBatch; +use crate::mpsc::support::RepeatedTasks; +use crate::mpsc::support::TASK_SHAPES; +use crate::mpsc::support::TaskShape; // Fixed sample sizes keep cold-start costs from skewing Divan's iteration count. #[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Flume], sample_size = 512)] diff --git a/benchmarks/benches/ecosystem/mpsc/reservation.rs b/benchmarks/benches/ecosystem/mpsc/reservation.rs index 2f4502a9..fbbd8637 100644 --- a/benchmarks/benches/ecosystem/mpsc/reservation.rs +++ b/benchmarks/benches/ecosystem/mpsc/reservation.rs @@ -25,14 +25,14 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::adapters::Asyncband; -use super::adapters::BoundedMpsc; -use super::adapters::Tokio; -use super::support::BATCH_MESSAGES; -use super::support::ConcurrentMpsc; -use super::support::RepeatedTasks; -use super::support::TASK_SHAPES; -use super::support::TaskShape; +use crate::mpsc::adapters::Asyncband; +use crate::mpsc::adapters::BoundedMpsc; +use crate::mpsc::adapters::Tokio; +use crate::mpsc::support::BATCH_MESSAGES; +use crate::mpsc::support::ConcurrentMpsc; +use crate::mpsc::support::RepeatedTasks; +use crate::mpsc::support::TASK_SHAPES; +use crate::mpsc::support::TaskShape; trait Reservable: BoundedMpsc { type Permit<'a>: Send; diff --git a/benchmarks/benches/ecosystem/mpsc/support.rs b/benchmarks/benches/ecosystem/mpsc/support.rs index 37453f00..14dde891 100644 --- a/benchmarks/benches/ecosystem/mpsc/support.rs +++ b/benchmarks/benches/ecosystem/mpsc/support.rs @@ -28,9 +28,13 @@ use std::thread::JoinHandle; pub use benchmarks::channels::BATCH_MESSAGES; pub use benchmarks::channels::BOUNDED_CAPACITY; use divan::black_box; +use tokio::runtime::Builder; +use tokio::runtime::Runtime; +use tokio::sync::Notify; +use tokio::sync::mpsc; -use super::adapters::BoundedMpsc; -use super::adapters::UnboundedMpsc; +use crate::mpsc::adapters::BoundedMpsc; +use crate::mpsc::adapters::UnboundedMpsc; pub const PRODUCER_COUNTS: &[usize] = &[1, 4]; #[derive(Clone, Copy)] @@ -248,15 +252,15 @@ impl> Drop for RepeatedBatch { // block_on instead measures worker-to-caller thread handoffs, which is a separate workload. enum Receiver { Task { - start: Arc, - completed: tokio::sync::mpsc::UnboundedReceiver, + start: Arc, + completed: mpsc::UnboundedReceiver, }, External(C::Receiver), } // Control messages coordinate each batch without forwarding measured payloads. pub struct RepeatedTasks { - runtime: tokio::runtime::Runtime, + runtime: Runtime, receiver: Receiver, start: Arc, stop: Arc, @@ -279,11 +283,9 @@ impl RepeatedTasks { ) -> Self { assert_eq!(BATCH_MESSAGES % producer_count, 0); let runtime = if worker_threads == 0 { - tokio::runtime::Builder::new_current_thread() - .build() - .unwrap() + Builder::new_current_thread().build().unwrap() } else { - tokio::runtime::Builder::new_multi_thread() + Builder::new_multi_thread() .worker_threads(worker_threads) .build() .unwrap() @@ -316,8 +318,8 @@ impl RepeatedTasks { let receiver = if external_receiver { Receiver::External(receiver) } else { - let request = Arc::new(tokio::sync::Notify::new()); - let (completed_tx, completed) = tokio::sync::mpsc::unbounded_channel(); + let request = Arc::new(Notify::new()); + let (completed_tx, completed) = mpsc::unbounded_channel(); let request_rx = request.clone(); let start = start.clone(); let stop = stop.clone(); diff --git a/benchmarks/benches/ecosystem/mpsc/unbounded.rs b/benchmarks/benches/ecosystem/mpsc/unbounded.rs index 60f36117..0e1b6784 100644 --- a/benchmarks/benches/ecosystem/mpsc/unbounded.rs +++ b/benchmarks/benches/ecosystem/mpsc/unbounded.rs @@ -26,19 +26,21 @@ use benchmarks::support::bench_context; use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; +use tokio::runtime::Builder; +use tokio::sync::Notify; -use super::adapters::AsyncChannel; -use super::adapters::Asyncband; -use super::adapters::Flume; -use super::adapters::Tokio; -use super::adapters::UnboundedMpsc; -use super::support::BATCH_MESSAGES; -use super::support::PRODUCER_COUNTS; -use super::support::RepeatedBatch; -use super::support::RepeatedTasks; -use super::support::TASK_SHAPES; -use super::support::TaskShape; -use super::support::Unbounded; +use crate::mpsc::adapters::AsyncChannel; +use crate::mpsc::adapters::Asyncband; +use crate::mpsc::adapters::Flume; +use crate::mpsc::adapters::Tokio; +use crate::mpsc::adapters::UnboundedMpsc; +use crate::mpsc::support::BATCH_MESSAGES; +use crate::mpsc::support::PRODUCER_COUNTS; +use crate::mpsc::support::RepeatedBatch; +use crate::mpsc::support::RepeatedTasks; +use crate::mpsc::support::TASK_SHAPES; +use crate::mpsc::support::TaskShape; +use crate::mpsc::support::Unbounded; #[divan::bench(types = [Asyncband, Tokio, AsyncChannel, Flume])] fn ready_send_receive(bencher: Bencher) { @@ -185,14 +187,12 @@ fn scheduled_bursts_inline>( ) { assert_eq!(BATCH_MESSAGES % burst_messages, 0); assert_eq!(burst_messages % producers, 0); - let runtime = tokio::runtime::Builder::new_multi_thread() + let runtime = Builder::new_multi_thread() .worker_threads(4) .build() .unwrap(); let (sender, mut receiver) = C::channel(); - let start: Vec<_> = (0..producers) - .map(|_| Arc::new(tokio::sync::Notify::new())) - .collect(); + let start: Vec<_> = (0..producers).map(|_| Arc::new(Notify::new())).collect(); let stop = Arc::new(AtomicBool::new(false)); let workers: Vec<_> = start .iter() diff --git a/benchmarks/benches/ecosystem/watch/paths.rs b/benchmarks/benches/ecosystem/watch/paths.rs index 0be7228e..aa66807a 100644 --- a/benchmarks/benches/ecosystem/watch/paths.rs +++ b/benchmarks/benches/ecosystem/watch/paths.rs @@ -27,9 +27,9 @@ use benchmarks::support::poll_ready; use divan::Bencher; use divan::black_box; -use super::adapters::Asyncband; -use super::adapters::Tokio; -use super::adapters::Watch; +use crate::watch::adapters::Asyncband; +use crate::watch::adapters::Tokio; +use crate::watch::adapters::Watch; const RECEIVER_COUNTS: &[usize] = &[2, 4, 8, 32]; diff --git a/benchmarks/benches/primitives/blocking/block_on.rs b/benchmarks/benches/primitives/blocking/block_on.rs index d7524bba..8faa57af 100644 --- a/benchmarks/benches/primitives/blocking/block_on.rs +++ b/benchmarks/benches/primitives/blocking/block_on.rs @@ -21,7 +21,7 @@ use std::task::Poll; use std::task::Waker; use std::thread; -use asyncband::blocking::FutureExt as _; +use asyncband::blocking::FutureExt; use divan::Bencher; use divan::black_box; diff --git a/benchmarks/benches/primitives/blocking/wait_timeout.rs b/benchmarks/benches/primitives/blocking/wait_timeout.rs index 1cd1efd7..c0e62725 100644 --- a/benchmarks/benches/primitives/blocking/wait_timeout.rs +++ b/benchmarks/benches/primitives/blocking/wait_timeout.rs @@ -17,7 +17,7 @@ use std::time::Duration; -use asyncband::blocking::FutureExt as _; +use asyncband::blocking::FutureExt; use divan::Bencher; use divan::black_box; diff --git a/benchmarks/benches/primitives/mpmc/bounded.rs b/benchmarks/benches/primitives/mpmc/bounded.rs index 13d36ef5..5c3e52ac 100644 --- a/benchmarks/benches/primitives/mpmc/bounded.rs +++ b/benchmarks/benches/primitives/mpmc/bounded.rs @@ -25,7 +25,7 @@ use benchmarks::support::poll_ready; use divan::Bencher; use divan::black_box; -use super::FAST_SAMPLE_SIZE; +use crate::mpmc::FAST_SAMPLE_SIZE; #[divan::bench(sample_size = FAST_SAMPLE_SIZE)] fn try_send_then_try_recv(bencher: Bencher) { diff --git a/benchmarks/benches/primitives/mpmc/unbounded.rs b/benchmarks/benches/primitives/mpmc/unbounded.rs index 59bec9b6..5e71cc56 100644 --- a/benchmarks/benches/primitives/mpmc/unbounded.rs +++ b/benchmarks/benches/primitives/mpmc/unbounded.rs @@ -25,7 +25,7 @@ use benchmarks::support::poll_ready; use divan::Bencher; use divan::black_box; -use super::FAST_SAMPLE_SIZE; +use crate::mpmc::FAST_SAMPLE_SIZE; #[divan::bench(sample_size = FAST_SAMPLE_SIZE)] fn send_then_try_recv(bencher: Bencher) { diff --git a/benchmarks/benches/primitives/once_map/initialize.rs b/benchmarks/benches/primitives/once_map/initialize.rs index 3218a361..fe378fbb 100644 --- a/benchmarks/benches/primitives/once_map/initialize.rs +++ b/benchmarks/benches/primitives/once_map/initialize.rs @@ -27,12 +27,12 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::support::BATCH_SAMPLE_SIZE; -use super::support::BATCH_SIZES; -use super::support::BenchMap; -use super::support::FAST_SAMPLE_SIZE; -use super::support::READY_ENTRY_COUNTS; -use super::support::ready_map; +use crate::once_map::support::BATCH_SAMPLE_SIZE; +use crate::once_map::support::BATCH_SIZES; +use crate::once_map::support::BenchMap; +use crate::once_map::support::FAST_SAMPLE_SIZE; +use crate::once_map::support::READY_ENTRY_COUNTS; +use crate::once_map::support::ready_map; const ABSENT_KEY: usize = usize::MAX; diff --git a/benchmarks/benches/primitives/once_map/lookup.rs b/benchmarks/benches/primitives/once_map/lookup.rs index 3b708753..bfef1cca 100644 --- a/benchmarks/benches/primitives/once_map/lookup.rs +++ b/benchmarks/benches/primitives/once_map/lookup.rs @@ -20,13 +20,13 @@ use benchmarks::support::poll_ready; use divan::Bencher; use divan::black_box; -use super::support::CONTENDED_SAMPLE_SIZE; -use super::support::HOT_ENTRY_COUNTS; -use super::support::READY_ENTRY_COUNTS; -use super::support::THREAD_COUNTS; -use super::support::distributed_absent_key; -use super::support::distributed_ready_key; -use super::support::ready_map; +use crate::once_map::support::CONTENDED_SAMPLE_SIZE; +use crate::once_map::support::HOT_ENTRY_COUNTS; +use crate::once_map::support::READY_ENTRY_COUNTS; +use crate::once_map::support::THREAD_COUNTS; +use crate::once_map::support::distributed_absent_key; +use crate::once_map::support::distributed_ready_key; +use crate::once_map::support::ready_map; #[divan::bench(threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] fn get_hit_same_key(bencher: Bencher) { diff --git a/benchmarks/benches/primitives/once_map/maintenance.rs b/benchmarks/benches/primitives/once_map/maintenance.rs index e939ca13..39c8e8e1 100644 --- a/benchmarks/benches/primitives/once_map/maintenance.rs +++ b/benchmarks/benches/primitives/once_map/maintenance.rs @@ -19,11 +19,11 @@ use benchmarks::support::defer_input_drop; use divan::Bencher; use divan::black_box; -use super::support::BATCH_SAMPLE_SIZE; -use super::support::FAST_SAMPLE_SIZE; -use super::support::NONEMPTY_ENTRY_COUNTS; -use super::support::READY_ENTRY_COUNTS; -use super::support::ready_map; +use crate::once_map::support::BATCH_SAMPLE_SIZE; +use crate::once_map::support::FAST_SAMPLE_SIZE; +use crate::once_map::support::NONEMPTY_ENTRY_COUNTS; +use crate::once_map::support::READY_ENTRY_COUNTS; +use crate::once_map::support::ready_map; #[divan::bench(args = NONEMPTY_ENTRY_COUNTS, sample_size = BATCH_SAMPLE_SIZE)] fn remove_hit(bencher: Bencher, ready_entries: usize) { diff --git a/benchmarks/benches/primitives/singleflight/coalescing.rs b/benchmarks/benches/primitives/singleflight/coalescing.rs index a52ec48b..dd244910 100644 --- a/benchmarks/benches/primitives/singleflight/coalescing.rs +++ b/benchmarks/benches/primitives/singleflight/coalescing.rs @@ -25,9 +25,9 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::support::BATCH_SAMPLE_SIZE; -use super::support::BATCH_SIZES; -use super::support::BenchGroup; +use crate::singleflight::support::BATCH_SAMPLE_SIZE; +use crate::singleflight::support::BATCH_SIZES; +use crate::singleflight::support::BenchGroup; // Measures the complete successful fan-in: one leader is suspended, every duplicate joins it, and // all callers receive the cloned result after the leader completes. diff --git a/benchmarks/benches/primitives/singleflight/leaders.rs b/benchmarks/benches/primitives/singleflight/leaders.rs index 28355dcd..ca1675ad 100644 --- a/benchmarks/benches/primitives/singleflight/leaders.rs +++ b/benchmarks/benches/primitives/singleflight/leaders.rs @@ -26,13 +26,13 @@ use divan::Bencher; use divan::black_box; use divan::counter::ItemsCount; -use super::support::BATCH_SAMPLE_SIZE; -use super::support::BATCH_SIZES; -use super::support::BenchGroup; -use super::support::CONTENDED_SAMPLE_SIZE; -use super::support::FAST_SAMPLE_SIZE; -use super::support::THREAD_COUNTS; -use super::support::unique_thread_key; +use crate::singleflight::support::BATCH_SAMPLE_SIZE; +use crate::singleflight::support::BATCH_SIZES; +use crate::singleflight::support::BenchGroup; +use crate::singleflight::support::CONTENDED_SAMPLE_SIZE; +use crate::singleflight::support::FAST_SAMPLE_SIZE; +use crate::singleflight::support::THREAD_COUNTS; +use crate::singleflight::support::unique_thread_key; // Reuse the group so table capacity survives completed calls. #[divan::bench(sample_size = FAST_SAMPLE_SIZE)] diff --git a/benchmarks/src/channels/adapters.rs b/benchmarks/src/channels/adapters.rs index 9ef3fd0c..488fc021 100644 --- a/benchmarks/src/channels/adapters.rs +++ b/benchmarks/src/channels/adapters.rs @@ -19,8 +19,10 @@ use std::future::Future; use std::marker::PhantomData; use asyncband::blocking::FutureExt; +use asyncband::mpmc; +use asyncband::spmc; -use super::BOUNDED_CAPACITY; +use crate::channels::BOUNDED_CAPACITY; pub struct Mpmc; pub struct Spmc; @@ -41,11 +43,11 @@ pub trait Channel: Send + Sync + 'static { } impl Channel for Bounded { - type Sender = asyncband::spmc::BoundedSender; - type Receiver = asyncband::spmc::BoundedReceiver; + type Sender = spmc::BoundedSender; + type Receiver = spmc::BoundedReceiver; fn channel() -> (Self::Sender, Self::Receiver) { - asyncband::spmc::bounded(CAPACITY) + spmc::bounded(CAPACITY) } async fn send(sender: &mut Self::Sender, value: usize) { @@ -58,11 +60,11 @@ impl Channel for Bounded { } impl Channel for Unbounded { - type Sender = asyncband::spmc::UnboundedSender; - type Receiver = asyncband::spmc::UnboundedReceiver; + type Sender = spmc::UnboundedSender; + type Receiver = spmc::UnboundedReceiver; fn channel() -> (Self::Sender, Self::Receiver) { - asyncband::spmc::unbounded() + spmc::unbounded() } async fn send(sender: &mut Self::Sender, value: usize) { @@ -139,11 +141,11 @@ pub trait UnboundedMpmc: Send + Sync + 'static { } impl BoundedMpmc for Mpmc { - type Receiver = asyncband::mpmc::BoundedReceiver; - type Sender = asyncband::mpmc::BoundedSender; + type Receiver = mpmc::BoundedReceiver; + type Sender = mpmc::BoundedSender; fn channel(capacity: usize) -> (Self::Sender, Self::Receiver) { - asyncband::mpmc::bounded(capacity) + mpmc::bounded(capacity) } async fn send_async(sender: &Self::Sender, value: usize) { @@ -159,11 +161,11 @@ impl BoundedMpmc for Mpmc { } impl UnboundedMpmc for Mpmc { - type Receiver = asyncband::mpmc::UnboundedReceiver; - type Sender = asyncband::mpmc::UnboundedSender; + type Receiver = mpmc::UnboundedReceiver; + type Sender = mpmc::UnboundedSender; fn channel() -> (Self::Sender, Self::Receiver) { - asyncband::mpmc::unbounded() + mpmc::unbounded() } fn send(sender: &Self::Sender, value: usize) { diff --git a/benchmarks/src/channels/mod.rs b/benchmarks/src/channels/mod.rs index b70e390e..3eb3adf6 100644 --- a/benchmarks/src/channels/mod.rs +++ b/benchmarks/src/channels/mod.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use tokio::runtime::Builder; use tokio::runtime::Runtime; pub mod adapters; @@ -26,11 +27,9 @@ pub const BOUNDED_CAPACITY: usize = 64; pub fn runtime(worker_threads: usize) -> Runtime { if worker_threads == 0 { - tokio::runtime::Builder::new_current_thread() - .build() - .unwrap() + Builder::new_current_thread().build().unwrap() } else { - tokio::runtime::Builder::new_multi_thread() + Builder::new_multi_thread() .worker_threads(worker_threads) .build() .unwrap() diff --git a/benchmarks/src/channels/mpmc.rs b/benchmarks/src/channels/mpmc.rs index d7582f36..8348058d 100644 --- a/benchmarks/src/channels/mpmc.rs +++ b/benchmarks/src/channels/mpmc.rs @@ -25,10 +25,10 @@ use divan::black_box; use tokio::runtime::Runtime; use tokio::task::JoinSet; -use super::BATCH_MESSAGES; -use super::adapters::BoundedMpmc; -use super::adapters::Channel; -use super::adapters::UnboundedMpmc; +use crate::channels::BATCH_MESSAGES; +use crate::channels::adapters::BoundedMpmc; +use crate::channels::adapters::Channel; +use crate::channels::adapters::UnboundedMpmc; #[derive(Clone, Copy, Debug)] pub struct Topology { diff --git a/benchmarks/src/channels/spmc.rs b/benchmarks/src/channels/spmc.rs index 2e99c5c1..efdc7d29 100644 --- a/benchmarks/src/channels/spmc.rs +++ b/benchmarks/src/channels/spmc.rs @@ -19,22 +19,23 @@ use std::sync::Arc; use divan::black_box; use tokio::runtime::Runtime; +use tokio::sync::Barrier; use tokio::task::JoinSet; -use super::BATCH_MESSAGES; -use super::adapters::Channel; +use crate::channels::BATCH_MESSAGES; +use crate::channels::adapters::Channel; pub const CONSUMERS: &[usize] = &[1, 2, 4, 8]; pub struct TaskBatch { - start: Arc, + start: Arc, tasks: JoinSet<(usize, usize)>, } impl TaskBatch { pub fn new(runtime: &Runtime, consumers: usize) -> Self { let (mut sender, receiver) = C::channel(); - let start = Arc::new(tokio::sync::Barrier::new(consumers + 2)); + let start = Arc::new(Barrier::new(consumers + 2)); let mut tasks = JoinSet::new(); let producer_start = start.clone(); tasks.spawn_on( diff --git a/examples/src/lazy_cell_boxed_vs_inline.rs b/examples/src/lazy_cell_boxed_vs_inline.rs index 5bc1d8ca..2b1ca67b 100644 --- a/examples/src/lazy_cell_boxed_vs_inline.rs +++ b/examples/src/lazy_cell_boxed_vs_inline.rs @@ -18,6 +18,7 @@ //! Choose how to store a lazy initializer: box its future to keep the cell movable, pin the cell //! locally to store the future inline, or share a pinned cell across tasks. +use std::pin::pin; use std::sync::Arc; use asyncband::once::LazyCell; @@ -32,7 +33,7 @@ async fn main() { // Returning the async future directly stores it inline. The caller pins the cell before the // first poll, so the future stays at a stable address without a heap allocation. let inline = LazyCell::new(async || "inline".to_owned()); - let mut inline = std::pin::pin!(inline); + let mut inline = pin!(inline); assert_eq!(LazyCell::force_pin(inline.as_ref()).await, "inline"); LazyCell::force_pin_mut(inline.as_mut()) diff --git a/examples/src/once_cell_vs_lazy_cell.rs b/examples/src/once_cell_vs_lazy_cell.rs index caad33a2..c4f339d1 100644 --- a/examples/src/once_cell_vs_lazy_cell.rs +++ b/examples/src/once_cell_vs_lazy_cell.rs @@ -20,6 +20,7 @@ //! cancellation. use std::future::Ready; +use std::future::ready; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -32,7 +33,7 @@ static ONCE_ENDPOINT: OnceCell = OnceCell::new(); static LAZY_ENDPOINT: LazyCell> = LazyCell::new(load_default_endpoint); fn load_default_endpoint() -> Ready { - std::future::ready("https://service.example".to_owned()) + ready("https://service.example".to_owned()) } #[tokio::main(flavor = "current_thread")] diff --git a/examples/src/phaser_rounds.rs b/examples/src/phaser_rounds.rs index b8a35424..2b39d87b 100644 --- a/examples/src/phaser_rounds.rs +++ b/examples/src/phaser_rounds.rs @@ -25,6 +25,8 @@ //! These are coordination patterns for jobs such as parallel imports. The work within a batch is //! represented by a yield; the examples focus on when workers can join, proceed, and leave. +use std::future::ready; + use asyncband::phaser::Closed; use asyncband::phaser::Phaser; use asyncband::phaser::PhaserParticipant; @@ -125,7 +127,7 @@ async fn cancellation_retry() -> Result<(), Closed> { biased; result = participant.wait() => panic!("peer has not arrived: {result:?}"), // Poll the wait once, then cancel it deterministically without a timer or sleep. - _ = std::future::ready(()) => {} + _ = ready(()) => {} } assert_eq!(phaser.arrived_parties(), 1); peer.arrive()?; diff --git a/examples/src/shared_completion.rs b/examples/src/shared_completion.rs index 4087927b..5570b1ff 100644 --- a/examples/src/shared_completion.rs +++ b/examples/src/shared_completion.rs @@ -21,6 +21,8 @@ //! require one channel per consumer and manual fan-out, and a late consumer would need separate //! result storage. Completion retains the result and exposes it to every observer. +use std::ptr; + use asyncband::completion; #[tokio::main(flavor = "current_thread")] @@ -44,7 +46,7 @@ async fn main() { let late_observer = completion.clone(); let late_result = late_observer.wait().await.unwrap(); - assert!(std::ptr::eq(dashboard_result, audit_result)); - assert!(std::ptr::eq(dashboard_result, late_result)); + assert!(ptr::eq(dashboard_result, audit_result)); + assert!(ptr::eq(dashboard_result, late_result)); println!("{late_result}"); } diff --git a/tests-integration/src/lib.rs b/tests-integration/src/lib.rs index 58e9fc89..a60bec49 100644 --- a/tests-integration/src/lib.rs +++ b/tests-integration/src/lib.rs @@ -16,16 +16,22 @@ // under the License. use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; +use std::panic::resume_unwind; use std::pin::Pin; use std::sync::Arc; use std::sync::Mutex; use std::sync::OnceLock; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; +use std::sync::mpsc; use std::task::Context; use std::task::Poll; use std::task::Wake; use std::task::Waker; +use std::thread; +use std::time::Duration; use tokio::runtime::Runtime; @@ -115,20 +121,20 @@ pub fn waker_on_drop(callback: impl FnOnce() + Send + 'static) -> Waker { } pub fn assert_completes_without_deadlock(test: impl FnOnce() + Send + 'static) { - let (finished_tx, finished_rx) = std::sync::mpsc::channel(); - let worker = std::thread::spawn(move || { - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(test)); + let (finished_tx, finished_rx) = mpsc::channel(); + let worker = thread::spawn(move || { + let result = catch_unwind(AssertUnwindSafe(test)); finished_tx.send(result).unwrap(); }); #[cfg(not(miri))] let result = finished_rx - .recv_timeout(std::time::Duration::from_secs(10)) + .recv_timeout(Duration::from_secs(10)) .expect("test did not finish"); // Miri detects deadlock itself; its interpretation time must not determine test success. #[cfg(miri)] let result = finished_rx.recv().expect("test did not finish"); worker.join().unwrap(); if let Err(payload) = result { - std::panic::resume_unwind(payload); + resume_unwind(payload); } } diff --git a/tests-integration/tests/blocking_test.rs b/tests-integration/tests/blocking_test.rs index 5f5e7511..9afcf7a4 100644 --- a/tests-integration/tests/blocking_test.rs +++ b/tests-integration/tests/blocking_test.rs @@ -16,6 +16,8 @@ // under the License. use std::future::Future; +use std::future::pending; +use std::future::poll_fn; use std::pin::Pin; use std::sync::Arc; use std::sync::Mutex; @@ -26,7 +28,7 @@ use std::task::Waker; use std::thread; use std::time::Duration; -use asyncband::blocking::FutureExt as _; +use asyncband::blocking::FutureExt; // Long timeouts in this test target are watchdogs for detecting a stalled test process, not // assertions about elapsed-time precision. @@ -35,24 +37,18 @@ const TEST_WATCHDOG: Duration = Duration::from_secs(5); #[test] fn block_on_supports_method_and_ufcs_syntax() { assert_eq!(async { 42 }.block_on(), 42); - assert_eq!(asyncband::blocking::FutureExt::block_on(async { 7 }), 7); + assert_eq!(FutureExt::block_on(async { 7 }), 7); } #[test] fn wait_timeout_polls_before_checking_the_deadline() { assert_eq!(async { 42 }.wait_timeout(Duration::ZERO), Some(42)); - assert_eq!( - std::future::pending::<()>().wait_timeout(Duration::ZERO), - None - ); + assert_eq!(pending::<()>().wait_timeout(Duration::ZERO), None); } #[test] fn pending_future_times_out() { - assert_eq!( - std::future::pending::<()>().wait_timeout(Duration::from_millis(1)), - None - ); + assert_eq!(pending::<()>().wait_timeout(Duration::from_millis(1)), None); } #[test] @@ -72,7 +68,7 @@ fn nested_waits_use_independent_notifications() { let (done_tx, done_rx) = mpsc::channel(); let worker = thread::spawn(move || { let mut outer_polled = false; - let output = std::future::poll_fn(|outer_context| { + let output = poll_fn(|outer_context| { if outer_polled { Poll::Ready(42) } else { @@ -80,7 +76,7 @@ fn nested_waits_use_independent_notifications() { outer_context.waker().wake_by_ref(); let mut inner_polled = false; - let inner_output = std::future::poll_fn(|inner_context| { + let inner_output = poll_fn(|inner_context| { if inner_polled { Poll::Ready(7) } else { diff --git a/tests-integration/tests/broadcast_test/bounded.rs b/tests-integration/tests/broadcast_test/bounded.rs index 5423783e..23e6a8af 100644 --- a/tests-integration/tests/broadcast_test/bounded.rs +++ b/tests-integration/tests/broadcast_test/bounded.rs @@ -16,9 +16,12 @@ // under the License. use std::future::Future; +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; use std::sync::Arc; use std::sync::Barrier; use std::task::Context; +use std::task::Poll; use std::task::Waker; use std::thread; @@ -291,7 +294,7 @@ fn parked_recv_that_reclaims_wakes_a_blocked_sender() { // Reclaim through the `recv` future rather than `try_recv`: it is a separate call site, and a // release wired into only one of them would strand this producer. let mut recv = Box::pin(rx.recv()); - assert_eq!(poll_once(recv.as_mut()), std::task::Poll::Ready(Ok(0))); + assert_eq!(poll_once(recv.as_mut()), Poll::Ready(Ok(0))); drop(recv); assert_eq!(tracker.count(), 1); @@ -535,7 +538,7 @@ fn bounded_parked_recv_wakes_when_the_last_sender_drops() { assert_eq!(tracker.count(), 1); assert_eq!( poll_once(recv.as_mut()), - std::task::Poll::Ready(Err(RecvError::Disconnected)) + Poll::Ready(Err(RecvError::Disconnected)) ); } @@ -602,9 +605,7 @@ fn bounded_panicking_clone_leaves_the_channel_consistent() { .unwrap(); // Two receivers share the payload, so `try_recv` has to clone it. - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - rx1.try_recv().map(|msg| msg.value) - })); + let result = catch_unwind(AssertUnwindSafe(|| rx1.try_recv().map(|msg| msg.value))); assert!(result.is_err()); // The panicking `try_recv` call still consumed the message for `rx1`, and left the channel @@ -636,7 +637,7 @@ fn panicking_payload_destructor_still_releases_capacity() { // Dropping `rx2` reclaims all three retained messages and their destructors panic. The // capacity they released must already have reached the parked producer by then. - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(rx2))); + let result = catch_unwind(AssertUnwindSafe(|| drop(rx2))); assert!(result.is_err()); assert!( diff --git a/tests-integration/tests/broadcast_test/unbounded.rs b/tests-integration/tests/broadcast_test/unbounded.rs index 3324e4d9..fd70417a 100644 --- a/tests-integration/tests/broadcast_test/unbounded.rs +++ b/tests-integration/tests/broadcast_test/unbounded.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -264,9 +266,7 @@ fn panicking_clone_leaves_the_channel_consistent() { }); // Two receivers share the payload, so `try_recv` has to clone it. - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - rx1.try_recv().map(|msg| msg.value) - })); + let result = catch_unwind(AssertUnwindSafe(|| rx1.try_recv().map(|msg| msg.value))); assert!(result.is_err()); // The panicking `try_recv` call still consumed the message for `rx1`, and left the channel diff --git a/tests-integration/tests/completion_test.rs b/tests-integration/tests/completion_test.rs index 4ea0afc1..8e411ef8 100644 --- a/tests-integration/tests/completion_test.rs +++ b/tests-integration/tests/completion_test.rs @@ -16,6 +16,7 @@ // under the License. use std::cell::Cell; +use std::ptr; use std::sync::Arc; use std::sync::Barrier; use std::task::Poll; @@ -45,13 +46,13 @@ fn all_observers_borrow_the_same_non_clone_value() { let second_value = FutureExt::block_on(second.wait()).unwrap(); let repeated = FutureExt::block_on(first.wait()).unwrap(); assert_eq!(first_value.0.as_str(), "ready"); - assert!(std::ptr::eq(first_value, second_value)); - assert!(std::ptr::eq(first_value, repeated)); + assert!(ptr::eq(first_value, second_value)); + assert!(ptr::eq(first_value, repeated)); let late = first.clone(); drop(first); let late_value = FutureExt::block_on(late.wait()).unwrap(); - assert!(std::ptr::eq(second_value, late_value)); + assert!(ptr::eq(second_value, late_value)); } #[test] diff --git a/tests-integration/tests/composition_test.rs b/tests-integration/tests/composition_test.rs index 64ac539b..f6b14dda 100644 --- a/tests-integration/tests/composition_test.rs +++ b/tests-integration/tests/composition_test.rs @@ -20,7 +20,7 @@ use std::thread; use std::time::Duration; use asyncband::barrier::Barrier; -use asyncband::blocking::FutureExt as _; +use asyncband::blocking::FutureExt; use asyncband::mutex::Mutex; use asyncband::oneshot; diff --git a/tests-integration/tests/latch_test.rs b/tests-integration/tests/latch_test.rs index 7a4e6c6c..ba37f83c 100644 --- a/tests-integration/tests/latch_test.rs +++ b/tests-integration/tests/latch_test.rs @@ -21,6 +21,7 @@ use std::task::Waker; use asyncband::latch::Latch; use tests_integration::WakeCounter; +use tokio::sync::Barrier; #[test] fn countdown_operations_saturate_at_zero() { @@ -97,7 +98,7 @@ async fn concurrent_arrivals_complete_the_latch() { const TASKS: usize = 32; let latch = Arc::new(Latch::new(TASKS as u32)); - let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let start = Arc::new(Barrier::new(TASKS + 1)); let mut tasks = Vec::with_capacity(TASKS); for _ in 0..TASKS { diff --git a/tests-integration/tests/lazy_cell_test.rs b/tests-integration/tests/lazy_cell_test.rs index 9c1d4aed..d08f4bef 100644 --- a/tests-integration/tests/lazy_cell_test.rs +++ b/tests-integration/tests/lazy_cell_test.rs @@ -17,10 +17,15 @@ use std::cell::Cell; use std::future::Ready; +use std::future::pending; +use std::future::poll_fn; +use std::future::ready; +use std::pin::pin; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; +use std::task::Poll; use asyncband::once::LazyCell; use tokio::sync::Notify; @@ -123,7 +128,7 @@ async fn from_future_does_not_poll_until_forced() { 42 } }); - let lazy = std::pin::pin!(lazy); + let lazy = pin!(lazy); assert_eq!(polls.load(Ordering::SeqCst), 0); assert_eq!(LazyCell::force_pin(lazy.as_ref()).await, &42); @@ -205,7 +210,7 @@ async fn dropping_cell_drops_suspended_attempt() { move || { Box::pin(async move { started.notify_one(); - std::future::pending::<()>().await; + pending::<()>().await; drop(held); 42 }) @@ -284,9 +289,9 @@ async fn initializer_poll_panic_permanently_poisons_cell() { async fn completed_future_drop_panic_permanently_poisons_cell() { let lazy = Arc::new(LazyCell::new(|| { let guard = DropPanic; - std::future::poll_fn(move |_| { + poll_fn(move |_| { let _ = &guard; - std::task::Poll::Ready(42) + Poll::Ready(42) }) })); @@ -307,7 +312,7 @@ async fn completed_future_drop_panic_permanently_poisons_cell() { #[tokio::test] async fn initializer_creation_panic_permanently_poisons_cell() { - let lazy = Arc::new(LazyCell::new(|| -> std::future::Ready { + let lazy = Arc::new(LazyCell::new(|| -> Ready { panic!("initializer creation panic") })); @@ -386,7 +391,7 @@ async fn initializer_need_not_be_sync() { } fn static_initializer() -> Ready { - std::future::ready(42) + ready(42) } static STATIC_LAZY: LazyCell> = LazyCell::new(static_initializer); @@ -406,7 +411,7 @@ async fn inline_future_supports_local_non_send_state() { *local + 1 } }); - let mut lazy = std::pin::pin!(lazy); + let mut lazy = pin!(lazy); assert_eq!(*LazyCell::force_pin(lazy.as_ref()).await, 42); assert_eq!(LazyCell::get(&lazy), Some(&42)); @@ -431,7 +436,7 @@ async fn cancellation_preserves_inline_initialization_future() { 42 } }); - let lazy = std::pin::pin!(lazy); + let lazy = pin!(lazy); tokio::select! { biased; diff --git a/tests-integration/tests/mpmc_test/concurrency.rs b/tests-integration/tests/mpmc_test/concurrency.rs index ae20a747..dbcbede8 100644 --- a/tests-integration/tests/mpmc_test/concurrency.rs +++ b/tests-integration/tests/mpmc_test/concurrency.rs @@ -21,7 +21,7 @@ use std::time::Duration; use asyncband::mpmc; use tokio::task::JoinHandle; -use super::Receiver; +use crate::Receiver; const PRODUCERS: usize = 8; const CONSUMERS: usize = 8; diff --git a/tests-integration/tests/mpmc_test/notification.rs b/tests-integration/tests/mpmc_test/notification.rs index f81f8953..e5fffee5 100644 --- a/tests-integration/tests/mpmc_test/notification.rs +++ b/tests-integration/tests/mpmc_test/notification.rs @@ -26,7 +26,7 @@ use tests_integration::WakeCounter; use tests_integration::expect_ready; use tests_integration::poll_with; -use super::Receiver; +use crate::Receiver; #[derive(Debug)] enum NotifiedReceiver { @@ -39,7 +39,7 @@ fn receiver_notifications>( channel: impl Fn() -> (S, R), send: impl Fn(&S, usize), ) { - use NotifiedReceiver::*; + use crate::notification::NotifiedReceiver::*; for action in [Receive, Cancel, TakeValueThenCancel] { let (sender, receiver) = channel(); diff --git a/tests-integration/tests/mpsc_test/callbacks.rs b/tests-integration/tests/mpsc_test/callbacks.rs index fa6e1120..b522c5af 100644 --- a/tests-integration/tests/mpsc_test/callbacks.rs +++ b/tests-integration/tests/mpsc_test/callbacks.rs @@ -15,6 +15,9 @@ // specific language governing permissions and limitations // under the License. +use std::array; +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; @@ -244,7 +247,7 @@ fn bounded_disconnect_finishes_cleanup_when_a_value_destructor_panics() { } let (tx, rx) = mpsc::bounded(3); - let drops = Arc::new(std::array::from_fn(|_| AtomicUsize::new(0))); + let drops = Arc::new(array::from_fn(|_| AtomicUsize::new(0))); for id in 0..3 { assert!( tx.try_send(Value { @@ -270,7 +273,7 @@ fn bounded_disconnect_finishes_cleanup_when_a_value_destructor_panics() { 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()); + assert!(catch_unwind(AssertUnwindSafe(|| drop(rx))).is_err()); for (_, wakes) in ¬ifications { assert_eq!(wakes.count(), 1); } diff --git a/tests-integration/tests/mpsc_test/concurrency.rs b/tests-integration/tests/mpsc_test/concurrency.rs index e77d0d71..24facfd4 100644 --- a/tests-integration/tests/mpsc_test/concurrency.rs +++ b/tests-integration/tests/mpsc_test/concurrency.rs @@ -15,8 +15,11 @@ // specific language governing permissions and limitations // under the License. +use std::array; use std::future::Future; +use std::future::poll_fn; use std::pin::Pin; +use std::pin::pin; use std::sync::Arc; use std::sync::Barrier; use std::sync::atomic::AtomicUsize; @@ -33,6 +36,7 @@ use tests_integration::WakeCounter; use tests_integration::poll_once; use tests_integration::poll_with; use tests_integration::test_runtime; +use tokio::sync::oneshot; use tokio_test::assert_ok; #[test] @@ -54,7 +58,7 @@ fn publication_racing_with_close_drops_every_payload_once() { for _ in 0..if cfg!(miri) { 8 } else { 128 } { let (tx, rx) = mpsc::bounded(3); - let drops = Arc::new(std::array::from_fn(|_| AtomicUsize::new(0))); + let drops = Arc::new(array::from_fn(|_| AtomicUsize::new(0))); let start = Barrier::new(4); thread::scope(|scope| { for byte in 0..3 { @@ -297,12 +301,12 @@ fn unbounded_collects_from_multiple_producers() { #[cfg_attr(miri, ignore = "requires an OS-backed Tokio runtime")] async fn bounded_backpressure_progresses_on_an_executor() { let (tx, mut rx) = mpsc::bounded(1); - let (blocked_tx, blocked_rx) = tokio::sync::oneshot::channel(); + let (blocked_tx, blocked_rx) = oneshot::channel(); tx.send(1).await.unwrap(); let sender = tokio::spawn(async move { - let mut send = std::pin::pin!(tx.send(2)); - let first_poll = std::future::poll_fn(|cx| Poll::Ready(send.as_mut().poll(cx))).await; + let mut send = pin!(tx.send(2)); + let first_poll = poll_fn(|cx| Poll::Ready(send.as_mut().poll(cx))).await; assert!(first_poll.is_pending()); blocked_tx.send(()).unwrap(); send.await.unwrap(); diff --git a/tests-integration/tests/mpsc_test/reservation.rs b/tests-integration/tests/mpsc_test/reservation.rs index 80b44dd8..d9ea37db 100644 --- a/tests-integration/tests/mpsc_test/reservation.rs +++ b/tests-integration/tests/mpsc_test/reservation.rs @@ -17,6 +17,7 @@ use std::cell::Cell; use std::mem; +use std::thread; use asyncband::mpsc; use asyncband::mpsc::TryRecvError; @@ -139,7 +140,7 @@ fn receiver_drop_does_not_wait_for_held_or_forgotten_permits() { fn a_permit_can_publish_send_only_payloads_from_another_thread() { let (tx, mut rx) = mpsc::bounded(1); let permit = tx.try_reserve().unwrap(); - std::thread::scope(|scope| { + thread::scope(|scope| { scope .spawn(move || permit.send(Cell::new(42)).unwrap()) .join() @@ -154,7 +155,7 @@ fn concurrent_cancellation_preserves_capacity_and_message_order() { const MESSAGES: usize = if cfg!(miri) { 8 } else { 256 }; let (tx, mut rx) = mpsc::bounded(3); let mut out_of_order = 0; - std::thread::scope(|scope| { + thread::scope(|scope| { let mut workers = vec![]; for producer in 0..PRODUCERS { let tx = tx.clone(); @@ -164,7 +165,7 @@ fn concurrent_cancellation_preserves_capacity_and_message_order() { let permit = loop { match tx.try_reserve() { Ok(permit) => break permit, - Err(TrySendError::Full(())) => std::thread::yield_now(), + Err(TrySendError::Full(())) => thread::yield_now(), Err(TrySendError::Disconnected(())) => panic!("receiver is alive"), } }; @@ -186,7 +187,7 @@ fn concurrent_cancellation_preserves_capacity_and_message_order() { next[producer] += 1; count += 1; } - Err(TryRecvError::Empty) => std::thread::yield_now(), + Err(TryRecvError::Empty) => thread::yield_now(), Err(TryRecvError::Disconnected) => panic!("senders are alive"), } } diff --git a/tests-integration/tests/mutex_test.rs b/tests-integration/tests/mutex_test.rs index 91cac410..62b2baf1 100644 --- a/tests-integration/tests/mutex_test.rs +++ b/tests-integration/tests/mutex_test.rs @@ -15,7 +15,10 @@ // specific language governing permissions and limitations // under the License. +use std::panic::catch_unwind; +use std::pin::pin; use std::sync::Arc; +use std::task::Poll; use asyncband::mutex::*; use tests_integration::poll_once; @@ -71,12 +74,12 @@ async fn test_owned_guard_filter_map_failure() { fn test_guard_prevents_concurrent_access() { let mutex = Mutex::new(0); let guard = mutex.try_lock().unwrap(); - let mut waiting = std::pin::pin!(mutex.lock()); + let mut waiting = pin!(mutex.lock()); assert!(poll_once(waiting.as_mut()).is_pending()); drop(guard); - let std::task::Poll::Ready(second_guard) = poll_once(waiting.as_mut()) else { + let Poll::Ready(second_guard) = poll_once(waiting.as_mut()) else { panic!("queued lock should complete after the active guard drops"); }; assert_eq!(*second_guard, 0); @@ -89,7 +92,7 @@ fn test_lock_panic_safety() { let mutex = Arc::new(Mutex::new(0)); let mutex_clone = mutex.clone(); - let result = std::panic::catch_unwind(AssertUnwindSafe(move || { + let result = catch_unwind(AssertUnwindSafe(move || { let _guard = mutex_clone.try_lock().unwrap(); panic!("test panic"); })); diff --git a/tests-integration/tests/once_cell_test.rs b/tests-integration/tests/once_cell_test.rs index f9303397..b01ce1e9 100644 --- a/tests-integration/tests/once_cell_test.rs +++ b/tests-integration/tests/once_cell_test.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use std::future::pending; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; @@ -22,6 +23,7 @@ use std::sync::atomic::Ordering; use asyncband::once::OnceCell; use tests_integration::poll_once; +use tokio::sync::Barrier; struct DropFlag(Arc); @@ -50,7 +52,7 @@ async fn concurrent_callers_publish_one_value() { let cell = Arc::new(OnceCell::new()); let attempts = Arc::new(AtomicUsize::new(0)); - let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let start = Arc::new(Barrier::new(TASKS + 1)); let mut tasks = Vec::with_capacity(TASKS); for value in 0..TASKS { @@ -79,7 +81,7 @@ async fn concurrent_callers_publish_one_value() { #[tokio::test] async fn cancelled_initialization_can_be_retried() { let cell = OnceCell::new(); - let mut first = Box::pin(cell.get_or_init(async || std::future::pending::().await)); + let mut first = Box::pin(cell.get_or_init(async || pending::().await)); assert!(poll_once(first.as_mut()).is_pending()); drop(first); diff --git a/tests-integration/tests/once_map_test.rs b/tests-integration/tests/once_map_test.rs index 991aba1e..ad9b56e8 100644 --- a/tests-integration/tests/once_map_test.rs +++ b/tests-integration/tests/once_map_test.rs @@ -25,12 +25,13 @@ use std::sync::atomic::Ordering; use asyncband::once::OnceMap; use tests_integration::poll_once; +use tokio::sync::oneshot; #[tokio::test] async fn concurrent_compute_runs_once() { let map = OnceMap::new(); let count = AtomicUsize::new(0); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let first = map.compute("key", async || { count.fetch_add(1, Ordering::SeqCst); @@ -195,8 +196,8 @@ async fn discard_releases_the_removed_key_and_value_after_growth() { #[tokio::test] async fn remove_while_computing_allows_a_new_generation() { let map = Arc::new(OnceMap::new()); - let (started_tx, started_rx) = tokio::sync::oneshot::channel(); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (started_tx, started_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let map_clone = map.clone(); let task = tokio::spawn(async move { diff --git a/tests-integration/tests/once_test.rs b/tests-integration/tests/once_test.rs index 622f746e..f33835db 100644 --- a/tests-integration/tests/once_test.rs +++ b/tests-integration/tests/once_test.rs @@ -15,12 +15,14 @@ // specific language governing permissions and limitations // under the License. +use std::future::pending; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use asyncband::once::Once; use tests_integration::poll_once; +use tokio::sync::Barrier; #[tokio::test] async fn call_once_runs_only_one_initializer() { @@ -51,7 +53,7 @@ async fn concurrent_callers_run_one_initializer() { let once = Arc::new(Once::new()); let counter = Arc::new(AtomicUsize::new(0)); - let start = Arc::new(tokio::sync::Barrier::new(TASKS + 1)); + let start = Arc::new(Barrier::new(TASKS + 1)); let mut tasks = Vec::with_capacity(TASKS); for _ in 0..TASKS { @@ -79,7 +81,7 @@ async fn concurrent_callers_run_one_initializer() { #[tokio::test] async fn cancelled_initializer_can_be_retried() { let once = Once::new(); - let mut first = Box::pin(once.call_once(async || std::future::pending::<()>().await)); + let mut first = Box::pin(once.call_once(async || pending::<()>().await)); assert!(poll_once(first.as_mut()).is_pending()); drop(first); diff --git a/tests-integration/tests/oneshot_test/main.rs b/tests-integration/tests/oneshot_test/main.rs index 7dadc8e1..2d106cf0 100644 --- a/tests-integration/tests/oneshot_test/main.rs +++ b/tests-integration/tests/oneshot_test/main.rs @@ -34,10 +34,10 @@ use asyncband::blocking::FutureExt; use asyncband::oneshot; use asyncband::oneshot::TryRecvError; -use self::support::DropProbe; -use self::support::WakerProbe; -use self::support::spawn_named; -use self::support::spin_until; +use crate::support::DropProbe; +use crate::support::WakerProbe; +use crate::support::spawn_named; +use crate::support::spin_until; mod support; diff --git a/tests-integration/tests/oneshot_test/support.rs b/tests-integration/tests/oneshot_test/support.rs index 0b67c36b..a58c9685 100644 --- a/tests-integration/tests/oneshot_test/support.rs +++ b/tests-integration/tests/oneshot_test/support.rs @@ -29,6 +29,7 @@ use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Wake; use std::task::Waker; +use std::thread; use std::time::Duration; use std::time::Instant; @@ -92,12 +93,12 @@ impl Wake for WakerProbe { } } -pub(super) fn spawn_named(name: &str, f: F) -> std::thread::JoinHandle +pub(super) fn spawn_named(name: &str, f: F) -> thread::JoinHandle where F: FnOnce() -> T + Send + 'static, T: Send + 'static, { - std::thread::Builder::new() + thread::Builder::new() .name(name.to_owned()) .spawn(f) .unwrap() @@ -118,7 +119,7 @@ where assert!(Instant::now() < deadline, "timed out waiting for {label}"); if spins % 64 == 0 { - std::thread::yield_now(); + thread::yield_now(); } else { spin_loop(); } diff --git a/tests-integration/tests/phaser_test.rs b/tests-integration/tests/phaser_test.rs index af81580d..15900ac4 100644 --- a/tests-integration/tests/phaser_test.rs +++ b/tests-integration/tests/phaser_test.rs @@ -15,8 +15,10 @@ // specific language governing permissions and limitations // under the License. +use std::array; use std::panic; use std::task::Poll; +use std::thread; use asyncband::blocking::FutureExt; use asyncband::phaser::Phaser; @@ -24,6 +26,7 @@ use tests_integration::WakeCounter; use tests_integration::poll_once; use tests_integration::poll_with; use tests_integration::waker_on_drop; +use tokio::sync::oneshot; #[test] fn batch_registration_joins_one_observed_phase() { @@ -652,9 +655,9 @@ fn arrivals_publish_each_workers_writes_across_threads() { use std::sync::atomic::Ordering; let phaser = Phaser::new(); - let values = std::array::from_fn::<_, 4, _>(|_| AtomicUsize::new(0)); + let values = array::from_fn::<_, 4, _>(|_| AtomicUsize::new(0)); let participants = phaser.register(values.len()).unwrap(); - std::thread::scope(|scope| { + thread::scope(|scope| { for (id, mut participant) in participants.enumerate() { let values = &values; scope.spawn(move || { @@ -683,7 +686,7 @@ async fn a_failed_task_can_close_the_group_without_reporting_phase_completion() let mut worker = phaser.register_one().unwrap(); let failing = phaser.register_one().unwrap(); let observed = phaser.phase(); - let (arrived, arrival) = tokio::sync::oneshot::channel(); + let (arrived, arrival) = oneshot::channel(); let task = tokio::spawn(async move { worker.arrive().unwrap(); arrived.send(()).unwrap(); diff --git a/tests-integration/tests/pool_test/acquire.rs b/tests-integration/tests/pool_test/acquire.rs index 39276acc..d8e3ece6 100644 --- a/tests-integration/tests/pool_test/acquire.rs +++ b/tests-integration/tests/pool_test/acquire.rs @@ -16,6 +16,7 @@ // under the License. use std::future::Future; +use std::mem; use std::pin::pin; use std::sync::Arc; use std::task::Context; @@ -27,8 +28,8 @@ use asyncband::pool::bounded::Pool; use asyncband::pool::bounded::PoolConfig; use tests_integration::WakeCounter; -use super::support::Manager; -use super::support::ManagerError; +use crate::support::Manager; +use crate::support::ManagerError; fn ready(future: impl Future>) -> T { let wakes = Arc::new(WakeCounter::default()); @@ -81,8 +82,8 @@ fn cancel_waiter(return_before_cancel: bool) { drop(held); // Cancel a notified waiter without requiring FIFO notification order. if first_wakes.count() == 0 { - std::mem::swap(&mut first, &mut next); - std::mem::swap(&mut first_wakes, &mut next_wakes); + mem::swap(&mut first, &mut next); + mem::swap(&mut first_wakes, &mut next_wakes); } assert!(first_wakes.count() > 0); // Cancel the notified future without polling it to retrieve its permit. diff --git a/tests-integration/tests/pool_test/behavior.rs b/tests-integration/tests/pool_test/behavior.rs index 9d68269f..0c43659c 100644 --- a/tests-integration/tests/pool_test/behavior.rs +++ b/tests-integration/tests/pool_test/behavior.rs @@ -25,7 +25,7 @@ use asyncband::pool::QueueStrategy; use asyncband::pool::bounded; use asyncband::pool::unbounded; -use super::support::Manager; +use crate::support::Manager; #[test] #[should_panic(expected = "bounded pool max_size must be greater than zero")] diff --git a/tests-integration/tests/pool_test/recycle.rs b/tests-integration/tests/pool_test/recycle.rs index d877cc61..44020196 100644 --- a/tests-integration/tests/pool_test/recycle.rs +++ b/tests-integration/tests/pool_test/recycle.rs @@ -21,9 +21,9 @@ use tests_integration::expect_ready; use tests_integration::poll_once; use tests_integration::poll_with; -use super::support::Manager; -use super::support::ManagerError; -use super::support::ready; +use crate::support::Manager; +use crate::support::ManagerError; +use crate::support::ready; // Both pool variants promise the same object lifecycle during recycle cancellation. macro_rules! recycle_contract { diff --git a/tests-integration/tests/pool_test/replenish.rs b/tests-integration/tests/pool_test/replenish.rs index b0a351dc..facc7dbe 100644 --- a/tests-integration/tests/pool_test/replenish.rs +++ b/tests-integration/tests/pool_test/replenish.rs @@ -24,9 +24,9 @@ use tests_integration::expect_ready; use tests_integration::poll_once; use tests_integration::poll_with; -use super::support::Manager; -use super::support::ManagerError; -use super::support::ready; +use crate::support::Manager; +use crate::support::ManagerError; +use crate::support::ready; #[test] fn targets_account_for_idle_objects_and_checked_out_capacity() { diff --git a/tests-integration/tests/rwlock_test.rs b/tests-integration/tests/rwlock_test.rs index 226e2f42..ca30f49c 100644 --- a/tests-integration/tests/rwlock_test.rs +++ b/tests-integration/tests/rwlock_test.rs @@ -15,11 +15,13 @@ // specific language governing permissions and limitations // under the License. +use std::future::pending; use std::num::NonZeroUsize; use std::sync::Arc; use asyncband::rwlock::*; use tests_integration::poll_once; +use tokio::sync::Barrier; use tokio_test::assert_pending; use tokio_test::assert_ready; @@ -68,7 +70,7 @@ async fn concurrent_readers_and_writers_preserve_values() { const WRITERS: usize = 10; let rwlock = Arc::new(RwLock::new(0)); - let start = Arc::new(tokio::sync::Barrier::new(READERS + WRITERS + 1)); + let start = Arc::new(Barrier::new(READERS + WRITERS + 1)); let mut readers = Vec::with_capacity(READERS); let mut writers = Vec::with_capacity(WRITERS); @@ -361,7 +363,7 @@ fn cancelled_mapped_reader_admits_a_queued_writer() { let mut reader = Box::pin(async { let _mapped = RwLockReadGuard::map(rwlock.read().await, |values| &values[0]); - std::future::pending::<()>().await; + pending::<()>().await; }); assert_pending!(poll_once(reader.as_mut())); let mut writer = Box::pin(rwlock.write()); @@ -379,7 +381,7 @@ fn cancelled_mapped_writer_admits_a_queued_reader() { let mut writer = Box::pin(async { let mut mapped = RwLockWriteGuard::map(rwlock.write().await, |values| &mut values[0]); *mapped = 10; - std::future::pending::<()>().await; + pending::<()>().await; }); assert_pending!(poll_once(writer.as_mut())); let mut reader = Box::pin(rwlock.read()); @@ -417,7 +419,7 @@ async fn aborted_task_releases_its_owned_mapped_rwlock_guard() { *mapped = 10; let task = tokio::spawn(async move { let _mapped = mapped; - std::future::pending::<()>().await; + pending::<()>().await; }); assert!(rwlock.try_read().is_none()); diff --git a/tests-integration/tests/semaphore_test.rs b/tests-integration/tests/semaphore_test.rs index 02431f3c..db1903d9 100644 --- a/tests-integration/tests/semaphore_test.rs +++ b/tests-integration/tests/semaphore_test.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; use std::pin::pin; use std::sync::Arc; use std::task::Context; @@ -82,7 +84,7 @@ fn maximum_capacity_can_be_acquired_and_returned() { #[test] fn release_overflow_preserves_permits() { let s = Semaphore::new(usize::MAX); - let result = std::panic::catch_unwind(|| s.release(1)); + let result = catch_unwind(|| s.release(1)); assert!(result.is_err()); assert_eq!(s.available_permits(), usize::MAX); @@ -100,7 +102,7 @@ fn merge_overflow_panics_without_losing_borrowed_permits() { s.release(1); let second = s.try_acquire(1).unwrap(); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let result = catch_unwind(AssertUnwindSafe(|| { first.merge(second); })); @@ -117,7 +119,7 @@ fn merge_overflow_panics_without_losing_owned_permits() { s.release(1); let second = s.clone().try_acquire_owned(1).unwrap(); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let result = catch_unwind(AssertUnwindSafe(|| { first.merge(second); })); diff --git a/tests-integration/tests/shutdown_test.rs b/tests-integration/tests/shutdown_test.rs index 6ab4d211..8e5190fe 100644 --- a/tests-integration/tests/shutdown_test.rs +++ b/tests-integration/tests/shutdown_test.rs @@ -16,6 +16,7 @@ // under the License. use std::future::pending; +use std::future::ready; use std::pin::pin; use asyncband::blocking::FutureExt; @@ -122,7 +123,7 @@ fn test_disabled_select_branch_does_not_request_shutdown() { tokio::select! { _ = shutdown, if false => unreachable!(), - _ = std::future::ready(()) => {} + _ = ready(()) => {} } assert!(!watch.is_shutdown_requested()); diff --git a/tests-integration/tests/singleflight_test.rs b/tests-integration/tests/singleflight_test.rs index 284a51b4..961d5d3d 100644 --- a/tests-integration/tests/singleflight_test.rs +++ b/tests-integration/tests/singleflight_test.rs @@ -28,6 +28,7 @@ use std::task::Poll; use asyncband::singleflight::Group; use tests_integration::assert_completes_without_deadlock; use tests_integration::poll_once; +use tokio::sync::oneshot; #[tokio::test] async fn supports_non_clone_key() { @@ -43,7 +44,7 @@ async fn supports_non_clone_key() { async fn concurrent_work_is_coalesced() { let group = Group::new(); let counter = AtomicUsize::new(0); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let leader = group.work("key", || async { counter.fetch_add(1, Ordering::SeqCst); @@ -77,7 +78,7 @@ async fn concurrent_work_is_coalesced() { #[tokio::test] async fn pending_work_does_not_block_other_keys() { let group = Group::new(); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let first = group.work("key1", || async { release_rx.await.unwrap(); @@ -96,8 +97,8 @@ async fn pending_work_does_not_block_other_keys() { async fn forget_detaches_in_flight_work() { let group = Group::new(); let counter = AtomicUsize::new(0); - let (first_release_tx, first_release_rx) = tokio::sync::oneshot::channel(); - let (second_release_tx, second_release_rx) = tokio::sync::oneshot::channel(); + let (first_release_tx, first_release_rx) = oneshot::channel(); + let (second_release_tx, second_release_rx) = oneshot::channel(); let first = group.work("key".to_owned(), || async { counter.fetch_add(1, Ordering::SeqCst); @@ -142,7 +143,7 @@ async fn try_work_returns_value_without_caching_it() { async fn concurrent_try_work_is_coalesced() { let group = Group::new(); let counter = AtomicUsize::new(0); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let leader = group.try_work("key", || async { counter.fetch_add(1, Ordering::SeqCst); @@ -211,7 +212,7 @@ fn duplicate_key_destructor_can_forget_another_key() { assert_completes_without_deadlock(|| { for fallible in [false, true] { let group = Rc::new(Group::new()); - let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); let mut leader = pin!(group.work( Key { value: 7, diff --git a/tests-integration/tests/spmc_test/notification.rs b/tests-integration/tests/spmc_test/notification.rs index 51f0e0fb..5b654ced 100644 --- a/tests-integration/tests/spmc_test/notification.rs +++ b/tests-integration/tests/spmc_test/notification.rs @@ -30,7 +30,7 @@ use tests_integration::expect_ready; use tests_integration::poll_once; use tests_integration::poll_with; -use super::DropSpy; +use crate::DropSpy; #[test] fn bounded_one_send_wakes_one_of_eight_receivers_and_cancellation_hands_off() { diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 47972d9d..2d7c88c6 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -16,6 +16,9 @@ // under the License. use std::cell::Cell; +use std::convert::Infallible; +use std::fmt; +use std::future::Ready; use std::marker::PhantomPinned; use std::panic::RefUnwindSafe; use std::panic::UnwindSafe; @@ -63,7 +66,7 @@ struct PoolManager; impl ManageObject for PoolManager { type Object = i64; - type Error = std::convert::Infallible; + type Error = Infallible; async fn create(&self) -> Result { Ok(0) @@ -90,7 +93,7 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::(); - assert_send_and_sync::>>(); + assert_send_and_sync::>>(); assert_send_and_sync::(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -186,7 +189,7 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::(); assert_unpin::(); - assert_unpin::>>(); + assert_unpin::>>(); assert_unpin::(); assert_unpin::>(); assert_unpin::>(); @@ -274,7 +277,7 @@ fn mpsc_endpoints_keep_legacy_traits_regardless_of_payload() { #[test] fn unbounded_manual_manager_traits_do_not_depend_on_the_object() { fn assert_copy() {} - fn assert_debug() {} + fn assert_debug() {} assert_copy::>(); assert_debug::>(); diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs index d4e5d2e3..087f4a4e 100644 --- a/tests-integration/tests/watch_test.rs +++ b/tests-integration/tests/watch_test.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::panic::AssertUnwindSafe; +use std::panic::catch_unwind; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -110,9 +112,7 @@ fn panicking_clone_leaves_the_update_unseen() { .unwrap(); panic_next.store(true, Ordering::Relaxed); - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - FutureExt::block_on(rx.recv()) - })); + let result = catch_unwind(AssertUnwindSafe(|| FutureExt::block_on(rx.recv()))); assert!(result.is_err()); assert_eq!(rx.has_changed(), Ok(true));