sync: new internal semaphore based on intrusive lists (#2325)

## Motivation

Many of Tokio's synchronization primitives (`RwLock`, `Mutex`,
`Semaphore`, and the bounded MPSC channel) are based on the internal
semaphore implementation, called `semaphore_ll`. This semaphore type
provides a lower-level internal API for the semaphore implementation
than the public `Semaphore` type, and supports "batch" operations, where
waiters may acquire more than one permit at a time, and batches of
permits may be released back to the semaphore.

Currently, `semaphore_ll` uses an atomic singly-linked list for the
waiter queue. The linked list implementation is specific to the
semaphore. This implementation therefore requires a heap allocation for
every waiter in the queue. These allocations are owned by the semaphore,
rather than by the task awaiting permits from the semaphore. Critically,
they are only _deallocated_ when permits are released back to the
semaphore, at which point it dequeues as many waiters from the front of
the queue as can be satisfied with the released permits. If a task
attempts to acquire permits from the semaphore and is cancelled (such as
by timing out), their waiter nodes remain in the list until they are
dequeued while releasing permits. In cases where large numbers of tasks
are cancelled while waiting for permits, this results in extremely high
memory use for the semaphore (see #2237).

## Solution

@Matthias247 has proposed that Tokio adopt the approach used in his
`futures-intrusive` crate: using an _intrusive_ linked list to store the
wakers of tasks waiting on a synchronization primitive. In an intrusive
list, each list node is stored as part of the entry that node
represents, rather than in a heap allocation that owns the entry.
Because futures must be pinned in order to be polled, the necessary
invariant of such a list --- that entries may not move while in the list
--- may be upheld by making the waiter node `!Unpin`. In this approach,
the waiter node can be stored inline in the future, rather than
requiring  separate heap allocation, and cancelled futures may remove
their nodes from the list.

This branch adds a new semaphore implementation that uses the intrusive
list added to Tokio in #2210. The implementation is essentially a hybrid
of the old `semaphore_ll` and the semaphore used in `futures-intrusive`:
while a `Mutex` around the wait list is necessary, since the intrusive
list is not thread-safe, the permit state is stored outside of the mutex
and updated atomically. 

The mutex is acquired only when accessing the wait list — if a task 
can acquire sufficient permits without waiting, it does not need to
acquire the lock. When releasing permits, we iterate over the wait
list from the end of the queue until we run out of permits to release,
and split off all the nodes that received enough permits to wake up
into a separate list. Then, we can drain the new list and notify those
wakers *after* releasing the lock. Because the split operation only
modifies the pointers on the head node of the split-off list and the
new tail node of the old list, it is O(1) and does not require an
allocation to return a variable length number of waiters to notify.


Because of the intrusive list invariants, the API provided by the new
`batch_semaphore` is somewhat different than that of `semaphore_ll`. In
particular, the `Permit` type has been removed. This type was primarily
intended allow the reuse of a wait list node allocated on the heap.
Since the intrusive list means we can avoid heap-allocating waiters,
this is no longer necessary. Instead, acquiring permits is done by
polling an `Acquire` future returned by the `Semaphore` type. The use of
a future here ensures that the waiter node is always pinned while
waiting to acquire permits, and that a reference to the semaphore is
available to remove the waiter if the future is cancelled.
Unfortunately, the current implementation of the bounded MPSC requires a
`poll_acquire` operation, and has methods that call it while outside of
a pinned context. Therefore, I've left the old `semaphore_ll`
implementation in place to be used by the bounded MPSC, and updated the
`Mutex`, `RwLock`, and `Semaphore` APIs to use the new implementation.
Hopefully, a subsequent change can update the bounded MPSC to use the
new semaphore as well.

Fixes #2237

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

3 files changed, 423 insertions, 0 deletions

diff --git a/tokio/src/sync/tests/loom_semaphore_batch.rs b/tokio/src/sync/tests/loom_semaphore_batch.rs
new file mode 100644
index 00000000..4c1936c5
--- /dev/null
+++ b/tokio/src/sync/tests/loom_semaphore_batch.rs
@@ -0,0 +1,171 @@
+use crate::sync::batch_semaphore::*;
+
+use futures::future::poll_fn;
+use loom::future::block_on;
+use loom::sync::atomic::AtomicUsize;
+use loom::thread;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::Arc;
+use std::task::Poll::Ready;
+use std::task::{Context, Poll};
+
+#[test]
+fn basic_usage() {
+    const NUM: usize = 2;
+
+    struct Shared {
+        semaphore: Semaphore,
+        active: AtomicUsize,
+    }
+
+    async fn actor(shared: Arc<Shared>) {
+        shared.semaphore.acquire(1).await.unwrap();
+        let actual = shared.active.fetch_add(1, SeqCst);
+        assert!(actual <= NUM - 1);
+
+        let actual = shared.active.fetch_sub(1, SeqCst);
+        assert!(actual <= NUM);
+        shared.semaphore.release(1);
+    }
+
+    loom::model(|| {
+        let shared = Arc::new(Shared {
+            semaphore: Semaphore::new(NUM),
+            active: AtomicUsize::new(0),
+        });
+
+        for _ in 0..NUM {
+            let shared = shared.clone();
+
+            thread::spawn(move || {
+                block_on(actor(shared));
+            });
+        }
+
+        block_on(actor(shared));
+    });
+}
+
+#[test]
+fn release() {
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        {
+            let semaphore = semaphore.clone();
+            thread::spawn(move || {
+                block_on(semaphore.acquire(1)).unwrap();
+                semaphore.release(1);
+            });
+        }
+
+        block_on(semaphore.acquire(1)).unwrap();
+
+        semaphore.release(1);
+    });
+}
+
+#[test]
+fn basic_closing() {
+    const NUM: usize = 2;
+
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        for _ in 0..NUM {
+            let semaphore = semaphore.clone();
+
+            thread::spawn(move || {
+                for _ in 0..2 {
+                    block_on(semaphore.acquire(1)).map_err(|_| ())?;
+
+                    semaphore.release(1);
+                }
+
+                Ok::<(), ()>(())
+            });
+        }
+
+        semaphore.close();
+    });
+}
+
+#[test]
+fn concurrent_close() {
+    const NUM: usize = 3;
+
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(1));
+
+        for _ in 0..NUM {
+            let semaphore = semaphore.clone();
+
+            thread::spawn(move || {
+                block_on(semaphore.acquire(1)).map_err(|_| ())?;
+                semaphore.release(1);
+                semaphore.close();
+
+                Ok::<(), ()>(())
+            });
+        }
+    });
+}
+
+#[test]
+fn batch() {
+    let mut b = loom::model::Builder::new();
+    b.preemption_bound = Some(1);
+
+    b.check(|| {
+        let semaphore = Arc::new(Semaphore::new(10));
+        let active = Arc::new(AtomicUsize::new(0));
+        let mut ths = vec![];
+
+        for _ in 0..2 {
+            let semaphore = semaphore.clone();
+            let active = active.clone();
+
+            ths.push(thread::spawn(move || {
+                for n in &[4, 10, 8] {
+                    block_on(semaphore.acquire(*n)).unwrap();
+
+                    active.fetch_add(*n as usize, SeqCst);
+
+                    let num_active = active.load(SeqCst);
+                    assert!(num_active <= 10);
+
+                    thread::yield_now();
+
+                    active.fetch_sub(*n as usize, SeqCst);
+
+                    semaphore.release(*n as usize);
+                }
+            }));
+        }
+
+        for th in ths.into_iter() {
+            th.join().unwrap();
+        }
+
+        assert_eq!(10, semaphore.available_permits());
+    });
+}
+
+#[test]
+fn release_during_acquire() {
+    loom::model(|| {
+        let semaphore = Arc::new(Semaphore::new(10));
+        semaphore
+            .try_acquire(8)
+            .expect("try_acquire should succeed; semaphore uncontended");
+        let semaphore2 = semaphore.clone();
+        let thread = thread::spawn(move || block_on(semaphore2.acquire(4)).unwrap());
+
+        semaphore.release(8);
+        thread.join().unwrap();
+        semaphore.release(4);
+        assert_eq!(10, semaphore.available_permits());
+    })
+}
diff --git a/tokio/src/sync/tests/mod.rs b/tokio/src/sync/tests/mod.rs
index 7225ce9c..d571754c 100644
--- a/tokio/src/sync/tests/mod.rs
+++ b/tokio/src/sync/tests/mod.rs
@@ -1,6 +1,7 @@
 cfg_not_loom! {
     mod atomic_waker;
     mod semaphore_ll;
+    mod semaphore_batch;
 }
 
 cfg_loom! {
@@ -10,5 +11,6 @@ cfg_loom! {
     mod loom_mpsc;
     mod loom_notify;
     mod loom_oneshot;
+    mod loom_semaphore_batch;
     mod loom_semaphore_ll;
 }
diff --git a/tokio/src/sync/tests/semaphore_batch.rs b/tokio/src/sync/tests/semaphore_batch.rs
new file mode 100644
index 00000000..60f3f231
--- /dev/null
+++ b/tokio/src/sync/tests/semaphore_batch.rs
@@ -0,0 +1,250 @@
+use crate::sync::batch_semaphore::Semaphore;
+use tokio_test::*;
+
+#[test]
+fn poll_acquire_one_available() {
+    let s = Semaphore::new(100);
+    assert_eq!(s.available_permits(), 100);
+
+    // Polling for a permit succeeds immediately
+    assert_ready_ok!(task::spawn(s.acquire(1)).poll());
+    assert_eq!(s.available_permits(), 99);
+}
+
+#[test]
+fn poll_acquire_many_available() {
+    let s = Semaphore::new(100);
+    assert_eq!(s.available_permits(), 100);
+
+    // Polling for a permit succeeds immediately
+    assert_ready_ok!(task::spawn(s.acquire(5)).poll());
+    assert_eq!(s.available_permits(), 95);
+
+    assert_ready_ok!(task::spawn(s.acquire(5)).poll());
+    assert_eq!(s.available_permits(), 90);
+}
+
+#[test]
+fn try_acquire_one_available() {
+    let s = Semaphore::new(100);
+    assert_eq!(s.available_permits(), 100);
+
+    assert_ok!(s.try_acquire(1));
+    assert_eq!(s.available_permits(), 99);
+
+    assert_ok!(s.try_acquire(1));
+    assert_eq!(s.available_permits(), 98);
+}
+
+#[test]
+fn try_acquire_many_available() {
+    let s = Semaphore::new(100);
+    assert_eq!(s.available_permits(), 100);
+
+    assert_ok!(s.try_acquire(5));
+    assert_eq!(s.available_permits(), 95);
+
+    assert_ok!(s.try_acquire(5));
+    assert_eq!(s.available_permits(), 90);
+}
+
+#[test]
+fn poll_acquire_one_unavailable() {
+    let s = Semaphore::new(1);
+
+    // Acquire the first permit
+    assert_ready_ok!(task::spawn(s.acquire(1)).poll());
+    assert_eq!(s.available_permits(), 0);
+
+    let mut acquire_2 = task::spawn(s.acquire(1));
+    // Try to acquire the second permit
+    assert_pending!(acquire_2.poll());
+    assert_eq!(s.available_permits(), 0);
+
+    s.release(1);
+
+    assert_eq!(s.available_permits(), 0);
+    assert!(acquire_2.is_woken());
+    assert_ready_ok!(acquire_2.poll());
+    assert_eq!(s.available_permits(), 0);
+
+    s.release(1);
+    assert_eq!(s.available_permits(), 1);
+}
+
+#[test]
+fn poll_acquire_many_unavailable() {
+    let s = Semaphore::new(5);
+
+    // Acquire the first permit
+    assert_ready_ok!(task::spawn(s.acquire(1)).poll());
+    assert_eq!(s.available_permits(), 4);
+
+    // Try to acquire the second permit
+    let mut acquire_2 = task::spawn(s.acquire(5));
+    assert_pending!(acquire_2.poll());
+    assert_eq!(s.available_permits(), 0);
+
+    // Try to acquire the third permit
+    let mut acquire_3 = task::spawn(s.acquire(3));
+    assert_pending!(acquire_3.poll());
+    assert_eq!(s.available_permits(), 0);
+
+    s.release(1);
+
+    assert_eq!(s.available_permits(), 0);
+    assert!(acquire_2.is_woken());
+    assert_ready_ok!(acquire_2.poll());
+
+    assert!(!acquire_3.is_woken());
+    assert_eq!(s.available_permits(), 0);
+
+    s.release(1);
+    assert!(!acquire_3.is_woken());
+    assert_eq!(s.available_permits(), 0);
+
+    s.release(2);
+    assert!(acquire_3.is_woken());
+
+    assert_ready_ok!(acquire_3.poll());
+}
+
+#[test]
+fn try_acquire_one_unavailable() {
+    let s = Semaphore::new(1);
+
+    // Acquire the first permit
+    assert_ok!(s.try_acquire(1));
+    assert_eq!(s.available_permits(), 0);
+
+    assert_err!(s.try_acquire(1));
+
+    s.release(1);
+
+    assert_eq!(s.available_permits(), 1);
+    assert_ok!(s.try_acquire(1));
+
+    s.release(1);
+    assert_eq!(s.available_permits(), 1);
+}
+
+#[test]
+fn try_acquire_many_unavailable() {
+    let s = Semaphore::new(5);
+
+    // Acquire the first permit
+    assert_ok!(s.try_acquire(1));
+    assert_eq!(s.available_permits(), 4);
+
+    assert_err!(s.try_acquire(5));
+
+    s.release(1);
+    assert_eq!(s.available_permits(), 5);
+
+    assert_ok!(s.try_acquire(5));
+
+    s.release(1);
+    assert_eq!(s.available_permits(), 1);
+
+    s.release(1);
+    assert_eq!(s.available_permits(), 2);
+}
+
+#[test]
+fn poll_acquire_one_zero_permits() {
+    let s = Semaphore::new(0);
+    assert_eq!(s.available_permits(), 0);
+
+    // Try to acquire the permit
+    let mut acquire = task::spawn(s.acquire(1));
+    assert_pending!(acquire.poll());
+
+    s.release(1);
+
+    assert!(acquire.is_woken());
+    assert_ready_ok!(acquire.poll());
+}
+
+#[test]
+#[should_panic]
+fn validates_max_permits() {
+    use std::usize;
+    Semaphore::new((usize::MAX >> 2) + 1);
+}
+
+#[test]
+fn close_semaphore_prevents_acquire() {
+    let s = Semaphore::new(5);
+    s.close();
+
+    assert_eq!(5, s.available_permits());
+
+    assert_ready_err!(task::spawn(s.acquire(1)).poll());
+    assert_eq!(5, s.available_permits());
+
+    assert_ready_err!(task::spawn(s.acquire(1)).poll());
+    assert_eq!(5, s.available_permits());
+}
+
+#[test]
+fn close_semaphore_notifies_permit1() {
+    let s = Semaphore::new(0);
+    let mut acquire = task::spawn(s.acquire(1));
+
+    assert_pending!(acquire.poll());
+
+    s.close();
+
+    assert!(acquire.is_woken());
+    assert_ready_err!(acquire.poll());
+}
+
+#[test]
+fn close_semaphore_notifies_permit2() {
+    let s = Semaphore::new(2);
+
+    // Acquire a couple of permits
+    assert_ready_ok!(task::spawn(s.acquire(1)).poll());
+    assert_ready_ok!(task::spawn(s.acquire(1)).poll());
+
+    let mut acquire3 = task::spawn(s.acquire(1));
+    let mut acquire4 = task::spawn(s.acquire(1));
+    assert_pending!(acquire3.poll());
+    assert_pending!(acquire4.poll());
+
+    s.close();
+
+    assert!(acquire3.is_woken());
+    assert!(acquire4.is_woken());
+
+    assert_ready_err!(acquire3.poll());
+    assert_ready_err!(acquire4.poll());
+
+    assert_eq!(0, s.available_permits());
+
+    s.release(1);
+
+    assert_eq!(1, s.available_permits());
+
+    assert_ready_err!(task::spawn(s.acquire(1)).poll());
+
+    s.release(1);
+
+    assert_eq!(2, s.available_permits());
+}
+
+#[test]
+fn cancel_acquire_releases_permits() {
+    let s = Semaphore::new(10);
+    let _permit1 = s.try_acquire(4).expect("uncontended try_acquire succeeds");
+    assert_eq!(6, s.available_permits());
+
+    let mut acquire = task::spawn(s.acquire(8));
+    assert_pending!(acquire.poll());
+
+    assert_eq!(0, s.available_permits());
+    drop(acquire);
+
+    assert_eq!(6, s.available_permits());
+    assert_ok!(s.try_acquire(6));
+}