Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

sync: improve docs for watch channels #5954

Merged
merged 6 commits into from Aug 28, 2023
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 1 addition & 1 deletion tokio/src/sync/mod.rs
Expand Up @@ -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.
Expand Down
81 changes: 67 additions & 14 deletions tokio/src/sync/watch.rs
Expand Up @@ -10,24 +10,64 @@
//!
//! [`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 **latest** value stored in the channel and mark it as *seen*,
//! use [`Receiver::borrow_and_update()`].
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you think it's worth explicitly stating something like "borrow_and_update() should be preferred if the receiver intends to await notifications from changed()"?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I kinda added that to the examples section but maybe I should move it up to this section.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved it into a new section titled "borrow_and_update versus borrow". Hopefully that works.

//!
//! To access the latest value **without** changing its state to *seen*, use
hawkw marked this conversation as resolved.
Show resolved Hide resolved
//! [`Receiver::borrow()`]. (If the value has already been marked *seen*,
//! [`Receiver::borrow()`] is equivalent to [`Receiver::borrow_and_update()`].)
Comment on lines +21 to +22
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you think it's worth explicitly stating something like "borrow_and_update() should be preferred if the receiver intends to await notifications from changed(), but it is not required if the receiver is just looking at the current value of the shared state without waiting for changes"?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added this to the new section, and also mentioned why borrow can be more convenient (it's &self).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do you think we should maybe add a

Suggested change
//! [`Receiver::borrow()`]. (If the value has already been marked *seen*,
//! [`Receiver::borrow()`] is equivalent to [`Receiver::borrow_and_update()`].)
//! [`Receiver::borrow()`]. (If the value has already been marked *seen*,
//! [`Receiver::borrow()`] is equivalent to [`Receiver::borrow_and_update()`].)
//!
//! See [here](#borrow-and-update-versus-borrow) for more information on when
//! to use these methods.

or something?

//!
//! ## Change notifications
hawkw marked this conversation as resolved.
Show resolved Hide resolved
//!
//! The [`Receiver`] half provides an asynchronous [`changed`] method. This
//! method is ready when a new, *unseen* value is sent via the [`Sender`] half.
//!
//! * The [`changed`] method returns `Ok(())` on receiving a new value, or
//! `Err(_)` if the [`Sender`] has been closed.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

very tiny nit, not actually important: this returns a public named error type, so we could actually reference it:

Suggested change
//! `Err(_)` if the [`Sender`] has been closed.
//! `Err(`[`errors::RecvError`]`)` if the [`Sender`] has been closed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

//! * On completion, the [`changed`] method marks the new value as *seen*. If
//! [`Receiver::changed()`] is called again, it will not return immediately
//! unless a subsequent value is sent.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps it would make sense to add this point somewhere here?

  • If the latest value is unseen when calling changed, then changed will return immediately. If the latest message is seen, then it will sleep until a new message is sent.

I don't think the current way its written make this sufficiently clear.

Then the second point can also be simplified to this:

  • On completion, the changed method marks the new value as seen.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 to @Darksonn's comment. Also, perhaps some of this documentation should also be copied into the Receiver::changed() docs?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like this a lot. Updated with this + also a mention that the sender can be dropped.

//! * At creation, the initial value is considered *seen*. In other words,
//! [`Receiver::changed()`] will not return until a subsequent value is sent
//! via the [`Sender`] half.
//! * New [`Receiver`] instances can be created with [`Sender::subscribe()`].
//! The current value at the time the [`Receiver`] is created is considered
//! *seen*. [`Receiver::changed()`] will not return until a subsequent value
//! is sent.
//!
//! # Examples
//!
//! In a loop with [`Receiver::changed()`], [`Receiver::borrow_and_update()`] is
//! recommended 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.
//!
//! 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.
hawkw marked this conversation as resolved.
Show resolved Hide resolved
//! loop {
//! println!("{}! ", *rx.borrow_and_update());
//! if rx.changed().await.is_err() {
//! break;
//! }
//! }
//! });
//!
//! sleep(Duration::from_millis(100)).await;
//! tx.send("world")?;
//! # Ok(())
//! # }
Expand All @@ -39,8 +79,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
//!
Expand All @@ -50,11 +90,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;

Expand Down Expand Up @@ -374,19 +418,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.
/// 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(())
/// # }
/// ```
Expand Down Expand Up @@ -595,7 +648,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());
Expand Down