Hedge

pub trait Policy<Request> {
    /// Called when the request is first received to determine if the request is retryable.
    fn clone_request(&self, req: &Request) -> Option<Request>;

    /// Called after the hedge timeout to determine if the hedge retry should be issued.
    fn can_retry(&self, req: &Request) -> bool;
}

type Histo = Arc<Mutex<RotatingHistogram>>;
type Service<S, P> = select::Select<
    SelectPolicy<P>,
    Latency<Histo, S>,
    Delay<DelayPolicy, AsyncFilter<Latency<Histo, S>, PolicyPredicate<P>>>,
>;

        // Clone the underlying service and wrap both copies in a middleware that
        // records the latencies in a rotating histogram.
        let recorded_a = Latency::new(histo.clone(), service.clone());
        let recorded_b = Latency::new(histo.clone(), service);

        // Check policy to see if the hedge request should be issued.
        let filtered = AsyncFilter::new(recorded_b, PolicyPredicate(policy.clone()));

        // Delay the second request by a percentile of the recorded request latency
        // histogram.
        let delay_policy = DelayPolicy {
            histo: histo.clone(),
            latency_percentile,
        };
        let delayed = Delay::new(delay_policy, filtered);

        // If the request is retryable, issue two requests -- the second one delayed
        // by a latency percentile.  Use the first result to complete.
        let select_policy = SelectPolicy {
            policy,
            histo,
            min_data_points,
        };
        Hedge(Select::new(select_policy, recorded_a, delayed))

    fn call(&mut self, request: Request) -> Self::Future {
        let b_fut = if let Some(cloned_req) = self.policy.clone_request(&request) {
            Some(self.b.call(cloned_req))
        } else {
            None
        };
        ResponseFuture {
            a_fut: self.a.call(request),
            b_fut,
        }
    }

    fn clone_request(&self, req: &Request) -> Option<Request> {
        self.policy.clone_request(req).filter(|_| {
            let mut locked = self.histo.lock().unwrap();
            // Do not attempt a retry if there are insufficiently many data
            // points in the histogram.
            locked.read().len() >= self.min_data_points
        })
    }

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::time::Instant;
use tower::hedge::{Hedge, Policy};
use tower::{BoxError, Service, ServiceExt, service_fn};

#[derive(Clone)]
struct Always;
impl Policy<u32> for Always {
  fn clone_request(&self, req: &u32) -> Option<u32> {
    Some(*req)
  }
  fn can_retry(&self, _: &u32) -> bool {
    true
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() -> Result<(), BoxError> {
  let t0 = Instant::now();
  let attempts = Arc::new(AtomicU32::new(0));
  // The first attempt succeeds after 500 ms; the hedged attempt fails at once.
  let inner = service_fn(move |req: u32| {
    let n = attempts.fetch_add(1, Ordering::SeqCst) + 1;
    async move {
      if n == 1 {
        tokio::time::sleep(Duration::from_millis(500)).await;
        Ok::<_, BoxError>(req)
      } else {
        Err::<u32, BoxError>("hedged attempt failed".into())
      }
    }
  });
  // Ten recorded latencies of 100 ms, min_data_points = 5, quantile 0.9.
  let mut hedge =
    Hedge::new_with_mock_latencies(inner, Always, 5, 0.9, Duration::from_secs(60), &[100; 10]);
  let res = hedge.ready().await?.call(7).await;
  println!("t={}ms result = {:?}", t0.elapsed().as_millis(), res.map_err(|e| e.to_string()));
  Ok(())
}

t=100ms result = Err("hedged attempt failed")

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Calling self.service.poll_ready would reserve a slot for the delayed request,
        // potentially well in advance of actually making it.  Instead, signal readiness here and
        // treat the service as a Oneshot in the future.
        Poll::Ready(Ok(()))
    }

                StateProj::Delaying { delay, req } => {
                    ready!(delay.poll(cx));
                    let req = req.take().expect("Missing request in delay");
                    let svc = this.service.take().expect("Missing service in delay");
                    let fut = Oneshot::new(svc, req);
                    this.state.set(State::called(fut));
                }

    fn check(&mut self, request: Request) -> Self::Future {
        if self.0.can_retry(&request) {
            Either::Left(future::ready(Ok(request)))
        } else {
            // If the hedge retry should not be issued, we simply want to wait
            // for the result of the original request.  Therefore we don't want
            // to return an error here.  Instead, we use future::pending to ensure
            // that the original request wins the select.
            Either::Right(future::pending())
        }
    }

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        let rsp = ready!(this.inner.poll(cx)).map_err(Into::into)?;
        let duration = Instant::now().saturating_duration_since(*this.start);
        this.rec.record(duration);
        Poll::Ready(Ok(rsp))
    }

