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

bbr and buffers fixes

Browse Source
This commit is contained in:
Трапезников Кирилл Иванович
2026-04-10 15:00:14 +10:00
parent 11d2b4e1d9
commit a503122ace
7 changed files with 384 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/net/connection.rs
+29 -13
View File
@@ -1,4 +1,5 @@
use bytes::{Buf, Bytes, BytesMut};
use bytes::{Buf, Bytes};
use std::collections::VecDeque;
use netrunner_core::{
net::{NetworkConfig, UDP_IDLE_TIMEOUT},
rawcast::{LocalProtocol, RawCastFrame},
@@ -50,7 +51,8 @@ pub enum ConnectionState {
pub struct TcpConnection {
core: ConnectionCore<Bytes>,
state: ConnectionState,
pending_data: BytesMut,
// 🔥 ФИКС: Используем очередь чанков, чтобы не копировать память
pending_data: VecDeque<Bytes>,
handshake_rx: Option<oneshot::Receiver<()>>,
chunk_buf: Vec<u8>,
server_eof: bool,
@@ -72,7 +74,7 @@ impl TcpConnection {
let conn = Self {
core,
state: ConnectionState::Handshaking,
pending_data: BytesMut::new(),
pending_data: VecDeque::new(),
handshake_rx: Some(handshake_rx),
chunk_buf: vec![0u8; NetworkConfig::global().tcp_chunk_size],
server_eof: false,
@@ -163,13 +165,14 @@ impl TcpConnection {
if !self.server_eof {
loop {
if self.pending_data.len() >= socket.send_capacity() {
// Если буфер сокета заполнился, нет смысла выкачивать канал MPSC
if !socket.can_send() {
break;
}
match self.core.rx.try_recv() {
Ok(data) => {
self.pending_data.extend_from_slice(&data);
self.pending_data.push_back(data);
}
Err(mpsc::error::TryRecvError::Empty) => break,
Err(mpsc::error::TryRecvError::Disconnected) => {
@@ -181,14 +184,27 @@ impl TcpConnection {
}
}
if !self.pending_data.is_empty() && socket.can_send() {
match socket.send_slice(&self.pending_data) {
Ok(n) => {
self.pending_data.advance(n);
}
Err(e) => {
debug!(%self.core.handle, "Smoltcp send error: {:?}", e);
// 🔥 ФИКС: Отправляем данные кусками без копирования памяти
while socket.can_send() {
if let Some(mut chunk) = self.pending_data.pop_front() {
match socket.send_slice(&chunk) {
Ok(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;
}
}
} else {
break; // Нет больше данных для отправки
}
}
@@ -396,4 +412,4 @@ impl IcmpResponder {
}
}
}
}
}
client/src/net/engine.rs
+22 -15
View File
@@ -15,7 +15,7 @@ use std::{
time::Instant as StdInstant,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use tokio::sync::mpsc::{self, Receiver, Sender};
use tokio::time::{Duration, sleep};
use tun::{DeviceReader, DeviceWriter};
@@ -35,8 +35,8 @@ pub struct Engine {
socket_set: SocketSet<'static>,
manager: ConnectionManager,
device: VirtTunDevice,
to_smoltcp_tx: UnboundedSender<TokenBuffer>,
from_smoltcp_rx: Option<UnboundedReceiver<TokenBuffer>>,
to_smoltcp_tx: Sender<TokenBuffer>,
from_smoltcp_rx: Option<Receiver<TokenBuffer>>,
avail: Arc<AtomicBool>,
rx_from_tunnel: mpsc::Receiver<RawCastFrame>,
}
@@ -83,6 +83,7 @@ impl Engine {
let from_smoltcp_rx = self.from_smoltcp_rx.take().expect("Engine started twice");
Self::spawn_tun_writer(writer, from_smoltcp_rx);
loop {
// 1. Сначала обрабатываем всё, что накопилось в стеке
let mut repeat_poll = true;
@@ -122,16 +123,17 @@ impl Engine {
if let Some(token) = msg {
self.manager.try_create_socket_from_packet(&token, &mut self.socket_set);
// Если Muxer вернул "BUSY", smoltcp не запишет этот пакет,
// и mark_rx_available не сработает. Это ПРАВИЛЬНО.
if self.to_smoltcp_tx.send(token).is_ok() {
// 🔥 ФИКС: try_send вместо send.
// Если Muxer/smoltcp вернул "BUSY", smoltcp не запишет этот пакет.
// Очередь дропнет пакет, и ОС поймет, что нужно сбавить скорость TCP.
if self.to_smoltcp_tx.try_send(token).is_ok() {
self.device.mark_rx_available();
}
let mut count = 0;
while let Ok(token) = tun_to_engine_rx.try_recv() {
self.manager.try_create_socket_from_packet(&token, &mut self.socket_set);
if self.to_smoltcp_tx.send(token).is_ok() {
if self.to_smoltcp_tx.try_send(token).is_ok() {
self.device.mark_rx_available();
}
count += 1;
@@ -151,7 +153,7 @@ impl Engine {
fn spawn_tun_reader(
mut reader: DeviceReader,
to_engine: mpsc::Sender<TokenBuffer>, // 👈 ИЗМЕНЕНО ЗДЕСЬ
to_engine: mpsc::Sender<TokenBuffer>,
is_avail: Arc<AtomicBool>,
) {
tokio::spawn(async move {
@@ -181,19 +183,24 @@ impl Engine {
token.extend_from_slice(&buf[..n]);
// 🚨 ИСПОЛЬЗУЕМ .await. Это создаст идеальный Backpressure!
// Если Engine занят, TUN просто перестанет читать из ОС телефона!
if to_engine.send(token).await.is_ok() {
is_avail.store(true, Ordering::Release);
} else {
break;
// Если Engine занят, TUN просто перестанет читать из ОС.
match to_engine.try_send(token) {
Ok(_) => {
is_avail.store(true, Ordering::Release);
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
// Пакет отброшен (Drop). Идеальное поведение для борьбы с Bufferbloat.
}
Err(_) => break, // Канал закрыт
}
}
warn!("TUN Reader task stopped");
});
}
fn spawn_tun_writer(
mut writer: DeviceWriter,
mut from_smoltcp: UnboundedReceiver<TokenBuffer>,
mut from_smoltcp: Receiver<TokenBuffer>,
) {
tokio::spawn(async move {
debug!("TUN Writer task started");
@@ -371,4 +378,4 @@ impl EngineBuilder {
Ok((engine, tun))
}
}
}
client/src/net/socket_factory.rs
+15 -10
View File
@@ -87,21 +87,26 @@ impl SocketProvider for SmolSocketFactory {
}
fn reconfigure_tcp(&self, socket: &mut tcp::Socket, profile: TrafficProfile) {
// Выключаем Nagle везде, так как внутри Netrunner мы сами батчим данные в чанки (tcp_chunk_size),
// а BBR сам управляет плавностью отправки (Pacing). Nagle здесь будет только мешать.
socket.set_nagle_enabled(false);
match profile {
TrafficProfile::Interactive => {
socket.set_nagle_enabled(false);
TrafficProfile::Interactive | TrafficProfile::Dns => {
// Для интерактивного трафика (SSH, RDP, DNS) нам нужна минимальная задержка.
// Отвечаем ACK'ами мгновенно.
socket.set_ack_delay(None);
}
TrafficProfile::Bulk => {
socket.set_nagle_enabled(false);
socket.set_ack_delay(Some(Duration::from_millis(15)));
}
_ => {
socket.set_nagle_enabled(false);
socket.set_ack_delay(None);
TrafficProfile::Bulk | TrafficProfile::Default => {
// Для тяжелого трафика ставим крошечный delay (1 мс).
// Это снизит нагрузку на CPU (будет меньше syscall-ов на отправку голых ACK),
// но при этом 1 мс не испортит расчеты RTT и RateSample для BBR.
socket.set_ack_delay(Some(Duration::from_millis(1)));
}
}
socket.set_congestion_control(CongestionControl::Cubic);
// Включаем нашу ракету
socket.set_congestion_control(CongestionControl::Bbr);
}
fn create_udp(&self, profile: TrafficProfile) -> udp::Socket<'static> {
client/src/tun/device.rs
+12 -17
View File
@@ -79,8 +79,9 @@ impl DerefMut for TokenBuffer {
pub struct VirtTunDevice {
capabilities: DeviceCapabilities,
rx_queue: mpsc::UnboundedReceiver<TokenBuffer>,
tx_queue: mpsc::UnboundedSender<TokenBuffer>,
// 🔥 ФИКС: Используем ограниченные каналы вместо Unbounded
rx_queue: mpsc::Receiver<TokenBuffer>,
tx_queue: mpsc::Sender<TokenBuffer>,
rx_avail: Arc<AtomicBool>,
rx_bytes: u64,
@@ -102,16 +103,16 @@ impl VirtTunDevice {
capabilities: DeviceCapabilities,
) -> (
Self,
mpsc::UnboundedSender<TokenBuffer>,
mpsc::UnboundedReceiver<TokenBuffer>,
mpsc::Sender<TokenBuffer>,
mpsc::Receiver<TokenBuffer>,
Arc<AtomicBool>,
) {
let (to_smoltcp_tx, to_smoltcp_rx) = mpsc::unbounded_channel();
let (from_smoltcp_tx, from_smoltcp_rx) = mpsc::unbounded_channel();
// 🔥 ФИКС: Ограничиваем очередь физики до 128 пакетов
let (to_smoltcp_tx, to_smoltcp_rx) = mpsc::channel(128);
let (from_smoltcp_tx, from_smoltcp_rx) = mpsc::channel(128);
let rx_avail = Arc::new(AtomicBool::new(false));
let now = StdInstant::now();
let device = Self {
capabilities,
rx_queue: to_smoltcp_rx,
@@ -131,7 +132,6 @@ impl VirtTunDevice {
last_log_time: now,
};
(device, to_smoltcp_tx, from_smoltcp_rx, rx_avail)
}
@@ -168,10 +168,8 @@ impl VirtTunDevice {
fn check_and_log_stats(&mut self) {
let now = StdInstant::now();
if now.duration_since(self.last_log_time).as_secs() >= 5 {
let stats = self.get_stats();
info!(
"TunDevice Traffic: RX: {:.2} MB ({} pkts) | TX: {:.2} MB ({} pkts) | Speed: ↓{:.2} MB/s, ↑{:.2} MB/s",
stats.rx_bytes as f64 / 1_048_576.0,
@@ -181,7 +179,6 @@ impl VirtTunDevice {
stats.rx_speed_mb_s,
stats.tx_speed_mb_s
);
self.last_log_time = now;
}
}
@@ -193,7 +190,6 @@ impl Device for VirtTunDevice {
fn receive(&mut self, _timestamp: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
self.check_and_log_stats();
if let Ok(buffer) = self.rx_queue.try_recv() {
let len = buffer.len() as u64;
self.rx_bytes += buffer.len() as u64;
@@ -201,7 +197,6 @@ impl Device for VirtTunDevice {
GLOBAL_TX_BYTES.fetch_add(len as u64, Ordering::Relaxed);
GLOBAL_TX_PACKETS.fetch_add(1, Ordering::Relaxed);
let rx = Self::RxToken {
buffer,
phantom_device: PhantomData,
@@ -237,7 +232,6 @@ 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
@@ -256,8 +250,9 @@ impl phy::TxToken for VirtTxToken<'_> {
GLOBAL_RX_BYTES.fetch_add(len as u64, Ordering::Relaxed);
GLOBAL_RX_PACKETS.fetch_add(1, Ordering::Relaxed);
let _ = self.0.tx_queue.send(buffer);
// 🔥 ФИКС: Используем try_send. Если буфер полон (сеть тормозит), дропаем пакет!
// Это сигнализирует TCP-стеку об ограничении канала.
let _ = self.0.tx_queue.try_send(buffer);
result
}
}
}