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

fixed dedadlocks and buffer changes

Browse Source
This commit is contained in:
Kirill
2026-03-31 10:34:15 +07:00
parent 71acb52386
commit dadf165da6
10 changed files with 145 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/bridge.rs
+7 -9
View File
@@ -34,7 +34,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
frame_type: FrameType::Data,
data: buf.split().freeze(),
};
if muxer.send_to_network(msg).await.is_err() { break; }
if muxer.send_to_network(msg).is_err() { break; }
}
Err(e) => {
error!(stream_id, error = %e, "Socket read error");
@@ -61,13 +61,11 @@ pub(crate) async fn run_tcp_bridge<R, W>(
}
}
let _ = muxer
.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
})
.await;
let _ = muxer.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
});
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
@@ -97,7 +95,7 @@ pub(crate) async fn run_udp_bridge(
// КРИТИЧЕСКИЙ МОМЕНТ: передаем true (is_udp),
// чтобы Muxer выбрал правильную ногу для реалтайм трафика
if let Err(e) = muxer.send_data_safe(stream_id, data, true).await {
if let Err(e) = muxer.send_data_safe(stream_id, data, true) {
error!("❌ [UDP {}] Failed to send to tunnel: {}", stream_id, e);
break;
}
core/src/net/connection/connection.rs
+3 -5
View File
@@ -282,17 +282,15 @@ impl ClientHandler {
});
// Отправляем управляющий фрейм с адресом
let _ = muxer_inner.send_control(stream_id, f_type, payload).await;
let _ = muxer_inner.send_control(stream_id, f_type, payload);
}
FrameType::Data | FrameType::UdpData => {
let _ = muxer_inner.send_data_safe(stream_id, payload, is_udp).await;
let _ = muxer_inner.send_data_safe(stream_id, payload, is_udp);
}
FrameType::Close => {
let _ = muxer_inner
.send_control(stream_id, FrameType::Close, Bytes::new())
.await;
let _ = muxer_inner.send_control(stream_id, FrameType::Close, Bytes::new());
muxer_inner.remove_stream(stream_id);
}
core/src/net/connection/handler.rs
+7 -8
View File
@@ -71,8 +71,9 @@ impl StreamHandler {
let elapsed = start.elapsed();
info!(stream_id, target = %target_str, "✅ [TCP] Established in {:?}. Starting bridge.", elapsed);
let (r, w) = stream.into_split();
run_tcp_bridge(stream_id, r, w, muxer, v_rx).await;
run_tcp_bridge(stream_id, r, w, muxer.clone(), v_rx).await;
info!(stream_id, "🔚 [TCP Worker] Bridge task finished");
Self::close_stream(&muxer, stream_id).await;
}
Ok(Err(e)) => {
error!(stream_id, target = %target_str, error = %e, "❌ [TCP] Connection failed");
@@ -86,7 +87,7 @@ impl StreamHandler {
});
} else {
info!(stream_id, "📲 [TCP] Dispatching payload to local stack");
self.muxer.dispatch_to_local(stream_id, payload).await;
self.muxer.dispatch_to_local(stream_id, payload);
}
}
@@ -143,19 +144,19 @@ impl StreamHandler {
stream_id,
"📲 [UDP] Dispatching connection payload to local stack"
);
self.muxer.dispatch_to_local(stream_id, payload).await;
self.muxer.dispatch_to_local(stream_id, payload);
}
}
async fn on_data(&self, stream_id: u32, payload: Bytes) {
// Здесь info может быть избыточным при большой нагрузке, но для отладки полезно
debug!(stream_id, "📦 [TCP Data] Size: {} bytes", payload.len());
self.muxer.dispatch_to_local(stream_id, payload).await;
self.muxer.dispatch_to_local(stream_id, payload);
}
async fn on_udp_data(&self, stream_id: u32, payload: Bytes) {
debug!(stream_id, "📦 [UDP Data] Size: {} bytes", payload.len());
self.muxer.dispatch_to_local(stream_id, payload).await;
self.muxer.dispatch_to_local(stream_id, payload);
}
async fn on_close(&self, stream_id: u32) {
@@ -168,9 +169,7 @@ impl StreamHandler {
stream_id,
"📡 [Control] Sending CLOSE signal to remote peer"
);
let _ = muxer
.send_control(stream_id, FrameType::Close, Bytes::new())
.await;
let _ = muxer.send_control(stream_id, FrameType::Close, Bytes::new());
muxer.remove_stream(stream_id);
}
}
core/src/net/connection/muxer.rs
+46 -27
View File
@@ -7,7 +7,6 @@ use tokio::sync::mpsc::Sender;
use crate::net::network::NetworkConfig;
use crate::nrxp::FrameType;
// Вспомогательная структура для отдельного TLS-соединения
#[derive(Clone)]
struct MuxLeg {
control_tx: Sender<MuxMessage>,
@@ -52,11 +51,9 @@ impl Muxer {
streams: Arc::new(DashMap::new()),
id_gen: Arc::new(IdGenerator::new(is_client)),
leg_selector: Arc::new(AtomicU32::new(0)),
// leg_id_gen удален!
}
}
// ТЕПЕРЬ leg_id ПЕРЕДАЕТСЯ СНАРУЖИ
pub fn add_leg(
&self,
leg_id: u32,
@@ -82,7 +79,6 @@ impl Muxer {
self.legs.len()
}
// Возвращаем кортеж (leg_id, MuxLeg), чтобы знать, кого удалять в случае ошибки
fn select_leg(&self, frame_type: &FrameType) -> Option<(u32, MuxLeg)> {
if self.legs.is_empty() {
return None;
@@ -90,7 +86,6 @@ impl Muxer {
match frame_type {
FrameType::UdpData | FrameType::UdpConnect => {
// Пытаемся взять UDP ногу (1), если её нет — берем TCP (0) как фоллбэк
if let Some(leg) = self.legs.get(&1) {
Some((1, leg.clone()))
} else if let Some(leg) = self.legs.get(&0) {
@@ -100,7 +95,6 @@ impl Muxer {
}
}
_ => {
// Пытаемся взять TCP ногу (0), фоллбэк на UDP (1)
if let Some(leg) = self.legs.get(&0) {
Some((0, leg.clone()))
} else if let Some(leg) = self.legs.get(&1) {
@@ -112,7 +106,8 @@ impl Muxer {
}
}
pub async fn send_to_network(&self, message: MuxMessage) -> Result<(), String> {
// ТЕПЕРЬ СИНХРОННАЯ ФУНКЦИЯ БЕЗ БЛОКИРОВОК
pub fn send_to_network(&self, message: MuxMessage) -> Result<(), String> {
let (leg_id, leg) = self
.select_leg(&message.frame_type)
.ok_or_else(|| "MUXER: No active legs available".to_string())?;
@@ -125,16 +120,27 @@ impl Muxer {
_ => &leg.data_tx,
};
// Если канал мертв (TunnelEngine упал), удаляем ногу из роутинга!
if let Err(e) = target_tx.send(message).await {
self.remove_leg(leg_id);
return Err(format!("MUXER: Leg {} died during send: {}", leg_id, e));
// Используем try_send вместо send().await
match target_tx.try_send(message) {
Ok(_) => Ok(()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
// Если очередь переполнена - просто дропаем пакет!
// Возвращаем Ok(()), чтобы не убивать соединение.
netrunner_logger::warn!(
leg_id,
"MUXER: Network queue full! Dropping outbound packet."
);
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.remove_leg(leg_id);
Err(format!("MUXER: Leg {} died during send", leg_id))
}
}
Ok(())
}
pub async fn send_data_safe(
// ТЕПЕРЬ СИНХРОННАЯ ФУНКЦИЯ БЕЗ БЛОКИРОВОК
pub fn send_data_safe(
&self,
stream_id: u32,
mut data: Bytes,
@@ -153,19 +159,16 @@ impl Muxer {
self.send_to_network(MuxMessage {
stream_id,
frame_type,
frame_type: frame_type.clone(),
data: chunk,
})
.await?;
if !data.is_empty() {
tokio::task::yield_now().await;
}
})?;
// task::yield_now().await удален, так как отправка теперь мгновенная
}
Ok(())
}
pub async fn send_control(
// ТЕПЕРЬ СИНХРОННАЯ ФУНКЦИЯ БЕЗ БЛОКИРОВОК
pub fn send_control(
&self,
stream_id: u32,
f_type: FrameType,
@@ -176,7 +179,6 @@ impl Muxer {
frame_type: f_type,
data,
})
.await
}
pub fn register_stream(&self, stream_id: u32, tx: Sender<Bytes>) {
@@ -187,11 +189,28 @@ impl Muxer {
self.streams.remove(&stream_id);
}
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
if let Some(tx) = self.streams.get(&stream_id) {
if let Err(_) = tx.send(data).await {
self.remove_stream(stream_id);
// ТЕПЕРЬ СИНХРОННАЯ ФУНКЦИЯ БЕЗ БЛОКИРОВОК
pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
// Узнаем, мертв ли канал, не блокируя DashMap
let is_closed = if let Some(tx) = self.streams.get(&stream_id) {
match tx.try_send(data) {
Ok(_) => false,
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
netrunner_logger::warn!(
stream_id,
"MUXER: Local worker queue full! Dropping inbound packet."
);
false
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => true,
}
} else {
false
};
// Удаляем стрим только после того, как отпустили ссылку на DashMap
if is_closed {
self.remove_stream(stream_id);
}
}
}