test(agent): 新增 dump_stream 码流诊断工具
用途：不依赖 viewer 的 Minifb 窗口，直接抓 agent 下发的码流并逐帧打印 NAL 结构与解码结果，用于定位解码恢复类问题。 能力 - 打印每帧的 seq / keyframe / 字节数 / NAL 类型（SPS PPS IDR P） - --skip-at N：模拟主控端「积压跳帧」——读出但不解码，用于验证跳帧是否 切断参考链（实测确认：跳掉一帧后所有 P 帧持续解码失败，直到收到 IDR） - 交叉校验帧头 keyframe 标志与码流实际 NAL 类型是否一致 - 末尾补发 KeyFrameReq 并观察 IDR 是否真的回来 用法 cargo run --release -p agent --bin dump_stream -- --port 47800 --code 123456 cargo run --release -p agent --bin dump_stream -- --port 47800 --code 123456 --skip-at 1 注意：KeyFrameReq 必须写在流 2（isend）——agent 的读循环只监听流 2， 写到流 1 会被静默忽略（本工具第一版踩过，误判「修复无效」）。 这个工具是发现 nvenc FORCEIDR 缺陷的直接手段，保留以备回归排查。
This commit is contained in:
郑博文
parent
e814cb64e7
commit
e778ff73cc
272
agent/src/bin/dump_stream.rs
Normal file
272
agent/src/bin/dump_stream.rs
Normal file
@ -0,0 +1,272 @@
|
||||
//! 临时诊断工具：抓 agent 下发的码流，逐帧打印 NAL 结构并试解码。
|
||||
//!
|
||||
//! 目的：定位「首帧能解、之后连续 decode failed」的根因，不靠猜。
|
||||
//!
|
||||
//! 用法：
|
||||
//! cargo run --release -p agent --bin dump_stream -- --port 47805 --code t
|
||||
//! cargo run --release -p agent --bin dump_stream -- --port 47805 --code t --skip-at 2
|
||||
//!
|
||||
//! `--skip-at N`：模拟主控端的「积压跳帧」行为——第 N 帧读出但不喂给解码器，
|
||||
//! 用来验证「跳帧是否直接导致后续所有 P 帧解不出来」。
|
||||
|
||||
use anyhow::{bail, Result};
|
||||
use common::{read_control, read_video_frame, write_control, ControlMsg};
|
||||
use openh264::decoder::Decoder;
|
||||
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
|
||||
use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
|
||||
use rustls::{DigitallySignedStruct, SignatureScheme};
|
||||
use std::net::ToSocketAddrs;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct SkipVerify;
|
||||
impl ServerCertVerifier for SkipVerify {
|
||||
fn verify_server_cert(
|
||||
&self,
|
||||
_e: &CertificateDer<'_>,
|
||||
_i: &[CertificateDer<'_>],
|
||||
_s: &ServerName<'_>,
|
||||
_o: &[u8],
|
||||
_n: UnixTime,
|
||||
) -> Result<ServerCertVerified, rustls::Error> {
|
||||
Ok(ServerCertVerified::assertion())
|
||||
}
|
||||
fn verify_tls12_signature(
|
||||
&self,
|
||||
_m: &[u8],
|
||||
_c: &CertificateDer<'_>,
|
||||
_d: &DigitallySignedStruct,
|
||||
) -> Result<HandshakeSignatureValid, rustls::Error> {
|
||||
Ok(HandshakeSignatureValid::assertion())
|
||||
}
|
||||
fn verify_tls13_signature(
|
||||
&self,
|
||||
_m: &[u8],
|
||||
_c: &CertificateDer<'_>,
|
||||
_d: &DigitallySignedStruct,
|
||||
) -> Result<HandshakeSignatureValid, rustls::Error> {
|
||||
Ok(HandshakeSignatureValid::assertion())
|
||||
}
|
||||
fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
|
||||
vec![
|
||||
SignatureScheme::RSA_PKCS1_SHA256,
|
||||
SignatureScheme::ECDSA_NISTP256_SHA256,
|
||||
SignatureScheme::ED25519,
|
||||
SignatureScheme::RSA_PSS_SHA256,
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
/// 扫 Annex-B 起始码，返回 NAL 类型列表（如 [7, 8, 5]）
|
||||
fn nal_types(payload: &[u8]) -> Vec<u8> {
|
||||
let mut out = Vec::new();
|
||||
let mut i = 0usize;
|
||||
while i + 3 < payload.len() {
|
||||
let sc = if payload[i..].starts_with(&[0, 0, 0, 1]) {
|
||||
4
|
||||
} else if payload[i..].starts_with(&[0, 0, 1]) {
|
||||
3
|
||||
} else {
|
||||
i += 1;
|
||||
continue;
|
||||
};
|
||||
match payload.get(i + sc) {
|
||||
Some(&b) => {
|
||||
out.push(b & 0x1F);
|
||||
i += sc + 1;
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn describe(types: &[u8]) -> String {
|
||||
let has = |t: u8| types.contains(&t);
|
||||
let mut s = String::new();
|
||||
if has(7) {
|
||||
s.push_str("SPS ");
|
||||
}
|
||||
if has(8) {
|
||||
s.push_str("PPS ");
|
||||
}
|
||||
if has(5) {
|
||||
s.push_str("IDR ");
|
||||
}
|
||||
if has(1) {
|
||||
s.push_str("P ");
|
||||
}
|
||||
if has(6) {
|
||||
s.push_str("SEI ");
|
||||
}
|
||||
if s.is_empty() {
|
||||
s.push_str("(未知)");
|
||||
}
|
||||
format!("{}[{}]", s.trim(), types.len())
|
||||
}
|
||||
|
||||
fn arg(name: &str) -> Option<String> {
|
||||
let a: Vec<String> = std::env::args().collect();
|
||||
a.iter()
|
||||
.position(|x| x == name)
|
||||
.and_then(|i| a.get(i + 1))
|
||||
.cloned()
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<()> {
|
||||
rustls::crypto::ring::default_provider()
|
||||
.install_default()
|
||||
.ok();
|
||||
|
||||
let port: u16 = arg("--port")
|
||||
.map(|s| s.parse())
|
||||
.transpose()?
|
||||
.unwrap_or(47800);
|
||||
let code = arg("--code").unwrap_or_else(|| "123456".into());
|
||||
let skip_at: Option<usize> = arg("--skip-at").map(|s| s.parse()).transpose()?;
|
||||
let total: usize = arg("--frames")
|
||||
.map(|s| s.parse())
|
||||
.transpose()?
|
||||
.unwrap_or(30);
|
||||
|
||||
let tls = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(Arc::new(SkipVerify))
|
||||
.with_no_client_auth();
|
||||
let qc: quinn::crypto::rustls::QuicClientConfig = Arc::new(tls).try_into()?;
|
||||
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));
|
||||
let mut client_config = quinn::ClientConfig::new(Arc::new(qc));
|
||||
client_config.transport_config(Arc::new(transport));
|
||||
let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse()?)?;
|
||||
endpoint.set_default_client_config(client_config);
|
||||
|
||||
let addr = ("127.0.0.1", port)
|
||||
.to_socket_addrs()?
|
||||
.next()
|
||||
.ok_or_else(|| anyhow::anyhow!("解析失败"))?;
|
||||
let conn = endpoint.connect(addr, "localhost")?.await?;
|
||||
println!("[dump] connected");
|
||||
|
||||
let (mut send, mut recv) = conn.open_bi().await?;
|
||||
write_control(
|
||||
&mut send,
|
||||
&ControlMsg::Auth {
|
||||
code,
|
||||
proto_version: common::PROTO_VERSION,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
match read_control(&mut recv).await? {
|
||||
ControlMsg::AuthOk {
|
||||
width,
|
||||
height,
|
||||
proto_version,
|
||||
} => println!("[dump] auth ok {width}x{height} proto={proto_version}"),
|
||||
other => bail!("认证失败: {other:?}"),
|
||||
}
|
||||
|
||||
let (mut isend, _irecv) = conn.open_bi().await?;
|
||||
|
||||
// 入会先请求一次 IDR，和 viewer 行为一致。
|
||||
// 注意：KeyFrameReq 必须走**流 2**（isend），agent 的读循环只监听流 2。
|
||||
write_control(&mut isend, &ControlMsg::KeyFrameReq).await?;
|
||||
|
||||
let mut decoder = Decoder::new()?;
|
||||
let mut sent_kfreq = false;
|
||||
|
||||
println!("\n序号 seq kf 载荷字节 NAL 结构 解码结果");
|
||||
println!("{}", "-".repeat(72));
|
||||
|
||||
for i in 0..total {
|
||||
let (hdr, payload) = match tokio::time::timeout(
|
||||
Duration::from_secs(10),
|
||||
read_video_frame(&mut recv),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(r) => r?,
|
||||
Err(_) => {
|
||||
println!("[{i}] 读帧超时（桌面可能完全静止）");
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
let types = nal_types(&payload);
|
||||
let is_idr = types.contains(&5);
|
||||
|
||||
// 交叉校验：帧头 keyframe 标志必须与码流实际 NAL 类型一致。
|
||||
// 这条不一致意味着时间戳/序号契约或编码器标志有 bug，比画面花屏更难排查。
|
||||
if hdr.keyframe != is_idr {
|
||||
println!(
|
||||
" ⚠️ 帧头 keyframe={} 但码流实际 IDR={}（seq={}）",
|
||||
hdr.keyframe, is_idr, hdr.seq
|
||||
);
|
||||
}
|
||||
|
||||
// 模拟主控端积压跳帧：读出但不解码
|
||||
if Some(i) == skip_at {
|
||||
println!(
|
||||
"[{i:2}] {:3} {:<5} {:8} {:<14} ⏭ 跳过解码（模拟积压跳帧）",
|
||||
hdr.seq,
|
||||
hdr.keyframe,
|
||||
payload.len(),
|
||||
describe(&types)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
let res = match decoder.decode(&payload) {
|
||||
Ok(Some(_)) => "✅ 解出画面".to_string(),
|
||||
Ok(None) => "○ 无输出（缓冲中）".to_string(),
|
||||
Err(e) => format!("❌ {e}"),
|
||||
};
|
||||
|
||||
println!(
|
||||
"[{i:2}] {:3} {:<5} {:8} {:<14} {}",
|
||||
hdr.seq,
|
||||
hdr.keyframe,
|
||||
payload.len(),
|
||||
describe(&types),
|
||||
res
|
||||
);
|
||||
|
||||
// 和 viewer 一样：解码失败就请求关键帧
|
||||
if res.starts_with('❌') && !sent_kfreq {
|
||||
sent_kfreq = true;
|
||||
write_control(&mut isend, &ControlMsg::KeyFrameReq).await?;
|
||||
println!(" ↑ 已发 KeyFrameReq");
|
||||
}
|
||||
}
|
||||
|
||||
// 诊断：静止桌面下再请求一次关键帧，看是否真能拿到 IDR
|
||||
println!("\n[dump] 再发一次 KeyFrameReq，观察 10 帧…");
|
||||
write_control(&mut isend, &ControlMsg::KeyFrameReq).await?;
|
||||
for i in 0..10 {
|
||||
match tokio::time::timeout(Duration::from_secs(5), read_video_frame(&mut recv)).await {
|
||||
Ok(Ok((hdr, payload))) => {
|
||||
let types = nal_types(&payload);
|
||||
println!(
|
||||
"[k{i:2}] seq={:3} kf={:<5} {:6}B {}",
|
||||
hdr.seq,
|
||||
hdr.keyframe,
|
||||
payload.len(),
|
||||
describe(&types)
|
||||
);
|
||||
if types.contains(&5) {
|
||||
println!(" ↑ 收到 IDR");
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
println!("[k{i:2}] 超时/流关闭");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let _ = conn.close(0u32.into(), b"done");
|
||||
Ok(())
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user