replace linkerd-concurrency-limit with Tower (#1107)

tower-rs/tower#574 added a `GlobalConcurrencyLimitLayer` which has the
same behavior as our `linkerd-concurrency-limit` crate (sharing a single
semaphore across all instances of a service). This was published in
Tower 0.4.8, and now that we've updated to that version, we can remove
our implementation in favor of upstream.

Author: hawkw

Cargo.lock

@@ -665,7 +665,6 @@ dependencies = [
  "ipnet",
  "linkerd-addr",
  "linkerd-cache",
- "linkerd-concurrency-limit",
  "linkerd-conditional",
  "linkerd-detect",
  "linkerd-dns",
@@ -844,19 +843,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "linkerd-concurrency-limit"
-version = "0.1.0"
-dependencies = [
- "futures",
- "linkerd-stack",
- "pin-project",
- "tokio",
- "tokio-util",
- "tower",
- "tracing",
-]
-
 [[package]]
 name = "linkerd-conditional"
 version = "0.1.0"

Cargo.toml

@@ -15,7 +15,6 @@ members = [
     "linkerd/app/test",
     "linkerd/app",
     "linkerd/cache",
-    "linkerd/concurrency-limit",
     "linkerd/conditional",
     "linkerd/detect",
     "linkerd/transport-header",

linkerd/app/core/Cargo.toml

@@ -22,7 +22,6 @@ futures = { version = "0.3", default-features = false }
 ipnet = "2.3"
 linkerd-addr = { path = "../../addr" }
 linkerd-cache = { path = "../../cache" }
-linkerd-concurrency-limit = { path = "../../concurrency-limit" }
 linkerd-conditional = { path = "../../conditional" }
 linkerd-dns = { path = "../../dns" }
 linkerd-detect = { path = "../../detect" }
@@ -78,6 +77,7 @@ features = [
     "spawn-ready",
     "timeout",
     "util",
+    "limit",
 ]
 
 [target.'cfg(target_os = "linux")'.dependencies]

linkerd/app/core/src/svc.rs

@@ -2,7 +2,6 @@
 
 pub use crate::proxy::http;
 use crate::{cache, Error};
-pub use linkerd_concurrency_limit::ConcurrencyLimit;
 use linkerd_error::Recover;
 use linkerd_exp_backoff::{ExponentialBackoff, ExponentialBackoffStream};
 pub use linkerd_reconnect::NewReconnect;
@@ -21,7 +20,10 @@ use tower::{
     layer::util::{Identity, Stack as Pair},
     make::MakeService,
 };
-pub use tower::{layer::Layer, service_fn as mk, spawn_ready::SpawnReady, Service, ServiceExt};
+pub use tower::{
+    layer::Layer, limit::GlobalConcurrencyLimitLayer as ConcurrencyLimitLayer, service_fn as mk,
+    spawn_ready::SpawnReady, Service, ServiceExt,
+};
 
 #[derive(Copy, Clone, Debug)]
 pub struct AlwaysReconnect(ExponentialBackoff);

linkerd/app/inbound/src/http/mod.rs

@@ -75,7 +75,7 @@ impl<H> Inbound<H> {
                     // to `NewRouter`) and the concurrency limit need not be
                     // driven outside of the request path, so there's no need
                     // for SpawnReady
-                    .push(svc::ConcurrencyLimit::layer(max_in_flight_requests))
+                    .push(svc::ConcurrencyLimitLayer::new(max_in_flight_requests))
                     .push(svc::FailFast::layer("HTTP Server", dispatch_timeout))
                     .push(rt.metrics.http_errors.clone())
                     // Synthesizes responses for proxy errors.

linkerd/app/outbound/src/http/server.rs

@@ -47,7 +47,7 @@ impl<N> Outbound<N> {
                     // background task to ensure it becomes ready without new
                     // requests being processed.
                     .push(svc::layer::mk(svc::SpawnReady::new))
-                    .push(svc::ConcurrencyLimit::layer(max_in_flight_requests))
+                    .push(svc::ConcurrencyLimitLayer::new(max_in_flight_requests))
                     .push(svc::FailFast::layer("HTTP Server", dispatch_timeout))
                     .push_spawn_buffer(buffer_capacity)
                     .push(rt.metrics.http_errors.clone())

linkerd/app/outbound/src/ingress.rs

@@ -206,7 +206,7 @@ impl Outbound<svc::BoxNewHttp<http::Endpoint>> {
                     // The concurrency-limit can force the service into fail-fast, but it need not
                     // be driven to readiness on a background task (i.e., by `SpawnReady`).
                     // Otherwise, the inner service is always ready (because it's a router).
-                    .push(svc::ConcurrencyLimit::layer(max_in_flight_requests))
+                    .push(svc::ConcurrencyLimitLayer::new(max_in_flight_requests))
                     .push(svc::FailFast::layer("Ingress server", dispatch_timeout))
                     .push(rt.metrics.http_errors.clone())
                     .push(errors::layer())