From ea296bbdc843406d4c6f7001e738aacf6e0ed23b Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sat, 11 Jul 2026 21:36:01 +0300 Subject: [PATCH] Replace per-session pool trimming with pressure hysteresis Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/maestro/mod.rs | 1 + src/metrics.rs | 10 ++++++++++ src/proxy/direct_buffer_budget.rs | 30 +++++++++++++++++++++++++++++- src/proxy/direct_relay.rs | 1 - src/proxy/middle_relay/session.rs | 1 - src/stats/me_getters.rs | 6 ++++++ src/stats/mod.rs | 1 + src/stats/writer_counters.rs | 8 ++++++++ 8 files changed, 55 insertions(+), 3 deletions(-) diff --git a/src/maestro/mod.rs b/src/maestro/mod.rs index a848fda..5143f9c 100644 --- a/src/maestro/mod.rs +++ b/src/maestro/mod.rs @@ -899,6 +899,7 @@ async fn run_telemt_core( ); spawn_direct_buffer_budget_controller( direct_buffer_budget, + buffer_pool.clone(), stats.clone(), shared_state.clone(), config.server.max_connections, diff --git a/src/metrics.rs b/src/metrics.rs index a7e008a..acbe502 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -595,6 +595,16 @@ async fn render_metrics( "telemt_buffer_pool_buffers_total{{kind=\"in_use\"}} {}", stats.get_buffer_pool_in_use_gauge() ); + let _ = writeln!( + out, + "# HELP telemt_buffer_pool_events_total Buffer-pool allocation lifecycle events" + ); + let _ = writeln!(out, "# TYPE telemt_buffer_pool_events_total counter"); + let _ = writeln!( + out, + "telemt_buffer_pool_events_total{{event=\"replaced_nonstandard\"}} {}", + stats.get_buffer_pool_replaced_nonstandard_total() + ); let direct_budget = shared_state.direct_buffer_budget.snapshot(); let _ = writeln!( diff --git a/src/proxy/direct_buffer_budget.rs b/src/proxy/direct_buffer_budget.rs index bb25a04..44ed537 100644 --- a/src/proxy/direct_buffer_budget.rs +++ b/src/proxy/direct_buffer_budget.rs @@ -5,6 +5,7 @@ use std::time::Duration; use tokio::sync::watch; use crate::stats::Stats; +use crate::stream::BufferPool; use super::shared_state::ProxySharedState; @@ -21,6 +22,8 @@ const AUTO_HARD_FALLBACK_BYTES: usize = 512 * 1024 * 1024; const TARGET_FLOOR_MIN_BYTES: usize = 16 * 1024 * 1024; const CONTROL_INTERVAL: Duration = Duration::from_secs(1); const HEALTHY_RECOVERY_SAMPLES: u8 = 30; +const BUFFER_POOL_TRIM_LOW_WATERMARK: usize = 64; +const BUFFER_POOL_TRIM_HIGH_WATERMARK: usize = 128; #[derive(Debug, Clone, Copy, Default)] /// Lock-free observability snapshot of the Direct copy-buffer envelope. @@ -337,9 +340,10 @@ pub(crate) async fn resolve_direct_buffer_hard_limit(configured: usize) -> usize align_down(derived as usize).max(DIRECT_BUFFER_UNIT_BYTES) } -/// Starts the single control-plane task that adjusts the runtime target. +/// Starts the single control-plane task for Direct budget and shared pool pressure. pub(crate) fn spawn_direct_buffer_budget_controller( budget: Arc, + buffer_pool: Arc, stats: Arc, shared: Arc, max_connections: u32, @@ -351,6 +355,13 @@ pub(crate) fn spawn_direct_buffer_budget_controller( let mut previous_denied = 0u64; let mut previous_fallback = 0u64; let mut previous_rejected = 0u64; + let pool_trim_low = buffer_pool + .max_buffers() + .min(BUFFER_POOL_TRIM_LOW_WATERMARK); + let pool_trim_high = buffer_pool + .max_buffers() + .min(BUFFER_POOL_TRIM_HIGH_WATERMARK); + let mut pool_trim_armed = true; loop { interval.tick().await; @@ -380,6 +391,23 @@ pub(crate) fn spawn_direct_buffer_budget_controller( || fallback_delta > 0 || rejected_delta > 0; + if !pressure { + pool_trim_armed = true; + } else if pool_trim_armed && buffer_pool.pooled() > pool_trim_high { + buffer_pool.trim_to(pool_trim_low); + pool_trim_armed = false; + } + + let pool_snapshot = buffer_pool.stats(); + stats.set_buffer_pool_gauges( + pool_snapshot.pooled, + pool_snapshot.allocated, + pool_snapshot.allocated.saturating_sub(pool_snapshot.pooled), + ); + stats.set_buffer_pool_replaced_nonstandard_total( + pool_snapshot.replaced_nonstandard, + ); + let headroom_target = if sample.total_bytes == 0 { snapshot.hard_limit_bytes } else { diff --git a/src/proxy/direct_relay.rs b/src/proxy/direct_relay.rs index 5f46272..d3a9d73 100644 --- a/src/proxy/direct_relay.rs +++ b/src/proxy/direct_relay.rs @@ -400,7 +400,6 @@ where Err(e) => debug!(user = %user, error = %e, "Direct relay ended with error"), } - buffer_pool_trim.trim_to(buffer_pool_trim.max_buffers().min(64)); let pool_snapshot = buffer_pool_trim.stats(); stats.set_buffer_pool_gauges( pool_snapshot.pooled, diff --git a/src/proxy/middle_relay/session.rs b/src/proxy/middle_relay/session.rs index ab90b97..371a5c5 100644 --- a/src/proxy/middle_relay/session.rs +++ b/src/proxy/middle_relay/session.rs @@ -823,7 +823,6 @@ where clear_relay_idle_candidate_in(shared.as_ref(), conn_id); me_pool.registry().unregister(conn_id).await; - buffer_pool.trim_to(buffer_pool.max_buffers().min(64)); let pool_snapshot = buffer_pool.stats(); stats.set_buffer_pool_gauges( pool_snapshot.pooled, diff --git a/src/stats/me_getters.rs b/src/stats/me_getters.rs index 4ab6dac..34a4e08 100644 --- a/src/stats/me_getters.rs +++ b/src/stats/me_getters.rs @@ -204,6 +204,12 @@ impl Stats { self.buffer_pool_in_use_gauge.load(Ordering::Relaxed) } + /// Returns the count of non-standard buffers replaced before pooling. + pub fn get_buffer_pool_replaced_nonstandard_total(&self) -> u64 { + self.buffer_pool_replaced_nonstandard_total + .load(Ordering::Relaxed) + } + pub fn get_me_c2me_send_full_total(&self) -> u64 { self.me_c2me_send_full_total.load(Ordering::Relaxed) } diff --git a/src/stats/mod.rs b/src/stats/mod.rs index b56ee5b..ba71113 100644 --- a/src/stats/mod.rs +++ b/src/stats/mod.rs @@ -274,6 +274,7 @@ pub struct Stats { buffer_pool_pooled_gauge: AtomicU64, buffer_pool_allocated_gauge: AtomicU64, buffer_pool_in_use_gauge: AtomicU64, + buffer_pool_replaced_nonstandard_total: AtomicU64, // C2ME enqueue observability me_c2me_send_full_total: AtomicU64, me_c2me_send_high_water_total: AtomicU64, diff --git a/src/stats/writer_counters.rs b/src/stats/writer_counters.rs index f8c852c..02c74ec 100644 --- a/src/stats/writer_counters.rs +++ b/src/stats/writer_counters.rs @@ -552,6 +552,14 @@ impl Stats { } } + /// Publishes the cumulative count of non-standard pool buffer replacements. + pub fn set_buffer_pool_replaced_nonstandard_total(&self, value: usize) { + if self.telemetry_me_allows_normal() { + self.buffer_pool_replaced_nonstandard_total + .store(value as u64, Ordering::Relaxed); + } + } + pub fn increment_me_c2me_send_full_total(&self) { if self.telemetry_me_allows_normal() { self.me_c2me_send_full_total.fetch_add(1, Ordering::Relaxed);