diff --git a/examples/stereo/src/app/master.rs b/examples/stereo/src/app/master.rs index b290a23..796dd95 100644 --- a/examples/stereo/src/app/master.rs +++ b/examples/stereo/src/app/master.rs @@ -91,7 +91,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> { let mut opus_out = vec![0u8; 1500]; let mut seq = 0u32; - let delay_us = 200_000; + // 播放延迟: 只需覆盖网络延迟 + 时钟偏移 + let delay_us = 100_000; // 100ms 基础延迟 let frame_duration_us = (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128; @@ -142,7 +143,8 @@ pub async fn run_master(master_role: ChannelRole) -> Result<()> { stream_start_seq = seq; } - // 计算该帧应当播放的基准时间（相对于流开始） + // 计算该帧应当播放的目标时间 + // target_ts = 数据包发送时间 + 播放延迟 let target_ts = stream_start_ts + ((seq - stream_start_seq) as u128 * frame_duration_us) + delay_us; diff --git a/examples/stereo/src/app/slave.rs b/examples/stereo/src/app/slave.rs index fd634d6..98f5c0d 100644 --- a/examples/stereo/src/app/slave.rs +++ b/examples/stereo/src/app/slave.rs @@ -6,7 +6,6 @@ use crate::audio::player::AudioPlayer; use crate::net::discovery::Discovery; use crate::net::network::SlaveNetwork; use crate::net::protocol::{AudioPacket, ChannelRole, ControlPacket}; -use crate::utils::jitter_buffer::JitterBuffer; use crate::utils::sync::{ClockSync, now_us}; use anyhow::{Result, anyhow}; use std::sync::Arc; @@ -62,7 +61,6 @@ async fn handle_connection(role: ChannelRole) -> Result<()> { }; let player = AudioPlayer::new(&config)?; let mut codec = OpusCodec::new(&config)?; - let mut jitter = JitterBuffer::new(50_000, 3); let clock = Arc::new(Mutex::new(ClockSync::new(100))); // 用于通知主循环 TCP 已断开的消息通道 @@ -85,7 +83,7 @@ async fn handle_connection(role: ChannelRole) -> Result<()> { let _ = d_tx_ping.send(()).await; // 通知主线程 TCP 失败 break; } - tokio::time::sleep(Duration::from_secs(1)).await; + tokio::time::sleep(Duration::from_millis(200)).await; seq += 1; } }); @@ -130,7 +128,7 @@ async fn handle_connection(role: ChannelRole) -> Result<()> { } }); - // 8. 播放主循环 + // 8. 播放提示 println!("✅ 主节点已连接，音频串流中..."); let role_str = role.to_string(); tokio::spawn(async move { @@ -144,9 +142,9 @@ async fn handle_connection(role: ChannelRole) -> Result<()> { .await; }); + // 9. 播放主循环 let mut pcm_buf = vec![0i16; config.frame_size]; let mut last_seq: Option = None; - let mut last_packet_time = now_us(); loop { // 检查 TCP 是否已断开 @@ -154,48 +152,53 @@ async fn handle_connection(role: ChannelRole) -> Result<()> { return Err(anyhow!("主节点已断开: {}", master_tcp_addr)); } - // 填充 Jitter Buffer - while let Ok(pkt) = audio_rx.try_recv() { + // 接收数据包 + if let Ok(pkt) = audio_rx.try_recv() { let now = now_us(); - // 如果超过 500ms 没有收到包，认为是新流开始，重置状态 - if now - last_packet_time > 500_000 { - jitter.clear(); - last_seq = None; - codec = OpusCodec::new(&config)?; - let _ = player.prepare(); + let current_server_time = clock.lock().await.to_server_time(now); + + // 检查包是否迟到（目标时间已过） + if current_server_time > pkt.timestamp { + let late_ms = (current_server_time - pkt.timestamp) / 1000; + if late_ms > 50 { + // 迟到超过 50ms，直接丢弃 + continue; + } + // 轻微迟到（<50ms），尝试播放 } - last_packet_time = now; - jitter.push(pkt); - } - let now = now_us(); - let current_server_time = clock.lock().await.to_server_time(now); + last_seq = Some(pkt.seq); - if let Some((seq, data)) = jitter.pop_frame(current_server_time) { - if let Some(last) = last_seq { - let loss_count = seq.wrapping_sub(last) as i32 - 1; - if loss_count > 0 { - // 1. 优先尝试 FEC 恢复最近丢失的那一帧 - // Opus 的 FEC 数据存储在当前包(data)中，用于恢复“前一帧” - if let Ok(len) = codec.decode_fec(&data, &mut pcm_buf) { - let _ = player.write(&pcm_buf[..len]); - } + // 精确等待到播放时间 + // 策略: 距离远时 sleep，距离近时忙等 + loop { + let now = now_us(); + let current_server_time_precise = clock.lock().await.to_server_time(now); - // 2. 如果丢包超过 1 帧，剩下的帧只能靠丢包补偿(PLC) - for _ in 0..(loss_count - 1) { - if let Ok(len) = codec.decode_loss(&mut pcm_buf) { - let _ = player.write(&pcm_buf[..len]); - } - } + if current_server_time_precise >= pkt.timestamp { + break; + } + + let wait_us = (pkt.timestamp - current_server_time_precise) as u64; + + if wait_us > 5000 { + // 等待时间 >5ms: sleep 大部分时间 + tokio::time::sleep(Duration::from_micros(wait_us - 3000)).await; + } else if wait_us > 1000 { + // 等待时间 1-5ms: yield 让出 CPU + tokio::task::yield_now().await; + } else { + // 等待时间 <1ms: 直接 break，让播放时机稍微早一点点 + break; } } - last_seq = Some(seq); - // 3. 正常解码当前帧 - let len = codec.decode(&data, &mut pcm_buf)?; - player.write(&pcm_buf[..len])?; + // 解码并播放 + if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) { + let _ = player.write(&pcm_buf[..len]); + } } else { - tokio::time::sleep(Duration::from_millis(10)).await; + tokio::time::sleep(Duration::from_micros(100)).await; } } } diff --git a/examples/stereo/src/utils/jitter_buffer.rs b/examples/stereo/src/utils/jitter_buffer.rs deleted file mode 100644 index 50d900e..0000000 --- a/examples/stereo/src/utils/jitter_buffer.rs +++ /dev/null @@ -1,100 +0,0 @@ -use std::cmp::Ordering; -use std::cmp::Reverse; -use std::collections::BinaryHeap; - -use crate::net::protocol::AudioPacket; - -#[derive(Debug)] -struct OrderedPacket { - seq: u32, - timestamp: u128, - data: Vec, -} - -// 仅针对序列号进行比较，处理回绕逻辑 -impl PartialEq for OrderedPacket { - fn eq(&self, other: &Self) -> bool { - self.seq == other.seq - } -} - -impl Eq for OrderedPacket {} - -impl PartialOrd for OrderedPacket { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -impl Ord for OrderedPacket { - fn cmp(&self, other: &Self) -> Ordering { - // 使用 wrapping_sub 处理 u32 回绕 (Rollover) - let diff = self.seq.wrapping_sub(other.seq) as i32; - diff.cmp(&0) - } -} - -pub struct JitterBuffer { - // 使用 Reverse 将 BinaryHeap 变为小顶堆，避免手动实现大量 Trait - buffer: BinaryHeap>, - last_played_seq: Option, - pub target_delay_us: u128, - min_packets: usize, // 最小缓冲包数，防止微小抖动 -} - -impl JitterBuffer { - pub fn new(target_delay_us: u128, min_packets: usize) -> Self { - Self { - buffer: BinaryHeap::with_capacity(32), // 预分配初始容量 - last_played_seq: None, - target_delay_us, - min_packets, - } - } - - pub fn push(&mut self, packet: AudioPacket) { - // 1. 处理序列号回绕的丢包逻辑 - if let Some(last) = self.last_played_seq { - let diff = packet.seq.wrapping_sub(last) as i32; - if diff <= 0 { - return; // 这是一个延迟到达的旧包，直接丢弃 - } - } - - self.buffer.push(Reverse(OrderedPacket { - seq: packet.seq, - timestamp: packet.timestamp, - data: packet.data, - })); - } - - pub fn pop_frame(&mut self, current_time: u128) -> Option<(u32, Vec)> { - // 2. 预缓冲逻辑：如果包量太少，先不播放，等待填充 - if self.buffer.len() < self.min_packets && self.last_played_seq.is_none() { - return None; - } - - // 3. 检查堆顶元素 - if let Some(Reverse(pkt)) = self.buffer.peek() { - // 判断是否到达播放时间（考虑目标延迟） - if current_time >= pkt.timestamp + (self.target_delay_us / 1000) { - let Reverse(pkt) = self.buffer.pop().unwrap(); - self.last_played_seq = Some(pkt.seq); - return Some((pkt.seq, pkt.data)); - } - } - None - } - - /// 如果缓冲区堆积过大，可以主动跳帧以降低延迟 - pub fn shrink_to_fit_latency(&mut self, max_size: usize) { - while self.buffer.len() > max_size { - self.buffer.pop(); - } - } - - pub fn clear(&mut self) { - self.buffer.clear(); - self.last_played_seq = None; - } -} diff --git a/examples/stereo/src/utils/mod.rs b/examples/stereo/src/utils/mod.rs index bf98177..8eaa5c4 100644 --- a/examples/stereo/src/utils/mod.rs +++ b/examples/stereo/src/utils/mod.rs @@ -1,3 +1,2 @@ pub mod alsa; -pub mod jitter_buffer; pub mod sync; diff --git a/examples/stereo/src/utils/sync.rs b/examples/stereo/src/utils/sync.rs index beac2f3..354ab8d 100644 --- a/examples/stereo/src/utils/sync.rs +++ b/examples/stereo/src/utils/sync.rs @@ -10,10 +10,38 @@ pub fn now_us() -> u128 { } /// 时钟同步管理器，用于计算主从节点间的时钟偏移 +/// 采用改进的 NTP 算法 + Kalman 滤波思想 pub struct ClockSync { - offsets: VecDeque, + /// 偏移量样本窗口 + offsets: VecDeque, + /// 当前估计的时钟偏移 (server_time - client_time) pub current_offset: i128, + /// RTT 样本窗口 + rtts: VecDeque, + /// 当前估计的最小 RTT + min_rtt: i128, + /// 窗口大小 window_size: usize, + /// 时钟漂移率 (ppm: parts per million) + /// 正值表示从节点时钟比主节点快 + drift_rate: f64, + /// 上次更新时间 + last_update_time: u128, + /// 漂移率估计窗口 + drift_samples: VecDeque, +} + +#[derive(Clone, Copy)] +struct OffsetSample { + offset: i128, + rtt: i128, + timestamp: u128, +} + +#[derive(Clone, Copy)] +struct DriftSample { + offset: i128, + timestamp: u128, } impl ClockSync { @@ -21,43 +49,176 @@ impl ClockSync { Self { offsets: VecDeque::with_capacity(window_size), current_offset: 0, + rtts: VecDeque::with_capacity(window_size), + min_rtt: i128::MAX, window_size, + drift_rate: 0.0, + last_update_time: now_us(), + drift_samples: VecDeque::with_capacity(60), // 保留 60 秒的样本 } } - /// 更新时钟偏移估计 + /// 更新时钟偏移估计 (NTP 算法) + /// + /// NTP 时间戳标记: + /// t1 = client_send_ts : 客户端发送 Ping 的时间 + /// t2 = server_ts : 服务器接收 Ping 的时间 + /// t3 = server_ts : 服务器发送 Pong 的时间 (假设处理时间忽略不计) + /// t4 = client_recv_ts : 客户端接收 Pong 的时间 + /// + /// RTT = (t4 - t1) - (t3 - t2) = (t4 - t1) (因为 t3 = t2) + /// Offset = ((t2 - t1) + (t3 - t4)) / 2 = ((t2 - t1) + (t2 - t4)) / 2 + /// = t2 - (t1 + t4) / 2 pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) { - let rtt = (client_recv_ts - client_send_ts) as i128; - // 基础过滤：如果 RTT 过大则忽略 (例如局域网内 > 100ms) - if rtt > 100_000 { + let t1 = client_send_ts as i128; + let t2 = server_ts as i128; + let t4 = client_recv_ts as i128; + + let rtt = t4 - t1; + + // 过滤异常 RTT (局域网内 > 100ms 视为异常) + if rtt < 0 || rtt > 100_000 { return; } - // 时钟偏移 = 主节点时间 - 从节点时间 - // 假设主节点收到 Ping 的时间点在 (发送时间 + 接收时间) / 2 - let estimated_server_time = server_ts as i128 + rtt / 2; - let offset = estimated_server_time - client_recv_ts as i128; + // 计算时钟偏移: offset = server_time - client_time + // offset = t2 - (t1 + t4) / 2 + let offset = t2 - (t1 + t4) / 2; - self.offsets.push_back(offset); + // 更新 RTT 窗口 + self.rtts.push_back(rtt); + if self.rtts.len() > self.window_size { + self.rtts.pop_front(); + } + self.min_rtt = *self.rtts.iter().min().unwrap_or(&rtt); + + // 更新偏移量窗口 + let sample = OffsetSample { + offset, + rtt, + timestamp: client_recv_ts, + }; + self.offsets.push_back(sample); if self.offsets.len() > self.window_size { self.offsets.pop_front(); } - // 计算中位数偏移，以增强抗干扰能力 - let mut sorted: Vec = self.offsets.iter().cloned().collect(); - sorted.sort_unstable(); - if !sorted.is_empty() { - self.current_offset = sorted[sorted.len() / 2]; + // 偏移量估计: 使用低 RTT 样本的中位数 + // 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确 + let mut low_rtt_offsets: Vec = self + .offsets + .iter() + .filter(|s| s.rtt <= self.min_rtt + 5000) // 5ms 容差 + .map(|s| s.offset) + .collect(); + + if !low_rtt_offsets.is_empty() { + low_rtt_offsets.sort_unstable(); + let new_offset = low_rtt_offsets[low_rtt_offsets.len() / 2]; + + // 漂移率估计 + self.estimate_drift(new_offset, client_recv_ts); + + // 平滑更新偏移量 (避免突变) + let alpha = 0.3; // 低通滤波系数 + self.current_offset = + (alpha * new_offset as f64 + (1.0 - alpha) * self.current_offset as f64) as i128; + } + + self.last_update_time = client_recv_ts; + } + + /// 估计时钟漂移率 + /// 时钟漂移率 = d(offset) / dt + fn estimate_drift(&mut self, offset: i128, timestamp: u128) { + self.drift_samples.push_back(DriftSample { offset, timestamp }); + if self.drift_samples.len() > 60 { + self.drift_samples.pop_front(); + } + + // 至少需要 10 秒的数据才能估计漂移 + if self.drift_samples.len() < 10 { + return; + } + + // 使用线性回归估计漂移率 + let first = self.drift_samples.front().unwrap(); + let last = self.drift_samples.back().unwrap(); + + let dt = (last.timestamp - first.timestamp) as f64; + let d_offset = (last.offset - first.offset) as f64; + + if dt > 10_000_000.0 { + // 超过 10 秒 + // drift_rate 单位: 微秒/秒 = ppm + let new_drift = d_offset / (dt / 1_000_000.0); + + // 平滑更新漂移率 + let beta = 0.1; + self.drift_rate = beta * new_drift + (1.0 - beta) * self.drift_rate; } } /// 将本地时间转换为服务器(主节点)时间 + /// 考虑时钟漂移补偿 pub fn to_server_time(&self, client_time: u128) -> u128 { - (client_time as i128 + self.current_offset) as u128 + let base_server_time = (client_time as i128 + self.current_offset) as u128; + + // 漂移补偿: 根据距离上次同步的时间，补偿时钟漂移 + let elapsed_since_update = client_time.saturating_sub(self.last_update_time) as f64; + let drift_correction = (self.drift_rate * elapsed_since_update / 1_000_000.0) as i128; + + (base_server_time as i128 + drift_correction) as u128 } /// 将服务器(主节点)时间转换为本地时间 pub fn to_client_time(&self, server_time: u128) -> u128 { + // 简化版本，不考虑漂移补偿 (播放时主要用 to_server_time) (server_time as i128 - self.current_offset) as u128 } + + /// 获取当前估计的 RTT (微秒) + pub fn get_rtt(&self) -> i128 { + self.min_rtt + } + + /// 获取当前时钟漂移率 (ppm) + pub fn get_drift_rate(&self) -> f64 { + self.drift_rate + } + + /// 获取同步质量评估 (0-100, 越高越好) + pub fn get_sync_quality(&self) -> u8 { + if self.offsets.is_empty() { + return 0; + } + + // 基于 RTT 稳定性和偏移量方差评估 + let rtt_variance = self.calculate_variance(&self.rtts.iter().copied().collect::>()); + let offset_variance = self.calculate_variance( + &self.offsets.iter().map(|s| s.offset).collect::>(), + ); + + // RTT 越稳定，方差越小，质量越高 + let rtt_score = ((100_000.0 - rtt_variance.min(100_000.0)) / 100_000.0 * 50.0) as u8; + let offset_score = ((50_000.0 - offset_variance.min(50_000.0)) / 50_000.0 * 50.0) as u8; + + rtt_score + offset_score + } + + fn calculate_variance(&self, samples: &[i128]) -> f64 { + if samples.is_empty() { + return 0.0; + } + let mean = samples.iter().sum::() as f64 / samples.len() as f64; + let variance = samples + .iter() + .map(|&x| { + let diff = x as f64 - mean; + diff * diff + }) + .sum::() + / samples.len() as f64; + variance.sqrt() + } }