No good way to reliably notify all consumers about a change

[arthurprs]

Is your feature request related to a problem? Please describe.

Consider an use-case where a producer wants to notify consumers about a change, even if they were doing something else at the time of the notification.

At first Notify and Notify::notify_waiters seem like a fit, but it's not. Because notify_waiters doesn't make the permit available for waiters that are not waiting at the time of the notification.

C: subscribe to S by cloning the Notify
P: notifier.notify_waiters()
C: notifier.notified().await <<< waits even if the permit was provided above

Option 1

One way to get the expected behavior is to use a broadcast channel with capacity of 1.

C: subscribe to S by calling sender.subscribe()
P: sender.send(())
C: receiver.recv().await <<< wakes up correctly, with either Err(Lagged(_)) or Ok(())

This works but involves a a couple locks inside the Broadcast channel unnecessarily.

Option 2

A Watch works too but it's unergonomic to keep both Sender and Receivers around, as there's no equivalent to subscribe()

Describe the solution you'd like

Option 1

Add Notifier::notify_all which will notify all existing waiters, not only the ones awaiting.

Option 2

Consider exposing Watch::subscribe which already exists internally. With it this use case can be modeled ergonomically/performantly with a Sender::subscribe() + Receiver::changed().await

tokio/tokio/src/sync/watch.rs

Line 413 in d4075a4

pub(crate) fn subscribe(&self) -> Receiver<T> {

Describe alternatives you've considered

See description above, which mentions 2 options, each with its set of drawbacks.

Additional context

I did find #3066 seems to describe a similar need but was closed with a partial(?) solution in the form of Notifier::notify_awaiters which only takes into account current waiters.

[Darksonn]

Add Notifier::notify_all which will notify all existing waiters, not only the ones awaiting.

This is not possible because unlike the channels, the Notify utility does not know how many waiters there are. It is shared by putting it in an Arc, not by cloning the Notify itself.

[Wesmania]

Would changing signature of notified to fn(&self) -> impl Future<Output = ()> + '_ help? This way, Notified struct could insert itself into a waiter queue immediately and could be used like so:

struct Event {
    occured: Arc<Mutex<bool>>,
    notify: Notify,
}

impl Event {
    fn set(&self) {
        *self.occured.lock().unwrap() = true;
        self.notify.notify_waiters();
    }

    fn wait(&self) -> impl Future<Output = ()> {
        let guard = self.occured.lock().unwrap();
        let wait = self.notify.notified();
        let occured = *guard;
        async move {
            if occured {
                return;
            } else {
                wait.await;
            }
        }
    }
}

[arthurprs]

Something like that works, and is very similar to Watch with no value. But occurred needs to be a counter (that is also tracked in Event) so each Event instance can track if it's up to date or not.

[Wesmania]

Actually I'm wrong, notify already works like that since Notified is constructed as part of the future. Sorry for the confusion.

[Darksonn]

The subscribe method has been exposed on the watch channel, so I'm closing this.