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

updates muxer legs select, buf sizes and constantsa

Browse Source
This commit is contained in:
Kirill
2026-04-11 22:48:33 +07:00
parent 6009eca8a7
commit 732f55c01d
12 changed files with 441 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/config.rs
+10 -10
View File
@@ -6,6 +6,7 @@ pub static GLOBAL_NET_CONFIG: OnceLock<NetworkConfig> = OnceLock::new();
#[derive(Debug, Clone)]
pub struct NetworkConfig {
pub mtu: usize,
pub connection_buf_size: usize,
pub tcp_buffer_size: usize,
pub udp_buffer_size: usize,
@@ -35,31 +36,30 @@ pub struct NetworkConfig {
}
impl NetworkConfig {
pub fn new(system_mtu: usize) -> Self {
Self {
mtu: system_mtu,
tcp_buffer_size: 64 * 1024, // Для чтения из физического сокета (нормально)
connection_buf_size: 1024 * 1024 * 2,
tcp_buffer_size: 512 * 1024,
udp_buffer_size: 64 * 1024,
tcp_chunk_size: 4 * 1024, // Оставляем 16 КБ
tcp_chunk_size: system_mtu - 100,
// 🔥 Зажимаем программные очереди (Убиваем Hidden Bloat)
client_muxer_capacity: 8,
client_tun_capacity: 16,
client_stream_capacity: 16,
client_virtual_stream_capacity: 128,
client_virtual_stream_capacity: 32,
server_muxer_capacity: 64, // Кардинально режем
server_muxer_capacity: 64, // Кардинально режем
server_stream_capacity: 32,
// 🔥 Расширяем TCP окна под BBR (Разблокируем Gigabit на дальние дистанции)
tcp_rx_heavy: 512 * 1024, //512
tcp_tx_heavy: 1 * 1024 * 1024, // 1 MB
tcp_rx_heavy: 256 * 1024, //512
tcp_tx_heavy: 256 * 1024, // 1 MB
tcp_rx_light: 16 * 1024,
tcp_tx_light: 64 * 1024,
tcp_rx_light: 32 * 1024,
tcp_tx_light: 32 * 1024,
udp_buf_heavy: 256 * 1024,
udp_meta_heavy: 512,
core/src/net/connection/connection.rs
+171 -55
View File
@@ -1,13 +1,20 @@
use std::{net::Ipv4Addr, sync::Arc};
use crate::{
crypto::{ChaChaCipher, SessionKeys}, net::{
FALLBACK_CONNECT_TIMEOUT, LEG_RECONNECT_DELAY, LEG_STAGGER_DELAY, MAX_TUNNEL_LEGS, NetworkConfig, SECURE_HANDSHAKE_TIMEOUT, STEALTH_FALLBACK_HOST, TLS_HELLO_TIMEOUT, TOPOLOGY_PRINT_INTERVAL, connection::{
crypto::{ChaChaCipher, SessionKeys},
net::{
connection::{
engine::TunnelEngine,
handler::{RemoteOpener, StreamHandler},
muxer::Muxer,
}
}, nrxp::{Codec, ErrorAction, Frame, FrameType, TlsBridge}, rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame}, tlseng::{BrowserProfile, ServerProfile}
},
NetworkConfig, FALLBACK_CONNECT_TIMEOUT, LEG_RECONNECT_DELAY, LEG_STAGGER_DELAY,
MAX_TUNNEL_LEGS, SECURE_HANDSHAKE_TIMEOUT, STEALTH_FALLBACK_HOST, TLS_HELLO_TIMEOUT,
TOPOLOGY_PRINT_INTERVAL,
},
nrxp::{Codec, ErrorAction, Frame, FrameType, TlsBridge},
rawcast::{LocalProtocol, RawCastAdapter, RawCastFrame},
tlseng::{BrowserProfile, ServerProfile},
};
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
@@ -29,7 +36,9 @@ pub struct SessionManager {
impl SessionManager {
pub fn new() -> Self {
Self { sessions: DashMap::new() }
Self {
sessions: DashMap::new(),
}
}
pub fn generate_id() -> String {
@@ -69,13 +78,11 @@ impl Connection {
Self {
inbound,
outbound,
read_buf: BytesMut::with_capacity(NetworkConfig::global().tcp_buffer_size),
read_buf: BytesMut::with_capacity(NetworkConfig::global().connection_buf_size),
}
}
}
type StreamContext = (u32, Ipv4Addr, u16, LocalProtocol);
pub struct ClientHandler;
impl ClientHandler {
async fn establish_leg(
@@ -84,24 +91,39 @@ impl ClientHandler {
muxer: Arc<Muxer>,
session_id: &str,
) -> Result<(), String> {
let leg_name = if leg_id % 2 == 0 { "TCP-Leg" } else { "UDP-Leg" };
let leg_name = if leg_id % 2 == 0 {
"TCP-Leg"
} else {
"UDP-Leg"
};
let mut addrs = tokio::net::lookup_host(remote_proxy_addr)
.await
.map_err(|e| format!("DNS resolution failed: {}", e))?;
let addr = addrs.next().ok_or_else(|| format!("No IPs found for {}", remote_proxy_addr))?;
let addr = addrs
.next()
.ok_or_else(|| format!("No IPs found for {}", remote_proxy_addr))?;
// --- Шаг 1: Создание сырого сокета через socket2 ---
let domain = if addr.is_ipv4() { Domain::IPV4 } else { Domain::IPV6 };
let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
.map_err(|e| e.to_string())?;
let domain = if addr.is_ipv4() {
Domain::IPV4
} else {
Domain::IPV6
};
let socket =
Socket::new(domain, Type::STREAM, Some(Protocol::TCP)).map_err(|e| e.to_string())?;
// --- Шаг 2: Настройка кроссплатформенных опций ---
socket.set_nonblocking(true).map_err(|e| e.to_string())?;
socket.set_nodelay(true).map_err(|e| e.to_string())?;
// 🔥 Шаг 3: Установка TCP_NOTSENT_LOWAT через libc (Linux, Android, iOS, macOS)
#[cfg(any(target_os = "linux", target_os = "android", target_os = "ios", target_os = "macos"))]
#[cfg(any(
target_os = "linux",
target_os = "android",
target_os = "ios",
target_os = "macos"
))]
unsafe {
use std::os::fd::AsRawFd;
@@ -114,31 +136,45 @@ impl ClientHandler {
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
);
if ret != 0 {
warn!("Could not set TCP_NOTSENT_LOWAT: {}", std::io::Error::last_os_error());
warn!(
"Could not set TCP_NOTSENT_LOWAT: {}",
std::io::Error::last_os_error()
);
}
}
// --- Шаг 4: Подключение и конвертация в TcpStream для Tokio ---
let _ = socket.connect(&addr.into());
let _ = socket.connect(&addr.into());
let std_stream: std::net::TcpStream = socket.into();
let stream = TcpStream::from_std(std_stream).map_err(|e| e.to_string())?;
let mut conn = Connection::new(stream);
let mut session_keys = SessionKeys::new(true);
let ch = TlsBridge::wrap_client_hello(&BrowserProfile::CHROME_131, "ubuntu.com", &session_keys);
let ch =
TlsBridge::wrap_client_hello(&BrowserProfile::CHROME_131, "ubuntu.com", &session_keys);
conn.outbound.write_all(&ch).await.map_err(|e| e.to_string())?;
conn.outbound
.write_all(&ch)
.await
.map_err(|e| e.to_string())?;
loop {
match TlsBridge::unpack_handshake(&mut conn.read_buf) {
Ok(Some(msg)) => {
session_keys.update_keys(msg.random(), msg.extensions(), false)
session_keys
.update_keys(msg.random(), msg.extensions(), false)
.map_err(|e| format!("Keys update error: {}", e))?;
break;
}
Ok(None) => {
let n = conn.inbound.read_buf(&mut conn.read_buf).await.map_err(|e| e.to_string())?;
if n == 0 { return Err(format!("EOF on {}", leg_name)); }
let n = conn
.inbound
.read_buf(&mut conn.read_buf)
.await
.map_err(|e| e.to_string())?;
if n == 0 {
return Err(format!("EOF on {}", leg_name));
}
}
Err(e) => return Err(format!("TLS error on {}: {:?}", leg_name, e)),
}
@@ -151,14 +187,18 @@ impl ClientHandler {
let (rx_codec, mut tx_codec) = codec.split();
let handshake_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
let encrypted_handshake = tx_codec.encode_frame(0, FrameType::Handshake, handshake_payload)
let encrypted_handshake = tx_codec
.encode_frame(0, FrameType::Handshake, handshake_payload)
.map_err(|e| format!("Failed to encrypt Handshake: {:?}", e))?;
conn.outbound.write_all(&encrypted_handshake).await.map_err(|e| e.to_string())?;
conn.outbound
.write_all(&encrypted_handshake)
.await
.map_err(|e| e.to_string())?;
let (control_tx, control_rx) = mpsc::channel(NetworkConfig::global().client_muxer_capacity);
let (data_tx, data_rx) = mpsc::channel(NetworkConfig::global().client_muxer_capacity);
muxer.add_leg(leg_id, control_tx, data_tx);
let handler = Arc::new(StreamHandler::new(muxer.clone(), None));
let engine = TunnelEngine {
leg_id,
@@ -183,7 +223,8 @@ impl ClientHandler {
) -> Result<(), String> {
let session_id = SessionManager::generate_id();
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
let registry: Arc<DashMap<u32, (u64, Ipv4Addr, u16, LocalProtocol)>> = Arc::new(DashMap::new());
let registry: Arc<DashMap<u32, (u64, Ipv4Addr, u16, LocalProtocol)>> =
Arc::new(DashMap::new());
let local_to_global: Arc<DashMap<u64, u32>> = Arc::new(DashMap::new());
for id in 0..MAX_TUNNEL_LEGS {
@@ -205,7 +246,9 @@ impl ClientHandler {
tokio::spawn(async move {
while let Some(m_stats) = m_weak.upgrade() {
tokio::time::sleep(TOPOLOGY_PRINT_INTERVAL).await;
if m_stats.active_legs_count() == 0 { break; }
if m_stats.active_legs_count() == 0 {
break;
}
m_stats.perform_health_check().await;
m_stats.print_topology_tree();
}
@@ -223,9 +266,18 @@ impl ClientHandler {
FrameType::Connect | FrameType::UdpConnect => {
let global_stream_id = muxer_inner.next_stream_id();
local_to_global.insert(local_socket_id, global_stream_id);
registry.insert(global_stream_id, (local_socket_id, raw_frame.dst_ip, raw_frame.dst_port, raw_frame.protocol));
registry.insert(
global_stream_id,
(
local_socket_id,
raw_frame.dst_ip,
raw_frame.dst_port,
raw_frame.protocol,
),
);
let (v_tx, mut v_rx) = mpsc::channel(NetworkConfig::global().client_stream_capacity);
let (v_tx, mut v_rx) =
mpsc::channel(NetworkConfig::global().client_stream_capacity);
muxer_inner.register_stream(global_stream_id, v_tx);
let tx_to_tun = tx_to_engine.clone();
@@ -234,25 +286,48 @@ impl ClientHandler {
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 out_f_type = if proto == LocalProtocol::Udp { FrameType::UdpData } else { FrameType::Data };
let mock_nrxp = Frame::new(orig_local_id as u32, out_f_type, back_payload);
if let Ok(raw) = RawCastAdapter::from_nrxp(mock_nrxp, ip, port, proto == LocalProtocol::Udp) {
let out_f_type = if proto == LocalProtocol::Udp {
FrameType::UdpData
} else {
FrameType::Data
};
let mock_nrxp = Frame::new(
orig_local_id as u32,
out_f_type,
back_payload,
);
if let Ok(raw) = RawCastAdapter::from_nrxp(
mock_nrxp,
ip,
port,
proto == LocalProtocol::Udp,
) {
let _ = tx_to_tun.send(raw).await;
}
}
}
});
let _ = muxer_inner.send_control(global_stream_id, f_type, payload).await;
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 _ = muxer_inner.send_data_safe(*id, payload, raw_frame.protocol == LocalProtocol::Udp).await;
let _ = muxer_inner
.send_data_safe(
*id,
payload,
raw_frame.protocol == LocalProtocol::Udp,
)
.await;
}
}
FrameType::Close => {
if let Some(kv) = local_to_global.remove(&local_socket_id) {
let global_stream_id = kv.1;
let _ = muxer_inner.send_control(global_stream_id, FrameType::Close, Bytes::new()).await;
let _ = muxer_inner
.send_control(global_stream_id, FrameType::Close, Bytes::new())
.await;
muxer_inner.remove_stream(global_stream_id);
registry.remove(&global_stream_id);
}
@@ -273,12 +348,23 @@ pub struct ServerHandler {
impl ServerHandler {
pub fn new(connection: Connection) -> Self {
Self { conn: connection, session_manager: Arc::new(SessionManager::new()) }
Self {
conn: connection,
session_manager: Arc::new(SessionManager::new()),
}
}
async fn handle_stealth_fallback(mut client_inbound: OwnedReadHalf, mut client_outbound: OwnedWriteHalf, initial_data: Bytes) {
async fn handle_stealth_fallback(
mut client_inbound: OwnedReadHalf,
mut client_outbound: OwnedWriteHalf,
initial_data: Bytes,
) {
info!(target = %STEALTH_FALLBACK_HOST, "Stealth fallback: bridging to Target");
let target_stream = tokio::time::timeout(FALLBACK_CONNECT_TIMEOUT, TcpStream::connect(STEALTH_FALLBACK_HOST)).await;
let target_stream = tokio::time::timeout(
FALLBACK_CONNECT_TIMEOUT,
TcpStream::connect(STEALTH_FALLBACK_HOST),
)
.await;
if let Ok(Ok(target_server)) = target_stream {
let (mut server_read, mut server_write) = target_server.into_split();
@@ -288,7 +374,8 @@ impl ServerHandler {
let _ = tokio::io::copy_bidirectional(
&mut tokio::io::join(&mut client_inbound, &mut client_outbound),
&mut tokio::io::join(&mut server_read, &mut server_write),
).await;
)
.await;
}
}
}
@@ -298,7 +385,11 @@ impl TunnelHandler for ServerHandler {
async fn run(self) -> Result<(), String> {
info!("Acting as TLS Server with Stealth Fallback");
let Connection { mut inbound, mut outbound, mut read_buf } = self.conn;
let Connection {
mut inbound,
mut outbound,
mut read_buf,
} = self.conn;
let mut session_keys = SessionKeys::new(false);
// --- PHASE 1: TLS Hello & Protocol Identification ---
@@ -308,11 +399,16 @@ impl TunnelHandler for ServerHandler {
match TlsBridge::unpack_handshake(&mut read_buf) {
Ok(Some(client_msg)) => {
info!("✅ Valid Netrunner ClientHello detected");
match TlsBridge::wrap_server_hello(&client_msg, &mut session_keys, &ServerProfile::MODERN) {
match TlsBridge::wrap_server_hello(
&client_msg,
&mut session_keys,
&ServerProfile::MODERN,
) {
Ok(sh) => break sh,
Err(e) => {
if e.execute_strategy() == ErrorAction::Redirect {
Self::handle_stealth_fallback(inbound, outbound, buf_snapshot).await;
Self::handle_stealth_fallback(inbound, outbound, buf_snapshot)
.await;
return Ok(());
}
return Err("ServerHello Generation Failed".into());
@@ -320,7 +416,9 @@ impl TunnelHandler for ServerHandler {
}
}
Ok(None) => {
let res = tokio::time::timeout(TLS_HELLO_TIMEOUT, inbound.read_buf(&mut read_buf)).await;
let res =
tokio::time::timeout(TLS_HELLO_TIMEOUT, inbound.read_buf(&mut read_buf))
.await;
match res {
Ok(Ok(0)) => return Err("Client closed".into()),
Ok(Ok(_)) => continue,
@@ -342,9 +440,12 @@ impl TunnelHandler for ServerHandler {
};
// --- PHASE 2: Send Server Hello ---
outbound.write_all(&hello).await.map_err(|e| e.to_string())?;
outbound
.write_all(&hello)
.await
.map_err(|e| e.to_string())?;
// --- PHASE 3: Secure Handshake & Engine Startup ---
// --- PHASE 3: Secure Handshake & Engine Startup ---
let (tx_key, tx_iv, rx_key, rx_iv) = session_keys.get_aead_parameters();
let mut cipher = ChaChaCipher::new();
cipher.set_keys(tx_key, tx_iv, rx_key, rx_iv);
@@ -357,11 +458,17 @@ impl TunnelHandler for ServerHandler {
match rx_codec.decode_inbound(&mut read_buf) {
Ok(Some(frame)) => {
if frame.header.frame_type == FrameType::Handshake {
let parts: Vec<&str> = std::str::from_utf8(&frame.payload).unwrap_or("").split(':').collect();
let parts: Vec<&str> = std::str::from_utf8(&frame.payload)
.unwrap_or("")
.split(':')
.collect();
if parts.len() == 2 {
let sid = parts[0].to_string();
let lid: u32 = parts[1].parse().unwrap_or(0);
info!("🤝 Secure Handshake verified! Session: {}, Leg: {}", sid, lid);
info!(
"🤝 Secure Handshake verified! Session: {}, Leg: {}",
sid, lid
);
break (sid, lid);
}
}
@@ -369,11 +476,16 @@ impl TunnelHandler for ServerHandler {
}
Ok(None) => {
// Ждем новых данных из сети
let n = tokio::time::timeout(SECURE_HANDSHAKE_TIMEOUT, inbound.read_buf(&mut read_buf)).await
.map_err(|_| "Timeout waiting for Handshake")?.map_err(|e| e.to_string())?;
let n = tokio::time::timeout(
SECURE_HANDSHAKE_TIMEOUT,
inbound.read_buf(&mut read_buf),
)
.await
.map_err(|_| "Timeout waiting for Handshake")?
.map_err(|e| e.to_string())?;
if n == 0 {
return Err("Client closed connection before Handshake".into());
if n == 0 {
return Err("Client closed connection before Handshake".into());
}
}
Err(e) => {
@@ -390,7 +502,9 @@ impl TunnelHandler for ServerHandler {
let (data_tx, data_rx) = mpsc::channel(NetworkConfig::global().server_muxer_capacity);
muxer.add_leg(leg_id, control_tx, data_tx);
let opener = Arc::new(RemoteOpener { muxer: muxer.clone() });
let opener = Arc::new(RemoteOpener {
muxer: muxer.clone(),
});
let handler = Arc::new(StreamHandler::new(muxer.clone(), Some(opener)));
let engine = TunnelEngine {
@@ -408,7 +522,9 @@ impl TunnelHandler for ServerHandler {
let res = engine.run().await;
muxer.remove_leg(leg_id);
if muxer.active_legs_count() == 0 { self.session_manager.remove(&session_id); }
if muxer.active_legs_count() == 0 {
self.session_manager.remove(&session_id);
}
res
}
}
}
core/src/net/connection/engine.rs
+22 -21
View File
@@ -36,12 +36,11 @@ impl TunnelEngine {
let outbound = self.outbound;
let read_buf = self.read_buf;
// Распаковываем наши кодеки
let mut rx_codec = self.rx_codec;
let mut tx_codec = self.tx_codec;
let control_rx = self.control_rx;
let data_rx = self.data_rx;
let mut control_rx = self.control_rx;
let mut data_rx = self.data_rx;
let handler = self.handler;
let leg_id = self.leg_id;
@@ -78,7 +77,6 @@ impl TunnelEngine {
let mut frames = Vec::new();
// Чтение из RxCodec без блокировок!
loop {
match rx_codec.decode_inbound(&mut read_buf) {
Ok(Some(frame)) => frames.push(frame),
@@ -98,10 +96,15 @@ impl TunnelEngine {
}
for frame in frames {
if frame.header.frame_type == FrameType::Heartbeat {
// ИСПРАВЛЕНИЕ: Перехватываем PONG для замера RTT, но НЕ ДЕЛАЕМ continue,
// чтобы фрейм дошел до хендлера (если нужно)
if frame.header.frame_type == FrameType::Handshake && frame.payload.as_ref() == b"PONG" {
muxer.record_pong(leg_id).await;
} else if frame.header.frame_type == FrameType::Heartbeat {
// Хартбиты тоже можем использовать для RTT, если они двусторонние
muxer.record_pong(leg_id).await;
continue;
}
handler.handle(frame).await;
}
}
@@ -112,8 +115,6 @@ impl TunnelEngine {
let writer_handle = tokio::spawn(async move {
let mut outbound = outbound;
let mut control_rx = control_rx;
let mut data_rx = data_rx;
let mut heartbeat = tokio::time::interval(HEALTH_CHECK_INTERVAL);
loop {
@@ -174,18 +175,10 @@ impl TunnelEngine {
let mut packets = Vec::new();
if frame_type == FrameType::UdpData {
match tx_codec.encode_frame(stream_id, frame_type.clone(), data) {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for UDP datagram");
return Err(format!("Encryption error: {:?}", e));
}
}
} else {
// ИСПРАВЛЕНИЕ: Чанкуем ТОЛЬКО Data. Контрольные фреймы отправляем целиком.
if frame_type == FrameType::Data {
while !data.is_empty() {
let chunk_size =
std::cmp::min(data.len(), NetworkConfig::global().tcp_chunk_size);
let chunk_size = std::cmp::min(data.len(), NetworkConfig::global().tcp_chunk_size);
let chunk = data.split_to(chunk_size);
match tx_codec.encode_frame(stream_id, frame_type.clone(), chunk) {
@@ -196,6 +189,15 @@ impl TunnelEngine {
}
}
}
} else {
// Для UdpData, Connect, Handshake, Heartbeat, Close чанкование запрещено
match tx_codec.encode_frame(stream_id, frame_type.clone(), data) {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for control/udp frame");
return Err(format!("Encryption error: {:?}", e));
}
}
}
for pkt in packets {
@@ -205,7 +207,6 @@ impl TunnelEngine {
})?;
}
debug!(stream_id, "Outbound packet sent successfully");
Ok(())
}
}
}
core/src/net/connection/handler.rs
+23 -6
View File
@@ -39,7 +39,9 @@ impl RemoteOpener {
stream_id,
"❌ [Remote] Target connection failed: {}", target
);
let _ = muxer.send_control(stream_id, FrameType::Close, Bytes::new());
let _ = muxer
.send_control(stream_id, FrameType::Close, Bytes::new())
.await;
}
}
muxer.remove_stream(stream_id);
@@ -76,9 +78,6 @@ impl StreamHandler {
match frame.header.frame_type {
FrameType::Heartbeat => {
// Проверяем: если мы Opener (т.е. мы Сервер), то отвечаем.
// Если мы Клиент (opener == None), то мы сюда вообще не должны дойти
// (т.к. перехватили в TunnelEngine), но на всякий случай логируем.
if self.opener.is_some() {
trace!(stream_id, "💓 [Server] Ping received, sending Pong");
let _ = self
@@ -86,7 +85,23 @@ impl StreamHandler {
.send_control(stream_id, FrameType::Heartbeat, Bytes::new())
.await;
} else {
trace!(stream_id, "💓 [Client] Pong received (captured by Engine)");
trace!(stream_id, "💓 [Client] Pong received");
}
}
// ИСПРАВЛЕНИЕ: Добавлена обработка PING/PONG для Health Check
FrameType::Handshake => {
if frame.payload.as_ref() == b"PING" {
trace!(
stream_id,
"🤝 [Tunnel] Health Check PING received, replying PONG"
);
let _ = self
.muxer
.send_control(stream_id, FrameType::Handshake, Bytes::from("PONG"))
.await;
} else if frame.payload.as_ref() == b"PONG" {
trace!(stream_id, "🤝 [Tunnel] Health Check PONG received");
}
}
@@ -111,6 +126,7 @@ impl StreamHandler {
_ => debug!(stream_id, "Unhandled frame: {:?}", frame.header.frame_type),
}
}
async fn handle_conn_request(&self, stream_id: u32, payload: Bytes, is_udp: bool) {
let target = String::from_utf8_lossy(&payload).to_string();
@@ -137,7 +153,8 @@ impl StreamHandler {
let _ = self
.muxer
.send_control(stream_id, FrameType::Close, Bytes::new());
.send_control(stream_id, FrameType::Close, Bytes::new())
.await;
}
}
}
core/src/net/connection/muxer.rs
+54 -45
View File
@@ -34,12 +34,12 @@ impl MuxLeg {
fn congestion_factor(&self) -> f64 {
let cap = self.data_tx.capacity() as f64;
let max = crate::net::NetworkConfig::global().client_muxer_capacity as f64;
// Защита от деления на ноль, если конфигурация задана криво
if max <= 0.0 {
return 0.0;
}
1.0 - (cap / max)
}
}
@@ -114,7 +114,8 @@ impl Muxer {
pub fn remove_leg(&self, leg_id: u32) {
self.legs.remove(&leg_id);
// Удаляем все привязки стримов к этой ноге, чтобы они перебалансировались
self.stream_bindings.retain(|_, target_leg| *target_leg != leg_id);
self.stream_bindings
.retain(|_, target_leg| *target_leg != leg_id);
info!(leg_id, "MUXER: Leg removed and bindings cleared");
}
@@ -127,7 +128,6 @@ impl Muxer {
return None;
}
// 1. Ищем существующую привязку (Sticky Session)
if let Some(leg_id_ref) = self.stream_bindings.get(&stream_id) {
let leg_id = *leg_id_ref;
if let Some(leg) = self.legs.get(&leg_id) {
@@ -135,18 +135,15 @@ impl Muxer {
}
}
// 2. Если привязки нет — выбираем новую ногу
let is_udp = matches!(frame_type, FrameType::UdpData | FrameType::UdpConnect);
let mut candidates: Vec<(u32, MuxLeg)> = self
.legs
.iter()
.map(|kv| (*kv.key(), kv.value().clone()))
// Фильтруем по типу: четные ID — TCP, нечетные — UDP
.filter(|(id, _)| if is_udp { id % 2 != 0 } else { id % 2 == 0 })
.collect();
// Fallback: если подходящих по типу ног нет, берем любые
if candidates.is_empty() {
candidates = self
.legs
@@ -155,26 +152,24 @@ impl Muxer {
.collect();
}
// 🔥 Умная балансировка: RTT + Штраф за забитость канала
candidates.sort_by(|(_, leg_a), (_, leg_b)| {
let rtt_a = leg_a.stats.rtt_ms.load(Ordering::Relaxed) as f64;
let rtt_b = leg_b.stats.rtt_ms.load(Ordering::Relaxed) as f64;
let score_a = rtt_a + (leg_a.congestion_factor() * 500.0);
let score_b = rtt_b + (leg_b.congestion_factor() * 500.0);
score_a.partial_cmp(&score_b).unwrap_or(std::cmp::Ordering::Equal)
let score_a = rtt_a + (leg_a.congestion_factor() * 2000.0);
let score_b = rtt_b + (leg_b.congestion_factor() * 2000.0);
score_a
.partial_cmp(&score_b)
.unwrap_or(std::cmp::Ordering::Equal)
});
// ИСПРАВЛЕНИЕ: Убрал агрессивную блокировку "в 2 раза быстрее",
// из-за которой вторая нога голодала и отваливалась по таймаутам.
// Теперь балансировка полагается на congestion_factor и скоринг плавно.
let pool_size = std::cmp::min(candidates.len(), MUXER_POOL_SIZE);
if pool_size == 0 {
return None;
}
// Выбираем из топа лучших
let (selected_id, selected_leg) = candidates[stream_id as usize % pool_size].clone();
// Запоминаем выбор для этого стрима
self.stream_bindings.insert(stream_id, selected_id);
Some((selected_id, selected_leg))
@@ -218,7 +213,11 @@ impl Muxer {
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);
stream_ref
.value()
.1
.tx_bytes
.fetch_add(size, Ordering::Relaxed);
}
Ok(())
@@ -259,7 +258,8 @@ impl Muxer {
}
pub fn register_stream(&self, stream_id: u32, tx: Sender<Bytes>) {
self.streams.insert(stream_id, (tx, Arc::new(StreamStats::default())));
self.streams
.insert(stream_id, (tx, Arc::new(StreamStats::default())));
}
pub fn remove_stream(&self, stream_id: u32) {
@@ -269,7 +269,10 @@ impl Muxer {
}
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let stream_opt = self.streams.get(&stream_id).map(|s| (s.value().0.clone(), s.value().1.clone()));
let stream_opt = self
.streams
.get(&stream_id)
.map(|s| (s.value().0.clone(), s.value().1.clone()));
if let Some((tx, stats)) = stream_opt {
let size = data.len() as u64;
@@ -293,16 +296,17 @@ impl Muxer {
}
pub async fn perform_health_check(&self) {
let legs: Vec<(u32, Sender<MuxMessage>)> = self
.legs
.iter()
.map(|k| (*k.key(), k.value().control_tx.clone()))
.collect();
// Берем список ID заранее, чтобы не держать lock DashMap
let leg_ids: Vec<u32> = self.legs.iter().map(|kv| *kv.key()).collect();
for leg_id in leg_ids {
let Some(leg) = self.legs.get(&leg_id) else {
continue;
};
let tx = leg.control_tx.clone();
for (leg_id, tx) in legs {
let probe_stream_id = self.id_gen.next();
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::channel(2);
self.register_stream(probe_stream_id, probe_tx);
let msg = MuxMessage {
@@ -313,22 +317,27 @@ impl Muxer {
let start = std::time::Instant::now();
if tx.try_send(msg).is_ok() {
match tokio::time::timeout(HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => {
let rtt = start.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
debug!(leg_id, rtt, "✅ Leg Health Check OK");
}
}
_ => {
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(5000, Ordering::Relaxed);
warn!(leg_id, "❌ Leg Health Check Timeout");
}
// 🔥 ФИКС: Если мы даже PING не можем отправить в очередь - нога уже труп
if tx.try_send(msg).is_err() {
warn!(leg_id, "❌ MUXER: Leg queue overflow, killing leg");
self.remove_leg(leg_id);
self.remove_stream(probe_stream_id);
continue;
}
match tokio::time::timeout(HEALTH_CHECK_TIMEOUT, probe_rx.recv()).await {
Ok(Some(_)) => {
let rtt = start.elapsed().as_millis() as u32;
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(rtt, Ordering::Relaxed);
debug!(leg_id, rtt, "✅ Leg Health Check OK");
}
}
_ => {
// 🔥 ФИКС: Если таймаут - УДАЛЯЕМ НОГУ. Хватит быть зомби.
warn!(leg_id, "❌ Leg Health Check Timeout - Evicting leg");
self.remove_leg(leg_id);
}
}
self.remove_stream(probe_stream_id);
}
@@ -429,4 +438,4 @@ impl Muxer {
}
println!();
}
}
}
core/src/net/constants.rs
+4 -4
View File
@@ -9,10 +9,10 @@ pub const MUXER_POOL_SIZE: usize = 3; // Из скольких лучших Leg-
pub const TCP_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20); // Время на установку SYN/ACK
pub const UDP_IDLE_TIMEOUT: Duration = Duration::from_secs(60); // Смерть UDP сессии без данных
pub const GLOBAL_IDLE_TIMEOUT: Duration = Duration::from_secs(120); // Очистка Tracker-ом
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(10); // Частота пинга Leg-ов
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(15);
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(5); // Пауза перед реконнектом Leg
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(300); // Таймаут задач-бриджей
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(5); // Частота пинга Leg-ов
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(3);
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(3); // Пауза перед реконнектом Leg
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30); // Таймаут задач-бриджей
// --- Сетевые порты ---
pub const DNS_PORT: u16 = 53;