tsiry-sandratraina.com
Headless Rust gRPC daemon to drive Bluetooth, HLS/DASH playback, and snapcast/shairport-sync/squeezelite on Raspberry Pi audio rigs.
Linux: use alsa-rs directly instead of cpal-alsa mmap path
cpal's ALSA backend uses Access::MMapInterleaved, which crashes inside
libasound's pulse plugin on Raspberry Pi OS (build_output_stream →
snd_pcm_mmap_begin SIGSEGV). aplay / speaker-test work because they
use snd_pcm_writei. Match that path.
- New crates/stream/src/alsa_sink.rs: alsa::pcm::PCM with
Access::RWInterleaved, dedicated writer thread draining a bounded
mpsc channel (cap 64 ≈ 3s buffer), EPIPE/xrun recovery via
pcm.try_recover. Linux only.
- player::build_sink: route AudioOutput::Cpal to AlsaSink on Linux,
CpalSink elsewhere. Proto variant stays Cpal for cross-platform
API parity.
- crates/stream/Cargo.toml: cpal moves to non-Linux target deps,
alsa = "0.9" added for Linux. Smaller dep tree on the Pi.
- CLI: drop --cpal-device. The named-device path walked cpal's
output_devices() enumeration, which triggers libasound plugin
probes that segfault on the JACK plugin destructor. ALSA routing
belongs in ~/.asoundrc, not a CLI knob.
- cpal_sink.rs: trace logs around each cpal call kept for the
non-Linux diagnostic path.
author
Tsiry Sandratraina
date
(Jun 10, 2026, 11:19 PM +0300)
commit
1c4b62e8
1c4b62e887ee8bf3d4e9c4a6078ed45e7b7edd1e
parent
90e39750
90e397500081dde1ad016926e13cadfb02807797
+244
-7
+1
Cargo.lock
+1
Cargo.lock
+10
-1
crates/stream/Cargo.toml
+10
-1
crates/stream/Cargo.toml
···
7
7
anyhow = { workspace = true }
8
8
async-trait = { workspace = true }
9
9
bytes = { workspace = true }
10
-
cpal = "0.15"
11
10
dash-mpd = { version = "0.18", default-features = false, features = ["scte35"] }
12
11
m3u8-rs = "6"
13
12
once_cell = { workspace = true }
···
17
16
tokio = { workspace = true }
18
17
tracing = { workspace = true }
19
18
url = { workspace = true }
19
+
20
+
# On Linux we go straight to libasound via the `alsa` crate (the
21
+
# `Access::RWInterleaved` / `snd_pcm_writei` path), bypassing cpal's
22
+
# mmap-based ALSA backend. cpal's mmap mode crashes inside libasound's
23
+
# pulse plugin on Raspberry Pi OS.
24
+
[target.'cfg(target_os = "linux")'.dependencies]
25
+
alsa = "0.9"
26
+
27
+
[target.'cfg(not(target_os = "linux"))'.dependencies]
28
+
cpal = "0.15"
+194
crates/stream/src/alsa_sink.rs
+194
crates/stream/src/alsa_sink.rs
···
1
+
//! Direct libasound sink (Linux only).
2
+
//!
3
+
//! We use `alsa::pcm::PCM` with `Access::RWInterleaved` + `snd_pcm_writei`,
4
+
//! exactly the way `aplay` does. That avoids cpal-alsa's mmap mode, which
5
+
//! crashes inside libasound's pulse plugin on Raspberry Pi OS (`build_output_stream`
6
+
//! → `snd_pcm_mmap_begin` SIGSEGV).
7
+
//!
8
+
//! Architecture: a dedicated OS thread owns the PCM handle and drains a
9
+
//! bounded mpsc channel. The player's tokio task pushes samples onto the
10
+
//! channel without ever touching libasound, so an ALSA stall can't wedge a
11
+
//! tokio worker.
12
+
13
+
use anyhow::{Context, Result};
14
+
use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
15
+
use std::sync::{Arc, Mutex};
16
+
use std::thread::JoinHandle;
17
+
18
+
use crate::sink::{AudioSink, PcmFormat};
19
+
20
+
const CHANNEL_CAPACITY: usize = 64;
21
+
const PERIOD_FRAMES: alsa::pcm::Frames = 1024; // ~23 ms @ 44.1 kHz
22
+
const BUFFER_FRAMES: alsa::pcm::Frames = 8192; // ~185 ms @ 44.1 kHz
23
+
24
+
enum Msg {
25
+
SetFormat(PcmFormat),
26
+
Write(Vec<i16>),
27
+
Close,
28
+
}
29
+
30
+
pub struct AlsaSink {
31
+
device_name: String,
32
+
tx: Mutex<Option<SyncSender<Msg>>>,
33
+
handle: Mutex<Option<JoinHandle<()>>>,
34
+
}
35
+
36
+
impl AlsaSink {
37
+
fn new(device: Option<String>) -> Arc<Self> {
38
+
let device_name = device
39
+
.filter(|s| !s.is_empty())
40
+
.unwrap_or_else(|| "default".to_string());
41
+
let (tx, rx) = sync_channel::<Msg>(CHANNEL_CAPACITY);
42
+
let device_for_thread = device_name.clone();
43
+
let handle = std::thread::Builder::new()
44
+
.name("zerod-alsa".into())
45
+
.spawn(move || run_writer(device_for_thread, rx))
46
+
.expect("spawn alsa writer thread");
47
+
Arc::new(Self {
48
+
device_name,
49
+
tx: Mutex::new(Some(tx)),
50
+
handle: Mutex::new(Some(handle)),
51
+
})
52
+
}
53
+
}
54
+
55
+
impl AudioSink for AlsaSink {
56
+
fn set_format(&self, fmt: PcmFormat) -> Result<()> {
57
+
if let Some(tx) = self.tx.lock().unwrap().as_ref() {
58
+
// Drop the format change if the writer thread is wedged — we'd
59
+
// rather emit a glitch than block the tokio worker indefinitely.
60
+
let _ = tx.try_send(Msg::SetFormat(fmt));
61
+
}
62
+
Ok(())
63
+
}
64
+
65
+
fn write(&self, samples: &[i16]) -> Result<()> {
66
+
if samples.is_empty() {
67
+
return Ok(());
68
+
}
69
+
let Some(tx) = self.tx.lock().unwrap().clone() else {
70
+
return Ok(());
71
+
};
72
+
// Bounded blocking send: applies back-pressure to the decode loop
73
+
// when the alsa thread can't keep up. At 50ms chunks × 64 capacity
74
+
// that's ~3s of buffer before we block.
75
+
tx.send(Msg::Write(samples.to_vec()))
76
+
.context("alsa: writer thread is gone")?;
77
+
Ok(())
78
+
}
79
+
80
+
fn close(&self) {
81
+
let tx = self.tx.lock().unwrap().take();
82
+
if let Some(tx) = tx {
83
+
let _ = tx.send(Msg::Close);
84
+
}
85
+
if let Some(h) = self.handle.lock().unwrap().take() {
86
+
let _ = h.join();
87
+
}
88
+
}
89
+
}
90
+
91
+
impl Drop for AlsaSink {
92
+
fn drop(&mut self) {
93
+
self.close();
94
+
}
95
+
}
96
+
97
+
fn open_pcm(device: &str, fmt: PcmFormat) -> Result<alsa::PCM> {
98
+
let pcm = alsa::PCM::new(device, alsa::Direction::Playback, false)
99
+
.with_context(|| format!("alsa: open PCM {device}"))?;
100
+
{
101
+
let hwp = alsa::pcm::HwParams::any(&pcm).context("alsa: HwParams::any")?;
102
+
hwp.set_access(alsa::pcm::Access::RWInterleaved)
103
+
.context("alsa: set_access")?;
104
+
hwp.set_format(alsa::pcm::Format::s16())
105
+
.context("alsa: set_format")?;
106
+
hwp.set_channels(fmt.channels as u32)
107
+
.with_context(|| format!("alsa: set_channels({})", fmt.channels))?;
108
+
hwp.set_rate(fmt.sample_rate, alsa::ValueOr::Nearest)
109
+
.with_context(|| format!("alsa: set_rate({})", fmt.sample_rate))?;
110
+
let _ = hwp.set_buffer_size_near(BUFFER_FRAMES);
111
+
let _ = hwp.set_period_size_near(PERIOD_FRAMES, alsa::ValueOr::Nearest);
112
+
pcm.hw_params(&hwp).context("alsa: hw_params apply")?;
113
+
}
114
+
pcm.prepare().context("alsa: prepare")?;
115
+
Ok(pcm)
116
+
}
117
+
118
+
fn run_writer(device: String, rx: Receiver<Msg>) {
119
+
let mut pcm: Option<alsa::PCM> = None;
120
+
let mut current_fmt: Option<PcmFormat> = None;
121
+
tracing::info!("stream/alsa: writer thread started (device={device})");
122
+
123
+
while let Ok(msg) = rx.recv() {
124
+
match msg {
125
+
Msg::SetFormat(fmt) => {
126
+
if current_fmt == Some(fmt) {
127
+
continue;
128
+
}
129
+
if let Some(old) = pcm.take() {
130
+
let _ = old.drain();
131
+
}
132
+
match open_pcm(&device, fmt) {
133
+
Ok(new) => {
134
+
tracing::info!(
135
+
"stream/alsa: opened {} at {} Hz × {} ch",
136
+
device,
137
+
fmt.sample_rate,
138
+
fmt.channels
139
+
);
140
+
pcm = Some(new);
141
+
current_fmt = Some(fmt);
142
+
}
143
+
Err(e) => {
144
+
tracing::error!("stream/alsa: open failed: {e:#}");
145
+
// Keep `pcm` as None — subsequent writes are dropped
146
+
// until a successful set_format reopens.
147
+
}
148
+
}
149
+
}
150
+
Msg::Write(samples) => {
151
+
let Some(p) = pcm.as_ref() else { continue };
152
+
let Some(fmt) = current_fmt else { continue };
153
+
let ch = (fmt.channels as usize).max(1);
154
+
let io = match p.io_i16() {
155
+
Ok(io) => io,
156
+
Err(e) => {
157
+
tracing::error!("stream/alsa: io_i16: {e}");
158
+
continue;
159
+
}
160
+
};
161
+
let mut offset = 0usize;
162
+
while offset < samples.len() {
163
+
let frame_start = offset / ch;
164
+
let chunk = &samples[offset..];
165
+
match io.writei(chunk) {
166
+
Ok(n) => {
167
+
offset += n * ch;
168
+
}
169
+
Err(e) => {
170
+
tracing::warn!(
171
+
"stream/alsa: writei error at frame {frame_start}: {e}, recovering"
172
+
);
173
+
if let Err(re) = p.try_recover(e, true) {
174
+
tracing::error!("stream/alsa: recover failed: {re}");
175
+
break;
176
+
}
177
+
}
178
+
}
179
+
}
180
+
}
181
+
Msg::Close => break,
182
+
}
183
+
}
184
+
185
+
if let Some(p) = pcm {
186
+
let _ = p.drain();
187
+
}
188
+
tracing::info!("stream/alsa: writer thread exiting");
189
+
}
190
+
191
+
/// Factory used by the player. Spawns the writer thread and returns the sink.
192
+
pub fn build(device: Option<String>) -> Result<Arc<AlsaSink>> {
193
+
Ok(AlsaSink::new(device))
194
+
}
+17
crates/stream/src/cpal_sink.rs
+17
crates/stream/src/cpal_sink.rs
···
57
57
}
58
58
59
59
fn open_stream(self: &std::sync::Arc<Self>) -> Result<()> {
60
+
tracing::info!("stream/cpal: open_stream begin (device={:?})", self.device_name);
61
+
62
+
tracing::debug!("stream/cpal: cpal::default_host()");
60
63
let host = cpal::default_host();
64
+
tracing::info!("stream/cpal: host id = {:?}", host.id());
65
+
66
+
tracing::debug!("stream/cpal: selecting output device");
61
67
let device = match self.device_name.as_deref() {
62
68
None | Some("") => host
63
69
.default_output_device()
···
68
74
.find(|d| d.name().map(|n| n == name).unwrap_or(false))
69
75
.ok_or_else(|| anyhow!("cpal: device {name} not found"))?,
70
76
};
77
+
let device_name = device.name().unwrap_or_else(|_| "<unnamed>".to_string());
78
+
tracing::info!("stream/cpal: selected device = {}", device_name);
71
79
80
+
tracing::debug!("stream/cpal: querying default_output_config()");
72
81
let config = device
73
82
.default_output_config()
74
83
.context("cpal: default output config")?;
···
76
85
let stream_config: cpal::StreamConfig = config.into();
77
86
let out_rate = stream_config.sample_rate.0;
78
87
let out_channels = stream_config.channels as u32;
88
+
tracing::info!(
89
+
"stream/cpal: default config = {} Hz × {} ch, sample_format = {:?}",
90
+
out_rate,
91
+
out_channels,
92
+
sample_format
93
+
);
79
94
self.out_rate.store(out_rate, Ordering::SeqCst);
80
95
self.out_channels.store(out_channels, Ordering::SeqCst);
81
96
···
103
118
other => return Err(anyhow!("cpal: unsupported sample format {other:?}")),
104
119
}
105
120
.context("cpal: build output stream")?;
121
+
tracing::info!("stream/cpal: build_output_stream ok, calling stream.play()");
106
122
stream.play().context("cpal: start stream")?;
123
+
tracing::info!("stream/cpal: stream.play() ok");
107
124
108
125
*self.stream.lock().unwrap() = Some(StreamHolder(stream));
109
126
tracing::info!(
+3
crates/stream/src/lib.rs
+3
crates/stream/src/lib.rs
···
9
9
//! Lifted from rockbox-zig/crates/hls and rewired to a trait-based output so
10
10
//! the gRPC layer can pick `cpal | stdout | pipe` per-stream at runtime.
11
11
12
+
#[cfg(target_os = "linux")]
13
+
mod alsa_sink;
14
+
#[cfg(not(target_os = "linux"))]
12
15
mod cpal_sink;
13
16
mod decoder;
14
17
mod demux;
+12
-2
crates/stream/src/player.rs
+12
-2
crates/stream/src/player.rs
···
10
10
use tokio::runtime::Runtime;
11
11
use tokio::task::JoinHandle;
12
12
13
+
#[cfg(target_os = "linux")]
14
+
use crate::alsa_sink;
15
+
#[cfg(not(target_os = "linux"))]
13
16
use crate::cpal_sink;
14
17
use crate::decoder::decode_segment;
15
18
use crate::demux;
···
125
128
fn build_sink(out: &AudioOutput) -> Result<Arc<dyn AudioSink>> {
126
129
match out {
127
130
AudioOutput::Cpal { device } => {
128
-
let sink = cpal_sink::build(device.clone())?;
129
-
Ok(sink as Arc<dyn AudioSink>)
131
+
// On Linux we go straight to libasound via alsa-rs to dodge
132
+
// cpal-alsa's mmap path (which segfaults inside the pulse
133
+
// plugin on Raspberry Pi OS). The proto variant stays named
134
+
// `Cpal` for cross-platform consistency.
135
+
#[cfg(target_os = "linux")]
136
+
let sink = alsa_sink::build(device.clone())? as Arc<dyn AudioSink>;
137
+
#[cfg(not(target_os = "linux"))]
138
+
let sink = cpal_sink::build(device.clone())? as Arc<dyn AudioSink>;
139
+
Ok(sink)
130
140
}
131
141
AudioOutput::Stdout => Ok(Arc::new(StdoutSink::new()) as Arc<dyn AudioSink>),
132
142
AudioOutput::Pipe { path } => {
+7
-4
src/main.rs
+7
-4
src/main.rs
···
83
83
output: OutputArg,
84
84
#[arg(long)]
85
85
pipe_path: Option<String>,
86
-
#[arg(long)]
87
-
cpal_device: Option<String>,
88
86
},
89
87
Pause,
90
88
Resume,
···
296
294
let ch = channel(host, port).await?;
297
295
let mut client = pb::stream_service_client::StreamServiceClient::new(ch);
298
296
match cmd {
299
-
StreamCmd::Play { url, output, pipe_path, cpal_device } => {
297
+
StreamCmd::Play { url, output, pipe_path } => {
300
298
let output = match output {
301
299
OutputArg::Cpal => pb::AudioOutput::Cpal,
302
300
OutputArg::Stdout => pb::AudioOutput::Stdout,
···
304
302
} as i32;
305
303
client
306
304
.play(attach_token(
307
-
Request::new(pb::PlayRequest { url, output, pipe_path, cpal_device }),
305
+
Request::new(pb::PlayRequest {
306
+
url,
307
+
output,
308
+
pipe_path,
309
+
cpal_device: None,
310
+
}),
308
311
&token,
309
312
))
310
313
.await?;