Support pipewire as host

[Decodetalkers]

Add support to pipewire

You can test it with pipewire feature open

Pipewire support use config to define rates. So the default config of cpal with pipewire can be changed through config like following.

cat ~/.config/pipewire/pipewire.conf.d/custom-rates.conf
context.properties = {
    default.clock.rate          = 48000
    default.clock.allowed-rates = [ 44100 48000 88200 96000 176400 192000 ]
}

Still problems left:

*once we do 'pipewire::init', we can only dequeue it after the whole thread. even put the function to another thread, the function still works.. But seems we can run init many times.. (Ok, seems it is not a problem, because in when we call init, the init action will only be called once. I think it will be ok)
*The crates by pipewire need edition 2024. I think that should be another pr.

[Decodetalkers]

Seems pipewire need edition 2024..

[Decodetalkers]

@roderickvd can you help review this pr? Thanks, and when can this crate be upgraded to edition 2024? I would also want to help

[roderickvd]

@roderickvd can you help review this pr?

Definitely will help you review it. Need a bit more time.

Thanks, and when can this crate be upgraded to edition 2024? I would also want to help

Actually cpal itself doesn't need to be upgraded to Rust 2024, it just needs a MSRV of Rust of 1.85 or higher to support dependencies that are Rust 2024 already.

When we can I'd like to stick cpal to Rust 2021 so we keep our MSRV down.

[SuperKenVery]

Super cool man! Happy to see cpal having better linux support, opening the possibility of loopback recording on Linux!

[Decodetalkers]

ok, only one ci that I cannot fix

[Decodetalkers]

ok cross-rs based on ubuntu20.04, so

[Be-ing]

How does this differ from #692?

[roderickvd]

Thanks so much for this work! Let's focus on getting it merged for v0.18, and so I spent some time on a code review.

[roderickvd]

That sounds smart - but it does not seem implemented?

[roderickvd]

Writing samples to an immutable &[u8] and then casting away its immutability violates aliasing rules. Try this:

let Some(samples) = buf_data.data_mut() else { return; };
let requested_bytes = buf_data.chunk().size() as usize;
let n_samples = requested_bytes / user_data.sample_format.sample_size();
let frames = requested_bytes / stride;
let data = samples.as_mut_ptr() as *mut ();
let mut data = unsafe { Data::from_parts(data, n_samples, user_data.sample_format) };

[Decodetalkers]

It won't work.. and this way is the way written in pipewire documents..

[Decodetalkers]

https://gitlab.freedesktop.org/pipewire/pipewire-rs/-/blob/main/pipewire/examples/tone.rs

the len of buf_data.chunk().size() here will always be 0, so we should use the len of samples

I also tried to use maxSize, take this as reference. https://docs.pipewire.org/audio-src-ring_8c-example.html, but it solves nothing.

[roderickvd]

You are right about chunk().size and maxsize() not working. When you bump to v0_3_49 I believe you should get a requested() function which does do what we're looking for.

Besides this there's the point of the as_mut_ptr() for proper aliasing (no UB was apparent because it's marked as unsafe but it's not right to write to memory which is defined to be immutable):

.process(|stream, user_data| match stream.dequeue_buffer() {                                                                                                                                 
    None => (user_data.error_callback)(StreamError::BufferUnderrun),                                                                                                                         
    Some(mut buffer) => {                                                                                                                                                                    
        // Read the requested frame count before mutably borrowing datas_mut().                                                                                                              
        // `requested()` is available from PipeWire 0.3.49 (`v0_3_49` feature).                                                                                                              
        let requested = buffer.requested() as usize;                                                                                                                                         
                                                                                                                                                                                               
        let datas = buffer.datas_mut();                                                                                                                                                      
        if datas.is_empty() {                                                                                                                                                                
            return;                                                                                                                                                                          
        }
        let buf_data = &mut datas[0];
        let n_channels = user_data.format.channels();

        let Some(samples) = buf_data.data() else {
            return;
        };
        let stride = user_data.sample_format.sample_size() * n_channels as usize;

        // Honor the frame count PipeWire requests this cycle, capped by the
        // mapped buffer capacity to guard against any mismatch.
        let frames = requested.min(buf_data.maxsize() as usize / stride);
        let n_samples = frames * n_channels as usize;

        // SAFETY: `samples` is a mutable slice into a PipeWire-mapped buffer.
        // `n_samples` is bounded by `buf_data.maxsize()` and `sample_format`
        // matches the format negotiated in the `param_changed` callback.
        let data = samples.as_mut_ptr() as *mut ();
        let mut data =
            unsafe { Data::from_parts(data, n_samples, user_data.sample_format) };

        if let Err(err) = user_data.publish_data_out(frames, &mut data) {
            (user_data.error_callback)(StreamError::BackendSpecific { err });
        }

        let chunk = buf_data.chunk_mut();
        *chunk.offset_mut() = 0;
        *chunk.stride_mut() = stride as i32;
        *chunk.size_mut() = (frames * stride) as u32;
    }
})

[Decodetalkers]

Thank you very much! finally solved it!

[roderickvd]

...instead of buf_data.data() does buf_data.data_mut() also work?

The mutable variant should be used on the output buffer, while the input buffer can remain immutable: buf_data.data() and samples.as_ptr().

[Decodetalkers]

The data is already mutable, not data_mut api.

[yara-blue]

Pipewire docs specify no allocations or file access should be done by any callback when RT_PROCESS is set. I don't think this is currently in the CPAL contract? Consider documenting this or making this a feature flag/or option for this specific host. (I do think RT_PROCESS is really valuable to have as an option).

For context Rodio currently does file access on in the callback. Though maybe it should not ... there is a lot it should not do :)

source: https://docs.pipewire.org/page_streams.html

[Decodetalkers]

ok, I will remove it and add a comment

[roderickvd]

@Decodetalkers there's the feature audio_thread_priority that you could gate this option behind.

@yara-blue yeah... your session title "dragging a ten year old crate into 2026" is hitting harder the more we think about it 😉 ideally the decoders should work on another thread.

[Decodetalkers]

audio_thread_priority is not in the features, it seems a bug.. Is it?

[roderickvd]

Yes, probably lost during rebasing somewhere. Feel free to re-add to Cargo.toml:

# Audio thread priority elevation
# Raises the audio callback thread to real-time priority for lower latency and fewer glitches
# Requires: On Linux, either rtkit or appropriate user permissions (e.g. limits.conf or capabilities)
# Platform: Linux, DragonFly BSD, FreeBSD, NetBSD, Windows
audio_thread_priority = ["dep:audio_thread_priority"]

[Decodetalkers]

ok. seems there are already so many commits, I prefer add the feature of RT_PROCESS later in another pr. If rebase this branch, I may be hard to read the timeline

[roderickvd]

Both ways fine for me.

[Be-ing]

I don't think this is currently in the CPAL contract?

Allocations and filesystem access shouldn't be done in any audio callback, regardless of the backend API or whether that's documented or not, or the priority of the thread.

[Decodetalkers]

@Frando I was wandering. I know sink can also be output, but why source is input here?

[roderickvd]

I see you’re addressing the review point by point - thanks! I reviewed changes that were marked as resolved, and commented where I think we still have got work to do.