cache: Ensure that actively held services are not evicted (#768)

Cache eviction is currently triggered when a service has not processed
new requests for some idle timeout. This is fragile for caches that may
process a few long-lived requests. In practice, we would prefer to only
start tracking idleness when there are no *active* requests on a
service.

This change restructures the cache to return services that are wrapped
with tracking handles, rather than passing a tracking handle into the
inner service. When a returned service is dropped, it spawns a
background task that retains the handle for an idle timeout and, if no
new instances have acquire the handle have that timeout, removes the
service from the cache.

This change reduces latency as well as CPU and memory utilization in
load tests. Furthermore, it ultimately eliminates the need for a specialized
buffer implementation.

Fixes linkerd/linkerd2#5334

Author: olix0r

Cargo.lock

@@ -1011,6 +1011,7 @@ dependencies = [
  "http-body",
  "hyper",
  "linkerd2-app-core",
+ "linkerd2-identity",
  "regex 0.1.80",
  "tokio 0.3.5",
  "tokio-test 0.3.0",
@@ -1044,9 +1045,11 @@ dependencies = [
  "linkerd2-error",
  "linkerd2-stack",
  "parking_lot",
- "tokio 0.2.23",
+ "tokio 0.3.5",
  "tower",
  "tracing",
+ "tracing-futures",
+ "tracing-subscriber",
 ]
 
 [[package]]

linkerd/app/core/src/svc.rs

@@ -78,18 +78,6 @@ impl<L> Layers<L> {
         self.push(buffer::SpawnBufferLayer::new(capacity))
     }
 
-    pub fn push_spawn_buffer_with_idle_timeout<Req, Rsp>(
-        self,
-        capacity: usize,
-        idle_timeout: Duration,
-    ) -> Layers<Pair<L, buffer::SpawnBufferLayer<Req, Rsp>>>
-    where
-        Req: Send + 'static,
-        Rsp: Send + 'static,
-    {
-        self.push(buffer::SpawnBufferLayer::new(capacity).with_idle_timeout(idle_timeout))
-    }
-
     // Makes the service eagerly process and fail requests after the given timeout.
     pub fn push_failfast(self, timeout: Duration) -> Layers<Pair<L, timeout::FailFastLayer>> {
         self.push(timeout::FailFastLayer::new(timeout))
@@ -239,14 +227,13 @@ impl<S> Stack<S> {
         self.push(http::insert::target::layer())
     }
 
-    pub fn cache<T, L, U>(self, track: L) -> Stack<cache::Cache<T, cache::layer::NewTrack<L, S>>>
+    pub fn push_cache<T>(self, idle: Duration) -> Stack<cache::Cache<T, S>>
     where
-        T: Eq + std::hash::Hash + Send + 'static,
-        S: NewService<T> + Clone,
-        L: tower::layer::Layer<cache::layer::Track<S>> + Clone,
-        L::Service: NewService<T, Service = U>,
+        T: Eq + std::hash::Hash + Send + Sync + 'static,
+        S: NewService<T> + 'static,
+        S::Service: Send + Sync + 'static,
     {
-        self.push(cache::CacheLayer::new(track))
+        self.push(cache::Cache::layer(idle))
     }
 
     /// Push a service that either calls the inner service if it is ready, or

linkerd/app/inbound/src/lib.rs

@@ -24,7 +24,7 @@ use linkerd2_app_core::{
     },
     reconnect,
     spans::SpanConverter,
-    svc::{self},
+    svc,
     transport::{self, io, listen, tls},
     Error, NameAddr, NameMatch, TraceContext, DST_OVERRIDE_HEADER,
 };
@@ -165,8 +165,9 @@ impl Config {
             Error = Error,
             Future = impl Send,
         > + Clone
-                      + Unpin
-                      + Send,
+                      + Send
+                      + Sync
+                      + Unpin,
     > + Unpin
            + Clone
            + Send
@@ -275,14 +276,17 @@ impl Config {
         // fails, skips that stack to forward to the local endpoint.
         svc::stack(switch_loopback)
             .check_new_service::<Target, http::Request<http::boxed::BoxBody>>()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .push(metrics.stack.layer(stack_labels("logical")))
-                        .box_http_response(),
-                ),
+            .push_on_response(
+                svc::layers()
+                    .push_failfast(dispatch_timeout)
+                    .push_spawn_buffer(buffer_capacity)
+                    .push(metrics.stack.layer(stack_labels("logical"))),
+            )
+            .push_cache(cache_max_idle_age)
+            .push_on_response(
+                svc::layers()
+                    .push(http::Retain::layer())
+                    .box_http_response(),
             )
             // Boxing is necessary purely to limit the link-time overhead of
             // having enormous types.
@@ -358,9 +362,9 @@ impl Config {
             .push(svc::layer::mk(|inner| {
                 svc::stack::NewRouter::new(RequestTarget::from, inner)
             }))
+            .check_new_service::<TcpAccept, http::Request<_>>()
             // Used by tap.
             .push_http_insert_target()
-            .check_new_service::<TcpAccept, http::Request<_>>()
             .push_on_response(
                 svc::layers()
                     .push(http_admit_request)