diff --git a/packages/client-v2/src/app/client/mod.rs b/packages/client-v2/src/app/client/mod.rs index 20a08a4..d08f621 100644 --- a/packages/client-v2/src/app/client/mod.rs +++ b/packages/client-v2/src/app/client/mod.rs @@ -279,6 +279,7 @@ impl Client { println!("[Client] Starting playback..."); let handle = PlaybackPipeline::spawn( config, + session.clock.clone(), session.audio_socket.clone(), session.cancel.clone(), ); diff --git a/packages/client-v2/src/app/client/pipeline.rs b/packages/client-v2/src/app/client/pipeline.rs index 7056dd7..fcd4ea6 100644 --- a/packages/client-v2/src/app/client/pipeline.rs +++ b/packages/client-v2/src/app/client/pipeline.rs @@ -22,13 +22,12 @@ use crate::audio::codec::OpusCodec; use crate::audio::config::AudioConfig; use crate::audio::player::AudioPlayer; use crate::audio::recorder::AudioRecorder; +use crate::net::jitter_buffer::{JitterBuffer, JitterConfig}; use crate::net::network::AudioSocket; use crate::net::protocol::AudioPacket; use crate::net::sync::now_us; -use std::collections::VecDeque; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -36,18 +35,15 @@ use tokio_util::sync::CancellationToken; /// 管道句柄 - 用于控制正在运行的音频管道 pub struct PipelineHandle { cancel: CancellationToken, - /// 用于通知阻塞线程停止 - stop_flag: Arc, } impl PipelineHandle { - fn new(cancel: CancellationToken, stop_flag: Arc) -> Self { - Self { cancel, stop_flag } + fn new(cancel: CancellationToken) -> Self { + Self { cancel } } /// 停止管道 pub fn stop(&self) { - self.stop_flag.store(true, Ordering::SeqCst); self.cancel.cancel(); } @@ -82,13 +78,12 @@ impl RecordPipeline { parent_cancel: CancellationToken, ) -> PipelineHandle { let cancel = parent_cancel.child_token(); - let stop_flag = Arc::new(AtomicBool::new(false)); - let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone()); + let handle = PipelineHandle::new(cancel.clone()); let token = cancel.clone(); tokio::spawn(async move { - if let Err(e) = Self::run(config, socket, target, token, stop_flag).await { + if let Err(e) = Self::run(config, socket, target, token).await { eprintln!("[RecordPipeline] Error: {}", e); } }); @@ -101,15 +96,13 @@ impl RecordPipeline { socket: Arc, target: SocketAddr, cancel: CancellationToken, - stop_flag: Arc, ) -> anyhow::Result<()> { // 创建 PCM 数据通道 let (pcm_tx, mut pcm_rx) = mpsc::channel::>(32); // 启动 ALSA 录音线程(阻塞 I/O) let recorder_config = config.clone(); - let recorder_stop = stop_flag.clone(); - + let channels = recorder_config.channels as usize; std::thread::spawn(move || { let recorder = match AudioRecorder::new(&recorder_config) { Ok(r) => r, @@ -119,13 +112,17 @@ impl RecordPipeline { } }; - let mut buf = vec![0i16; recorder_config.frame_size]; + let mut buf = + vec![0i16; recorder_config.frame_size * recorder_config.channels as usize]; - // 使用 stop_flag 来优雅退出 - while !recorder_stop.load(Ordering::SeqCst) { + loop { match recorder.read(&mut buf) { Ok(n) if n > 0 => { - if pcm_tx.blocking_send(buf[..n].to_vec()).is_err() { + let actual_samples = n * channels; + if pcm_tx + .blocking_send(buf[..actual_samples].to_vec()) + .is_err() + { break; } } @@ -153,11 +150,11 @@ impl RecordPipeline { pcm = pcm_rx.recv() => { match pcm { Some(samples) => { - if let Ok(len) = codec.encode(&samples, &mut opus_buf) { + if let Ok(encoded_len) = codec.encode(&samples, &mut opus_buf) { let packet = AudioPacket { seq: 0, timestamp: 0, - data: opus_buf[..len].to_vec(), + data: opus_buf[..encoded_len].to_vec(), }; let _ = socket.send(&packet, target).await; } @@ -189,17 +186,17 @@ impl PlaybackPipeline { /// * `parent_cancel` - 父级取消令牌 pub fn spawn( config: AudioConfig, + clock: Arc>, socket: Arc, parent_cancel: CancellationToken, ) -> PipelineHandle { let cancel = parent_cancel.child_token(); - let stop_flag = Arc::new(AtomicBool::new(false)); - let handle = PipelineHandle::new(cancel.clone(), stop_flag.clone()); + let handle = PipelineHandle::new(cancel.clone()); let token = cancel.clone(); tokio::spawn(async move { - if let Err(e) = Self::run(config, socket, token, stop_flag).await { + if let Err(e) = Self::run(config, clock, socket, token).await { eprintln!("[PlaybackPipeline] Error: {}", e); } }); @@ -209,91 +206,72 @@ impl PlaybackPipeline { async fn run( config: AudioConfig, + clock: Arc>, socket: Arc, cancel: CancellationToken, - stop_flag: Arc, ) -> anyhow::Result<()> { - // 创建 PCM 数据通道 - let (pcm_tx, pcm_rx) = mpsc::channel::(128); + // 1. 创建 PCM 通道 + let (pcm_tx, mut pcm_rx) = mpsc::channel::>(64); - // 启动 ALSA 播放线程(阻塞 I/O) + // 2. 专用播放线程 let player_config = config.clone(); - let player_stop = stop_flag.clone(); + std::thread::spawn(move || { + let player = match AudioPlayer::new(&player_config) { + Ok(p) => p, + Err(e) => return eprintln!("[PlaybackPipeline] Player init error: {}", e), + }; - println!("[PlaybackPipeline] Started"); - - // 主循环:从 UDP 接收,解码后发送给播放线程 - let mut udp_buf = vec![0u8; 4096]; - let mut last_time = now_us(); - tokio::spawn(async move { - loop { - match socket.recv(&mut udp_buf).await { - Ok((packet, _src)) => { - let now = now_us(); - let diff = now - last_time; - println!("Received packet now:{} diff:{}ms", now, diff / 1000); - last_time = now; - - if let Err(e) = pcm_tx.send(packet).await { - eprintln!("[PlaybackPipeline] PCM channel send error: {}", e); - break; - } - } - Err(e) => { - eprintln!("[PlaybackPipeline] Recv error: {}", e); - } + // 当 pcm_tx 在异步任务中被 drop,这里会自动退出 + while let Some(samples) = pcm_rx.blocking_recv() { + if let Err(e) = player.write(&samples) { + eprintln!("[PlaybackPipeline] Write error: {}", e); + break; } } + println!("[PlaybackPipeline] Player thread exited naturally"); }); - let player = match AudioPlayer::new(&player_config) { - Ok(p) => p, - Err(e) => { - eprintln!("[PlaybackPipeline] Failed to create player: {}", e); - return Err(e); - } - }; + // 3. Opus 解码器与 Jitter Buffer + let mut codec = OpusCodec::new(&config)?; + let mut jitter_buffer = JitterBuffer::new(JitterConfig::default()); + let mut pcm_frame = vec![0i16; config.frame_size * config.channels as usize]; - // 使用 blocking_recv 在线程中接收 - let mut rx = pcm_rx; - // 创建 Opus 解码器 - let mut codec = OpusCodec::new(&config).unwrap(); - let mut pcm_buf = vec![0i16; config.frame_size * config.channels as usize]; + // 提高定时精度 + let mut ticker = tokio::time::interval(std::time::Duration::from_millis(5)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - let mut jitter_buffer: VecDeque = VecDeque::new(); + println!("[PlaybackPipeline] Running"); - let mut start_time = 0u128; - let frame_duration_us = - (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128; - - while !player_stop.load(Ordering::SeqCst) { - while let Ok(p) = rx.try_recv() { - jitter_buffer.push_back(p); - } - if let Some(pck) = jitter_buffer.front() { - let now = now_us(); - if start_time == 0 { - start_time = now; + // 4. 主逻辑循环 + // 使用 loop + select,当 cancel 触发时直接 break + let mut udp_buf = vec![0u8; 4096]; + loop { + tokio::select! { + // 优先级 1: 外部取消 + _ = cancel.cancelled() => { + break; } - let target_client_time = start_time + (pck.seq as u128) * frame_duration_us; + // 优先级 2: 网络接收 + result = socket.recv(&mut udp_buf) => { + if let Ok((packet,_)) = result { + let arrival_time = clock.lock().to_server_time(now_us()); + jitter_buffer.push(packet, arrival_time); + } + } - if now >= target_client_time { - let packet = jitter_buffer.pop_front().unwrap(); - let samples = codec.decode(&packet.data, &mut pcm_buf)?; - player.write(&pcm_buf[..samples * config.channels as usize])?; - } else if target_client_time - now > 500_000 { - // Too far in the future, maybe clock jumped? - jitter_buffer.pop_front(); - } else { - // Wait until it's time - let wait = (target_client_time - now) as u64; - if wait > 1000 { - tokio::time::sleep(Duration::from_micros(wait)).await; + // 优先级 3: 播放调度 + _ = ticker.tick() => { + let current_time = clock.lock().to_server_time(now_us()); + while let Some(packet) = jitter_buffer.pop(current_time) { + if let Ok(samples) = codec.decode(&packet.data, &mut pcm_frame) { + let samples = pcm_frame[..samples * config.channels as usize].to_vec(); + if pcm_tx.try_send(samples).is_err() { + break; + } + } } } - } else { - tokio::time::sleep(Duration::from_millis(5)).await; } } diff --git a/packages/client-v2/src/app/server/stream.rs b/packages/client-v2/src/app/server/stream.rs index 1284171..54aba20 100644 --- a/packages/client-v2/src/app/server/stream.rs +++ b/packages/client-v2/src/app/server/stream.rs @@ -78,7 +78,7 @@ pub struct StreamSender { // 配置参数 input_len: usize, frame_duration_us: u128, - max_lead_us: u128, // 允许的最大超前时间,例如 1_000_000 (1s) + max_lead_us: u128, // 允许的最大超前时间,例如 500_000 (1s) } impl StreamSender { @@ -105,7 +105,7 @@ impl StreamSender { stream_start_ts: None, input_len, frame_duration_us, - max_lead_us: 1_000_000, + max_lead_us: 500_000, }) } @@ -259,7 +259,7 @@ impl RecorderStream { ) -> anyhow::Result<()> { let mut writer = WavWriter::create(&filename, config.sample_rate, config.channels)?; let mut codec = OpusCodec::new(&config)?; - let mut pcm = vec![0i16; config.frame_size]; + let mut pcm = vec![0i16; config.frame_size * config.channels as usize]; println!( "[Recorder] Started -> {} (filter: {:?})", @@ -280,8 +280,9 @@ impl RecorderStream { } // 解码并写入 - if let Ok(n) = codec.decode(&frame.packet.data, &mut pcm) { - let _ = writer.write_samples(&pcm[..n]); + if let Ok(samples_per_channel) = codec.decode(&frame.packet.data, &mut pcm) { + let total_samples = samples_per_channel * config.channels as usize; + let _ = writer.write_samples(&pcm[..total_samples])?; } } Err(broadcast::error::RecvError::Lagged(n)) => { diff --git a/packages/client-v2/src/bin/server.rs b/packages/client-v2/src/bin/server.rs index e9de390..d0a9cfe 100644 --- a/packages/client-v2/src/bin/server.rs +++ b/packages/client-v2/src/bin/server.rs @@ -103,13 +103,25 @@ async fn main() -> anyhow::Result<()> { .await; println!(" ✅ Event broadcasted"); + // 5. 测试音频录制 + println!("\n5️⃣ Testing Audio Recording (10 seconds)..."); + match server.start_record(addr, AudioConfig::voice_16k()).await { + Ok(_) => { + println!(" ⏺️ Recording started..."); + tokio::time::sleep(std::time::Duration::from_secs(10)).await; + server.stop_record(addr).await?; + println!(" ⏹️ Recording stopped"); + } + Err(e) => println!(" ❌ Recording failed: {}", e), + } + // 6. 测试音频播放(如果有测试文件) println!("\n6️⃣ Testing Audio Playback..."); if std::path::Path::new("temp/test.wav").exists() { match server.start_play(addr, "temp/test.wav").await { Ok(_) => { println!(" ▶️ Playback started..."); - tokio::time::sleep(std::time::Duration::from_secs(60)).await; + tokio::time::sleep(std::time::Duration::from_secs(10)).await; server.stop_play(addr).await?; println!(" ⏹️ Playback stopped"); } @@ -119,18 +131,6 @@ async fn main() -> anyhow::Result<()> { println!(" ⚠️ No test file found at temp/test.wav, skipping..."); } - // 5. 测试音频录制 - println!("\n5️⃣ Testing Audio Recording (5 seconds)..."); - match server.start_record(addr, AudioConfig::voice_16k()).await { - Ok(_) => { - println!(" ⏺️ Recording started..."); - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - server.stop_record(addr).await?; - println!(" ⏹️ Recording stopped"); - } - Err(e) => println!(" ❌ Recording failed: {}", e), - } - println!("\n═══════════════════════════════════════════════════════"); println!("✅ All tests completed!"); println!("\nServer status:"); diff --git a/packages/client-v2/src/net/jitter_buffer.rs b/packages/client-v2/src/net/jitter_buffer.rs new file mode 100644 index 0000000..ece9a09 --- /dev/null +++ b/packages/client-v2/src/net/jitter_buffer.rs @@ -0,0 +1,334 @@ +//! # Jitter Buffer - 抖动缓冲区 +//! +//! 用于音频流的抖动缓冲和包重排序。 +//! +//! ## 功能 +//! - 自适应缓冲区大小 +//! - 乱序包重排 +//! - 丢包检测和统计 +//! - 延迟统计 + +use crate::net::protocol::AudioPacket; +use std::cmp::Ordering; +use std::collections::{BTreeMap, VecDeque}; + +impl PartialEq for AudioPacket { + fn eq(&self, other: &Self) -> bool { + self.timestamp == other.timestamp + } +} +impl Eq for AudioPacket {} +impl Ord for AudioPacket { + fn cmp(&self, other: &Self) -> Ordering { + other.timestamp.cmp(&self.timestamp) + } +} +impl PartialOrd for AudioPacket { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +/// Jitter Buffer 统计信息 +#[derive(Debug, Clone, Default)] +pub struct JitterStats { + /// 总接收包数 + pub received: u64, + /// 总丢包数 + pub lost: u64, + /// 总播放包数 + pub played: u64, + /// 迟到的包数(到达时已过播放时间) + pub late: u64, + /// 重复包数 + pub duplicate: u64, + /// 当前缓冲区大小(包数) + pub buffer_size: usize, + /// 最小延迟(微秒) + pub min_delay: u128, + /// 最大延迟(微秒) + pub max_delay: u128, + /// 平均延迟(微秒) + pub avg_delay: u128, +} + +impl JitterStats { + /// 计算丢包率(百分比) + pub fn loss_rate(&self) -> f64 { + if self.received + self.lost == 0 { + return 0.0; + } + (self.lost as f64 / (self.received + self.lost) as f64) * 100.0 + } +} + +/// Jitter Buffer 配置 +#[derive(Debug, Clone)] +pub struct JitterConfig { + /// 最小缓冲区大小(包数) + pub min_buffer_size: usize, + /// 最大缓冲区大小(包数) + pub max_buffer_size: usize, + /// 目标缓冲区大小(包数) + pub target_buffer_size: usize, + /// 自适应调整间隔(包数) + pub adapt_interval: usize, + /// 最大容忍延迟(微秒) + pub max_tolerable_delay: u128, +} + +impl Default for JitterConfig { + fn default() -> Self { + Self { + min_buffer_size: 2, + max_buffer_size: 20, + target_buffer_size: 5, + adapt_interval: 50, + max_tolerable_delay: 100_000, // 100ms + } + } +} + +/// Jitter Buffer - 用于音频流的抖动缓冲 +pub struct JitterBuffer { + /// 配置 + config: JitterConfig, + /// 缓冲区(按时间戳排序) + buffer: BTreeMap, + /// 统计信息 + stats: JitterStats, + /// 期望的下一个序列号 + expected_seq: u32, + /// 是否已接收第一个包 + first_packet_received: bool, + /// 延迟样本窗口(用于自适应) + delay_samples: VecDeque, + /// 自适应计数器 + adapt_counter: usize, + /// 上次播放的时间戳 + last_played_timestamp: u128, +} + +impl JitterBuffer { + /// 创建新的 Jitter Buffer + pub fn new(config: JitterConfig) -> Self { + Self { + config, + buffer: BTreeMap::new(), + stats: JitterStats::default(), + expected_seq: 0, + first_packet_received: false, + delay_samples: VecDeque::with_capacity(100), + adapt_counter: 0, + last_played_timestamp: 0, + } + } + + /// 使用默认配置创建 + pub fn default() -> Self { + Self::new(JitterConfig::default()) + } + + /// 插入音频包 + pub fn push(&mut self, packet: AudioPacket, arrival_time: u128) { + // 检测重复包 + if self.buffer.contains_key(&packet.timestamp) { + self.stats.duplicate += 1; + return; + } + + // 初始化序列号 + if !self.first_packet_received { + self.expected_seq = packet.seq.wrapping_add(1); + self.first_packet_received = true; + } else { + // 检测丢包 + let seq_diff = packet.seq.wrapping_sub(self.expected_seq); + if seq_diff > 0 && seq_diff < 1000 { + // 允许一定的序列号跳跃(处理回环) + self.stats.lost += seq_diff as u64; + } + self.expected_seq = packet.seq.wrapping_add(1); + } + + // 检查是否迟到 + if self.last_played_timestamp > 0 && packet.timestamp < self.last_played_timestamp { + self.stats.late += 1; + return; + } + + // 超过最大容忍延迟,直接丢弃 + if packet.timestamp < arrival_time + && arrival_time - packet.timestamp > self.config.max_tolerable_delay + { + self.stats.late += 1; + return; + } + + let delay = if packet.timestamp >= arrival_time { + packet.timestamp - arrival_time + } else { + 0 // 包到达时已经过了播放时间 + }; + + self.update_delay_stats(delay); + + // 插入缓冲区 + self.buffer.insert(packet.timestamp, packet); + self.stats.received += 1; + self.stats.buffer_size = self.buffer.len(); + + // 自适应调整 + self.adapt_counter += 1; + if self.adapt_counter >= self.config.adapt_interval { + self.adapt_buffer_size(); + self.adapt_counter = 0; + } + } + + /// 获取下一个应该播放的包 + /// + /// # Arguments + /// * `current_time` - 当前服务器时间 + /// + /// # Returns + /// * `Some(packet)` - 如果有包应该播放 + /// * `None` - 如果缓冲区为空或还没到播放时间 + pub fn pop(&mut self, current_time: u128) -> Option { + // 如果缓冲区为空,直接返回 + if self.buffer.is_empty() { + return None; + } + + // 检查缓冲区是否达到最小大小 + if self.buffer.len() < self.config.target_buffer_size && !self.should_drain() { + return None; + } + + // 获取最早的包 + if let Some((×tamp, _)) = self.buffer.iter().next() { + // 检查是否到达播放时间 + if current_time >= timestamp { + let packet = self.buffer.remove(×tamp).unwrap(); + self.last_played_timestamp = timestamp; + self.stats.played += 1; + self.stats.buffer_size = self.buffer.len(); + return Some(packet); + } + } + + None + } + + /// 强制获取下一个包(无论时间) + pub fn pop_next(&mut self) -> Option { + if let Some((×tamp, _)) = self.buffer.iter().next() { + let packet = self.buffer.remove(×tamp).unwrap(); + self.last_played_timestamp = timestamp; + self.stats.played += 1; + self.stats.buffer_size = self.buffer.len(); + return Some(packet); + } + None + } + + /// 查看下一个包的播放时间(不移除) + pub fn peek_next_timestamp(&self) -> Option { + self.buffer.keys().next().copied() + } + + /// 获取缓冲区大小 + pub fn len(&self) -> usize { + self.buffer.len() + } + + /// 检查缓冲区是否为空 + pub fn is_empty(&self) -> bool { + self.buffer.is_empty() + } + + /// 获取统计信息 + pub fn stats(&self) -> &JitterStats { + &self.stats + } + + /// 重置统计信息 + pub fn reset_stats(&mut self) { + self.stats = JitterStats::default(); + self.stats.buffer_size = self.buffer.len(); + } + + /// 清空缓冲区 + pub fn clear(&mut self) { + self.buffer.clear(); + self.stats.buffer_size = 0; + self.first_packet_received = false; + self.last_played_timestamp = 0; + } + + /// 检查是否应该排空缓冲区(处理长时间没有新包的情况) + fn should_drain(&self) -> bool { + // 如果缓冲区有包且已经等了很久,就开始播放 + !self.buffer.is_empty() && self.buffer.len() >= self.config.min_buffer_size + } + + /// 更新延迟统计 + fn update_delay_stats(&mut self, delay: u128) { + // 更新最小/最大延迟 + if self.stats.received == 0 { + self.stats.min_delay = delay; + self.stats.max_delay = delay; + self.stats.avg_delay = delay; + } else { + self.stats.min_delay = self.stats.min_delay.min(delay); + self.stats.max_delay = self.stats.max_delay.max(delay); + // 滑动平均 + self.stats.avg_delay = (self.stats.avg_delay * 9 + delay) / 10; + } + + // 保存延迟样本用于自适应 + self.delay_samples.push_back(delay); + if self.delay_samples.len() > 100 { + self.delay_samples.pop_front(); + } + } + + /// 自适应调整缓冲区大小 + fn adapt_buffer_size(&mut self) { + if self.delay_samples.len() < 10 { + return; + } + + // 计算延迟方差(抖动) + let avg = self.stats.avg_delay; + let variance: f64 = self + .delay_samples + .iter() + .map(|&d| { + let diff = d as i128 - avg as i128; + (diff * diff) as f64 + }) + .sum::() + / self.delay_samples.len() as f64; + + let jitter = variance.sqrt(); + + // 根据抖动调整目标缓冲区大小 + // 抖动大 -> 增加缓冲区 + // 抖动小 -> 减少缓冲区 + let target = if jitter > 50_000.0 { + // 高抖动(>50ms 标准差) + self.config.target_buffer_size + 2 + } else if jitter < 10_000.0 { + // 低抖动(<10ms 标准差) + self.config.target_buffer_size.saturating_sub(1) + } else { + self.config.target_buffer_size + }; + + // 限制在最小/最大范围内 + self.config.target_buffer_size = target + .max(self.config.min_buffer_size) + .min(self.config.max_buffer_size); + } +} diff --git a/packages/client-v2/src/net/mod.rs b/packages/client-v2/src/net/mod.rs index 5f4f8d5..0674a96 100644 --- a/packages/client-v2/src/net/mod.rs +++ b/packages/client-v2/src/net/mod.rs @@ -7,10 +7,13 @@ //! - `network` - 底层网络连接 //! - `protocol` - 通信协议定义 //! - `rpc` - RPC 调用管理 +//! - `sync` - 时间同步工具 +//! - `jitter_buffer` - 抖动缓冲区实现 pub mod command; pub mod discovery; pub mod event; +pub mod jitter_buffer; pub mod network; pub mod protocol; pub mod rpc;