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::watch: Add has_changed flag and method to Ref #4758

Merged
Changes from 2 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
42 changes: 34 additions & 8 deletions tokio/src/sync/watch.rs
Expand Up @@ -112,6 +112,20 @@ pub struct Sender<T> {
#[derive(Debug)]
pub struct Ref<'a, T> {
inner: RwLockReadGuard<'a, T>,
updated: bool,
}

impl<'a, T> Ref<'a, T> {
/// Indicates if the shared value has been detected as _changed_
/// **and** marked as seen.
///
/// The result is `true` only if this reference has been obtained
/// from [`Receiver::borrow_and_update()`] in case changes have been
/// detected and marked as seen. In all other cases the result will
/// be `false`.
pub fn updated(&self) -> bool {
self.updated
}
Copy link
Contributor

Choose a reason for hiding this comment

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

It seems more useful to have this function return whether the value was marked as seen before the call. Then the value is also useful for the borrow method.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

This would mean to

  • add an invocation of load() to borrow() and
  • rename the function to has_changed().

I was unsure if the extra load() in borrow() is desired. It probably doesn't have any noticeable effect and a more consistent interface might be preferable.

Copy link
Contributor Author

@uklotzde uklotzde Jun 9, 2022

Choose a reason for hiding this comment

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

Done: d081f83 (fixed typo) b4fef98

I also added a comprehensive example.

}

#[derive(Debug)]
Expand Down Expand Up @@ -332,15 +346,17 @@ impl<T> Receiver<T> {
/// ```
pub fn borrow(&self) -> Ref<'_, T> {
let inner = self.shared.value.read().unwrap();
Ref { inner }
// Changes are neither detected nor marked as seen by this function
let updated = false;
Ref { inner, updated }
}

/// Returns a reference to the most recently sent value and mark that value
/// Returns a reference to the most recently sent value and marks that value
/// as seen.
///
/// This method marks the value as seen, so [`changed`] will not return
/// immediately if the newest value is one previously returned by
/// `borrow_and_update`.
/// This method marks the current value as seen as indicated by [`Ref::updated`].
/// Subsequent calls to [`changed`] will not return immediately until the
/// [`Sender`] has modified the shared value again.
///
/// Outstanding borrows hold a read lock. This means that long lived borrows
/// could cause the send half to block. It is recommended to keep the borrow
Expand All @@ -366,8 +382,16 @@ impl<T> Receiver<T> {
/// [`changed`]: Receiver::changed
pub fn borrow_and_update(&mut self) -> Ref<'_, T> {
let inner = self.shared.value.read().unwrap();
self.version = self.shared.state.load().version();
Ref { inner }
// After obtaining a read-lock no concurrent writes could occur
// and the loaded version matches that of the borrowed reference.
let new_version = self.shared.state.load().version();
let updated = if self.version != new_version {
self.version = new_version;
true
} else {
false
};
Ref { inner, updated }
}

/// Checks if this channel contains a message that this receiver has not yet
Expand Down Expand Up @@ -725,7 +749,9 @@ impl<T> Sender<T> {
/// ```
pub fn borrow(&self) -> Ref<'_, T> {
let inner = self.shared.value.read().unwrap();
Ref { inner }
// The sender/producer always sees the current value
let updated = false;
Ref { inner, updated }
}

/// Checks if the channel has been closed. This happens when all receivers
Expand Down