Connection deadlock under bidirectional bulk data: both peers stop reading while a PONG/PING/SETTINGS write is pending (poll_ready gates poll_next) #975

//! Bounded reduction of #6073: hyper HTTP/2 client and streaming echo server,
//! adaptive window on/off per side, many concurrent POST/echo streams. A
//! watchdog reports any in-flight stream with no progress for STALL_SECS.
//!
//! usage: adaptive_repro <client_adaptive 0|1> <server_adaptive 0|1> <body_bytes>
//!        <streams> <secs> <reps>
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::{Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use tokio::net::{TcpListener, TcpStream};

const STALL_SECS: u64 = 5;

struct Tls {
    acceptor: tokio_rustls::TlsAcceptor,
    connector: tokio_rustls::TlsConnector,
}

fn make_tls() -> Tls {
    let _ = rustls::crypto::ring::default_provider().install_default();
    let ck = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let cert = ck.cert.der().clone();
    let key = rustls::pki_types::PrivateKeyDer::Pkcs8(ck.signing_key.serialize_der().into());
    let mut server = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(vec![cert.clone()], key)
        .unwrap();
    server.alpn_protocols = vec![b"h2".to_vec()];
    let mut roots = rustls::RootCertStore::empty();
    roots.add(cert).unwrap();
    let mut client = rustls::ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth();
    client.alpn_protocols = vec![b"h2".to_vec()];
    Tls {
        acceptor: tokio_rustls::TlsAcceptor::from(Arc::new(server)),
        connector: tokio_rustls::TlsConnector::from(Arc::new(client)),
    }
}
const DRAIN_LIMIT: Duration = Duration::from_secs(30);

fn now_ms(start: Instant) -> u64 {
    start.elapsed().as_millis() as u64
}

struct StreamState {
    in_flight: AtomicUsize,
    last_progress_ms: AtomicU64,
    received: AtomicU64,
    /// 0 = request sent, waiting for response headers; 1 = headers received.
    phase: AtomicUsize,
}

async fn serve(listener: TcpListener, adaptive: bool, collect: bool, tls: Option<tokio_rustls::TlsAcceptor>) {
    loop {
        let Ok((tcp, _)) = listener.accept().await else { return };
        tcp.set_nodelay(true).ok();
        let tls = tls.clone();
        tokio::spawn(async move {
            let mut b = hyper::server::conn::http2::Builder::new(TokioExecutor::new());
            b.timer(TokioTimer::new());
            if adaptive {
                b.adaptive_window(true);
            } else {
                b.initial_stream_window_size(8 << 20).initial_connection_window_size(32 << 20);
            }
            b.max_frame_size(1 << 20);
            let svc = hyper::service::service_fn(move |req: Request<Incoming>| async move {
                let body: http_body_util::combinators::BoxBody<Bytes, hyper::Error> = if collect {
                    // Like the old proto_backend: read the whole request first.
                    let bytes = req.into_body().collect().await.map(|c| c.to_bytes()).unwrap_or_default();
                    Full::new(bytes).map_err(|never| match never {}).boxed()
                } else {
                    req.into_body().boxed()
                };
                Ok::<_, std::convert::Infallible>(Response::new(body))
            });
            match tls {
                Some(acceptor) => {
                    let Ok(stream) = acceptor.accept(tcp).await else { return };
                    let _ = b.serve_connection(TokioIo::new(stream), svc).await;
                }
                None => {
                    let _ = b.serve_connection(TokioIo::new(tcp), svc).await;
                }
            }
        });
    }
}

async fn one_rep(client_adaptive: bool, server_adaptive: bool, size: usize, streams: usize, secs: u64, collect: bool, per_conn: usize) -> (u64, usize, bool) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let tls = (std::env::var("TLS").as_deref() == Ok("1")).then(make_tls);
    let server = tokio::spawn(serve(listener, server_adaptive, collect, tls.as_ref().map(|t| t.acceptor.clone())));

    let num_conns = streams.div_ceil(per_conn).max(1);
    let mut senders = Vec::new();
    let mut drivers = Vec::new();
    for _ in 0..num_conns {
        let tcp = TcpStream::connect(addr).await.unwrap();
        tcp.set_nodelay(true).ok();
        let mut b = hyper::client::conn::http2::Builder::new(TokioExecutor::new());
        b.timer(TokioTimer::new());
        if client_adaptive {
            b.adaptive_window(true);
        } else {
            b.initial_stream_window_size(8 << 20).initial_connection_window_size(32 << 20);
        }
        b.max_frame_size(1 << 20);
        if let Ok(ms) = std::env::var("KEEPALIVE_MS") {
            b.keep_alive_interval(Duration::from_millis(ms.parse().unwrap()))
                .keep_alive_while_idle(true);
        }
        let sender = match &tls {
            Some(t) => {
                let name = rustls::pki_types::ServerName::try_from("localhost").unwrap();
                let stream = t.connector.connect(name, tcp).await.unwrap();
                let (sender, conn) = b.handshake::<_, Full<Bytes>>(TokioIo::new(stream)).await.unwrap();
                drivers.push(tokio::spawn(async move { let _ = conn.await; }));
                sender
            }
            None => {
                let (sender, conn) = b.handshake::<_, Full<Bytes>>(TokioIo::new(tcp)).await.unwrap();
                drivers.push(tokio::spawn(async move { let _ = conn.await; }));
                sender
            }
        };
        senders.push(sender);
    }

    let start = Instant::now();
    let deadline = start + Duration::from_secs(secs);
    let payload = Bytes::from(vec![7u8; size]);
    let completed = Arc::new(AtomicU64::new(0));
    let states: Vec<Arc<StreamState>> = (0..streams)
        .map(|_| Arc::new(StreamState { in_flight: AtomicUsize::new(0), last_progress_ms: AtomicU64::new(0), received: AtomicU64::new(0), phase: AtomicUsize::new(0) }))
        .collect();

    let mut workers = Vec::new();
    for (i, state) in states.iter().enumerate() {
        let mut sender = senders[i % num_conns].clone();
        let payload = payload.clone();
        let state = Arc::clone(state);
        let completed = Arc::clone(&completed);
        workers.push(tokio::spawn(async move {
            while Instant::now() < deadline {
                state.in_flight.store(1, Ordering::Relaxed);
                state.received.store(0, Ordering::Relaxed);
                state.phase.store(0, Ordering::Relaxed);
                state.last_progress_ms.store(now_ms(start), Ordering::Relaxed);
                let req = Request::post("/echo").body(Full::new(payload.clone())).unwrap();
                let Ok(resp) = sender.send_request(req).await else { break };
                state.phase.store(1, Ordering::Relaxed);
                state.last_progress_ms.store(now_ms(start), Ordering::Relaxed);
                let mut body = resp.into_body();
                while let Some(frame) = body.frame().await {
                    let Ok(frame) = frame else { break };
                    if let Some(d) = frame.data_ref() {
                        state.received.fetch_add(d.len() as u64, Ordering::Relaxed);
                        state.last_progress_ms.store(now_ms(start), Ordering::Relaxed);
                    }
                }
                state.in_flight.store(0, Ordering::Relaxed);
                completed.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }

    // Watchdog
    let wd_states = states.clone();
    let wd_senders = senders.clone();
    let wd_num_conns = num_conns;
    let wd_port = addr.port();
    let stalls = Arc::new(AtomicUsize::new(0));
    let wd_stalls = Arc::clone(&stalls);
    let watchdog = tokio::spawn(async move {
        let mut reported = vec![false; wd_states.len()];
        let mut probed = vec![false; wd_num_conns];
        loop {
            tokio::time::sleep(Duration::from_millis(250)).await;
            let t = now_ms(start);
            for (i, s) in wd_states.iter().enumerate() {
                let idle = t.saturating_sub(s.last_progress_ms.load(Ordering::Relaxed));
                if s.in_flight.load(Ordering::Relaxed) == 1 && idle >= STALL_SECS * 1000 {
                    if !reported[i] {
                        reported[i] = true;
                        wd_stalls.fetch_add(1, Ordering::Relaxed);
                        if !probed[i % wd_num_conns] {
                            probed[i % wd_num_conns] = true;
                            let mut probe = wd_senders[i % wd_num_conns].clone();
                            let conn = i % wd_num_conns;
                            tokio::spawn(async move {
                                // Headers-only exchange: not flow controlled.
                                let req = Request::post("/echo").body(Full::new(Bytes::new())).unwrap();
                                let t0 = Instant::now();
                                let outcome = tokio::time::timeout(Duration::from_secs(3), async {
                                    let resp = probe.send_request(req).await?;
                                    resp.into_body().collect().await.map(|_| ())
                                }).await;
                                if outcome.is_err() {
                                    if let Ok(out) = std::process::Command::new("netstat").args(["-an", "-p", "tcp"]).output() {
                                        let text = String::from_utf8_lossy(&out.stdout);
                                        eprintln!("  SOCKETS (Recv-Q Send-Q local foreign) for server port {wd_port}:");
                                        for line in text.lines().filter(|l| l.contains(&format!(".{wd_port} ")) && l.contains("ESTABLISHED")) {
                                            eprintln!("    {line}");
                                        }
                                    }
                                }
                                eprintln!("  PROBE conn#{conn}: {} after {}ms", match outcome { Ok(Ok(())) => "empty-body request COMPLETED (driver alive)".to_string(), Ok(Err(e)) => format!("error {e}"), Err(_) => "TIMED OUT (driver not progressing)".to_string() }, t0.elapsed().as_millis());
                            });
                        }
                        eprintln!("  STALL stream#{i} at t={t}ms idle={idle}ms phase={} received={}/{size}", if s.phase.load(Ordering::Relaxed) == 0 { "awaiting-headers" } else { "awaiting-body" }, s.received.load(Ordering::Relaxed));
                    }
                } else {
                    reported[i] = false;
                }
            }
        }
    });

    let drain_deadline = tokio::time::Instant::from_std(deadline + DRAIN_LIMIT);
    let mut hung = false;
    for mut w in workers {
        if tokio::time::timeout_at(drain_deadline, &mut w).await.is_err() {
            w.abort();
            hung = true;
        }
    }
    watchdog.abort();
    for d in drivers { d.abort(); }
    server.abort();
    (completed.load(Ordering::Relaxed), stalls.load(Ordering::Relaxed), hung)
}

#[tokio::main]
async fn main() {
    if let Ok(path) = std::env::var("TRACE_FILE") {
        let file = std::fs::File::create(path).unwrap();
        tracing_subscriber::fmt()
            .with_env_filter(tracing_subscriber::EnvFilter::new(
                std::env::var("TRACE_FILTER").unwrap_or_else(|_| "h2::codec=debug,h2::proto=trace,hyper::proto::h2=trace".to_string()),
            ))
            .with_writer(std::sync::Mutex::new(file))
            .with_ansi(false)
            .with_thread_ids(true)
            .init();
    }
    let a: Vec<String> = std::env::args().collect();
    let ca = a[1] == "1";
    let sa = a[2] == "1";
    let size: usize = a[3].parse().unwrap();
    let streams: usize = a[4].parse().unwrap();
    let secs: u64 = a[5].parse().unwrap();
    let reps: usize = a[6].parse().unwrap();
    let collect = a.get(7).map(|m| m == "collect").unwrap_or(false);
    let per_conn: usize = a.get(8).map(|v| v.parse().unwrap()).unwrap_or(usize::MAX);
    let (mut stalled_reps, mut hung_reps) = (0, 0);
    for rep in 0..reps {
        let t = Instant::now();
        let (done, stalls, hung) = one_rep(ca, sa, size, streams, secs, collect, per_conn).await;
        if stalls > 0 { stalled_reps += 1; }
        if hung { hung_reps += 1; }
        println!("rep={rep} client_adaptive={ca} server_adaptive={sa} size={size} streams={streams} completed={done} stalls={stalls} hung_drain={hung} wall={:.1}s", t.elapsed().as_secs_f64());
    }
    println!("SUMMARY collect={collect} per_conn={per_conn} client_adaptive={ca} server_adaptive={sa} size={size} streams={streams} reps={reps} stalled_reps={stalled_reps} hung_reps={hung_reps}");
}