diff --git a/viewer/src/main.rs b/viewer/src/main.rs index 48873a0..8521538 100644 --- a/viewer/src/main.rs +++ b/viewer/src/main.rs @@ -253,6 +253,10 @@ 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; + let mut last_arr: Option = None; tokio::spawn(async move { let mut size_logged = false; loop { @@ -260,25 +264,55 @@ fn session( Ok((hdr, payload)) => { stats.bytes.fetch_add(payload.len() as u64, Ordering::Relaxed); stats.frames.fetch_add(1, Ordering::Relaxed); + // 与上一帧的到达间隔:≤100ms 才算"突发到达"。 + // 桌面静止时 agent 以 2Hz 空闲重发,帧龄天然 ≥500ms, + // 不能按陈旧帧丢弃——否则静止桌面所有帧被跳过 → + // 永久黑屏 + KeyFrameReq/IDR 死循环(v0.8 实测教训) + let burst = matches!(last_arr, Some(t) if t.elapsed() <= Duration::from_millis(100)); + last_arr = Some(Instant::now()); // 端到端延迟 = 接收时刻 - 时钟偏移 - 采集时刻(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) { - let old = stats.e2e.load(Ordering::Relaxed) as i64; - let sm = if stats.e2e_valid.load(Ordering::Relaxed) { - (old * 4 + e) / 5 - } else { - e - }; - stats.e2e.store(sm as u64, Ordering::Relaxed); - stats.e2e_valid.store(true, Ordering::Relaxed); - est = Some(sm); + // 积压判定必须同时满足:突发到达(背靠背)+ 帧龄>250ms。 + // QUIC 可靠有序,积压须逐帧读出但可不解码——跳过解码 + // 让队列以网络速度排空 + stale = burst && e > 250; + // 空闲重发帧的帧龄不代表显示延迟,不计入 EMA + if burst || e < 250 { + let old = stats.e2e.load(Ordering::Relaxed) as i64; + let sm = if stats.e2e_valid.load(Ordering::Relaxed) { + (old * 4 + e) / 5 + } else { + e + }; + stats.e2e.store(sm as u64, Ordering::Relaxed); + stats.e2e_valid.store(true, Ordering::Relaxed); + est = Some(sm); + } } } stats.on_frame_arrival(est); + if stale { + skip_stale += 1; + if skip_stale % 60 == 1 { + eprintln!( + "[viewer] backlog: stale burst frame skipped (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 +324,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 +375,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 +402,7 @@ fn session( push(&msg_tx, ControlMsg::KeyFrameReq); let mut window = Window::new( - "mvp-remote v0.7-LowLatency (Esc 退出)", + "mvp-remote v0.8.1-CatchUp (Esc 退出)", w, h, WindowOptions {