Extract tower-load from tower-balance (#285)

The tower-balance crate includes the `Load` and `Instrument` traits,
which are likely useful outside of balancers; and certainly have no
tight coupling with any specific balancer implementation. This change
extracts these protocol-agnostic traits into a dedicated crate.

The `Load` trait includes a latency-aware _PeakEWMA_ load strategy as
well as a simple _PendingRequests_ strategy for latency-agnostic
applications.

The `Instrument` trait is used by both of these strategies to track
in-flight requests without knowing protocol details. It is expected that
protocol-specific crates will provide, for instance, HTTP
time-to-first-byte latency strategies.

A default `NoInstrument` implementation tracks the a request until its
response future is satisfied.

This crate should only be published once tower-balance is published.

Part of #286

Cargo.toml

@@ -9,6 +9,7 @@ members = [
   "tower-hedge",
   "tower-layer",
   "tower-limit",
+  "tower-load",
   "tower-load-shed",
   "tower-reconnect",
   "tower-retry",

azure-pipelines.yml

@@ -20,6 +20,7 @@ jobs:
       - tower-hedge
       - tower-layer
       - tower-limit
+      - tower-load
       - tower-load-shed
       - tower-reconnect
       - tower-retry

tower-balance/Cargo.toml

@@ -28,8 +28,9 @@ indexmap = "1.0.2"
 log = "0.4.1"
 rand = "0.6.5"
 tokio-timer = "0.2.4"
-tower-service = "0.2.0"
 tower-discover = "0.1.0"
+tower-load = { version = "0.1.0", path = "../tower-load" }
+tower-service = "0.2.0"
 tower-util = "0.1.0"
 
 [dev-dependencies]

tower-balance/examples/demo.rs

@@ -5,14 +5,14 @@ use futures::{future, stream, Async, Future, Poll, Stream};
 use hdrsample::Histogram;
 use rand::{self, Rng};
 use std::time::{Duration, Instant};
-use std::{cmp, hash};
 use tokio::{runtime, timer};
 use tower::{
     discover::{Change, Discover},
     limit::concurrency::ConcurrencyLimit,
     Service, ServiceExt,
 };
 use tower_balance as lb;
+use tower_load as load;
 
 const REQUESTS: usize = 50_000;
 const CONCURRENCY: usize = 500;
@@ -30,7 +30,6 @@ static MAX_ENDPOINT_LATENCIES: [Duration; 10] = [
     Duration::from_millis(500),
     Duration::from_millis(1000),
 ];
-static WEIGHTS: [f64; 10] = [1.0, 1.0, 1.0, 0.5, 1.5, 0.5, 1.5, 1.0, 1.0, 1.0];
 
 struct Summary {
     latencies: Histogram<u64>,
@@ -50,57 +49,24 @@ fn main() {
         print!("{}ms, ", l);
     }
     println!("]");
-    print!("WEIGHTS=[");
-    for w in &WEIGHTS {
-        print!("{}, ", w);
-    }
-    println!("]");
 
     let mut rt = runtime::Runtime::new().unwrap();
 
-    // Show weighted behavior first...
-
     let fut = future::lazy(move || {
         let decay = Duration::from_secs(10);
         let d = gen_disco();
-        let pe = lb::Balance::p2c(lb::WithWeighted::from(lb::load::WithPeakEwma::new(
+        let pe = lb::Balance::p2c(load::PeakEwmaDiscover::new(
             d,
             DEFAULT_RTT,
             decay,
-            lb::load::NoInstrument,
-        )));
-        run("P2C+PeakEWMA w/ weights", pe)
-    });
-
-    let fut = fut.then(move |_| {
-        let d = gen_disco();
-        let ll = lb::Balance::p2c(lb::WithWeighted::from(lb::load::WithPendingRequests::new(
-            d,
-            lb::load::NoInstrument,
-        )));
-        run("P2C+LeastLoaded w/ weights", ll)
-    });
-
-    // Then run through standard comparisons...
-
-    let fut = fut.then(move |_| {
-        let decay = Duration::from_secs(10);
-        let d = gen_disco();
-        let pe = lb::Balance::p2c(lb::load::WithPeakEwma::new(
-            d,
-            DEFAULT_RTT,
-            decay,
-            lb::load::NoInstrument,
+            load::NoInstrument,
         ));
         run("P2C+PeakEWMA", pe)
     });
 
     let fut = fut.then(move |_| {
         let d = gen_disco();
-        let ll = lb::Balance::p2c(lb::load::WithPendingRequests::new(
-            d,
-            lb::load::NoInstrument,
-        ));
+        let ll = lb::Balance::p2c(load::PendingRequestsDiscover::new(d, load::NoInstrument));
         run("P2C+LeastLoaded", ll)
     });
 
@@ -115,26 +81,7 @@ fn main() {
 
 type Error = Box<dyn std::error::Error + Send + Sync>;
 
-#[derive(Clone, Debug, PartialEq)]
-struct Key {
-    instance: usize,
-    weight: lb::Weight,
-}
-
-impl cmp::Eq for Key {}
-
-impl hash::Hash for Key {
-    fn hash<H: hash::Hasher>(&self, state: &mut H) {
-        self.instance.hash(state);
-        // Ignore weight.
-    }
-}
-
-impl lb::HasWeight for Key {
-    fn weight(&self) -> lb::Weight {
-        self.weight
-    }
-}
+type Key = usize;
 
 struct Disco<S>(Vec<(Key, S)>);
 
@@ -163,14 +110,8 @@ fn gen_disco() -> impl Discover<
     Disco(
         MAX_ENDPOINT_LATENCIES
             .iter()
-            .zip(WEIGHTS.iter())
             .enumerate()
-            .map(|(instance, (latency, weight))| {
-                let key = Key {
-                    instance,
-                    weight: (*weight).into(),
-                };
-
+            .map(|(instance, latency)| {
                 let svc = tower::service_fn(move |_| {
                     let start = Instant::now();
 
@@ -186,7 +127,7 @@ fn gen_disco() -> impl Discover<
                         })
                 });
 
-                (key, ConcurrencyLimit::new(svc, ENDPOINT_CAPACITY))
+                (instance, ConcurrencyLimit::new(svc, ENDPOINT_CAPACITY))
             })
             .collect(),
     )

tower-balance/src/lib.rs

@@ -10,25 +10,18 @@ use log::{debug, trace};
 use rand::{rngs::SmallRng, SeedableRng};
 use std::fmt;
 use tower_discover::Discover;
+use tower_load::Load;
 use tower_service::Service;
 
 pub mod choose;
 pub mod error;
 pub mod future;
-pub mod load;
 pub mod pool;
 
 #[cfg(test)]
 mod test;
 
-pub use self::{
-    choose::Choose,
-    load::{
-        weight::{HasWeight, Weight, Weighted, WithWeighted},
-        Load,
-    },
-    pool::Pool,
-};
+pub use self::{choose::Choose, pool::Pool};
 
 use self::{error::Error, future::ResponseFuture};
 

tower-load/CHANGELOG.md

@@ -0,0 +1,3 @@
+# 0.1.0 (unreleased)
+
+- Initial release