sync: provide API for notifying multiple waiters.

[carllerche]

There currently is no async equivalent to the synchronous CondVar::notify_all. Roughly, Notify provides similar functionality, but only has a notify_one(). Notify currently has a notify_all() method, but it is crate private. This API has not been made public due to API questions to work through.

First, Notify currently uses an approach similar to park / unpark. When notify_one() is called and there are no waiters, it stores a pending notification. The next time a task waits on Notify, the task consumes the pending notification and the wait completes. This behavior may result in spurious wakeups but is designed to be a lightweight mechamism to avoid race condition without requiring locking in many cases.

However, the park / unpark behavior does not translate to a "notify many" operation. Implementing a useful "notify many" operation is a bit tricky. Consider std::sync::CondVar. In order to wait on a CondVar, the caller must first acquire a mutex. The mutex, paired with CondVar is how races are handled. The downside of using a Mutex is that it is fairly coarse. Usually, with regards to mutual exclusion, in "broadcast" style synchronization, the caller wishes to ensure mutual exclusion between producers and consumers and not with other consumers. Additionally, using an "async" mutex for this doesn't make a ton of sense either as the critical section to guard is usually small and synchronous. It usually makes more sense to use a std Mutex to ensure mutual exclusion, but we need an async-aware notification strategy. See sync::Watch as an example of this.

The private notify_waiters() API handles the race by requiring the consumer to acquire the notification future before checking the state. If the state is as expected and the consumer drops the notification future, otherwise, the consumer waits on the notification future. Unfortunatley, it isn't that simple. In order to register for the notification, the consumer must also poll the notification future. This requires pinning and results in a poor API. I believe that this problem can be solved by storing a counter internally tracking the number of calls to notify_*. When the notification future is created, it loads the current count. When the future is awaited, if the current count is greater than the one loaded on creation, the future completes immediately.

Additionally, notify_waiters() and notify_one() are fairly different behaviors. Conflating them in a single notification primitive may not make sense. I am not aware of any use cases that require mixing notify_one() and notify_waiters(). It may make more sense to provide a separate type for doing broadcast style notifications.

Actions

• determine if the counter strategy can resolve the poor notify_waiters() API.
• decide if notify_waiters() should be included on Notify, or a separate type should be created.

Refs: #2739, #3061

[kaimast]

Awesome! Maybe eventually extend the text in the documentation for tokio::sync as well explaining how to port existing code using CondVars.

[hawkw]

Additionally, notify_waiters() and notify_one() are fairly different behaviors. Conflating them in a single notification primitive may not make sense. I am not aware of any use cases that require mixing notify_one() and notify_waiters(). It may make more sense to provide a separate type for doing broadcast style notifications.

I could go either on this. I think you're right that I can't immediately think of use-cases that require both. On the other hand, if we have two separate public API types for "notify one" and "notify all" behaviors, but they share the underlying internal implementation, and someone does have a use-case for both in the same place, this seems a little inefficient. If a user wants to wake tasks with either a "notify one" or a "notify all" operation, they have to express interest in two separate types, maintaining two separate wait lists...

One potential use-case I can think of for both types of notification in one place is something that notifies a single waiter at a time during normal operation but notifies all waiters when it's dropped/shut down/canceled.

[zaharidichev]

Is that something I can work on? Starting with the counter approach for notify_waiters() ?

[carllerche]

@zaharidichev sure. I believe that based on @hawkw's comments, we should include notify_waiters() on Notify itself.

Assuming the counter strategy works, we should increase the Notify internal state from AtomicU8 to AtomicUsize. Then use all bits not needed for the existing state (0, 1, 2, so 2 bits) for the counter. It will probably be easiest to reserve the LSBs for the state and use the MSBs for the counter, then we can just let it wrap.

Existing internal usage of notify_waiters() will need to be updated to avoid the initial pin/poll.

Does the counter strategy make sense?

[zaharidichev]

@carllerche Yes all that makes sense.

[zaharidichev]

I believe that this problem can be solved by storing a counter internally tracking the number of calls to notify_*. When the notification future is created, it loads the current count. When the future is awaited, if the current count is greater than the one loaded on creation, the future completes immediately.

@carllerche Not sure what you mean by "notify_*". I do not think tracking the number of calls to notify_one is right. We only need to count the calls to notify_waiters because otherwise we can end up with multiple Notified futures getting completed because of a single call to notify_one.

[carllerche]

@zaharidichev very possible. The NOTIFIED state probably handles the case of notify_one.