balance: Specialize the balancer for P2C

tower-balance/Cargo.toml

@@ -29,6 +29,7 @@ log = "0.4.1"
 rand = "0.6.5"
 tokio-timer = "0.2.4"
 tower-discover = "0.1.0"
+tower-layer = "0.1.0"
 tower-load = { version = "0.1.0", path = "../tower-load" }
 tower-service = "0.2.0"
 tower-util = "0.1.0"
@@ -43,3 +44,4 @@ tokio-executor = "0.1.2"
 tower = { version = "0.1", path = "../tower" }
 tower-buffer = { version = "0.1", path = "../tower-buffer" }
 tower-limit = { version = "0.1", path = "../tower-limit" }
+tower-test = { version = "0.1.0", path = "../tower-test" }

tower-balance/examples/demo.rs

@@ -14,7 +14,7 @@ use tower::{
 use tower_balance as lb;
 use tower_load as load;
 
-const REQUESTS: usize = 50_000;
+const REQUESTS: usize = 100_000;
 const CONCURRENCY: usize = 500;
 const DEFAULT_RTT: Duration = Duration::from_millis(30);
 static ENDPOINT_CAPACITY: usize = CONCURRENCY;
@@ -55,24 +55,22 @@ fn main() {
     let fut = future::lazy(move || {
         let decay = Duration::from_secs(10);
         let d = gen_disco();
-        let pe = lb::Balance::p2c(load::PeakEwmaDiscover::new(
+        let pe = lb::p2c::Balance::from_entropy(load::PeakEwmaDiscover::new(
             d,
             DEFAULT_RTT,
             decay,
             load::NoInstrument,
         ));
-        run("P2C+PeakEWMA", pe)
+        run("P2C+PeakEWMA...", pe)
     });
 
     let fut = fut.then(move |_| {
         let d = gen_disco();
-        let ll = lb::Balance::p2c(load::PendingRequestsDiscover::new(d, load::NoInstrument));
-        run("P2C+LeastLoaded", ll)
-    });
-
-    let fut = fut.and_then(move |_| {
-        let rr = lb::Balance::round_robin(gen_disco());
-        run("RoundRobin", rr)
+        let ll = lb::p2c::Balance::from_entropy(load::PendingRequestsDiscover::new(
+            d,
+            load::NoInstrument,
+        ));
+        run("P2C+LeastLoaded...", ll)
     });
 
     rt.spawn(fut);
@@ -133,14 +131,14 @@ fn gen_disco() -> impl Discover<
     )
 }
 
-fn run<D, C>(name: &'static str, lb: lb::Balance<D, C>) -> impl Future<Item = (), Error = ()>
+fn run<D>(name: &'static str, lb: lb::p2c::Balance<D>) -> impl Future<Item = (), Error = ()>
 where
     D: Discover + Send + 'static,
     D::Error: Into<Error>,
     D::Key: Send,
-    D::Service: Service<Req, Response = Rsp, Error = Error> + Send,
+    D::Service: Service<Req, Response = Rsp, Error = Error> + load::Load + Send,
     <D::Service as Service<Req>>::Future: Send,
-    C: lb::Choose<D::Key, D::Service> + Send + 'static,
+    <D::Service as load::Load>::Metric: std::fmt::Debug,
 {
     println!("{}", name);
 

tower-balance/src/error.rs

@@ -1,17 +1,20 @@
+//! Error types
+
 use std::fmt;
 
 pub(crate) type Error = Box<dyn std::error::Error + Send + Sync>;
 
+/// An error returned when the balancer's endpoint discovery stream fails.
 #[derive(Debug)]
-pub struct Balance(pub(crate) Error);
+pub struct Discover(pub(crate) Error);
 
-impl fmt::Display for Balance {
+impl fmt::Display for Discover {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "load balancing discover error: {}", self.0)
+        write!(f, "load balancer discovery error: {}", self.0)
     }
 }
 
-impl std::error::Error for Balance {
+impl std::error::Error for Discover {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         Some(&*self.0)
     }