Bug 1567457 - Update cubeb-pulse-rs to version 3a748a2. r=kinetik

Reviewed upstream by :achronop, :kinetik, :chunmin in:
https://github.com/djg/cubeb-pulse-rs/pull/41
https://github.com/djg/cubeb-pulse-rs/pull/42

Differential Revision: https://phabricator.services.mozilla.com/D38661

--HG--
extra : moz-landing-system : lando

media/libcubeb/cubeb-pulse-rs/Cargo.toml

@@ -15,3 +15,4 @@ cubeb-backend = "0.5"
 pulse-ffi = { path = "pulse-ffi" }
 pulse = { path = "pulse-rs" }
 semver = "^0.6"
+ringbuf = "0.1"

media/libcubeb/cubeb-pulse-rs/README_MOZILLA

@@ -5,4 +5,4 @@ Makefile.in build files for the Mozilla build system.
 
 The cubeb-pulse-rs git repository is: https://github.com/djg/cubeb-pulse-rs.git
 
-The git commit ID used was 17c1629c323ff24d656ff9449bf50d6758aafc1a (2019-01-24 07:50:09 +1300)
+The git commit ID used was 3a748a2df25658f1c8c5a475b9dd2ae4561b174b (2019-07-19 21:44:07 +1200)

media/libcubeb/cubeb-pulse-rs/pulse-ffi/src/ffi_funcs.rs

@@ -84,6 +84,7 @@ mod static_fns {
                                       new_balance: c_float)
                                       -> *mut pa_cvolume;
         pub fn pa_frame_size(spec: *const pa_sample_spec) -> usize;
+        pub fn pa_sample_size(spec: *const pa_sample_spec) -> usize;
         pub fn pa_mainloop_api_once(m: *mut pa_mainloop_api,
                                     callback: pa_mainloop_api_once_cb_t,
                                     userdata: *mut c_void);
@@ -365,6 +366,13 @@ mod dynamic_fns {
                 }
                 fp
             };
