Bug 1730499 - mach vendor rust. r=cubeb-reviewers,chunmin

Depends on D126188

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

.cargo/config.in

@@ -25,7 +25,7 @@ rev = "a69df9836b1ef536727195209013b9ad6b132618"
 [source."https://github.com/mozilla/cubeb-pulse-rs"]
 git = "https://github.com/mozilla/cubeb-pulse-rs"
 replace-with = "vendored-sources"
-rev = "3ad5978575f501ab10b1753626f176f1bba3f584"
+rev = "b5aea956e1478b9d0c67911a35fe0aa9d544307a"
 
 [source."https://github.com/mozilla/cubeb-coreaudio-rs"]
 git = "https://github.com/mozilla/cubeb-coreaudio-rs"

third_party/rust/cubeb-pulse/.cargo-checksum.json

@@ -1 +1 @@
-{"files":{".editorconfig":"bf047bd1da10cabb99eea666d1e57c321eba4716dccb3e4ed0e2c5fe3ca53858",".github/workflows/build.yml":"95d0d2542c04f0c932f58591b92c3051db5c95657bf5f24b6a6110f7b667568d","AUTHORS":"0e0ac930a68ce2f6b876126b195add177f0d3886facb9260f4d9b69f1988f0cc","Cargo.toml":"56e90cb82ec36ead07e551a28fc2455fa658fa8308c3d73f8d856d85bfcd2122","LICENSE":"44c6b5ae5ec3fe2fbc608b00e6f4896f4d2d5c7e525fcbaa3eaa3cf2f3d5a983","README.md":"0079450bb4b013bac065ed1750851e461a3710ebad1f323817da1cb82db0bc4f","src/backend/context.rs":"8969b300850b56ace07abe570940e2099774bf0bf20325be1d3ec0f5f408c8ce","src/backend/cork_state.rs":"4a0f1afc7d9f333dac89218cc56d7d32fbffb487cd48c1c9a4e03d79cb3b5e28","src/backend/intern.rs":"11ca424e4eb77f8eb9fd5a6717d1e791facf9743156a8534f0016fcf64d57b0f","src/backend/mod.rs":"d5da05348bf1a7f65c85b14372964a49dc4849f0aee96c75e2c18b51fb03fcaf","src/backend/stream.rs":"deac28f4840dbeeb195deff8734d594e53fbca3b65545a54613e9e6fa22d7dfc","src/capi.rs":"fa0fa020f0d0efe55aa0fc3596405e8407bbe2cbe6c7a558345304e6da87994e","src/lib.rs":"b41bbdc562cbfb130ed7c1e53fe69944774f515705341d8ce48a2f82c8c0c2c5"},"package":null}
+{"files":{".editorconfig":"bf047bd1da10cabb99eea666d1e57c321eba4716dccb3e4ed0e2c5fe3ca53858",".github/workflows/build.yml":"95d0d2542c04f0c932f58591b92c3051db5c95657bf5f24b6a6110f7b667568d","AUTHORS":"0e0ac930a68ce2f6b876126b195add177f0d3886facb9260f4d9b69f1988f0cc","Cargo.toml":"56e90cb82ec36ead07e551a28fc2455fa658fa8308c3d73f8d856d85bfcd2122","LICENSE":"44c6b5ae5ec3fe2fbc608b00e6f4896f4d2d5c7e525fcbaa3eaa3cf2f3d5a983","README.md":"0079450bb4b013bac065ed1750851e461a3710ebad1f323817da1cb82db0bc4f","src/backend/context.rs":"8969b300850b56ace07abe570940e2099774bf0bf20325be1d3ec0f5f408c8ce","src/backend/cork_state.rs":"4a0f1afc7d9f333dac89218cc56d7d32fbffb487cd48c1c9a4e03d79cb3b5e28","src/backend/intern.rs":"11ca424e4eb77f8eb9fd5a6717d1e791facf9743156a8534f0016fcf64d57b0f","src/backend/mod.rs":"d5da05348bf1a7f65c85b14372964a49dc4849f0aee96c75e2c18b51fb03fcaf","src/backend/stream.rs":"c1457baa0b44156364599c4de655ee4438e5291892ef9386936d0b708f8bf777","src/capi.rs":"fa0fa020f0d0efe55aa0fc3596405e8407bbe2cbe6c7a558345304e6da87994e","src/lib.rs":"b41bbdc562cbfb130ed7c1e53fe69944774f515705341d8ce48a2f82c8c0c2c5"},"package":null}

third_party/rust/cubeb-pulse/src/backend/stream.rs

@@ -275,6 +275,7 @@ pub struct PulseStream<'ctx> {
     drain_timer: *mut pa_time_event,
     output_sample_spec: pulse::SampleSpec,
     input_sample_spec: pulse::SampleSpec,
