proxy/router: Create a separate `cache` module (#920)

The router's `Inner` type contains a map of routes. Recently, this map's
capacity has become constrained to prevent leakage for long-running
processes.

This change prepares for a fuller LRU implementation by moving the
router's `Inner` type to a new (tested) module, `cache`.

proxy/router/src/cache.rs

@@ -0,0 +1,128 @@
+use indexmap::IndexMap;
+use std::hash::Hash;
+
+// Reexported so IndexMap isn't exposed.
+pub use indexmap::Equivalent;
+
+/// A cache for routes
+///
+/// ## Assumptions
+///
+/// - `access` is common;
+/// - `store` is less common;
+/// - `capacity` is large enough..
+///
+/// ## Complexity
+///
+/// - `access` computes in O(1) time (amortized average).
+/// - `store` computes in O(1) time (average).
+// TODO LRU
+pub struct Cache<K: Hash + Eq, V> {
+    vals: IndexMap<K, V>,
+    capacity: usize,
+}
+
+/// A handle to a `Cache` that has capacity for at least one additional value.
+pub struct Reserve<'a, K: Hash + Eq + 'a, V: 'a> {
+    vals: &'a mut IndexMap<K, V>,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct CapacityExhausted {
+    pub capacity: usize,
+}
+
+impl<K: Hash + Eq, V> Cache<K, V> {
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            capacity,
+            vals: IndexMap::default(),
+        }
+    }
+
+    /// Accesses a route.
+    // TODO track access times for each entry.
+    pub fn access<Q>(&mut self, key: &Q) -> Option<&mut V>
+    where
+        Q: Hash + Equivalent<K>,
+    {
+        self.vals.get_mut(key)
+    }
+
+    /// Ensures that there is capacity to store an additional route.
+    ///
+    /// An error is returned if there is no available capacity.
+    // TODO evict old entries
+    pub fn reserve(&mut self) -> Result<Reserve<K, V>, CapacityExhausted> {
+        let avail = self.capacity - self.vals.len();
+        if avail == 0 {
+            // TODO If the cache is full, evict the oldest inactive route. If all
+            // routes are active, fail the request.
+            return Err(CapacityExhausted {
+                capacity: self.capacity,
+            });
+        }
+
+        Ok(Reserve {
+            vals: &mut self.vals,
+        })
+    }
+}
+
+impl<'a, K: Hash + Eq + 'a, V: 'a> Reserve<'a, K, V> {
+    /// Stores a route in the cache.
+    pub fn store(self, key: K, val: V) {
+        self.vals.insert(key, val);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use test_util::MultiplyAndAssign;
+
+    #[test]
+    fn reserve_and_store() {
+        let mut cache = Cache::<_, MultiplyAndAssign>::new(2);
+
+        {
+            let r = cache.reserve().expect("reserve");
+            r.store(1, MultiplyAndAssign::default());
+        }
+        assert_eq!(cache.vals.len(), 1);
+
+        {
+            let r = cache.reserve().expect("reserve");
+            r.store(2, MultiplyAndAssign::default());
+        }
+        assert_eq!(cache.vals.len(), 2);
+
+        assert_eq!(
+            cache.reserve().err(),
+            Some(CapacityExhausted { capacity: 2 })
+        );
+        assert_eq!(cache.vals.len(), 2);
+    }
+
+    #[test]
+    fn store_and_access() {
+        let mut cache = Cache::<_, MultiplyAndAssign>::new(2);
+
+        assert!(cache.access(&1).is_none());
+        assert!(cache.access(&2).is_none());
+
+        {
+            let r = cache.reserve().expect("reserve");
+            r.store(1, MultiplyAndAssign::default());
+        }
+        assert!(cache.access(&1).is_some());
+        assert!(cache.access(&2).is_none());
+
+        {
+            let r = cache.reserve().expect("reserve");
+            r.store(2, MultiplyAndAssign::default());
+        }
+        assert!(cache.access(&1).is_some());
+        assert!(cache.access(&2).is_some());
+    }
+}

proxy/router/src/lib.rs

@@ -3,13 +3,16 @@ extern crate indexmap;
 extern crate tower_service;
 
 use futures::{Future, Poll};
-use indexmap::IndexMap;
 use tower_service::Service;
 
 use std::{error, fmt, mem};
 use std::hash::Hash;
 use std::sync::{Arc, Mutex};
 
+mod cache;
+
+use self::cache::Cache;
+
 /// Routes requests based on a configurable `Key`.
 pub struct Router<T>
 where T: Recognize,
@@ -75,12 +78,6 @@ where T: Recognize,
     cache: Mutex<Cache<T::Key, T::Service>>,
 }
 
-struct Cache<K: Hash + Eq, V>
-{
-    routes: IndexMap<K, V>,
-    capacity: usize,
-}
-
 enum State<T>
 where T: Recognize,
 {
@@ -100,10 +97,7 @@ where T: Recognize
         Router {
             inner: Arc::new(Inner {
                 recognize,
-                cache: Mutex::new(Cache {
-                    routes: IndexMap::default(),
-                    capacity,
-                }),
+                cache: Mutex::new(Cache::new(capacity)),
             }),
         }
     }
@@ -140,17 +134,18 @@ where T: Recognize,
         let cache = &mut *self.inner.cache.lock().expect("lock router cache");
 
         // First, try to load a cached route for `key`.