pub struct RotatingHistogram {
    read: Histogram<u64>,
    write: Histogram<u64>,
    last_rotation: Instant,
    period: Duration,
}

    fn maybe_rotate(&mut self) {
        let delta = Instant::now().saturating_duration_since(self.last_rotation);
        // TODO: replace with delta.duration_div when it becomes stable.
        let rotations = (nanos(delta) / nanos(self.period)) as u32;
        if rotations >= 2 {
            trace!("Time since last rotation is {:?}.  clearing!", delta);
            self.clear();
        } else if rotations == 1 {
            trace!("Time since last rotation is {:?}. rotating!", delta);
            self.rotate();
        }
        self.last_rotation += self.period * rotations;
    }

    fn delay(&self, _req: &Request) -> Duration {
        let mut locked = self.histo.lock().unwrap();
        let millis = locked
            .read()
            .value_at_quantile(self.latency_percentile.into());
        Duration::from_millis(millis)
    }

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::time::Instant;
use tower::hedge::{Hedge, Policy};
use tower::{BoxError, Service, ServiceExt, service_fn};

#[derive(Clone)]
struct Always;
impl Policy<u32> for Always {
  fn clone_request(&self, req: &u32) -> Option<u32> {
    Some(*req)
  }
  fn can_retry(&self, _: &u32) -> bool {
    true
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() -> Result<(), BoxError> {
  let t0 = Instant::now();
  let calls = Arc::new(AtomicU32::new(0));
  let c = calls.clone();
  let inner = service_fn(move |req: u32| {
    c.fetch_add(1, Ordering::SeqCst);
    println!("  t={}ms inner call req={req}", t0.elapsed().as_millis());
    async move {
      tokio::time::sleep(Duration::from_millis(50)).await;
      Ok::<_, BoxError>(req)
    }
  });
  // min_data_points = 0: no history needed before hedging.
  let mut hedge = Hedge::new(inner, Always, 0, 0.9, Duration::from_secs(1));
  for req in 1..=3 {
    hedge.ready().await?.call(req).await?;
  }
  println!("3 requests -> {} inner calls", calls.load(Ordering::SeqCst));
  Ok(())
}

  t=0ms inner call req=1
  t=0ms inner call req=1
  t=50ms inner call req=2
  t=50ms inner call req=2
  t=100ms inner call req=3
  t=100ms inner call req=3
3 requests -> 6 inner calls

use std::collections::HashSet;
use std::convert::Infallible;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use futures::future::join_all;
use tokio::time::{Instant, sleep_until};
use tower::hedge::{Hedge, Policy};
use tower::{BoxError, Service, ServiceExt, service_fn};

// Request: (id, latency of the first attempt in ms). Any later attempt with
// the same id takes 10 ms.
type Req = (u32, u64);

#[derive(Clone)]
struct Always;
impl Policy<Req> for Always {
  fn clone_request(&self, req: &Req) -> Option<Req> {
    Some(*req)
  }
  fn can_retry(&self, _: &Req) -> bool {
    true
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() -> Result<(), BoxError> {
  let t0 = Instant::now();
  let ms = move || t0.elapsed().as_millis();
  let seen = Arc::new(Mutex::new(HashSet::new()));
  let inner = service_fn(move |(id, slow): Req| {
    let first = seen.lock().unwrap().insert(id);
    let lat = if first { slow } else { 10 };
    if id >= 100 {
      println!("  t={:>4}ms  inner call id={id} attempt={}", ms(), if first { 1 } else { 2 });
    }
    async move {
      tokio::time::sleep(Duration::from_millis(lat)).await;
      Ok::<_, Infallible>(lat)
    }
  });
  // min_data_points = 5, quantile 0.5, period = 1s
  let mut hedge = Hedge::new(inner, Always, 5, 0.5, Duration::from_secs(1));

  // Ten 100 ms requests at once, then the histogram's write half has 10 points.
  let mut futs = Vec::new();
  for id in 0..10 {
    futs.push(hedge.ready().await?.call((id, 100)));
  }
  join_all(futs).await;
  println!("t={:>4}ms  10 requests of 100ms done", ms());

  for (id, at) in [(100, 100), (101, 1200), (102, 3500)] {
    sleep_until(t0 + Duration::from_millis(at)).await;
    println!("t={:>4}ms  send id={id} (first attempt 500ms)", ms());
    let lat = hedge.ready().await?.call((id, 500)).await?;
    println!("t={:>4}ms  id={id} answered by the {lat}ms attempt", ms());
  }
  Ok(())
}

t= 100ms  10 requests of 100ms done
t= 100ms  send id=100 (first attempt 500ms)
  t= 100ms  inner call id=100 attempt=1
t= 600ms  id=100 answered by the 500ms attempt
t=1200ms  send id=101 (first attempt 500ms)
  t=1200ms  inner call id=101 attempt=1
  t=1300ms  inner call id=101 attempt=2
t=1310ms  id=101 answered by the 10ms attempt
t=3500ms  send id=102 (first attempt 500ms)
  t=3500ms  inner call id=102 attempt=1
t=4000ms  id=102 answered by the 500ms attempt