1 | #![cfg_attr (not(feature = "sync" ), allow(dead_code, unreachable_pub))] |
2 | |
3 | //! A single-producer, multi-consumer channel that only retains the *last* sent |
4 | //! value. |
5 | //! |
6 | //! This channel is useful for watching for changes to a value from multiple |
7 | //! points in the code base, for example, changes to configuration values. |
8 | //! |
9 | //! # Usage |
10 | //! |
11 | //! [`channel`] returns a [`Sender`] / [`Receiver`] pair. These are the producer |
12 | //! and consumer halves of the channel. The channel is created with an initial |
13 | //! value. |
14 | //! |
15 | //! Each [`Receiver`] independently tracks the last value *seen* by its caller. |
16 | //! |
17 | //! To access the **current** value stored in the channel and mark it as *seen* |
18 | //! by a given [`Receiver`], use [`Receiver::borrow_and_update()`]. |
19 | //! |
20 | //! To access the current value **without** marking it as *seen*, use |
21 | //! [`Receiver::borrow()`]. (If the value has already been marked *seen*, |
22 | //! [`Receiver::borrow()`] is equivalent to [`Receiver::borrow_and_update()`].) |
23 | //! |
24 | //! For more information on when to use these methods, see |
25 | //! [here](#borrow_and_update-versus-borrow). |
26 | //! |
27 | //! ## Change notifications |
28 | //! |
29 | //! The [`Receiver`] half provides an asynchronous [`changed`] method. This |
30 | //! method is ready when a new, *unseen* value is sent via the [`Sender`] half. |
31 | //! |
32 | //! * [`Receiver::changed()`] returns `Ok(())` on receiving a new value, or |
33 | //! `Err(`[`error::RecvError`]`)` if the [`Sender`] has been dropped. |
34 | //! * If the current value is *unseen* when calling [`changed`], then |
35 | //! [`changed`] will return immediately. If the current value is *seen*, then |
36 | //! it will sleep until either a new message is sent via the [`Sender`] half, |
37 | //! or the [`Sender`] is dropped. |
38 | //! * On completion, the [`changed`] method marks the new value as *seen*. |
39 | //! * At creation, the initial value is considered *seen*. In other words, |
40 | //! [`Receiver::changed()`] will not return until a subsequent value is sent. |
41 | //! * New [`Receiver`] instances can be created with [`Sender::subscribe()`]. |
42 | //! The current value at the time the [`Receiver`] is created is considered |
43 | //! *seen*. |
44 | //! |
45 | //! ## `borrow_and_update` versus `borrow` |
46 | //! |
47 | //! If the receiver intends to await notifications from [`changed`] in a loop, |
48 | //! [`Receiver::borrow_and_update()`] should be preferred over |
49 | //! [`Receiver::borrow()`]. This avoids a potential race where a new value is |
50 | //! sent between [`changed`] being ready and the value being read. (If |
51 | //! [`Receiver::borrow()`] is used, the loop may run twice with the same value.) |
52 | //! |
53 | //! If the receiver is only interested in the current value, and does not intend |
54 | //! to wait for changes, then [`Receiver::borrow()`] can be used. It may be more |
55 | //! convenient to use [`borrow`](Receiver::borrow) since it's an `&self` |
56 | //! method---[`borrow_and_update`](Receiver::borrow_and_update) requires `&mut |
57 | //! self`. |
58 | //! |
59 | //! # Examples |
60 | //! |
61 | //! The following example prints `hello! world! `. |
62 | //! |
63 | //! ``` |
64 | //! use tokio::sync::watch; |
65 | //! use tokio::time::{Duration, sleep}; |
66 | //! |
67 | //! # async fn dox() -> Result<(), Box<dyn std::error::Error>> { |
68 | //! let (tx, mut rx) = watch::channel("hello" ); |
69 | //! |
70 | //! tokio::spawn(async move { |
71 | //! // Use the equivalent of a "do-while" loop so the initial value is |
72 | //! // processed before awaiting the `changed()` future. |
73 | //! loop { |
74 | //! println!("{}! " , *rx.borrow_and_update()); |
75 | //! if rx.changed().await.is_err() { |
76 | //! break; |
77 | //! } |
78 | //! } |
79 | //! }); |
80 | //! |
81 | //! sleep(Duration::from_millis(100)).await; |
82 | //! tx.send("world" )?; |
83 | //! # Ok(()) |
84 | //! # } |
85 | //! ``` |
86 | //! |
87 | //! # Closing |
88 | //! |
89 | //! [`Sender::is_closed`] and [`Sender::closed`] allow the producer to detect |
90 | //! when all [`Receiver`] handles have been dropped. This indicates that there |
91 | //! is no further interest in the values being produced and work can be stopped. |
92 | //! |
93 | //! The value in the channel will not be dropped until the sender and all |
94 | //! receivers have been dropped. |
95 | //! |
96 | //! # Thread safety |
97 | //! |
98 | //! Both [`Sender`] and [`Receiver`] are thread safe. They can be moved to other |
99 | //! threads and can be used in a concurrent environment. Clones of [`Receiver`] |
100 | //! handles may be moved to separate threads and also used concurrently. |
101 | //! |
102 | //! [`Sender`]: crate::sync::watch::Sender |
103 | //! [`Receiver`]: crate::sync::watch::Receiver |
104 | //! [`changed`]: crate::sync::watch::Receiver::changed |
105 | //! [`Receiver::changed()`]: crate::sync::watch::Receiver::changed |
106 | //! [`Receiver::borrow()`]: crate::sync::watch::Receiver::borrow |
107 | //! [`Receiver::borrow_and_update()`]: |
108 | //! crate::sync::watch::Receiver::borrow_and_update |
109 | //! [`channel`]: crate::sync::watch::channel |
110 | //! [`Sender::is_closed`]: crate::sync::watch::Sender::is_closed |
111 | //! [`Sender::closed`]: crate::sync::watch::Sender::closed |
112 | //! [`Sender::subscribe()`]: crate::sync::watch::Sender::subscribe |
113 | |
114 | use crate::sync::notify::Notify; |
115 | |
116 | use crate::loom::sync::atomic::AtomicUsize; |
117 | use crate::loom::sync::atomic::Ordering::Relaxed; |
118 | use crate::loom::sync::{Arc, RwLock, RwLockReadGuard}; |
119 | use std::fmt; |
120 | use std::mem; |
121 | use std::ops; |
122 | use std::panic; |
123 | |
124 | /// Receives values from the associated [`Sender`](struct@Sender). |
125 | /// |
126 | /// Instances are created by the [`channel`](fn@channel) function. |
127 | /// |
128 | /// To turn this receiver into a `Stream`, you can use the [`WatchStream`] |
129 | /// wrapper. |
130 | /// |
131 | /// [`WatchStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.WatchStream.html |
132 | #[derive(Debug)] |
133 | pub struct Receiver<T> { |
134 | /// Pointer to the shared state |
135 | shared: Arc<Shared<T>>, |
136 | |
137 | /// Last observed version |
138 | version: Version, |
139 | } |
140 | |
141 | /// Sends values to the associated [`Receiver`](struct@Receiver). |
142 | /// |
143 | /// Instances are created by the [`channel`](fn@channel) function. |
144 | #[derive(Debug)] |
145 | pub struct Sender<T> { |
146 | shared: Arc<Shared<T>>, |
147 | } |
148 | |
149 | /// Returns a reference to the inner value. |
150 | /// |
151 | /// Outstanding borrows hold a read lock on the inner value. This means that |
152 | /// long-lived borrows could cause the producer half to block. It is recommended |
153 | /// to keep the borrow as short-lived as possible. Additionally, if you are |
154 | /// running in an environment that allows `!Send` futures, you must ensure that |
155 | /// the returned `Ref` type is never held alive across an `.await` point, |
156 | /// otherwise, it can lead to a deadlock. |
157 | /// |
158 | /// The priority policy of the lock is dependent on the underlying lock |
159 | /// implementation, and this type does not guarantee that any particular policy |
160 | /// will be used. In particular, a producer which is waiting to acquire the lock |
161 | /// in `send` might or might not block concurrent calls to `borrow`, e.g.: |
162 | /// |
163 | /// <details><summary>Potential deadlock example</summary> |
164 | /// |
165 | /// ```text |
166 | /// // Task 1 (on thread A) | // Task 2 (on thread B) |
167 | /// let _ref1 = rx.borrow(); | |
168 | /// | // will block |
169 | /// | let _ = tx.send(()); |
170 | /// // may deadlock | |
171 | /// let _ref2 = rx.borrow(); | |
172 | /// ``` |
173 | /// </details> |
174 | #[derive(Debug)] |
175 | pub struct Ref<'a, T> { |
176 | inner: RwLockReadGuard<'a, T>, |
177 | has_changed: bool, |
178 | } |
179 | |
180 | impl<'a, T> Ref<'a, T> { |
181 | /// Indicates if the borrowed value is considered as _changed_ since the last |
182 | /// time it has been marked as seen. |
183 | /// |
184 | /// Unlike [`Receiver::has_changed()`], this method does not fail if the channel is closed. |
185 | /// |
186 | /// When borrowed from the [`Sender`] this function will always return `false`. |
187 | /// |
188 | /// # Examples |
189 | /// |
190 | /// ``` |
191 | /// use tokio::sync::watch; |
192 | /// |
193 | /// #[tokio::main] |
194 | /// async fn main() { |
195 | /// let (tx, mut rx) = watch::channel("hello" ); |
196 | /// |
197 | /// tx.send("goodbye" ).unwrap(); |
198 | /// // The sender does never consider the value as changed. |
199 | /// assert!(!tx.borrow().has_changed()); |
200 | /// |
201 | /// // Drop the sender immediately, just for testing purposes. |
202 | /// drop(tx); |
203 | /// |
204 | /// // Even if the sender has already been dropped... |
205 | /// assert!(rx.has_changed().is_err()); |
206 | /// // ...the modified value is still readable and detected as changed. |
207 | /// assert_eq!(*rx.borrow(), "goodbye" ); |
208 | /// assert!(rx.borrow().has_changed()); |
209 | /// |
210 | /// // Read the changed value and mark it as seen. |
211 | /// { |
212 | /// let received = rx.borrow_and_update(); |
213 | /// assert_eq!(*received, "goodbye" ); |
214 | /// assert!(received.has_changed()); |
215 | /// // Release the read lock when leaving this scope. |
216 | /// } |
217 | /// |
218 | /// // Now the value has already been marked as seen and could |
219 | /// // never be modified again (after the sender has been dropped). |
220 | /// assert!(!rx.borrow().has_changed()); |
221 | /// } |
222 | /// ``` |
223 | pub fn has_changed(&self) -> bool { |
224 | self.has_changed |
225 | } |
226 | } |
227 | |
228 | struct Shared<T> { |
229 | /// The most recent value. |
230 | value: RwLock<T>, |
231 | |
232 | /// The current version. |
233 | /// |
234 | /// The lowest bit represents a "closed" state. The rest of the bits |
235 | /// represent the current version. |
236 | state: AtomicState, |
237 | |
238 | /// Tracks the number of `Receiver` instances. |
239 | ref_count_rx: AtomicUsize, |
240 | |
241 | /// Notifies waiting receivers that the value changed. |
242 | notify_rx: big_notify::BigNotify, |
243 | |
244 | /// Notifies any task listening for `Receiver` dropped events. |
245 | notify_tx: Notify, |
246 | } |
247 | |
248 | impl<T: fmt::Debug> fmt::Debug for Shared<T> { |
249 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
250 | let state = self.state.load(); |
251 | f.debug_struct("Shared" ) |
252 | .field("value" , &self.value) |
253 | .field("version" , &state.version()) |
254 | .field("is_closed" , &state.is_closed()) |
255 | .field("ref_count_rx" , &self.ref_count_rx) |
256 | .finish() |
257 | } |
258 | } |
259 | |
260 | pub mod error { |
261 | //! Watch error types. |
262 | |
263 | use std::error::Error; |
264 | use std::fmt; |
265 | |
266 | /// Error produced when sending a value fails. |
267 | #[derive(PartialEq, Eq, Clone, Copy)] |
268 | pub struct SendError<T>(pub T); |
269 | |
270 | // ===== impl SendError ===== |
271 | |
272 | impl<T> fmt::Debug for SendError<T> { |
273 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { |
274 | f.debug_struct("SendError" ).finish_non_exhaustive() |
275 | } |
276 | } |
277 | |
278 | impl<T> fmt::Display for SendError<T> { |
279 | fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { |
280 | write!(fmt, "channel closed" ) |
281 | } |
282 | } |
283 | |
284 | impl<T> Error for SendError<T> {} |
285 | |
286 | /// Error produced when receiving a change notification. |
287 | #[derive(Debug, Clone)] |
288 | pub struct RecvError(pub(super) ()); |
289 | |
290 | // ===== impl RecvError ===== |
291 | |
292 | impl fmt::Display for RecvError { |
293 | fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { |
294 | write!(fmt, "channel closed" ) |
295 | } |
296 | } |
297 | |
298 | impl Error for RecvError {} |
299 | } |
300 | |
301 | mod big_notify { |
302 | use super::Notify; |
303 | use crate::sync::notify::Notified; |
304 | |
305 | // To avoid contention on the lock inside the `Notify`, we store multiple |
306 | // copies of it. Then, we use either circular access or randomness to spread |
307 | // out threads over different `Notify` objects. |
308 | // |
309 | // Some simple benchmarks show that randomness performs slightly better than |
310 | // circular access (probably due to contention on `next`), so we prefer to |
311 | // use randomness when Tokio is compiled with a random number generator. |
312 | // |
313 | // When the random number generator is not available, we fall back to |
314 | // circular access. |
315 | |
316 | pub(super) struct BigNotify { |
317 | #[cfg (not(all(not(loom), feature = "sync" , any(feature = "rt" , feature = "macros" ))))] |
318 | next: std::sync::atomic::AtomicUsize, |
319 | inner: [Notify; 8], |
320 | } |
321 | |
322 | impl BigNotify { |
323 | pub(super) fn new() -> Self { |
324 | Self { |
325 | #[cfg (not(all( |
326 | not(loom), |
327 | feature = "sync" , |
328 | any(feature = "rt" , feature = "macros" ) |
329 | )))] |
330 | next: std::sync::atomic::AtomicUsize::new(0), |
331 | inner: Default::default(), |
332 | } |
333 | } |
334 | |
335 | pub(super) fn notify_waiters(&self) { |
336 | for notify in &self.inner { |
337 | notify.notify_waiters(); |
338 | } |
339 | } |
340 | |
341 | /// This function implements the case where randomness is not available. |
342 | #[cfg (not(all(not(loom), feature = "sync" , any(feature = "rt" , feature = "macros" ))))] |
343 | pub(super) fn notified(&self) -> Notified<'_> { |
344 | let i = self.next.fetch_add(1, std::sync::atomic::Ordering::Relaxed) % 8; |
345 | self.inner[i].notified() |
346 | } |
347 | |
348 | /// This function implements the case where randomness is available. |
349 | #[cfg (all(not(loom), feature = "sync" , any(feature = "rt" , feature = "macros" )))] |
350 | pub(super) fn notified(&self) -> Notified<'_> { |
351 | let i = crate::runtime::context::thread_rng_n(8) as usize; |
352 | self.inner[i].notified() |
353 | } |
354 | } |
355 | } |
356 | |
357 | use self::state::{AtomicState, Version}; |
358 | mod state { |
359 | use crate::loom::sync::atomic::AtomicUsize; |
360 | use crate::loom::sync::atomic::Ordering; |
361 | |
362 | const CLOSED_BIT: usize = 1; |
363 | |
364 | // Using 2 as the step size preserves the `CLOSED_BIT`. |
365 | const STEP_SIZE: usize = 2; |
366 | |
367 | /// The version part of the state. The lowest bit is always zero. |
368 | #[derive(Copy, Clone, Debug, Eq, PartialEq)] |
369 | pub(super) struct Version(usize); |
370 | |
371 | /// Snapshot of the state. The first bit is used as the CLOSED bit. |
372 | /// The remaining bits are used as the version. |
373 | /// |
374 | /// The CLOSED bit tracks whether the Sender has been dropped. Dropping all |
375 | /// receivers does not set it. |
376 | #[derive(Copy, Clone, Debug)] |
377 | pub(super) struct StateSnapshot(usize); |
378 | |
379 | /// The state stored in an atomic integer. |
380 | /// |
381 | /// The `Sender` uses `Release` ordering for storing a new state |
382 | /// and the `Receiver`s use `Acquire` ordering for loading the |
383 | /// current state. This ensures that written values are seen by |
384 | /// the `Receiver`s for a proper handover. |
385 | #[derive(Debug)] |
386 | pub(super) struct AtomicState(AtomicUsize); |
387 | |
388 | impl Version { |
389 | /// Decrements the version. |
390 | pub(super) fn decrement(&mut self) { |
391 | // Using a wrapping decrement here is required to ensure that the |
392 | // operation is consistent with `std::sync::atomic::AtomicUsize::fetch_add()` |
393 | // which wraps on overflow. |
394 | self.0 = self.0.wrapping_sub(STEP_SIZE); |
395 | } |
396 | |
397 | pub(super) const INITIAL: Self = Version(0); |
398 | } |
399 | |
400 | impl StateSnapshot { |
401 | /// Extract the version from the state. |
402 | pub(super) fn version(self) -> Version { |
403 | Version(self.0 & !CLOSED_BIT) |
404 | } |
405 | |
406 | /// Is the closed bit set? |
407 | pub(super) fn is_closed(self) -> bool { |
408 | (self.0 & CLOSED_BIT) == CLOSED_BIT |
409 | } |
410 | } |
411 | |
412 | impl AtomicState { |
413 | /// Create a new `AtomicState` that is not closed and which has the |
414 | /// version set to `Version::INITIAL`. |
415 | pub(super) fn new() -> Self { |
416 | AtomicState(AtomicUsize::new(Version::INITIAL.0)) |
417 | } |
418 | |
419 | /// Load the current value of the state. |
420 | /// |
421 | /// Only used by the receiver and for debugging purposes. |
422 | /// |
423 | /// The receiver side (read-only) uses `Acquire` ordering for a proper handover |
424 | /// of the shared value with the sender side (single writer). The state is always |
425 | /// updated after modifying and before releasing the (exclusive) lock on the |
426 | /// shared value. |
427 | pub(super) fn load(&self) -> StateSnapshot { |
428 | StateSnapshot(self.0.load(Ordering::Acquire)) |
429 | } |
430 | |
431 | /// Increment the version counter. |
432 | pub(super) fn increment_version_while_locked(&self) { |
433 | // Use `Release` ordering to ensure that the shared value |
434 | // has been written before updating the version. The shared |
435 | // value is still protected by an exclusive lock during this |
436 | // method. |
437 | self.0.fetch_add(STEP_SIZE, Ordering::Release); |
438 | } |
439 | |
440 | /// Set the closed bit in the state. |
441 | pub(super) fn set_closed(&self) { |
442 | self.0.fetch_or(CLOSED_BIT, Ordering::Release); |
443 | } |
444 | } |
445 | } |
446 | |
447 | /// Creates a new watch channel, returning the "send" and "receive" handles. |
448 | /// |
449 | /// All values sent by [`Sender`] will become visible to the [`Receiver`] handles. |
450 | /// Only the last value sent is made available to the [`Receiver`] half. All |
451 | /// intermediate values are dropped. |
452 | /// |
453 | /// # Examples |
454 | /// |
455 | /// The following example prints `hello! world! `. |
456 | /// |
457 | /// ``` |
458 | /// use tokio::sync::watch; |
459 | /// use tokio::time::{Duration, sleep}; |
460 | /// |
461 | /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> { |
462 | /// let (tx, mut rx) = watch::channel("hello" ); |
463 | /// |
464 | /// tokio::spawn(async move { |
465 | /// // Use the equivalent of a "do-while" loop so the initial value is |
466 | /// // processed before awaiting the `changed()` future. |
467 | /// loop { |
468 | /// println!("{}! " , *rx.borrow_and_update()); |
469 | /// if rx.changed().await.is_err() { |
470 | /// break; |
471 | /// } |
472 | /// } |
473 | /// }); |
474 | /// |
475 | /// sleep(Duration::from_millis(100)).await; |
476 | /// tx.send("world" )?; |
477 | /// # Ok(()) |
478 | /// # } |
479 | /// ``` |
480 | /// |
481 | /// [`Sender`]: struct@Sender |
482 | /// [`Receiver`]: struct@Receiver |
483 | pub fn channel<T>(init: T) -> (Sender<T>, Receiver<T>) { |
484 | let shared = Arc::new(Shared { |
485 | value: RwLock::new(init), |
486 | state: AtomicState::new(), |
487 | ref_count_rx: AtomicUsize::new(1), |
488 | notify_rx: big_notify::BigNotify::new(), |
489 | notify_tx: Notify::new(), |
490 | }); |
491 | |
492 | let tx = Sender { |
493 | shared: shared.clone(), |
494 | }; |
495 | |
496 | let rx = Receiver { |
497 | shared, |
498 | version: Version::INITIAL, |
499 | }; |
500 | |
501 | (tx, rx) |
502 | } |
503 | |
504 | impl<T> Receiver<T> { |
505 | fn from_shared(version: Version, shared: Arc<Shared<T>>) -> Self { |
506 | // No synchronization necessary as this is only used as a counter and |
507 | // not memory access. |
508 | shared.ref_count_rx.fetch_add(1, Relaxed); |
509 | |
510 | Self { shared, version } |
511 | } |
512 | |
513 | /// Returns a reference to the most recently sent value. |
514 | /// |
515 | /// This method does not mark the returned value as seen, so future calls to |
516 | /// [`changed`] may return immediately even if you have already seen the |
517 | /// value with a call to `borrow`. |
518 | /// |
519 | /// Outstanding borrows hold a read lock on the inner value. This means that |
520 | /// long-lived borrows could cause the producer half to block. It is recommended |
521 | /// to keep the borrow as short-lived as possible. Additionally, if you are |
522 | /// running in an environment that allows `!Send` futures, you must ensure that |
523 | /// the returned `Ref` type is never held alive across an `.await` point, |
524 | /// otherwise, it can lead to a deadlock. |
525 | /// |
526 | /// The priority policy of the lock is dependent on the underlying lock |
527 | /// implementation, and this type does not guarantee that any particular policy |
528 | /// will be used. In particular, a producer which is waiting to acquire the lock |
529 | /// in `send` might or might not block concurrent calls to `borrow`, e.g.: |
530 | /// |
531 | /// <details><summary>Potential deadlock example</summary> |
532 | /// |
533 | /// ```text |
534 | /// // Task 1 (on thread A) | // Task 2 (on thread B) |
535 | /// let _ref1 = rx.borrow(); | |
536 | /// | // will block |
537 | /// | let _ = tx.send(()); |
538 | /// // may deadlock | |
539 | /// let _ref2 = rx.borrow(); | |
540 | /// ``` |
541 | /// </details> |
542 | /// |
543 | /// For more information on when to use this method versus |
544 | /// [`borrow_and_update`], see [here](self#borrow_and_update-versus-borrow). |
545 | /// |
546 | /// [`changed`]: Receiver::changed |
547 | /// [`borrow_and_update`]: Receiver::borrow_and_update |
548 | /// |
549 | /// # Examples |
550 | /// |
551 | /// ``` |
552 | /// use tokio::sync::watch; |
553 | /// |
554 | /// let (_, rx) = watch::channel("hello" ); |
555 | /// assert_eq!(*rx.borrow(), "hello" ); |
556 | /// ``` |
557 | pub fn borrow(&self) -> Ref<'_, T> { |
558 | let inner = self.shared.value.read().unwrap(); |
559 | |
560 | // After obtaining a read-lock no concurrent writes could occur |
561 | // and the loaded version matches that of the borrowed reference. |
562 | let new_version = self.shared.state.load().version(); |
563 | let has_changed = self.version != new_version; |
564 | |
565 | Ref { inner, has_changed } |
566 | } |
567 | |
568 | /// Returns a reference to the most recently sent value and marks that value |
569 | /// as seen. |
570 | /// |
571 | /// This method marks the current value as seen. Subsequent calls to [`changed`] |
572 | /// will not return immediately until the [`Sender`] has modified the shared |
573 | /// value again. |
574 | /// |
575 | /// Outstanding borrows hold a read lock on the inner value. This means that |
576 | /// long-lived borrows could cause the producer half to block. It is recommended |
577 | /// to keep the borrow as short-lived as possible. Additionally, if you are |
578 | /// running in an environment that allows `!Send` futures, you must ensure that |
579 | /// the returned `Ref` type is never held alive across an `.await` point, |
580 | /// otherwise, it can lead to a deadlock. |
581 | /// |
582 | /// The priority policy of the lock is dependent on the underlying lock |
583 | /// implementation, and this type does not guarantee that any particular policy |
584 | /// will be used. In particular, a producer which is waiting to acquire the lock |
585 | /// in `send` might or might not block concurrent calls to `borrow`, e.g.: |
586 | /// |
587 | /// <details><summary>Potential deadlock example</summary> |
588 | /// |
589 | /// ```text |
590 | /// // Task 1 (on thread A) | // Task 2 (on thread B) |
591 | /// let _ref1 = rx1.borrow_and_update(); | |
592 | /// | // will block |
593 | /// | let _ = tx.send(()); |
594 | /// // may deadlock | |
595 | /// let _ref2 = rx2.borrow_and_update(); | |
596 | /// ``` |
597 | /// </details> |
598 | /// |
599 | /// For more information on when to use this method versus [`borrow`], see |
600 | /// [here](self#borrow_and_update-versus-borrow). |
601 | /// |
602 | /// [`changed`]: Receiver::changed |
603 | /// [`borrow`]: Receiver::borrow |
604 | pub fn borrow_and_update(&mut self) -> Ref<'_, T> { |
605 | let inner = self.shared.value.read().unwrap(); |
606 | |
607 | // After obtaining a read-lock no concurrent writes could occur |
608 | // and the loaded version matches that of the borrowed reference. |
609 | let new_version = self.shared.state.load().version(); |
610 | let has_changed = self.version != new_version; |
611 | |
612 | // Mark the shared value as seen by updating the version |
613 | self.version = new_version; |
614 | |
615 | Ref { inner, has_changed } |
616 | } |
617 | |
618 | /// Checks if this channel contains a message that this receiver has not yet |
619 | /// seen. The new value is not marked as seen. |
620 | /// |
621 | /// Although this method is called `has_changed`, it does not check new |
622 | /// messages for equality, so this call will return true even if the new |
623 | /// message is equal to the old message. |
624 | /// |
625 | /// Returns an error if the channel has been closed. |
626 | /// # Examples |
627 | /// |
628 | /// ``` |
629 | /// use tokio::sync::watch; |
630 | /// |
631 | /// #[tokio::main] |
632 | /// async fn main() { |
633 | /// let (tx, mut rx) = watch::channel("hello" ); |
634 | /// |
635 | /// tx.send("goodbye" ).unwrap(); |
636 | /// |
637 | /// assert!(rx.has_changed().unwrap()); |
638 | /// assert_eq!(*rx.borrow_and_update(), "goodbye" ); |
639 | /// |
640 | /// // The value has been marked as seen |
641 | /// assert!(!rx.has_changed().unwrap()); |
642 | /// |
643 | /// drop(tx); |
644 | /// // The `tx` handle has been dropped |
645 | /// assert!(rx.has_changed().is_err()); |
646 | /// } |
647 | /// ``` |
648 | pub fn has_changed(&self) -> Result<bool, error::RecvError> { |
649 | // Load the version from the state |
650 | let state = self.shared.state.load(); |
651 | if state.is_closed() { |
652 | // The sender has dropped. |
653 | return Err(error::RecvError(())); |
654 | } |
655 | let new_version = state.version(); |
656 | |
657 | Ok(self.version != new_version) |
658 | } |
659 | |
660 | /// Marks the state as changed. |
661 | /// |
662 | /// After invoking this method [`has_changed()`](Self::has_changed) |
663 | /// returns `true` and [`changed()`](Self::changed) returns |
664 | /// immediately, regardless of whether a new value has been sent. |
665 | /// |
666 | /// This is useful for triggering an initial change notification after |
667 | /// subscribing to synchronize new receivers. |
668 | pub fn mark_changed(&mut self) { |
669 | self.version.decrement(); |
670 | } |
671 | |
672 | /// Waits for a change notification, then marks the newest value as seen. |
673 | /// |
674 | /// If the newest value in the channel has not yet been marked seen when |
675 | /// this method is called, the method marks that value seen and returns |
676 | /// immediately. If the newest value has already been marked seen, then the |
677 | /// method sleeps until a new message is sent by the [`Sender`] connected to |
678 | /// this `Receiver`, or until the [`Sender`] is dropped. |
679 | /// |
680 | /// This method returns an error if and only if the [`Sender`] is dropped. |
681 | /// |
682 | /// For more information, see |
683 | /// [*Change notifications*](self#change-notifications) in the module-level documentation. |
684 | /// |
685 | /// # Cancel safety |
686 | /// |
687 | /// This method is cancel safe. If you use it as the event in a |
688 | /// [`tokio::select!`](crate::select) statement and some other branch |
689 | /// completes first, then it is guaranteed that no values have been marked |
690 | /// seen by this call to `changed`. |
691 | /// |
692 | /// [`Sender`]: struct@Sender |
693 | /// |
694 | /// # Examples |
695 | /// |
696 | /// ``` |
697 | /// use tokio::sync::watch; |
698 | /// |
699 | /// #[tokio::main] |
700 | /// async fn main() { |
701 | /// let (tx, mut rx) = watch::channel("hello" ); |
702 | /// |
703 | /// tokio::spawn(async move { |
704 | /// tx.send("goodbye" ).unwrap(); |
705 | /// }); |
706 | /// |
707 | /// assert!(rx.changed().await.is_ok()); |
708 | /// assert_eq!(*rx.borrow_and_update(), "goodbye" ); |
709 | /// |
710 | /// // The `tx` handle has been dropped |
711 | /// assert!(rx.changed().await.is_err()); |
712 | /// } |
713 | /// ``` |
714 | pub async fn changed(&mut self) -> Result<(), error::RecvError> { |
715 | changed_impl(&self.shared, &mut self.version).await |
716 | } |
717 | |
718 | /// Waits for a value that satisfies the provided condition. |
719 | /// |
720 | /// This method will call the provided closure whenever something is sent on |
721 | /// the channel. Once the closure returns `true`, this method will return a |
722 | /// reference to the value that was passed to the closure. |
723 | /// |
724 | /// Before `wait_for` starts waiting for changes, it will call the closure |
725 | /// on the current value. If the closure returns `true` when given the |
726 | /// current value, then `wait_for` will immediately return a reference to |
727 | /// the current value. This is the case even if the current value is already |
728 | /// considered seen. |
729 | /// |
730 | /// The watch channel only keeps track of the most recent value, so if |
731 | /// several messages are sent faster than `wait_for` is able to call the |
732 | /// closure, then it may skip some updates. Whenever the closure is called, |
733 | /// it will be called with the most recent value. |
734 | /// |
735 | /// When this function returns, the value that was passed to the closure |
736 | /// when it returned `true` will be considered seen. |
737 | /// |
738 | /// If the channel is closed, then `wait_for` will return a `RecvError`. |
739 | /// Once this happens, no more messages can ever be sent on the channel. |
740 | /// When an error is returned, it is guaranteed that the closure has been |
741 | /// called on the last value, and that it returned `false` for that value. |
742 | /// (If the closure returned `true`, then the last value would have been |
743 | /// returned instead of the error.) |
744 | /// |
745 | /// Like the `borrow` method, the returned borrow holds a read lock on the |
746 | /// inner value. This means that long-lived borrows could cause the producer |
747 | /// half to block. It is recommended to keep the borrow as short-lived as |
748 | /// possible. See the documentation of `borrow` for more information on |
749 | /// this. |
750 | /// |
751 | /// [`Receiver::changed()`]: crate::sync::watch::Receiver::changed |
752 | /// |
753 | /// # Examples |
754 | /// |
755 | /// ``` |
756 | /// use tokio::sync::watch; |
757 | /// |
758 | /// #[tokio::main] |
759 | /// |
760 | /// async fn main() { |
761 | /// let (tx, _rx) = watch::channel("hello" ); |
762 | /// |
763 | /// tx.send("goodbye" ).unwrap(); |
764 | /// |
765 | /// // here we subscribe to a second receiver |
766 | /// // now in case of using `changed` we would have |
767 | /// // to first check the current value and then wait |
768 | /// // for changes or else `changed` would hang. |
769 | /// let mut rx2 = tx.subscribe(); |
770 | /// |
771 | /// // in place of changed we have use `wait_for` |
772 | /// // which would automatically check the current value |
773 | /// // and wait for changes until the closure returns true. |
774 | /// assert!(rx2.wait_for(|val| *val == "goodbye" ).await.is_ok()); |
775 | /// assert_eq!(*rx2.borrow(), "goodbye" ); |
776 | /// } |
777 | /// ``` |
778 | pub async fn wait_for( |
779 | &mut self, |
780 | mut f: impl FnMut(&T) -> bool, |
781 | ) -> Result<Ref<'_, T>, error::RecvError> { |
782 | let mut closed = false; |
783 | loop { |
784 | { |
785 | let inner = self.shared.value.read().unwrap(); |
786 | |
787 | let new_version = self.shared.state.load().version(); |
788 | let has_changed = self.version != new_version; |
789 | self.version = new_version; |
790 | |
791 | if !closed || has_changed { |
792 | let result = panic::catch_unwind(panic::AssertUnwindSafe(|| f(&inner))); |
793 | match result { |
794 | Ok(true) => { |
795 | return Ok(Ref { inner, has_changed }); |
796 | } |
797 | Ok(false) => { |
798 | // Skip the value. |
799 | } |
800 | Err(panicked) => { |
801 | // Drop the read-lock to avoid poisoning it. |
802 | drop(inner); |
803 | // Forward the panic to the caller. |
804 | panic::resume_unwind(panicked); |
805 | // Unreachable |
806 | } |
807 | }; |
808 | } |
809 | } |
810 | |
811 | if closed { |
812 | return Err(error::RecvError(())); |
813 | } |
814 | |
815 | // Wait for the value to change. |
816 | closed = changed_impl(&self.shared, &mut self.version).await.is_err(); |
817 | } |
818 | } |
819 | |
820 | /// Returns `true` if receivers belong to the same channel. |
821 | /// |
822 | /// # Examples |
823 | /// |
824 | /// ``` |
825 | /// let (tx, rx) = tokio::sync::watch::channel(true); |
826 | /// let rx2 = rx.clone(); |
827 | /// assert!(rx.same_channel(&rx2)); |
828 | /// |
829 | /// let (tx3, rx3) = tokio::sync::watch::channel(true); |
830 | /// assert!(!rx3.same_channel(&rx2)); |
831 | /// ``` |
832 | pub fn same_channel(&self, other: &Self) -> bool { |
833 | Arc::ptr_eq(&self.shared, &other.shared) |
834 | } |
835 | |
836 | cfg_process_driver! { |
837 | pub(crate) fn try_has_changed(&mut self) -> Option<Result<(), error::RecvError>> { |
838 | maybe_changed(&self.shared, &mut self.version) |
839 | } |
840 | } |
841 | } |
842 | |
843 | fn maybe_changed<T>( |
844 | shared: &Shared<T>, |
845 | version: &mut Version, |
846 | ) -> Option<Result<(), error::RecvError>> { |
847 | // Load the version from the state |
848 | let state = shared.state.load(); |
849 | let new_version = state.version(); |
850 | |
851 | if *version != new_version { |
852 | // Observe the new version and return |
853 | *version = new_version; |
854 | return Some(Ok(())); |
855 | } |
856 | |
857 | if state.is_closed() { |
858 | // The sender has been dropped. |
859 | return Some(Err(error::RecvError(()))); |
860 | } |
861 | |
862 | None |
863 | } |
864 | |
865 | async fn changed_impl<T>( |
866 | shared: &Shared<T>, |
867 | version: &mut Version, |
868 | ) -> Result<(), error::RecvError> { |
869 | crate::trace::async_trace_leaf().await; |
870 | |
871 | loop { |
872 | // In order to avoid a race condition, we first request a notification, |
873 | // **then** check the current value's version. If a new version exists, |
874 | // the notification request is dropped. |
875 | let notified = shared.notify_rx.notified(); |
876 | |
877 | if let Some(ret) = maybe_changed(shared, version) { |
878 | return ret; |
879 | } |
880 | |
881 | notified.await; |
882 | // loop around again in case the wake-up was spurious |
883 | } |
884 | } |
885 | |
886 | impl<T> Clone for Receiver<T> { |
887 | fn clone(&self) -> Self { |
888 | let version = self.version; |
889 | let shared = self.shared.clone(); |
890 | |
891 | Self::from_shared(version, shared) |
892 | } |
893 | } |
894 | |
895 | impl<T> Drop for Receiver<T> { |
896 | fn drop(&mut self) { |
897 | // No synchronization necessary as this is only used as a counter and |
898 | // not memory access. |
899 | if 1 == self.shared.ref_count_rx.fetch_sub(1, Relaxed) { |
900 | // This is the last `Receiver` handle, tasks waiting on `Sender::closed()` |
901 | self.shared.notify_tx.notify_waiters(); |
902 | } |
903 | } |
904 | } |
905 | |
906 | impl<T> Sender<T> { |
907 | /// Creates the sending-half of the [`watch`] channel. |
908 | /// |
909 | /// See documentation of [`watch::channel`] for errors when calling this function. |
910 | /// Beware that attempting to send a value when there are no receivers will |
911 | /// return an error. |
912 | /// |
913 | /// [`watch`]: crate::sync::watch |
914 | /// [`watch::channel`]: crate::sync::watch |
915 | /// |
916 | /// # Examples |
917 | /// ``` |
918 | /// let sender = tokio::sync::watch::Sender::new(0u8); |
919 | /// assert!(sender.send(3).is_err()); |
920 | /// let _rec = sender.subscribe(); |
921 | /// assert!(sender.send(4).is_ok()); |
922 | /// ``` |
923 | pub fn new(init: T) -> Self { |
924 | let (tx, _) = channel(init); |
925 | tx |
926 | } |
927 | |
928 | /// Sends a new value via the channel, notifying all receivers. |
929 | /// |
930 | /// This method fails if the channel is closed, which is the case when |
931 | /// every receiver has been dropped. It is possible to reopen the channel |
932 | /// using the [`subscribe`] method. However, when `send` fails, the value |
933 | /// isn't made available for future receivers (but returned with the |
934 | /// [`SendError`]). |
935 | /// |
936 | /// To always make a new value available for future receivers, even if no |
937 | /// receiver currently exists, one of the other send methods |
938 | /// ([`send_if_modified`], [`send_modify`], or [`send_replace`]) can be |
939 | /// used instead. |
940 | /// |
941 | /// [`subscribe`]: Sender::subscribe |
942 | /// [`SendError`]: error::SendError |
943 | /// [`send_if_modified`]: Sender::send_if_modified |
944 | /// [`send_modify`]: Sender::send_modify |
945 | /// [`send_replace`]: Sender::send_replace |
946 | pub fn send(&self, value: T) -> Result<(), error::SendError<T>> { |
947 | // This is pretty much only useful as a hint anyway, so synchronization isn't critical. |
948 | if 0 == self.receiver_count() { |
949 | return Err(error::SendError(value)); |
950 | } |
951 | |
952 | self.send_replace(value); |
953 | Ok(()) |
954 | } |
955 | |
956 | /// Modifies the watched value **unconditionally** in-place, |
957 | /// notifying all receivers. |
958 | /// |
959 | /// This can be useful for modifying the watched value, without |
960 | /// having to allocate a new instance. Additionally, this |
961 | /// method permits sending values even when there are no receivers. |
962 | /// |
963 | /// Prefer to use the more versatile function [`Self::send_if_modified()`] |
964 | /// if the value is only modified conditionally during the mutable borrow |
965 | /// to prevent unneeded change notifications for unmodified values. |
966 | /// |
967 | /// # Panics |
968 | /// |
969 | /// This function panics when the invocation of the `modify` closure panics. |
970 | /// No receivers are notified when panicking. All changes of the watched |
971 | /// value applied by the closure before panicking will be visible in |
972 | /// subsequent calls to `borrow`. |
973 | /// |
974 | /// # Examples |
975 | /// |
976 | /// ``` |
977 | /// use tokio::sync::watch; |
978 | /// |
979 | /// struct State { |
980 | /// counter: usize, |
981 | /// } |
982 | /// let (state_tx, state_rx) = watch::channel(State { counter: 0 }); |
983 | /// state_tx.send_modify(|state| state.counter += 1); |
984 | /// assert_eq!(state_rx.borrow().counter, 1); |
985 | /// ``` |
986 | pub fn send_modify<F>(&self, modify: F) |
987 | where |
988 | F: FnOnce(&mut T), |
989 | { |
990 | self.send_if_modified(|value| { |
991 | modify(value); |
992 | true |
993 | }); |
994 | } |
995 | |
996 | /// Modifies the watched value **conditionally** in-place, |
997 | /// notifying all receivers only if modified. |
998 | /// |
999 | /// This can be useful for modifying the watched value, without |
1000 | /// having to allocate a new instance. Additionally, this |
1001 | /// method permits sending values even when there are no receivers. |
1002 | /// |
1003 | /// The `modify` closure must return `true` if the value has actually |
1004 | /// been modified during the mutable borrow. It should only return `false` |
1005 | /// if the value is guaranteed to be unmodified despite the mutable |
1006 | /// borrow. |
1007 | /// |
1008 | /// Receivers are only notified if the closure returned `true`. If the |
1009 | /// closure has modified the value but returned `false` this results |
1010 | /// in a *silent modification*, i.e. the modified value will be visible |
1011 | /// in subsequent calls to `borrow`, but receivers will not receive |
1012 | /// a change notification. |
1013 | /// |
1014 | /// Returns the result of the closure, i.e. `true` if the value has |
1015 | /// been modified and `false` otherwise. |
1016 | /// |
1017 | /// # Panics |
1018 | /// |
1019 | /// This function panics when the invocation of the `modify` closure panics. |
1020 | /// No receivers are notified when panicking. All changes of the watched |
1021 | /// value applied by the closure before panicking will be visible in |
1022 | /// subsequent calls to `borrow`. |
1023 | /// |
1024 | /// # Examples |
1025 | /// |
1026 | /// ``` |
1027 | /// use tokio::sync::watch; |
1028 | /// |
1029 | /// struct State { |
1030 | /// counter: usize, |
1031 | /// } |
1032 | /// let (state_tx, mut state_rx) = watch::channel(State { counter: 1 }); |
1033 | /// let inc_counter_if_odd = |state: &mut State| { |
1034 | /// if state.counter % 2 == 1 { |
1035 | /// state.counter += 1; |
1036 | /// return true; |
1037 | /// } |
1038 | /// false |
1039 | /// }; |
1040 | /// |
1041 | /// assert_eq!(state_rx.borrow().counter, 1); |
1042 | /// |
1043 | /// assert!(!state_rx.has_changed().unwrap()); |
1044 | /// assert!(state_tx.send_if_modified(inc_counter_if_odd)); |
1045 | /// assert!(state_rx.has_changed().unwrap()); |
1046 | /// assert_eq!(state_rx.borrow_and_update().counter, 2); |
1047 | /// |
1048 | /// assert!(!state_rx.has_changed().unwrap()); |
1049 | /// assert!(!state_tx.send_if_modified(inc_counter_if_odd)); |
1050 | /// assert!(!state_rx.has_changed().unwrap()); |
1051 | /// assert_eq!(state_rx.borrow_and_update().counter, 2); |
1052 | /// ``` |
1053 | pub fn send_if_modified<F>(&self, modify: F) -> bool |
1054 | where |
1055 | F: FnOnce(&mut T) -> bool, |
1056 | { |
1057 | { |
1058 | // Acquire the write lock and update the value. |
1059 | let mut lock = self.shared.value.write().unwrap(); |
1060 | |
1061 | // Update the value and catch possible panic inside func. |
1062 | let result = panic::catch_unwind(panic::AssertUnwindSafe(|| modify(&mut lock))); |
1063 | match result { |
1064 | Ok(modified) => { |
1065 | if !modified { |
1066 | // Abort, i.e. don't notify receivers if unmodified |
1067 | return false; |
1068 | } |
1069 | // Continue if modified |
1070 | } |
1071 | Err(panicked) => { |
1072 | // Drop the lock to avoid poisoning it. |
1073 | drop(lock); |
1074 | // Forward the panic to the caller. |
1075 | panic::resume_unwind(panicked); |
1076 | // Unreachable |
1077 | } |
1078 | }; |
1079 | |
1080 | self.shared.state.increment_version_while_locked(); |
1081 | |
1082 | // Release the write lock. |
1083 | // |
1084 | // Incrementing the version counter while holding the lock ensures |
1085 | // that receivers are able to figure out the version number of the |
1086 | // value they are currently looking at. |
1087 | drop(lock); |
1088 | } |
1089 | |
1090 | self.shared.notify_rx.notify_waiters(); |
1091 | |
1092 | true |
1093 | } |
1094 | |
1095 | /// Sends a new value via the channel, notifying all receivers and returning |
1096 | /// the previous value in the channel. |
1097 | /// |
1098 | /// This can be useful for reusing the buffers inside a watched value. |
1099 | /// Additionally, this method permits sending values even when there are no |
1100 | /// receivers. |
1101 | /// |
1102 | /// # Examples |
1103 | /// |
1104 | /// ``` |
1105 | /// use tokio::sync::watch; |
1106 | /// |
1107 | /// let (tx, _rx) = watch::channel(1); |
1108 | /// assert_eq!(tx.send_replace(2), 1); |
1109 | /// assert_eq!(tx.send_replace(3), 2); |
1110 | /// ``` |
1111 | pub fn send_replace(&self, mut value: T) -> T { |
1112 | // swap old watched value with the new one |
1113 | self.send_modify(|old| mem::swap(old, &mut value)); |
1114 | |
1115 | value |
1116 | } |
1117 | |
1118 | /// Returns a reference to the most recently sent value |
1119 | /// |
1120 | /// Outstanding borrows hold a read lock on the inner value. This means that |
1121 | /// long-lived borrows could cause the producer half to block. It is recommended |
1122 | /// to keep the borrow as short-lived as possible. Additionally, if you are |
1123 | /// running in an environment that allows `!Send` futures, you must ensure that |
1124 | /// the returned `Ref` type is never held alive across an `.await` point, |
1125 | /// otherwise, it can lead to a deadlock. |
1126 | /// |
1127 | /// # Examples |
1128 | /// |
1129 | /// ``` |
1130 | /// use tokio::sync::watch; |
1131 | /// |
1132 | /// let (tx, _) = watch::channel("hello" ); |
1133 | /// assert_eq!(*tx.borrow(), "hello" ); |
1134 | /// ``` |
1135 | pub fn borrow(&self) -> Ref<'_, T> { |
1136 | let inner = self.shared.value.read().unwrap(); |
1137 | |
1138 | // The sender/producer always sees the current version |
1139 | let has_changed = false; |
1140 | |
1141 | Ref { inner, has_changed } |
1142 | } |
1143 | |
1144 | /// Checks if the channel has been closed. This happens when all receivers |
1145 | /// have dropped. |
1146 | /// |
1147 | /// # Examples |
1148 | /// |
1149 | /// ``` |
1150 | /// let (tx, rx) = tokio::sync::watch::channel(()); |
1151 | /// assert!(!tx.is_closed()); |
1152 | /// |
1153 | /// drop(rx); |
1154 | /// assert!(tx.is_closed()); |
1155 | /// ``` |
1156 | pub fn is_closed(&self) -> bool { |
1157 | self.receiver_count() == 0 |
1158 | } |
1159 | |
1160 | /// Completes when all receivers have dropped. |
1161 | /// |
1162 | /// This allows the producer to get notified when interest in the produced |
1163 | /// values is canceled and immediately stop doing work. |
1164 | /// |
1165 | /// # Cancel safety |
1166 | /// |
1167 | /// This method is cancel safe. Once the channel is closed, it stays closed |
1168 | /// forever and all future calls to `closed` will return immediately. |
1169 | /// |
1170 | /// # Examples |
1171 | /// |
1172 | /// ``` |
1173 | /// use tokio::sync::watch; |
1174 | /// |
1175 | /// #[tokio::main] |
1176 | /// async fn main() { |
1177 | /// let (tx, rx) = watch::channel("hello" ); |
1178 | /// |
1179 | /// tokio::spawn(async move { |
1180 | /// // use `rx` |
1181 | /// drop(rx); |
1182 | /// }); |
1183 | /// |
1184 | /// // Waits for `rx` to drop |
1185 | /// tx.closed().await; |
1186 | /// println!("the `rx` handles dropped" ) |
1187 | /// } |
1188 | /// ``` |
1189 | pub async fn closed(&self) { |
1190 | crate::trace::async_trace_leaf().await; |
1191 | |
1192 | while self.receiver_count() > 0 { |
1193 | let notified = self.shared.notify_tx.notified(); |
1194 | |
1195 | if self.receiver_count() == 0 { |
1196 | return; |
1197 | } |
1198 | |
1199 | notified.await; |
1200 | // The channel could have been reopened in the meantime by calling |
1201 | // `subscribe`, so we loop again. |
1202 | } |
1203 | } |
1204 | |
1205 | /// Creates a new [`Receiver`] connected to this `Sender`. |
1206 | /// |
1207 | /// All messages sent before this call to `subscribe` are initially marked |
1208 | /// as seen by the new `Receiver`. |
1209 | /// |
1210 | /// This method can be called even if there are no other receivers. In this |
1211 | /// case, the channel is reopened. |
1212 | /// |
1213 | /// # Examples |
1214 | /// |
1215 | /// The new channel will receive messages sent on this `Sender`. |
1216 | /// |
1217 | /// ``` |
1218 | /// use tokio::sync::watch; |
1219 | /// |
1220 | /// #[tokio::main] |
1221 | /// async fn main() { |
1222 | /// let (tx, _rx) = watch::channel(0u64); |
1223 | /// |
1224 | /// tx.send(5).unwrap(); |
1225 | /// |
1226 | /// let rx = tx.subscribe(); |
1227 | /// assert_eq!(5, *rx.borrow()); |
1228 | /// |
1229 | /// tx.send(10).unwrap(); |
1230 | /// assert_eq!(10, *rx.borrow()); |
1231 | /// } |
1232 | /// ``` |
1233 | /// |
1234 | /// The most recent message is considered seen by the channel, so this test |
1235 | /// is guaranteed to pass. |
1236 | /// |
1237 | /// ``` |
1238 | /// use tokio::sync::watch; |
1239 | /// use tokio::time::Duration; |
1240 | /// |
1241 | /// #[tokio::main] |
1242 | /// async fn main() { |
1243 | /// let (tx, _rx) = watch::channel(0u64); |
1244 | /// tx.send(5).unwrap(); |
1245 | /// let mut rx = tx.subscribe(); |
1246 | /// |
1247 | /// tokio::spawn(async move { |
1248 | /// // by spawning and sleeping, the message is sent after `main` |
1249 | /// // hits the call to `changed`. |
1250 | /// # if false { |
1251 | /// tokio::time::sleep(Duration::from_millis(10)).await; |
1252 | /// # } |
1253 | /// tx.send(100).unwrap(); |
1254 | /// }); |
1255 | /// |
1256 | /// rx.changed().await.unwrap(); |
1257 | /// assert_eq!(100, *rx.borrow()); |
1258 | /// } |
1259 | /// ``` |
1260 | pub fn subscribe(&self) -> Receiver<T> { |
1261 | let shared = self.shared.clone(); |
1262 | let version = shared.state.load().version(); |
1263 | |
1264 | // The CLOSED bit in the state tracks only whether the sender is |
1265 | // dropped, so we do not need to unset it if this reopens the channel. |
1266 | Receiver::from_shared(version, shared) |
1267 | } |
1268 | |
1269 | /// Returns the number of receivers that currently exist. |
1270 | /// |
1271 | /// # Examples |
1272 | /// |
1273 | /// ``` |
1274 | /// use tokio::sync::watch; |
1275 | /// |
1276 | /// #[tokio::main] |
1277 | /// async fn main() { |
1278 | /// let (tx, rx1) = watch::channel("hello" ); |
1279 | /// |
1280 | /// assert_eq!(1, tx.receiver_count()); |
1281 | /// |
1282 | /// let mut _rx2 = rx1.clone(); |
1283 | /// |
1284 | /// assert_eq!(2, tx.receiver_count()); |
1285 | /// } |
1286 | /// ``` |
1287 | pub fn receiver_count(&self) -> usize { |
1288 | self.shared.ref_count_rx.load(Relaxed) |
1289 | } |
1290 | } |
1291 | |
1292 | impl<T> Drop for Sender<T> { |
1293 | fn drop(&mut self) { |
1294 | self.shared.state.set_closed(); |
1295 | self.shared.notify_rx.notify_waiters(); |
1296 | } |
1297 | } |
1298 | |
1299 | // ===== impl Ref ===== |
1300 | |
1301 | impl<T> ops::Deref for Ref<'_, T> { |
1302 | type Target = T; |
1303 | |
1304 | fn deref(&self) -> &T { |
1305 | self.inner.deref() |
1306 | } |
1307 | } |
1308 | |
1309 | #[cfg (all(test, loom))] |
1310 | mod tests { |
1311 | use futures::future::FutureExt; |
1312 | use loom::thread; |
1313 | |
1314 | // test for https://github.com/tokio-rs/tokio/issues/3168 |
1315 | #[test] |
1316 | fn watch_spurious_wakeup() { |
1317 | loom::model(|| { |
1318 | let (send, mut recv) = crate::sync::watch::channel(0i32); |
1319 | |
1320 | send.send(1).unwrap(); |
1321 | |
1322 | let send_thread = thread::spawn(move || { |
1323 | send.send(2).unwrap(); |
1324 | send |
1325 | }); |
1326 | |
1327 | recv.changed().now_or_never(); |
1328 | |
1329 | let send = send_thread.join().unwrap(); |
1330 | let recv_thread = thread::spawn(move || { |
1331 | recv.changed().now_or_never(); |
1332 | recv.changed().now_or_never(); |
1333 | recv |
1334 | }); |
1335 | |
1336 | send.send(3).unwrap(); |
1337 | |
1338 | let mut recv = recv_thread.join().unwrap(); |
1339 | let send_thread = thread::spawn(move || { |
1340 | send.send(2).unwrap(); |
1341 | }); |
1342 | |
1343 | recv.changed().now_or_never(); |
1344 | |
1345 | send_thread.join().unwrap(); |
1346 | }); |
1347 | } |
1348 | |
1349 | #[test] |
1350 | fn watch_borrow() { |
1351 | loom::model(|| { |
1352 | let (send, mut recv) = crate::sync::watch::channel(0i32); |
1353 | |
1354 | assert!(send.borrow().eq(&0)); |
1355 | assert!(recv.borrow().eq(&0)); |
1356 | |
1357 | send.send(1).unwrap(); |
1358 | assert!(send.borrow().eq(&1)); |
1359 | |
1360 | let send_thread = thread::spawn(move || { |
1361 | send.send(2).unwrap(); |
1362 | send |
1363 | }); |
1364 | |
1365 | recv.changed().now_or_never(); |
1366 | |
1367 | let send = send_thread.join().unwrap(); |
1368 | let recv_thread = thread::spawn(move || { |
1369 | recv.changed().now_or_never(); |
1370 | recv.changed().now_or_never(); |
1371 | recv |
1372 | }); |
1373 | |
1374 | send.send(3).unwrap(); |
1375 | |
1376 | let recv = recv_thread.join().unwrap(); |
1377 | assert!(recv.borrow().eq(&3)); |
1378 | assert!(send.borrow().eq(&3)); |
1379 | |
1380 | send.send(2).unwrap(); |
1381 | |
1382 | thread::spawn(move || { |
1383 | assert!(recv.borrow().eq(&2)); |
1384 | }); |
1385 | assert!(send.borrow().eq(&2)); |
1386 | }); |
1387 | } |
1388 | } |
1389 | |