From 11d91b4cef3955b4528e67a744c815bbd9045624 Mon Sep 17 00:00:00 2001 From: QwQBiG Date: Fri, 28 Aug 2026 23:48:21 +0800 Subject: [PATCH] feat(watch): add a latest-state channel --- CHANGELOG.md | 1 + README.md | 1 + asyncband/Cargo.toml | 1 + asyncband/src/channel/mod.rs | 2 + asyncband/src/channel/watch/error.rs | 70 +++++ asyncband/src/channel/watch/mod.rs | 331 ++++++++++++++++++++++ asyncband/src/channel/watch/tests.rs | 78 +++++ asyncband/src/internal/mod.rs | 3 + asyncband/src/internal/waitset.rs | 6 + asyncband/src/lib.rs | 3 + benchmarks/Cargo.toml | 1 + benchmarks/asyncband/main.rs | 1 + benchmarks/asyncband/watch/mod.rs | 100 +++++++ benchmarks/ecosystem/main.rs | 1 + benchmarks/ecosystem/watch/adapters.rs | 94 ++++++ benchmarks/ecosystem/watch/mod.rs | 19 ++ benchmarks/ecosystem/watch/paths.rs | 112 ++++++++ tests-integration/Cargo.toml | 1 + tests-integration/tests/traits_test.rs | 13 + tests-integration/tests/watch_test.rs | 378 +++++++++++++++++++++++++ 20 files changed, 1216 insertions(+) create mode 100644 asyncband/src/channel/watch/error.rs create mode 100644 asyncband/src/channel/watch/mod.rs create mode 100644 asyncband/src/channel/watch/tests.rs create mode 100644 benchmarks/asyncband/watch/mod.rs create mode 100644 benchmarks/ecosystem/watch/adapters.rs create mode 100644 benchmarks/ecosystem/watch/mod.rs create mode 100644 benchmarks/ecosystem/watch/paths.rs create mode 100644 tests-integration/tests/watch_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 66b5f57..932d7e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ All notable changes to this project will be documented in this file. * Add an opt-in `asyncband::blocking::FutureExt` bridge with `block_on` and `wait_timeout` methods for waiting on runtime-agnostic futures from synchronous code. * Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`. * Add opt-in `asyncband::once::LazyCell` for values that own one asynchronous initializer and preserve its in-flight future across caller cancellation. +* Add an opt-in latest-state channel under `asyncband::watch`. ### Breaking changes diff --git a/README.md b/README.md index 1a65d56..1fbb23d 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | Channels | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value between two tasks. | | | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver. | | | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | +| | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Retain the latest state and coalesce intermediate updates. | | Resource reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | | Workload coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | | | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce concurrent calls for the same key. | diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index aafe547..9b831d7 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -62,6 +62,7 @@ semaphore = [] shutdown = ["latch", "waitgroup"] singleflight = ["dep:hashbrown", "once-cell"] waitgroup = [] +watch = [] [dependencies] hashbrown = { workspace = true, default-features = false, features = [ diff --git a/asyncband/src/channel/mod.rs b/asyncband/src/channel/mod.rs index bfe3be0..c2cdf58 100644 --- a/asyncband/src/channel/mod.rs +++ b/asyncband/src/channel/mod.rs @@ -21,3 +21,5 @@ pub mod broadcast; pub mod mpsc; #[cfg(feature = "oneshot")] pub mod oneshot; +#[cfg(feature = "watch")] +pub mod watch; diff --git a/asyncband/src/channel/watch/error.rs b/asyncband/src/channel/watch/error.rs new file mode 100644 index 0000000..1f0aa05 --- /dev/null +++ b/asyncband/src/channel/watch/error.rs @@ -0,0 +1,70 @@ +// 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::any::type_name; +use std::fmt; + +/// An error returned when sending on a watch channel without any receivers. +/// +/// The value that could not be sent can be retrieved with [`SendError::into_inner`]. +#[derive(Clone, PartialEq, Eq)] +pub struct SendError(T); + +impl SendError { + /// Returns a reference to the value that could not be sent. + pub fn as_inner(&self) -> &T { + &self.0 + } + + /// Consumes the error and returns the value that could not be sent. + pub fn into_inner(self) -> T { + self.0 + } + + pub(super) fn new(value: T) -> Self { + Self(value) + } +} + +impl fmt::Display for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("sending on a closed channel") + } +} + +impl fmt::Debug for SendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "SendError<{}>(..)", type_name::()) + } +} + +impl std::error::Error for SendError {} + +/// An error returned when every sender has disconnected and no unseen value remains. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RecvError { + /// Every sender has disconnected, so the current value will never change again. + Disconnected, +} + +impl fmt::Display for RecvError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("receiving on a closed channel") + } +} + +impl std::error::Error for RecvError {} diff --git a/asyncband/src/channel/watch/mod.rs b/asyncband/src/channel/watch/mod.rs new file mode 100644 index 0000000..2f2efda --- /dev/null +++ b/asyncband/src/channel/watch/mod.rs @@ -0,0 +1,331 @@ +// 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 channel that retains and distributes the latest state. +//! +//! Every receiver independently tracks whether it has observed the current value. Intermediate +//! updates may be coalesced, so a slow receiver observes the latest state rather than every update. +//! The receiver returned by [`channel`] considers the initial value observed, as does a receiver +//! created by [`Sender::subscribe`]. Cloning a receiver preserves the source receiver's observed +//! version and then tracks future observations independently. +//! +//! [`Receiver::borrow`] returns an owning [`Arc`] snapshot without marking the current version +//! observed. [`Receiver::borrow_and_update`] and [`Receiver::changed`] explicitly mark the returned +//! version observed. Because snapshots do not retain the channel's internal lock, they may be kept +//! or moved independently while senders continue publishing newer values. +//! +//! If every sender disconnects after publishing a final unseen value, each receiver can still +//! observe that value once before [`RecvError::Disconnected`] is reported. +//! +//! # Examples +//! +//! ``` +//! use asyncband::watch; +//! +//! # #[tokio::main] +//! # async fn main() { +//! let (tx, mut rx) = watch::channel(0); +//! tx.send(1).unwrap(); +//! tx.send(2).unwrap(); +//! +//! assert_eq!(*rx.changed().await.unwrap(), 2); +//! assert_eq!(rx.has_changed(), Ok(false)); +//! # } +//! ``` + +mod error; + +#[cfg(test)] +mod tests; + +use std::fmt; +use std::future::Future; +use std::mem; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Context; +use std::task::Poll; + +pub use self::error::RecvError; +pub use self::error::SendError; +use crate::internal::mutex::Mutex; +use crate::internal::waitset::WaitSet; +use crate::internal::waitset::WakerToken; + +/// Creates a watch channel with an initial value. +/// +/// The receiver returned by this function considers the initial value already observed. +/// +/// # Examples +/// +/// ``` +/// use asyncband::watch; +/// +/// let (_tx, rx) = watch::channel("ready"); +/// assert_eq!(&*rx.borrow(), &"ready"); +/// ``` +pub fn channel(initial: T) -> (Sender, Receiver) { + let shared = Arc::new(Shared { + state: Mutex::new(State { + value: Arc::new(initial), + version: 0, + senders: 1, + receivers: 1, + waiters: WaitSet::new(), + }), + }); + let sender = Sender { + shared: shared.clone(), + }; + let receiver = Receiver { shared, seen: 0 }; + (sender, receiver) +} + +struct Shared { + state: Mutex>, +} + +struct State { + value: Arc, + version: u64, + senders: usize, + receivers: usize, + waiters: WaitSet, +} + +/// A sending endpoint of a watch channel. +pub struct Sender { + shared: Arc>, +} + +impl Clone for Sender { + fn clone(&self) -> Self { + let mut state = self.shared.state.lock(); + state.senders = state + .senders + .checked_add(1) + .expect("watch sender count overflowed"); + drop(state); + Self { + shared: self.shared.clone(), + } + } +} + +impl fmt::Debug for Sender { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Sender").finish_non_exhaustive() + } +} + +impl Drop for Sender { + fn drop(&mut self) { + let wakers = { + let mut state = self.shared.state.lock(); + state.senders = state + .senders + .checked_sub(1) + .expect("watch sender count underflowed"); + (state.senders == 0 && !state.waiters.is_empty()).then(|| state.waiters.take_wakers()) + }; + if let Some(wakers) = wakers { + for waker in wakers { + waker.wake(); + } + } + } +} + +impl Sender { + /// Publishes a new current value. + /// + /// If no receivers remain, the value is returned and the retained value is left unchanged. + /// + /// # Panics + /// + /// Panics if the internal version counter overflows. + pub fn send(&self, value: T) -> Result<(), SendError> { + let (wakers, replaced) = { + let mut state = self.shared.state.lock(); + if state.receivers == 0 { + return Err(SendError::new(value)); + } + let version = state + .version + .checked_add(1) + .expect("watch channel version counter overflowed"); + let replaced = mem::replace(&mut state.value, Arc::new(value)); + state.version = version; + let wakers = (!state.waiters.is_empty()).then(|| state.waiters.take_wakers()); + (wakers, replaced) + }; + if let Some(wakers) = wakers { + for waker in wakers { + waker.wake(); + } + } + drop(replaced); + Ok(()) + } + + /// Creates a receiver that considers the current value already observed. + pub fn subscribe(&self) -> Receiver { + let mut state = self.shared.state.lock(); + state.receivers = state + .receivers + .checked_add(1) + .expect("watch receiver count overflowed"); + let seen = state.version; + drop(state); + Receiver { + shared: self.shared.clone(), + seen, + } + } + + /// Returns the number of active receivers. + pub fn receiver_count(&self) -> usize { + self.shared.state.lock().receivers + } +} + +/// A receiving endpoint of a watch channel. +/// +/// Each receiver independently tracks the latest version it has observed. +pub struct Receiver { + shared: Arc>, + seen: u64, +} + +impl Clone for Receiver { + fn clone(&self) -> Self { + let mut state = self.shared.state.lock(); + state.receivers = state + .receivers + .checked_add(1) + .expect("watch receiver count overflowed"); + drop(state); + Self { + shared: self.shared.clone(), + seen: self.seen, + } + } +} + +impl fmt::Debug for Receiver { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Receiver").finish_non_exhaustive() + } +} + +impl Drop for Receiver { + fn drop(&mut self) { + let mut state = self.shared.state.lock(); + state.receivers = state + .receivers + .checked_sub(1) + .expect("watch receiver count underflowed"); + } +} + +impl Receiver { + /// Returns a snapshot of the current value without marking it observed. + pub fn borrow(&self) -> Arc { + self.shared.state.lock().value.clone() + } + + /// Returns a snapshot of the current value and marks its version observed. + pub fn borrow_and_update(&mut self) -> Arc { + let state = self.shared.state.lock(); + self.seen = state.version; + state.value.clone() + } + + /// Returns whether a version newer than the last observed version exists. + /// + /// An unseen final version is reported before disconnection, even if every sender has already + /// been dropped. + pub fn has_changed(&self) -> Result { + let state = self.shared.state.lock(); + if state.version != self.seen { + Ok(true) + } else if state.senders == 0 { + Err(RecvError::Disconnected) + } else { + Ok(false) + } + } + + /// Waits for a newer version and returns its latest snapshot. + /// + /// Intermediate updates may be coalesced. This method is cancel safe: until it returns, no + /// version is marked observed by the call. + pub async fn changed(&mut self) -> Result, RecvError> { + Changed { + receiver: self, + registration: None, + } + .await + } + + /// Returns whether every sender has been dropped. + /// + /// This does not mark the current version observed, so it may return `true` while a final + /// unseen value is still available through [`Receiver::changed`]. + pub fn is_disconnected(&self) -> bool { + self.shared.state.lock().senders == 0 + } +} + +struct Changed<'a, T> { + receiver: &'a mut Receiver, + registration: Option, +} + +impl Future for Changed<'_, T> { + type Output = Result, RecvError>; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + let (poll, retired_waker) = { + let mut state = this.receiver.shared.state.lock(); + if state.version != this.receiver.seen { + let retired = state.waiters.unregister_waker(&mut this.registration); + this.receiver.seen = state.version; + (Poll::Ready(Ok(state.value.clone())), retired) + } else if state.senders == 0 { + let retired = state.waiters.unregister_waker(&mut this.registration); + (Poll::Ready(Err(RecvError::Disconnected)), retired) + } else { + let retired = state.waiters.register_waker(&mut this.registration, cx); + (Poll::Pending, retired) + } + }; + drop(retired_waker); + poll + } +} + +impl Drop for Changed<'_, T> { + fn drop(&mut self) { + let waker = { + let mut state = self.receiver.shared.state.lock(); + state.waiters.unregister_waker(&mut self.registration) + }; + drop(waker); + } +} diff --git a/asyncband/src/channel/watch/tests.rs b/asyncband/src/channel/watch/tests.rs new file mode 100644 index 0000000..688a472 --- /dev/null +++ b/asyncband/src/channel/watch/tests.rs @@ -0,0 +1,78 @@ +// 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 super::*; + +#[test] +#[should_panic(expected = "watch channel version counter overflowed")] +fn send_panics_on_version_overflow() { + let (tx, _rx) = channel(0); + tx.shared.state.lock().version = u64::MAX; + tx.send(1).unwrap(); +} + +#[test] +#[should_panic(expected = "watch sender count overflowed")] +fn sender_clone_panics_on_count_overflow() { + let (tx, _rx) = channel(0); + tx.shared.state.lock().senders = usize::MAX; + let _ = tx.clone(); +} + +#[test] +#[should_panic(expected = "watch receiver count overflowed")] +fn receiver_clone_panics_on_count_overflow() { + let (tx, rx) = channel(0); + tx.shared.state.lock().receivers = usize::MAX; + let _ = rx.clone(); +} + +#[test] +#[should_panic(expected = "watch receiver count overflowed")] +fn subscribe_panics_on_count_overflow() { + let (tx, _rx) = channel(0); + tx.shared.state.lock().receivers = usize::MAX; + let _ = tx.subscribe(); +} + +#[test] +fn concurrent_senders_commit_every_version() { + const SENDERS: usize = 4; + const SENDS_PER_THREAD: usize = 1_000; + + let (tx, rx) = channel(0); + let workers = (0..SENDERS) + .map(|sender_index| { + let tx = tx.clone(); + std::thread::spawn(move || { + for offset in 0..SENDS_PER_THREAD { + tx.send(sender_index * SENDS_PER_THREAD + offset).unwrap(); + } + }) + }) + .collect::>(); + + for worker in workers { + worker.join().unwrap(); + } + + assert_eq!( + tx.shared.state.lock().version, + (SENDERS * SENDS_PER_THREAD) as u64 + ); + assert!(*rx.borrow() < SENDERS * SENDS_PER_THREAD); +} diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 2374f39..e2676a8 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -27,6 +27,7 @@ pub(crate) mod atomic_waker; feature = "rwlock", feature = "semaphore", feature = "waitgroup", + feature = "watch", ))] // `WaitList` and `WaitSet` use different `Arena` operations. A single-primitive build therefore // leaves part of this shared API unused, while the all-feature build uses it. @@ -60,6 +61,7 @@ pub(crate) mod value_cell; feature = "rwlock", feature = "semaphore", feature = "waitgroup", + feature = "watch", ))] pub(crate) mod mutex; @@ -89,6 +91,7 @@ pub(crate) mod waitlist; feature = "latch", feature = "once", feature = "waitgroup", + feature = "watch", ))] // `barrier` constructs a wait set with `with_capacity`, while countdown-based primitives use // `new`. One constructor is therefore unused in every single-primitive build. diff --git a/asyncband/src/internal/waitset.rs b/asyncband/src/internal/waitset.rs index 39db636..8f8cd7b 100644 --- a/asyncband/src/internal/waitset.rs +++ b/asyncband/src/internal/waitset.rs @@ -55,6 +55,12 @@ impl WaitSet { } } + /// Returns whether no wakers are currently registered. + #[inline] + pub fn is_empty(&self) -> bool { + self.waiters.is_empty() + } + /// Takes all registered wakers as an owning iterator without waking them. #[inline] pub fn take_wakers(&mut self) -> impl Iterator + 'static { diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 2bbc468..4ee8c22 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -67,6 +67,7 @@ //! | Channels | [`oneshot`] | `oneshot` | Send one value between two tasks. | //! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver. | //! | | [`broadcast`] | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | +//! | | [`watch`] | `watch` | Retain the latest state and coalesce intermediate updates. | //! | Resource reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | //! | Workload coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Control concurrent access with permits. | //! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce concurrent calls for the same key. | @@ -149,6 +150,8 @@ pub mod shutdown; pub mod singleflight; #[cfg(feature = "waitgroup")] pub mod waitgroup; +#[cfg(feature = "watch")] +pub use self::channel::watch; #[cfg(all(test, any(feature = "once-map", feature = "singleflight")))] mod test_support; diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 15a724f..44accc9 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -43,6 +43,7 @@ asyncband = { workspace = true, features = [ "shutdown", "singleflight", "waitgroup", + "watch", ] } divan = { workspace = true } flume = { workspace = true, features = ["async"] } diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index c383615..980c875 100644 --- a/benchmarks/asyncband/main.rs +++ b/benchmarks/asyncband/main.rs @@ -32,6 +32,7 @@ mod shutdown; mod singleflight; mod support; mod waitgroup; +mod watch; fn main() { divan::main(); diff --git a/benchmarks/asyncband/watch/mod.rs b/benchmarks/asyncband/watch/mod.rs new file mode 100644 index 0000000..d8914c1 --- /dev/null +++ b/benchmarks/asyncband/watch/mod.rs @@ -0,0 +1,100 @@ +// 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 asyncband::watch; +use divan::Bencher; +use divan::black_box; + +use crate::support::bench_context; +use crate::support::poll_pending; +use crate::support::poll_pinned_ready; +use crate::support::poll_ready; + +const RECEIVER_COUNTS: &[usize] = &[1, 2, 4, 8, 32]; + +#[divan::bench] +fn borrow_current(bencher: Bencher) { + let (sender, receiver) = watch::channel(1usize); + bencher.bench_local(|| black_box(*receiver.borrow())); + black_box(sender); +} + +#[divan::bench] +fn send_and_borrow(bencher: Bencher) { + let (sender, receiver) = watch::channel(0usize); + bencher.bench_local(|| { + sender.send(black_box(1usize)).unwrap(); + black_box(*receiver.borrow()) + }); +} + +#[divan::bench] +fn send_and_borrow_and_update(bencher: Bencher) { + let (sender, mut receiver) = watch::channel(0usize); + bencher.bench_local(|| { + sender.send(black_box(1usize)).unwrap(); + black_box(*receiver.borrow_and_update()) + }); +} + +#[divan::bench] +fn ready_changed(bencher: Bencher) { + let mut context = bench_context(); + let (sender, mut receiver) = watch::channel(0usize); + bencher.bench_local(|| { + sender.send(black_box(1usize)).unwrap(); + black_box(*poll_ready(receiver.changed(), &mut context).unwrap()) + }); +} + +#[divan::bench] +fn notify_pending(bencher: Bencher) { + let mut context = bench_context(); + let (sender, mut receiver) = watch::channel(0usize); + bencher.bench_local(|| { + let mut changed = pin!(receiver.changed()); + poll_pending(changed.as_mut(), &mut context); + sender.send(black_box(1usize)).unwrap(); + black_box(*poll_pinned_ready(changed.as_mut(), &mut context).unwrap()) + }); +} + +#[divan::bench(args = RECEIVER_COUNTS)] +fn notify_pending_fanout(bencher: Bencher, receiver_count: usize) { + let mut context = bench_context(); + let (sender, first) = watch::channel(0usize); + let mut receivers = Vec::with_capacity(receiver_count); + receivers.push(first); + receivers.extend((1..receiver_count).map(|_| sender.subscribe())); + + bencher.bench_local(|| { + let mut changed = receivers + .iter_mut() + .map(|receiver| Box::pin(receiver.changed())) + .collect::>(); + for future in &mut changed { + poll_pending(future.as_mut(), &mut context); + } + + sender.send(black_box(1usize)).unwrap(); + for mut future in changed { + black_box(*poll_pinned_ready(future.as_mut(), &mut context).unwrap()); + } + }); +} diff --git a/benchmarks/ecosystem/main.rs b/benchmarks/ecosystem/main.rs index 6fd069a..29a744c 100644 --- a/benchmarks/ecosystem/main.rs +++ b/benchmarks/ecosystem/main.rs @@ -17,6 +17,7 @@ mod broadcast; mod mpsc; +mod watch; #[allow(dead_code)] #[path = "../asyncband/support.rs"] diff --git a/benchmarks/ecosystem/watch/adapters.rs b/benchmarks/ecosystem/watch/adapters.rs new file mode 100644 index 0000000..a351b78 --- /dev/null +++ b/benchmarks/ecosystem/watch/adapters.rs @@ -0,0 +1,94 @@ +// 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::pin::Pin; + +pub struct Asyncband; +pub struct Tokio; + +pub trait Watch: Send + Sync + 'static { + type Sender: Send + 'static; + type Receiver: Send + 'static; + + fn channel(receiver_count: usize) -> (Self::Sender, Vec); + fn send(sender: &Self::Sender, value: usize); + fn borrow(receiver: &Self::Receiver) -> usize; + fn borrow_and_update(receiver: &mut Self::Receiver) -> usize; + fn changed(receiver: &mut Self::Receiver) -> Pin + '_>>; +} + +impl Watch for Asyncband { + type Sender = asyncband::watch::Sender; + type Receiver = asyncband::watch::Receiver; + + fn channel(receiver_count: usize) -> (Self::Sender, Vec) { + let (sender, first) = asyncband::watch::channel(0); + let mut receivers = Vec::with_capacity(receiver_count); + receivers.push(first); + receivers.extend((1..receiver_count).map(|_| sender.subscribe())); + (sender, receivers) + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).unwrap(); + } + + fn borrow(receiver: &Self::Receiver) -> usize { + *receiver.borrow() + } + + fn borrow_and_update(receiver: &mut Self::Receiver) -> usize { + *receiver.borrow_and_update() + } + + fn changed(receiver: &mut Self::Receiver) -> Pin + '_>> { + Box::pin(async move { *receiver.changed().await.unwrap() }) + } +} + +impl Watch for Tokio { + type Sender = tokio::sync::watch::Sender; + type Receiver = tokio::sync::watch::Receiver; + + fn channel(receiver_count: usize) -> (Self::Sender, Vec) { + let (sender, first) = tokio::sync::watch::channel(0); + let mut receivers = Vec::with_capacity(receiver_count); + receivers.push(first); + receivers.extend((1..receiver_count).map(|_| sender.subscribe())); + (sender, receivers) + } + + fn send(sender: &Self::Sender, value: usize) { + sender.send(value).unwrap(); + } + + fn borrow(receiver: &Self::Receiver) -> usize { + *receiver.borrow() + } + + fn borrow_and_update(receiver: &mut Self::Receiver) -> usize { + *receiver.borrow_and_update() + } + + fn changed(receiver: &mut Self::Receiver) -> Pin + '_>> { + Box::pin(async move { + receiver.changed().await.unwrap(); + *receiver.borrow() + }) + } +} diff --git a/benchmarks/ecosystem/watch/mod.rs b/benchmarks/ecosystem/watch/mod.rs new file mode 100644 index 0000000..c9afe70 --- /dev/null +++ b/benchmarks/ecosystem/watch/mod.rs @@ -0,0 +1,19 @@ +// 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. + +mod adapters; +mod paths; diff --git a/benchmarks/ecosystem/watch/paths.rs b/benchmarks/ecosystem/watch/paths.rs new file mode 100644 index 0000000..f5df6d9 --- /dev/null +++ b/benchmarks/ecosystem/watch/paths.rs @@ -0,0 +1,112 @@ +// 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. + +// Asyncband returns an owning Arc snapshot, while Tokio returns a read-lock-backed Ref from its +// borrow methods and returns () from changed. These benchmarks immediately read a usize and release +// either representation. The changed adapter also reads Tokio's current value so both sides finish +// with the observed value, but their ownership and lock-lifetime contracts remain intentionally +// different. + +use std::future::Future; +use std::pin::Pin; +use std::task::Context; +use std::task::Poll; + +use divan::Bencher; +use divan::black_box; + +use super::adapters::Asyncband; +use super::adapters::Tokio; +use super::adapters::Watch; +use crate::support::bench_context; +use crate::support::poll_ready; + +const RECEIVER_COUNTS: &[usize] = &[1, 2, 4, 8, 32]; + +fn poll_erased_pending( + mut future: Pin<&mut dyn Future>, + context: &mut Context<'_>, +) { + assert!(future.as_mut().poll(context).is_pending()); +} + +fn poll_erased_ready( + mut future: Pin<&mut dyn Future>, + context: &mut Context<'_>, +) -> usize { + match future.as_mut().poll(context) { + Poll::Ready(output) => output, + Poll::Pending => panic!("benchmark future should be ready"), + } +} + +#[divan::bench(types = [Asyncband, Tokio])] +fn borrow_current(bencher: Bencher) { + let (sender, mut receivers) = C::channel(1); + let receiver = receivers.pop().unwrap(); + bencher.bench_local(|| black_box(C::borrow(&receiver))); + black_box(sender); +} + +#[divan::bench(types = [Asyncband, Tokio])] +fn send_and_borrow(bencher: Bencher) { + let (sender, mut receivers) = C::channel(1); + let receiver = receivers.pop().unwrap(); + bencher.bench_local(|| { + C::send(&sender, black_box(1)); + black_box(C::borrow(&receiver)) + }); +} + +#[divan::bench(types = [Asyncband, Tokio])] +fn send_and_borrow_and_update(bencher: Bencher) { + let (sender, mut receivers) = C::channel(1); + let mut receiver = receivers.pop().unwrap(); + bencher.bench_local(|| { + C::send(&sender, black_box(1)); + black_box(C::borrow_and_update(&mut receiver)) + }); +} + +#[divan::bench(types = [Asyncband, Tokio])] +fn ready_changed(bencher: Bencher) { + let mut context = bench_context(); + let (sender, mut receivers) = C::channel(1); + let mut receiver = receivers.pop().unwrap(); + bencher.bench_local(|| { + C::send(&sender, black_box(1)); + black_box(poll_ready(C::changed(&mut receiver), &mut context)) + }); +} + +#[divan::bench(types = [Asyncband, Tokio], args = RECEIVER_COUNTS)] +fn notify_pending_fanout(bencher: Bencher, receiver_count: usize) { + let mut context = bench_context(); + let (sender, mut receivers) = C::channel(receiver_count); + + bencher.bench_local(|| { + let mut changed = receivers.iter_mut().map(C::changed).collect::>(); + for future in &mut changed { + poll_erased_pending(future.as_mut(), &mut context); + } + + C::send(&sender, black_box(1)); + for mut future in changed { + black_box(poll_erased_ready(future.as_mut(), &mut context)); + } + }); +} diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index b66fda0..a43060d 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -45,6 +45,7 @@ asyncband = { workspace = true, features = [ "shutdown", "singleflight", "waitgroup", + "watch", ] } pollster = { workspace = true, features = ["macro"] } tokio-test = { workspace = true } diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 847de16..2a62838 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -43,6 +43,7 @@ use asyncband::shutdown::ShutdownWatch; use asyncband::singleflight; use asyncband::waitgroup::Wait; use asyncband::waitgroup::WaitGroup; +use asyncband::watch; struct PoolManager; @@ -102,16 +103,24 @@ 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::(); } #[test] fn movable_public_types_are_send() { fn assert_send() {} + fn assert_send_value(_: T) {} assert_send::>>(); assert_send::>(); assert_send::>(); assert_send::>>(); + + let (_tx, mut rx) = watch::channel(0); + assert_send_value(rx.changed()); } #[test] @@ -154,6 +163,10 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); + assert_unpin::>(); + assert_unpin::(); } #[test] diff --git a/tests-integration/tests/watch_test.rs b/tests-integration/tests/watch_test.rs new file mode 100644 index 0000000..3aac616 --- /dev/null +++ b/tests-integration/tests/watch_test.rs @@ -0,0 +1,378 @@ +// 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::pin::Pin; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; +use std::task::Wake; +use std::task::Waker; +use std::thread; +use std::time::Duration; + +use asyncband::watch; + +struct TrackWake(AtomicUsize); + +impl Wake for TrackWake { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } +} + +struct WakeCallback(Mutex>>); + +impl Wake for WakeCallback { + fn wake(self: Arc) { + let callback = self.0.lock().unwrap().take(); + if let Some(callback) = callback { + callback(); + } + } +} + +struct DropCallbackWake(Mutex>>); + +// This test needs a custom waker whose final `Arc` drop is observable. +#[allow(clippy::manual_noop_waker)] +impl Wake for DropCallbackWake { + fn wake(self: Arc) {} +} + +impl Drop for DropCallbackWake { + fn drop(&mut self) { + if let Some(callback) = self.0.get_mut().unwrap().take() { + callback(); + } + } +} + +struct ReentrantDrop(Option>); + +impl Drop for ReentrantDrop { + fn drop(&mut self) { + if let Some(sender) = &self.0 { + let _ = sender.receiver_count(); + } + } +} + +fn poll_with(future: Pin<&mut F>, waker: &Waker) -> Poll { + future.poll(&mut Context::from_waker(waker)) +} + +#[test] +fn initial_value_is_observed_and_updates_coalesce() { + let (tx, mut rx) = watch::channel(0); + + assert_eq!(*rx.borrow(), 0); + assert_eq!(rx.has_changed(), Ok(false)); + + tx.send(1).unwrap(); + tx.send(2).unwrap(); + + assert_eq!(rx.has_changed(), Ok(true)); + assert_eq!(*pollster::block_on(rx.changed()).unwrap(), 2); + assert_eq!(rx.has_changed(), Ok(false)); +} + +#[test] +fn equal_values_still_create_a_new_version() { + let (tx, mut rx) = watch::channel(1); + + tx.send(1).unwrap(); + + assert_eq!(rx.has_changed(), Ok(true)); + assert_eq!(*pollster::block_on(rx.changed()).unwrap(), 1); +} + +#[test] +fn borrow_does_not_consume_but_borrow_and_update_does() { + let (tx, mut rx) = watch::channel(0); + tx.send(1).unwrap(); + + assert_eq!(*rx.borrow(), 1); + assert_eq!(rx.has_changed(), Ok(true)); + assert_eq!(*rx.borrow_and_update(), 1); + assert_eq!(rx.has_changed(), Ok(false)); +} + +#[test] +fn cloned_receivers_inherit_then_advance_independently() { + let (tx, mut first) = watch::channel(0); + tx.send(1).unwrap(); + let mut second = first.clone(); + + assert_eq!(*pollster::block_on(first.changed()).unwrap(), 1); + assert_eq!(first.has_changed(), Ok(false)); + assert_eq!(second.has_changed(), Ok(true)); + assert_eq!(*pollster::block_on(second.changed()).unwrap(), 1); + + tx.send(2).unwrap(); + assert_eq!(*first.borrow_and_update(), 2); + assert_eq!(second.has_changed(), Ok(true)); +} + +#[test] +fn subscriptions_start_at_the_current_version() { + let (tx, _rx) = watch::channel(0); + tx.send(1).unwrap(); + let mut subscribed = tx.subscribe(); + + assert_eq!(*subscribed.borrow(), 1); + assert_eq!(subscribed.has_changed(), Ok(false)); + + tx.send(2).unwrap(); + assert_eq!(*pollster::block_on(subscribed.changed()).unwrap(), 2); +} + +#[test] +fn final_unseen_value_is_reported_before_disconnection() { + let (tx, mut first) = watch::channel(0); + let mut second = first.clone(); + tx.send(1).unwrap(); + drop(tx); + + assert!(first.is_disconnected()); + assert_eq!(first.has_changed(), Ok(true)); + assert_eq!(*first.borrow(), 1); + assert_eq!(first.has_changed(), Ok(true)); + assert_eq!(*pollster::block_on(first.changed()).unwrap(), 1); + assert_eq!(first.has_changed(), Err(watch::RecvError::Disconnected)); + + assert_eq!(*pollster::block_on(second.changed()).unwrap(), 1); + assert_eq!( + pollster::block_on(second.changed()), + Err(watch::RecvError::Disconnected) + ); +} + +#[test] +fn sending_without_receivers_returns_the_value_and_preserves_current() { + let (tx, rx) = watch::channel(String::from("initial")); + assert_eq!(tx.receiver_count(), 1); + drop(rx); + assert_eq!(tx.receiver_count(), 0); + + let error = tx.send(String::from("unsent")).unwrap_err(); + assert_eq!(error.as_inner(), "unsent"); + assert_eq!(error.into_inner(), "unsent"); + + let mut replacement = tx.subscribe(); + assert_eq!(tx.receiver_count(), 1); + assert_eq!(&*replacement.borrow(), "initial"); + assert_eq!(replacement.has_changed(), Ok(false)); + + tx.send(String::from("accepted")).unwrap(); + assert_eq!(&*replacement.borrow_and_update(), "accepted"); +} + +#[test] +fn cancelling_changed_releases_its_waker_without_consuming() { + let (tx, mut rx) = watch::channel(0); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let baseline = Arc::strong_count(&tracker); + let mut changed = Box::pin(rx.changed()); + + assert!(poll_with(changed.as_mut(), &waker).is_pending()); + assert_eq!(Arc::strong_count(&tracker), baseline + 1); + drop(changed); + assert_eq!(Arc::strong_count(&tracker), baseline); + + tx.send(1).unwrap(); + assert_eq!(tracker.0.load(Ordering::Relaxed), 0); + assert_eq!(*pollster::block_on(rx.changed()).unwrap(), 1); +} + +#[test] +fn cancelling_after_wake_still_leaves_the_update_unseen() { + let (tx, mut rx) = watch::channel(0); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut changed = Box::pin(rx.changed()); + + assert!(poll_with(changed.as_mut(), &waker).is_pending()); + tx.send(1).unwrap(); + assert_eq!(tracker.0.load(Ordering::Relaxed), 1); + drop(changed); + + assert_eq!(*pollster::block_on(rx.changed()).unwrap(), 1); +} + +#[test] +fn one_update_wakes_every_waiting_receiver_once() { + let (tx, mut first) = watch::channel(0); + let mut second = first.clone(); + let first_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let second_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first_tracker.clone()); + let second_waker = Waker::from(second_tracker.clone()); + let mut first_changed = Box::pin(first.changed()); + let mut second_changed = Box::pin(second.changed()); + + assert!(poll_with(first_changed.as_mut(), &first_waker).is_pending()); + assert!(poll_with(second_changed.as_mut(), &second_waker).is_pending()); + + tx.send(1).unwrap(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 1); + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 1); + assert_eq!( + poll_with(first_changed.as_mut(), &first_waker), + Poll::Ready(Ok(Arc::new(1))) + ); + assert_eq!( + poll_with(second_changed.as_mut(), &second_waker), + Poll::Ready(Ok(Arc::new(1))) + ); +} + +#[test] +fn only_the_last_sender_drop_wakes_a_waiter() { + let (tx, mut rx) = watch::channel(()); + let other = tx.clone(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut changed = Box::pin(rx.changed()); + + assert!(poll_with(changed.as_mut(), &waker).is_pending()); + drop(tx); + assert_eq!(tracker.0.load(Ordering::Relaxed), 0); + drop(other); + assert_eq!(tracker.0.load(Ordering::Relaxed), 1); + assert_eq!( + poll_with(changed.as_mut(), &waker), + Poll::Ready(Err(watch::RecvError::Disconnected)) + ); +} + +#[test] +fn dropping_a_stale_changed_future_keeps_a_new_waiter_registered() { + let (tx, mut first) = watch::channel(0); + let mut second = first.clone(); + let first_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first_tracker.clone()); + let mut first_changed = Box::pin(first.changed()); + + assert!(poll_with(first_changed.as_mut(), &first_waker).is_pending()); + tx.send(1).unwrap(); + assert_eq!(first_tracker.0.load(Ordering::Relaxed), 1); + + assert_eq!(*second.borrow_and_update(), 1); + let second_tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let second_waker = Waker::from(second_tracker.clone()); + let mut second_changed = Box::pin(second.changed()); + assert!(poll_with(second_changed.as_mut(), &second_waker).is_pending()); + + drop(first_changed); + tx.send(2).unwrap(); + + assert_eq!(second_tracker.0.load(Ordering::Relaxed), 1); + assert_eq!( + poll_with(second_changed.as_mut(), &second_waker), + Poll::Ready(Ok(Arc::new(2))) + ); +} + +#[test] +fn replaced_values_are_dropped_outside_the_channel_lock() { + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); + let worker = thread::spawn(move || { + let (tx, _rx) = watch::channel(ReentrantDrop(None)); + tx.send(ReentrantDrop(Some(tx.clone()))).unwrap(); + tx.send(ReentrantDrop(None)).unwrap(); + finished_tx.send(()).unwrap(); + }); + + finished_rx + .recv_timeout(Duration::from_secs(10)) + .expect("value destructor deadlocked against the watch lock"); + worker.join().unwrap(); +} + +#[test] +fn wake_callbacks_run_outside_the_channel_lock() { + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); + let worker = thread::spawn(move || { + let (tx, mut rx) = watch::channel(0); + let callback_sender = tx.clone(); + let waker = Waker::from(Arc::new(WakeCallback(Mutex::new(Some(Box::new( + move || { + let _ = callback_sender.receiver_count(); + }, + )))))); + let mut changed = Box::pin(rx.changed()); + + assert!(poll_with(changed.as_mut(), &waker).is_pending()); + tx.send(1).unwrap(); + assert_eq!( + poll_with(changed.as_mut(), &waker), + Poll::Ready(Ok(Arc::new(1))) + ); + drop(changed); + + let observer = rx.clone(); + let waker = Waker::from(Arc::new(WakeCallback(Mutex::new(Some(Box::new( + move || { + assert!(observer.is_disconnected()); + }, + )))))); + let mut changed = Box::pin(rx.changed()); + assert!(poll_with(changed.as_mut(), &waker).is_pending()); + drop(tx); + assert_eq!( + poll_with(changed.as_mut(), &waker), + Poll::Ready(Err(watch::RecvError::Disconnected)) + ); + finished_tx.send(()).unwrap(); + }); + + finished_rx + .recv_timeout(Duration::from_secs(10)) + .expect("wake callback deadlocked against the watch lock"); + worker.join().unwrap(); +} + +#[test] +fn replaced_wakers_are_dropped_outside_the_channel_lock() { + let (finished_tx, finished_rx) = std::sync::mpsc::channel(); + let worker = thread::spawn(move || { + let (tx, mut rx) = watch::channel(0); + let callback_sender = tx.clone(); + let old_waker = Waker::from(Arc::new(DropCallbackWake(Mutex::new(Some(Box::new( + move || { + let _ = callback_sender.receiver_count(); + }, + )))))); + let mut changed = Box::pin(rx.changed()); + assert!(poll_with(changed.as_mut(), &old_waker).is_pending()); + drop(old_waker); + + let replacement = Waker::from(Arc::new(TrackWake(AtomicUsize::new(0)))); + assert!(poll_with(changed.as_mut(), &replacement).is_pending()); + finished_tx.send(()).unwrap(); + }); + + finished_rx + .recv_timeout(Duration::from_secs(10)) + .expect("replaced waker destructor deadlocked against the watch lock"); + worker.join().unwrap(); +}