chore: prepare v0.3.6 release (#3276)

Co-authored-by: @tijsvd

.github/workflows/ci.yml

@@ -1,8 +1,8 @@
 on:
   push:
-    branches: ["master"]
+    branches: ["v0.3.x"]
   pull_request:
-    branches: ["master"]
+    branches: ["v0.3.x"]
 
 name: CI
 

tokio/CHANGELOG.md

@@ -1,3 +1,16 @@
+# 0.3.6 (December 14, 2020)
+
+### Fixed
+- rt: fix deadlock in shutdown (#3228)
+- rt: fix panic in task abort when off rt (#3159)
+- sync: make `add_permits` panic with usize::MAX >> 3 permits (#3188)
+- time: Fix race condition in timer drop (#3229)
+- watch: fix spurious wakeup (#3244)
+
+### Added
+- example: add back udp-codec example (#3205)
+- net: add `TcpStream::into_std` (#3189)
+
 # 0.3.5 (November 30, 2020)
 
 ### Fixed

tokio/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.3.x" git tag.
-version = "0.3.5"
+version = "0.3.6"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio/0.3.5/tokio/"
+documentation = "https://docs.rs/tokio/0.3.6/tokio/"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
 description = """

tokio/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio/0.3.5")]
+#![doc(html_root_url = "https://docs.rs/tokio/0.3.6")]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,

tokio/src/sync/watch.rs

@@ -53,10 +53,10 @@
 
 use crate::sync::Notify;
 
+use crate::loom::sync::atomic::AtomicUsize;
+use crate::loom::sync::atomic::Ordering::{Relaxed, SeqCst};
+use crate::loom::sync::{Arc, RwLock, RwLockReadGuard};
 use std::ops;
-use std::sync::atomic::AtomicUsize;
-use std::sync::atomic::Ordering::{Relaxed, SeqCst};
-use std::sync::{Arc, RwLock, RwLockReadGuard};
 
 /// Receives values from the associated [`Sender`](struct@Sender).
 ///
@@ -241,19 +241,19 @@ impl<T> Receiver<T> {
     /// }
     /// ```
     pub async fn changed(&mut self) -> Result<(), error::RecvError> {
-        // In order to avoid a race condition, we first request a notification,
-        // **then** check the current value's version. If a new version exists,
-        // the notification request is dropped.
-        let notified = self.shared.notify_rx.notified();
+        loop {
+            // In order to avoid a race condition, we first request a notification,
+            // **then** check the current value's version. If a new version exists,
+            // the notification request is dropped.
+            let notified = self.shared.notify_rx.notified();
 
-        if let Some(ret) = maybe_changed(&self.shared, &mut self.version) {
-            return ret;
+            if let Some(ret) = maybe_changed(&self.shared, &mut self.version) {
+                return ret;
+            }
+
+            notified.await;
+            // loop around again in case the wake-up was spurious
         }
-
-        notified.await;
-
-        maybe_changed(&self.shared, &mut self.version)
-            .expect("[bug] failed to observe change after notificaton.")
     }
 }
 
@@ -390,3 +390,44 @@ impl<T> ops::Deref for Ref<'_, T> {
         self.inner.deref()
     }
 }
+
+#[cfg(all(test, loom))]
+mod tests {
+    use futures::future::FutureExt;
+    use loom::thread;
+
+    // test for https://github.com/tokio-rs/tokio/issues/3168
+    #[test]
+    fn watch_spurious_wakeup() {
+        loom::model(|| {
+            let (send, mut recv) = crate::sync::watch::channel(0i32);
+
+            send.send(1).unwrap();
+
+            let send_thread = thread::spawn(move || {
+                send.send(2).unwrap();
+                send
+            });
+
+            recv.changed().now_or_never();
+
+            let send = send_thread.join().unwrap();
+            let recv_thread = thread::spawn(move || {
+                recv.changed().now_or_never();
+                recv.changed().now_or_never();
+                recv
+            });
+
+            send.send(3).unwrap();
+
+            let mut recv = recv_thread.join().unwrap();
+            let send_thread = thread::spawn(move || {
+                send.send(2).unwrap();
+            });
+
+            recv.changed().now_or_never();
+
+            send_thread.join().unwrap();
+        });
+    }
+}