diff --git a/agent/src/capture.rs b/agent/src/capture.rs index d6e141e..e6daf81 100644 --- a/agent/src/capture.rs +++ b/agent/src/capture.rs @@ -275,7 +275,8 @@ pub fn desktop_size() -> Result<(u32, u32)> { /// 采集主循环:独立线程运行;连接断开(通道关闭)时退出。 /// 通道为 watch 单槽(latest-wins):发送即覆盖,网络慢时不产生积压。 -pub fn run(tx: Sender>>, keyreq: Arc) { +/// congestion:发送侧 write 超时置位 → 跳过当前帧(不编码不发送),注入率自适应降速。 +pub fn run(tx: Sender>>, keyreq: Arc, congestion: Arc) { let start = Instant::now(); loop { let mut dup = match Duplication::new() { @@ -333,6 +334,7 @@ pub fn run(tx: Sender>>, keyreq: Arc) { let mut stat_frames = 0u64; let mut stat_convert_ms = 0u64; let mut stat_encode_ms = 0u64; + let mut stat_skipped = 0u64; // 拥塞跳帧计数 macro_rules! emit { ($payload:expr) => {{ @@ -357,6 +359,16 @@ pub fn run(tx: Sender>>, keyreq: Arc) { Ok(true) => { // 主控端请求关键帧(解码异常恢复) let force_key = keyreq.swap(false, Ordering::Relaxed); + // 拥塞反馈:发送侧排队深 → 跳过本帧(NVENC 少输入一帧,参考链连续; + // 解码端收到的帧序 = 编码端实际输出序,无花屏风险) + if congestion.swap(false, Ordering::Relaxed) { + stat_skipped += 1; + let el = iter_start.elapsed(); + if el < min_frame { + std::thread::sleep(min_frame - el); + } + continue; + } match &mut encoder { EncoderKind::Nvenc(enc) => { let enc_start = Instant::now(); @@ -412,12 +424,14 @@ pub fn run(tx: Sender>>, keyreq: Arc) { } if stat_frames > 0 && stat_frames % 120 == 0 { eprintln!( - "[capture] avg convert {}ms, encode {}ms/frame (last 120)", + "[capture] avg convert {}ms, encode {}ms/frame, skipped {} (last 120)", stat_convert_ms / 120, - stat_encode_ms / 120 + stat_encode_ms / 120, + stat_skipped ); stat_convert_ms = 0; stat_encode_ms = 0; + stat_skipped = 0; } // 帧间隔节流到 ~60fps let el = iter_start.elapsed(); diff --git a/agent/src/main.rs b/agent/src/main.rs index 276f660..c96df0e 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -43,12 +43,13 @@ async fn main() -> Result<()> { // 传输参数调优: // - 收发窗口放大,避免高 RTT 下吞吐被 窗口/RTT 死锁(默认 512KB 流窗口) - // - 发送缓冲上限 512KB:QUIC 内部排队 ≈ 3 帧 @5Mbps,E2E 延迟天花板 ~150ms; - // 配合上游 latest-wins 丢帧,超量部分在进 QUIC 前就被丢弃 + // - 发送缓冲上限 64KB:QUIC 内部排队硬上限(实测 WiFi ~3Mbps 吞吐 → E2E ≤ ~170ms; + // 突发吞吐 = 64KB/7ms ≈ 72Mbps,不限制带宽)。注意此窗口为连接级,流 2 输入/Ping + // 数据量极小不受影响。配合下方拥塞反馈跳帧,把排队压在数帧以内 let mut transport = quinn::TransportConfig::default(); transport.receive_window(quinn::VarInt::from_u32(16 * 1024 * 1024)); transport.stream_receive_window(quinn::VarInt::from_u32(8 * 1024 * 1024)); - transport.send_window(512 * 1024); + transport.send_window(64 * 1024); let mut server_config = quinn::ServerConfig::with_single_cert(vec![cert], key.into())?; server_config.transport_config(Arc::new(transport)); @@ -87,7 +88,10 @@ async fn main() -> Result<()> { let (tx, mut frames_rx) = watch::channel::>>(None); let keyreq = Arc::new(AtomicBool::new(false)); let keyreq_capture = keyreq.clone(); - std::thread::spawn(move || capture::run(tx, keyreq_capture)); + // 拥塞反馈标志:发送侧 write 超过单帧预算(20ms) → 置位,采集线程跳帧降注入率 + let congestion = Arc::new(AtomicBool::new(false)); + let congestion_capture = congestion.clone(); + std::thread::spawn(move || capture::run(tx, keyreq_capture, congestion_capture)); // 上行控制消息(Pong)经通道交给主循环写流 2,避免多任务争抢 let (ctl_tx, mut ctl_rx) = mpsc::unbounded_channel::(); @@ -127,7 +131,14 @@ async fn main() -> Result<()> { break; // 采集线程退出 } if let Some(data) = frames_rx.borrow().clone() { - if send.write_all(&data).await.is_err() { + let wt = std::time::Instant::now(); + let ok = send.write_all(&data).await.is_ok(); + // write 耗时 > 单帧预算(16.7ms 的 1.2 倍) → QUIC 窗口排队深, + // 通知采集线程跳帧,把注入率降到网络实际吞吐水平 + if wt.elapsed() > std::time::Duration::from_millis(20) { + congestion.store(true, Ordering::Relaxed); + } + if !ok { break; } } diff --git a/viewer/src/main.rs b/viewer/src/main.rs index ac39c1f..36ba386 100644 --- a/viewer/src/main.rs +++ b/viewer/src/main.rs @@ -362,7 +362,7 @@ fn session( }; let mut window = Window::new( - "mvp-remote v0.6-HUD (Esc 退出)", + "mvp-remote v0.7-LowLatency (Esc 退出)", w, h, WindowOptions {