diff --git a/viewer/src/main.rs b/viewer/src/main.rs index 48873a0..19eaac8 100644 --- a/viewer/src/main.rs +++ b/viewer/src/main.rs @@ -253,6 +253,9 @@ fn session( let latest = latest.clone(); let mut decoder = openh264::decoder::Decoder::new() .map_err(|e| anyhow::anyhow!("decoder init failed: {e}"))?; + let mut rgb_buf: Vec = Vec::new(); + let mut last_kfreq = Instant::now(); + let mut skip_stale = 0u64; tokio::spawn(async move { let mut size_logged = false; loop { @@ -262,11 +265,16 @@ fn session( stats.frames.fetch_add(1, Ordering::Relaxed); // 端到端延迟 = 接收时刻 - 时钟偏移 - 采集时刻(EMA 平滑) let mut est = None; + let mut stale = false; if stats.offset_valid.load(Ordering::Relaxed) { let e = now_ms() as i64 - stats.offset.load(Ordering::Relaxed) - hdr.ts_ms as i64; if (0..5000).contains(&e) { + // 积压保护:QUIC 视频流可靠有序,积压必须逐帧读出, + // 但陈旧帧(>250ms)可以不解码——跳过解码让积压以 + // 网络速度排空,否则消化速率≈到达速率永远追不回来 + stale = e > 250; let old = stats.e2e.load(Ordering::Relaxed) as i64; let sm = if stats.e2e_valid.load(Ordering::Relaxed) { (old * 4 + e) / 5 @@ -279,6 +287,22 @@ fn session( } } stats.on_frame_arrival(est); + if stale { + skip_stale += 1; + if skip_stale % 120 == 1 { + eprintln!( + "[viewer] backlog: stale frame skipped (e2e>250ms, total {skip_stale})" + ); + } + // 限频请求关键帧:积压排空后从 IDR 重新衔接画面 + if last_kfreq.elapsed() >= Duration::from_millis(200) { + last_kfreq = Instant::now(); + if let Ok(mut slot) = latest.lock() { + *slot = Some((0, 0, Vec::new())); + } + } + continue; + } match decoder.decode(&payload) { Ok(Some(yuv)) => { let (dw, dh) = yuv.dimensions(); @@ -290,11 +314,12 @@ fn session( hdr.width, hdr.height ); } - let mut rgb = vec![0u8; dw as usize * dh as usize * 3]; - yuv.write_rgb8(&mut rgb); + rgb_buf.clear(); + rgb_buf.resize(dw as usize * dh as usize * 3, 0); + yuv.write_rgb8(&mut rgb_buf); // latest-wins:直接覆盖旧帧,无积压 if let Ok(mut slot) = latest.lock() { - *slot = Some((dw, dh, rgb)); + *slot = Some((dw, dh, rgb_buf.clone())); } } Ok(None) => {} @@ -340,7 +365,10 @@ fn session( stats.offset_valid.store(true, Ordering::Relaxed); } Ok(_) => {} - Err(_) => break, + Err(e) => { + eprintln!("[viewer] control stream closed: {e} (RTT/LOSS 将停止更新)"); + break; + } } } }); @@ -364,7 +392,7 @@ fn session( push(&msg_tx, ControlMsg::KeyFrameReq); let mut window = Window::new( - "mvp-remote v0.7-LowLatency (Esc 退出)", + "mvp-remote v0.8-CatchUp (Esc 退出)", w, h, WindowOptions {