sync: ensure Mutex, RwLock, and Semaphore futures are Send + Sync (#2375)

Previously, the `Mutex::lock`, `RwLock::{read, write}`, and
`Semaphore::acquire` futures in `tokio::sync` implemented `Send + Sync`
automatically. This was by virtue of being implemented using a `poll_fn`
that only closed over `Send + Sync` types. However, this broke in
PR #2325, which rewrote those types using the new `batch_semaphore`.
Now, they await an `Acquire` future, which contains a `Waiter`, which
internally contains an `UnsafeCell`, and thus does not implement `Sync`.

Since removing previously implemented traits breaks existing code, this
inadvertantly caused a breaking change. There were tests ensuring that
the `Mutex`, `RwLock`, and `Semaphore` types themselves were `Send +
Sync`, but no tests that the _futures they return_ implemented those
traits.

I've fixed this by adding an explicit impl of `Sync` for the
`batch_semaphore::Acquire` future. Since the `Waiter` type held by this
struct is only accessed when borrowed mutably, it is safe for it to
implement `Sync`.

Additionally, I've added to the bounds checks for the effected
`tokio::sync` types to ensure that returned futures continue to
implement `Send + Sync` in the future.

tokio/CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.2.16 (April 3, 2020)
+
+### Fixes
+
+- sync: fix a regression where `Mutex`, `Semaphore`, and `RwLock` futures no
+  longer implement `Sync` (#2375)
+
+
 # 0.2.15 (April 2, 2020)
 
 ### Fixes

tokio/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.2.x" git tag.
-version = "0.2.15"
+version = "0.2.16"
 edition = "2018"
 authors = ["Tokio Contributors <team@tokio.rs>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio/0.2.15/tokio/"
+documentation = "https://docs.rs/tokio/0.2.16/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.2.15")]
+#![doc(html_root_url = "https://docs.rs/tokio/0.2.16")]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,

tokio/src/sync/batch_semaphore.rs

@@ -463,6 +463,13 @@ impl Drop for Acquire<'_> {
     }
 }
 
+// Safety: the `Acquire` future is not `Sync` automatically because it contains
+// a `Waiter`, which, in turn, contains an `UnsafeCell`. However, the
+// `UnsafeCell` is only accessed when the future is borrowed mutably (either in
+// `poll` or in `drop`). Therefore, it is safe (although not particularly
+// _useful_) for the future to be borrowed immutably across threads.
+unsafe impl Sync for Acquire<'_> {}
+
 // ===== impl AcquireError ====
 
 impl AcquireError {

tokio/src/sync/mutex.rs

@@ -137,8 +137,15 @@ impl Error for TryLockError {}
 fn bounds() {
     fn check_send<T: Send>() {}
     fn check_unpin<T: Unpin>() {}
+    // This has to take a value, since the async fn's return type is unnameable.
+    fn check_send_sync_val<T: Send + Sync>(_t: T) {}
+    fn check_send_sync<T: Send + Sync>() {}
     check_send::<MutexGuard<'_, u32>>();
     check_unpin::<Mutex<u32>>();
+    check_send_sync::<Mutex<u32>>();
+
+    let mutex = Mutex::new(1);
+    check_send_sync_val(mutex.lock());
 }
 
 impl<T> Mutex<T> {

tokio/src/sync/rwlock.rs

@@ -133,6 +133,9 @@ fn bounds() {
     fn check_send<T: Send>() {}
     fn check_sync<T: Sync>() {}
     fn check_unpin<T: Unpin>() {}
+    // This has to take a value, since the async fn's return type is unnameable.
+    fn check_send_sync_val<T: Send + Sync>(_t: T) {}
+
     check_send::<RwLock<u32>>();
     check_sync::<RwLock<u32>>();
     check_unpin::<RwLock<u32>>();
@@ -142,6 +145,10 @@ fn bounds() {
 
     check_sync::<RwLockWriteGuard<'_, u32>>();
     check_unpin::<RwLockWriteGuard<'_, u32>>();
+
+    let rwlock = RwLock::new(0);
+    check_send_sync_val(rwlock.read());
+    check_send_sync_val(rwlock.write());
 }
 
 // As long as T: Send + Sync, it's fine to send and share RwLock<T> between threads.

tokio/src/sync/semaphore.rs

@@ -39,8 +39,15 @@ pub struct TryAcquireError(());
 #[cfg(not(loom))]
 fn bounds() {
     fn check_unpin<T: Unpin>() {}
+    // This has to take a value, since the async fn's return type is unnameable.
+    fn check_send_sync_val<T: Send + Sync>(_t: T) {}
+    fn check_send_sync<T: Send + Sync>() {}
     check_unpin::<Semaphore>();
     check_unpin::<SemaphorePermit<'_>>();
+    check_send_sync::<Semaphore>();
+
+    let semaphore = Semaphore::new(0);
+    check_send_sync_val(semaphore.acquire());
 }
 
 impl Semaphore {