sync: Remove readiness assertion in `watch::Receiver::changed() (#2839)

*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.

1 files changed, 19 insertions, 3 deletions

diff --git a/tokio/src/sync/tests/loom_watch.rs b/tokio/src/sync/tests/loom_watch.rs
index 7944cab8..c575b5b6 100644
--- a/tokio/src/sync/tests/loom_watch.rs
+++ b/tokio/src/sync/tests/loom_watch.rs
@@ -6,14 +6,30 @@ use loom::thread;
 #[test]
 fn smoke() {
     loom::model(|| {
-        let (tx, mut rx) = watch::channel(1);
+        let (tx, mut rx1) = watch::channel(1);
+        let mut rx2 = rx1.clone();
+        let mut rx3 = rx1.clone();
+        let mut rx4 = rx1.clone();
+        let mut rx5 = rx1.clone();
 
         let th = thread::spawn(move || {
             tx.send(2).unwrap();
         });
 
-        block_on(rx.changed()).unwrap();
-        assert_eq!(*rx.borrow(), 2);
+        block_on(rx1.changed()).unwrap();
+        assert_eq!(*rx1.borrow(), 2);
+
+        block_on(rx2.changed()).unwrap();
+        assert_eq!(*rx2.borrow(), 2);
+
+        block_on(rx3.changed()).unwrap();
+        assert_eq!(*rx3.borrow(), 2);
+
+        block_on(rx4.changed()).unwrap();
+        assert_eq!(*rx4.borrow(), 2);
+
+        block_on(rx5.changed()).unwrap();
+        assert_eq!(*rx5.borrow(), 2);
 
         th.join().unwrap();
     })