-        if let Some(service) = cache.routes.get_mut(&key) {
+        if let Some(service) = cache.access(&key) {
             return ResponseFuture::new(service.call(request));
         }
 
         // Since there wasn't a cached route, ensure that there is capacity for a
         // new one.
-        if cache.routes.len() == cache.capacity {
-            // TODO If the cache is full, evict the oldest inactive route. If all
-            // routes are active, fail the request.
-            return ResponseFuture::no_capacity(cache.capacity);
-        }
+        let reserve = match cache.reserve() {
+            Ok(r) => r,
+            Err(cache::CapacityExhausted { capacity }) => {
+                return ResponseFuture::no_capacity(capacity);
+            }
+        };
 
         // Bind a new route, send the request on the route, and cache the route.
         let mut service = match self.inner.recognize.bind_service(&key) {
@@ -159,7 +154,8 @@ where T: Recognize,
         };
 
         let response = service.call(request);
-        cache.routes.insert(key, service);
+        reserve.store(key, service);
+
         ResponseFuture::new(response)
     }
 }
@@ -256,18 +252,17 @@ where
 }
 
 #[cfg(test)]
-mod tests {
+mod test_util {
     use futures::{Poll, Future, future};
     use tower_service::Service;
-    use super::{Error, Router};
 
-    struct Recognize;
+    pub struct Recognize;
 
-    struct MultiplyAndAssign(usize);
+    pub struct MultiplyAndAssign(usize);
 
-    enum Request {
+    pub enum Request {
         NotRecognized,
-        Recgonized(usize),
+        Recognized(usize),
     }
 
     impl super::Recognize for Recognize {
@@ -281,7 +276,7 @@ mod tests {
         fn recognize(&self, req: &Self::Request) -> Option<Self::Key> {
             match *req {
                 Request::NotRecognized => None,
-                Request::Recgonized(n) => Some(n),
+                Request::Recognized(n) => Some(n),
             }
         }
 
@@ -303,22 +298,40 @@ mod tests {
         fn call(&mut self, req: Self::Request) -> Self::Future {
             let n = match req {
                 Request::NotRecognized => unreachable!(),
-                Request::Recgonized(n) => n,
+                Request::Recognized(n) => n,
             };
             self.0 *= n;
             future::ok(self.0)
         }
     }
 
-    impl Router<Recognize> {
-        fn call_ok(&mut self, req: Request) -> usize {
+    impl From<usize> for Request {
+        fn from(n: usize) -> Request {
+            Request::Recognized(n)
+        }
+    }
+
+    impl Default for MultiplyAndAssign {
+        fn default() -> Self {
+            MultiplyAndAssign(1)
+        }
+    }
+
+    impl super::Router<Recognize> {
+        pub fn call_ok(&mut self, req: Request) -> usize {
             self.call(req).wait().expect("should route")
         }
 
-        fn call_err(&mut self, req: Request) -> super::Error<(), ()> {
+        pub fn call_err(&mut self, req: Request) -> super::Error<(), ()> {
             self.call(req).wait().expect_err("should not route")
         }
     }
+}
+
+#[cfg(test)]
+mod tests {
+    use test_util::*;
+    use super::{Error, Router};
 
     #[test]
     fn invalid() {
@@ -332,10 +345,10 @@ mod tests {
     fn cache_limited_by_capacity() {
         let mut router = Router::new(Recognize, 1);
 
-        let rsp = router.call_ok(Request::Recgonized(2));
+        let rsp = router.call_ok(2.into());
         assert_eq!(rsp, 2);
 
-        let rsp = router.call_err(Request::Recgonized(3));
+        let rsp = router.call_err(3.into());
         assert_eq!(rsp, Error::NoCapacity(1));
     }
 
@@ -343,10 +356,10 @@ mod tests {
     fn services_cached() {
         let mut router = Router::new(Recognize, 1);
 
-        let rsp = router.call_ok(Request::Recgonized(2));
+        let rsp = router.call_ok(2.into());
         assert_eq!(rsp, 2);
 
-        let rsp = router.call_ok(Request::Recgonized(2));
+        let rsp = router.call_ok(2.into());
         assert_eq!(rsp, 4);
     }
 }

proxy/src/inbound.rs

@@ -31,6 +31,18 @@ impl<B> Inbound<B> {
     }
 }
 
+impl<B> Clone for Inbound<B>
+where
+    B: tower_h2::Body + 'static,
+{
+    fn clone(&self) -> Self {
+        Self {
+            bind: self.bind.clone(),
+            default_addr: self.default_addr.clone(),
+        }
+    }
+}
+
 impl<B> Recognize for Inbound<B>
 where
     B: tower_h2::Body + 'static,

proxy/src/outbound.rs

@@ -32,6 +32,12 @@ pub struct Outbound<B> {
 
 const MAX_IN_FLIGHT: usize = 10_000;
 
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum Destination {
+    Hostname(DnsNameAndPort),
+    ImplicitOriginalDst(SocketAddr),
+}
+
 // ===== impl Outbound =====
 
 impl<B> Outbound<B> {
@@ -47,10 +53,17 @@ impl<B> Outbound<B> {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-pub enum Destination {
-    Hostname(DnsNameAndPort),
-    ImplicitOriginalDst(SocketAddr),
+impl<B> Clone for Outbound<B>
+where
+    B: tower_h2::Body + 'static,
+{
+    fn clone(&self) -> Self {
+        Self {
+            bind: self.bind.clone(),
+            discovery: self.discovery.clone(),
+            bind_timeout: self.bind_timeout.clone(),
+        }
+    }
 }
 
 impl<B> Recognize for Outbound<B>