From 7bb450a4020ec1fd246f3bc903898184b8b19f0b Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 24 Aug 2026 00:47:48 +0800 Subject: [PATCH] refactor(broadcast): remove overflow channel --- CHANGELOG.md | 2 +- HISTORY.md | 1 - README.md | 1 - asyncband/Cargo.toml | 1 - asyncband/src/broadcast/mod.rs | 24 - asyncband/src/broadcast/overflow/mod.rs | 548 ---------------------- asyncband/src/broadcast/overflow/tests.rs | 86 ---- asyncband/src/internal/mod.rs | 10 +- asyncband/src/internal/rwlock.rs | 56 --- asyncband/src/lib.rs | 4 +- benchmarks/Cargo.toml | 1 - benchmarks/broadcast.rs | 255 ---------- benchmarks/main.rs | 1 - tests-integration/Cargo.toml | 1 - tests-integration/tests/broadcast_test.rs | 350 -------------- tests-integration/tests/traits_test.rs | 9 - 16 files changed, 4 insertions(+), 1346 deletions(-) delete mode 100644 asyncband/src/broadcast/mod.rs delete mode 100644 asyncband/src/broadcast/overflow/mod.rs delete mode 100644 asyncband/src/broadcast/overflow/tests.rs delete mode 100644 asyncband/src/internal/rwlock.rs delete mode 100644 benchmarks/broadcast.rs delete mode 100644 tests-integration/tests/broadcast_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 12c9357..a0d21b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ All notable changes to this project will be documented in this file. * Gate all exported primitives behind opt-in Cargo features and enable no features by default; downstream dependencies must explicitly enable the APIs they use. * Remove `admission::FairShare` and its `admission` Cargo feature from the feature set. * Remove the `asyncband::atomicbox` module and its `AtomicBox` and `AtomicOptionBox` types from the public API. +* Remove the lossy `broadcast::overflow` channel and its `broadcast` Cargo feature; future broadcast APIs will use explicit bounded and unbounded lossless semantics. * Remove `Semaphore::try_acquire_and_forget`, `Semaphore::acquire_and_forget`, `Semaphore::try_acquire_owned_and_forget`, and `Semaphore::acquire_owned_and_forget`; acquire a permit and call its `forget` method instead. * Rename `oneshot::Sender::is_closed` and `oneshot::Receiver::is_closed` to `is_disconnected`. * Replace `Semaphore::forget` with `Semaphore::drain_permits` and `Semaphore::forget_exact` with `Semaphore::reduce_permits`; permit-level `forget` methods are unchanged. @@ -24,7 +25,6 @@ All notable changes to this project will be documented in this file. ### Bug fixes * Release cancelled wait registrations promptly and reclaim fulfilled `Semaphore::reduce_permits` debt nodes. -* Serialize broadcast publication so receivers cannot observe reserved slots or messages overwritten out of sequence. ### Improvements diff --git a/HISTORY.md b/HISTORY.md index 54143c7..55a9cc2 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -6,7 +6,6 @@ Asyncband collects runtime-agnostic synchronization and coordination tools infor - `barrier::Barrier` is inspired by [`std::sync::Barrier`](https://doc.rust-lang.org/std/sync/struct.Barrier.html) and [`tokio::sync::Barrier`](https://docs.rs/tokio/latest/tokio/sync/struct.Barrier.html), with a different implementation based on the internal `WaitSet` primitive. - The single-future polling loop in `blocking` is adapted from [`pollster`](https://github.com/zesterer/pollster), its parker caching strategy follows [`futures-lite`](https://github.com/smol-rs/futures-lite), and its private parker state machine is adapted from [`parking`](https://github.com/smol-rs/parking) 2.2.1. -- `broadcast::overflow::channel` is derived from [`tokio::sync::broadcast::channel`](https://docs.rs/tokio/latest/tokio/sync/broadcast/fn.channel.html), with a different implementation based on the internal `WaitSet` primitive. - `condvar::Condvar` is inspired by [`std::sync::Condvar`](https://doc.rust-lang.org/std/sync/struct.Condvar.html) and [`async_std::sync::Condvar`](https://docs.rs/async-std/latest/async_std/sync/struct.Condvar.html), with a fair FIFO waiter queue and standard non-buffered notification semantics. - `latch::Latch` is inspired by [`latches`](https://github.com/mirromutth/latches), with a different implementation based on the internal `CountdownState` primitive. - `mutex::Mutex` is derived from [`tokio::sync::Mutex`](https://docs.rs/tokio/latest/tokio/sync/struct.Mutex.html). diff --git a/README.md b/README.md index 539c150..674c87c 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,6 @@ The crate enables no APIs by default. Categories describe each API's primary pur | Channels | [`oneshot::channel`](https://docs.rs/asyncband/*/asyncband/oneshot/fn.channel.html) | `oneshot` | Send one value between two tasks. | | | [`mpsc::bounded`](https://docs.rs/asyncband/*/asyncband/mpsc/fn.bounded.html) | `mpsc` | Send values from multiple producers through a bounded channel. | | | [`mpsc::unbounded`](https://docs.rs/asyncband/*/asyncband/mpsc/fn.unbounded.html) | `mpsc` | Send values from multiple producers through an unbounded channel. | -| | [`broadcast::overflow`](https://docs.rs/asyncband/*/asyncband/broadcast/overflow/) | `broadcast` | Broadcast values and report when slow receivers miss overwritten items. | | Resource reuse | [`pool::bounded`](https://docs.rs/asyncband/*/asyncband/pool/bounded/) | `pool` | Reuse managed objects up to a configured capacity. | | | [`pool::unbounded`](https://docs.rs/asyncband/*/asyncband/pool/unbounded/) | `pool` | Reuse manually supplied or manager-created objects. | | Workload control | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index 4f45d5f..ce31add 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -40,7 +40,6 @@ default = [] barrier = [] blocking = [] -broadcast = [] condvar = ["mutex"] latch = [] mpsc = [] diff --git a/asyncband/src/broadcast/mod.rs b/asyncband/src/broadcast/mod.rs deleted file mode 100644 index 96b8d1b..0000000 --- a/asyncband/src/broadcast/mod.rs +++ /dev/null @@ -1,24 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -//! A multi-producer multi-consumer broadcast channel. -//! -//! This module provides broadcast channels in one of the following policies: -//! -//! * [`overflow`]: when the channel is full, the oldest messages are overwritten. - -pub mod overflow; diff --git a/asyncband/src/broadcast/overflow/mod.rs b/asyncband/src/broadcast/overflow/mod.rs deleted file mode 100644 index 176379a..0000000 --- a/asyncband/src/broadcast/overflow/mod.rs +++ /dev/null @@ -1,548 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -//! A multi-producer multi-consumer broadcast channel. -//! -//! This channel supports multiple senders and multiple receivers. Each message sent by any -//! sender is received by all receivers. If a receiver falls behind, it may miss messages, -//! which is reported via [`RecvError::Lagged`]. -//! -//! # Examples -//! -//! Basic usage: -//! -//! ``` -//! use asyncband::broadcast::overflow; -//! -//! # #[tokio::main] -//! # async fn main() { -//! let (tx, mut rx1) = overflow::channel(16); -//! let mut rx2 = tx.subscribe(); -//! -//! tx.send(10); -//! tx.send(20); -//! -//! assert_eq!(rx1.recv().await, Ok(10)); -//! assert_eq!(rx1.recv().await, Ok(20)); -//! assert_eq!(rx2.recv().await, Ok(10)); -//! assert_eq!(rx2.recv().await, Ok(20)); -//! # } -//! ``` -//! -//! Handling lag: -//! -//! ``` -//! use asyncband::broadcast::overflow; -//! use asyncband::broadcast::overflow::RecvError; -//! -//! # #[tokio::main] -//! # async fn main() { -//! let (tx, mut rx) = overflow::channel(2); -//! -//! tx.send(1); -//! tx.send(2); -//! tx.send(3); // overwrites the oldest message (1) -//! -//! assert_eq!(rx.recv().await, Err(RecvError::Lagged(1))); -//! assert_eq!(rx.recv().await, Ok(2)); -//! assert_eq!(rx.recv().await, Ok(3)); -//! # } -//! ``` - -use std::fmt; -use std::future::Future; -use std::pin::Pin; -use std::sync::Arc; -use std::sync::atomic::AtomicU64; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; -use std::task::Context; -use std::task::Poll; -use std::task::Waker; - -use crate::internal::mutex::Mutex; -use crate::internal::rwlock::RwLock; -use crate::internal::waitset::WaitRegistration; -use crate::internal::waitset::WaitSet; - -#[cfg(test)] -mod tests; - -/// Creates a new broadcast channel with the given hint `capacity`. The actual capacity may be -/// greater than the provided `capacity`. -/// -/// See [module-level documentation](self) for broadcast channel semantics. -/// -/// # Panics -/// -/// Panics if `capacity` is 0. -/// -/// # Examples -/// -/// ``` -/// use asyncband::broadcast::overflow; -/// -/// let (tx, mut rx) = overflow::channel(16); -/// tx.send(10); -/// assert_eq!(rx.try_recv(), Ok(10)); -/// ``` -pub fn channel(capacity: usize) -> (Sender, Receiver) { - assert!(capacity > 0, "capacity must be greater than 0"); - - let capacity = capacity.next_power_of_two(); - let mask = capacity - 1; - - let mut buffer = Vec::with_capacity(capacity); - for _ in 0..capacity { - buffer.push(RwLock::new(Slot { - msg: None, - version: 0, - })); - } - - let shared = Arc::new(Shared { - buffer: buffer.into_boxed_slice(), - capacity, - mask, - tail: AtomicU64::new(0), - state: Mutex::new(State { - waiters: WaitSet::new(), - }), - senders: AtomicUsize::new(1), - }); - let sender = Sender { - shared: shared.clone(), - }; - let receiver = Receiver { shared, head: 0 }; - (sender, receiver) -} - -/// Error returned by [`Receiver::recv`]. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum RecvError { - /// The receiver lagged too far behind. - /// - /// The count is the number of messages skipped. The receiver's internal cursor has been - /// advanced to the oldest available message. - Lagged(u64), - /// The sender has become disconnected, and there will never be any more data received on it. - Disconnected, -} - -impl fmt::Display for RecvError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - RecvError::Lagged(n) => write!(f, "receiver has been lagged by {n}"), - RecvError::Disconnected => write!(f, "receiving on a closed channel"), - } - } -} - -impl std::error::Error for RecvError {} - -/// Error returned by [`Receiver::try_recv`]. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum TryRecvError { - /// This channel is currently empty, but the sender(s) have not yet disconnected, so data may - /// yet become available. - Empty, - /// The receiver lagged too far behind. - /// - /// The count is the number of messages skipped. The receiver's internal cursor has been - /// advanced to the oldest available message. - Lagged(u64), - /// The sender has become disconnected, and there will never be any more data received on it. - Disconnected, -} - -impl fmt::Display for TryRecvError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - TryRecvError::Empty => write!(f, "receiving on an empty channel"), - TryRecvError::Lagged(n) => write!(f, "receiver has been lagged by {n}"), - TryRecvError::Disconnected => write!(f, "receiving on a closed channel"), - } - } -} - -impl std::error::Error for TryRecvError {} - -#[derive(Debug)] -struct Slot { - /// The message. `None` if the slot is empty (initial state only). - msg: Option, - /// The absolute version of the message in this slot. - version: u64, -} - -struct Shared { - buffer: Box<[RwLock>]>, - capacity: usize, - mask: usize, - /// The next sequence after the contiguous prefix of fully published slots. - tail: AtomicU64, - /// Serializes senders and makes publishing atomic with draining or registering waiters. - state: Mutex, - /// Number of active senders. - senders: AtomicUsize, -} - -struct State { - /// Receivers waiting for a new message. - waiters: WaitSet, -} - -fn wake_waiters(wakers: impl IntoIterator) { - for waker in wakers { - waker.wake(); - } -} - -/// A sender handle to the broadcast channel. -/// -/// The sender can be cloned to create multiple producers. When all senders are dropped, -/// the channel is closed. -pub struct Sender { - shared: Arc>, -} - -impl Clone for Sender { - fn clone(&self) -> Self { - self.shared.senders.fetch_add(1, Ordering::Release); - Self { - shared: self.shared.clone(), - } - } -} - -impl Drop for Sender { - fn drop(&mut self) { - match self.shared.senders.fetch_sub(1, Ordering::AcqRel) { - 1 => { - // If this is the last sender, we need to wake up the receiver so it can - // observe the disconnected state. - let wakers = self.shared.state.lock().waiters.take_wakers(); - wake_waiters(wakers); - } - _ => { - // there are still other senders left, do nothing - } - } - } -} - -impl Sender { - /// Broadcasts a value to all active receivers. - /// - /// This operation is non-blocking. If the channel buffer is full, the oldest message - /// in the buffer is overwritten. Any receiver that was waiting for that overwritten - /// message will receive a [`RecvError::Lagged`] error on its next call to `recv`. - /// - /// # Examples - /// - /// ``` - /// use asyncband::broadcast::overflow; - /// - /// let (tx, mut rx) = overflow::channel(16); - /// tx.send(10); - /// assert_eq!(rx.try_recv(), Ok(10)); - /// ``` - pub fn send(&self, msg: T) { - let wakers = { - let mut state = self.shared.state.lock(); - let tail = self.shared.tail.load(Ordering::Relaxed); - let idx = (tail as usize) & self.shared.mask; - - let mut slot = self.shared.buffer[idx].write(); - slot.msg = Some(msg); - slot.version = tail; - - // Publish the completed slot before releasing its write lock. A receiver that sees the - // new tail either held the old slot lock first or waits until the new value is - // complete. - self.shared - .tail - .store(tail.wrapping_add(1), Ordering::Release); - drop(slot); - - state.waiters.take_wakers() - }; - - // Notify all waiting receivers. - wake_waiters(wakers); - } - - /// Creates a new receiver that starts receiving messages from the current tail of the channel. - /// - /// # Examples - /// - /// ``` - /// use asyncband::broadcast::overflow; - /// use asyncband::broadcast::overflow::TryRecvError; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let (tx, _) = overflow::channel(16); - /// tx.send(10); - /// - /// let mut rx = tx.subscribe(); - /// assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); - /// tx.send(20); - /// assert_eq!(rx.recv().await, Ok(20)); - /// # } - /// ``` - pub fn subscribe(&self) -> Receiver { - // Receiver starts at the current tail. - let head = self.shared.tail.load(Ordering::Acquire); - let shared = self.shared.clone(); - Receiver { shared, head } - } -} - -/// A receiver handle to the broadcast channel. -/// -/// The receiver can be cloned to create multiple consumers. Each receiver sees every -/// message sent to the channel (unless it lags behind). -pub struct Receiver { - shared: Arc>, - head: u64, -} - -impl Clone for Receiver { - fn clone(&self) -> Self { - Self { - shared: self.shared.clone(), - head: self.head, - } - } -} - -impl Receiver { - /// Receives the next value for this receiver. - /// - /// # Returns - /// - /// * `Ok(T)`: The next message. - /// * `Err(RecvError::Lagged(u64))`: The receiver lagged behind. The internal cursor is advanced - /// to the oldest available message. The count indicates how many messages were skipped. - /// * `Err(RecvError::Disconnected)`: All senders have been dropped and no more messages are - /// available. - /// - /// # Examples - /// - /// ``` - /// use asyncband::broadcast::overflow; - /// - /// # #[tokio::main] - /// # async fn main() { - /// let (tx, mut rx) = overflow::channel(16); - /// tx.send(10); - /// assert_eq!(rx.recv().await, Ok(10)); - /// # } - /// ``` - pub async fn recv(&mut self) -> Result { - Recv { - receiver: self, - registration: None, - } - .await - } - - /// Attempts to receive the next value for this receiver without blocking. - /// - /// # Returns - /// - /// * `Ok(T)`: The next message. - /// * `Err(TryRecvError::Empty)`: No message is currently available. - /// * `Err(TryRecvError::Lagged(u64))`: The receiver lagged behind. The internal cursor is - /// advanced to the oldest available message. The count indicates how many messages were - /// skipped. - /// * `Err(TryRecvError::Disconnected)`: All senders have been dropped and no more messages are - /// available. - /// - /// # Examples - /// - /// ``` - /// use asyncband::broadcast::overflow; - /// - /// let (tx, mut rx) = overflow::channel(16); - /// tx.send(10); - /// assert_eq!(rx.try_recv(), Ok(10)); - /// ``` - pub fn try_recv(&mut self) -> Result { - let shared = &self.shared; - let cap = shared.capacity as u64; - - loop { - let tail = shared.tail.load(Ordering::Acquire); - let head = self.head; - - // diff represents how far behind the head is from the tail. - let diff = tail.wrapping_sub(head); - - // 1. Check for Lag - if diff > cap { - let missed = diff - cap; - self.head = tail.wrapping_sub(cap); - return Err(TryRecvError::Lagged(missed)); - } - - // 2. Check if a message is available - if diff > 0 { - let idx = (head as usize) & shared.mask; - let slot = shared.buffer[idx].read(); - - if slot.version == head { - return if let Some(msg) = &slot.msg { - self.head = head.wrapping_add(1); - Ok(msg.clone()) - } else { - Err(TryRecvError::Empty) - }; - } - - drop(slot); - - // The slot may have been overwritten after the first tail snapshot. Publication - // happens while holding the slot write lock, so a fresh tail now includes that - // overwrite and produces an accurate lag count. - let tail = shared.tail.load(Ordering::Acquire); - let diff = tail.wrapping_sub(head); - if diff > cap { - let missed = diff - cap; - self.head = tail.wrapping_sub(cap); - return Err(TryRecvError::Lagged(missed)); - } - - return Err(TryRecvError::Empty); - } - - // 3. No message available (diff == 0). Check for Closed. - if shared.senders.load(Ordering::Acquire) == 0 { - // Observing the final sender drop synchronizes with all preceding sends, but the - // first tail snapshot predates that acquire. Reload it before declaring the - // channel drained so a published final message cannot be hidden by closure. - if shared.tail.load(Ordering::Acquire) != head { - continue; - } - return Err(TryRecvError::Disconnected); - } - - return Err(TryRecvError::Empty); - } - } -} - -impl Receiver { - /// Re-subscribes to the channel, returning a new receiver that starts receiving messages - /// from the *current* tail of the channel. - /// - /// This is useful if the receiver has lagged too far behind and wants to jump to the latest - /// message, skipping everything in between. - /// - /// # Examples - /// - /// ``` - /// use asyncband::broadcast::overflow; - /// - /// let (tx, mut rx) = overflow::channel(2); - /// tx.send(1); - /// tx.send(2); - /// - /// let mut rx2 = rx.resubscribe(); - /// tx.send(3); - /// - /// assert_eq!(rx2.try_recv(), Ok(3)); - /// ``` - pub fn resubscribe(&self) -> Self { - // Resubscribe starts at the current tail. - let head = self.shared.tail.load(Ordering::Acquire); - let shared = self.shared.clone(); - Self { shared, head } - } -} - -struct Recv<'a, T> { - receiver: &'a mut Receiver, - registration: Option, -} - -impl Future for Recv<'_, T> { - type Output = Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Self { - receiver, - registration, - } = self.get_mut(); - - loop { - // Senders publish data or closure before draining the current wake epoch. Once a - // result is observable, Drop does not need to lock the waiter set again. - match receiver.try_recv() { - Ok(val) => { - *registration = None; - return Poll::Ready(Ok(val)); - } - Err(TryRecvError::Lagged(n)) => { - *registration = None; - return Poll::Ready(Err(RecvError::Lagged(n))); - } - Err(TryRecvError::Disconnected) => { - *registration = None; - return Poll::Ready(Err(RecvError::Disconnected)); - } - Err(TryRecvError::Empty) => {} - } - - let shared = &receiver.shared; - let mut state = shared.state.lock(); - - // Double check tail to avoid race conditions. - if shared.tail.load(Ordering::Acquire) != receiver.head { - // New message arrived while acquiring the lock. Retry. - drop(state); - continue; - } - - // Check for Closed - // Use Acquire to ensure we see all writes before the sender dropped. - if shared.senders.load(Ordering::Acquire) == 0 { - *registration = None; - return Poll::Ready(Err(RecvError::Disconnected)); - } - - // Register Waker - let replaced_waker = state.waiters.register_waker(registration, cx); - drop(state); - drop(replaced_waker); - return Poll::Pending; - } - } -} - -impl Drop for Recv<'_, T> { - fn drop(&mut self) { - if self.registration.is_some() { - let removed_waker = { - let mut state = self.receiver.shared.state.lock(); - state.waiters.unregister_waker(&mut self.registration) - }; - drop(removed_waker); - } - } -} diff --git a/asyncband/src/broadcast/overflow/tests.rs b/asyncband/src/broadcast/overflow/tests.rs deleted file mode 100644 index da49b0c..0000000 --- a/asyncband/src/broadcast/overflow/tests.rs +++ /dev/null @@ -1,86 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::sync::atomic::Ordering; - -use super::*; - -// These tests stay next to the implementation because they inspect private state. - -#[tokio::test] -async fn sequence_number_wraparound() { - let (tx, mut rx) = channel(4); - let mut rx2 = rx.clone(); - - let boundary = u64::MAX - 2; - tx.shared.tail.store(boundary, Ordering::Release); - rx.head = boundary; - - tx.send(1); - assert_eq!(rx.recv().await, Ok(1)); - - for value in 2..=8 { - tx.send(value); - } - - assert_eq!(rx.recv().await, Err(RecvError::Lagged(3))); - for value in 5..=8 { - assert_eq!(rx.recv().await, Ok(value)); - } - - assert_eq!(rx2.recv().await, Err(RecvError::Lagged(1))); - for value in 5..=8 { - assert_eq!(rx2.recv().await, Ok(value)); - } -} - -#[tokio::test] -async fn sequence_number_wraparound_exactly_overwritten() { - let (tx, mut rx) = channel(4); - let mut rx2 = rx.clone(); - - let boundary = u64::MAX - 2; - tx.shared.tail.store(boundary, Ordering::Release); - rx.head = boundary; - - tx.send(1); - assert_eq!(rx.recv().await, Ok(1)); - - for value in 2..=5 { - tx.send(value); - } - - assert_eq!(rx.recv().await, Ok(2)); - // Wrapping the complete u64 space creates an ABA ambiguity. At 10^9 messages per second this - // takes roughly 584 years, so the implementation accepts it in favor of cheaper arithmetic. - assert_eq!(rx2.recv().await, Ok(4)); -} - -#[test] -fn capacity_is_rounded_to_a_power_of_two() { - let (tx, _) = channel::<()>(3); - assert_eq!(tx.shared.capacity, 4); - assert_eq!(tx.shared.mask, 3); - - let (tx, _) = channel::<()>(4); - assert_eq!(tx.shared.capacity, 4); - assert_eq!(tx.shared.mask, 3); - - let (tx, _) = channel::<()>(5); - assert_eq!(tx.shared.capacity, 8); - assert_eq!(tx.shared.mask, 7); -} diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 995db3c..caac4a2 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -20,7 +20,6 @@ pub(crate) mod atomic_option_box; #[cfg(any( feature = "barrier", - feature = "broadcast", feature = "latch", feature = "mpsc", feature = "mutex", @@ -45,7 +44,6 @@ pub(crate) mod once_table; #[cfg(any( feature = "barrier", - feature = "broadcast", feature = "latch", feature = "mpsc", feature = "mutex", @@ -55,9 +53,6 @@ pub(crate) mod once_table; ))] pub(crate) mod mutex; -#[cfg(feature = "broadcast")] -pub(crate) mod rwlock; - #[cfg(any( feature = "mpsc", feature = "mutex", @@ -80,12 +75,11 @@ pub(crate) mod waitlist; #[cfg(any( feature = "barrier", - feature = "broadcast", feature = "latch", feature = "once", feature = "waitgroup", ))] -// `barrier` constructs a wait set with `with_capacity`, while broadcast and countdown-based -// primitives use `new`. One constructor is therefore unused in every single-primitive build. +// `barrier` constructs a wait set with `with_capacity`, while countdown-based primitives use +// `new`. One constructor is therefore unused in every single-primitive build. #[allow(dead_code)] pub(crate) mod waitset; diff --git a/asyncband/src/internal/rwlock.rs b/asyncband/src/internal/rwlock.rs deleted file mode 100644 index 6d24dea..0000000 --- a/asyncband/src/internal/rwlock.rs +++ /dev/null @@ -1,56 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::sync::PoisonError; - -pub struct RwLock(std::sync::RwLock); - -impl RwLock { - pub const fn new(t: T) -> Self { - Self(std::sync::RwLock::new(t)) - } -} - -impl RwLock { - pub fn read(&self) -> std::sync::RwLockReadGuard<'_, T> { - self.0.read().unwrap_or_else(PoisonError::into_inner) - } - - pub fn write(&self) -> std::sync::RwLockWriteGuard<'_, T> { - self.0.write().unwrap_or_else(PoisonError::into_inner) - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - - use crate::internal::rwlock::RwLock; - - #[test] - fn test_poison_rwlock() { - let rwlock = Arc::new(RwLock::new(42)); - let r = rwlock.clone(); - let handle = std::thread::spawn(move || { - let _guard = r.write(); - panic!("poison"); - }); - let _ = handle.join(); - assert_eq!(*rwlock.read(), 42); - assert_eq!(*rwlock.write(), 42); - } -} diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 851434f..ea458cd 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -56,7 +56,7 @@ //! | Protect shared state | [`mutex::Mutex`], [`rwlock::RwLock`], [`condvar::Condvar`] | `mutex`, `rwlock`, `condvar` | //! | Initialize values once | [`once::Once`], [`once::OnceCell`], [`once::OnceMap`] | `once`, `once-cell`, `once-map` | //! | Coordinate tasks | [`barrier::Barrier`], [`latch::Latch`], [`waitgroup::WaitGroup`], [`shutdown`] | `barrier`, `latch`, `waitgroup`, `shutdown` | -//! | Send values | [`oneshot::channel`], [`mpsc::bounded`], [`mpsc::unbounded`], [`broadcast::overflow`] | `oneshot`, `mpsc`, `broadcast` | +//! | Send values | [`oneshot::channel`], [`mpsc::bounded`], [`mpsc::unbounded`] | `oneshot`, `mpsc` | //! | Reuse managed objects | [`pool::bounded`], [`pool::unbounded`] | `pool` | //! | Control workloads | [`semaphore::Semaphore`], [`singleflight::Group`] | `semaphore`, `singleflight` | //! | Wait from synchronous code | [`blocking::FutureExt`] | `blocking` | @@ -94,8 +94,6 @@ mod internal; pub mod barrier; #[cfg(feature = "blocking")] pub mod blocking; -#[cfg(feature = "broadcast")] -pub mod broadcast; #[cfg(feature = "condvar")] pub mod condvar; #[cfg(feature = "latch")] diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 15b6faa..14f1453 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -26,7 +26,6 @@ rust-version.workspace = true asyncband = { workspace = true, features = [ "barrier", "blocking", - "broadcast", "condvar", "latch", "mpsc", diff --git a/benchmarks/broadcast.rs b/benchmarks/broadcast.rs deleted file mode 100644 index 87d9870..0000000 --- a/benchmarks/broadcast.rs +++ /dev/null @@ -1,255 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::pin::pin; -use std::sync::Arc; -use std::sync::Barrier; -use std::thread; -use std::thread::JoinHandle; - -use asyncband::broadcast::overflow; -use divan::Bencher; -use divan::black_box; - -use super::support::bench_context; -use super::support::poll_pending; -use super::support::poll_pinned_ready; - -const RECEIVER_COUNTS: &[usize] = &[1, 8, 32]; -const CONCURRENCY_COUNTS: &[usize] = &[1, 2, 4, 8]; -const CONCURRENT_BATCH_SIZE: usize = 4096; - -struct ConcurrentSend { - _receiver: overflow::Receiver, - start: Arc, - done: Arc, - workers: Vec>, -} - -impl ConcurrentSend { - fn new(sender_count: usize) -> Self { - let (sender, receiver) = overflow::channel(CONCURRENT_BATCH_SIZE); - let ready = Arc::new(Barrier::new(sender_count + 1)); - let start = Arc::new(Barrier::new(sender_count + 1)); - let done = Arc::new(Barrier::new(sender_count + 1)); - let sends_per_worker = CONCURRENT_BATCH_SIZE / sender_count; - let mut workers = Vec::with_capacity(sender_count); - - for worker_index in 0..sender_count { - let sender = sender.clone(); - let ready = ready.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - ready.wait(); - start.wait(); - let first = worker_index * sends_per_worker; - for value in first..first + sends_per_worker { - sender.send(black_box(value)); - } - done.wait(); - })); - } - drop(sender); - ready.wait(); - - Self { - _receiver: receiver, - start, - done, - workers, - } - } - - fn run(&mut self) { - self.start.wait(); - self.done.wait(); - } -} - -impl Drop for ConcurrentSend { - fn drop(&mut self) { - for worker in self.workers.drain(..) { - worker.join().unwrap(); - } - } -} - -struct ConcurrentFanout { - sender: overflow::Sender, - start: Arc, - done: Arc, - workers: Vec>, -} - -impl ConcurrentFanout { - fn new(receiver_count: usize) -> Self { - let (sender, receiver) = overflow::channel(CONCURRENT_BATCH_SIZE); - let mut receivers = Vec::with_capacity(receiver_count); - receivers.push(receiver); - for _ in 1..receiver_count { - receivers.push(receivers[0].clone()); - } - - let ready = Arc::new(Barrier::new(receiver_count + 1)); - let start = Arc::new(Barrier::new(receiver_count + 1)); - let done = Arc::new(Barrier::new(receiver_count + 1)); - let mut workers = Vec::with_capacity(receiver_count); - - for mut receiver in receivers { - let ready = ready.clone(); - let start = start.clone(); - let done = done.clone(); - workers.push(thread::spawn(move || { - ready.wait(); - start.wait(); - let result = (0..CONCURRENT_BATCH_SIZE).try_for_each(|_| { - receiver.try_recv().map(|value| { - black_box(value); - }) - }); - done.wait(); - result.unwrap(); - })); - } - ready.wait(); - - Self { - sender, - start, - done, - workers, - } - } - - fn run(&mut self) { - for value in 0..CONCURRENT_BATCH_SIZE { - self.sender.send(black_box(value)); - } - self.start.wait(); - self.done.wait(); - } -} - -impl Drop for ConcurrentFanout { - fn drop(&mut self) { - for worker in self.workers.drain(..) { - worker.join().unwrap(); - } - } -} - -#[divan::bench] -fn send_overwrite(bencher: Bencher) { - let (sender, receiver) = overflow::channel::(1); - bencher.bench_local(|| sender.send(black_box(1))); - black_box(receiver); -} - -#[divan::bench] -fn try_recv_empty(bencher: Bencher) { - let (sender, mut receiver) = overflow::channel::(1); - bencher.bench_local(|| black_box(receiver.try_recv())); - black_box(sender); -} - -#[divan::bench] -fn send_and_try_recv(bencher: Bencher) { - let (sender, mut receiver) = overflow::channel(1); - bencher.bench_local(|| { - sender.send(black_box(1)); - black_box(receiver.try_recv().unwrap()) - }); -} - -#[divan::bench( - args = CONCURRENCY_COUNTS, - sample_count = 50, - sample_size = 1, - counters = [CONCURRENT_BATCH_SIZE] -)] -fn concurrent_send(bencher: Bencher, sender_count: usize) { - bencher - .with_inputs(|| ConcurrentSend::new(sender_count)) - .bench_local_refs(ConcurrentSend::run); -} - -#[divan::bench( - args = CONCURRENCY_COUNTS, - sample_count = 50, - sample_size = 1, - counters = [CONCURRENT_BATCH_SIZE] -)] -fn concurrent_fanout(bencher: Bencher, receiver_count: usize) { - bencher - .with_inputs(|| ConcurrentFanout::new(receiver_count)) - .bench_local_refs(ConcurrentFanout::run); -} - -#[divan::bench] -fn cancel_pending(bencher: Bencher) { - let mut context = bench_context(); - - bencher.bench_local(|| { - let (sender, mut receiver) = overflow::channel::(1); - { - let mut recv = pin!(receiver.recv()); - poll_pending(recv.as_mut(), &mut context); - } - black_box((sender, receiver)) - }); -} - -#[divan::bench] -fn deliver_to_waiter(bencher: Bencher) { - let mut context = bench_context(); - - bencher.bench_local(|| { - let (sender, mut receiver) = overflow::channel(1); - let mut recv = pin!(receiver.recv()); - poll_pending(recv.as_mut(), &mut context); - - sender.send(black_box(1usize)); - let value = poll_pinned_ready(recv.as_mut(), &mut context).unwrap(); - black_box(value) - }); -} - -#[divan::bench(args = RECEIVER_COUNTS)] -fn deliver_to_receiver_batch(bencher: Bencher, receiver_count: usize) { - let mut context = bench_context(); - - bencher.bench_local(|| { - let (sender, receiver) = overflow::channel(1); - let mut receivers = (0..receiver_count) - .map(|_| receiver.resubscribe()) - .collect::>(); - let mut recvs = receivers - .iter_mut() - .map(|receiver| Box::pin(receiver.recv())) - .collect::>(); - for recv in &mut recvs { - poll_pending(recv.as_mut(), &mut context); - } - - sender.send(black_box(1usize)); - for mut recv in recvs { - let value = poll_pinned_ready(recv.as_mut(), &mut context).unwrap(); - black_box(value); - } - }); -} diff --git a/benchmarks/main.rs b/benchmarks/main.rs index 43c5d32..9c3e2eb 100644 --- a/benchmarks/main.rs +++ b/benchmarks/main.rs @@ -17,7 +17,6 @@ mod barrier; mod blocking; -mod broadcast; mod condvar; mod latch; mod mpsc; diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 658dba3..27c0594 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -29,7 +29,6 @@ tokio = { workspace = true, features = ["full"] } asyncband = { workspace = true, features = [ "barrier", "blocking", - "broadcast", "condvar", "latch", "mpsc", diff --git a/tests-integration/tests/broadcast_test.rs b/tests-integration/tests/broadcast_test.rs deleted file mode 100644 index 56ab69d..0000000 --- a/tests-integration/tests/broadcast_test.rs +++ /dev/null @@ -1,350 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -use std::future::Future; -use std::sync::Arc; -use std::sync::Barrier; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; -use std::task::Context; -use std::task::Wake; -use std::task::Waker; -use std::thread; - -use asyncband::broadcast::overflow::*; - -struct TrackWake(AtomicUsize); - -impl Wake for TrackWake { - fn wake(self: Arc) { - self.0.fetch_add(1, Ordering::Relaxed); - } -} - -#[derive(Debug)] -struct PanicOnDrop { - value: u64, - panic: bool, - panicked: Arc, -} - -impl Clone for PanicOnDrop { - fn clone(&self) -> Self { - Self { - value: self.value, - panic: false, - panicked: self.panicked.clone(), - } - } -} - -impl Drop for PanicOnDrop { - fn drop(&mut self) { - if self.panic && !self.panicked.swap(true, Ordering::Relaxed) { - panic!("panic while replacing a broadcast slot"); - } - } -} - -#[tokio::test] -async fn test_broadcast_basic() { - let (tx, mut rx1) = channel(10); - let mut rx2 = rx1.clone(); - - tx.send(10); - tx.send(20); - - assert_eq!(rx1.recv().await, Ok(10)); - assert_eq!(rx1.recv().await, Ok(20)); - assert_eq!(rx2.recv().await, Ok(10)); - assert_eq!(rx2.recv().await, Ok(20)); -} - -#[tokio::test] -async fn test_broadcast_lagged() { - let (tx, mut rx) = channel(2); - - tx.send(1); - tx.send(2); - tx.send(3); - - // Overwrites 1. Rx lagged by 1 (missed msg '1'). - // Rx should return Lagged(1) and catch up to 2 (oldest valid). - assert_eq!(rx.recv().await, Err(RecvError::Lagged(1))); - assert_eq!(rx.recv().await, Ok(2)); - assert_eq!(rx.recv().await, Ok(3)); -} - -#[tokio::test] -async fn test_broadcast_lagged_multi() { - let (tx, mut rx) = channel(2); - - tx.send(1); - tx.send(2); - tx.send(3); - tx.send(4); - - // Overwrites 1 and 2. Missed 2 messages. - assert_eq!(rx.recv().await, Err(RecvError::Lagged(2))); - assert_eq!(rx.recv().await, Ok(3)); - assert_eq!(rx.recv().await, Ok(4)); -} - -#[tokio::test] -async fn test_broadcast_closed() { - let (tx, mut rx) = channel::<()>(10); - drop(tx); - assert_eq!(rx.recv().await, Err(RecvError::Disconnected)); -} - -#[tokio::test] -async fn test_wait_mechanism() { - let (tx, mut rx) = channel(10); - - let handle = tokio::spawn(async move { rx.recv().await }); - - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - tx.send(42); - - assert_eq!(handle.await.unwrap(), Ok(42)); -} - -#[test] -fn cancelled_recv_releases_its_waker() { - let (tx, mut rx) = channel::<()>(1); - let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); - let waker = Waker::from(tracker.clone()); - let baseline = Arc::strong_count(&tracker); - let mut context = Context::from_waker(&waker); - let mut recv = Box::pin(rx.recv()); - - assert!(recv.as_mut().poll(&mut context).is_pending()); - assert_eq!(Arc::strong_count(&tracker), baseline + 1); - - drop(recv); - assert_eq!(Arc::strong_count(&tracker), baseline); - - tx.send(()); - assert_eq!(tracker.0.load(Ordering::Relaxed), 0); - assert_eq!(rx.try_recv(), Ok(())); -} - -#[tokio::test] -async fn test_subscribe() { - let (tx, _rx) = channel(10); - let mut rx = tx.subscribe(); - - tx.send(100); - assert_eq!(rx.recv().await, Ok(100)); -} - -#[tokio::test] -async fn test_resubscribe() { - let (tx, mut rx) = channel(2); - - tx.send(1); - tx.send(2); - - let mut rx2 = rx.resubscribe(); - - // rx sees 1, 2 - // rx2 sees nothing yet (starts at tail=2) - - tx.send(3); - - assert_eq!(rx.recv().await, Err(RecvError::Lagged(1))); - assert_eq!(rx.recv().await, Ok(2)); - assert_eq!(rx2.recv().await, Ok(3)); -} - -#[tokio::test] -async fn test_try_recv() { - let (tx, mut rx) = channel(16); - - // Empty - assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); - - // Success - tx.send(10); - assert_eq!(rx.try_recv(), Ok(10)); - assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); - - // Closed - drop(tx); - assert_eq!(rx.try_recv(), Err(TryRecvError::Disconnected)); -} - -#[tokio::test] -async fn test_try_recv_lagged() { - let (tx, mut rx) = channel(2); - tx.send(1); - tx.send(2); - tx.send(3); - - assert_eq!(rx.try_recv(), Err(TryRecvError::Lagged(1))); - assert_eq!(rx.try_recv(), Ok(2)); - assert_eq!(rx.try_recv(), Ok(3)); - assert_eq!(rx.try_recv(), Err(TryRecvError::Empty)); -} - -#[test] -fn panicking_send_does_not_publish_an_unwritten_slot() { - let panicked = Arc::new(AtomicBool::new(false)); - let (tx, mut rx) = channel(1); - tx.send(PanicOnDrop { - value: 1, - panic: true, - panicked: panicked.clone(), - }); - - let received = rx.try_recv().unwrap(); - assert_eq!(received.value, 1); - drop(received); - - let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - tx.send(PanicOnDrop { - value: 2, - panic: false, - panicked: panicked.clone(), - }); - })); - assert!(result.is_err()); - assert!(panicked.load(Ordering::Relaxed)); - - assert!(matches!(rx.try_recv(), Err(TryRecvError::Empty))); - tx.send(PanicOnDrop { - value: 3, - panic: false, - panicked: panicked.clone(), - }); - assert_eq!(rx.try_recv().unwrap().value, 3); - - drop(tx); - assert!(matches!(rx.try_recv(), Err(TryRecvError::Disconnected))); -} - -#[test] -fn concurrent_overwrite_preserves_sequence_and_lag_count() { - const MESSAGE_COUNT: u64 = 200_000; - - let (tx, mut rx) = channel(2); - let producer = thread::spawn(move || { - for value in 0..MESSAGE_COUNT { - tx.send(value); - } - }); - - let mut next = 0_u64; - loop { - match rx.try_recv() { - Ok(value) => { - assert_eq!(value, next); - next = next.wrapping_add(1); - } - Err(TryRecvError::Lagged(missed)) => { - assert!(missed > 0); - next = next.wrapping_add(missed); - } - Err(TryRecvError::Empty) => thread::yield_now(), - Err(TryRecvError::Disconnected) => break, - } - } - - producer.join().unwrap(); - assert_eq!(next, MESSAGE_COUNT); -} - -#[test] -fn concurrent_receivers_observe_the_same_sequence() { - const MESSAGE_COUNT: usize = 4096; - const RECEIVER_COUNT: usize = 8; - - let (tx, receiver) = channel(MESSAGE_COUNT); - let mut receivers = Vec::with_capacity(RECEIVER_COUNT); - receivers.push(receiver); - for _ in 1..RECEIVER_COUNT { - receivers.push(receivers[0].clone()); - } - - let ready = Arc::new(Barrier::new(RECEIVER_COUNT + 1)); - let workers = receivers - .into_iter() - .map(|mut receiver| { - let ready = ready.clone(); - thread::spawn(move || { - ready.wait(); - let mut received = Vec::with_capacity(MESSAGE_COUNT); - loop { - match receiver.try_recv() { - Ok(value) => received.push(value), - Err(TryRecvError::Empty) => thread::yield_now(), - Err(TryRecvError::Disconnected) => return received, - Err(TryRecvError::Lagged(missed)) => { - panic!("receiver unexpectedly lagged by {missed}") - } - } - } - }) - }) - .collect::>(); - - ready.wait(); - for value in 0..MESSAGE_COUNT { - tx.send(value); - } - drop(tx); - - let expected = (0..MESSAGE_COUNT).collect::>(); - for worker in workers { - assert_eq!(worker.join().unwrap(), expected); - } -} - -#[tokio::test] -async fn test_multi_senders_concurrent() { - let (tx, mut rx) = channel(100); - let tx1 = tx.clone(); - let tx2 = tx.clone(); - - tokio::spawn(async move { - for i in 0..10 { - tx1.send(i); - } - }); - - tokio::spawn(async move { - for i in 10..20 { - tx2.send(i); - } - }); - - // Main tx can also send - for i in 20..30 { - tx.send(i); - } - drop(tx); - - let mut received = Vec::new(); - while let Ok(n) = rx.recv().await { - received.push(n); - } - received.sort(); - - let expected = (0..30).collect::>(); - assert_eq!(received, expected); -} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index f2e4a99..14d4f1a 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -16,7 +16,6 @@ // under the License. use asyncband::barrier::Barrier; -use asyncband::broadcast; use asyncband::condvar::Condvar; use asyncband::latch::Latch; use asyncband::mpsc; @@ -82,10 +81,6 @@ 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::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -130,10 +125,6 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); - assert_unpin::>(); - assert_unpin::>(); - assert_unpin::(); - assert_unpin::(); assert_unpin::>(); assert_unpin::>(); assert_unpin::>();