nineap/netrunner-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bufferbloat fixes, deadlocks and buf sizes change

Browse Source
This commit is contained in:
Kirill
2026-04-16 13:14:16 +07:00
parent f5f1d20f54
commit 2d70f33257
12 changed files with 191 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/config.rs
+16 -11
View File
@@ -30,25 +30,30 @@ impl NetworkConfig {
pub fn new(system_mtu: usize) -> Self {
Self {
mtu: system_mtu,
connection_buf_size: 1024 * 1024 * 4,
tcp_buffer_size: 1024 * 1024,
udp_buffer_size: 64 * 1024,
// Для сервера 64KB ок, для клиента поднимем до 256KB, чтобы не тормозить чтение с TUN
connection_buf_size: 256 * 1024,
// 256KB — это золотая середина. Позволяет держать ~10Мбит на стрим при пинге 200мс.
tcp_buffer_size: 256 * 1024,
udp_buffer_size: 128 * 1024,
tcp_chunk_size: system_mtu - 100,
// 🔥 Канал расширен для сглаживания микро-обрывов связи
channel_capacity: 2048,
// Емкость каналов: 512 пакетов (~700КБ).
// Этого достаточно, чтобы сгладить лаги радио-эфира на телефоне.
channel_capacity: 128,
// Расширяем TCP окна под BBR
// Окна smoltcp (важно для Download)
// Увеличиваем до 512KB для тяжелых профилей
tcp_rx_heavy: 256 * 1024,
tcp_tx_heavy: 256 * 1024,
tcp_rx_light: 32 * 1024,
tcp_tx_light: 32 * 1024,
tcp_rx_light: 64 * 1024,
tcp_tx_light: 64 * 1024,
udp_buf_heavy: 256 * 1024,
udp_meta_heavy: 512,
udp_buf_light: 16 * 1024,
udp_meta_light: 32,
udp_meta_heavy: 1024, // Больше метаданных для мелких UDP пакетов
udp_buf_light: 32 * 1024,
udp_meta_light: 64,
}
}
core/src/net/connection/bridge.rs
+39 -74
View File
@@ -21,7 +21,6 @@ impl Drop for StreamGuard {
self.muxer.remove_stream(self.stream_id);
}
}
pub(crate) async fn run_tcp_bridge<R, W>(
stream_id: u32,
mut reader: R,
@@ -36,86 +35,52 @@ pub(crate) async fn run_tcp_bridge<R, W>(
stream_id,
muxer: muxer.clone(),
};
let buf_size = NetworkConfig::global().tcp_buffer_size;
let mut buf = BytesMut::with_capacity(NetworkConfig::global().tcp_buffer_size);
// Создаем отдельный канал для упорядоченной отправки в туннель
let (tx_to_mux, mut rx_from_bridge) = mpsc::channel::<Bytes>(16);
loop {
buf.reserve(NetworkConfig::global().tcp_buffer_size);
let select_res = timeout(BRIDGE_IDLE_TIMEOUT, async {
tokio::select! {
res = reader.read_buf(&mut buf) => {
match res {
Ok(0) => {
debug!(stream_id, "TCP Socket reached EOF");
return Ok(false);
}
Ok(_) => {
let msg = MuxMessage {
stream_id,
frame_type: FrameType::Data,
data: buf.split().freeze(),
};
if muxer.send_to_network(msg).await.is_err() {
// 🔥 ФИКС: Если нет живых ног туннеля — просто дропаем пакет!
// Мы НЕ закрываем мост (return Ok(false)),
// TCP-стек на клиенте сам сделает ретрансмиссию
warn!(stream_id, "All tunnel legs dead. Dropping packet. TCP will retransmit.");
}
Ok(true)
}
Err(e) => {
error!(stream_id, error = %e, "TCP Socket read error");
Err(e.to_string())
}
}
}
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
if data.is_empty() { return Ok(true); }
if let Err(e) = writer.write_all(&data).await {
error!(stream_id, error = %e, "TCP Socket write error");
return Err(e.to_string());
}
Ok(true)
}
None => {
debug!(stream_id, "Virtual channel closed (Muxer removed stream)");
Ok(false)
}
}
}
}
})
.await;
match select_res {
Ok(Ok(true)) => continue,
Ok(Ok(false)) => break,
Ok(Err(e)) => {
debug!(stream_id, "Bridge closing due to error: {}", e);
break;
}
Err(_) => {
warn!(stream_id, "TCP Bridge IDLE timeout reached. Evicting task.");
// Задача-отправщик: гарантирует порядок и не блокирует основной цикл моста
let m_clone = muxer.clone();
tokio::spawn(async move {
while let Some(data) = rx_from_bridge.recv().await {
if let Err(_) = m_clone.send_data_safe(stream_id, data, false).await {
break;
}
}
});
let mut buf = BytesMut::with_capacity(buf_size);
loop {
if buf.capacity() < 16384 {
buf.reserve(buf_size);
}
tokio::select! {
// Читаем из Интернета -> В очередь отправки (Upload)
res = reader.read_buf(&mut buf) => {
match res {
Ok(0) => break,
Ok(_) => {
let data = buf.split().freeze();
if tx_to_mux.send(data).await.is_err() { break; }
}
Err(_) => break,
}
}
// Читаем из Туннеля -> В Интернет (Download)
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
if data.is_empty() { continue; }
if writer.write_all(&data).await.is_err() { break; }
}
None => break,
}
}
}
}
let _ = muxer
.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
})
.await;
tokio::time::sleep(Duration::from_millis(50)).await;
}
pub(crate) async fn run_udp_bridge(
stream_id: u32,
socket: UdpSocket,
core/src/net/connection/connection.rs
+26 -13
View File
@@ -12,11 +12,11 @@ use crate::{
MAX_TUNNEL_LEGS, SECURE_HANDSHAKE_TIMEOUT, STEALTH_FALLBACK_HOST, TLS_HELLO_TIMEOUT,
TOPOLOGY_PRINT_INTERVAL,
},
nrxp::{Codec, ErrorAction, Frame, FrameType, TlsBridge},
nrxp::{Codec, Frame, FrameType, TlsBridge},
rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame},
tlseng::{BrowserProfile, ServerProfile},
};
use bytes::{Buf, Bytes, BytesMut};
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use netrunner_logger::{debug, error, info, warn};
use rand::Rng;
@@ -173,7 +173,6 @@ impl ClientHandler {
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
let (data_tx, data_rx) = mpsc::channel::<MuxMessage>(cap);
// Клонируем Sender до передачи в add_leg, чтобы использовать для remove_leg
let control_tx_clone = control_tx.clone();
muxer.add_leg(leg_id, control_tx, data_tx);
@@ -193,8 +192,6 @@ impl ClientHandler {
let run_result = engine.run().await;
// 🔥 ФИКС: Гарантированно вычищаем ногу по завершению работы,
// используя безопасный same_channel чек
muxer.remove_leg(leg_id, &control_tx_clone);
run_result.map_err(|e| e.to_string())?;
@@ -219,7 +216,6 @@ impl ClientHandler {
loop {
interval.tick().await;
let current_ip = Self::get_local_ip();
// 🔥 ФИКС: Trigger also if we completely lost network (current_ip == None)
if current_ip != last_ip {
netrunner_logger::warn!(
"🌐 Network Change Detected: {:?} -> {:?}",
@@ -287,10 +283,13 @@ impl ClientHandler {
let tx_to_tun = tx_to_engine.clone();
let reg = registry.clone();
let l2g = local_to_global.clone(); // 🔥 Клонируем для очистки
tokio::spawn(async move {
while let Some(back_payload) = v_rx.recv().await {
if let Some(r) = reg.get(&global_stream_id) {
let (orig_local_id, ip, port, proto) = *r;
let route_info = reg.get(&global_stream_id).map(|r| *r);
if let Some((orig_local_id, ip, port, proto)) = route_info {
let out_f_type = if proto == LocalProtocol::Udp {
FrameType::UdpData
} else {
@@ -311,16 +310,31 @@ impl ClientHandler {
}
}
}
// 🔥 ФИКС УТЕЧКИ ПАМЯТИ: Сборщик мусора
// Если цикл завершился (Muxer удалил v_tx), стираем мертвые сессии
if let Some((_, (orig_local_id, _, _, _))) =
reg.remove(&global_stream_id)
{
l2g.remove(&orig_local_id);
debug!(
global_stream_id,
"🧹 Garbage Collector: Cleaned up dead registry stream"
);
}
});
let _ = muxer_inner
.send_control(global_stream_id, f_type, payload)
.await;
}
FrameType::Data | FrameType::UdpData => {
if let Some(id) = local_to_global.get(&local_socket_id) {
let global_id = local_to_global.get(&local_socket_id).map(|id| *id);
if let Some(id) = global_id {
let _ = muxer_inner
.send_data_safe(
*id,
id,
payload,
raw_frame.protocol == LocalProtocol::Udp,
)
@@ -353,10 +367,10 @@ pub struct ServerHandler {
}
impl ServerHandler {
pub fn new(connection: Connection) -> Self {
pub fn new(connection: Connection, session_manager: Arc<SessionManager>) -> Self {
Self {
conn: connection,
session_manager: Arc::new(SessionManager::new()),
session_manager,
}
}
@@ -521,7 +535,6 @@ impl TunnelHandler for ServerHandler {
let res = engine.run().await;
// 🔥 ФИКС: Безопасное удаление ноги и на сервере
muxer.remove_leg(leg_id, &control_tx_clone);
if muxer.active_legs_count() == 0 {
core/src/net/connection/engine.rs
+7 -1
View File
@@ -56,6 +56,11 @@ impl TunnelEngine {
let mut inbound = inbound;
loop {
if read_buf.is_empty() {
read_buf.clear();
}
read_buf.reserve(16384);
tokio::select! {
_ = token_reader.cancelled() => {
info!("Reader Task: Shutdown signal received.");
@@ -141,6 +146,8 @@ impl TunnelEngine {
res = writer_handle => res.unwrap_or_else(|e| Err(format!("Writer panic: {}", e))),
};
token.cancel();
if let Err(e) = &res {
error!("TunnelEngine critical failure: {}", e);
}
@@ -181,7 +188,6 @@ impl TunnelEngine {
}
for pkt in packets {
// 🔥 ФИКС: Увеличен таймаут до 10 секунд (Mobile RRC Transitions)
let write_future = outbound.write_all(&pkt);
if let Err(_) =
tokio::time::timeout(std::time::Duration::from_secs(10), write_future).await
core/src/net/connection/mod.rs
+1 -1
View File
@@ -4,5 +4,5 @@ mod engine;
mod handler;
mod muxer;
pub use connection::{ClientHandler, Connection, ServerHandler, TunnelHandler};
pub use connection::{ClientHandler, Connection, ServerHandler, SessionManager, TunnelHandler};
pub use muxer::GLOBAL_MIN_RTT;
core/src/net/connection/muxer.rs
+31 -29
View File
@@ -1,6 +1,6 @@
use bytes::Bytes;
use dashmap::DashMap;
use netrunner_logger::{debug, info, trace, warn};
use netrunner_logger::{info, trace, warn};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
@@ -54,9 +54,9 @@ impl IdGenerator {
#[derive(Clone)]
pub struct MuxMessage {
pub stream_id: u32,
pub frame_type: FrameType,
pub data: Bytes,
pub(crate) stream_id: u32,
pub(crate) frame_type: FrameType,
pub(crate) data: Bytes,
}
pub static GLOBAL_MIN_RTT: AtomicU32 = AtomicU32::new(250);
@@ -102,11 +102,13 @@ impl Muxer {
stats: Arc::new(LegStats::default()),
},
);
info!(leg_id, "MUXER: Leg registered (Total: {})", self.legs.len());
info!(
leg_id,
"MUXER: Physical TCP+TLS leg registered (Total: {})",
self.legs.len()
);
}
// 🔥 ФИКС: Умное удаление ноги. Мы проверяем, не была ли эта нога уже перезаписана
// более новым подключением (чтобы не удалить живую ногу по ошибке)
pub fn remove_leg(&self, leg_id: u32, tx: &Sender<MuxMessage>) {
let should_remove = if let Some(leg) = self.legs.get(&leg_id) {
leg.control_tx.same_channel(tx)
@@ -116,7 +118,10 @@ impl Muxer {
if should_remove {
self.legs.remove(&leg_id);
info!(leg_id, "MUXER: Leg removed safely, streams will re-balance");
info!(
leg_id,
"MUXER: TCP leg removed safely, streams will re-balance"
);
} else {
trace!(
leg_id,
@@ -126,7 +131,7 @@ impl Muxer {
}
pub fn remove_all_legs(&self) {
warn!("🚨 MUXER: Emergency reset! Removing all legs due to network change.");
warn!("🚨 MUXER: Emergency reset! Removing all physical legs due to network change.");
self.legs.clear();
self.stream_bindings.clear();
}
@@ -181,7 +186,7 @@ impl Muxer {
let rtt = start_time.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
trace!(leg_id, rtt, "💓 [Muxer] RTT updated for leg");
trace!(leg_id, rtt, "💓 [Muxer] RTT updated for physical leg");
let min_rtt = self
.legs
@@ -208,7 +213,7 @@ impl Muxer {
let (leg_id, leg) = match self.select_leg(&message.frame_type, message.stream_id) {
Some(l) => l,
None => return Err("MUXER: No active legs available".to_string()),
None => return Err("MUXER: No active physical legs available".to_string()),
};
let target_tx = match message.frame_type {
@@ -220,7 +225,7 @@ impl Muxer {
};
match tokio::time::timeout(
std::time::Duration::from_secs(2),
std::time::Duration::from_secs(10),
target_tx.send(message.clone()),
)
.await
@@ -243,7 +248,7 @@ impl Muxer {
Err(_) => {
warn!(
message.stream_id,
"Physical TX full for 2s! Leg {} is dead. Evicting.", leg_id
"Physical TCP TX full for 10s! Leg {} is dead. Evicting.", leg_id
);
self.remove_leg(leg_id, &leg.control_tx);
continue;
@@ -271,7 +276,7 @@ impl Muxer {
.await
}
pub async fn send_control(
pub(crate) async fn send_control(
&self,
stream_id: u32,
f_type: FrameType,
@@ -303,17 +308,16 @@ impl Muxer {
if let Some((tx, stats)) = stream_opt {
let size = data.len() as u64;
match tx.try_send(data) {
// 🔥 КЛЮЧЕВОЙ ФИКС: Используем .send().await
// Мы больше не выбрасываем пакеты! Мы ждем, пока освободится очередь.
// Это создает естественный TCP Backpressure на стороне сервера.
match tx.send(data).await {
Ok(_) => {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
trace!(
stream_id,
"Local RX queue full. Dropped packet for inner-TCP backpressure."
);
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
Err(_) => {
// Канал закрыт (клиент отвалился)
self.remove_stream(stream_id);
}
}
@@ -360,10 +364,10 @@ impl Muxer {
match tokio::time::timeout(crate::net::HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => {
trace!(leg_id, "✅ Leg Health Check OK");
trace!(leg_id, "✅ TCP Leg Health Check OK");
}
_ => {
warn!(leg_id, "❌ Leg Health Check FAIL/Timeout - Evicting");
warn!(leg_id, "❌ TCP Leg Health Check FAIL/Timeout - Evicting");
self.remove_leg(leg_id, &tx);
}
}
@@ -407,7 +411,6 @@ impl Muxer {
total_tx += tx;
total_rx += rx;
let leg_type = if id % 2 == 0 { "TCP" } else { "UDP" };
let rtt_str = if rtt == 0 {
"N/A".to_string()
} else {
@@ -415,9 +418,8 @@ impl Muxer {
};
legs_info.push(format!(
" ├─ Leg {} ({}) ─ ⇡ {:<9} | ⇣ {:<9} [RTT: {}]",
" ├─ Leg {} (TCP+TLS) ─ ⇡ {:<9} | ⇣ {:<9} [RTT: {}]",
id,
leg_type,
Self::format_size(tx),
Self::format_size(rx),
rtt_str
@@ -430,7 +432,7 @@ impl Muxer {
Self::format_size(total_rx)
));
out.push_str(&format!(
"├─ 🦵 Physical Legs (Active: {})\n",
"├─ 🦵 Physical Connections (Active: {})\n",
legs_info.len()
));
@@ -444,7 +446,7 @@ impl Muxer {
let streams_count = self.streams.len();
out.push_str(&format!(
"└─ 🔀 Virtual Streams (Active: {})\n",
"└─ 🔀 Virtual Streams (TCP/UDP multiplexed: {})\n",
streams_count
));
core/src/net/constants.rs
+1 -1
View File
@@ -10,7 +10,7 @@ pub const GLOBAL_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
// 🔥 ФИКС: Ускоряем обнаружение мертвой сети при переключении Wi-Fi -> LTE
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(3); // Было 7
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(4); // Было 10
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(10); // Было 10
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(2); // Было 3
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
core/src/net/mod.rs
+3 -1
View File
@@ -3,5 +3,7 @@ mod connection;
mod constants;
pub use config::NetworkConfig;
pub use connection::{ClientHandler, Connection, ServerHandler, TunnelHandler, GLOBAL_MIN_RTT};
pub use connection::{
ClientHandler, Connection, ServerHandler, SessionManager, TunnelHandler, GLOBAL_MIN_RTT,
};
pub use constants::*;