Repository: tokio-rs/tokio
File: tokio/src/sync/watch.rs (Rust)

Code before:
    /// async fn main() {
    ///     let (tx, _rx) = watch::channel("hello");
    ///
    ///     tx.send("goodbye").unwrap();
    ///     
    ///     // here we subscribe to a second receiver
    ///     // now in case of using `changed` we would have
    ///     // to first check the current value and then wait
    ///     // for changes or else `changed` would hang.
    ///     let mut rx2 = tx.subscribe();
    ///     
    ///     // in place of changed we have use `wait_for`
    ///     // which would automatically check the current value
    ///     // and wait for changes until the closure returns true.
    ///     assert!(rx2.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx2.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        loop {
            if f(&self.borrow_and_update()) {
                return Ok(self.borrow());
            }

            match self.changed().await {
                Ok(_) => {}
                Err(e) => {
                    // check if the version has changed since the last time we
                    // called the closure.
                    if self.shared.state.load().version() == self.version {
                        return Err(e);
                    }
                    // the version has changed, so we need to call the closure
                    if f(&self.borrow_and_update()) {
                        return Ok(self.borrow());
                    }
                    return Err(e);
                }
            }
        }
    }

    /// Returns `true` if receivers belong to the same channel.
    ///
    /// # Examples
    ///
    /// ```
    /// let (tx, rx) = tokio::sync::watch::channel(true);

Review comment (bug):
I didn't notice this initially, but this is wrong. You need to reuse the `Ref` object.
```Rust
{
    let val = self.borrow_and_update();
    if f(&val) {
        return Ok(val);
    }
}
```
When you lock the `RwLock` twice like this, you risk that someone changed the value between the call to `self.borrow_and_update()` and `self.borrow()`. That would be wrong since we promise to return the same value that we passed to the closure.

It's possible that you're lucky here and that the `borrow_and_update` lock guard is still alive by the time you call `self.borrow()`, meaning that a write lock could not have been taken, preventing modifications. However, in that case, you can end up with a deadlock because a writer might be in front of the `self.borrow()` call in the lock queue.

Code after:
    ///     // which would automatically check the current value
    ///     // and wait for changes until the closure returns true.
    ///     assert!(rx2.wait_for(|val| *val == "goodbye").await.is_ok());
    ///     assert_eq!(*rx2.borrow(), "goodbye");
    /// }
    /// ```
    pub async fn wait_for(
        &mut self,
        mut f: impl FnMut(&T) -> bool,
    ) -> Result<Ref<'_, T>, error::RecvError> {
        let mut closed = false;
        loop {
            {
                let inner = self.shared.value.read().unwrap();

                let new_version = self.shared.state.load().version();
                let has_changed = self.version != new_version;
                self.version = new_version;

                if (!closed || has_changed) && f(&inner) {
                    return Ok(Ref { inner, has_changed });
                }
            }

            if closed {
                return Err(error::RecvError(()));
            }

            // Wait for the value to change.
            closed = changed_impl(&self.shared, &mut self.version).await.is_err();
        }
    }

    /// Returns `true` if receivers belong to the same channel.
    ///
    /// # Examples
    ///
    /// ```
    /// let (tx, rx) = tokio::sync::watch::channel(true);
    /// let rx2 = rx.clone();
    /// assert!(rx.same_channel(&rx2));
    ///
    /// let (tx3, rx3) = tokio::sync::watch::channel(true);
    /// assert!(!rx3.same_channel(&rx2));
    /// ```
    pub fn same_channel(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.shared, &other.shared)
    }

    cfg_process_driver! {
        pub(crate) fn try_has_changed(&mut self) -> Option<Result<(), error::RecvError>> {