refactor: improve networking layer with enhanced engine, muxer, and diagnostics
- Refactored connection engine with better state management - Improved muxer with enhanced protocol handling - Updated connection bridge and diagnostics - Added new network constants and configuration options - Enhanced session tracking and error handling - Updated dependencies in Cargo.toml
This commit is contained in:
Kirill
+11
-4
@@ -40,10 +40,17 @@ impl NetworkConfig {
|
||||
const BULK_WINDOW_SEGMENTS: usize = 128;
|
||||
const LIGHT_WINDOW_SEGMENTS: usize = 32;
|
||||
|
||||
// How many MTU-sized packets the Tokio mpsc channel should hold.
|
||||
// At MTU 1450 and 128 slots: ~185 KB per channel — enough to absorb
|
||||
// ~15 ms of jitter at 100 Mbps without spawning backpressure tasks.
|
||||
const CHANNEL_PACKETS: usize = 128;
|
||||
// How many messages the Tokio mpsc channels hold.
|
||||
//
|
||||
// 🔥 ANTI-BUFFERBLOAT: this is the dominant app-layer queue on every
|
||||
// tunnel leg. A single server→leg data message can be up to one read
|
||||
// buffer (~180 KB), so 128 slots meant up to ~23 MB of in-flight data
|
||||
// QUEUED per leg. After a speedtest that reservoir is full of data for
|
||||
// streams the app already closed; the downlink wastes seconds draining
|
||||
// it (observed: mux_dispatch no_stream ≫ ok, RTT → 1.3 s, tunnel "dies").
|
||||
// 16 slots bounds the per-leg queue ~8× lower so it drains in ~1 s and
|
||||
// RTT stays low, while still keeping the writer fed for full throughput.
|
||||
const CHANNEL_PACKETS: usize = 16;
|
||||
|
||||
// Payload bytes per segment (no IP/TCP headers in the smoltcp buffer).
|
||||
let seg = mtu.saturating_sub(40).max(512); // subtract typical IP+TCP overhead
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::net::connection::muxer::Muxer;
|
||||
use crate::net::{NetworkConfig, BRIDGE_IDLE_TIMEOUT};
|
||||
use crate::net::connection::muxer::{adaptive_write_timeout, Muxer};
|
||||
use crate::net::{
|
||||
NetworkConfig, BRIDGE_IDLE_TIMEOUT, BRIDGE_READ_CHUNK, BRIDGE_STREAM_WRITE_TIMEOUT,
|
||||
STREAM_PAUSE_BUDGET, STREAM_PAUSE_RETRY,
|
||||
};
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use netrunner_logger::{debug, error, info, warn};
|
||||
use std::time::Instant;
|
||||
use tokio::net::UdpSocket;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::timeout;
|
||||
@@ -34,7 +38,6 @@ pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
stream_id,
|
||||
muxer: muxer.clone(),
|
||||
};
|
||||
let buf_size = NetworkConfig::global().tcp_buffer_size;
|
||||
let token = CancellationToken::new();
|
||||
|
||||
// Upload: internet → tunnel.
|
||||
@@ -45,10 +48,13 @@ pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
let token = token.clone();
|
||||
async move {
|
||||
let mut reader = reader;
|
||||
let mut buf = BytesMut::with_capacity(buf_size);
|
||||
// Read in ≤ BRIDGE_READ_CHUNK (one-frame) units so a single data
|
||||
// message can't be huge. Combined with CHANNEL_PACKETS this byte-bounds
|
||||
// the per-leg queue and keeps post-speedtest bufferbloat small.
|
||||
let mut buf = BytesMut::with_capacity(BRIDGE_READ_CHUNK);
|
||||
loop {
|
||||
if buf.capacity() < 16384 {
|
||||
buf.reserve(buf_size);
|
||||
if buf.capacity() - buf.len() < BRIDGE_READ_CHUNK {
|
||||
buf.reserve(BRIDGE_READ_CHUNK);
|
||||
}
|
||||
tokio::select! {
|
||||
biased;
|
||||
@@ -57,7 +63,31 @@ pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(_) => {
|
||||
let data = buf.split().freeze();
|
||||
if muxer.send_data_safe(stream_id, data, false).await.is_err() {
|
||||
// 🔥 GRACEFUL PAUSE (anti-domino).
|
||||
// send_data_safe already fails over between live legs;
|
||||
// it only errors when EVERY leg is down. In that case we
|
||||
// do NOT close the stream — we hold this chunk and retry
|
||||
// while the engine reconnects, bounded by STREAM_PAUSE_BUDGET.
|
||||
// Because we stop reading meanwhile, TCP back-pressure
|
||||
// naturally pauses the source instead of dropping data.
|
||||
let deadline = Instant::now() + STREAM_PAUSE_BUDGET;
|
||||
let mut delivered = false;
|
||||
loop {
|
||||
if muxer.send_data_safe(stream_id, data.clone(), false).await.is_ok() {
|
||||
delivered = true;
|
||||
break;
|
||||
}
|
||||
if Instant::now() >= deadline {
|
||||
warn!(stream_id, "Stream pause budget exceeded — no leg recovered, closing");
|
||||
break;
|
||||
}
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = token.cancelled() => break,
|
||||
_ = tokio::time::sleep(STREAM_PAUSE_RETRY) => {}
|
||||
}
|
||||
}
|
||||
if !delivered {
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -85,7 +115,11 @@ pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
None => break,
|
||||
Some(data) => {
|
||||
if data.is_empty() { continue; }
|
||||
match timeout(crate::net::BRIDGE_STREAM_WRITE_TIMEOUT, writer.write_all(&data)).await {
|
||||
// Adaptive: BRIDGE_STREAM_WRITE_TIMEOUT is the floor, but
|
||||
// under high RTT we grant the slow local socket more drain
|
||||
// time before declaring it stuck and closing the stream.
|
||||
let write_timeout = adaptive_write_timeout(BRIDGE_STREAM_WRITE_TIMEOUT);
|
||||
match timeout(write_timeout, writer.write_all(&data)).await {
|
||||
Ok(Ok(_)) => {}
|
||||
_ => break,
|
||||
}
|
||||
@@ -118,17 +152,27 @@ pub(crate) async fn run_udp_bridge(
|
||||
};
|
||||
|
||||
let config = NetworkConfig::global();
|
||||
let mut buf = vec![0u8; config.udp_buffer_size];
|
||||
let dgram_cap = config.udp_buffer_size;
|
||||
// 🔥 ZERO-COPY: receive directly into BytesMut spare capacity and hand the
|
||||
// datagram downstream via split().freeze() (ownership transfer, no memcpy).
|
||||
// Replaces `vec![0u8; N]` + `Bytes::copy_from_slice` (one full copy/datagram).
|
||||
let mut buf = BytesMut::with_capacity(dgram_cap);
|
||||
|
||||
info!(stream_id, "🌉 UDP Bridge active");
|
||||
|
||||
loop {
|
||||
// Guarantee room for a whole datagram so recv_buf never truncates it.
|
||||
if buf.capacity() - buf.len() < dgram_cap {
|
||||
buf.reserve(dgram_cap);
|
||||
}
|
||||
let select_res = timeout(BRIDGE_IDLE_TIMEOUT, async {
|
||||
tokio::select! {
|
||||
res = socket.recv(&mut buf) => {
|
||||
res = socket.recv_buf(&mut buf) => {
|
||||
match res {
|
||||
Ok(n) if n > 0 => {
|
||||
let data = Bytes::copy_from_slice(&buf[..n]);
|
||||
// Ownership transfer: the just-received bytes are moved
|
||||
// out with no copy; buf is left empty for the next reserve.
|
||||
let data = buf.split().freeze();
|
||||
if let Err(e) = muxer.send_data_safe(stream_id, data, true).await {
|
||||
warn!(stream_id, "UDP Tunnel legs dead. Dropping packet: {}", e);
|
||||
// 🔥 ФИКС: Опять же, не обрываем стрим из-за мертвого туннеля!
|
||||
|
||||
@@ -378,7 +378,61 @@ impl ClientHandler {
|
||||
|
||||
let muxer_inner = muxer.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(raw_frame) = rx_from_engine.recv().await {
|
||||
// Per-socket upload backlog. When a stream's up_tx is momentarily full we
|
||||
// stash the frame here and KEEP PROCESSING other streams — so one slow
|
||||
// upload can no longer head-of-line-block the shared loop, and we never
|
||||
// kill a healthy stream. Only a single stream sustaining more than
|
||||
// UPLOAD_PENDING_CAP buffered frames triggers bounded back-pressure
|
||||
// (a one-frame blocking send) to keep memory bounded.
|
||||
const UPLOAD_PENDING_CAP: usize = 64;
|
||||
let mut pending_upload: std::collections::HashMap<
|
||||
u64,
|
||||
std::collections::VecDeque<Bytes>,
|
||||
> = std::collections::HashMap::new();
|
||||
|
||||
loop {
|
||||
// Flush existing per-socket backlogs first (fully non-blocking).
|
||||
if !pending_upload.is_empty() {
|
||||
pending_upload.retain(|sid, q| {
|
||||
match local_to_upload_tx.get(sid).map(|r| r.value().clone()) {
|
||||
Some(up_tx) => {
|
||||
while let Some(front) = q.pop_front() {
|
||||
match up_tx.try_send(front) {
|
||||
Ok(_) => {}
|
||||
Err(mpsc::error::TrySendError::Full(p)) => {
|
||||
q.push_front(p);
|
||||
break;
|
||||
}
|
||||
Err(mpsc::error::TrySendError::Closed(_)) => {
|
||||
q.clear();
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
!q.is_empty() // keep the entry only if still backlogged
|
||||
}
|
||||
None => false, // socket gone — drop its backlog
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Wait for the next frame; while backlogged, also wake on a short timer
|
||||
// to retry the flush as the uplink drains.
|
||||
let raw_frame = if pending_upload.is_empty() {
|
||||
match rx_from_engine.recv().await {
|
||||
Some(f) => f,
|
||||
None => break,
|
||||
}
|
||||
} else {
|
||||
tokio::select! {
|
||||
f = rx_from_engine.recv() => match f {
|
||||
Some(f) => f,
|
||||
None => break,
|
||||
},
|
||||
_ = tokio::time::sleep(std::time::Duration::from_millis(5)) => continue,
|
||||
}
|
||||
};
|
||||
|
||||
if let Ok(nrxp_frame) = RawCastAdapter::to_nrxp(raw_frame.clone()) {
|
||||
let local_socket_id = raw_frame.socket_id;
|
||||
let f_type = nrxp_frame.header.frame_type;
|
||||
@@ -446,7 +500,11 @@ impl ClientHandler {
|
||||
}
|
||||
});
|
||||
|
||||
let (up_tx, mut up_rx) = mpsc::channel::<Bytes>(cap);
|
||||
// Per-stream upload buffer, kept deeper than the (deliberately
|
||||
// small, anti-bufferbloat) default so upload bursts — e.g. a
|
||||
// speedtest over a slow uplink — are absorbed here and the shared
|
||||
// rx_from_engine loop rarely has to apply back-pressure on it.
|
||||
let (up_tx, mut up_rx) = mpsc::channel::<Bytes>(cap.max(32));
|
||||
local_to_upload_tx.insert(local_socket_id, up_tx);
|
||||
|
||||
let m_clone = muxer_inner.clone();
|
||||
@@ -469,13 +527,44 @@ impl ClientHandler {
|
||||
.get(&local_socket_id)
|
||||
.map(|r| r.value().clone())
|
||||
{
|
||||
// .send().await blocks the upload task when the muxer leg
|
||||
// is saturated, creating back-pressure back to smoltcp
|
||||
// (no drops → no unnecessary retransmits → lower jitter).
|
||||
let _ = up_tx.send(payload).await;
|
||||
// PER-STREAM, no shared-loop HOL: if this stream already has a
|
||||
// backlog, queue behind it (preserve order). Otherwise try a
|
||||
// non-blocking send; on full, START a backlog and keep serving
|
||||
// OTHER streams. (Replaces the old 2 s blocking grace that
|
||||
// stalled the whole loop and then killed healthy streams.)
|
||||
if let Some(q) = pending_upload.get_mut(&local_socket_id) {
|
||||
q.push_back(payload);
|
||||
} else {
|
||||
match up_tx.try_send(payload) {
|
||||
Ok(_) => {}
|
||||
Err(mpsc::error::TrySendError::Closed(_)) => {}
|
||||
Err(mpsc::error::TrySendError::Full(p)) => {
|
||||
let mut q = std::collections::VecDeque::with_capacity(16);
|
||||
q.push_back(p);
|
||||
pending_upload.insert(local_socket_id, q);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Bounded back-pressure: only if THIS stream's backlog exceeds
|
||||
// the cap (sustained uplink-bound overload) do we block on a
|
||||
// single frame, so memory stays bounded. Never closes the
|
||||
// stream; other streams were already flushed at the loop top.
|
||||
let over_cap = pending_upload
|
||||
.get(&local_socket_id)
|
||||
.map_or(false, |q| q.len() > UPLOAD_PENDING_CAP);
|
||||
if over_cap {
|
||||
let front = pending_upload
|
||||
.get_mut(&local_socket_id)
|
||||
.and_then(|q| q.pop_front());
|
||||
if let Some(front) = front {
|
||||
let _ = up_tx.send(front).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
FrameType::Close => {
|
||||
pending_upload.remove(&local_socket_id);
|
||||
if let Some(kv) = local_to_global.remove(&local_socket_id) {
|
||||
let global_stream_id = kv.1;
|
||||
|
||||
|
||||
@@ -230,8 +230,9 @@ impl TunnelEngine {
|
||||
|
||||
for frame in frames {
|
||||
if frame.header.frame_type == FrameType::Heartbeat {
|
||||
let m = muxer.clone();
|
||||
tokio::spawn(async move { m.record_pong(leg_id).await; });
|
||||
// record_pong does no .await internally, so run it inline:
|
||||
// a spawn+Arc-clone per PONG was pure scheduler churn.
|
||||
muxer.record_pong(leg_id).await;
|
||||
}
|
||||
let _ = handler.handle(frame).await;
|
||||
}
|
||||
@@ -247,7 +248,6 @@ impl TunnelEngine {
|
||||
let mut heartbeat = tokio::time::interval(HEALTH_CHECK_INTERVAL);
|
||||
|
||||
let mut pending_data: Option<MuxMessage> = None;
|
||||
let interleave_chunk = TUNNEL_INTERLEAVE_CHUNK;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
@@ -286,6 +286,12 @@ impl TunnelEngine {
|
||||
_ = std::future::ready(()), if pending_data.is_some() => {
|
||||
let mut msg = pending_data.take().unwrap();
|
||||
|
||||
// #4 Adaptive batch: under high RTT take a bigger interleave
|
||||
// chunk so more frames coalesce into one write in
|
||||
// handle_outbound (#3); at low RTT stay small for fairness.
|
||||
let interleave_chunk = crate::net::connection::muxer::adaptive_batch_chunk(
|
||||
TUNNEL_INTERLEAVE_CHUNK,
|
||||
);
|
||||
let chunk_size = std::cmp::min(msg.data.len(), interleave_chunk);
|
||||
let chunk_data = msg.data.split_to(chunk_size);
|
||||
|
||||
@@ -439,23 +445,47 @@ impl TunnelEngine {
|
||||
}
|
||||
}
|
||||
|
||||
for pkt in packets {
|
||||
let write_future = outbound.write_all(&pkt);
|
||||
// 💡 ИЗМЕНЕНО: Увеличен таймаут отправки до 20 секунд для совместимости с агрессивным BBR
|
||||
if let Err(_) =
|
||||
tokio::time::timeout(std::time::Duration::from_secs(20), write_future).await
|
||||
{
|
||||
error!(stream_id, "🔥 Physical leg STUCK on write. Killing leg.");
|
||||
// Increment counter; the call site in run() emits the full event
|
||||
// with the correct leg_id since handle_outbound is a static fn.
|
||||
crate::net::diagnostics::DIAG_COUNTERS
|
||||
.tunnel_write_stalls
|
||||
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
return Err(AppError::new(
|
||||
ERR_INFRA_TIMEOUT,
|
||||
"Таймаут отправки",
|
||||
"Physical leg STUCK on write",
|
||||
));
|
||||
// Adaptive write deadline: floor of 20 s (BBR-friendly), but scales with
|
||||
// the live RTT so a high-latency path (RTT > 2.5 s) doesn't trip a flat
|
||||
// timeout on a leg that is slow rather than dead. Killing such a leg is
|
||||
// what set off the leg-drop → stream-close cascade.
|
||||
let write_timeout = crate::net::connection::muxer::adaptive_write_timeout(
|
||||
std::time::Duration::from_secs(20),
|
||||
);
|
||||
// #3 Syscall batching (sendmmsg-analog for a TCP byte stream): when a Data
|
||||
// message produced several MAX_FRAME_PAYLOAD frames, coalesce them into ONE
|
||||
// contiguous buffer and issue a single write_all instead of N — fewer
|
||||
// User→Kernel transitions under exactly the high-throughput conditions that
|
||||
// were producing tunnel_write_stuck. The single-frame case (control/UDP and
|
||||
// ≤16 KB payloads) keeps the zero-copy direct write with no extra copy.
|
||||
let stuck = || -> AppError {
|
||||
error!(stream_id, "🔥 Physical leg STUCK on write. Killing leg.");
|
||||
// Increment counter; the call site in run() emits the full event
|
||||
// with the correct leg_id since handle_outbound is a static fn.
|
||||
crate::net::diagnostics::DIAG_COUNTERS
|
||||
.tunnel_write_stalls
|
||||
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
AppError::new(
|
||||
ERR_INFRA_TIMEOUT,
|
||||
"Таймаут отправки",
|
||||
"Physical leg STUCK on write",
|
||||
)
|
||||
};
|
||||
|
||||
if packets.len() == 1 {
|
||||
let write_future = outbound.write_all(&packets[0]);
|
||||
if tokio::time::timeout(write_timeout, write_future).await.is_err() {
|
||||
return Err(stuck());
|
||||
}
|
||||
} else if !packets.is_empty() {
|
||||
let total: usize = packets.iter().map(|p| p.len()).sum();
|
||||
let mut batch = BytesMut::with_capacity(total);
|
||||
for pkt in &packets {
|
||||
batch.extend_from_slice(pkt);
|
||||
}
|
||||
let write_future = outbound.write_all(&batch);
|
||||
if tokio::time::timeout(write_timeout, write_future).await.is_err() {
|
||||
return Err(stuck());
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
|
||||
@@ -1,15 +1,16 @@
|
||||
use arc_swap::ArcSwap;
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
use netrunner_logger::{info, instrument, trace, warn, AppError, ERR_INFRA_TIMEOUT};
|
||||
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::Instant;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::sync::mpsc::Sender;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::net::diagnostics::{self, DiagnosticsEvent, LegMetrics, TunnelMetrics, DIAG_COUNTERS};
|
||||
use crate::net::{DISPATCH_TO_LOCAL_TIMEOUT, MAX_TUNNEL_LEGS};
|
||||
use crate::net::{INITIAL_RTT_MS, MUXER_CONGESTION_WEIGHT};
|
||||
use crate::net::INITIAL_RTT_MS;
|
||||
use crate::nrxp::FrameType;
|
||||
|
||||
#[derive(Default, Debug)]
|
||||
@@ -66,11 +67,43 @@ pub struct MuxMessage {
|
||||
|
||||
pub static GLOBAL_MIN_RTT: AtomicU32 = AtomicU32::new(INITIAL_RTT_MS);
|
||||
|
||||
/// Write timeout that scales with the observed network RTT.
|
||||
///
|
||||
/// On a healthy path (RTT ~50 ms) this stays at `floor`. When the path degrades
|
||||
/// to multi-second RTT (the > 2500 ms peaks seen in production), a flat 20 s
|
||||
/// timeout fires on a leg that is merely *slow*, not dead — and a killed leg
|
||||
/// triggers the leg-drop → stream-close cascade ("domino effect"). Allowing
|
||||
/// ~8 RTT of drain time (capped at 60 s) keeps slow-but-alive legs from being
|
||||
/// evicted under high latency, while still reaping genuinely stuck sockets.
|
||||
pub fn adaptive_write_timeout(floor: Duration) -> Duration {
|
||||
let rtt_ms = GLOBAL_MIN_RTT.load(Ordering::Relaxed) as u64;
|
||||
let scaled = Duration::from_millis(rtt_ms.saturating_mul(8));
|
||||
scaled.clamp(floor, Duration::from_secs(60))
|
||||
}
|
||||
|
||||
/// Interleave/batch chunk size that grows with RTT.
|
||||
///
|
||||
/// At low RTT keep the `base` (snappy, fair interleaving); under high RTT — where
|
||||
/// the bandwidth-delay product is large — write bigger batches per pass so more
|
||||
/// 16 KB frames coalesce into a single contiguous socket write (see
|
||||
/// `handle_outbound`), cutting the number of `write()` syscalls under exactly the
|
||||
/// conditions that were producing `tunnel_write_stuck`.
|
||||
///
|
||||
/// 1× at ≤250 ms, +1× per extra 250 ms of RTT, capped at 4×.
|
||||
pub fn adaptive_batch_chunk(base: usize) -> usize {
|
||||
let rtt_ms = GLOBAL_MIN_RTT.load(Ordering::Relaxed) as usize;
|
||||
let factor = (1 + rtt_ms / 250).clamp(1, 4);
|
||||
base.saturating_mul(factor)
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Muxer {
|
||||
legs: Arc<DashMap<u32, MuxLeg>>,
|
||||
// 🔥 ОПТИМИЗАЦИЯ: Lock-Free кэш для горячего пути
|
||||
active_legs_cache: Arc<RwLock<Arc<Vec<MuxLeg>>>>,
|
||||
// 🔥 ОПТИМИЗАЦИЯ: полностью lock-free кэш горячего пути.
|
||||
// ArcSwap: чтение (load_full) — атомарный bump Arc без блокировок; запись
|
||||
// (store) реже и тоже неблокирующая. Заменил RwLock<Arc<Vec>> — у которого
|
||||
// чтение брало read-guard.
|
||||
active_legs_cache: Arc<ArcSwap<Vec<MuxLeg>>>,
|
||||
|
||||
// Добавили CancellationToken для предотвращения утечек памяти (Зомби-задач)
|
||||
streams: Arc<DashMap<u32, (Sender<Bytes>, Arc<StreamStats>, CancellationToken)>>,
|
||||
@@ -78,24 +111,29 @@ pub struct Muxer {
|
||||
pending_pings: Arc<DashMap<u32, Instant>>,
|
||||
id_gen: Arc<IdGenerator>,
|
||||
session_id: Arc<String>,
|
||||
/// Rotating cursor for round-robin leg selection among similar-quality legs,
|
||||
/// so a burst of new streams spreads across legs instead of all binding to
|
||||
/// the single current-best one (thundering herd).
|
||||
rr_counter: Arc<AtomicU32>,
|
||||
}
|
||||
|
||||
impl Muxer {
|
||||
pub fn new(is_client: bool, session_id: String) -> Self {
|
||||
Self {
|
||||
legs: Arc::new(DashMap::new()),
|
||||
active_legs_cache: Arc::new(RwLock::new(Arc::new(Vec::new()))),
|
||||
active_legs_cache: Arc::new(ArcSwap::from_pointee(Vec::new())),
|
||||
streams: Arc::new(DashMap::new()),
|
||||
stream_bindings: Arc::new(DashMap::new()),
|
||||
id_gen: Arc::new(IdGenerator::new(is_client)),
|
||||
pending_pings: Arc::new(DashMap::new()),
|
||||
session_id: Arc::new(session_id),
|
||||
rr_counter: Arc::new(AtomicU32::new(0)),
|
||||
}
|
||||
}
|
||||
|
||||
fn update_legs_cache(&self) {
|
||||
let new_cache: Vec<MuxLeg> = self.legs.iter().map(|kv| kv.value().clone()).collect();
|
||||
*self.active_legs_cache.write().unwrap() = Arc::new(new_cache);
|
||||
self.active_legs_cache.store(Arc::new(new_cache));
|
||||
}
|
||||
|
||||
pub fn add_leg(
|
||||
@@ -126,6 +164,13 @@ impl Muxer {
|
||||
);
|
||||
}
|
||||
|
||||
/// Drop every stream→leg binding that points at `leg_id`. Stale bindings to a
|
||||
/// removed leg force `select_leg` to re-balance each affected stream onto a
|
||||
/// healthy leg on its next send, instead of repeatedly probing the dead one.
|
||||
fn clear_bindings_for_leg(&self, leg_id: u32) {
|
||||
self.stream_bindings.retain(|_, bound_leg| *bound_leg != leg_id);
|
||||
}
|
||||
|
||||
pub fn remove_leg(&self, leg_id: u32, tx: &Sender<MuxMessage>) {
|
||||
let should_remove = self
|
||||
.legs
|
||||
@@ -133,6 +178,9 @@ impl Muxer {
|
||||
.map_or(false, |leg| leg.control_tx.same_channel(tx));
|
||||
if should_remove {
|
||||
self.legs.remove(&leg_id);
|
||||
// Unbind streams BEFORE refreshing the cache so a concurrent
|
||||
// select_leg never re-binds a stream to the leg we are evicting.
|
||||
self.clear_bindings_for_leg(leg_id);
|
||||
self.update_legs_cache();
|
||||
info!(
|
||||
leg_id,
|
||||
@@ -143,6 +191,7 @@ impl Muxer {
|
||||
|
||||
pub fn force_remove_leg(&self, leg_id: u32) {
|
||||
if self.legs.remove(&leg_id).is_some() {
|
||||
self.clear_bindings_for_leg(leg_id);
|
||||
self.update_legs_cache();
|
||||
info!(leg_id, "MUXER: TCP leg force-removed on engine exit");
|
||||
}
|
||||
@@ -159,36 +208,56 @@ impl Muxer {
|
||||
}
|
||||
|
||||
fn select_leg(&self, stream_id: u32) -> Option<MuxLeg> {
|
||||
// 1. Читаем кэш (это Arc, поэтому clone здесь — это просто инкремент счетчика, не копирование данных)
|
||||
let legs = self.active_legs_cache.read().unwrap().clone();
|
||||
if legs.is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
// 2. Если поток уже привязан к леге, используем её (Sticky Connection)
|
||||
// 1. FAST PATH (hot, per data frame): a bound stream resolves its leg by
|
||||
// id straight from the legs map — no full-cache Arc clone and no vector
|
||||
// scan. Reading `legs` (source of truth, not the cached snapshot) also
|
||||
// transparently picks up a leg that reconnected under the same id.
|
||||
if let Some(leg_id_ref) = self.stream_bindings.get(&stream_id) {
|
||||
let leg_id = *leg_id_ref;
|
||||
if let Some(leg) = legs.iter().find(|l| l.id == leg_id) {
|
||||
if let Some(leg) = self.legs.get(&leg_id) {
|
||||
return Some(leg.clone());
|
||||
}
|
||||
// Bound leg disappeared — fall through and re-pick a fresh one below.
|
||||
}
|
||||
|
||||
// 2. New (or re-homed) stream: load the leg set and choose.
|
||||
let legs = self.active_legs_cache.load_full();
|
||||
if legs.is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
// 3. O(N) поиск лучшей леги без сортировки всего вектора.
|
||||
// Consider all available legs so the 4th leg is not permanently starved.
|
||||
// MUXER_POOL_SIZE is kept for topology printing but no longer limits
|
||||
// leg selection: sticky bindings already prevent hot-leg thrashing.
|
||||
let selected_leg = legs
|
||||
.iter()
|
||||
.min_by(|a, b| {
|
||||
let score_a = a.stats.rtt_ms.load(Ordering::Relaxed) as f64
|
||||
+ (a.congestion_factor() * MUXER_CONGESTION_WEIGHT);
|
||||
let score_b = b.stats.rtt_ms.load(Ordering::Relaxed) as f64
|
||||
+ (b.congestion_factor() * MUXER_CONGESTION_WEIGHT);
|
||||
score_a
|
||||
.partial_cmp(&score_b)
|
||||
.unwrap_or(std::cmp::Ordering::Equal)
|
||||
})
|
||||
.cloned();
|
||||
// RTT-DOMINANT score: a leg's latency sets the scale, congestion only
|
||||
// modulates within legs of similar RTT. A drastically slower leg is never
|
||||
// preferred over a fast one, even when the fast leg is congested. (The old
|
||||
// additive `rtt + congestion*2000` could score a busy 160 ms leg WORSE
|
||||
// than an idle 1300 ms one, routing new streams onto the laggy leg.)
|
||||
let score = |leg: &MuxLeg| -> f64 {
|
||||
let rtt = (leg.stats.rtt_ms.load(Ordering::Relaxed) as f64).max(1.0);
|
||||
rtt * (1.0 + leg.congestion_factor())
|
||||
};
|
||||
|
||||
let best = legs.iter().map(|l| score(l)).fold(f64::MAX, f64::min);
|
||||
|
||||
// Candidate set = every leg within 2× of the best score. Drastically
|
||||
// worse (slow / bufferbloated) legs are excluded; near-equal legs are all
|
||||
// eligible. We then ROUND-ROBIN across the candidates so a burst of new
|
||||
// streams (speedtest / multi-connection upload opening many sockets at
|
||||
// once, before congestion registers) spreads across legs instead of all
|
||||
// binding to the single current-best leg — which previously left one leg
|
||||
// saturated and the others idle (low aggregate upload + stop-start stalls).
|
||||
let candidates: Vec<&MuxLeg> = legs.iter().filter(|&l| score(l) <= best * 2.0).collect();
|
||||
|
||||
let selected_leg = if candidates.is_empty() {
|
||||
None
|
||||
} else {
|
||||
let idx =
|
||||
self.rr_counter.fetch_add(1, Ordering::Relaxed) as usize % candidates.len();
|
||||
Some(candidates[idx].clone())
|
||||
};
|
||||
|
||||
if let Some(leg) = selected_leg {
|
||||
self.stream_bindings.insert(stream_id, leg.id);
|
||||
@@ -230,47 +299,75 @@ impl Muxer {
|
||||
}
|
||||
|
||||
#[instrument(skip(self, message), fields(session_id = %self.session_id, stream_id = message.stream_id, frame = ?message.frame_type))]
|
||||
pub async fn send_to_network(&self, message: MuxMessage) -> Result<(), AppError> {
|
||||
let leg = match self.select_leg(message.stream_id) {
|
||||
Some(l) => l,
|
||||
None => {
|
||||
return Err(AppError::new(
|
||||
ERR_INFRA_TIMEOUT,
|
||||
"Нет связи",
|
||||
"No active legs",
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
pub async fn send_to_network(&self, mut message: MuxMessage) -> Result<(), AppError> {
|
||||
let is_data = matches!(message.frame_type, FrameType::Data | FrameType::UdpData);
|
||||
let stream_id = message.stream_id;
|
||||
let size = message.data.len() as u64;
|
||||
|
||||
if is_data {
|
||||
// 💡 ДАННЫЕ: Используем .send().await для создания Backpressure
|
||||
match leg.data_tx.send(message).await {
|
||||
Ok(_) => {
|
||||
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
|
||||
if let Some(stream_ref) = self.streams.get(&stream_id) {
|
||||
stream_ref
|
||||
.value()
|
||||
.1
|
||||
.tx_bytes
|
||||
.fetch_add(size, Ordering::Relaxed);
|
||||
// 🔥 ANTI-DOMINO FAILOVER.
|
||||
// A single leg dropping must NOT close the stream. We evict the dead
|
||||
// leg, unbind the stream, and retry on the next-best leg. Only when
|
||||
// *every* leg is gone do we return Err — and the bridge treats that
|
||||
// as "pause & buffer", not "close" (see run_tcp_bridge). The loop is
|
||||
// bounded: remove_leg drops the leg from the cache, so select_leg can
|
||||
// never hand back the same dead leg, and it terminates at None.
|
||||
loop {
|
||||
let leg = match self.select_leg(message.stream_id) {
|
||||
Some(l) => l,
|
||||
None => {
|
||||
return Err(AppError::new(
|
||||
ERR_INFRA_TIMEOUT,
|
||||
"Нет связи",
|
||||
"No active legs",
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let stream_id = message.stream_id;
|
||||
let size = message.data.len() as u64;
|
||||
|
||||
// 💡 ДАННЫЕ: Используем .send().await для создания Backpressure
|
||||
match leg.data_tx.send(message).await {
|
||||
Ok(_) => {
|
||||
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
|
||||
if let Some(stream_ref) = self.streams.get(&stream_id) {
|
||||
stream_ref
|
||||
.value()
|
||||
.1
|
||||
.tx_bytes
|
||||
.fetch_add(size, Ordering::Relaxed);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
Err(send_err) => {
|
||||
// Recover the payload from the failed send so the retry
|
||||
// on another leg does not lose the chunk.
|
||||
message = send_err.0;
|
||||
DIAG_COUNTERS.upload_fails.fetch_add(1, Ordering::Relaxed);
|
||||
diagnostics::send_diag_event(DiagnosticsEvent::UploadFailed {
|
||||
stream_id,
|
||||
reason: "data channel closed (leg dropped) — failing over".into(),
|
||||
});
|
||||
// Evict the dead leg (also unbinds its streams) so the
|
||||
// next select_leg re-balances onto a healthy leg.
|
||||
self.remove_leg(leg.id, &leg.control_tx);
|
||||
// loop → pick another leg, or return Err if none remain.
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Err(_) => {
|
||||
DIAG_COUNTERS.upload_fails.fetch_add(1, Ordering::Relaxed);
|
||||
diagnostics::send_diag_event(DiagnosticsEvent::UploadFailed {
|
||||
stream_id,
|
||||
reason: "data channel closed (leg dropped)".into(),
|
||||
});
|
||||
self.remove_leg(leg.id, &leg.control_tx);
|
||||
Err(AppError::new(ERR_INFRA_TIMEOUT, "Обрыв", "Leg closed"))
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let leg = match self.select_leg(message.stream_id) {
|
||||
Some(l) => l,
|
||||
None => {
|
||||
return Err(AppError::new(
|
||||
ERR_INFRA_TIMEOUT,
|
||||
"Нет связи",
|
||||
"No active legs",
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let stream_id = message.stream_id;
|
||||
let size = message.data.len() as u64;
|
||||
// Close and Heartbeat frames MUST be delivered reliably (.send().await).
|
||||
// Close: dropping it leaks stream resources.
|
||||
// Heartbeat (PONG): dropping it via try_send causes the health-check
|
||||
@@ -380,15 +477,16 @@ impl Muxer {
|
||||
self.stream_bindings.remove(&stream_id);
|
||||
}
|
||||
|
||||
// ORDERING CONTRACT: callers MUST .await this; the caller (TunnelEngine reader)
|
||||
// is a spawned task, so blocking here creates correct back-pressure all the way
|
||||
// back to the kernel TCP socket buffer. Never spawn a task to deliver data
|
||||
// from this function — that breaks in-order delivery guarantees.
|
||||
// ORDERING CONTRACT: in-order delivery — never spawn a task to deliver data
|
||||
// from this function.
|
||||
//
|
||||
// TIMEOUT GUARD: if the stream's receive channel stays full for longer than
|
||||
// DISPATCH_TO_LOCAL_TIMEOUT the stream is forcibly closed. Without this a
|
||||
// single slow consumer (app socket buffer full, background app paused, etc.)
|
||||
// would block the engine reader and starve every other stream on the same leg.
|
||||
// HEAD-OF-LINE GUARD: the hot path is a non-blocking try_send, so one slow or
|
||||
// dead stream can NEVER block the shared per-leg reader. (A finished speedtest
|
||||
// socket the app stopped reading used to back its channel up and freeze EVERY
|
||||
// other download on that leg, because the reader awaited here for up to 10 s.)
|
||||
// Only a genuinely-full channel gets a SHORT grace wait (DISPATCH_TO_LOCAL_
|
||||
// TIMEOUT); if it is still full that ONE stream is closed so the leg keeps
|
||||
// serving everyone else.
|
||||
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
|
||||
let size = data.len() as u64;
|
||||
|
||||
@@ -397,22 +495,53 @@ impl Muxer {
|
||||
(val.0.clone(), val.1.clone())
|
||||
});
|
||||
|
||||
if let Some((tx, stats)) = tx_and_stats {
|
||||
match tokio::time::timeout(DISPATCH_TO_LOCAL_TIMEOUT, tx.send(data)).await {
|
||||
Ok(Ok(_)) => {
|
||||
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
|
||||
}
|
||||
Ok(Err(_)) => { /* receiver already closed — stream gone */ }
|
||||
Err(_) => {
|
||||
// Bridge isn't consuming: app socket full or app paused too long.
|
||||
// Close the stream to free the leg for all other streams.
|
||||
warn!(
|
||||
stream_id,
|
||||
"dispatch_to_local: stream stalled for {:?}, closing",
|
||||
DISPATCH_TO_LOCAL_TIMEOUT
|
||||
);
|
||||
self.remove_stream(stream_id);
|
||||
}
|
||||
let Some((tx, stats)) = tx_and_stats else {
|
||||
// No stream registered for this id (already closed / never opened).
|
||||
DIAG_COUNTERS
|
||||
.mux_dispatch_no_stream
|
||||
.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
};
|
||||
|
||||
// Fast path: deliver without awaiting → zero head-of-line blocking.
|
||||
let data = match tx.try_send(data) {
|
||||
Ok(()) => {
|
||||
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
|
||||
DIAG_COUNTERS.mux_dispatch_ok.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
}
|
||||
// Receiver already closed — stream gone.
|
||||
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
|
||||
DIAG_COUNTERS
|
||||
.mux_dispatch_recv_closed
|
||||
.fetch_add(1, Ordering::Relaxed);
|
||||
return;
|
||||
}
|
||||
// Channel full: recover the payload and fall through to a bounded wait.
|
||||
Err(tokio::sync::mpsc::error::TrySendError::Full(data)) => data,
|
||||
};
|
||||
|
||||
match tokio::time::timeout(DISPATCH_TO_LOCAL_TIMEOUT, tx.send(data)).await {
|
||||
Ok(Ok(_)) => {
|
||||
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
|
||||
DIAG_COUNTERS.mux_dispatch_ok.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Ok(Err(_)) => {
|
||||
DIAG_COUNTERS
|
||||
.mux_dispatch_recv_closed
|
||||
.fetch_add(1, Ordering::Relaxed);
|
||||
}
|
||||
Err(_) => {
|
||||
// Consumer stayed full past the grace window: close just this one
|
||||
// stream so the leg keeps serving everyone else.
|
||||
DIAG_COUNTERS
|
||||
.mux_dispatch_full_closed
|
||||
.fetch_add(1, Ordering::Relaxed);
|
||||
warn!(
|
||||
stream_id,
|
||||
"dispatch_to_local: stream stalled for {:?}, closing", DISPATCH_TO_LOCAL_TIMEOUT
|
||||
);
|
||||
self.remove_stream(stream_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -524,7 +653,7 @@ impl Muxer {
|
||||
let mut total_rx = 0;
|
||||
let mut legs_info = Vec::new();
|
||||
|
||||
let cached_legs = self.active_legs_cache.read().unwrap().clone();
|
||||
let cached_legs = self.active_legs_cache.load_full();
|
||||
for leg in cached_legs.iter() {
|
||||
let tx = leg.stats.tx_bytes.load(Ordering::Relaxed);
|
||||
let rx = leg.stats.rx_bytes.load(Ordering::Relaxed);
|
||||
@@ -594,14 +723,32 @@ impl Muxer {
|
||||
Self::format_size(rx)
|
||||
));
|
||||
}
|
||||
|
||||
// ── Pipeline health counters (cumulative) ────────────────────────────
|
||||
let c = &DIAG_COUNTERS;
|
||||
out.push_str(&format!(
|
||||
"├─ 📥 Mux dispatch: ok={} no_stream={} full_closed={} recv_closed={}\n",
|
||||
c.mux_dispatch_ok.load(Ordering::Relaxed),
|
||||
c.mux_dispatch_no_stream.load(Ordering::Relaxed),
|
||||
c.mux_dispatch_full_closed.load(Ordering::Relaxed),
|
||||
c.mux_dispatch_recv_closed.load(Ordering::Relaxed),
|
||||
));
|
||||
out.push_str(&format!(
|
||||
"└─ 📤 Upload/legs: upload_fails={} ctrl_full_drops={} write_stalls={} leg_disconnects={}",
|
||||
c.upload_fails.load(Ordering::Relaxed),
|
||||
c.control_full_drops.load(Ordering::Relaxed),
|
||||
c.tunnel_write_stalls.load(Ordering::Relaxed),
|
||||
c.leg_disconnects.load(Ordering::Relaxed),
|
||||
));
|
||||
|
||||
info!("\n{}", out);
|
||||
}
|
||||
|
||||
/// Collect a point-in-time snapshot of tunnel metrics for diagnostics.
|
||||
/// Lock-free: reads only atomics and the RwLock-protected legs cache.
|
||||
/// Lock-free: reads only atomics and the ArcSwap-backed legs cache.
|
||||
pub fn snapshot_tunnel_metrics(&self) -> TunnelMetrics {
|
||||
let global_min_rtt = crate::net::GLOBAL_MIN_RTT.load(Ordering::Relaxed);
|
||||
let cached_legs = self.active_legs_cache.read().unwrap().clone();
|
||||
let cached_legs = self.active_legs_cache.load_full();
|
||||
|
||||
let active_legs: Vec<LegMetrics> = cached_legs
|
||||
.iter()
|
||||
|
||||
@@ -21,10 +21,20 @@ pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
/// If the app's receive buffer stays full longer than this, the connection
|
||||
/// is closed to unblock the tunnel leg for other streams.
|
||||
pub const BRIDGE_STREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
/// Max time dispatch_to_local will block waiting for a stream's receive channel.
|
||||
/// Protects the engine reader (and thus the entire tunnel leg) from being stuck
|
||||
/// behind one slow stream's backlog. On timeout the stream is forcibly closed.
|
||||
pub const DISPATCH_TO_LOCAL_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
/// While *every* tunnel leg is momentarily down (all reconnecting), an upload
|
||||
/// stream holds its current chunk and retries instead of closing — turning a
|
||||
/// leg outage into a short pause rather than a mass stream reset. This bounds
|
||||
/// how long a stream will wait before it finally gives up and closes.
|
||||
pub const STREAM_PAUSE_BUDGET: Duration = Duration::from_secs(30);
|
||||
/// Poll interval while a paused upload stream waits for a leg to come back.
|
||||
pub const STREAM_PAUSE_RETRY: Duration = Duration::from_millis(250);
|
||||
/// Grace window dispatch_to_local waits when a stream's receive channel is full
|
||||
/// before closing that ONE stream. The hot path now uses try_send (no await), so
|
||||
/// this applies only to a genuinely backed-up consumer; kept short so a slow or
|
||||
/// dead stream (e.g. a finished speedtest socket the app stopped reading) can
|
||||
/// never head-of-line-block the shared per-leg reader and freeze every other
|
||||
/// download on that leg (was 10 s — caused multi-second download stalls).
|
||||
pub const DISPATCH_TO_LOCAL_TIMEOUT: Duration = Duration::from_millis(300);
|
||||
pub const TLS_HELLO_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
pub const SECURE_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
pub const FALLBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
@@ -79,16 +89,24 @@ pub const TUNNEL_MAX_BUFFER_SIZE: usize = 1024 * 1024;
|
||||
pub const TUNNEL_READ_RESERVE: usize = 16 * 1024;
|
||||
/// Maximum bytes written per stream in a single interleaved write pass.
|
||||
pub const TUNNEL_INTERLEAVE_CHUNK: usize = 16 * 1024;
|
||||
/// Max bytes a stream bridge reads per pass before producing a data message.
|
||||
/// Bounds the size of a single MuxMessage so the per-leg queue is byte-bounded
|
||||
/// (CHANNEL_PACKETS × this), keeping post-speedtest bufferbloat small. One NRXP
|
||||
/// frame is 16 KB, so reading in 16 KB units also aligns with the wire framing.
|
||||
pub const BRIDGE_READ_CHUNK: usize = 16 * 1024;
|
||||
|
||||
// ── Tunnel leg TCP socket tuning ─────────────────────────────────────────────
|
||||
/// OS-level TCP send buffer for each tunnel leg. The default (4–8 MB on
|
||||
/// Linux/Android) can hold seconds of data at typical mobile speeds, causing
|
||||
/// severe jitter. 256 KB limits extra queuing to ~80 ms at 25 Mbit/s per leg
|
||||
/// while still providing enough headroom for TCP slow-start.
|
||||
pub const TUNNEL_SOCKET_SNDBUF: u32 = 256 * 1024;
|
||||
/// severe jitter. 128 KB limits extra queuing to ~40 ms at 25 Mbit/s per leg
|
||||
/// while still providing enough headroom for TCP slow-start. (Halved from
|
||||
/// 256 KB to cut post-speedtest bufferbloat — see CHANNEL_PACKETS.)
|
||||
pub const TUNNEL_SOCKET_SNDBUF: u32 = 128 * 1024;
|
||||
/// OS-level TCP receive buffer for each tunnel leg. Larger than the send
|
||||
/// buffer so the receiver can absorb bursts without dropping packets.
|
||||
pub const TUNNEL_SOCKET_RCVBUF: u32 = 512 * 1024;
|
||||
/// buffer so the receiver can absorb bursts without dropping packets, but
|
||||
/// bounded to keep stale in-flight download data (for already-closed streams)
|
||||
/// small so the tunnel recovers in ~1 s after a heavy download.
|
||||
pub const TUNNEL_SOCKET_RCVBUF: u32 = 256 * 1024;
|
||||
|
||||
// ── Smoltcp socket defaults ──────────────────────────────────────────────────
|
||||
/// Packet slots for the ICMP socket's RX and TX packet buffers.
|
||||
|
||||
@@ -166,6 +166,15 @@ pub struct DiagnosticsCounters {
|
||||
pub control_full_drops: AtomicU64,
|
||||
pub tunnel_write_stalls: AtomicU64,
|
||||
pub stream_errors: AtomicU64,
|
||||
// ── Download dispatch funnel (muxer.dispatch_to_local) ──────────────────
|
||||
/// Frames handed to a local stream's channel successfully.
|
||||
pub mux_dispatch_ok: AtomicU64,
|
||||
/// Frames dropped because no stream is registered for that id (gone/unknown).
|
||||
pub mux_dispatch_no_stream: AtomicU64,
|
||||
/// Streams closed because their channel stayed full past the grace window.
|
||||
pub mux_dispatch_full_closed: AtomicU64,
|
||||
/// Frames dropped because the stream's receiver was already closed.
|
||||
pub mux_dispatch_recv_closed: AtomicU64,
|
||||
}
|
||||
|
||||
impl DiagnosticsCounters {
|
||||
@@ -177,6 +186,10 @@ impl DiagnosticsCounters {
|
||||
control_full_drops: AtomicU64::new(0),
|
||||
tunnel_write_stalls: AtomicU64::new(0),
|
||||
stream_errors: AtomicU64::new(0),
|
||||
mux_dispatch_ok: AtomicU64::new(0),
|
||||
mux_dispatch_no_stream: AtomicU64::new(0),
|
||||
mux_dispatch_full_closed: AtomicU64::new(0),
|
||||
mux_dispatch_recv_closed: AtomicU64::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user