From 5b49955e33bc6f5405afd289f938482704d3d1bc Mon Sep 17 00:00:00 2001 From: Breadway Date: Wed, 5 Aug 2026 19:04:03 +0800 Subject: [PATCH] Fix Cast Streaming teardown crash, bitrate collapse, and HLS latency Three independent problems found by reading the mirroring paths end to end. 1. Segfault on Cast session teardown. CastStreamSender::SchedulePoll self-reschedules every 100ms with a raw `this` and was never cancelled, so the one task that can be scheduled to run *after* an already-queued teardown task would dereference the just-reset `environment` unique_ptr (Environment::task_runner() dereferences a member immediately) --- a hard null deref on openscreen's TaskRunner thread. TaskRunnerImpl's shutdown has an explicit flushing phase and PlatformClientPosix::ShutDown()'s quit task queues behind whatever is already pending, so this is a race the teardown path can lose. Latch a `shutting_down` atomic before posting teardown and check it in the poll and in every other posted task. 2. Encoder bitrate collapsing to the floor within seconds. The control loop set `target = 0.85 * estimate` once a second unconditionally. openscreen's BandwidthEstimator deliberately under-estimates capacity whenever the transmit rate is below it and documents the required TCP-like response; multiplying the target by <=0.85 every second instead walks 4000 kbps past 1500 in ~6s and pins it at the floor on a healthy LAN. Replaced with proper AIMD (hold on a zero/unknown estimate, back off below it, probe up 10%/s otherwise), clamped to 1000..8000 kbps, with unit tests. 3. DLNA/HLS latency. Segment length is max(target-duration, GOP), so target-duration=2 with a 2s GOP gave 2s segments, ~6s of renderer buffer, plus 3 segments of stale video waited for before handover. 1s segments (GOP halved to make that reachable), shorter playlist, and wait for 2 segments instead of 3. Also hardened two paths into openscreen's fatal OSP_CHECK on strictly increasing RTP timestamps: pull_encoded_frame no longer substitutes 0 for a missing PTS (it skips the buffer), and facade.cc drops non-monotonic capture times at the FFI boundary. Either could previously abort the daemon outright. CastMirrorSession now owns the Arc instead of leaving its lifetime to whichever detached pump thread dropped the last clone, so the blocking FFI destroy happens at a defined point in stop() with the pump joins ordered around it. --- breadcast-caststream-sys/src/facade.cc | 63 ++++++++ breadcast-core/src/pipeline/mod.rs | 68 ++++++--- breadcastd/src/cast_mirror.rs | 204 +++++++++++++++++++++---- breadcastd/src/dlna_mirror.rs | 8 +- 4 files changed, 299 insertions(+), 44 deletions(-) diff --git a/breadcast-caststream-sys/src/facade.cc b/breadcast-caststream-sys/src/facade.cc index 753c231..1a84e64 100644 --- a/breadcast-caststream-sys/src/facade.cc +++ b/breadcast-caststream-sys/src/facade.cc @@ -44,6 +44,34 @@ struct CastStreamSender { bool have_origin = false; int64_t origin_capture_time_us = 0; + // The capture timestamp of the last frame actually handed to + // Sender::EnqueueFrame. openscreen enforces strictly-increasing RTP + // timestamps with a *fatal* OSP_CHECK_GT (sender_impl.cc), not an error + // return -- so a single frame arriving with a non-increasing capture time + // (a buffer with no PTS, which the GStreamer side substitutes 0 for; a + // clock reset on portal source change; any encoder that ever reorders + // output) would abort the whole process. Dropping such a frame instead + // costs at most one frame of video. Only touched on the TaskRunner + // thread. + bool have_last_capture_time = false; + int64_t last_capture_time_us = 0; + + // Set by breadcast_caststream_sender_destroy *before* it posts its + // teardown task, and checked by the self-rescheduling poll below. + // + // Without this, the poll task (posted with a 100ms delay, so it is the one + // task that can be scheduled to run *after* an already-queued teardown + // task) dereferences `environment` after teardown has reset it -- a null + // `unique_ptr`, whose `task_runner()` accessor immediately + // dereferences a member -- i.e. a hard SIGSEGV on openscreen's TaskRunner + // thread. If it lands even later it is a use-after-free of `this`, since + // destroy() `delete`s this struct once the teardown task completes. + // TaskRunnerImpl's shutdown has an explicit "flushing phase" that keeps + // running runnable tasks, and PlatformClientPosix::ShutDown()'s quit task + // is queued *behind* whatever is already pending, so this is a race the + // teardown path can and does lose. + std::atomic shutting_down{false}; + // negotiated uses acquire/release so that once // breadcast_caststream_sender_enqueue_frame observes it true (from an // arbitrary caller thread), `session->video_sender()` is guaranteed @@ -62,8 +90,14 @@ struct CastStreamSender { BreadcastOnPictureLostFn rust_on_picture_lost = nullptr; void SchedulePoll() { + if (shutting_down.load(std::memory_order_acquire) || !environment) { + return; + } environment->task_runner().PostTaskWithDelay( [this] { + if (shutting_down.load(std::memory_order_acquire)) { + return; + } if (session && session->video_sender()) { needs_key_frame.store(session->video_sender()->NeedsKeyFrame(), std::memory_order_relaxed); @@ -186,6 +220,10 @@ CastStreamSender* breadcast_caststream_sender_create( void breadcast_caststream_sender_negotiate(CastStreamSender* sender) { sender->environment->task_runner().PostTask([sender] { + if (sender->shutting_down.load(std::memory_order_acquire) || + !sender->session) { + return; + } sender->session->Negotiate(); sender->SchedulePoll(); }); @@ -202,6 +240,10 @@ void breadcast_caststream_sender_on_message(CastStreamSender* sender, auto ns = std::make_shared(message_namespace, message_namespace_len); auto body = std::make_shared(message, message_len); sender->environment->task_runner().PostTask([sender, source, ns, body] { + if (sender->shutting_down.load(std::memory_order_acquire) || + !sender->message_port) { + return; + } sender->message_port->DeliverMessage(*source, *ns, *body); }); } @@ -225,15 +267,30 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender, using namespace openscreen; using namespace openscreen::cast; + if (sender->shutting_down.load(std::memory_order_acquire) || + !sender->session) { + return; + } Sender* video_sender = sender->session->video_sender(); if (!video_sender) { return; } + // See `last_capture_time_us`: openscreen aborts the process (fatal + // OSP_CHECK, not an error return) if RTP timestamps ever fail to + // strictly increase, so a non-monotonic capture time has to be dropped + // here rather than passed through. + if (sender->have_last_capture_time && + capture_time_us <= sender->last_capture_time_us) { + return; + } + if (!sender->have_origin) { sender->have_origin = true; sender->origin_capture_time_us = capture_time_us; } + sender->have_last_capture_time = true; + sender->last_capture_time_us = capture_time_us; const FrameId frame_id = video_sender->GetNextFrameId(); const FrameId referenced_frame_id = @@ -277,6 +334,12 @@ void breadcast_caststream_sender_destroy(CastStreamSender* sender) { if (!sender) { return; } + // Latched *before* the teardown task is posted so the self-rescheduling + // poll (see CastStreamSender::shutting_down) stops re-arming itself and + // no longer touches `environment`/`session` -- both of which the teardown + // task below is about to reset out from under it. + sender->shutting_down.store(true, std::memory_order_release); + // These must be torn down on the TaskRunner thread (they hold raw // references into it and into `environment`), so hop over there and block // until it's done before shutting the TaskRunner itself down. diff --git a/breadcast-core/src/pipeline/mod.rs b/breadcast-core/src/pipeline/mod.rs index abf8b89..0a6ce94 100644 --- a/breadcast-core/src/pipeline/mod.rs +++ b/breadcast-core/src/pipeline/mod.rs @@ -62,15 +62,37 @@ pub fn build_video_pipeline(video_node_id: u32, output_dir: &Path) -> Result Result, bool, i64)>> { - let sample = match appsink.pull_sample() { - Ok(sample) => sample, - Err(_) if appsink.is_eos() => return Ok(None), - Err(e) => bail!("appsink pull_sample failed: {e}"), - }; - let buffer = sample.buffer().context("pulled sample had no buffer")?; - let map = buffer.map_readable().context("failed to map sample buffer readable")?; - let is_key_frame = !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT); - // `.unwrap_or(0)` rather than propagating a missing PTS as an error: - // CastStreamSender::enqueue_frame only needs monotonically-increasing, - // real-elapsed-time-proportional values (see its doc comment) -- an - // occasional buffer with no PTS shouldn't abort an otherwise-live - // stream over it. - let capture_time_us = buffer.pts().map(|t| t.useconds() as i64).unwrap_or(0); - Ok(Some((map.as_slice().to_vec(), is_key_frame, capture_time_us))) + loop { + let sample = match appsink.pull_sample() { + Ok(sample) => sample, + Err(_) if appsink.is_eos() => return Ok(None), + Err(e) => bail!("appsink pull_sample failed: {e}"), + }; + let buffer = sample.buffer().context("pulled sample had no buffer")?; + let Some(capture_time_us) = buffer.pts().map(|t| t.useconds() as i64) else { + tracing::debug!("skipped an encoded frame with no PTS"); + continue; + }; + let map = buffer.map_readable().context("failed to map sample buffer readable")?; + let is_key_frame = !buffer.flags().contains(gst::BufferFlags::DELTA_UNIT); + return Ok(Some((map.as_slice().to_vec(), is_key_frame, capture_time_us))); + } } /// Sends an upstream "force key unit" event from `appsink`, propagating to diff --git a/breadcastd/src/cast_mirror.rs b/breadcastd/src/cast_mirror.rs index 4326d33..3d37eab 100644 --- a/breadcastd/src/cast_mirror.rs +++ b/breadcastd/src/cast_mirror.rs @@ -24,11 +24,43 @@ use rust_cast::channels::receiver::CastDeviceApp; use crate::daemon::DaemonCommand; +/// The encoder's starting target bitrate, in kbps -- must match the +/// `bitrate=` property `build_video_pipeline_for_streaming` builds the +/// `vah264enc` with, since [`bitrate_control_step`] treats it as the value +/// already in effect at t=0. +const INITIAL_BITRATE_KBPS: u32 = 4000; +/// Never encode below this. 720p30 below roughly 1 Mbps is a wall of +/// blocking artifacts -- if the link genuinely can't carry that, dropping +/// frames is a better failure mode than shipping unwatchable video. +const MIN_BITRATE_KBPS: u32 = 1000; +/// Never encode above this, regardless of how much headroom the estimator +/// reports. Matches `VideoParams::default().max_bitrate_bps`, i.e. what the +/// OFFER told the receiver to expect. +const MAX_BITRATE_KBPS: u32 = 8000; + pub struct CastMirrorSession { pipeline: gst::Pipeline, session: CastSession, capture: Option, - threads: Vec>, + /// The FFI Cast Streaming session. Held here (rather than only inside + /// the pump-thread closures, as an earlier version did) so its + /// `Drop` -- which calls `breadcast_caststream_sender_destroy` and + /// blocks until openscreen's threads stop -- happens at an explicit, + /// deterministic point in [`Self::stop`], instead of "whichever + /// detached pump thread happened to drop the last `Arc`." + sender: Option>, + /// Forwards inbound CASTV2 `urn:x-cast:com.google.cast.webrtc` messages + /// (the ANSWER) into the FFI session. Ends when [`CastSession::stop`] + /// closes the raw-message channel. Holds an `Arc`. + message_pump: Option>, + /// Forwards outbound FFI events (the OFFER) onto the CASTV2 connection. + /// Ends only once the `CastStreamSender` itself is dropped (that is what + /// closes the event channel), so it must be joined *after* `sender` is + /// dropped, not before -- joining it first would deadlock. + event_pump: Option>, + /// Pulls encoded frames from the appsink into the FFI session. Ends on + /// pipeline EOS/flush. Holds an `Arc`. + frame_pump: Option>, } impl CastMirrorSession { @@ -78,9 +110,7 @@ impl CastMirrorSession { .context("failed to start the Cast Streaming session")?; let sender = Arc::new(sender); - let mut threads = Vec::new(); - - threads.push({ + let message_pump = { let sender = sender.clone(); std::thread::spawn(move || { while let Some(msg) = raw_messages.recv() { @@ -89,10 +119,10 @@ impl CastMirrorSession { } } }) - }); + }; let negotiated = Arc::new(AtomicBool::new(false)); - threads.push({ + let event_pump = { let session = session.clone(); let negotiated = negotiated.clone(); std::thread::spawn(move || { @@ -109,7 +139,7 @@ impl CastMirrorSession { } } }) - }); + }; tracing::info!(device = %device.name, "sending Cast Streaming OFFER"); sender.negotiate(); @@ -127,8 +157,9 @@ impl CastMirrorSession { pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?; tracing::info!(device = %device.name, "mirroring started"); - threads.push({ + let frame_pump = { let device_name = device.name.clone(); + let sender = sender.clone(); std::thread::spawn(move || { let result = frame_pump_loop(&appsink, &encoder, &sender); if let Err(e) = result { @@ -140,15 +171,35 @@ impl CastMirrorSession { // thread than drop the notification. let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded); }) - }); + }; - Ok(Self { pipeline, session, capture: Some(capture), threads }) + Ok(Self { + pipeline, + session, + capture: Some(capture), + sender: Some(sender), + message_pump: Some(message_pump), + event_pump: Some(event_pump), + frame_pump: Some(frame_pump), + }) } - /// Tears down the session: stops the pipeline (which unblocks the frame - /// pump thread's blocking `appsink.pull_sample()` call), stops the - /// CASTV2 session (which ends its io thread, closing the channels the - /// other two pump threads block on), then joins every thread. + /// Tears down the session. Order matters and is not interchangeable: + /// + /// 1. Pipeline to `Null` -- unblocks the frame pump's blocking + /// `appsink.pull_sample()`, so it can exit and release its + /// `Arc`. + /// 2. Stop the CASTV2 session -- ends its io thread, closing the + /// raw-message channel the message pump blocks on, so it too can exit + /// and release its `Arc`. + /// 3. Join those two. After this, no thread is calling into the FFI + /// session and `self.sender` holds the only remaining `Arc`. + /// 4. Drop `self.sender` -- runs `breadcast_caststream_sender_destroy` + /// (blocking until openscreen's threads stop) at a point where + /// nothing else can be mid-call into it, and closes the FFI event + /// channel. + /// 5. Only *then* join the event pump, which blocks on that channel and + /// would deadlock if joined before step 4. pub async fn stop(mut self) { if let Err(e) = self.pipeline.set_state(gst::State::Null) { tracing::warn!(error = ?e, "failed to stop the encode pipeline cleanly"); @@ -161,31 +212,93 @@ impl CastMirrorSession { tracing::warn!(error = ?e, "failed to cleanly close the portal capture session"); } } - for thread in self.threads.drain(..) { - // These threads all end once the pipeline/session teardown - // above propagates to them (see this method's own doc comment) - // -- `spawn_blocking` just keeps `.join()`'s wait off the async - // runtime's worker threads. - if let Err(panic) = tokio::task::spawn_blocking(move || thread.join()).await { - tracing::warn!(error = ?panic, "mirror session pump thread join task panicked"); - } + + join_pump(self.frame_pump.take(), "frame").await; + join_pump(self.message_pump.take(), "message").await; + + // Step 4: the blocking FFI teardown, kept off the async runtime's + // worker threads for the same reason the joins are. + if let Some(sender) = self.sender.take() { + let _ = tokio::task::spawn_blocking(move || drop(sender)).await; } + + join_pump(self.event_pump.take(), "event").await; } } +/// `spawn_blocking` just keeps `.join()`'s wait off the async runtime's +/// worker threads. +async fn join_pump(handle: Option>, what: &str) { + let Some(handle) = handle else { return }; + if let Err(panic) = tokio::task::spawn_blocking(move || handle.join()).await { + tracing::warn!(pump = what, error = ?panic, "mirror session pump thread join task panicked"); + } +} + +/// One step of the encoder-bitrate congestion-control loop: given the +/// currently-applied target and openscreen's latest bandwidth estimate, +/// returns the new target in kbps. +/// +/// openscreen's `BandwidthEstimator` deliberately *under*-estimates capacity +/// whenever the transmit rate is below it (see its class comment in +/// `vendor/openscreen/cast/streaming/impl/bandwidth_estimator.h`), and +/// prescribes a TCP-like response: cut hard when the estimate is below the +/// current target, ramp back up *gradually* when it's above. An earlier +/// version of this loop instead did `target = 0.85 * estimate` every second +/// unconditionally, which multiplies the target by <= 0.85 once a second +/// with no way back up -- 4000 kbps collapses past 1500 within ~6 seconds +/// and pins at the floor, which is exactly the "low quality / compression +/// artifacts" symptom, on a perfectly healthy LAN. +/// +/// An estimate of 0 means "not enough recent data to say" (documented +/// return value), and must leave the target alone rather than be treated as +/// a zero-bandwidth link. +fn bitrate_control_step(current_kbps: u32, estimate_bps: i32) -> u32 { + if estimate_bps <= 0 { + return current_kbps; + } + let estimate_kbps = (estimate_bps / 1000) as u32; + let next = if estimate_kbps < current_kbps { + // Below target: back off immediately to just under the estimate. + ((estimate_kbps as f64) * 0.85) as u32 + } else { + // Headroom: probe upward by 10% per second, not straight to the + // estimate -- the estimate is a lower bound, and jumping to it + // oscillates. + current_kbps + current_kbps / 10 + }; + next.clamp(MIN_BITRATE_KBPS, MAX_BITRATE_KBPS) +} + fn frame_pump_loop( appsink: &gstreamer_app::AppSink, encoder: &gst::Element, sender: &CastStreamSender, ) -> Result<()> { let mut last_bitrate_update = std::time::Instant::now(); + let mut current_kbps = INITIAL_BITRATE_KBPS; + // `needs_key_frame()` is a snapshot of an atomic the C++ side only + // refreshes every 100ms, so it stays true for several frames after a + // request has already been sent upstream. Firing a force-key-unit event + // per frame in that window makes the encoder emit a burst of IDRs, which + // under CBR eats the whole bitrate budget and produces a visible quality + // dip on every picture-loss report. One request per refresh window is + // enough. + let mut last_key_frame_request: Option = None; loop { let Some((data, is_key_frame, capture_time_us)) = pull_encoded_frame(appsink)? else { return Ok(()); // EOS -- pipeline was set to Null, or the portal source ended }; - if sender.needs_key_frame() && !is_key_frame { + if sender.needs_key_frame() + && !is_key_frame + && last_key_frame_request.is_none_or(|t| t.elapsed() >= std::time::Duration::from_millis(250)) + { request_key_frame(appsink); + last_key_frame_request = Some(std::time::Instant::now()); + } + if is_key_frame { + last_key_frame_request = None; } if let Err(e) = sender.enqueue_frame(&data, is_key_frame, capture_time_us) { @@ -193,10 +306,49 @@ fn frame_pump_loop( } if last_bitrate_update.elapsed() >= std::time::Duration::from_secs(1) { - let bps = sender.estimated_bandwidth_bps(); - let target_kbps = ((bps as f64 * 0.85) / 1000.0).max(500.0) as u32; - set_video_bitrate_kbps(encoder, target_kbps); + let next_kbps = bitrate_control_step(current_kbps, sender.estimated_bandwidth_bps()); + if next_kbps != current_kbps { + current_kbps = next_kbps; + set_video_bitrate_kbps(encoder, current_kbps); + } last_bitrate_update = std::time::Instant::now(); } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_zero_estimate_leaves_the_target_alone() { + assert_eq!(bitrate_control_step(4000, 0), 4000); + assert_eq!(bitrate_control_step(4000, -1), 4000); + } + + #[test] + fn headroom_ramps_up_gradually_and_is_capped() { + assert_eq!(bitrate_control_step(4000, 20_000_000), 4400); + assert_eq!(bitrate_control_step(MAX_BITRATE_KBPS, 20_000_000), MAX_BITRATE_KBPS); + } + + #[test] + fn a_low_estimate_backs_off_but_not_below_the_floor() { + assert_eq!(bitrate_control_step(4000, 2_000_000), 1700); + assert_eq!(bitrate_control_step(4000, 100_000), MIN_BITRATE_KBPS); + } + + /// The regression this loop exists to prevent: a *steady* estimate at + /// roughly the current encode rate must hold the target there (AIMD + /// oscillates a little around it, which is fine), not ratchet it down + /// once per second the way `target = 0.85 * estimate` did -- that + /// reached the floor in about a dozen iterations. + #[test] + fn a_steady_estimate_does_not_spiral_downward() { + let mut kbps = 4000; + for _ in 0..60 { + kbps = bitrate_control_step(kbps, 4_000_000); + assert!(kbps >= 3000, "target spiralled down to {kbps} kbps on a steady 4 Mbps estimate"); + } + } +} diff --git a/breadcastd/src/dlna_mirror.rs b/breadcastd/src/dlna_mirror.rs index 73262de..c3c68e2 100644 --- a/breadcastd/src/dlna_mirror.rs +++ b/breadcastd/src/dlna_mirror.rs @@ -81,7 +81,13 @@ impl DlnaMirrorSession { .context("failed to start the HLS HTTP server")?; let stream_url = http.url(lan_ip, "playlist.m3u8"); - wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 3, Duration::from_secs(20)) + // Two segments, not three: this is a "don't hand the renderer a 404 + // playlist" guard, and every segment waited for here is a segment of + // already-stale video sitting between the renderer and live (see + // `build_video_pipeline`'s note on HLS latency). Two is the minimum + // that still proves the encoder is genuinely producing output rather + // than having emitted one segment and stalled. + wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 2, Duration::from_secs(20)) .await .context("encode pipeline never produced playable HLS segments")?;