+            PA_SAMPLE_SIZE = {
+                let fp = dlsym(h, cstr!("pa_sample_size"));
+                if fp.is_null() {
+                    return None;
+                }
+                fp
+            };
             PA_MAINLOOP_API_ONCE = {
                 let fp = dlsym(h, cstr!("pa_mainloop_api_once"));
                 if fp.is_null() {
@@ -999,6 +1007,12 @@ mod dynamic_fns {
         (::std::mem::transmute::<_, extern "C" fn(*const pa_sample_spec) -> usize>(PA_FRAME_SIZE))(spec)
     }
 
+    static mut PA_SAMPLE_SIZE: *mut ::libc::c_void = 0 as *mut _;
+    #[inline]
+    pub unsafe fn pa_sample_size(spec: *const pa_sample_spec) -> usize {
+        (::std::mem::transmute::<_, extern "C" fn(*const pa_sample_spec) -> usize>(PA_SAMPLE_SIZE))(spec)
+    }
+
     static mut PA_MAINLOOP_API_ONCE: *mut ::libc::c_void = 0 as *mut _;
     #[inline]
     pub unsafe fn pa_mainloop_api_once(m: *mut pa_mainloop_api,

media/libcubeb/cubeb-pulse-rs/pulse-rs/src/context.rs

@@ -95,7 +95,7 @@ impl Context {
     pub fn set_state_callback<CB>(&self, _: CB, userdata: *mut c_void)
         where CB: Fn(&Context, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context, userdata: *mut c_void)
@@ -146,7 +146,7 @@ impl Context {
     pub fn drain<CB>(&self, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context, userdata: *mut c_void)
@@ -167,7 +167,7 @@ impl Context {
     pub fn rttime_new<CB>(&self, usec: USec, _: CB, userdata: *mut c_void) -> *mut ffi::pa_time_event
         where CB: Fn(&MainloopApi, *mut ffi::pa_time_event, &TimeVal, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(a: *mut ffi::pa_mainloop_api,
@@ -191,7 +191,7 @@ impl Context {
     pub fn get_server_info<CB>(&self, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, Option<&ServerInfo>, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context, i: *const ffi::pa_server_info, userdata: *mut c_void)
@@ -219,7 +219,7 @@ impl Context {
         CB: Fn(&Context, *const SinkInfo, i32, *mut c_void),
         CS: Into<Option<&'str CStr>>,
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context,
@@ -246,7 +246,7 @@ impl Context {
     pub fn get_sink_info_list<CB>(&self, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, *const SinkInfo, i32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context,
@@ -270,7 +270,7 @@ impl Context {
     pub fn get_sink_input_info<CB>(&self, idx: u32, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, *const SinkInputInfo, i32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context,
@@ -294,7 +294,7 @@ impl Context {
     pub fn get_source_info_list<CB>(&self, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, *const SourceInfo, i32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context,
@@ -323,7 +323,7 @@ impl Context {
                                      -> Result<Operation>
         where CB: Fn(&Context, i32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context, success: c_int, userdata: *mut c_void)
@@ -344,7 +344,7 @@ impl Context {
     pub fn subscribe<CB>(&self, m: SubscriptionMask, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Context, i32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context, success: c_int, userdata: *mut c_void)
@@ -371,7 +371,7 @@ impl Context {
     pub fn set_subscribe_callback<CB>(&self, _: CB, userdata: *mut c_void)
         where CB: Fn(&Context, SubscriptionEvent, u32, *mut c_void)
     {
-        debug_assert_eq!(::std::mem::size_of::<CB>(), 0);
+        assert_eq!(::std::mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(c: *mut ffi::pa_context,

media/libcubeb/cubeb-pulse-rs/pulse-rs/src/lib.rs

@@ -639,12 +639,16 @@ impl ProplistExt for SourceInfo {
 
 pub trait SampleSpecExt {
     fn frame_size(&self) -> usize;
+    fn sample_size(&self) -> usize;
 }
 
 impl SampleSpecExt for SampleSpec {
     fn frame_size(&self) -> usize {
         unsafe { ffi::pa_frame_size(self) }
     }
+    fn sample_size(&self) -> usize {
+        unsafe { ffi::pa_sample_size(self) }
+    }
 }
 
 pub trait USecExt {

media/libcubeb/cubeb-pulse-rs/pulse-rs/src/stream.rs

@@ -184,7 +184,7 @@ impl Stream {
     pub fn update_timing_info<CB>(&self, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Stream, i32, *mut c_void)
     {
-        debug_assert_eq!(mem::size_of::<CB>(), 0);
+        assert_eq!(mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(s: *mut ffi::pa_stream, success: c_int, userdata: *mut c_void)
@@ -219,7 +219,7 @@ impl Stream {
     pub fn set_state_callback<CB>(&self, _: CB, userdata: *mut c_void)
         where CB: Fn(&Stream, *mut c_void)
     {
-        debug_assert_eq!(mem::size_of::<CB>(), 0);
+        assert_eq!(mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(s: *mut ffi::pa_stream, userdata: *mut c_void)
@@ -247,7 +247,7 @@ impl Stream {
     pub fn set_write_callback<CB>(&self, _: CB, userdata: *mut c_void)
         where CB: Fn(&Stream, usize, *mut c_void)
     {
-        debug_assert_eq!(mem::size_of::<CB>(), 0);
+        assert_eq!(mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(s: *mut ffi::pa_stream, nbytes: usize, userdata: *mut c_void)
@@ -275,7 +275,7 @@ impl Stream {
     pub fn set_read_callback<CB>(&self, _: CB, userdata: *mut c_void)
         where CB: Fn(&Stream, usize, *mut c_void)
     {
-        debug_assert_eq!(mem::size_of::<CB>(), 0);
+        assert_eq!(mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(s: *mut ffi::pa_stream, nbytes: usize, userdata: *mut c_void)
@@ -297,7 +297,7 @@ impl Stream {
     pub fn cork<CB>(&self, b: i32, _: CB, userdata: *mut c_void) -> Result<Operation>
         where CB: Fn(&Stream, i32, *mut c_void)
     {
-        debug_assert_eq!(mem::size_of::<CB>(), 0);
+        assert_eq!(mem::size_of::<CB>(), 0);
 
         // See: A note about `wrapped` functions
         unsafe extern "C" fn wrapped<F>(s: *mut ffi::pa_stream, success: c_int, userdata: *mut c_void)

media/libcubeb/cubeb-pulse-rs/src/backend/stream.rs

@@ -12,8 +12,20 @@ use pulse_ffi::*;
 use std::{mem, ptr};
 use std::ffi::{CStr, CString};
 use std::os::raw::{c_long, c_void};
+use std::slice;
+use ringbuf::RingBuffer;
+
+use self::RingBufferConsumer::*;
+use self::RingBufferProducer::*;
+use self::LinearInputBuffer::*;
 
 const PULSE_NO_GAIN: f32 = -1.0;
+// When running duplex callbacks, the input data is fed to a ring buffer, and then later copied to
+// a linear piece of memory is used to hold the input samples, so that they are passed to the audio
+// callback that delivers it to the callees. Their size depends on the buffer size requested
+// initially.  This is to be tuned when changing tlength and fragsize, but this value works for
+// now.
+const INPUT_BUFFER_CAPACITY: usize = 4096;
 
 /// Iterator interface to `ChannelLayout`.
 ///
@@ -112,6 +124,141 @@ impl Drop for Device {
     }
 }
 
+
+enum RingBufferConsumer {
+    IntegerRingBufferConsumer(ringbuf::Consumer<i16>),
+    FloatRingBufferConsumer(ringbuf::Consumer<f32>)
+}
+
+enum RingBufferProducer {
+    IntegerRingBufferProducer(ringbuf::Producer<i16>),
+    FloatRingBufferProducer(ringbuf::Producer<f32>)
+}
+
+enum LinearInputBuffer {
+    IntegerLinearInputBuffer(Vec<i16>),
+    FloatLinearInputBuffer(Vec<f32>)
+}
+
+struct BufferManager {
+    consumer: RingBufferConsumer,
+    producer: RingBufferProducer,
+    linear_input_buffer: LinearInputBuffer
+}
+
+impl BufferManager {
+    // When opening a duplex stream, the sample-spec are guaranteed to match. It's ok to have
+    // either the input or output sample-spec here.
+    fn new(sample_spec: &pulse::SampleSpec) -> BufferManager {
+        if sample_spec.format == PA_SAMPLE_S16BE ||
+           sample_spec.format == PA_SAMPLE_S16LE  {
+                let ring = RingBuffer::<i16>::new(INPUT_BUFFER_CAPACITY);
+                let (prod, cons) = ring.split();
+                return BufferManager {
+                    producer: IntegerRingBufferProducer(prod),
+                    consumer: IntegerRingBufferConsumer(cons),
+                    linear_input_buffer: IntegerLinearInputBuffer(Vec::<i16>::with_capacity(INPUT_BUFFER_CAPACITY))
+                };
+            } else {
+                let ring = RingBuffer::<f32>::new(INPUT_BUFFER_CAPACITY);
+                let (prod, cons) = ring.split();
+                return BufferManager {
+                    producer: FloatRingBufferProducer(prod),
+                    consumer: FloatRingBufferConsumer(cons),
+                    linear_input_buffer: FloatLinearInputBuffer(Vec::<f32>::with_capacity(INPUT_BUFFER_CAPACITY))
+                };
+            }
+    }
+
+    fn push_input_data(&mut self, input_data: *const c_void, read_samples: usize) {
+        match &mut self.producer {
+            RingBufferProducer::FloatRingBufferProducer(p) => {
+                let input_data = unsafe { slice::from_raw_parts::<f32>(input_data as *const f32, read_samples) };
+                match p.push_slice(input_data) {
+                    Ok(_) => { }
+                    Err(_) => {
+                        // do nothing: the data are ignored. This happens when underruning the
+                        // output callback.
+                    }
+                }
+            }
+            RingBufferProducer::IntegerRingBufferProducer(p) => {
+                let input_data = unsafe { slice::from_raw_parts::<i16>(input_data as *const i16, read_samples) };
+                match p.push_slice(input_data) {
+                    Ok(_) => { }
+                    Err(_) => {
+                        // do nothing: the data are ignored. This happens when underruning the
+                        // output callback.
+                    }
+                }
+            }
+        }
+    }
+
+    fn pull_input_data(&mut self, input_data: *mut c_void, needed_samples: usize) {
+        match &mut self.consumer {
+            IntegerRingBufferConsumer(p) => {
+                let mut input: &mut[i16] = unsafe { slice::from_raw_parts_mut::<i16>(input_data as *mut i16, needed_samples) };
+                match p.pop_slice(&mut input) {
+                    Ok(read) => {
+                        if read < needed_samples {
+                            for i in 0..(needed_samples - read) {
+                                input[read + i] = 0;
+                            }
+                        }
+                    }
+                    Err(_) => {
+                        // Buffer empty
+                        for i in input.iter_mut() {
+                            *i = 0;
+                        }
+                    }
+                }
+            }
+            FloatRingBufferConsumer(p) => {
+                let mut input: &mut[f32] = unsafe { slice::from_raw_parts_mut::<f32>(input_data as *mut f32, needed_samples) };
+                match p.pop_slice(&mut input) {
+                    Ok(read) => {
+                        if read < needed_samples {
+                            for i in 0..(needed_samples - read) {
+                                input[read + i] = 0.;
+                            }
+                        }
+                    }
+                    Err(_) => {
+                        // Buffer empty
+                        for i in input.iter_mut() {
+                            *i = 0.;
+                        }
+                    }
+                }
+            }
+        }
+    }
+    fn get_linear_input_data(&mut self, nsamples: usize) -> *const c_void {
+        let p: *mut c_void;
+        match &mut self.linear_input_buffer {
+            LinearInputBuffer::IntegerLinearInputBuffer(b) => {
+                b.resize(nsamples, 0);
+                p = b.as_mut_ptr() as *mut c_void;
+            }
+            LinearInputBuffer::FloatLinearInputBuffer(b) => {
+                b.resize(nsamples, 0.);
+                p = b.as_mut_ptr() as *mut c_void;
+            }
+        }
+        self.pull_input_data(p, nsamples);
+
+        return p;
+    }
+}
+
+impl std::fmt::Debug for BufferManager {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "")
+    }
+}
+
 #[repr(C)]
 #[derive(Debug)]
 pub struct PulseStream<'ctx> {
@@ -127,6 +274,7 @@ pub struct PulseStream<'ctx> {
     shutdown: bool,
     volume: f32,
     state: ffi::cubeb_state,
+    input_buffer_manager: Option<BufferManager>
 }
 
 impl<'ctx> PulseStream<'ctx> {
@@ -177,13 +325,11 @@ impl<'ctx> PulseStream<'ctx> {
                 if !read_data.is_null() {
                     let in_frame_size = stm.input_sample_spec.frame_size();
                     let read_frames = read_size / in_frame_size;
+                    let read_samples = read_size / stm.input_sample_spec.sample_size();
 
                     if stm.output_stream.is_some() {
-                        // input/capture + output/playback operation
-                        let out_frame_size = stm.output_sample_spec.frame_size();
-                        let write_size = read_frames * out_frame_size;
-                        // Offer full duplex data for writing
-                        stm.trigger_user_callback(read_data, write_size);
+                        // duplex stream: push the input data to the ring buffer.
+                        stm.input_buffer_manager.as_mut().unwrap().push_input_data(read_data, read_samples);
                     } else {
                         // input/capture only operation. Call callback directly
                         let got = unsafe {
@@ -221,7 +367,11 @@ impl<'ctx> PulseStream<'ctx> {
                 return;
             }
 
-            if stm.input_stream.is_none() {
+            if stm.input_stream.is_some() {
+                let nsamples = nbytes / stm.output_sample_spec.sample_size();
+                let p = stm.input_buffer_manager.as_mut().unwrap().get_linear_input_data(nsamples);
+                stm.trigger_user_callback(p, nbytes);
+            } else {
                 // Output/playback only operation.
                 // Write directly to output
                 debug_assert!(stm.output_stream.is_some());
@@ -242,6 +392,7 @@ impl<'ctx> PulseStream<'ctx> {
             shutdown: false,
             volume: PULSE_NO_GAIN,
             state: ffi::CUBEB_STATE_ERROR,
+            input_buffer_manager: None
         });
 
         if let Some(ref context) = stm.context.context {
@@ -310,6 +461,11 @@ impl<'ctx> PulseStream<'ctx> {
                 }
             }
 
+            // Duplex, set up the ringbuffer
+            if input_stream_params.is_some() && output_stream_params.is_some() {
+                stm.input_buffer_manager = Some(BufferManager::new(&stm.input_sample_spec))
+            }
+
             let r = if stm.wait_until_ready() {
                 /* force a timing update now, otherwise timing info does not become valid
                 until some point after initialization has completed. */
@@ -406,10 +562,10 @@ impl<'ctx> StreamOps for PulseStream<'ctx> {
         self.shutdown = false;
         self.cork(CorkState::uncork() | CorkState::notify());
 
-        if self.output_stream.is_some() && self.input_stream.is_none() {
-            /* On output only case need to manually call user cb once in order to make
-             * things roll. This is done via a defer event in order to execute it
-             * from PA server thread. */
+        if self.output_stream.is_some() {
+            /* When doing output-only or duplex, we need to manually call user cb once in order to
+             * make things roll. This is done via a defer event in order to execute it from PA
+             * server thread. */
             self.context.mainloop.lock();
             self.context
                 .mainloop
@@ -797,6 +953,7 @@ impl<'ctx> PulseStream<'ctx> {
         true
     }
 
+
     #[cfg_attr(feature = "cargo-clippy", allow(cyclomatic_complexity))]
     fn trigger_user_callback(&mut self, input_data: *const c_void, nbytes: usize) {
         fn drained_cb(
@@ -941,6 +1098,7 @@ fn context_success(_: &pulse::Context, success: i32, u: *mut c_void) {
 }
 
 fn set_buffering_attribute(latency_frames: u32, sample_spec: &pa_sample_spec) -> pa_buffer_attr {
+    // When changing this, change the constant INPUT_BUFFER_CAPACITY to reflect the new sizes.
     let tlength = latency_frames * sample_spec.frame_size() as u32;
     let minreq = tlength / 4;
     let battr = pa_buffer_attr {

media/libcubeb/cubeb-pulse-rs/src/lib.rs

@@ -10,6 +10,7 @@ extern crate cubeb_backend;
 extern crate pulse;
 extern crate pulse_ffi;
 extern crate semver;
+extern crate ringbuf;
 
 mod capi;
 mod backend;