itsscb/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2025-10-01 12:20:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

sync: improve docs for watch channels (#5954)

Browse Source 
## Motivation

I found the watch docs as written to be somewhat confusing.

* It wasn't clear to me whether values are marked seen or not at
  creation/subscribe time.
* The example also confused me a bit, suggesting a while loop when a
  do-while loop is generally more correct.
* I noticed a potential race with `borrow` that is no longer an issue
  with `borrow_and_update`.

## Solution

Update the documentation for the watch module to try and make all this
clearer.
This commit is contained in:
Rain 2023-08-28 16:13:07 -07:00 committed by GitHub
parent fb3028f3a2
commit cb1e10b745
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 90 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tokio/src/sync/mod.rs
View File

@ -383,7 +383,7 @@
//! sleep.set(time::sleep_until(op_start + conf.timeout));
//! }
//! _ = rx.changed() => {
//! conf = rx.borrow().clone();
//! conf = rx.borrow_and_update().clone();
//!
//! // The configuration has been updated. Update the
//! // `sleep` using the new `timeout` value.

103
tokio/src/sync/watch.rs
View File

@ -10,24 +10,75 @@
//!
//! [`channel`] returns a [`Sender`] / [`Receiver`] pair. These are the producer
//! and consumer halves of the channel. The channel is created with an initial
//! value. The **latest** value stored in the channel is accessed with
//! [`Receiver::borrow()`]. Awaiting [`Receiver::changed()`] waits for a new
//! value to be sent by the [`Sender`] half.
//! value.
//!
//! Each [`Receiver`] independently tracks the last value *seen* by its caller.
//!
//! To access the **current** value stored in the channel and mark it as *seen*
//! by a given [`Receiver`], use [`Receiver::borrow_and_update()`].
//!
//! To access the current value **without** marking it as *seen*, use
//! [`Receiver::borrow()`]. (If the value has already been marked *seen*,
//! [`Receiver::borrow()`] is equivalent to [`Receiver::borrow_and_update()`].)
//!
//! For more information on when to use these methods, see
//! [here](#borrow_and_update-versus-borrow).
//!
//! ## Change notifications
//!
//! The [`Receiver`] half provides an asynchronous [`changed`] method. This
//! method is ready when a new, *unseen* value is sent via the [`Sender`] half.
//!
//! * [`Receiver::changed()`] returns `Ok(())` on receiving a new value, or
//! `Err(`[`error::RecvError`]`)` if the [`Sender`] has been dropped.
//! * If the current value is *unseen* when calling [`changed`], then
//! [`changed`] will return immediately. If the current value is *seen*, then
//! it will sleep until either a new message is sent via the [`Sender`] half,
//! or the [`Sender`] is dropped.
//! * On completion, the [`changed`] method marks the new value as *seen*.
//! * At creation, the initial value is considered *seen*. In other words,
//! [`Receiver::changed()`] will not return until a subsequent value is sent.
//! * New [`Receiver`] instances can be created with [`Sender::subscribe()`].
//! The current value at the time the [`Receiver`] is created is considered
//! *seen*.
//!
//! ## `borrow_and_update` versus `borrow`
//!
//! If the receiver intends to await notifications from [`changed`] in a loop,
//! [`Receiver::borrow_and_update()`] should be preferred over
//! [`Receiver::borrow()`]. This avoids a potential race where a new value is
//! sent between [`changed`] being ready and the value being read. (If
//! [`Receiver::borrow()`] is used, the loop may run twice with the same value.)
//!
//! If the receiver is only interested in the current value, and does not intend
//! to wait for changes, then [`Receiver::borrow()`] can be used. It may be more
//! convenient to use [`borrow`](Receiver::borrow) since it's an `&self`
//! method---[`borrow_and_update`](Receiver::borrow_and_update) requires `&mut
//! self`.
//!
//! # Examples
//!
//! The following example prints `hello! world! `.
//!
//! ```
//! use tokio::sync::watch;
//! use tokio::time::{Duration, sleep};
//!
//! # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
//! let (tx, mut rx) = watch::channel("hello");
//!
//! tokio::spawn(async move {
//! while rx.changed().await.is_ok() {
//! println!("received = {:?}", *rx.borrow());
//! // Use the equivalent of a "do-while" loop so the initial value is
//! // processed before awaiting the `changed()` future.
//! loop {
//! println!("{}! ", *rx.borrow_and_update());
//! if rx.changed().await.is_err() {
//! break;
//! }
//! }
//! });
//!
//! sleep(Duration::from_millis(100)).await;
//! tx.send("world")?;
//! # Ok(())
//! # }
@ -39,8 +90,8 @@
//! when all [`Receiver`] handles have been dropped. This indicates that there
//! is no further interest in the values being produced and work can be stopped.
//!
//! The value in the channel will not be dropped until the sender and all receivers
//! have been dropped.
//! The value in the channel will not be dropped until the sender and all
//! receivers have been dropped.
//!
//! # Thread safety
//!
@ -50,11 +101,15 @@
//!
//! [`Sender`]: crate::sync::watch::Sender
//! [`Receiver`]: crate::sync::watch::Receiver
//! [`changed`]: crate::sync::watch::Receiver::changed
//! [`Receiver::changed()`]: crate::sync::watch::Receiver::changed
//! [`Receiver::borrow()`]: crate::sync::watch::Receiver::borrow
//! [`Receiver::borrow_and_update()`]:
//! crate::sync::watch::Receiver::borrow_and_update
//! [`channel`]: crate::sync::watch::channel
//! [`Sender::is_closed`]: crate::sync::watch::Sender::is_closed
//! [`Sender::closed`]: crate::sync::watch::Sender::closed
//! [`Sender::subscribe()`]: crate::sync::watch::Sender::subscribe
use crate::sync::notify::Notify;
@ -374,19 +429,28 @@ mod state {
///
/// # Examples
///
/// The following example prints `hello! world! `.
///
/// ```
/// use tokio::sync::watch;
/// use tokio::time::{Duration, sleep};
///
/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let (tx, mut rx) = watch::channel("hello");
/// let (tx, mut rx) = watch::channel("hello");
///
/// tokio::spawn(async move {
/// while rx.changed().await.is_ok() {
/// println!("received = {:?}", *rx.borrow());
/// tokio::spawn(async move {
/// // Use the equivalent of a "do-while" loop so the initial value is
/// // processed before awaiting the `changed()` future.
/// loop {
/// println!("{}! ", *rx.borrow_and_update());
/// if rx.changed().await.is_err() {
/// break;
/// }
/// });
/// }
/// });
///
/// tx.send("world")?;
/// sleep(Duration::from_millis(100)).await;
/// tx.send("world")?;
/// # Ok(())
/// # }
/// ```
@ -453,7 +517,11 @@ impl<T> Receiver<T> {
/// ```
/// </details>
///
/// For more information on when to use this method versus
/// [`borrow_and_update`], see [here](self#borrow_and_update-versus-borrow).
///
/// [`changed`]: Receiver::changed
/// [`borrow_and_update`]: Receiver::borrow_and_update
///
/// # Examples
///
@ -505,7 +573,11 @@ impl<T> Receiver<T> {
/// ```
/// </details>
///
/// For more information on when to use this method versus [`borrow`], see
/// [here](self#borrow_and_update-versus-borrow).
///
/// [`changed`]: Receiver::changed
/// [`borrow`]: Receiver::borrow
pub fn borrow_and_update(&mut self) -> Ref<'_, T> {
let inner = self.shared.value.read().unwrap();
@ -572,6 +644,9 @@ impl<T> Receiver<T> {
///
/// This method returns an error if and only if the [`Sender`] is dropped.
///
/// For more information, see
/// [*Change notifications*](self#change-notifications) in the module-level documentation.
///
/// # Cancel safety
///
/// This method is cancel safe. If you use it as the event in a
@ -595,7 +670,7 @@ impl<T> Receiver<T> {
/// });
///
/// assert!(rx.changed().await.is_ok());
/// assert_eq!(*rx.borrow(), "goodbye");
/// assert_eq!(*rx.borrow_and_update(), "goodbye");
///
/// // The `tx` handle has been dropped
/// assert!(rx.changed().await.is_err());