diff --git a/crates/pulsing-actor/Cargo.toml b/crates/pulsing-actor/Cargo.toml index a6f866108..d54097160 100644 --- a/crates/pulsing-actor/Cargo.toml +++ b/crates/pulsing-actor/Cargo.toml @@ -100,6 +100,10 @@ path = "../../examples/rust/behavior_counter.rs" name = "behavior_fsm" path = "../../examples/rust/behavior_fsm.rs" +[[example]] +name = "http2_pool_expand" +path = "../../examples/rust/http2_pool_expand.rs" + [[bench]] name = "tracing_overhead" diff --git a/crates/pulsing-actor/src/transport/http2/config.rs b/crates/pulsing-actor/src/transport/http2/config.rs index f1ab6f9a8..efb8eda1d 100644 --- a/crates/pulsing-actor/src/transport/http2/config.rs +++ b/crates/pulsing-actor/src/transport/http2/config.rs @@ -1,10 +1,11 @@ //! HTTP/2 transport configuration. -use crate::error::Result; use std::time::Duration; #[cfg(feature = "tls")] use super::tls::TlsConfig; +#[cfg(feature = "tls")] +use crate::error::Result; /// HTTP/2 transport configuration. #[derive(Debug, Clone)] @@ -17,6 +18,8 @@ pub struct Http2Config { pub connect_timeout: Duration, pub request_timeout: Duration, pub stream_timeout: Duration, + /// Initial per-host connection cap. The pool grows 2x on demand; this is + /// the starting point, not a hard ceiling. pub max_connections_per_host: usize, pub keepalive_interval: Option, pub keepalive_timeout: Duration, @@ -41,7 +44,7 @@ impl Default for Http2Config { connect_timeout: Duration::from_secs(5), request_timeout: Duration::from_secs(30), stream_timeout: Duration::from_secs(300), - max_connections_per_host: 10, + max_connections_per_host: 8, keepalive_interval: Some(Duration::from_secs(30)), keepalive_timeout: Duration::from_secs(10), http2_prior_knowledge: true, diff --git a/crates/pulsing-actor/src/transport/http2/pool.rs b/crates/pulsing-actor/src/transport/http2/pool.rs index 689e93d85..a01f80716 100644 --- a/crates/pulsing-actor/src/transport/http2/pool.rs +++ b/crates/pulsing-actor/src/transport/http2/pool.rs @@ -12,7 +12,7 @@ use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::net::TcpStream; -use tokio::sync::{Mutex, RwLock, Semaphore}; +use tokio::sync::{Mutex, OwnedSemaphorePermit, RwLock, Semaphore}; /// Connection pool statistics. #[derive(Debug, Default)] @@ -21,6 +21,7 @@ pub struct PoolStats { pub connections_closed: AtomicU64, pub connections_reused: AtomicU64, pub connection_errors: AtomicU64, + pub pool_expansions: AtomicU64, pub active_connections: AtomicUsize, pub idle_connections: AtomicUsize, } @@ -32,6 +33,7 @@ impl PoolStats { "connections_closed": self.connections_closed.load(Ordering::Relaxed), "connections_reused": self.connections_reused.load(Ordering::Relaxed), "connection_errors": self.connection_errors.load(Ordering::Relaxed), + "pool_expansions": self.pool_expansions.load(Ordering::Relaxed), "active_connections": self.active_connections.load(Ordering::Relaxed), "idle_connections": self.idle_connections.load(Ordering::Relaxed), }) @@ -54,10 +56,12 @@ pub struct PooledConnection { pub last_used: Instant, pub request_count: u64, pub state: ConnectionState, + /// Released on Drop; idle-reuse via `try_get_existing` does NOT consume new permits. + _permit: OwnedSemaphorePermit, } impl PooledConnection { - fn new(sender: http2::SendRequest>) -> Self { + fn new(sender: http2::SendRequest>, permit: OwnedSemaphorePermit) -> Self { let now = Instant::now(); Self { sender, @@ -65,6 +69,7 @@ impl PooledConnection { last_used: now, request_count: 0, state: ConnectionState::Idle, + _permit: permit, } } @@ -130,7 +135,7 @@ pub struct PoolConfig { impl Default for PoolConfig { fn default() -> Self { Self { - max_connections_per_host: 10, + max_connections_per_host: 8, min_idle_per_host: 1, max_total_connections: 100, connect_timeout: Duration::from_secs(5), @@ -159,6 +164,8 @@ struct HostPool { connections: Vec>>, /// Semaphore for limiting concurrent connections semaphore: Arc, + /// Current per-host live connection cap + current_cap: usize, /// Last connection error time last_error: Option, /// Consecutive error count @@ -166,10 +173,13 @@ struct HostPool { } impl HostPool { - fn new(max_connections: usize) -> Self { + fn new(initial_cap: usize) -> Self { + let cap = initial_cap.clamp(1, Semaphore::MAX_PERMITS); + Self { - connections: Vec::with_capacity(max_connections), - semaphore: Arc::new(Semaphore::new(max_connections)), + connections: Vec::with_capacity(cap), + semaphore: Arc::new(Semaphore::new(cap)), + current_cap: cap, last_error: None, error_count: 0, } @@ -184,6 +194,22 @@ impl HostPool { self.error_count = 0; } + /// Double the host pool capacity up to Tokio's semaphore limit. + fn expand(&mut self) -> Option<(usize, usize)> { + let old = self.current_cap; + let target = old.saturating_mul(2).min(Semaphore::MAX_PERMITS); + let added = target.saturating_sub(old); + + if added == 0 { + return None; + } + + self.semaphore.add_permits(added); + self.current_cap = target; + + Some((old, target)) + } + /// Check if we should back off from connecting to this host fn should_backoff(&self) -> bool { if let Some(last_error) = self.last_error { @@ -304,13 +330,14 @@ impl ConnectionPool { host_pool.semaphore.clone() }; - // Acquire permit (limits concurrent connections per host) - let _permit = semaphore - .try_acquire() - .map_err(|_| anyhow::anyhow!("Connection pool exhausted for {}", addr))?; + // Acquire permit (limits live connections per host) + let permit = match semaphore.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => self.expand_and_acquire(addr, &semaphore).await?, + }; // Create the connection - let result = self.create_connection_inner(addr).await; + let result = self.create_connection_inner(addr, permit).await; // Update host pool state { @@ -333,7 +360,8 @@ impl ConnectionPool { { let mut pools = self.pools.write().await; if let Some(host_pool) = pools.get_mut(&addr) { - if host_pool.connections.len() < self.config.max_connections_per_host { + // invariant: permit was acquired above, so cap always has room + if host_pool.connections.len() < host_pool.current_cap { host_pool.connections.push(conn.clone()); } } @@ -352,8 +380,62 @@ impl ConnectionPool { }) } + async fn expand_and_acquire( + &self, + addr: SocketAddr, + semaphore: &Arc, + ) -> anyhow::Result { + // `did_expand_here` is true only when *this* task performed the expand; + // a concurrent task may have expanded already, in which case we just + // skip and wait on the existing permits. + let did_expand_here = { + let mut pools = self.pools.write().await; + let host_pool = pools + .get_mut(&addr) + .ok_or_else(|| anyhow::anyhow!("Host pool for {} disappeared", addr))?; + + if host_pool.semaphore.available_permits() == 0 { + match host_pool.expand() { + Some((old, new)) => { + tracing::warn!( + addr = %addr, + old_cap = old, + new_cap = new, + "HTTP/2 connection pool exhausted, expanding capacity 2x" + ); + true + } + None => { + return Err(anyhow::anyhow!( + "Connection pool for {} reached Semaphore::MAX_PERMITS", + addr + )); + } + } + } else { + false + } + }; + + if did_expand_here { + self.stats.pool_expansions.fetch_add(1, Ordering::Relaxed); + } + + tokio::time::timeout( + self.http2_config.connect_timeout, + semaphore.clone().acquire_owned(), + ) + .await + .map_err(|_| anyhow::anyhow!("Timed out waiting for connection pool permit on {}", addr))? + .map_err(|_| anyhow::anyhow!("Semaphore closed for {}", addr)) + } + /// Create the actual TCP + HTTP/2 connection - async fn create_connection_inner(&self, addr: SocketAddr) -> anyhow::Result { + async fn create_connection_inner( + &self, + addr: SocketAddr, + permit: OwnedSemaphorePermit, + ) -> anyhow::Result { let stream = tokio::time::timeout(self.http2_config.connect_timeout, TcpStream::connect(addr)) .await @@ -384,7 +466,7 @@ impl ConnectionPool { } }); - let mut pooled = PooledConnection::new(sender); + let mut pooled = PooledConnection::new(sender, permit); pooled.mark_used(); return Ok(pooled); } @@ -402,7 +484,7 @@ impl ConnectionPool { } }); - let mut pooled = PooledConnection::new(sender); + let mut pooled = PooledConnection::new(sender, permit); pooled.mark_used(); Ok(pooled) @@ -527,11 +609,50 @@ impl Drop for ConnectionGuard { #[cfg(test)] mod tests { use super::*; + use hyper::service::service_fn; + use hyper::{Request, Response}; + use std::convert::Infallible; + use std::future::Future; + use std::task::{Context, Poll}; + use tokio::net::TcpListener; + + async fn spawn_h2c_server() -> SocketAddr { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let addr = listener.local_addr().unwrap(); + + tokio::spawn(async move { + loop { + let (stream, peer_addr) = match listener.accept().await { + Ok(accepted) => accepted, + Err(_) => break, + }; + + tokio::spawn(async move { + let io = TokioIo::new(stream); + let service = service_fn(|_req: Request| async { + Ok::<_, Infallible>(Response::new(Full::new(Bytes::new()))) + }); + let builder = hyper::server::conn::http2::Builder::new(TokioExecutor::new()); + let conn = builder.serve_connection(io, service); + + if let Err(e) = conn.await { + tracing::debug!( + peer = %peer_addr, + error = %e, + "Test HTTP/2 connection closed" + ); + } + }); + } + }); + + addr + } #[test] fn test_pool_config_default() { let config = PoolConfig::default(); - assert_eq!(config.max_connections_per_host, 10); + assert_eq!(config.max_connections_per_host, 8); assert_eq!(config.min_idle_per_host, 1); assert!(config.max_connection_age.is_some()); } @@ -541,10 +662,12 @@ mod tests { let stats = PoolStats::default(); stats.connections_created.fetch_add(1, Ordering::Relaxed); stats.connections_reused.fetch_add(5, Ordering::Relaxed); + stats.pool_expansions.fetch_add(2, Ordering::Relaxed); let json = stats.to_json(); assert_eq!(json["connections_created"], 1); assert_eq!(json["connections_reused"], 5); + assert_eq!(json["pool_expansions"], 2); } #[test] @@ -563,10 +686,192 @@ mod tests { assert_eq!(pool.error_count, 0); } + #[test] + fn test_host_pool_expand_doubles_capacity() { + let mut pool = HostPool::new(2); + let sem = pool.semaphore.clone(); + + let _p1 = sem.clone().try_acquire_owned().unwrap(); + let _p2 = sem.clone().try_acquire_owned().unwrap(); + assert!(sem.clone().try_acquire_owned().is_err()); + + let (old, new) = pool.expand().unwrap(); + assert_eq!((old, new), (2, 4)); + assert_eq!(pool.current_cap, 4); + assert_eq!(sem.available_permits(), 2); + + let _p3 = sem.clone().try_acquire_owned().unwrap(); + let _p4 = sem.clone().try_acquire_owned().unwrap(); + assert!(sem.clone().try_acquire_owned().is_err()); + } + + #[test] + fn test_host_pool_expand_respects_max_permits() { + let mut pool = HostPool::new(1); + pool.current_cap = Semaphore::MAX_PERMITS; + + assert!(pool.expand().is_none()); + } + + #[tokio::test] + async fn test_expand_and_acquire_times_out_when_expanded_permit_is_taken() { + let addr: SocketAddr = "127.0.0.1:12345".parse().unwrap(); + let http2_config = Http2Config::default().connect_timeout(Duration::from_millis(10)); + let pool = ConnectionPool::new(http2_config); + + let semaphore = { + let mut pools = pool.pools.write().await; + let host_pool = pools.entry(addr).or_insert_with(|| HostPool::new(1)); + host_pool.semaphore.clone() + }; + + let _held = semaphore.clone().try_acquire_owned().unwrap(); + let mut queued = Box::pin(semaphore.clone().acquire_owned()); + let waker = futures::task::noop_waker_ref(); + let mut cx = Context::from_waker(waker); + assert!(matches!(queued.as_mut().poll(&mut cx), Poll::Pending)); + + let err = pool.expand_and_acquire(addr, &semaphore).await.unwrap_err(); + assert!( + err.to_string() + .contains("Timed out waiting for connection pool permit"), + "unexpected error: {err}" + ); + assert_eq!(pool.stats.pool_expansions.load(Ordering::Relaxed), 1); + + let queued_permit = queued.await.unwrap(); + drop(queued_permit); + } + + #[tokio::test] + async fn test_expand_and_acquire_uses_existing_permit_without_expanding() { + let addr: SocketAddr = "127.0.0.1:12346".parse().unwrap(); + let pool = ConnectionPool::new(Http2Config::default()); + + let semaphore = { + let mut pools = pool.pools.write().await; + let host_pool = pools.entry(addr).or_insert_with(|| HostPool::new(1)); + host_pool.semaphore.clone() + }; + + let permit = pool.expand_and_acquire(addr, &semaphore).await.unwrap(); + assert_eq!(pool.stats.pool_expansions.load(Ordering::Relaxed), 0); + assert_eq!(semaphore.available_permits(), 0); + + drop(permit); + assert_eq!(semaphore.available_permits(), 1); + } + + #[tokio::test] + async fn test_expand_and_acquire_errors_when_host_pool_disappears() { + let addr: SocketAddr = "127.0.0.1:12347".parse().unwrap(); + let pool = ConnectionPool::new(Http2Config::default()); + let semaphore = Arc::new(Semaphore::new(1)); + + let err = pool.expand_and_acquire(addr, &semaphore).await.unwrap_err(); + assert!( + err.to_string() + .contains("Host pool for 127.0.0.1:12347 disappeared"), + "unexpected error: {err}" + ); + } + + #[tokio::test] + async fn test_expand_and_acquire_errors_at_semaphore_max_permits() { + let addr: SocketAddr = "127.0.0.1:12348".parse().unwrap(); + let pool = ConnectionPool::new(Http2Config::default()); + + let (semaphore, _held) = { + let mut pools = pool.pools.write().await; + let host_pool = pools.entry(addr).or_insert_with(|| HostPool::new(1)); + let semaphore = host_pool.semaphore.clone(); + let held = semaphore.clone().try_acquire_owned().unwrap(); + host_pool.current_cap = Semaphore::MAX_PERMITS; + (semaphore, held) + }; + + let err = pool.expand_and_acquire(addr, &semaphore).await.unwrap_err(); + assert!( + err.to_string().contains("reached Semaphore::MAX_PERMITS"), + "unexpected error: {err}" + ); + assert_eq!(pool.stats.pool_expansions.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn test_expand_and_acquire_errors_when_semaphore_is_closed() { + let addr: SocketAddr = "127.0.0.1:12349".parse().unwrap(); + let pool = ConnectionPool::new(Http2Config::default()); + + let semaphore = { + let mut pools = pool.pools.write().await; + let host_pool = pools.entry(addr).or_insert_with(|| HostPool::new(1)); + host_pool.semaphore.clone() + }; + semaphore.close(); + + let err = pool.expand_and_acquire(addr, &semaphore).await.unwrap_err(); + assert!( + err.to_string().contains("Semaphore closed"), + "unexpected error: {err}" + ); + assert_eq!(pool.stats.pool_expansions.load(Ordering::Relaxed), 0); + } + #[tokio::test] async fn test_connection_pool_creation() { let pool = ConnectionPool::new(Http2Config::default()); let stats = pool.stats(); assert_eq!(stats.connections_created.load(Ordering::Relaxed), 0); } + + #[tokio::test] + async fn test_connection_pool_expands_and_releases_permits() { + let addr = spawn_h2c_server().await; + let http2_config = Http2Config::default().connect_timeout(Duration::from_secs(2)); + let pool_config = PoolConfig { + max_connections_per_host: 2, + max_connection_age: None, + max_idle_time: Some(Duration::from_millis(10)), + max_requests_per_connection: None, + ..Default::default() + }; + let pool = ConnectionPool::with_config(http2_config, pool_config); + + let (g1, g2, g3, g4) = tokio::join!( + pool.get_connection(addr), + pool.get_connection(addr), + pool.get_connection(addr), + pool.get_connection(addr), + ); + let guards = vec![g1.unwrap(), g2.unwrap(), g3.unwrap(), g4.unwrap()]; + + let info = pool.pool_info().await; + let hosts = info["hosts"].as_array().unwrap(); + assert_eq!(hosts.len(), 1); + assert_eq!(hosts[0]["total_connections"].as_u64().unwrap(), 4); + assert!( + info["stats"]["pool_expansions"].as_u64().unwrap() >= 1, + "pool_info did not report an expansion: {info}" + ); + + drop(guards); + // Wait past max_idle_time so cleanup() will treat all conns as stale. + tokio::time::sleep(Duration::from_millis(20)).await; + pool.cleanup().await; + + let (total_connections, current_cap, semaphore) = { + let pools = pool.pools.read().await; + let host_pool = pools.get(&addr).unwrap(); + ( + host_pool.connections.len(), + host_pool.current_cap, + host_pool.semaphore.clone(), + ) + }; + + assert_eq!(total_connections, 0); + assert_eq!(semaphore.available_permits(), current_cap); + let _permit = semaphore.try_acquire_owned().unwrap(); + } } diff --git a/examples/rust/http2_pool_expand.rs b/examples/rust/http2_pool_expand.rs new file mode 100644 index 000000000..26069f36e --- /dev/null +++ b/examples/rust/http2_pool_expand.rs @@ -0,0 +1,134 @@ +//! HTTP/2 Connection Pool 2x 自动扩容示例 +//! +//! 演示当并发请求数超过 `max_connections_per_host` 时,连接池会自动 2x 扩容, +//! 而不是抛出 "Connection pool exhausted" 错误。 +//! +//! Run: cargo run --example http2_pool_expand -p pulsing-actor + +use bytes::Bytes; +use http_body_util::Full; +use hyper::service::service_fn; +use hyper::{Request, Response}; +use hyper_util::rt::{TokioExecutor, TokioIo}; +use pulsing_actor::transport::http2::{ConnectionPool, Http2Config, PoolConfig}; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::net::TcpListener; + +const INITIAL_CAP: usize = 2; +const BURST_SIZE: usize = 8; +const HOLD_DURATION: Duration = Duration::from_millis(300); + +#[tokio::main] +async fn main() -> Result<(), Box> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "warn,pulsing_actor::transport::http2=warn".into()), + ) + .init(); + + let addr = spawn_h2c_server().await?; + println!("h2c server listening on {addr}"); + + let pool_config = PoolConfig { + max_connections_per_host: INITIAL_CAP, + // 关闭 cleanup 触发条件,让所有连接在 burst 期间稳稳存活 + max_connection_age: None, + max_idle_time: None, + max_requests_per_connection: None, + ..Default::default() + }; + let pool = Arc::new(ConnectionPool::with_config( + Http2Config::default(), + pool_config, + )); + + println!( + "\n=== burst test ===\n initial cap : {INITIAL_CAP}\n burst size : {BURST_SIZE}\n hold time : {HOLD_DURATION:?}\n expected : pool grows 2 -> 4 -> 8, all {BURST_SIZE} requests succeed\n" + ); + + let mut handles = Vec::with_capacity(BURST_SIZE); + for i in 0..BURST_SIZE { + let pool = pool.clone(); + handles.push(tokio::spawn(async move { + let started = Instant::now(); + let outcome = pool.get_connection(addr).await; + let elapsed = started.elapsed(); + match outcome { + Ok(guard) => { + println!(" [task {i:>2}] acquired in {elapsed:>8.2?}"); + tokio::time::sleep(HOLD_DURATION).await; + drop(guard); + Ok(()) + } + Err(e) => { + eprintln!(" [task {i:>2}] FAILED after {elapsed:>8.2?}: {e}"); + Err(e) + } + } + })); + } + + let mut ok = 0usize; + let mut err = 0usize; + for h in handles { + match h.await? { + Ok(()) => ok += 1, + Err(_) => err += 1, + } + } + + println!("\nresults: {ok} succeeded, {err} failed"); + + let info = pool.pool_info().await; + println!("\npool_info:\n{}", serde_json::to_string_pretty(&info)?); + + let expansions = info["stats"]["pool_expansions"].as_u64().unwrap_or(0); + let final_cap = info["hosts"] + .as_array() + .and_then(|a| a.first()) + .and_then(|h| h["total_connections"].as_u64()) + .unwrap_or(0); + + println!("\n--- verdict ---"); + if err == 0 && expansions >= 1 && final_cap as usize >= BURST_SIZE { + println!( + "PASS: pool expanded {expansions} time(s); all {BURST_SIZE} requests succeeded; \ + {final_cap} connections live (>= burst size)" + ); + Ok(()) + } else { + Err(format!( + "FAIL: succeeded={ok}, failed={err}, expansions={expansions}, live_conns={final_cap}" + ) + .into()) + } +} + +async fn spawn_h2c_server() -> std::io::Result { + let listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let addr = listener.local_addr()?; + + tokio::spawn(async move { + loop { + let (stream, _peer) = match listener.accept().await { + Ok(a) => a, + Err(_) => break, + }; + tokio::spawn(async move { + let io = TokioIo::new(stream); + let service = service_fn(|_req: Request| async { + Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok")))) + }); + let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new()) + .serve_connection(io, service) + .await; + }); + } + }); + + Ok(addr) +}