fix(stereo): 修复主从节点同步异常的问题

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;

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<u32> = 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 没有收到包，认为是新流开始，重置状态
+            let current_server_time = clock.lock().await.to_server_time(now);
-            if now - last_packet_time > 500_000 {
+
-                jitter.clear();
+            // 检查包是否迟到（目标时间已过）
-                last_seq = None;
+            if current_server_time > pkt.timestamp {
-                codec = OpusCodec::new(&config)?;
+                let late_ms = (current_server_time - pkt.timestamp) / 1000;
-                let _ = player.prepare();
+                if late_ms > 50 {
+                    // 迟到超过 50ms，直接丢弃
+                    continue;
+                }
+                // 轻微迟到（<50ms），尝试播放
             }
-            last_packet_time = now;
-            jitter.push(pkt);
-        }
 
-        let now = now_us();
+            last_seq = Some(pkt.seq);
-        let current_server_time = clock.lock().await.to_server_time(now);
 
-        if let Some((seq, data)) = jitter.pop_frame(current_server_time) {
+            // 精确等待到播放时间
-            if let Some(last) = last_seq {
+            // 策略: 距离远时 sleep，距离近时忙等
-                let loss_count = seq.wrapping_sub(last) as i32 - 1;
+            loop {
-                if loss_count > 0 {
+                let now = now_us();
-                    // 1. 优先尝试 FEC 恢复最近丢失的那一帧
+                let current_server_time_precise = clock.lock().await.to_server_time(now);
-                    // Opus 的 FEC 数据存储在当前包(data)中，用于恢复“前一帧”
-                    if let Ok(len) = codec.decode_fec(&data, &mut pcm_buf) {
-                        let _ = player.write(&pcm_buf[..len]);
-                    }
 
-                    // 2. 如果丢包超过 1 帧，剩下的帧只能靠丢包补偿(PLC)
+                if current_server_time_precise >= pkt.timestamp {
-                    for _ in 0..(loss_count - 1) {
+                    break;
-                        if let Ok(len) = codec.decode_loss(&mut pcm_buf) {
+                }
-                            let _ = player.write(&pcm_buf[..len]);
+
-                        }
+                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)?;
+            if let Ok(len) = codec.decode(&pkt.data, &mut pcm_buf) {
-            player.write(&pcm_buf[..len])?;
+                let _ = player.write(&pcm_buf[..len]);
+            }
         } else {
-            tokio::time::sleep(Duration::from_millis(10)).await;
+            tokio::time::sleep(Duration::from_micros(100)).await;
         }
     }
 }

examples/stereo/src/utils/jitter_buffer.rs

@@ -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<u8>,
-}
-
-// 仅针对序列号进行比较，处理回绕逻辑
-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<Ordering> {
-        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<Reverse<OrderedPacket>>,
-    last_played_seq: Option<u32>,
-    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<u8>)> {
-        // 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;
-    }
-}

examples/stereo/src/utils/mod.rs

@@ -1,3 +1,2 @@
 pub mod alsa;
-pub mod jitter_buffer;
 pub mod sync;

examples/stereo/src/utils/sync.rs

@@ -10,10 +10,38 @@ pub fn now_us() -> u128 {
 }
 
 /// 时钟同步管理器，用于计算主从节点间的时钟偏移
+/// 采用改进的 NTP 算法 + Kalman 滤波思想
 pub struct ClockSync {
-    offsets: VecDeque<i128>,
+    /// 偏移量样本窗口
+    offsets: VecDeque<OffsetSample>,
+    /// 当前估计的时钟偏移 (server_time - client_time)
     pub current_offset: i128,
+    /// RTT 样本窗口
+    rtts: VecDeque<i128>,
+    /// 当前估计的最小 RTT
+    min_rtt: i128,
+    /// 窗口大小
     window_size: usize,
+    /// 时钟漂移率 (ppm: parts per million)
+    /// 正值表示从节点时钟比主节点快
+    drift_rate: f64,
+    /// 上次更新时间
+    last_update_time: u128,
+    /// 漂移率估计窗口
+    drift_samples: VecDeque<DriftSample>,
+}
+
+#[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;
+        let t1 = client_send_ts as i128;
-        // 基础过滤：如果 RTT 过大则忽略 (例如局域网内 > 100ms)
+        let t2 = server_ts as i128;
-        if rtt > 100_000 {
+        let t4 = client_recv_ts as i128;
+
+        let rtt = t4 - t1;
+
+        // 过滤异常 RTT (局域网内 > 100ms 视为异常)
+        if rtt < 0 || rtt > 100_000 {
             return;
         }
 
-        // 时钟偏移 = 主节点时间 - 从节点时间
+        // 计算时钟偏移: offset = server_time - client_time
-        // 假设主节点收到 Ping 的时间点在 (发送时间 + 接收时间) / 2
+        // offset = t2 - (t1 + t4) / 2
-        let estimated_server_time = server_ts as i128 + rtt / 2;
+        let offset = t2 - (t1 + t4) / 2;
-        let offset = estimated_server_time - client_recv_ts as i128;
 
-        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();
         }
 
-        // 计算中位数偏移，以增强抗干扰能力
+        // 偏移量估计: 使用低 RTT 样本的中位数
-        let mut sorted: Vec<i128> = self.offsets.iter().cloned().collect();
+        // 原理: RTT 较小的样本受网络抖动影响小，时间测量更准确
-        sorted.sort_unstable();
+        let mut low_rtt_offsets: Vec<i128> = self
-        if !sorted.is_empty() {
+            .offsets
-            self.current_offset = sorted[sorted.len() / 2];
+            .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::<Vec<_>>());
+        let offset_variance = self.calculate_variance(
+            &self.offsets.iter().map(|s| s.offset).collect::<Vec<_>>(),
+        );
+
+        // 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::<i128>() as f64 / samples.len() as f64;
+        let variance = samples
+            .iter()
+            .map(|&x| {
+                let diff = x as f64 - mean;
+                diff * diff
+            })
+            .sum::<f64>()
+            / samples.len() as f64;
+        variance.sqrt()
+    }
 }