Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! Owns local device streams on the helper worker. Callbacks allocate no buffers and take no locks. 2 //! Small callbacks share full queue slots; overload discards stale media and resumes fresh audio. 3 //! Capture and actual rendered output carry device timing. 4 //! References start with worker service; unmute rejects earlier device capture buffers. 5 6 #[path = "audio_sink.rs"] 7 mod audio_sink; 8 #[path = "device_buffers.rs"] 9 mod buffers; 10 #[path = "capture_worker.rs"] 11 mod capture_worker; 12 #[path = "playback.rs"] 13 mod playback; 14 #[path = "playout.rs"] 15 mod playout; 16 #[path = "processing.rs"] 17 mod processing; 18 19 use std::io; 20 use std::sync::Arc; 21 use std::sync::atomic::AtomicU16; 22 use std::sync::atomic::Ordering; 23 use std::time::Duration; 24 use std::time::Instant; 25 26 use cpal::FromSample; 27 use cpal::Sample; 28 use cpal::SampleFormat; 29 use cpal::SizedSample; 30 use cpal::traits::DeviceTrait; 31 use cpal::traits::HostTrait; 32 use cpal::traits::StreamTrait; 33 34 use buffers::BLOCK; 35 use buffers::Buffers; 36 use buffers::CaptureBoundary; 37 use buffers::Frame; 38 use buffers::FramePacker; 39 use buffers::MAX_CALLBACK_FRAMES; 40 use buffers::Playback; 41 use buffers::QUEUE_CAPACITY; 42 use playback::PlaybackPort; 43 44 const MAX_CAPTURE_AGE: Duration = Duration::from_secs(/*secs*/ 1); 45 46 pub(super) struct Devices { 47 _input: cpal::Stream, 48 output: Option<cpal::Stream>, 49 output_device: cpal::Device, 50 input_frames: u32, 51 playback: PlaybackPort, 52 playout: Option<playout::Playout>, 53 worker: capture_worker::CaptureWorker, 54 } 55 56 macro_rules! stream { 57 ($format:expr, $build:ident, $($arg:expr),+) => { 58 match $format { 59 SampleFormat::I8 => $build::<i8>($($arg),+), 60 SampleFormat::I16 => $build::<i16>($($arg),+), 61 SampleFormat::I32 => $build::<i32>($($arg),+), 62 SampleFormat::I64 => $build::<i64>($($arg),+), 63 SampleFormat::U8 => $build::<u8>($($arg),+), 64 SampleFormat::U16 => $build::<u16>($($arg),+), 65 SampleFormat::U32 => $build::<u32>($($arg),+), 66 SampleFormat::U64 => $build::<u64>($($arg),+), 67 SampleFormat::F32 => $build::<f32>($($arg),+), 68 SampleFormat::F64 => $build::<f64>($($arg),+), 69 _ => Err(cpal::Error::from(cpal::ErrorKind::UnsupportedConfig)), 70 } 71 }; 72 } 73 74 impl Devices { 75 pub(super) fn take_state(&self) -> io::Result<codex_realtime_webrtc::AudioState> { 76 self.worker.buffers.take_state() 77 } 78 79 pub(super) fn open() -> io::Result<Self> { 80 let host = cpal::default_host(); 81 let input = host 82 .default_input_device() 83 .ok_or_else(|| io::Error::other("microphone unavailable"))?; 84 let output = host 85 .default_output_device() 86 .ok_or_else(|| io::Error::other("speaker unavailable"))?; 87 let input_config = input 88 .default_input_config() 89 .map_err(|_| io::Error::other("microphone configuration unavailable"))?; 90 let output_config = output 91 .default_output_config() 92 .map_err(|_| io::Error::other("speaker configuration unavailable"))?; 93 let input_stream_config = bounded_stream_config(&input_config)?; 94 let output_stream_config = bounded_stream_config(&output_config)?; 95 let buffers = Arc::new(Buffers::new( 96 input_config.sample_rate(), 97 output_config.sample_rate(), 98 )); 99 let playback = PlaybackPort::new(buffers.clone(), output_config.sample_rate()); 100 let processor = 101 processing::Processor::new(input_config.sample_rate(), output_config.sample_rate()) 102 .map_err(io::Error::other)?; 103 let (cpal::BufferSize::Fixed(input_frames), cpal::BufferSize::Fixed(output_frames)) = ( 104 input_stream_config.buffer_size, 105 output_stream_config.buffer_size, 106 ) else { 107 return Err(io::Error::other("audio callback size unavailable")); 108 }; 109 processor 110 .validate_callback_timing(input_frames, output_frames) 111 .map_err(io::Error::other)?; 112 let input = stream!( 113 input_config.sample_format(), 114 build_input, 115 &input, 116 &input_stream_config, 117 buffers.clone() 118 ) 119 .map_err(|_| io::Error::other("failed to open microphone"))?; 120 let output_stream = stream!( 121 output_config.sample_format(), 122 build_output, 123 &output, 124 &output_stream_config, 125 buffers.clone() 126 ) 127 .map_err(|_| io::Error::other("failed to open speaker"))?; 128 output_stream 129 .play() 130 .map_err(|_| io::Error::other("failed to start speaker"))?; 131 input 132 .play() 133 .map_err(|_| io::Error::other("failed to start microphone"))?; 134 Ok(Self { 135 _input: input, 136 output: Some(output_stream), 137 output_device: output, 138 input_frames, 139 playback, 140 playout: None, 141 worker: capture_worker::CaptureWorker { 142 buffers, 143 processor, 144 pending: Default::default(), 145 }, 146 }) 147 } 148 149 pub(super) fn set_controls( 150 &mut self, 151 controls: codex_realtime_webrtc::AudioControls, 152 ) -> io::Result<()> { 153 let buffers = self.worker.buffers.clone(); 154 let previous = buffers.speaker.load(Ordering::Acquire); 155 self.worker.set_controls(controls)?; 156 if previous != buffers.speaker.load(Ordering::Acquire) { 157 // Stop callbacks before native teardown can block the worker. 158 drop(self.output.take()); 159 drop(self.playout.take()); 160 // The new epoch cancelled writers before either teardown; wait for 161 // any in-flight write before resetting its queue. 162 let producer = self 163 .playback 164 .0 165 .producer 166 .lock() 167 .map_err(|_| io::Error::other("speaker writer failed"))?; 168 while buffers.playback.pop().is_some() {} 169 while buffers.rendered.pop().is_some() {} 170 buffers.queued.store(/*val*/ 0, Ordering::Release); 171 drop(producer); 172 while buffers.rendered.pop().is_some() {} 173 // Opening a Bluetooth microphone can change the speaker's format. 174 // Requery after stopping the old stream rather than restoring its 175 // pre-microphone rate, which the device may no longer support. 176 let output_config = self 177 .output_device 178 .default_output_config() 179 .map_err(|_| io::Error::other("speaker configuration unavailable"))?; 180 let output_stream_config = bounded_stream_config(&output_config)?; 181 let cpal::BufferSize::Fixed(output_frames) = output_stream_config.buffer_size else { 182 return Err(io::Error::other("audio callback size unavailable")); 183 }; 184 self.worker 185 .processor 186 .set_render_rate(output_config.sample_rate()) 187 .map_err(io::Error::other)?; 188 self.worker 189 .processor 190 .validate_callback_timing(self.input_frames, output_frames) 191 .map_err(io::Error::other)?; 192 self.playback = PlaybackPort::new(buffers.clone(), output_config.sample_rate()); 193 if !controls.speaker_suppressed { 194 self.playout = 195 Some(playout::Playout::new(self.playback.writer()).map_err(io::Error::other)?); 196 } 197 let output = stream!( 198 output_config.sample_format(), 199 build_output, 200 &self.output_device, 201 &output_stream_config, 202 buffers 203 ) 204 .map_err(|_| io::Error::other("failed to reset speaker"))?; 205 output 206 .play() 207 .map_err(|_| io::Error::other("failed to restart speaker"))?; 208 self.output = Some(output); 209 } 210 Ok(()) 211 } 212 213 pub(super) async fn service( 214 &mut self, 215 audio: &mut crate::audio_track::AudioTrack, 216 ) -> io::Result<usize> { 217 if let Some(playout) = &self.playout { 218 playout 219 .check() 220 .map_err(|_| io::Error::other(crate::service_failure::ServiceFailure::Playout))?; 221 } 222 self.worker.service(audio, Instant::now).await 223 } 224 225 pub(super) fn receive(&self, packet: crate::incoming::ReceivedRtp) -> io::Result<()> { 226 if let Some(playout) = &self.playout { 227 playout.push(packet).map_err(io::Error::other)?; 228 } 229 Ok(()) 230 } 231 } 232 233 impl Drop for Devices { 234 fn drop(&mut self) { 235 self.worker.buffers.failed.store(true, Ordering::Release); 236 drop(self.output.take()); 237 drop(self.playout.take()); 238 let _producer = self.playback.0.producer.lock().ok(); 239 } 240 } 241 242 fn bounded_stream_config( 243 supported: &cpal::SupportedStreamConfig, 244 ) -> io::Result<cpal::StreamConfig> { 245 if supported.channels() == 0 246 || supported.channels() > 32 247 || !(8_000..=384_000).contains(&supported.sample_rate()) 248 { 249 return Err(io::Error::other("unsupported audio device configuration")); 250 } 251 let cpal::SupportedBufferSize::Range { min, max } = *supported.buffer_size() else { 252 return Err(io::Error::other("audio callback size range unavailable")); 253 }; 254 let mut config = supported.config(); 255 let min = min.max(1); 256 let max = max.min(MAX_CALLBACK_FRAMES as u32); 257 if min > max { 258 return Err(io::Error::other("unsupported audio callback size range")); 259 } 260 // ALSA allocates two periods. A 20 ms ring can be smaller than one PipeWire 261 // graph cycle (e.g. 2048 frames at 48 kHz), silently losing capture samples 262 // every cycle. Give Linux a bounded 100 ms ring instead, while retaining 263 // 10 ms callbacks elsewhere and rejecting incompatible device ranges below. 264 let periods_per_second = if cfg!(target_os = "linux") { 20 } else { 100 }; 265 let frames = (config.sample_rate / periods_per_second).clamp(min, max); 266 let callback_duration = 267 Duration::from_secs_f64(f64::from(frames) / f64::from(config.sample_rate)); 268 // Backends may deliver smaller callbacks than requested. Packing makes queue 269 // coverage depend on samples, while an incomplete block adds bounded latency. 270 let packing_duration = Duration::from_secs_f64(BLOCK as f64 / f64::from(config.sample_rate)); 271 let queue_duration = packing_duration * QUEUE_CAPACITY as u32; 272 if callback_duration + packing_duration + super::DEVICE_SERVICE_INTERVAL >= MAX_CAPTURE_AGE 273 || queue_duration <= super::DEVICE_SERVICE_INTERVAL 274 { 275 return Err(io::Error::other("unsupported audio callback timing")); 276 } 277 config.buffer_size = cpal::BufferSize::Fixed(frames); 278 Ok(config) 279 } 280 281 fn record_peak(peak: &AtomicU16, sample: f32) { 282 peak.fetch_max( 283 (sample.abs().min(1.0) * f32::from(u16::MAX)) as u16, 284 Ordering::Relaxed, 285 ); 286 } 287 288 // CPAL reports recoverable underruns/overruns through the same callback as device loss. 289 fn handle_stream_error(buffers: &Buffers, error: cpal::Error) { 290 if error.kind() != cpal::ErrorKind::Xrun { 291 buffers.failed.store(true, Ordering::Release); 292 } 293 } 294 295 fn build_input<T>( 296 device: &cpal::Device, 297 config: &cpal::StreamConfig, 298 buffers: Arc<Buffers>, 299 ) -> Result<cpal::Stream, cpal::Error> 300 where 301 T: SizedSample, 302 f32: FromSample<T>, 303 { 304 let channels = usize::from(config.channels); 305 let rate = f64::from(config.sample_rate); 306 let failure = buffers.clone(); 307 let mut origin = None; 308 let mut boundary = CaptureBoundary::default(); 309 let mut capture = FramePacker::default(); 310 device.build_input_stream( 311 *config, 312 move |data: &[T], info: &cpal::InputCallbackInfo| { 313 let generation = buffers.microphone.load(Ordering::Acquire); 314 if generation % 2 == 1 { 315 capture.reset(); 316 return; 317 } 318 let timestamp = info.timestamp(); 319 let origin = origin.get_or_insert(timestamp.capture); 320 let (Some(callback), Some(captured)) = ( 321 timestamp.callback.checked_duration_since(*origin), 322 timestamp.capture.checked_duration_since(*origin), 323 ) else { 324 capture.reset(); 325 return; 326 }; 327 if !boundary.accepts(generation, callback, captured) { 328 capture.reset(); 329 return; 330 } 331 let start = Instant::now() 332 .checked_sub( 333 timestamp 334 .callback 335 .checked_duration_since(timestamp.capture) 336 .unwrap_or_default(), 337 ) 338 .unwrap_or_else(Instant::now); 339 capture.discard_capture_gap(start, rate); 340 for (index, chunk) in data.chunks(BLOCK * channels).enumerate() { 341 let mut frame = Frame { 342 samples: [0.0; BLOCK], 343 len: chunk.len() / channels, 344 at: start + Duration::from_secs_f64((index * BLOCK) as f64 / rate), 345 generation, 346 }; 347 for (output, input) in frame.samples.iter_mut().zip(chunk.chunks_exact(channels)) { 348 *output = input 349 .iter() 350 .map(|sample| f32::from_sample(*sample)) 351 .sum::<f32>() 352 / channels as f32; 353 if !output.is_finite() { 354 buffers.failed.store(true, Ordering::Release); 355 return; 356 } 357 record_peak(&buffers.microphone_peak, *output); 358 } 359 if !capture.push(frame, rate, &buffers.capture) { 360 capture.reset(); 361 buffers.capture_dropped.store(true, Ordering::Release); 362 return; 363 } 364 } 365 }, 366 move |error| handle_stream_error(&failure, error), 367 /*timeout*/ None, 368 ) 369 } 370 371 fn build_output<T>( 372 device: &cpal::Device, 373 config: &cpal::StreamConfig, 374 buffers: Arc<Buffers>, 375 ) -> Result<cpal::Stream, cpal::Error> 376 where 377 T: SizedSample + FromSample<f32>, 378 f32: FromSample<T>, 379 { 380 let channels = usize::from(config.channels); 381 let rate = f64::from(config.sample_rate); 382 let failure = buffers.clone(); 383 let mut output = OutputState::default(); 384 device.build_output_stream( 385 *config, 386 move |data: &mut [T], info: &cpal::OutputCallbackInfo| { 387 let timestamp = info.timestamp(); 388 let start = Instant::now() 389 + timestamp 390 .playback 391 .checked_duration_since(timestamp.callback) 392 .unwrap_or_default(); 393 render_output(data, channels, rate, start, &buffers, &mut output); 394 }, 395 move |error| handle_stream_error(&failure, error), 396 /*timeout*/ None, 397 ) 398 } 399 400 #[derive(Default)] 401 struct OutputState { 402 playback: Playback, 403 reference: FramePacker, 404 } 405 406 fn render_output<T>( 407 data: &mut [T], 408 channels: usize, 409 rate: f64, 410 start: Instant, 411 buffers: &Buffers, 412 output: &mut OutputState, 413 ) where 414 T: SizedSample + FromSample<f32>, 415 f32: FromSample<T>, 416 { 417 if !buffers.serviced.load(Ordering::Acquire) { 418 data.fill(T::from_sample(0.0)); 419 return; 420 } 421 for (index, chunk) in data.chunks_mut(BLOCK * channels).enumerate() { 422 let mut reference = Frame { 423 samples: [0.0; BLOCK], 424 len: chunk.len() / channels, 425 at: start + Duration::from_secs_f64((index * BLOCK) as f64 / rate), 426 generation: buffers.speaker.load(Ordering::Acquire), 427 }; 428 for (frame, sample) in chunk.chunks_mut(channels).zip(&mut reference.samples) { 429 let rendered = T::from_sample(output.playback.next(buffers).unwrap_or(0.0)); 430 frame.fill(rendered); 431 *sample = f32::from_sample(rendered); 432 record_peak(&buffers.speaker_peak, *sample); 433 } 434 if !output.reference.push(reference, rate, &buffers.rendered) { 435 output.reference.reset(); 436 buffers.render_dropped.store(true, Ordering::Release); 437 } 438 } 439 } 440 441 #[cfg(test)] 442 #[path = "devices_tests.rs"] 443 mod tests;