sync: make notify_waiters calls atomic

[satakuma]

Motivation

tokio::sync::Notify has an internal waiters queue locked behind a mutex. Notify::notify_waiters is a function to notify all waiters from this queue. It first acquires the lock, then removes waiters from the queue and wakes them up. However, the whole process is done in batches of 32 waiters to avoid deadlocks. The function has to release the lock before waking up a batch and re-acquire it before proceeding to the next one. In this short timespan, another thread can acquire the lock and modify the queue or the internal state. This leads to the following issues:

1. Spurious wake-ups. This is an issue reported in one call tokio::sync::Notify::notify_waiters() can wake two sequential tokio::sync::Notify::notified().await #5396. A single notify_waiters call can notify multiple futures which are created and awaited sequentially. For example, both futures from the following code can complete:

notify.notified().await;
notify.notified().await;

2. Inconsistency in results from poll. One could expect that sequentially polled futures will yield results consistent with each other. If one pending future returns Ready, then other futures which were pending should also be ready. As an example, let's say that we have two pending Notified futures: fut1 and fut2. If we call notify_waiters in parallel with the following, the assertion can fail:

let res1 = fut1.poll();
let res2 = fut2.poll();
assert!(res1.is_pending() || res2.is_ready()); // if res1 is ready, then res2 must also be ready

3. Lost notifications from subsequent notify_one calls. It is possible that calling notify_one after awaiting a future notified by notify_waiters will result in the notification being lost. As an example, let's say that we have a pending notify.notified() future fut. If we call notify_waiters in parallel with the following code, it can hang forever:

fut.await;
notify.notify_one();
notify.notified().await; // this should consume the notification from `notify_one`.

For more context, see the discussion in #5404 or in the related Discord thread.

Closes: #5396

Solution

The idea is to move all waiters to a secondary list, which is created in each call to notify_waiters. This list will be unavailable to modify from notify_one and other parts of the code, so the waiters contained in it will not consume other notifications, and it will be impossible to insert new waiters into it. This approach eliminates all presented issues.

The tricky part is how to allow waiters contained by a secondary list to remove themselves from it. It is possible to drop a waiter still contained by a secondary list, and in such case it should be able to unlink itself from the neighboring nodes from the list. To solve it, this PR adds a new variant of a linked list, which allows removing a node from it without knowing the list head's memory address. This is done by adding a special “guard” node, which makes the list circular.

To solve the issue with inconsistency in results from poll, this PR also adds some additional logic to poll_notified. A polled future has to know whether it is already included by a secondary list in notify_waiters and is staged to be notified. In such case it also has to remove itself from the list. To determine if it is contained by a secondary list, a future uses the counter of notify_waiters calls. If the current value of this counter is different from the value recorded by a future at the time, when it was being added to the queue, then there is an ongoing notify_waiters call. In such case the future removes itself from the secondary list.

[carllerche]

Given that this is a very specific requirement, maybe include a note saying that this is for the Notify LL and why it relies on it (in a few words). Otherwise, if I were reading this without the necessary context, I would find it very strange.

[carllerche]

This PR is looking good. I really like the guarded linked list approach here. I left some comments inline. I also am going to spend a bit more time thinking about whether moving the state atomic store matters.

[carllerche]

We probably want to pin guard to ensure it doesn't accidentally move (the pin! macro might work here). Also, could you add a big comment saying it is critical for safety that guard does not move and is not dropped until the guarded list is dropped?

[carllerche]

Why was this moved up? I'm not sure it matters, but it is not apparent it does not matter.

[carllerche]

The code is correct as long as the store happens while the mutex is held. Moving the store up increases the odds of a poll_notified succeeding early in some rare concurrent conditions.

[satakuma]

It was moved up for correctness. If we allowed to poll a pending future between chunks and observe the old counter value, then it would be possible to observe the inconsistency from the description (number 2.). This is because such future would return Pending, even though other waiters from the decoupled list could be already notified. notify_waiters_poll_consistency_many checks such scenarios.

[satakuma]

I applied suggestions from the review. I also removed unnecessary ManuallyDrop from the guarded linked list code because it could be a memory-leaking foot gun for link handles such as Arc.

[carllerche]

Looks good to me. I'll let @Darksonn take a look as well. Thanks!

[Darksonn]

We must clean up the linked list even if this call panics.

[carllerche]

Good call.

[satakuma]

That's a good catch. I moved the list to a new struct, which makes sure the list is cleaned up on drop.

[satakuma]

Note that this destructor does not call waker.wake(), which means the remaining waiters will never be woken up unless manually polled. However, this is already the case in the current code because one panicking waker from a batch can result in the whole batch never being notified, see the discussion linked in #4069.

[Darksonn]

Thanks. Looks good to me.