fast 请求等了 2000ms：Steer 的队头阻塞

tower = { version = "=0.5.3", features = ["steer", "buffer", "limit", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }

use std::time::Duration;
use tokio::time::{sleep, Instant};
use tower::buffer::Buffer;
use tower::steer::Steer;
use tower::util::BoxCloneSyncService;
use tower::{service_fn, BoxError, Service, ServiceBuilder, ServiceExt};

type Svc = BoxCloneSyncService<String, String, BoxError>;

// "slow" handles one request at a time and takes 1s; "fast" answers at once.
fn backends() -> Vec<Svc> {
  let slow = ServiceBuilder::new()
    .concurrency_limit(1)
    .service_fn(|r: String| async move {
      sleep(Duration::from_secs(1)).await;
      Ok::<_, BoxError>(format!("slow handled {r}"))
    });
  let fast = service_fn(|r: String| async move { Ok::<_, BoxError>(format!("fast handled {r}")) });
  vec![BoxCloneSyncService::new(slow), BoxCloneSyncService::new(fast)]
}

fn pick<S>(r: &String, _: &[S]) -> usize {
  if r.starts_with("slow") { 0 } else { 1 }
}

async fn run<S>(name: &str, steer: S)
where
  S: Service<String, Response = String, Error = BoxError> + Clone + Send + 'static,
  S::Future: Send,
{
  println!("{name}");
  let start = Instant::now();
  let mut tasks = Vec::new();
  for r in ["slow-1", "slow-2", "fast-1"] {
    let mut steer = steer.clone();
    tasks.push(tokio::spawn(async move {
      let res = steer.ready().await?.call(r.to_string()).await?;
      Ok::<_, BoxError>(format!("  {res} at {}ms", start.elapsed().as_millis()))
    }));
  }
  for t in tasks {
    println!("{}", t.await.unwrap().unwrap());
  }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  run("Steer over bare backends", Steer::new(backends(), pick)).await;

  let buffered: Vec<_> = backends().into_iter().map(|s| Buffer::new(s, 64)).collect();
  run("Steer over Buffer(64) backends", Steer::new(buffered, pick)).await;
}

Steer over bare backends
  slow handled slow-1 at 1000ms
  slow handled slow-2 at 2000ms
  fast handled fast-1 at 2000ms
Steer over Buffer(64) backends
  slow handled slow-1 at 1000ms
  slow handled slow-2 at 2000ms
  fast handled fast-1 at 0ms

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        loop {
            // must wait for *all* services to be ready.
            // this will cause head-of-line blocking unless the underlying services are always ready.
            if self.not_ready.is_empty() {
                return Poll::Ready(Ok(()));
            } else {
                if self.services[self.not_ready[0]]
                    .poll_ready(cx)?
                    .is_pending()
                {
                    return Poll::Pending;
                }

                self.not_ready.pop_front();
            }
        }
    }

    fn call(&mut self, req: Req) -> Self::Future {
        assert!(
            self.not_ready.is_empty(),
            "Steer must wait for all services to be ready. Did you forget to call poll_ready()?"
        );

        let idx = self.router.pick(&req, &self.services[..]);
        let cl = &mut self.services[idx];
        self.not_ready.push_back(idx);
        cl.call(req)
    }

slow: Buffer(64), fast: Buffer(1)
  slow-1 ready at 0ms
  slow-2 ready at 1000ms
  slow handled slow-1 at 1000ms
  fast-1 ready at 2000ms
  slow handled slow-2 at 2000ms
  fast handled fast-1 at 2000ms