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

Code before:
        pub(super) fn load(&self, ordering: Ordering) -> StateSnapshot {
            StateSnapshot(self.0.load(ordering))
        }

        /// Load the current value of the state.
        ///
        /// The receiver side (read-only) uses `Acquire` ordering for a proper handover
        /// with the sender side (single writer).
        pub(super) fn load_receiver(&self) -> StateSnapshot {
            StateSnapshot(self.0.load(Ordering::Acquire))
        }

        /// Increment the version counter.
        pub(super) fn increment_version(&self) {
            // Use `Release` ordering to ensure that storing the version
            // state is seen by the receiver side that uses `Acquire` for
            // loading the state.
            self.0.fetch_add(STEP_SIZE, Ordering::Release);
        }

        /// Set the closed bit in the state.
        pub(super) fn set_closed(&self) {
            // Use `Release` ordering to ensure that storing the version
            // state is seen by the receiver side that uses `Acquire` for
            // loading the state.
            self.0.fetch_or(CLOSED_BIT, Ordering::Release);
        }
    }
}

/// Creates a new watch channel, returning the "send" and "receive" handles.
///
/// All values sent by [`Sender`] will become visible to the [`Receiver`] handles.
/// Only the last value sent is made available to the [`Receiver`] half. All
/// intermediate values are dropped.
///
/// # 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 {
///     // Use the equivalent of a "do-while" loop so the initial value is
///     // processed before awaiting the `changed()` future.
///     loop {

Review comment (bug):
That said, it would arguably make sense to use something stronger than relaxed for `ref_count_rx`. Currently this assert could fail:
```
// thread 1
GLOBAL_BOOL.store(true, Relaxed);
drop(receiver);

// thread 2
if sender.is_closed() {
    assert!(GLOBAL_BOOL.load(Relaxed));
}
```

Code after:
        /// Only used by the receiver and for debugging purposes.
        ///
        /// The receiver side (read-only) uses `Acquire` ordering for a proper handover
        /// of the shared value with the sender side (single writer). The state is always
        /// updated after modifying and before releasing the (exclusive) lock on the
        /// shared value.
        pub(super) fn load(&self) -> StateSnapshot {
            StateSnapshot(self.0.load(Ordering::Acquire))
        }

        /// Increment the version counter.
        pub(super) fn increment_version_while_locked(&self) {
            // Use `Release` ordering to ensure that the shared value
            // has been written before updating the version. The shared
            // value is still protected by an exclusive lock during this
            // method.
            self.0.fetch_add(STEP_SIZE, Ordering::Release);
        }

        /// Set the closed bit in the state.
        pub(super) fn set_closed(&self) {
            self.0.fetch_or(CLOSED_BIT, Ordering::Release);
        }
    }
}

/// Creates a new watch channel, returning the "send" and "receive" handles.
///
/// All values sent by [`Sender`] will become visible to the [`Receiver`] handles.
/// Only the last value sent is made available to the [`Receiver`] half. All
/// intermediate values are dropped.
///
/// # 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 {
///     // 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;
///         }