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

muxer cleaning update, head on line blocking fix

Browse Source
This commit is contained in:
nineAp
2026-05-03 13:41:46 +07:00
parent 7fb40f6be1
commit ae557b4848
10 changed files with 110 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/net/connection.rs
+44 -37
View File
@@ -129,12 +129,12 @@ impl TcpConnection {
true
}
fn poll_and_process(&mut self, socket: &mut tcp::Socket, timestamp: smoltcp::time::Instant) {
fn poll_and_process(&mut self, socket: &mut tcp::Socket, timestamp: smoltcp::time::Instant) {
let current_rtt = GLOBAL_MIN_RTT.load(Ordering::Relaxed);
let rtt = smoltcp::time::Duration::from_millis(current_rtt as u64);
socket.set_tunnel_rtt(rtt);
// Читаем из браузера в Туннель
// 1. Читаем из браузера -> в Туннель (Upload)
while socket.can_recv() && self.core.tx.capacity() > 0 {
if let Ok(n) = socket.peek_slice(&mut self.chunk_buf, timestamp) {
if n == 0 {
@@ -146,9 +146,7 @@ impl TcpConnection {
Ok(_) => {
socket.recv_slice(&mut self.chunk_buf[..n]).unwrap();
}
Err(mpsc::error::TrySendError::Full(_)) => {
break;
}
Err(mpsc::error::TrySendError::Full(_)) => break,
Err(mpsc::error::TrySendError::Closed(_)) => {
self.server_eof = true;
break;
@@ -160,51 +158,60 @@ impl TcpConnection {
}
if !self.server_eof {
while socket.can_send() {
// Если есть недоотправленный кусок - берем его первым
if let Some(mut chunk) = self.pending_data.pop_front() {
match socket.send_slice(&chunk) {
Ok(n) => {
self.pending_bytes -= n;
if n < chunk.len() {
chunk.advance(n);
self.pending_data.push_front(chunk);
break; // Сокет заполнился
}
}
Err(e) => {
debug!(%self.core.handle, "Smoltcp send error: {:?}", e);
self.pending_data.push_front(chunk);
break;
}
loop {
if self.pending_bytes > 256 * 1024 {
netrunner_logger::warn!(%self.core.handle, "Local stream buffer >8MB! Force closing to save tunnel.");
self.server_eof = true;
socket.close();
break;
}
match self.core.rx.try_recv() {
Ok(data) => {
self.pending_bytes += data.len();
self.pending_data.push_back(data);
}
} else {
// Очередь пуста, пытаемся прочитать из канала туннеля
match self.core.rx.try_recv() {
Ok(data) => {
self.pending_bytes += data.len();
self.pending_data.push_back(data);
}
Err(mpsc::error::TryRecvError::Empty) => break,
Err(mpsc::error::TryRecvError::Disconnected) => {
debug!(%self.core.handle, "Tunnel channel closed (Server EOF)");
self.server_eof = true;
break;
}
Err(mpsc::error::TryRecvError::Empty) => break,
Err(mpsc::error::TryRecvError::Disconnected) => {
netrunner_logger::debug!(%self.core.handle, "Tunnel channel closed (Server EOF)");
self.server_eof = true;
break;
}
}
}
}
// 3. Отправляем в smoltcp то, что накопилось (если браузер готов принять)
while socket.can_send() {
if let Some(mut chunk) = self.pending_data.pop_front() {
match socket.send_slice(&chunk) {
Ok(n) => {
self.pending_bytes -= n;
if n < chunk.len() {
chunk.advance(n);
self.pending_data.push_front(chunk);
break; // Сокет браузера переполнен, ждем следующего тика
}
}
Err(e) => {
netrunner_logger::debug!(%self.core.handle, "Smoltcp send error: {:?}", e);
self.pending_data.push_front(chunk);
break;
}
}
} else {
break;
}
}
if self.server_eof && self.pending_data.is_empty() {
let state = socket.state();
if state == tcp::State::Established || state == tcp::State::CloseWait {
debug!(%self.core.handle, "All data flushed after server EOF, sending FIN to browser");
netrunner_logger::debug!(%self.core.handle, "All data flushed, sending FIN to browser");
socket.close();
}
}
}
pub fn app_pending_out_size(&self) -> usize {
self.pending_bytes
}
client/src/net/engine.rs
+16 -3
View File
@@ -88,6 +88,9 @@ impl Engine {
let inbound_map = self.manager.tracker.inbound_tx.clone();
let mut rx_tunnel = self.rx_from_tunnel.take().unwrap();
let tunnel_waker = Arc::new(tokio::sync::Notify::new());
let waker_clone = tunnel_waker.clone();
tokio::spawn(async move {
while let Some(frame) = rx_tunnel.recv().await {
if frame.event == RawCastEvent::Close {
@@ -98,9 +101,12 @@ impl Engine {
.get(&frame.socket_id)
.map(|ref_tx| ref_tx.clone());
// 2. Теперь спокойно ждем (Backpressure), не блокируя остальную систему
// 2. Спокойно ждем (Backpressure), не блокируя остальную систему
if let Some(tx) = tx_opt {
let _ = tx.send(frame.payload).await;
if tx.send(frame.payload).await.is_ok() {
// 🔥 2. БУДИМ ОСНОВНОЙ ЦИКЛ! Данные готовы к выдаче в TUN!
waker_clone.notify_one();
}
}
}
}
@@ -126,6 +132,8 @@ impl Engine {
let delay = self
.interface
.poll_delay(Self::current_time(), &self.socket_set);
// Можно смело ставить 5ms, так как Waker разбудит нас раньше, если нужно
let sleep_time = delay
.map(|d| std::cmp::min(Duration::from_micros(d.micros()), Duration::from_millis(5)))
.unwrap_or(Duration::from_millis(5));
@@ -134,6 +142,12 @@ impl Engine {
tokio::select! {
_ = sleep(sleep_time) => {}
// 🔥 3. ЛОВИМ ПРОБУЖДЕНИЕ ОТ ТУННЕЛЯ
_ = tunnel_waker.notified() => {
// Цикл пойдет на новую итерацию, вызовет self.poll()
// и мгновенно вытащит данные из канала в TUN-интерфейс
}
msg = tun_to_engine_rx.recv() => {
if let Some(token) = msg {
self.manager.try_create_socket_from_packet(&token, &mut self.socket_set);
@@ -152,7 +166,6 @@ impl Engine {
.poll(now, &mut self.device, &mut self.socket_set)
}
// ... [Остальные методы spawn_tun_reader, spawn_tun_writer и set_route остаются без изменений] ...
fn spawn_tun_reader(
mut reader: DeviceReader,
to_engine: mpsc::UnboundedSender<TokenBuffer>,
client/src/net/socket_factory.rs
+1 -1
View File
@@ -95,7 +95,7 @@ impl SocketProvider for SmolSocketFactory {
fn reconfigure_tcp(&self, socket: &mut tcp::Socket, profile: TrafficProfile) {
socket.set_nagle_enabled(false);
socket.set_ack_delay(Some(Duration::from_millis(50)));
socket.set_ack_delay(None);
socket.set_congestion_control(CongestionControl::Bbr);
}
client/src/tun/device.rs
+3 -2
View File
@@ -4,7 +4,7 @@ use std::{
ops::{Deref, DerefMut},
sync::{
Arc, LazyLock, Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
},
time::Instant as StdInstant,
};
@@ -230,6 +230,7 @@ impl phy::RxToken for VirtRxToken<'_> {
}
pub struct VirtTxToken<'a>(&'a mut VirtTunDevice);
impl phy::TxToken for VirtTxToken<'_> {
fn consume<R, F>(self, len: usize, f: F) -> R
where
@@ -248,7 +249,7 @@ impl phy::TxToken for VirtTxToken<'_> {
GLOBAL_RX_BYTES.fetch_add(len as u64, Ordering::Relaxed);
GLOBAL_RX_PACKETS.fetch_add(1, Ordering::Relaxed);
// 🔥 ФИКС: Отправляем в безлимитный канал. Никаких дропов внутри локальной машины!
// Просто шлем, `smoltcp` сам регулирует объемы своими внутренними окнами
let _ = self.0.tx_queue.send(buffer);
result
}