lock: Generalize to protect a guarded value (#431)

We used Tokio's Lock implementation in the router's cache
implementation, though we know it can leak under contention.

This change generalizes the Lock to return a guarded value (like Tokio's
lock). This change simplifies the Lock's state management: the Lock may
no longer hold a value, nor can it fail.

The `lock::Service` implementation now holds a `Result<Service,
ServiceError>` so that lock services may still broadcast the inner
service's failure.

Author: olix0r

Cargo.lock

@@ -1216,6 +1216,7 @@ dependencies = [
  "futures",
  "indexmap",
  "linkerd2-error",
+ "linkerd2-lock",
  "tokio",
  "tokio-sync",
  "tokio-timer",

linkerd/lock/src/error.rs

@@ -1,28 +1,9 @@
 pub use linkerd2_error::Error;
 use std::sync::Arc;
 
-#[derive(Debug)]
-pub struct Poisoned(());
-
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct ServiceError(Arc<Error>);
 
-// === impl Poisoned ===
-
-impl Poisoned {
-    pub fn new() -> Self {
-        Poisoned(())
-    }
-}
-
-impl std::fmt::Display for Poisoned {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "poisoned")
-    }
-}
-
-impl std::error::Error for Poisoned {}
-
 // === impl ServiceError ===
 
 impl ServiceError {

linkerd/lock/src/layer.rs

@@ -1,4 +1,4 @@
-use super::Lock;
+use super::LockService;
 
 #[derive(Clone, Debug, Default)]
 pub struct LockLayer(());
@@ -12,7 +12,7 @@ impl LockLayer {
 }
 
 impl<S> tower::layer::Layer<S> for LockLayer {
-    type Service = Lock<S>;
+    type Service = LockService<S>;
 
     fn layer(&self, inner: S) -> Self::Service {
         Self::Service::new(inner)

linkerd/lock/src/lib.rs

@@ -4,9 +4,14 @@
 
 pub mod error;
 mod layer;
+mod lock;
 mod service;
 mod shared;
 #[cfg(test)]
 mod test;
 
-pub use self::{layer::LockLayer, service::Lock};
+pub use self::{
+    layer::LockLayer,
+    lock::{Guard, Lock},
+    service::LockService,
+};

linkerd/lock/src/lock.rs

@@ -0,0 +1,122 @@
+use crate::shared::{Shared, Wait};
+use futures::Async;
+use std::sync::{Arc, Mutex};
+
+/// Provides mutually exclusive to a `T`-typed value, asynchronously.
+///
+/// Note that, when the lock is contested, waiters are notified in a LIFO
+/// fashion. This is done to minimize latency at the expense of unfairness.
+pub struct Lock<T> {
+    /// Set when this Lock is interested in acquiring the value.
+    waiting: Option<Wait>,
+    shared: Arc<Mutex<Shared<T>>>,
+}
+
+/// Guards access to a `T`-typed value, ensuring the value is released on Drop.
+pub struct Guard<T> {
+    /// Must always be Some; Used to reclaim the value in Drop.
+    value: Option<T>,
+
+    shared: Arc<Mutex<Shared<T>>>,
+}
+
+// === impl Lock ===
+
+impl<S> Lock<S> {
+    pub fn new(service: S) -> Self {
+        Self {
+            waiting: None,
+            shared: Arc::new(Mutex::new(Shared::new(service))),
+        }
+    }
+}
+
+impl<S> Clone for Lock<S> {
+    fn clone(&self) -> Self {
+        Self {
+            // Clones have an independent local lock state.
+            waiting: None,
+            shared: self.shared.clone(),
+        }
+    }
+}
+
+impl<T> Lock<T> {
+    fn guard(&self, value: T) -> Guard<T> {
+        Guard {
+            value: Some(value),
+            shared: self.shared.clone(),
+        }
+    }
+
+    pub fn poll_acquire(&mut self) -> Async<Guard<T>> {
+        let mut shared = self.shared.lock().expect("Lock poisoned");
+
+        loop {
+            self.waiting = match self.waiting {
+                // This instance has not registered interest in the lock.
+                None => match shared.acquire() {
+                    // This instance has exclusive access to the inner service.
+                    Some(value) => {
+                        // The state remains Released.
+                        return Async::Ready(self.guard(value));
+                    }
+                    None => Some(Wait::default()),
+                },
+
+                // This instance is interested in the lock.
+                Some(ref waiter) => match shared.poll_acquire(waiter) {
+                    Async::NotReady => return Async::NotReady,
+                    Async::Ready(value) => {
+                        self.waiting = None;
+                        return Async::Ready(self.guard(value));
+                    }
+                },
+            };
+        }
+    }
+}
+
+impl<T> Drop for Lock<T> {
+    fn drop(&mut self) {
+        if let Some(wait) = self.waiting.take() {
+            if let Ok(mut shared) = self.shared.lock() {
+                shared.release_waiter(wait);
+            }
+        }
+    }
+}
+
+impl<T> std::ops::Deref for Guard<T> {
+    type Target = T;
+    fn deref(&self) -> &Self::Target {
+        self.value.as_ref().expect("Value dropped from guard")
+    }
+}
+
+impl<T> std::ops::DerefMut for Guard<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.value.as_mut().expect("Value dropped from guard")
+    }
+}
+
+impl<T: std::fmt::Debug> std::fmt::Debug for Guard<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Guard({:?})", &**self)
+    }
+}
+
+impl<T: std::fmt::Display> std::fmt::Display for Guard<T> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(&**self, f)
+    }
+}
+
+impl<T> Drop for Guard<T> {
+    fn drop(&mut self) {
+        let value = self.value.take().expect("Guard may not be dropped twice");
+        if let Ok(mut shared) = self.shared.lock() {
+            shared.release_and_notify(value);
+        }
+    }
+}