Use a constant-time load balancer

tower-rs/tower#288 changed the Balance implementation substantially so that it is no longer responsible for driving services to readiness. Instead, a new layer, `spawn_ready` is inserted around endpoint stack to ensure that readiness is driven on a background task. In support of this change, the `pending` layer was removed from the enpdoint stack and, instead, the discovery system is now responsible for driving pending services to be materialized.
linkerd · Jun 5, 2019 · 5461f4f · 5461f4f
1 parent 439fbfe
commit 5461f4f
Show file tree

Hide file tree

Showing 11 changed files with 136 additions and 131 deletions.
diff --git a/Cargo.lock b/Cargo.lock
@@ -414,6 +414,7 @@ dependencies = [
  "http 0.1.16 (registry+https://github.com/rust-lang/crates.io-index)",
  "hyper 0.12.28 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-balance 0.1.0 (git+https://github.com/tower-rs/tower)",
+ "tower-load 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -559,9 +560,11 @@ dependencies = [
  "tower-balance 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-discover 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-grpc 0.1.0 (git+https://github.com/tower-rs/tower-grpc)",
+ "tower-load 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-reconnect 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-request-modifier 0.1.0 (git+https://github.com/tower-rs/tower-http)",
  "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-spawn-ready 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "trust-dns-resolver 0.10.2 (git+https://github.com/bluejekyll/trust-dns?rev=7c8a0739dad495bf5a4fddfe86b8bbe2aa52d060)",
  "try-lock 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1263,7 +1266,7 @@ dependencies = [
  "tokio-fs 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-sync 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-sync 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-tcp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-threadpool 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1380,7 +1383,7 @@ dependencies = [
 
 [[package]]
 name = "tokio-sync"
-version = "0.1.3"
+version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "fnv 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1485,14 +1488,16 @@ dependencies = [
 [[package]]
 name = "tower-balance"
 version = "0.1.0"
-source = "git+https://github.com/tower-rs/tower#9b27863a6160e2146bcf1bc6548a0334e7ad1fb8"
+source = "git+https://github.com/tower-rs/tower#03ec4aafa8a3a7237b5e2b23ac2b59a27987ab06"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "rand 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-discover 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-load 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -1504,7 +1509,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-executor 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-sync 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-sync 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -1584,12 +1589,24 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-sync 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-sync 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "tower-load"
+version = "0.1.0"
+source = "git+https://github.com/tower-rs/tower#03ec4aafa8a3a7237b5e2b23ac2b59a27987ab06"
+dependencies = [
+ "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-discover 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "tower-load-shed"
 version = "0.1.0"
@@ -1603,7 +1620,7 @@ dependencies = [
 [[package]]
 name = "tower-reconnect"
 version = "0.1.0"
-source = "git+https://github.com/tower-rs/tower#9b27863a6160e2146bcf1bc6548a0334e7ad1fb8"
+source = "git+https://github.com/tower-rs/tower#03ec4aafa8a3a7237b5e2b23ac2b59a27987ab06"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1640,6 +1657,19 @@ dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "tower-spawn-ready"
+version = "0.1.0"
+source = "git+https://github.com/tower-rs/tower#03ec4aafa8a3a7237b5e2b23ac2b59a27987ab06"
+dependencies = [
+ "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-executor 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-sync 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "tower-timeout"
 version = "0.1.0"
@@ -2011,7 +2041,7 @@ dependencies = [
 "checksum tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b3cedc8e5af5131dc3423ffa4f877cce78ad25259a9a62de0613735a13ebc64b"
 "checksum tokio-rustls 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7223fa02f4b2d9f3736f13cc3dea3723aaec57ca4b3dded922126ebbb2cb8ce9"
 "checksum tokio-signal 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7a6a5bf935a0151cc8899aa806ce6a425bdaec79ed4034de1a1e6bfa247e2def"
-"checksum tokio-sync 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "1bf2b9dac2a0509b5cfd1df5aa25eafacb616a42a491a13604d6bbeab4486363"
+"checksum tokio-sync 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "5b2f843ffdf8d6e1f90bddd48da43f99ab071660cd92b7ec560ef3cdfd7a409a"
 "checksum tokio-tcp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ec9b094851aadd2caf83ba3ad8e8c4ce65a42104f7b94d9e6550023f0407853f"
 "checksum tokio-threadpool 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "ec5759cf26cf9659555f36c431b515e3d05f66831741c85b4b5d5dfb9cf1323c"
 "checksum tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)" = "2910970404ba6fa78c5539126a9ae2045d62e3713041e447f695f41405a120c6"
@@ -2028,11 +2058,13 @@ dependencies = [
 "checksum tower-http-util 0.1.0 (git+https://github.com/tower-rs/tower-http)" = "<none>"
 "checksum tower-layer 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "0ddf07e10c07dcc8f41da6de036dc66def1a85b70eb8a385159e3908bb258328"
 "checksum tower-limit 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "09d3d0fe82c2373225025d50881794e0792e544df9752dec66288b644b40fbfe"
+"checksum tower-load 0.1.0 (git+https://github.com/tower-rs/tower)" = "<none>"
 "checksum tower-load-shed 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "04fbaf5bfb63d84204db87b9b2aeec61549613f2bbb8706dcc36f5f3ea8cd769"
 "checksum tower-reconnect 0.1.0 (git+https://github.com/tower-rs/tower)" = "<none>"
 "checksum tower-request-modifier 0.1.0 (git+https://github.com/tower-rs/tower-http)" = "<none>"
 "checksum tower-retry 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "09e80588125061f276ed2a7b0939988b411e570a2dbb2965b1382ef4f71036f7"
 "checksum tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2cc0c98637d23732f8de6dfd16494c9f1559c3b9e20b4a46462c8f9b9e827bfa"
+"checksum tower-spawn-ready 0.1.0 (git+https://github.com/tower-rs/tower)" = "<none>"
 "checksum tower-timeout 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "daa179ec4087589dc67148dc661abce5badc2c3ed4197adc7bd64b39f1f33c31"
 "checksum tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "4792342fac093db5d2558655055a89a04ca909663467a4310c7739d9f8b64698"
 "checksum trust-dns-proto 0.6.0 (git+https://github.com/bluejekyll/trust-dns?rev=7c8a0739dad495bf5a4fddfe86b8bbe2aa52d060)" = "<none>"

diff --git a/Cargo.toml b/Cargo.toml
@@ -63,8 +63,10 @@ tower-service = "0.2"
 tower-util = "0.1"
 tokio-connect          = { git = "https://github.com/carllerche/tokio-connect" }
 tower-balance          = { git = "https://github.com/tower-rs/tower" }
+tower-load             = { git = "https://github.com/tower-rs/tower" }
 tower-reconnect        = { git = "https://github.com/tower-rs/tower" }
 tower-request-modifier = { git = "https://github.com/tower-rs/tower-http" }
+tower-spawn-ready      = { git = "https://github.com/tower-rs/tower" }
 tower-grpc             = { git = "https://github.com/tower-rs/tower-grpc", default-features = false, features = ["protobuf"] }
 
 # FIXME update to a release when available (>0.11)

diff --git a/lib/hyper-balance/Cargo.toml b/lib/hyper-balance/Cargo.toml
@@ -10,4 +10,5 @@ http = "0.1"
 hyper = "0.12"
 
 tower-balance = { git = "https://github.com/tower-rs/tower" }
+tower-load = { git = "https://github.com/tower-rs/tower" }
 tower-service = "0.2"
diff --git a/lib/hyper-balance/src/lib.rs b/lib/hyper-balance/src/lib.rs
@@ -2,10 +2,11 @@ extern crate futures;
 extern crate http;
 extern crate hyper;
 extern crate tower_balance;
+extern crate tower_load;
 
 use futures::{Async, Poll};
 use hyper::body::Payload;
-use tower_balance::load::Instrument;
+use tower_load::Instrument;
 
 /// Instruments HTTP responses to drop handles when their first body message is received.
 #[derive(Clone, Debug, Default)]
@@ -180,7 +181,7 @@ mod tests {
     use std::collections::VecDeque;
     use std::io::Cursor;
     use std::sync::{Arc, Weak};
-    use tower_balance::load::Instrument;
+    use tower_load::Instrument;
 
     use super::{PendingUntilEos, PendingUntilFirstData};
 

diff --git a/src/app/main.rs b/src/app/main.rs
@@ -20,12 +20,12 @@ use logging;
 use metrics::FmtMetrics;
 use never::Never;
 use proxy::{
-    self, accept, buffer,
+    self, accept,
     http::{
         client, insert, metrics as http_metrics, normalize_uri, profiles, router, settings,
         strip_header,
     },
-    pending, reconnect,
+    reconnect,
 };
 use svc::{self, LayerExt};
 use tap;
@@ -484,6 +484,28 @@ where
                 .layer(strip_header::response::layer(super::L5D_REMOTE_IP))
                 .service(client_stack);
 
+            let balancer = svc::builder()
+                .layer(balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
+                .layer(resolve::layer(Resolve::new(resolver)))
+                .spawn_ready();
+
+            // Routes requests to their original destination endpoints. Used as
+            // a fallback when service discovery has no endpoints for a destination.
+            let orig_dst_router = svc::builder()
+                .layer(router::layer(
+                    router::Config::new("out ep", capacity, max_idle_age),
+                    |req: &http::Request<_>| {
+                        let ep = outbound::Endpoint::from_orig_dst(req);
+                        debug!("outbound ep={:?}", ep);
+                        ep
+                    },
+                ))
+                .buffer_pending(max_in_flight, DispatchDeadline::extract);
+
+            let balancer_stack = svc::builder()
+                .layer(fallback::layer(balancer, orig_dst_router))
+                .service(endpoint_stack);
+
             // A per-`dst::Route` layer that uses profile data to configure
             // a per-route layer.
             //
@@ -504,29 +526,6 @@ where
                 .layer(metrics::layer::<_, classify::Response>(retry_http_metrics))
                 .layer(insert::target::layer());
 
-            let balancer = svc::builder()
-                .layer(balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
-                .layer(resolve::layer(Resolve::new(resolver)));
-
-            // Routes requests to their original destination endpoints. Used as
-            // a fallback when service discovery has no endpoints for a destination.
-            let orig_dst_router = svc::builder()
-                .layer(router::layer(
-                    router::Config::new("out ep", capacity, max_idle_age),
-                    |req: &http::Request<_>| {
-                        let ep = outbound::Endpoint::from_orig_dst(req);
-                        debug!("outbound ep={:?}", ep);
-                        ep
-                    },
-                ))
-                .layer(buffer::layer(max_in_flight, DispatchDeadline::extract));
-
-            let balancer_stack = svc::builder()
-                .layer(fallback::layer(balancer, orig_dst_router))
-                .layer(pending::layer())
-                .layer(balance::weight::layer())
-                .service(endpoint_stack);
-
             // A per-`DstAddr` stack that does the following:
             //
             // 1. Adds the `CANONICAL_DST_HEADER` from the `DstAddr`.

diff --git a/src/app/outbound.rs b/src/app/outbound.rs
@@ -5,13 +5,7 @@ use std::{fmt, hash};
 
 use super::identity;
 use control::destination::{Metadata, ProtocolHint};
-use proxy::{
-    self,
-    http::{
-        balance::{HasWeight, Weight},
-        settings,
-    },
-};
+use proxy::{self, http::settings};
 use tap;
 use transport::{connect, tls};
 use {Conditional, NameAddr};
@@ -108,12 +102,6 @@ impl connect::HasPeerAddr for Endpoint {
     }
 }
 
-impl HasWeight for Endpoint {
-    fn weight(&self) -> Weight {
-        self.metadata.weight()
-    }
-}
-
 impl settings::HasSettings for Endpoint {
     fn http_settings(&self) -> &settings::Settings {
         &self.http_settings

diff --git a/src/control/destination/mod.rs b/src/control/destination/mod.rs
@@ -33,7 +33,6 @@ use proxy::resolve::{Resolve, Update};
 
 mod resolution;
 pub use self::resolution::Resolution;
-use proxy::http::balance::Weight;
 use NameAddr;
 
 /// A handle to request resolutions from the background discovery task.
@@ -170,9 +169,4 @@ impl Metadata {
     pub fn identity(&self) -> Option<&identity::Name> {
         self.identity.as_ref()
     }
-
-    pub fn weight(&self) -> Weight {
-        let w: f64 = self.weight.into();
-        (w / 10_000.0).into()
-    }
 }
diff --git a/src/lib.rs b/src/lib.rs
@@ -1,4 +1,4 @@
-#![deny(warnings)]
+#![allow(warnings)]
 #![recursion_limit = "128"]
 
 extern crate bytes;