+    // output frames count excluding pre-buffering
     output_frame_count: AtomicUsize,
     shutdown: bool,
     volume: f32,
@@ -375,12 +376,13 @@ impl<'ctx> PulseStream<'ctx> {
                 return;
             }
 
+            let nframes = nbytes / stm.output_sample_spec.frame_size();
+            let first_callback = stm.output_frame_count.fetch_add(nframes, Ordering::SeqCst) == 0;
             if stm.input_stream.is_some() {
-                let nframes = nbytes / stm.output_sample_spec.frame_size();
                 let nsamples_input = nframes * stm.input_sample_spec.channels as usize;
                 let input_buffer_manager = stm.input_buffer_manager.as_mut().unwrap();
 
-                if stm.output_frame_count.fetch_add(nframes, Ordering::SeqCst) == 0 {
+                if first_callback {
                     let buffered_input_frames = input_buffer_manager.available_samples()
                         / stm.input_sample_spec.channels as usize;
                     if buffered_input_frames > nframes {
@@ -436,7 +438,7 @@ impl<'ctx> PulseStream<'ctx> {
 
                         let battr = pa_buffer_attr {
                             maxlength: u32::max_value(),
-                            prebuf: 0,
+                            prebuf: u32::max_value(),
                             fragsize: u32::max_value(),
                             tlength: buffer_size_bytes * 2,
                             minreq: buffer_size_bytes / 4,
@@ -601,9 +603,31 @@ impl<'ctx> Drop for PulseStream<'ctx> {
 
 impl<'ctx> StreamOps for PulseStream<'ctx> {
     fn start(&mut self) -> Result<()> {
+        fn output_preroll(_: &pulse::MainloopApi, u: *mut c_void) {
+            let stm = unsafe { &mut *(u as *mut PulseStream) };
+            if !stm.shutdown {
+                let size = stm
+                    .output_stream
+                    .as_ref()
+                    .map_or(0, |s| s.writable_size().unwrap_or(0));
+                stm.trigger_user_callback(std::ptr::null(), size);
+            }
+        }
         self.shutdown = false;
         self.cork(CorkState::uncork() | CorkState::notify());
 
+        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
+                .get_api()
+                .once(output_preroll, self as *const _ as *mut _);
+            self.context.mainloop.unlock();
+        }
+
         Ok(())
     }
 
@@ -993,7 +1017,7 @@ impl<'ctx> PulseStream<'ctx> {
                             read_offset
                         );
                         let read_ptr = unsafe { (input_data as *const u8).add(read_offset) };
-                        let got = unsafe {
+                        let mut got = unsafe {
                             self.data_callback.unwrap()(
                                 self as *const _ as *mut _,
                                 self.user_ptr,
@@ -1033,15 +1057,40 @@ impl<'ctx> PulseStream<'ctx> {
                             }
                         }
 
+                        let should_drain = (got as usize) < size / frame_size;
+
+                        if should_drain && self.output_frame_count.load(Ordering::SeqCst) == 0 {
+                            // Draining during preroll, ensure `prebuf` frames are written so
+                            // the stream starts. If not, pad with a bit of silence.
+                            let prebuf_size_bytes = stm.get_buffer_attr().prebuf as usize;
+                            let got_bytes = got as usize * frame_size;
+                            if prebuf_size_bytes > got_bytes {
+                                let padding_bytes = prebuf_size_bytes - got_bytes;
+                                if padding_bytes + got_bytes <= size {
+                                    // A slice that starts after the data provided by the callback,
+                                    // with just enough room to provide a final buffer big enough.
+                                    let padding_buf: &mut [u8] = unsafe {
+                                        slice::from_raw_parts_mut::<u8>(
+                                            buffer.add(got_bytes) as *mut u8,
+                                            padding_bytes,
+                                        )
+                                    };
+                                    padding_buf.fill(0);
+                                    got += (padding_bytes / frame_size) as i64;
+                                }
+                            } else {
+                                cubeb_log!("Not enough room to pad up to prebuf when prebuffering.")
+                            }
+                        }
+
                         let r = stm.write(
                             buffer,
                             got as usize * frame_size,
                             0,
                             pulse::SeekMode::Relative,
                         );
-                        debug_assert!(r.is_ok());
 
-                        if (got as usize) < size / frame_size {
+                        if should_drain {
                             cubeb_logv!("Draining {} < {}", got, size / frame_size);
                             let latency = match stm.get_latency() {
                                 Ok(StreamLatency::Positive(l)) => l,
@@ -1073,6 +1122,8 @@ impl<'ctx> PulseStream<'ctx> {
                             return;
                         }
 
+                        debug_assert!(r.is_ok());
+
                         towrite -= size;
                     }
                 }