sync: simplify the broadcast channel (#2467)

Replace an ad hoc read/write lock with RwLock. Use
The parking_lot RwLock when possible. 

2 files changed, 39 insertions, 22 deletions

diff --git a/tokio/src/loom/std/mod.rs b/tokio/src/loom/std/mod.rs
index 595bdf60..60ee56ad 100644
--- a/tokio/src/loom/std/mod.rs
+++ b/tokio/src/loom/std/mod.rs
@@ -6,6 +6,8 @@ mod atomic_u32;
 mod atomic_u64;
 mod atomic_u8;
 mod atomic_usize;
+#[cfg(feature = "parking_lot")]
+mod parking_lot;
 mod unsafe_cell;
 
 pub(crate) mod cell {
@@ -41,24 +43,21 @@ pub(crate) mod rand {
 pub(crate) mod sync {
     pub(crate) use std::sync::Arc;
 
-    #[cfg(feature = "parking_lot")]
-    mod pl_wrappers;
-
     // Below, make sure all the feature-influenced types are exported for
     // internal use. Note however that some are not _currently_ named by
     // consuming code.
 
     #[cfg(feature = "parking_lot")]
     #[allow(unused_imports)]
-    pub(crate) use pl_wrappers::{Condvar, Mutex};
-
-    #[cfg(feature = "parking_lot")]
-    #[allow(unused_imports)]
-    pub(crate) use parking_lot::{MutexGuard, WaitTimeoutResult};
+    pub(crate) use crate::loom::std::parking_lot::{
+        Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard, WaitTimeoutResult,
+    };
 
     #[cfg(not(feature = "parking_lot"))]
     #[allow(unused_imports)]
-    pub(crate) use std::sync::{Condvar, Mutex, MutexGuard, WaitTimeoutResult};
+    pub(crate) use std::sync::{
+        Condvar, Mutex, MutexGuard, RwLock, RwLockReadGuard, WaitTimeoutResult,
+    };
 
     pub(crate) mod atomic {
         pub(crate) use crate::loom::std::atomic_ptr::AtomicPtr;
diff --git a/tokio/src/loom/std/sync/pl_wrappers.rs b/tokio/src/loom/std/parking_lot.rs
index 3be8ba1c..25d94af4 100644
--- a/tokio/src/loom/std/sync/pl_wrappers.rs
+++ b/tokio/src/loom/std/parking_lot.rs
@@ -6,25 +6,33 @@
 use std::sync::{LockResult, TryLockError, TryLockResult};
 use std::time::Duration;
 
-use parking_lot as pl;
+// Types that do not need wrapping
+pub(crate) use parking_lot::{MutexGuard, RwLockReadGuard, RwLockWriteGuard, WaitTimeoutResult};
 
 /// Adapter for `parking_lot::Mutex` to the `std::sync::Mutex` interface.
 #[derive(Debug)]
-pub(crate) struct Mutex<T: ?Sized>(pl::Mutex<T>);
+pub(crate) struct Mutex<T: ?Sized>(parking_lot::Mutex<T>);
+
+#[derive(Debug)]
+pub(crate) struct RwLock<T>(parking_lot::RwLock<T>);
+
+/// Adapter for `parking_lot::Condvar` to the `std::sync::Condvar` interface.
+#[derive(Debug)]
+pub(crate) struct Condvar(parking_lot::Condvar);
 
 impl<T> Mutex<T> {
     #[inline]
     pub(crate) fn new(t: T) -> Mutex<T> {
-        Mutex(pl::Mutex::new(t))
+        Mutex(parking_lot::Mutex::new(t))
     }
 
     #[inline]
-    pub(crate) fn lock(&self) -> LockResult<pl::MutexGuard<'_, T>> {
+    pub(crate) fn lock(&self) -> LockResult<MutexGuard<'_, T>> {
         Ok(self.0.lock())
     }
 
     #[inline]
-    pub(crate) fn try_lock(&self) -> TryLockResult<pl::MutexGuard<'_, T>> {
+    pub(crate) fn try_lock(&self) -> TryLockResult<MutexGuard<'_, T>> {
         match self.0.try_lock() {
             Some(guard) => Ok(guard),
             None => Err(TryLockError::WouldBlock),
@@ -35,14 +43,24 @@ impl<T> Mutex<T> {
     // provided here as needed.
 }
 
-/// Adapter for `parking_lot::Condvar` to the `std::sync::Condvar` interface.
-#[derive(Debug)]
-pub(crate) struct Condvar(pl::Condvar);
+impl<T> RwLock<T> {
+    pub(crate) fn new(t: T) -> RwLock<T> {
+        RwLock(parking_lot::RwLock::new(t))
+    }
+
+    pub(crate) fn read(&self) -> LockResult<RwLockReadGuard<'_, T>> {
+        Ok(self.0.read())
+    }
+
+    pub(crate) fn write(&self) -> LockResult<RwLockWriteGuard<'_, T>> {
+        Ok(self.0.write())
+    }
+}
 
 impl Condvar {
     #[inline]
     pub(crate) fn new() -> Condvar {
-        Condvar(pl::Condvar::new())
+        Condvar(parking_lot::Condvar::new())
     }
 
     #[inline]
@@ -58,8 +76,8 @@ impl Condvar {
     #[inline]
     pub(crate) fn wait<'a, T>(
         &self,
-        mut guard: pl::MutexGuard<'a, T>,
-    ) -> LockResult<pl::MutexGuard<'a, T>> {
+        mut guard: MutexGuard<'a, T>,
+    ) -> LockResult<MutexGuard<'a, T>> {
         self.0.wait(&mut guard);
         Ok(guard)
     }
@@ -67,9 +85,9 @@ impl Condvar {
     #[inline]
     pub(crate) fn wait_timeout<'a, T>(
         &self,
-        mut guard: pl::MutexGuard<'a, T>,
+        mut guard: MutexGuard<'a, T>,
         timeout: Duration,
-    ) -> LockResult<(pl::MutexGuard<'a, T>, pl::WaitTimeoutResult)> {
+    ) -> LockResult<(MutexGuard<'a, T>, WaitTimeoutResult)> {
         let wtr = self.0.wait_for(&mut guard, timeout);
         Ok((guard, wtr))
     }