rickub
nandi/jolt-nativepublic⑂ Fork 0
⑂ cfd3e3677bed92e80cfe447469a249cfdc0b1401
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/jolt-native.git
git clone ssh://git@rickub.com/nandi/jolt-native.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

mod.rs · 1543 lines · 54.7 KBRust Blame HistoryRaw
Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago1//! ALSA backend implementation.
2//!
3//! Default backend on Linux and BSD systems.
4
5extern crate alsa;
6extern crate libc;
7
8use std::{
9 cmp,
10 sync::{
11 atomic::{AtomicBool, AtomicUsize, Ordering},
12 Arc,
13 },
14 thread::{self, JoinHandle},
15 time::Duration,
16 vec::IntoIter as VecIntoIter,
17};
18
19use self::alsa::poll::Descriptors;
20pub use self::enumerate::Devices;
21
22use crate::{
23 host::fill_with_equilibrium,
24 iter::{SupportedInputConfigs, SupportedOutputConfigs},
25 traits::{DeviceTrait, HostTrait, StreamTrait},
26 BackendSpecificError, BufferSize, BuildStreamError, ChannelCount, Data,
27 DefaultStreamConfigError, DeviceDescription, DeviceDescriptionBuilder, DeviceDirection,
28 DeviceId, DeviceIdError, DeviceNameError, DevicesError, FrameCount, InputCallbackInfo,
29 OutputCallbackInfo, PauseStreamError, PlayStreamError, SampleFormat, SampleRate, StreamConfig,
30 StreamError, SupportedBufferSize, SupportedStreamConfig, SupportedStreamConfigRange,
31 SupportedStreamConfigsError,
32};
33
34mod enumerate;
35
36// ALSA Buffer Size Behavior
37// =========================
38//
39// ## ALSA Latency Model
40//
41// **Hardware vs Software Buffer**: ALSA maintains a software buffer in memory that feeds
42// a hardware buffer in the audio device. Audio latency is determined by how much data
43// sits in the software buffer before being transferred to hardware.
44//
45// **Period-Based Transfer**: ALSA transfers data in chunks called "periods". When one
46// period worth of data has been consumed by hardware, ALSA triggers a callback to refill
47// that period in the software buffer.
48//
49// ## BufferSize::Fixed Behavior
50//
51// When `BufferSize::Fixed(x)` is specified, cpal attempts to configure the period size
52// to approximately `x` frames to achieve the requested callback size. However, the
53// actual callback size may differ from the request:
54//
55// - ALSA may round the period size to hardware-supported values
56// - Different devices have different period size constraints
57// - The callback size is not guaranteed to exactly match the request
58// - If the requested size cannot be accommodated, ALSA will choose the nearest
59// supported configuration
60//
61// This mirrors the behavior documented in the cpal API where `BufferSize::Fixed(x)`
62// requests but does not guarantee a specific callback size.
63//
64// ## BufferSize::Default Behavior
65//
66// When `BufferSize::Default` is specified, cpal does NOT set explicit period size or
67// period count constraints, allowing the device/driver to choose sensible defaults.
68//
69// **Why not set defaults?** Different audio systems have different behaviors:
70//
71// - **Native ALSA hardware**: Typically chooses reasonable defaults (e.g., 512-2048
72// frame periods with 2-4 periods)
73//
74// - **PipeWire-ALSA plugin**: Allocates a large ring buffer (~1M frames at 48kHz) but
75// uses small periods (512-1024 frames). Critically, if you request `set_periods(2)`
76// without specifying period size, PipeWire calculates period = buffer/2, resulting
77// in pathologically large periods (~524K frames = 10 seconds). See issues #1029 and
78// #1036.
79//
80// By not constraining period configuration, PipeWire-ALSA can use its optimized defaults
81// (small periods with many-period buffer), while native ALSA hardware uses its own defaults.
82//
83// **Startup latency**: Regardless of buffer size, cpal uses double-buffering for startup
84// (start_threshold = 2 periods), ensuring low latency even with large multi-period ring
85// buffers.
86
87const DEFAULT_DEVICE: &str = "default";
88
89// Some ALSA plugins (e.g. alsaequal, certain USB drivers) are not reentrant.
90static ALSA_OPEN_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
91
92// TODO: Not yet defined in rust-lang/libc crate
93const LIBC_ENOTSUPP: libc::c_int = 524;
94
95/// The default Linux and BSD host type.
96#[derive(Debug, Clone)]
97pub struct Host {
98 inner: Arc<AlsaContext>,
99}
100
101impl Host {
102 pub fn new() -> Result<Self, crate::HostUnavailable> {
103 let inner = AlsaContext::new().map_err(|_| crate::HostUnavailable)?;
104 Ok(Host {
105 inner: Arc::new(inner),
106 })
107 }
108}
109
110impl HostTrait for Host {
111 type Devices = Devices;
112 type Device = Device;
113
114 fn is_available() -> bool {
115 // Assume ALSA is always available on Linux and BSD.
116 true
117 }
118
119 fn devices(&self) -> Result<Self::Devices, DevicesError> {
120 self.enumerate_devices()
121 }
122
123 fn device_by_id(&self, id: &crate::DeviceId) -> Option<Self::Device> {
124 let canonical_id = crate::DeviceId(id.0, canonical_pcm_id(&id.1));
125 self.devices()
126 .ok()?
127 .find(|d| d.id().ok().as_ref() == Some(&canonical_id))
128 }
129
130 fn default_input_device(&self) -> Option<Self::Device> {
131 Some(Device::default())
132 }
133
134 fn default_output_device(&self) -> Option<Self::Device> {
135 Some(Device::default())
136 }
137}
138
139/// Global count of active ALSA context instances.
140static ALSA_CONTEXT_COUNT: AtomicUsize = AtomicUsize::new(0);
141
142/// ALSA backend context shared between `Host`, `Device`, and `Stream` via `Arc`.
143#[derive(Debug)]
144pub(super) struct AlsaContext;
145
146impl AlsaContext {
147 fn new() -> Result<Self, alsa::Error> {
148 // Initialize global ALSA config cache on first context creation.
149 if ALSA_CONTEXT_COUNT.fetch_add(1, Ordering::SeqCst) == 0 {
150 alsa::config::update()?;
151 }
152 Ok(Self)
153 }
154}
155
156impl Drop for AlsaContext {
157 fn drop(&mut self) {
158 // Free the global ALSA config cache when the last context is dropped.
159 if ALSA_CONTEXT_COUNT.fetch_sub(1, Ordering::SeqCst) == 1 {
160 let _ = alsa::config::update_free_global();
161 }
162 }
163}
164
165impl DeviceTrait for Device {
166 type SupportedInputConfigs = SupportedInputConfigs;
167 type SupportedOutputConfigs = SupportedOutputConfigs;
168 type Stream = Stream;
169
170 // ALSA overrides name() to return pcm_id directly instead of from description
171 fn name(&self) -> Result<String, DeviceNameError> {
172 Device::name(self)
173 }
174
175 fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
176 Device::description(self)
177 }
178
179 fn id(&self) -> Result<DeviceId, DeviceIdError> {
180 Device::id(self)
181 }
182
183 // Override trait defaults to avoid opening devices during enumeration.
184 //
185 // ALSA does not guarantee transactional cleanup on failed snd_pcm_open(). Opening plugins like
186 // alsaequal that fail with EPERM can leak FDs, poisoning the ALSA backend for the process
187 // lifetime (subsequent device opens fail with EBUSY until process exit).
188 fn supports_input(&self) -> bool {
189 matches!(
190 self.direction,
191 DeviceDirection::Input | DeviceDirection::Duplex
192 )
193 }
194
195 fn supports_output(&self) -> bool {
196 matches!(
197 self.direction,
198 DeviceDirection::Output | DeviceDirection::Duplex
199 )
200 }
201
202 fn supported_input_configs(
203 &self,
204 ) -> Result<Self::SupportedInputConfigs, SupportedStreamConfigsError> {
205 Device::supported_input_configs(self)
206 }
207
208 fn supported_output_configs(
209 &self,
210 ) -> Result<Self::SupportedOutputConfigs, SupportedStreamConfigsError> {
211 Device::supported_output_configs(self)
212 }
213
214 fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
215 Device::default_input_config(self)
216 }
217
218 fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
219 Device::default_output_config(self)
220 }
221
222 fn build_input_stream_raw<D, E>(
223 &self,
224 conf: StreamConfig,
225 sample_format: SampleFormat,
226 data_callback: D,
227 error_callback: E,
228 timeout: Option<Duration>,
229 ) -> Result<Self::Stream, BuildStreamError>
230 where
231 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
232 E: FnMut(StreamError) + Send + 'static,
233 {
234 let stream_inner =
235 self.build_stream_inner(conf, sample_format, alsa::Direction::Capture)?;
236 let stream = Self::Stream::new_input(
237 Arc::new(stream_inner),
238 data_callback,
239 error_callback,
240 timeout,
241 );
242 Ok(stream)
243 }
244
245 fn build_output_stream_raw<D, E>(
246 &self,
247 conf: StreamConfig,
248 sample_format: SampleFormat,
249 data_callback: D,
250 error_callback: E,
251 timeout: Option<Duration>,
252 ) -> Result<Self::Stream, BuildStreamError>
253 where
254 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
255 E: FnMut(StreamError) + Send + 'static,
256 {
257 let stream_inner =
258 self.build_stream_inner(conf, sample_format, alsa::Direction::Playback)?;
259 let stream = Self::Stream::new_output(
260 Arc::new(stream_inner),
261 data_callback,
262 error_callback,
263 timeout,
264 );
265 Ok(stream)
266 }
267}
268
269#[derive(Debug)]
270struct TriggerSender(libc::c_int);
271
272#[derive(Debug)]
273struct TriggerReceiver(libc::c_int);
274
275impl TriggerSender {
276 fn wakeup(&self) {
277 let buf = 1u64;
278 loop {
279 let ret = unsafe { libc::write(self.0, &buf as *const u64 as *const _, 8) };
280 if ret == 8 {
281 return;
282 }
283 // write() can be interrupted by a signal before writing any bytes; retry.
284 assert_eq!(ret, -1, "wakeup: unexpected return value {ret}");
285 if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted {
286 panic!("wakeup: {}", std::io::Error::last_os_error());
287 }
288 }
289 }
290}
291
292impl TriggerReceiver {
293 fn clear_pipe(&self) {
294 let mut out = 0u64;
295 loop {
296 let ret = unsafe { libc::read(self.0, &mut out as *mut u64 as *mut _, 8) };
297 if ret == 8 {
298 return;
299 }
300 // read() can be interrupted by a signal before reading any bytes; retry.
301 assert_eq!(ret, -1, "clear_pipe: unexpected return value {ret}");
302 if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted {
303 panic!("clear_pipe: {}", std::io::Error::last_os_error());
304 }
305 }
306 }
307}
308
309fn trigger() -> (TriggerSender, Arc<TriggerReceiver>) {
310 let mut fds = [0, 0];
311 match unsafe { libc::pipe(fds.as_mut_ptr()) } {
312 0 => (TriggerSender(fds[1]), Arc::new(TriggerReceiver(fds[0]))),
313 _ => panic!("Could not create pipe"),
314 }
315}
316
317impl Drop for TriggerSender {
318 fn drop(&mut self) {
319 unsafe {
320 libc::close(self.0);
321 }
322 }
323}
324
325impl Drop for TriggerReceiver {
326 fn drop(&mut self) {
327 unsafe {
328 libc::close(self.0);
329 }
330 }
331}
332
333#[derive(Clone, Debug)]
334pub struct Device {
335 pcm_id: String,
336 desc: Option<String>,
337 direction: DeviceDirection,
338 _context: Arc<AlsaContext>,
339}
340
341impl PartialEq for Device {
342 fn eq(&self, other: &Self) -> bool {
343 self.pcm_id == other.pcm_id
344 }
345}
346
347impl Eq for Device {}
348
349impl std::hash::Hash for Device {
350 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
351 self.pcm_id.hash(state);
352 }
353}
354
355impl Device {
356 fn build_stream_inner(
357 &self,
358 conf: StreamConfig,
359 sample_format: SampleFormat,
360 stream_type: alsa::Direction,
361 ) -> Result<StreamInner, BuildStreamError> {
362 // Validate buffer size if Fixed is specified. This is necessary because
363 // `set_period_size_near()` with `ValueOr::Nearest` will accept ANY value and return the
364 // "nearest" supported value, which could be wildly different (e.g., requesting 4096 frames
365 // might return 512 frames if that's "nearest").
366 if let BufferSize::Fixed(requested_size) = conf.buffer_size {
367 // Note: We use `default_input_config`/`default_output_config` to get the buffer size
368 // range. This queries the CURRENT device (`self.pcm_id`), not the default device. The
369 // buffer size range is the same across all format configurations for a given device
370 // (see `supported_configs()`).
371 let supported_config = match stream_type {
372 alsa::Direction::Capture => self.default_input_config(),
373 alsa::Direction::Playback => self.default_output_config(),
374 };
375 if let Ok(config) = supported_config {
376 if let SupportedBufferSize::Range { min, max } = config.buffer_size {
377 if !(min..=max).contains(&requested_size) {
378 return Err(BuildStreamError::StreamConfigNotSupported);
379 }
380 }
381 }
382 }
383
384 let open_result = {
385 let _guard = ALSA_OPEN_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
386 alsa::pcm::PCM::new(&self.pcm_id, stream_type, true).map_err(|e| (e, e.errno()))
387 };
388 let handle = match open_result {
389 Err((_, libc::ENOENT))
390 | Err((_, libc::EPERM))
391 | Err((_, libc::ENODEV))
392 | Err((_, LIBC_ENOTSUPP)) => return Err(BuildStreamError::DeviceNotAvailable),
393 Err((_, libc::EBUSY)) | Err((_, libc::EAGAIN)) => {
394 return Err(BuildStreamError::DeviceBusy)
395 }
396 Err((_, libc::EINVAL)) => return Err(BuildStreamError::InvalidArgument),
397 Err((e, _)) => return Err(e.into()),
398 Ok(handle) => handle,
399 };
400
401 let can_pause = set_hw_params_from_format(&handle, conf, sample_format)?;
402 let period_samples = set_sw_params_from_format(&handle, conf, stream_type)?;
403
404 handle.prepare()?;
405
406 let num_descriptors = handle.count();
407 if num_descriptors == 0 {
408 let description = "poll descriptor count for stream was 0".to_string();
409 let err = BackendSpecificError { description };
410 return Err(err.into());
411 }
412
413 // Check to see if we can retrieve valid timestamps from the device.
414 // Related: https://bugs.freedesktop.org/show_bug.cgi?id=88503
415 let ts = handle.status()?.get_htstamp();
416 let creation_instant = std::time::Instant::now();
417 let use_hw_timestamps = !(ts.tv_sec == 0 && ts.tv_nsec == 0);
418
419 if let alsa::Direction::Capture = stream_type {
420 handle.start()?;
421 }
422
423 // Pre-compute a period-sized buffer filled with silence values.
424 let period_frames = period_samples / conf.channels as usize;
425 let frame_size = sample_format.sample_size() * conf.channels as usize;
426 let period_bytes = period_frames * frame_size;
427 let mut silence_template = vec![0u8; period_bytes].into_boxed_slice();
428
429 // Only fill buffer for unsigned formats that don't have a zero value for silence.
430 if sample_format.is_uint() {
431 fill_with_equilibrium(&mut silence_template, sample_format);
432 }
433
434 let stream_inner = StreamInner {
435 dropping: AtomicBool::new(false),
436 channel: handle,
437 sample_format,
438 num_descriptors,
439 conf,
440 period_samples,
441 period_frames,
442 frame_size,
443 silence_template,
444 can_pause,
445 creation_instant,
446 use_hw_timestamps,
447 _context: self._context.clone(),
448 };
449
450 Ok(stream_inner)
451 }
452
453 fn name(&self) -> Result<String, DeviceNameError> {
454 Ok(self.pcm_id.clone())
455 }
456
457 fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
458 let name = self
459 .desc
460 .as_ref()
461 .and_then(|desc| desc.lines().next())
462 .unwrap_or(&self.pcm_id)
463 .to_string();
464
465 let mut builder = DeviceDescriptionBuilder::new(name)
466 .driver(self.pcm_id.clone())
467 .direction(self.direction);
468
469 if let Some(ref desc) = self.desc {
470 let lines = desc
471 .lines()
472 .map(|line| line.trim().to_string())
473 .filter(|line| !line.is_empty())
474 .collect();
475 builder = builder.extended(lines);
476 }
477
478 Ok(builder.build())
479 }
480
481 fn id(&self) -> Result<DeviceId, DeviceIdError> {
482 Ok(DeviceId(crate::platform::HostId::Alsa, self.pcm_id.clone()))
483 }
484
485 fn supported_configs(
486 &self,
487 stream_t: alsa::Direction,
488 ) -> Result<VecIntoIter<SupportedStreamConfigRange>, SupportedStreamConfigsError> {
489 let open_result = {
490 let _guard = ALSA_OPEN_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
491 alsa::pcm::PCM::new(&self.pcm_id, stream_t, true).map_err(|e| (e, e.errno()))
492 };
493 let pcm = match open_result {
494 Err((_, libc::ENOENT))
495 | Err((_, libc::EPERM))
496 | Err((_, libc::ENODEV))
497 | Err((_, LIBC_ENOTSUPP)) => {
498 return Err(SupportedStreamConfigsError::DeviceNotAvailable)
499 }
500 Err((_, libc::EBUSY)) | Err((_, libc::EAGAIN)) => {
501 return Err(SupportedStreamConfigsError::DeviceBusy)
502 }
503 Err((_, libc::EINVAL)) => return Err(SupportedStreamConfigsError::InvalidArgument),
504 Err((e, _)) => return Err(e.into()),
505 Ok(pcm) => pcm,
506 };
507
508 let hw_params = alsa::pcm::HwParams::any(&pcm)?;
509
510 // Test both LE and BE formats to detect what the hardware actually supports.
511 // LE is listed first as it's the common case for most audio hardware.
512 // Hardware reports its supported formats regardless of CPU endianness.
513 const FORMATS: [(SampleFormat, alsa::pcm::Format); 23] = [
514 (SampleFormat::I8, alsa::pcm::Format::S8),
515 (SampleFormat::U8, alsa::pcm::Format::U8),
516 (SampleFormat::I16, alsa::pcm::Format::S16LE),
517 (SampleFormat::I16, alsa::pcm::Format::S16BE),
518 (SampleFormat::U16, alsa::pcm::Format::U16LE),
519 (SampleFormat::U16, alsa::pcm::Format::U16BE),
520 (SampleFormat::I24, alsa::pcm::Format::S24LE),
521 (SampleFormat::I24, alsa::pcm::Format::S24BE),
522 (SampleFormat::U24, alsa::pcm::Format::U24LE),
523 (SampleFormat::U24, alsa::pcm::Format::U24BE),
524 (SampleFormat::I32, alsa::pcm::Format::S32LE),
525 (SampleFormat::I32, alsa::pcm::Format::S32BE),
526 (SampleFormat::U32, alsa::pcm::Format::U32LE),
527 (SampleFormat::U32, alsa::pcm::Format::U32BE),
528 (SampleFormat::F32, alsa::pcm::Format::FloatLE),
529 (SampleFormat::F32, alsa::pcm::Format::FloatBE),
530 (SampleFormat::F64, alsa::pcm::Format::Float64LE),
531 (SampleFormat::F64, alsa::pcm::Format::Float64BE),
532 (SampleFormat::DsdU8, alsa::pcm::Format::DSDU8),
533 (SampleFormat::DsdU16, alsa::pcm::Format::DSDU16LE),
534 (SampleFormat::DsdU16, alsa::pcm::Format::DSDU16BE),
535 (SampleFormat::DsdU32, alsa::pcm::Format::DSDU32LE),
536 (SampleFormat::DsdU32, alsa::pcm::Format::DSDU32BE),
537 //SND_PCM_FORMAT_IEC958_SUBFRAME_LE,
538 //SND_PCM_FORMAT_IEC958_SUBFRAME_BE,
539 //SND_PCM_FORMAT_MU_LAW,
540 //SND_PCM_FORMAT_A_LAW,
541 //SND_PCM_FORMAT_IMA_ADPCM,
542 //SND_PCM_FORMAT_MPEG,
543 //SND_PCM_FORMAT_GSM,
544 //SND_PCM_FORMAT_SPECIAL,
545 //SND_PCM_FORMAT_S24_3LE,
546 //SND_PCM_FORMAT_S24_3BE,
547 //SND_PCM_FORMAT_U24_3LE,
548 //SND_PCM_FORMAT_U24_3BE,
549 //SND_PCM_FORMAT_S20_3LE,
550 //SND_PCM_FORMAT_S20_3BE,
551 //SND_PCM_FORMAT_U20_3LE,
552 //SND_PCM_FORMAT_U20_3BE,
553 //SND_PCM_FORMAT_S18_3LE,
554 //SND_PCM_FORMAT_S18_3BE,
555 //SND_PCM_FORMAT_U18_3LE,
556 //SND_PCM_FORMAT_U18_3BE,
557 ];
558
559 // Collect supported formats, deduplicating since we test both LE and BE variants.
560 // If hardware supports both endiannesses (rare), we only report the format once.
561 let mut supported_formats = Vec::new();
562 for &(sample_format, alsa_format) in FORMATS.iter() {
563 if hw_params.test_format(alsa_format).is_ok()
564 && !supported_formats.contains(&sample_format)
565 {
566 supported_formats.push(sample_format);
567 }
568 }
569
570 let min_rate = hw_params.get_rate_min()?;
571 let max_rate = hw_params.get_rate_max()?;
572
573 let sample_rates = if min_rate == max_rate || hw_params.test_rate(min_rate + 1).is_ok() {
574 vec![(min_rate, max_rate)]
575 } else {
576 let mut rates = Vec::new();
577 for &sample_rate in crate::COMMON_SAMPLE_RATES.iter() {
578 if hw_params.test_rate(sample_rate).is_ok() {
579 rates.push((sample_rate, sample_rate));
580 }
581 }
582
583 if rates.is_empty() {
584 vec![(min_rate, max_rate)]
585 } else {
586 rates
587 }
588 };
589
590 let min_channels = hw_params.get_channels_min()?;
591 let max_channels = hw_params.get_channels_max()?;
592
593 let max_channels = cmp::min(max_channels, 32); // TODO: limiting to 32 channels or too much stuff is returned
594 let supported_channels = (min_channels..max_channels + 1)
595 .filter_map(|num| {
596 if hw_params.test_channels(num).is_ok() {
597 Some(num as ChannelCount)
598 } else {
599 None
600 }
601 })
602 .collect::<Vec<_>>();
603
604 let (min_buffer_size, max_buffer_size) = hw_params_buffer_size_min_max(&hw_params);
605 let buffer_size_range = SupportedBufferSize::Range {
606 min: min_buffer_size,
607 max: max_buffer_size,
608 };
609
610 let mut output = Vec::with_capacity(
611 supported_formats.len() * supported_channels.len() * sample_rates.len(),
612 );
613 for &sample_format in supported_formats.iter() {
614 for &channels in supported_channels.iter() {
615 for &(min_rate, max_rate) in sample_rates.iter() {
616 output.push(SupportedStreamConfigRange {
617 channels,
618 min_sample_rate: min_rate,
619 max_sample_rate: max_rate,
620 buffer_size: buffer_size_range,
621 sample_format,
622 });
623 }
624 }
625 }
626
627 Ok(output.into_iter())
628 }
629
630 fn supported_input_configs(
631 &self,
632 ) -> Result<SupportedInputConfigs, SupportedStreamConfigsError> {
633 self.supported_configs(alsa::Direction::Capture)
634 }
635
636 fn supported_output_configs(
637 &self,
638 ) -> Result<SupportedOutputConfigs, SupportedStreamConfigsError> {
639 self.supported_configs(alsa::Direction::Playback)
640 }
641
642 // ALSA does not offer default stream formats, so instead we compare all supported formats by
643 // the `SupportedStreamConfigRange::cmp_default_heuristics` order and select the greatest.
644 fn default_config(
645 &self,
646 stream_t: alsa::Direction,
647 ) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
648 let mut formats: Vec<_> = {
649 match self.supported_configs(stream_t) {
650 Err(SupportedStreamConfigsError::DeviceNotAvailable) => {
651 return Err(DefaultStreamConfigError::DeviceNotAvailable);
652 }
653 Err(SupportedStreamConfigsError::DeviceBusy) => {
654 return Err(DefaultStreamConfigError::DeviceBusy);
655 }
656 Err(SupportedStreamConfigsError::InvalidArgument) => {
657 // this happens sometimes when querying for input and output capabilities, but
658 // the device supports only one
659 return Err(DefaultStreamConfigError::StreamTypeNotSupported);
660 }
661 Err(SupportedStreamConfigsError::BackendSpecific { err }) => {
662 return Err(err.into());
663 }
664 Ok(fmts) => fmts.collect(),
665 }
666 };
667
668 formats.sort_by(|a, b| a.cmp_default_heuristics(b));
669
670 match formats.into_iter().next_back() {
671 Some(f) => {
672 let min_r = f.min_sample_rate;
673 let max_r = f.max_sample_rate;
674 let mut format = f.with_max_sample_rate();
675 const HZ_44100: SampleRate = 44_100;
676 if min_r <= HZ_44100 && HZ_44100 <= max_r {
677 format.sample_rate = HZ_44100;
678 }
679 Ok(format)
680 }
681 None => Err(DefaultStreamConfigError::StreamTypeNotSupported),
682 }
683 }
684
685 fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
686 self.default_config(alsa::Direction::Capture)
687 }
688
689 fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
690 self.default_config(alsa::Direction::Playback)
691 }
692}
693
694impl Default for Device {
695 fn default() -> Self {
696 // "default" is a virtual ALSA device that redirects to the configured default. We cannot
697 // determine its actual capabilities without opening it, so we return Unknown direction.
698 Self {
699 pcm_id: DEFAULT_DEVICE.to_owned(),
700 desc: Some("Default Audio Device".to_string()),
701 direction: DeviceDirection::Unknown,
702 _context: Arc::new(
703 AlsaContext::new().expect("Failed to initialize ALSA configuration"),
704 ),
705 }
706 }
707}
708
709#[derive(Debug)]
710struct StreamInner {
711 // Flag used to check when to stop polling, regardless of the state of the stream
712 // (e.g. broken due to a disconnected device).
713 dropping: AtomicBool,
714
715 // The ALSA channel.
716 channel: alsa::pcm::PCM,
717
718 // When converting between file descriptors and `snd_pcm_t`, this is the number of
719 // file descriptors that this `snd_pcm_t` uses.
720 num_descriptors: usize,
721
722 // Format of the samples.
723 sample_format: SampleFormat,
724
725 // The configuration used to open this stream.
726 conf: StreamConfig,
727
728 // Cached values for performance in audio callback hot path
729 period_samples: usize,
730 period_frames: usize,
731 frame_size: usize,
732 silence_template: Box<[u8]>,
733
734 #[allow(dead_code)]
735 // Whether or not the hardware supports pausing the stream.
736 // TODO: We need an API to expose this. See #197, #284.
737 can_pause: bool,
738
739 // Whether to attempt hardware timestamps via `get_htstamp` / `get_trigger_htstamp`.
740 //
741 // When `true`, hardware timestamps are tried first on every callback and we fall back silently
742 // to `creation_instant` if they are transiently unavailable; e.g. the PulseAudio ALSA plugin
743 // returns `(0, 0)` for the first several periods after the stream is triggered.
744 use_hw_timestamps: bool,
745
746 // Timestamp origin used by the fallback path. Faster without `Option`.
747 creation_instant: std::time::Instant,
748
749 // Keep ALSA context alive to prevent premature ALSA config cleanup
750 _context: Arc<AlsaContext>,
751}
752
753// Assume that the ALSA library is built with thread safe option.
754unsafe impl Sync for StreamInner {}
755
756#[derive(Debug)]
757pub struct Stream {
758 /// The high-priority audio processing thread calling callbacks.
759 /// Option used for moving out in destructor.
760 thread: Option<JoinHandle<()>>,
761
762 /// Handle to the underlying stream for playback controls.
763 inner: Arc<StreamInner>,
764
765 /// Used to signal to stop processing.
766 trigger: TriggerSender,
767
768 /// Keeps the read end of the self-pipe alive for the lifetime of the Stream, so that
769 /// `trigger.wakeup()` never writes to a closed pipe, even if the worker exited early.
770 _rx: Arc<TriggerReceiver>,
771}
772
773// Compile-time assertion that Stream is Send and Sync
774crate::assert_stream_send!(Stream);
775crate::assert_stream_sync!(Stream);
776
777struct StreamWorkerContext {
778 descriptors: Box<[libc::pollfd]>,
779 transfer_buffer: Box<[u8]>,
780 poll_timeout: i32,
781}
782
783impl StreamWorkerContext {
784 fn new(poll_timeout: &Option<Duration>, stream: &StreamInner, rx: &TriggerReceiver) -> Self {
785 let poll_timeout: i32 = if let Some(d) = poll_timeout {
786 d.as_millis().try_into().unwrap()
787 } else {
788 -1 // Don't timeout, wait forever.
789 };
790
791 // Pre-allocate buffer to exactly one period size with proper equilibrium values.
792 let transfer_buffer = stream.silence_template.clone();
793
794 // Pre-allocate and initialize descriptors vector: 1 for self-pipe + stream.num_descriptors
795 // for ALSA. The descriptor count is constant for the lifetime of stream parameters, and
796 // poll() overwrites revents on each call, so we only need to set up fd and events once.
797 let total_descriptors = 1 + stream.num_descriptors;
798 let mut descriptors = vec![
799 libc::pollfd {
800 fd: 0,
801 events: 0,
802 revents: 0
803 };
804 total_descriptors
805 ]
806 .into_boxed_slice();
807
808 // Set up self-pipe descriptor at index 0
809 descriptors[0] = libc::pollfd {
810 fd: rx.0,
811 events: libc::POLLIN,
812 revents: 0,
813 };
814
815 // Set up ALSA descriptors starting at index 1
816 let filled = stream
817 .channel
818 .fill(&mut descriptors[1..])
819 .expect("Failed to fill ALSA descriptors");
820 debug_assert_eq!(filled, stream.num_descriptors);
821
822 Self {
823 descriptors,
824 transfer_buffer,
825 poll_timeout,
826 }
827 }
828}
829
830fn input_stream_worker(
831 rx: Arc<TriggerReceiver>,
832 stream: &StreamInner,
833 data_callback: &mut (dyn FnMut(&Data, &InputCallbackInfo) + Send + 'static),
834 error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
835 timeout: Option<Duration>,
836) {
837 boost_current_thread_priority(stream.conf.buffer_size, stream.conf.sample_rate);
838
839 let mut ctxt = StreamWorkerContext::new(&timeout, stream, &rx);
840 loop {
841 if stream.dropping.load(Ordering::Acquire) {
842 return;
843 }
844 let result = match poll_for_period(&rx, stream, &mut ctxt) {
845 Ok(Poll::Pending) => continue,
846 Ok(Poll::Ready {
847 status,
848 delay_frames,
849 }) => process_input(
850 stream,
851 &mut ctxt.transfer_buffer,
852 status,
853 delay_frames,
854 data_callback,
855 ),
856 Err(err) => Err(err),
857 };
858 if let Err(err) = result {
859 match err {
860 StreamError::BufferUnderrun => {
861 error_callback(StreamError::BufferUnderrun);
862 if let Err(err) = stream.channel.prepare() {
863 error_callback(err.into());
864 } else if let Err(err) = stream.channel.start() {
865 error_callback(err.into());
866 }
867 }
868 StreamError::DeviceNotAvailable => {
869 error_callback(StreamError::DeviceNotAvailable);
870 return;
871 }
872 err => error_callback(err),
873 }
874 }
875 }
876}
877
878fn output_stream_worker(
879 rx: Arc<TriggerReceiver>,
880 stream: &StreamInner,
881 data_callback: &mut (dyn FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static),
882 error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
883 timeout: Option<Duration>,
884) {
885 boost_current_thread_priority(stream.conf.buffer_size, stream.conf.sample_rate);
886
887 let mut ctxt = StreamWorkerContext::new(&timeout, stream, &rx);
888
889 loop {
890 if stream.dropping.load(Ordering::Acquire) {
891 return;
892 }
893 let result = match poll_for_period(&rx, stream, &mut ctxt) {
894 Ok(Poll::Pending) => continue,
895 Ok(Poll::Ready {
896 status,
897 delay_frames,
898 }) => process_output(
899 stream,
900 &mut ctxt.transfer_buffer,
901 status,
902 delay_frames,
903 data_callback,
904 ),
905 Err(err) => Err(err),
906 };
907 if let Err(err) = result {
908 match err {
909 StreamError::BufferUnderrun => {
910 error_callback(StreamError::BufferUnderrun);
911 if let Err(err) = stream.channel.prepare() {
912 error_callback(err.into());
913 }
914 // No need to call start() for output streams after prepare();
915 // ALSA automatically restarts them when the buffer is refilled
916 // and the stream is triggered again.
917 }
918 StreamError::DeviceNotAvailable => {
919 error_callback(StreamError::DeviceNotAvailable);
920 return;
921 }
922 err => error_callback(err),
923 }
924 }
925 }
926}
927
928#[cfg(feature = "audio_thread_priority")]
929fn boost_current_thread_priority(buffer_size: BufferSize, sample_rate: SampleRate) {
930 use audio_thread_priority::promote_current_thread_to_real_time;
931
932 let buffer_size = if let BufferSize::Fixed(buffer_size) = buffer_size {
933 buffer_size
934 } else {
935 // if the buffer size isn't fixed, let audio_thread_priority choose a sensible default value
936 0
937 };
938
939 if let Err(err) = promote_current_thread_to_real_time(buffer_size, sample_rate) {
940 eprintln!("Failed to promote audio thread to real-time priority: {err}");
941 }
942}
943
944#[cfg(not(feature = "audio_thread_priority"))]
945fn boost_current_thread_priority(_: BufferSize, _: SampleRate) {}
946
947/// Attempt hardware resume from a suspend event (`ESTRPIPE`).
948fn try_resume(channel: &alsa::PCM) -> Result<Poll, StreamError> {
949 match channel.resume() {
950 // device resumed successfully and will continue running on its own
951 Ok(()) => Ok(Poll::Pending),
952 // device is still resuming; poll again until it is ready.
953 Err(e) if e.errno() == libc::EAGAIN => Ok(Poll::Pending),
954 // hardware does not support soft resume
955 Err(e) if e.errno() == libc::ENOSYS => Err(StreamError::BufferUnderrun),
956 Err(e) => Err(e.into()),
957 }
958}
959
960enum Poll {
961 Pending,
962 Ready {
963 status: alsa::pcm::Status,
964 delay_frames: usize,
965 },
966}
967
968// This block is shared between both input and output stream worker functions.
969fn poll_for_period(
970 rx: &TriggerReceiver,
971 stream: &StreamInner,
972 ctxt: &mut StreamWorkerContext,
973) -> Result<Poll, StreamError> {
974 let StreamWorkerContext {
975 ref mut descriptors,
976 ref poll_timeout,
977 ..
978 } = *ctxt;
979
980 let res = alsa::poll::poll(descriptors, *poll_timeout)?;
981 if res == 0 {
982 // poll() returned 0: either a timeout or a spurious wakeup. Nothing to do.
983 return Ok(Poll::Pending);
984 }
985
986 if descriptors[0].revents != 0 {
987 // Self-pipe fired: the stream is being dropped. Clear the pipe and let the
988 // worker loop detect the dropping flag on the next iteration.
989 rx.clear_pipe();
990 return Ok(Poll::Pending);
991 }
992
993 let revents = stream.channel.revents(&descriptors[1..])?;
994 // No events: spurious wakeup, poll again.
995 if revents.is_empty() {
996 return Ok(Poll::Pending);
997 }
998 // POLLHUP/POLLNVAL: the device has been disconnected.
999 if revents.intersects(alsa::poll::Flags::HUP | alsa::poll::Flags::NVAL) {
1000 return Err(StreamError::DeviceNotAvailable);
1001 }
1002 // POLLERR signals an xrun or suspend; avail() below returns EPIPE/ESTRPIPE accordingly.
1003 // POLLIN/POLLOUT: data is ready, fall through to process it.
1004
1005 let status = stream.channel.status()?;
1006 let avail_frames = match stream.channel.avail() {
1007 // Xrun: recover via prepare() (+ start() for capture, handled by the worker).
1008 Err(err) if err.errno() == libc::EPIPE => return Err(StreamError::BufferUnderrun),
1009 // Suspend: try hardware resume first; fall back to prepare() if unsupported.
1010 Err(err) if err.errno() == libc::ESTRPIPE => return try_resume(&stream.channel),
1011 res => res,
1012 }? as usize;
1013 let delay_frames = match status.get_delay() {
1014 d if d < 0 => 0,
1015 d => d as usize,
1016 };
1017 let available_samples = avail_frames * stream.conf.channels as usize;
1018
1019 // ALSA can have spurious wakeups where poll returns but avail < avail_min.
1020 // This is documented to occur with dmix (timer-driven) and other plugins.
1021 // Verify we have room for at least one full period before processing.
1022 // See: https://bugzilla.kernel.org/show_bug.cgi?id=202499
1023 if available_samples < stream.period_samples {
1024 return Ok(Poll::Pending);
1025 }
1026
1027 Ok(Poll::Ready {
1028 status,
1029 delay_frames,
1030 })
1031}
1032
1033// Read input data from ALSA and deliver it to the user.
1034fn process_input(
1035 stream: &StreamInner,
1036 buffer: &mut [u8],
1037 status: alsa::pcm::Status,
1038 delay_frames: usize,
1039 data_callback: &mut (dyn FnMut(&Data, &InputCallbackInfo) + Send + 'static),
1040) -> Result<(), StreamError> {
1041 let mut frames_read = 0;
1042 while frames_read < stream.period_frames {
1043 match stream
1044 .channel
1045 .io_bytes()
1046 .readi(&mut buffer[frames_read * stream.frame_size..])
1047 {
1048 Ok(n) => frames_read += n,
1049 Err(err) if err.errno() == libc::EPIPE || err.errno() == libc::ESTRPIPE => {
1050 // EPIPE = xrun, ESTRPIPE = hardware suspend. Both require prepare()+restart;
1051 // attempting resume mid-loop with a partial transfer in the ring buffer is unsafe.
1052 return Err(StreamError::BufferUnderrun);
1053 }
1054 Err(err) if err.errno() == libc::ENODEV => return Err(StreamError::DeviceNotAvailable),
1055 Err(err) => return Err(err.into()),
1056 }
1057 }
1058 let data = buffer.as_mut_ptr() as *mut ();
1059 let data = unsafe { Data::from_parts(data, stream.period_samples, stream.sample_format) };
1060 let callback = if stream.use_hw_timestamps {
1061 stream_timestamp_hardware(&status)
1062 .or_else(|_| stream_timestamp_fallback(stream.creation_instant))
1063 } else {
1064 stream_timestamp_fallback(stream.creation_instant)
1065 }?;
1066 let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
1067 let capture = callback
1068 .sub(delay_duration)
1069 .ok_or_else(|| BackendSpecificError {
1070 description: "`capture` is earlier than representation supported by `StreamInstant`"
1071 .to_string(),
1072 })?;
1073 let timestamp = crate::InputStreamTimestamp { callback, capture };
1074 let info = crate::InputCallbackInfo { timestamp };
1075 data_callback(&data, &info);
1076
1077 Ok(())
1078}
1079
1080// Request data from the user's function and write it via ALSA.
1081fn process_output(
1082 stream: &StreamInner,
1083 buffer: &mut [u8],
1084 status: alsa::pcm::Status,
1085 delay_frames: usize,
1086 data_callback: &mut (dyn FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static),
1087) -> Result<(), StreamError> {
1088 // Buffer is always pre-filled with equilibrium, user overwrites what they want
1089 buffer.copy_from_slice(&stream.silence_template);
1090 {
1091 let data = buffer.as_mut_ptr() as *mut ();
1092 let mut data =
1093 unsafe { Data::from_parts(data, stream.period_samples, stream.sample_format) };
1094 let callback = if stream.use_hw_timestamps {
1095 stream_timestamp_hardware(&status)
1096 .or_else(|_| stream_timestamp_fallback(stream.creation_instant))
1097 } else {
1098 stream_timestamp_fallback(stream.creation_instant)
1099 }?;
1100 let delay_duration = frames_to_duration(delay_frames, stream.conf.sample_rate);
1101 let playback = callback
1102 .add(delay_duration)
1103 .ok_or_else(|| BackendSpecificError {
1104 description: "`playback` occurs beyond representation supported by `StreamInstant`"
1105 .to_string(),
1106 })?;
1107 let timestamp = crate::OutputStreamTimestamp { callback, playback };
1108 let info = crate::OutputCallbackInfo { timestamp };
1109 data_callback(&mut data, &info);
1110 }
1111
1112 let mut frames_written = 0;
1113 while frames_written < stream.period_frames {
1114 match stream
1115 .channel
1116 .io_bytes()
1117 .writei(&buffer[frames_written * stream.frame_size..])
1118 {
1119 Ok(n) => frames_written += n,
1120 Err(err) if err.errno() == libc::EPIPE || err.errno() == libc::ESTRPIPE => {
1121 // EPIPE = xrun, ESTRPIPE = hardware suspend. Both require prepare()+restart;
1122 // attempting resume mid-loop with a partial transfer in the ring buffer is unsafe.
1123 return Err(StreamError::BufferUnderrun);
1124 }
1125 Err(err) if err.errno() == libc::ENODEV => return Err(StreamError::DeviceNotAvailable),
1126 Err(err) => return Err(err.into()),
1127 }
1128 }
1129 Ok(())
1130}
1131
1132// Use hardware timestamps from ALSA.
1133//
1134// This ensures accurate timestamps based on actual hardware timing.
1135#[inline]
1136fn stream_timestamp_hardware(
1137 status: &alsa::pcm::Status,
1138) -> Result<crate::StreamInstant, BackendSpecificError> {
1139 let trigger_ts = status.get_trigger_htstamp();
1140 // trigger_htstamp records when the PCM stream started.
1141 // On the first few callbacks, it might not have been set yet,
1142 // which would yield a huge positive nanos nd cause non-monotonicity
1143 // once it is set. Bail out and let the caller use the fallback.
1144 // See https://github.com/RustAudio/cpal/issues/710
1145 if trigger_ts.tv_sec == 0 && trigger_ts.tv_nsec == 0 {
1146 return Err(BackendSpecificError {
1147 description: "trigger_htstamp not yet set".to_string(),
1148 });
1149 }
1150 let ts = status.get_htstamp();
1151 let nanos = timespec_diff_nanos(ts, trigger_ts);
1152 if nanos < 0 {
1153 let description = format!(
1154 "get_htstamp `{}.{}` was earlier than get_trigger_htstamp `{}.{}`",
1155 ts.tv_sec, ts.tv_nsec, trigger_ts.tv_sec, trigger_ts.tv_nsec
1156 );
1157 return Err(BackendSpecificError { description });
1158 }
1159 Ok(crate::StreamInstant::from_nanos(nanos))
1160}
1161
1162// Use elapsed duration since stream creation as fallback when hardware timestamps are unavailable.
1163//
1164// This ensures positive values that are compatible with our `StreamInstant` representation.
1165#[inline]
1166fn stream_timestamp_fallback(
1167 creation: std::time::Instant,
1168) -> Result<crate::StreamInstant, BackendSpecificError> {
1169 let now = std::time::Instant::now();
1170 let duration = now.duration_since(creation);
1171 crate::StreamInstant::from_nanos_i128(duration.as_nanos() as i128).ok_or(BackendSpecificError {
1172 description: "stream duration has exceeded `StreamInstant` representation".to_string(),
1173 })
1174}
1175
1176// Adapted from `timestamp2ns` here:
1177// https://fossies.org/linux/alsa-lib/test/audio_time.c
1178#[inline]
1179#[allow(clippy::unnecessary_cast)]
1180fn timespec_to_nanos(ts: libc::timespec) -> i64 {
1181 ts.tv_sec as i64 * 1_000_000_000 + ts.tv_nsec as i64
1182}
1183
1184// Adapted from `timediff` here:
1185// https://fossies.org/linux/alsa-lib/test/audio_time.c
1186#[inline]
1187fn timespec_diff_nanos(a: libc::timespec, b: libc::timespec) -> i64 {
1188 timespec_to_nanos(a) - timespec_to_nanos(b)
1189}
1190
1191// Convert the given duration in frames at the given sample rate to a `std::time::Duration`.
1192#[inline]
1193fn frames_to_duration(frames: usize, rate: crate::SampleRate) -> std::time::Duration {
1194 let secsf = frames as f64 / rate as f64;
1195 let secs = secsf as u64;
1196 let nanos = ((secsf - secs as f64) * 1_000_000_000.0) as u32;
1197 std::time::Duration::new(secs, nanos)
1198}
1199
1200impl Stream {
1201 fn new_input<D, E>(
1202 inner: Arc<StreamInner>,
1203 mut data_callback: D,
1204 mut error_callback: E,
1205 timeout: Option<Duration>,
1206 ) -> Stream
1207 where
1208 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
1209 E: FnMut(StreamError) + Send + 'static,
1210 {
1211 let (tx, rx) = trigger();
1212 let rx_thread = rx.clone();
1213 let stream = inner.clone();
1214 let thread = thread::Builder::new()
1215 .name("cpal_alsa_in".to_owned())
1216 .spawn(move || {
1217 input_stream_worker(
1218 rx_thread,
1219 &stream,
1220 &mut data_callback,
1221 &mut error_callback,
1222 timeout,
1223 );
1224 })
1225 .unwrap();
1226 Self {
1227 thread: Some(thread),
1228 inner,
1229 trigger: tx,
1230 _rx: rx,
1231 }
1232 }
1233
1234 fn new_output<D, E>(
1235 inner: Arc<StreamInner>,
1236 mut data_callback: D,
1237 mut error_callback: E,
1238 timeout: Option<Duration>,
1239 ) -> Stream
1240 where
1241 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
1242 E: FnMut(StreamError) + Send + 'static,
1243 {
1244 let (tx, rx) = trigger();
1245 let rx_thread = rx.clone();
1246 let stream = inner.clone();
1247 let thread = thread::Builder::new()
1248 .name("cpal_alsa_out".to_owned())
1249 .spawn(move || {
1250 output_stream_worker(
1251 rx_thread,
1252 &stream,
1253 &mut data_callback,
1254 &mut error_callback,
1255 timeout,
1256 );
1257 })
1258 .unwrap();
1259 Self {
1260 thread: Some(thread),
1261 inner,
1262 trigger: tx,
1263 _rx: rx,
1264 }
1265 }
1266}
1267
1268impl Drop for Stream {
1269 fn drop(&mut self) {
1270 self.inner.dropping.store(true, Ordering::Release);
1271 self.trigger.wakeup();
1272 if let Some(handle) = self.thread.take() {
1273 let _ = handle.join();
1274 }
1275 }
1276}
1277
1278impl StreamTrait for Stream {
1279 fn play(&self) -> Result<(), PlayStreamError> {
1280 self.inner.channel.pause(false).ok();
1281 Ok(())
1282 }
1283 fn pause(&self) -> Result<(), PauseStreamError> {
1284 self.inner.channel.pause(true).ok();
1285 Ok(())
1286 }
1287 fn buffer_size(&self) -> Option<FrameCount> {
1288 Some(self.inner.period_frames as FrameCount)
1289 }
1290}
1291
1292// Convert ALSA frames to FrameCount, clamping to valid range.
1293// ALSA Frames are i64 (64-bit) or i32 (32-bit).
1294fn clamp_frame_count(buffer_size: alsa::pcm::Frames) -> FrameCount {
1295 buffer_size.max(1).try_into().unwrap_or(FrameCount::MAX)
1296}
1297
1298fn hw_params_buffer_size_min_max(hw_params: &alsa::pcm::HwParams) -> (FrameCount, FrameCount) {
1299 let min_buf = hw_params
1300 .get_buffer_size_min()
1301 .map(clamp_frame_count)
1302 .unwrap_or(1);
1303 let max_buf = hw_params
1304 .get_buffer_size_max()
1305 .map(clamp_frame_count)
1306 .unwrap_or(FrameCount::MAX);
1307 (min_buf, max_buf)
1308}
1309
1310fn init_hw_params<'a>(
1311 pcm_handle: &'a alsa::pcm::PCM,
1312 config: StreamConfig,
1313 sample_format: SampleFormat,
1314) -> Result<alsa::pcm::HwParams<'a>, BackendSpecificError> {
1315 let hw_params = alsa::pcm::HwParams::any(pcm_handle)?;
1316 hw_params.set_access(alsa::pcm::Access::RWInterleaved)?;
1317
1318 // Determine which endianness the hardware actually supports for this format.
1319 // We prefer native endian (no conversion needed) but fall back to the opposite
1320 // endian if that's all the hardware supports (e.g., LE USB DAC on BE system).
1321 let alsa_format = sample_format_to_alsa_format(&hw_params, sample_format)?;
1322 hw_params.set_format(alsa_format)?;
1323
1324 hw_params.set_rate(config.sample_rate, alsa::ValueOr::Nearest)?;
1325 hw_params.set_channels(config.channels as u32)?;
1326 Ok(hw_params)
1327}
1328
1329/// Convert SampleFormat to the appropriate alsa::pcm::Format based on what the hardware supports.
1330/// Prefers native endian, falls back to non-native if that's all the hardware supports.
1331fn sample_format_to_alsa_format(
1332 hw_params: &alsa::pcm::HwParams,
1333 sample_format: SampleFormat,
1334) -> Result<alsa::pcm::Format, BackendSpecificError> {
1335 use alsa::pcm::Format;
1336
1337 // For each sample format, define (native_endian_format, opposite_endian_format) pairs
1338 let (native, opposite) = match sample_format {
1339 SampleFormat::I8 => return Ok(Format::S8), // No endianness
1340 SampleFormat::U8 => return Ok(Format::U8), // No endianness
1341 #[cfg(target_endian = "little")]
1342 SampleFormat::I16 => (Format::S16LE, Format::S16BE),
1343 #[cfg(target_endian = "big")]
1344 SampleFormat::I16 => (Format::S16BE, Format::S16LE),
1345 #[cfg(target_endian = "little")]
1346 SampleFormat::U16 => (Format::U16LE, Format::U16BE),
1347 #[cfg(target_endian = "big")]
1348 SampleFormat::U16 => (Format::U16BE, Format::U16LE),
1349 #[cfg(target_endian = "little")]
1350 SampleFormat::I24 => (Format::S24LE, Format::S24BE),
1351 #[cfg(target_endian = "big")]
1352 SampleFormat::I24 => (Format::S24BE, Format::S24LE),
1353 #[cfg(target_endian = "little")]
1354 SampleFormat::U24 => (Format::U24LE, Format::U24BE),
1355 #[cfg(target_endian = "big")]
1356 SampleFormat::U24 => (Format::U24BE, Format::U24LE),
1357 #[cfg(target_endian = "little")]
1358 SampleFormat::I32 => (Format::S32LE, Format::S32BE),
1359 #[cfg(target_endian = "big")]
1360 SampleFormat::I32 => (Format::S32BE, Format::S32LE),
1361 #[cfg(target_endian = "little")]
1362 SampleFormat::U32 => (Format::U32LE, Format::U32BE),
1363 #[cfg(target_endian = "big")]
1364 SampleFormat::U32 => (Format::U32BE, Format::U32LE),
1365 #[cfg(target_endian = "little")]
1366 SampleFormat::F32 => (Format::FloatLE, Format::FloatBE),
1367 #[cfg(target_endian = "big")]
1368 SampleFormat::F32 => (Format::FloatBE, Format::FloatLE),
1369 #[cfg(target_endian = "little")]
1370 SampleFormat::F64 => (Format::Float64LE, Format::Float64BE),
1371 #[cfg(target_endian = "big")]
1372 SampleFormat::F64 => (Format::Float64BE, Format::Float64LE),
1373 SampleFormat::DsdU8 => return Ok(Format::DSDU8),
1374 #[cfg(target_endian = "little")]
1375 SampleFormat::DsdU16 => (Format::DSDU16LE, Format::DSDU16BE),
1376 #[cfg(target_endian = "big")]
1377 SampleFormat::DsdU16 => (Format::DSDU16BE, Format::DSDU16LE),
1378 #[cfg(target_endian = "little")]
1379 SampleFormat::DsdU32 => (Format::DSDU32LE, Format::DSDU32BE),
1380 #[cfg(target_endian = "big")]
1381 SampleFormat::DsdU32 => (Format::DSDU32BE, Format::DSDU32LE),
1382 _ => {
1383 return Err(BackendSpecificError {
1384 description: format!("Sample format '{sample_format}' is not supported"),
1385 })
1386 }
1387 };
1388
1389 // Try native endian first (optimal - no conversion needed)
1390 if hw_params.test_format(native).is_ok() {
1391 return Ok(native);
1392 }
1393
1394 // Fall back to opposite endian if hardware only supports that
1395 if hw_params.test_format(opposite).is_ok() {
1396 return Ok(opposite);
1397 }
1398
1399 Err(BackendSpecificError {
1400 description: format!(
1401 "Sample format '{sample_format}' is not supported by hardware in any endianness"
1402 ),
1403 })
1404}
1405
1406fn set_hw_params_from_format(
1407 pcm_handle: &alsa::pcm::PCM,
1408 config: StreamConfig,
1409 sample_format: SampleFormat,
1410) -> Result<bool, BackendSpecificError> {
1411 let hw_params = init_hw_params(pcm_handle, config, sample_format)?;
1412
1413 // When BufferSize::Fixed(x) is specified, we configure double-buffering with
1414 // buffer_size = 2x and period_size = x. This provides consistent low-latency
1415 // behavior across different ALSA implementations and hardware.
1416 if let BufferSize::Fixed(buffer_frames) = config.buffer_size {
1417 hw_params.set_buffer_size_near((2 * buffer_frames) as alsa::pcm::Frames)?;
1418 hw_params
1419 .set_period_size_near(buffer_frames as alsa::pcm::Frames, alsa::ValueOr::Nearest)?;
1420 }
1421
1422 // Apply hardware parameters
1423 pcm_handle.hw_params(&hw_params)?;
1424
1425 // For BufferSize::Default, constrain to device's configured period with 2-period buffering.
1426 // PipeWire-ALSA picks a good period size but pairs it with many periods (huge buffer).
1427 // We need to re-initialize hw_params and set BOTH period and buffer to constrain properly.
1428 if config.buffer_size == BufferSize::Default {
1429 if let Ok(period) = hw_params.get_period_size() {
1430 // Re-initialize hw_params to clear previous constraints
1431 let hw_params = init_hw_params(pcm_handle, config, sample_format)?;
1432
1433 // Set both period (to device's chosen value) and buffer (to 2 periods)
1434 hw_params.set_period_size_near(period, alsa::ValueOr::Nearest)?;
1435 hw_params.set_buffer_size_near(2 * period)?;
1436
1437 // Re-apply with new constraints
1438 pcm_handle.hw_params(&hw_params)?;
1439 }
1440 }
1441
1442 Ok(hw_params.can_pause())
1443}
1444
1445fn set_sw_params_from_format(
1446 pcm_handle: &alsa::pcm::PCM,
1447 config: StreamConfig,
1448 stream_type: alsa::Direction,
1449) -> Result<usize, BackendSpecificError> {
1450 let sw_params = pcm_handle.sw_params_current()?;
1451
1452 let period_samples = {
1453 let (buffer, period) = pcm_handle.get_params()?;
1454 if buffer == 0 {
1455 return Err(BackendSpecificError {
1456 description: "initialization resulted in a null buffer".to_string(),
1457 });
1458 }
1459 let start_threshold = match stream_type {
1460 alsa::Direction::Playback => {
1461 // Start playback when 2 periods are filled. This ensures consistent low-latency
1462 // startup regardless of total buffer size (whether 2 or more periods).
1463 2 * period
1464 }
1465 alsa::Direction::Capture => 1,
1466 };
1467 sw_params.set_start_threshold(start_threshold as alsa::pcm::Frames)?;
1468 sw_params.set_avail_min(period as alsa::pcm::Frames)?;
1469
1470 period as usize * config.channels as usize
1471 };
1472
1473 sw_params.set_tstamp_mode(true)?;
1474 sw_params.set_tstamp_type(alsa::pcm::TstampType::MonotonicRaw)?;
1475
1476 // tstamp_type param cannot be changed after the device is opened.
1477 // The default tstamp_type value on most Linux systems is "monotonic",
1478 // let's try to use it if setting the tstamp_type fails.
1479 if pcm_handle.sw_params(&sw_params).is_err() {
1480 sw_params.set_tstamp_type(alsa::pcm::TstampType::Monotonic)?;
1481 pcm_handle.sw_params(&sw_params)?;
1482 }
1483
1484 Ok(period_samples)
1485}
1486
1487fn canonical_pcm_id(pcm_id: &str) -> String {
1488 if let Some((prefix, rest)) = pcm_id.split_once(':') {
1489 let (card_str, device_str) = match rest.split_once(',') {
1490 Some((c, d)) => (c.trim(), d.trim()),
1491 None => (rest.trim(), "0"),
1492 };
1493 if !card_str.contains('=') {
1494 if let Ok(device) = device_str.parse::<u32>() {
1495 return format!("{prefix}:CARD={card_str},DEV={device}");
1496 }
1497 }
1498 }
1499 pcm_id.to_owned()
1500}
1501
1502impl From<alsa::Error> for BackendSpecificError {
1503 fn from(err: alsa::Error) -> Self {
1504 Self {
1505 description: err.to_string(),
1506 }
1507 }
1508}
1509
1510impl From<alsa::Error> for BuildStreamError {
1511 fn from(err: alsa::Error) -> Self {
1512 let err: BackendSpecificError = err.into();
1513 err.into()
1514 }
1515}
1516
1517impl From<alsa::Error> for SupportedStreamConfigsError {
1518 fn from(err: alsa::Error) -> Self {
1519 let err: BackendSpecificError = err.into();
1520 err.into()
1521 }
1522}
1523
1524impl From<alsa::Error> for PlayStreamError {
1525 fn from(err: alsa::Error) -> Self {
1526 let err: BackendSpecificError = err.into();
1527 err.into()
1528 }
1529}
1530
1531impl From<alsa::Error> for PauseStreamError {
1532 fn from(err: alsa::Error) -> Self {
1533 let err: BackendSpecificError = err.into();
1534 err.into()
1535 }
1536}
1537
1538impl From<alsa::Error> for StreamError {
1539 fn from(err: alsa::Error) -> Self {
1540 let err: BackendSpecificError = err.into();
1541 err.into()
1542 }
1543}