sync: Remove readiness assertion in `watch::Receiver::changed() #2839
Conversation
In `watch::Receiver::changed` `Notified` was polled for the first time to ensure the waiter is registered while assuming that the first poll will always return `Pending`. It is the case however that another instance of `Notified` is dropped without receiving its notification, this "orphaned" notification can be used to satisfy another waiter without even registering it. This commit accounts for that scenario. Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>
Darksonn
added
A-tokio
There was a problem hiding this comment.
This looks right to me, thanks for the fix! @carllerche, anything I'm overlooking?
I had a couple minor nits, but the change seems right.
tokio/src/sync/watch.rs
Outdated
| // Polling the future here has dual purpose. The first one is to register | ||
| // the waiter so when `notify_waiters` is called it is notified. The second | ||
| // is to cover the case where another instance of `Notiified` has been dropped | ||
| // without receiving its notification. If that was the case polling the | ||
| // future for the first time will use this "lost" notification and return | ||
| // `Ready` immediatelly without registering any waiter |
There was a problem hiding this comment.
a couple typos and grammar nits:
| // Polling the future here has dual purpose. The first one is to register | |
| // the waiter so when `notify_waiters` is called it is notified. The second | |
| // is to cover the case where another instance of `Notiified` has been dropped | |
| // without receiving its notification. If that was the case polling the | |
| // future for the first time will use this "lost" notification and return | |
| // `Ready` immediatelly without registering any waiter | |
| // Polling the future here has a dual purpose. The first one is to register | |
| // the waiter so that it is notified when `notify_waiters` is called. The second | |
| // is to cover the case where another instance of `Notified` has been dropped | |
| // without receiving its notification. If this has happened, polling the future | |
| // for the first time will use this "lost" notification and return `Ready` | |
| // immediately without registering any waiter. |
tokio/src/sync/watch.rs
Outdated
| let aquired_lost_notification = | ||
| crate::future::poll_fn(|cx| match Pin::new(&mut notified).poll(cx) { | ||
| Poll::Ready(()) => Poll::Ready(true), | ||
| Poll::Pending => Poll::Ready(false), | ||
| }) | ||
| .await; | ||
|
|
||
| if aquired_lost_notification { |
There was a problem hiding this comment.
nit, take it or leave it: "acquired_lost_notification" is kind of a mouthful, what about something like
| let aquired_lost_notification = | |
| crate::future::poll_fn(|cx| match Pin::new(&mut notified).poll(cx) { | |
| Poll::Ready(()) => Poll::Ready(true), | |
| Poll::Pending => Poll::Ready(false), | |
| }) | |
| .await; | |
| if aquired_lost_notification { | |
| let already_notified = | |
| crate::future::poll_fn(|cx| match Pin::new(&mut notified).poll(cx) { | |
| Poll::Ready(()) => Poll::Ready(true), | |
| Poll::Pending => Poll::Ready(false), | |
| }) | |
| .await; | |
| if already_notified { |
(also, "acquired" is spelled wrong in the code, so we should fix that even if we don't change the name :) )
|
Ack, you are correct. There should be more tests 🙃 thanks for catching this. I don't think this will entirely solve the issue as we can concurrently send many updates and reproduce the problem. As I see it, we have two options: a) Update b) Implement a loop in |
That indeed makes the most sense to me as well. Just out of curiosity, how can I triger the problem after this change by sending concurrent updates. Can I have a loom test to exercise that part ? |
Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>
|
@hawkw I incorporated the suggestion to avoid resending the notifications if they have been trigerred by |
tokio/src/sync/notify.rs
Outdated
| // Notification trigerred by calling `notify_waiters` | ||
| AllWaiters, | ||
| // Notification trigerred by calling `notify_one` | ||
| OneWaiter, |
There was a problem hiding this comment.
typos:
| // Notification trigerred by calling `notify_waiters` | |
| AllWaiters, | |
| // Notification trigerred by calling `notify_one` | |
| OneWaiter, | |
| // Notification triggered by calling `notify_waiters` | |
| AllWaiters, | |
| // Notification triggered by calling `notify_one` | |
| OneWaiter, |
tokio/src/sync/notify.rs
Outdated
| // See if the node was notified but not received. In this case, the | ||
| // notification must be sent to another waiter. | ||
| // notification must be sent to another waiter, only if it was | ||
| // triggered via `notify_one` |
There was a problem hiding this comment.
I might rephrase this to something like
// See if the node was notified but not received. In this case, if
// the notification was triggered via `notify_one`, it must be sent
// to the next waiter.
tokio/src/sync/notify.rs
Outdated
| } | ||
| #[derive(Debug, Clone, Copy)] | ||
| enum NotificationType { |
There was a problem hiding this comment.
nit: i'd add a newline here (surprised that rustfmt doesn't do that?)
| } | |
| #[derive(Debug, Clone, Copy)] | |
| enum NotificationType { | |
| } | |
| #[derive(Debug, Clone, Copy)] | |
| enum NotificationType { |
Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>
Motivation
After chatting with @hawkw I decided to try and tackle #2800. They pointed me to a bit of code that might be useful when learning about how the
Notifytype works. I decided to play with loom and see how it works by writing a few tests. Then came around a problem with theWatchtype. The current loom smoke test forWatchexercises only one receiver. Changing the test to have more than one will result in a panic:This happens because in
watch::Receiver::changedNotifiedwas polled for the firt time to ensure that we register the waiter. The problem is that when there are more than oneNotifiedinstances tied to oneNotify, it is possible for aNotifiedto be dropped without receiving its notification. If that happens this notification is left for anotherNotifiedinstance to consume it. This means that it is not safe to assume that calling Notified::poll() for the first time shall always result in returningPending, even if we are never callingnotify_one.Solution
We handle the case where polling the
Notifiedfuture returnsReadyright away.Signed-off-by: Zahari Dichev zaharidichev@gmail.com