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")?;