Fix some panics after futures get dropped

If a Future gets dropped after being polled() but before gaining
ownership of the Mutex or RwLock, a panic would result when the owner
tried to transfer ownership to the dropped Future.  Fix the drop methods
to handle waiting Futures that have already disappeared.

Kudos to Kayle Gishen for reporting and diagnosing the bug

Fixes #24

src/mutex.rs

@@ -322,13 +322,15 @@ impl<T: ?Sized> Mutex<T> {
     fn unlock(&self) {
         let mut mtx_data = self.inner.mutex.lock().expect("sync::Mutex::lock");
         assert!(mtx_data.owned);
-        if let Some(tx) = mtx_data.waiters.pop_front() {
-            // Send ownership to the waiter
-            tx.send(()).expect("Sender::send");
-        } else {
-            // Relinquish ownership
-            mtx_data.owned = false;
+
+        while let Some(tx) = mtx_data.waiters.pop_front() {
+            if tx.send(()).is_ok() {
+                return;
+            }
+            // An error indicates that the waiter's future was dropped
         }
+        // Relinquish ownership
+        mtx_data.owned = false;
     }
 
     /// Returns true if the two `Mutex` point to the same data else false.

src/rwlock.rs

@@ -450,31 +450,38 @@ impl<T: ?Sized> RwLock<T> {
         assert!(lock_data.num_readers > 0);
         assert!(!lock_data.exclusive);
         assert_eq!(lock_data.read_waiters.len(), 0);
-        if lock_data.num_readers == 1 {
-            if let Some(tx) = lock_data.write_waiters.pop_front() {
-                lock_data.num_readers -= 1;
-                lock_data.exclusive = true;
-                tx.send(()).expect("Sender::send");
-                return;
+        lock_data.num_readers -= 1;
+        if lock_data.num_readers == 0 {
+            while let Some(tx) = lock_data.write_waiters.pop_front() {
+                if tx.send(()).is_ok() {
+                    lock_data.exclusive = true;
+                    return
+                }
             }
         }
-        lock_data.num_readers -= 1;
     }
 
     /// Release an exclusive lock of an `RwLock`.
     fn unlock_writer(&self) {
         let mut lock_data = self.inner.mutex.lock().expect("sync::Mutex::lock");
         assert!(lock_data.num_readers == 0);
         assert!(lock_data.exclusive);
-        if let Some(tx) = lock_data.write_waiters.pop_front() {
-            tx.send(()).expect("Sender::send");
-        } else {
-            lock_data.exclusive = false;
-            lock_data.num_readers += lock_data.read_waiters.len() as u32;
-            for tx in lock_data.read_waiters.drain(..) {
-                tx.send(()).expect("Sender::send");
+
+        // First try to wake up any writers
+        while let Some(tx) = lock_data.write_waiters.pop_front() {
+            if tx.send(()).is_ok() {
+                return;
             }
         }
+
+        // If there are no writers, try to wake up readers
+        lock_data.exclusive = false;
+        lock_data.num_readers += lock_data.read_waiters.len() as u32;
+        for tx in lock_data.read_waiters.drain(..) {
+            // Ignore errors, which are due to a reader's future getting
+            // dropped before it was ready
+            let _ = tx.send(());
+        }
     }
 }
 

tests/mutex.rs

@@ -47,10 +47,11 @@ fn mutex_eq_ptr_false() {
     assert!(!Mutex::ptr_eq(&mutex, &mutex_other));
 }
 
-// When a pending Mutex gets dropped, it should drain its channel and relinquish
-// ownership if a message was found.  If not, deadlocks may result.
+// When a Mutex gets dropped after gaining ownership but before being polled, it
+// should drain its channel and relinquish ownership if a message was found.  If
+// not, deadlocks may result.
 #[test]
-fn drop_before_poll_returns_ready() {
+fn drop_when_ready() {
     let mutex = Mutex::<u32>::new(0);
     let mut rt = current_thread::Runtime::new().unwrap();
 
@@ -70,6 +71,30 @@ fn drop_before_poll_returns_ready() {
     })).unwrap();
 }
 
+// When a pending Mutex gets dropped after being polled() but before gaining
+// ownership, ownership should pass on to the next waiter.
+#[test]
+fn drop_before_ready() {
+    let mutex = Mutex::<u32>::new(0);
+    let mut rt = current_thread::Runtime::new().unwrap();
+
+    rt.block_on(lazy(|| {
+        let mut fut1 = mutex.lock();
+        let guard1 = fut1.poll();    // fut1 immediately gets ownership
+        assert!(guard1.as_ref().unwrap().is_ready());
+        let mut fut2 = mutex.lock();
+        assert!(!fut2.poll().unwrap().is_ready());
+        let mut fut3 = mutex.lock();
+        assert!(!fut3.poll().unwrap().is_ready());
+        drop(fut2);                  // drop before gaining ownership
+        drop(guard1);                // ownership transfers to fut3
+        drop(fut1);
+        let guard3 = fut3.poll();
+        assert!(guard3.as_ref().unwrap().is_ready());
+        future::ok::<(), ()>(())
+    })).unwrap();
+}
+
 // Mutably dereference a uniquely owned Mutex
 #[test]
 fn get_mut() {

tests/rwlock.rs

@@ -10,8 +10,9 @@ use tokio::runtime::current_thread;
 use futures_locks::*;
 
 
-// When an exclusively owned but not yet polled RwLock future is dropped, it
-// should relinquish ownership.  If not, deadlocks may result.
+// When an exclusively owned RwLock future is dropped after gaining ownership
+// but before begin polled, it should relinquish ownership.  If not, deadlocks
+// may result.
 #[test]
 fn drop_exclusive_before_poll_returns_ready() {
     let rwlock = RwLock::<u32>::new(42);
@@ -32,8 +33,57 @@ fn drop_exclusive_before_poll_returns_ready() {
     })).unwrap();
 }
 
-// When an nonexclusively owned but not yet polled RwLock future is dropped, it
-// should relinquish ownership.  If not, deadlocks may result.
+// When a pending exclusive RwLock gets dropped after being polled() but before
+// gaining ownership, ownership should pass on to the next waiter.
+#[test]
+fn drop_exclusive_before_ready() {
+    let rwlock = RwLock::<u32>::new(42);
+    let mut rt = current_thread::Runtime::new().unwrap();
+
+    rt.block_on(lazy(|| {
+        let mut fut1 = rwlock.read();
+        let guard1 = fut1.poll();    // fut1 immediately gets ownership
+        assert!(guard1.as_ref().unwrap().is_ready());
+        let mut fut2 = rwlock.write();
+        assert!(!fut2.poll().unwrap().is_ready());
+        let mut fut3 = rwlock.write();
+        assert!(!fut3.poll().unwrap().is_ready());
+        drop(fut2);                  // drop before gaining ownership
+        drop(guard1);                // ownership transfers to fut3
+        drop(fut1);
+        let guard3 = fut3.poll();
+        assert!(guard3.as_ref().unwrap().is_ready());
+        future::ok::<(), ()>(())
+    })).unwrap();
+}
+
+// Like drop_exclusive_before_ready, but the rwlock is already locked in
+// exclusive mode.
+#[test]
+fn drop_exclusive_before_ready_2() {
+    let rwlock = RwLock::<u32>::new(42);
+    let mut rt = current_thread::Runtime::new().unwrap();
+
+    rt.block_on(lazy(|| {
+        let mut fut1 = rwlock.write();
+        let guard1 = fut1.poll();    // fut1 immediately gets ownership
+        assert!(guard1.as_ref().unwrap().is_ready());
+        let mut fut2 = rwlock.write();
+        assert!(!fut2.poll().unwrap().is_ready());
+        let mut fut3 = rwlock.write();
+        assert!(!fut3.poll().unwrap().is_ready());
+        drop(fut2);                  // drop before gaining ownership
+        drop(guard1);                // ownership transfers to fut3
+        drop(fut1);
+        let guard3 = fut3.poll();
+        assert!(guard3.as_ref().unwrap().is_ready());
+        future::ok::<(), ()>(())
+    })).unwrap();
+}
+
+// When a nonexclusively owned RwLock future is dropped after gaining ownership
+// but before begin polled, it should relinquish ownership.  If not, deadlocks
+// may result.
 #[test]
 fn drop_shared_before_poll_returns_ready() {
     let rwlock = RwLock::<u32>::new(42);
@@ -54,6 +104,30 @@ fn drop_shared_before_poll_returns_ready() {
     })).unwrap();
 }
 
+// When a pending shared RwLock gets dropped after being polled() but before
+// gaining ownership, ownership should pass on to the next waiter.
+#[test]
+fn drop_shared_before_ready() {
+    let rwlock = RwLock::<u32>::new(42);
+    let mut rt = current_thread::Runtime::new().unwrap();
+
+    rt.block_on(lazy(|| {
+        let mut fut1 = rwlock.write();
+        let guard1 = fut1.poll();    // fut1 immediately gets ownership
+        assert!(guard1.as_ref().unwrap().is_ready());
+        let mut fut2 = rwlock.read();
+        assert!(!fut2.poll().unwrap().is_ready());
+        let mut fut3 = rwlock.read();
+        assert!(!fut3.poll().unwrap().is_ready());
+        drop(fut2);                  // drop before gaining ownership
+        drop(guard1);                // ownership transfers to fut3
+        drop(fut1);
+        let guard3 = fut3.poll();
+        assert!(guard3.as_ref().unwrap().is_ready());
+        future::ok::<(), ()>(())
+    })).unwrap();
+}
+
 // Mutably dereference a uniquely owned RwLock
 #[test]
 fn get_mut() {