Breadway/breadcast
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Compare commits

base: Breadway:2cc131075209db0f10f26ee61b67f57b20e319e3
Branches Tags
Breadway:main
Breadway:fix/per-monitor-app-css
Breadway:v0.1.1-rc.1
Breadway:v0.1.0
..
compare: Breadway:9c6fc61f5351df8ff0094788a17dfbb5efbedb02
Branches Tags
Breadway:fix/per-monitor-app-css
Breadway:main
Breadway:v0.1.1-rc.1
Breadway:v0.1.0

No commits in common. "2cc131075209db0f10f26ee61b67f57b20e319e3" and "9c6fc61f5351df8ff0094788a17dfbb5efbedb02" have entirely different histories.
2cc1310752 ... 9c6fc61f53

14 changed files with 95 additions and 1076 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.gitignore vendored
View file

@ -31,7 +31,3 @@ logs/
*.pid
# Local hygiene notes (not for commit)
CLAUDE.md
# graphify knowledge-graph output (local tool cache, not for commit)
graphify-out/

170
breadcast-caststream-sys/src/facade.cc
View file

@ -1,7 +1,6 @@
#include "facade.h"
#include <atomic>
#include <chrono>
#include <future>
#include <memory>
#include <string>
@ -45,34 +44,6 @@ 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<Environment>`, 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<bool> 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
@ -81,41 +52,6 @@ struct CastStreamSender {
std::atomic<bool> needs_key_frame{true};
std::atomic<int32_t> estimated_bandwidth_bps{kDefaultBandwidthEstimateBps};
// Set (never cleared except by the one consuming read, see
// breadcast_caststream_sender_needs_key_frame below) whenever a frame gets
// silently dropped after already being encoded -- either
// Sender::EnqueueFrame() rejecting it (e.g. MAX_DURATION_IN_FLIGHT, the
// in-flight budget openscreen enforces) or the non-monotonic-capture-time
// guard below. Either way, vah264enc already encoded the *next* frame as a
// P-slice depending on the one that just got dropped -- the encoder has no
// idea the drop happened, since it happens downstream of encoding, at this
// FFI boundary. Left alone, that reference is now dangling: the receiver
// decodes it against whatever picture it last successfully received,
// producing a stuck or corrupted frame that only resolves at the next
// regularly-scheduled key frame (key-int-max=60 -- up to ~2s, longer still
// if that key frame is itself dropped the same way). Forcing a key frame
// on the very next enqueue turns "up to several seconds of corruption"
// into "one dropped frame, then a clean resync" -- a separate flag rather
// than reusing `needs_key_frame` directly because SchedulePoll's 100ms
// timer unconditionally overwrites that one with the Sender's own
// (unrelated) NeedsKeyFrame() reading, which would silently clobber this
// signal before frame_pump_loop ever observed it.
std::atomic<bool> frame_chain_broken{false};
// See BreadcastEnqueueStats in facade.h. Counters are reset by the reading
// call; gauges are overwritten at every enqueue attempt. All relaxed --
// these are diagnostics, and a torn read across two of them costs nothing
// but a slightly inconsistent log line.
std::atomic<int32_t> enqueue_ok{0};
std::atomic<int32_t> enqueue_payload_too_large{0};
std::atomic<int32_t> enqueue_id_span_limit{0};
std::atomic<int32_t> enqueue_max_duration_in_flight{0};
std::atomic<int32_t> dropped_non_monotonic{0};
std::atomic<int32_t> in_flight_frames{0};
std::atomic<int32_t> in_flight_ms{0};
std::atomic<int32_t> max_in_flight_ms{0};
std::atomic<int32_t> round_trip_time_ms{0};
// The caller's own user_data + callbacks, as passed to `_create`. Not
// called directly -- session.h/message_port_bridge.h are instead given
// trampolines below (with `this` as their user_data) so this struct can
@ -126,14 +62,8 @@ 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);
@ -256,10 +186,6 @@ 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();
});
@ -276,10 +202,6 @@ void breadcast_caststream_sender_on_message(CastStreamSender* sender,
auto ns = std::make_shared<std::string>(message_namespace, message_namespace_len);
auto body = std::make_shared<std::string>(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);
});
}
@ -303,32 +225,15 @@ 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) {
sender->frame_chain_broken.store(true, std::memory_order_relaxed);
sender->dropped_non_monotonic.fetch_add(1, std::memory_order_relaxed);
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 =
@ -349,74 +254,17 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
ByteView(owned_data->data(), owned_data->size()));
// EnqueueFrame()'s result (e.g. MAX_DURATION_IN_FLIGHT under backpressure)
// can't be propagated to enqueue_frame()'s caller synchronously -- by the
// time this runs, that call has already returned 0 (this call is posted,
// not immediate -- see facade.h's threading contract). What it *can* do
// is flag the drop so the next frame comes in clean -- see
// `frame_chain_broken`'s doc comment on why that matters here, not just
// for the encoder's bitrate.
// Sampled *before* the enqueue attempt, so these describe the state the
// Sender used to make its accept/reject decision below rather than the
// state after it. See BreadcastEnqueueStats in facade.h.
const auto to_ms = [](Clock::duration d) {
return static_cast<int32_t>(
std::chrono::duration_cast<std::chrono::milliseconds>(d).count());
};
sender->in_flight_frames.store(
static_cast<int32_t>(video_sender->GetInFlightFrameCount()),
std::memory_order_relaxed);
sender->in_flight_ms.store(
to_ms(video_sender->GetInFlightMediaDuration(rtp_timestamp)),
std::memory_order_relaxed);
sender->max_in_flight_ms.store(
to_ms(video_sender->GetMaxInFlightMediaDuration()),
std::memory_order_relaxed);
sender->round_trip_time_ms.store(
to_ms(video_sender->GetCurrentRoundTripTime()),
std::memory_order_relaxed);
switch (video_sender->EnqueueFrame(frame)) {
case Sender::OK:
sender->enqueue_ok.fetch_add(1, std::memory_order_relaxed);
break;
case Sender::PAYLOAD_TOO_LARGE:
sender->enqueue_payload_too_large.fetch_add(1, std::memory_order_relaxed);
sender->frame_chain_broken.store(true, std::memory_order_relaxed);
break;
case Sender::REACHED_ID_SPAN_LIMIT:
sender->enqueue_id_span_limit.fetch_add(1, std::memory_order_relaxed);
sender->frame_chain_broken.store(true, std::memory_order_relaxed);
break;
case Sender::MAX_DURATION_IN_FLIGHT:
sender->enqueue_max_duration_in_flight.fetch_add(1, std::memory_order_relaxed);
sender->frame_chain_broken.store(true, std::memory_order_relaxed);
break;
}
// isn't propagated to the caller: by the time this runs, enqueue_frame()
// has already returned 0 synchronously (this call is posted, not
// immediate -- see facade.h's threading contract). Backpressure here just
// means this one frame is dropped; the encoder finds out indirectly via
// needs_key_frame()/estimated_bandwidth_bps() polling.
(void)video_sender->EnqueueFrame(frame);
});
return 0;
}
void breadcast_caststream_sender_take_stats(CastStreamSender* sender,
BreadcastEnqueueStats* out) {
if (!sender || !out) {
return;
}
out->enqueue_ok = sender->enqueue_ok.exchange(0, std::memory_order_relaxed);
out->enqueue_payload_too_large =
sender->enqueue_payload_too_large.exchange(0, std::memory_order_relaxed);
out->enqueue_id_span_limit =
sender->enqueue_id_span_limit.exchange(0, std::memory_order_relaxed);
out->enqueue_max_duration_in_flight =
sender->enqueue_max_duration_in_flight.exchange(0, std::memory_order_relaxed);
out->dropped_non_monotonic =
sender->dropped_non_monotonic.exchange(0, std::memory_order_relaxed);
out->in_flight_frames = sender->in_flight_frames.load(std::memory_order_relaxed);
out->in_flight_ms = sender->in_flight_ms.load(std::memory_order_relaxed);
out->max_in_flight_ms = sender->max_in_flight_ms.load(std::memory_order_relaxed);
out->round_trip_time_ms = sender->round_trip_time_ms.load(std::memory_order_relaxed);
}
int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender) {
return sender->needs_key_frame.load(std::memory_order_relaxed) ? 1 : 0;
}
@ -429,12 +277,6 @@ 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.

53
breadcast-caststream-sys/src/facade.h
View file

@ -103,59 +103,6 @@ int32_t breadcast_caststream_sender_enqueue_frame(CastStreamSender* sender,
int32_t is_key_frame,
int64_t capture_time_us);
// A snapshot of why frames are (or aren't) making it into the Sender.
//
// The `enqueue_frame` entry point above cannot report this: it returns as
// soon as the frame is *posted* to openscreen's TaskRunner, long before
// Sender::EnqueueFrame actually runs and decides. So a caller watching only
// its return value sees a 100% success rate even while every frame is being
// rejected downstream -- which is exactly the blind spot that made a
// multi-second picture freeze look, from the sender's own counters, like a
// perfectly healthy 30fps stream.
//
// The four `enqueue_*` counters are cumulative-since-last-read: reading
// them resets them to zero, so a caller polling once a second gets per-second
// rates directly. The remaining fields are instantaneous gauges, sampled on
// the TaskRunner thread at the moment of the most recent enqueue attempt.
typedef struct BreadcastEnqueueStats {
// Sender::EnqueueFrame returned OK -- the frame is genuinely in flight.
int32_t enqueue_ok;
// Sender::PAYLOAD_TOO_LARGE -- the encoded access unit needs more RTP
// packets than the packetizer allows.
int32_t enqueue_payload_too_large;
// Sender::REACHED_ID_SPAN_LIMIT -- more than kMaxUnackedFrames (120)
// frames have gone unacknowledged.
int32_t enqueue_id_span_limit;
// Sender::MAX_DURATION_IN_FLIGHT -- the in-flight media window
// (see `in_flight_ms`/`max_in_flight_ms`) is full. The expected symptom
// of the receiver's acknowledgements stalling.
int32_t enqueue_max_duration_in_flight;
// Frames dropped by the facade before ever reaching EnqueueFrame, by the
// non-monotonic-capture-time guard in enqueue_frame.
int32_t dropped_non_monotonic;
// Sender::GetInFlightFrameCount() at the last enqueue attempt.
int32_t in_flight_frames;
// Sender::GetInFlightMediaDuration() at the last enqueue attempt, in ms --
// i.e. the media timespan between the oldest unacknowledged frame and the
// one being enqueued. Note this is a *timespan*, not a byte count: frame
// size has no bearing on it, so a large key frame is neither more nor less
// likely to be rejected than a small P-frame.
int32_t in_flight_ms;
// Sender::GetMaxInFlightMediaDuration() at the last enqueue attempt, in ms.
// A frame is rejected when `in_flight_ms` would exceed this. openscreen
// computes it as clamp(2*RTT, kMinSenderInFlight, playout_delay/3), so on
// a low-latency LAN it sits at the kMinSenderInFlight floor.
int32_t max_in_flight_ms;
// Sender::GetCurrentRoundTripTime() at the last enqueue attempt, in ms.
int32_t round_trip_time_ms;
} BreadcastEnqueueStats;
// Fills `out` with the current stats and resets the counters. Safe to call
// from any thread; cheap, non-blocking, lock-free.
void breadcast_caststream_sender_take_stats(CastStreamSender* sender,
BreadcastEnqueueStats* out);
// True (nonzero) if the receiver wants a key frame as soon as possible.
// Safe to poll frequently; cheap, non-blocking, lock-free.
int32_t breadcast_caststream_sender_needs_key_frame(CastStreamSender* sender);

33
breadcast-caststream-sys/src/lib.rs
View file

@ -45,29 +45,6 @@ pub type OnErrorFn =
pub type OnPictureLostFn = extern "C" fn(user_data: *mut c_void);
/// Mirrors `BreadcastEnqueueStats` in `facade.h` -- see that struct's doc
/// comment for what each field means and why they exist at all (short
/// version: `sender_enqueue_frame`'s return value reports only that the
/// frame was *posted* to openscreen's TaskRunner, never whether
/// `Sender::EnqueueFrame` subsequently accepted it, so it reads 100% success
/// even while every frame is being rejected).
///
/// The `enqueue_*`/`dropped_*` fields are counts since the previous
/// `sender_take_stats` call; the rest are instantaneous gauges.
#[repr(C)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct EnqueueStats {
pub enqueue_ok: i32,
pub enqueue_payload_too_large: i32,
pub enqueue_id_span_limit: i32,
pub enqueue_max_duration_in_flight: i32,
pub dropped_non_monotonic: i32,
pub in_flight_frames: i32,
pub in_flight_ms: i32,
pub max_in_flight_ms: i32,
pub round_trip_time_ms: i32,
}
unsafe extern "C" {
/// Returns null on failure (e.g. an unparseable `remote_ip`, or the
/// local UDP socket failed to bind).
@ -136,16 +113,6 @@ unsafe extern "C" {
capture_time_us: i64,
) -> i32;
/// Fills `out` with the current enqueue stats and resets the counters.
///
/// # Safety
/// `sender` must be live and `out` must be a valid, writable pointer to
/// an `EnqueueStats` for the duration of this call.
pub fn breadcast_caststream_sender_take_stats(
sender: *mut CastStreamSender,
out: *mut EnqueueStats,
);
/// # Safety
/// `sender` must be live.
pub fn breadcast_caststream_sender_needs_key_frame(sender: *mut CastStreamSender) -> i32;

29
breadcast-caststream-sys/src/session.cc
View file

@ -39,33 +39,6 @@ using openscreen::cast::VideoStream;
// audio-then-video streams collapses to just "index 0 is the video stream."
constexpr int kVideoStreamIndex = 0;
// The playout delay breadcast asks the receiver for -- the window between
// capture here and presentation there. Deliberately *not*
// openscreen::cast::kDefaultTargetPlayoutDelay (400ms), because that value
// turned out to be the binding constraint on throughput, not just on latency.
//
// SenderImpl::GetMaxInFlightMediaDuration() computes the sender's send window
// as clamp(2*RTT, kMinSenderInFlight, target_playout_delay/3). At a 400ms
// target that ceiling is 133ms -- about four frames at 30 FPS. Instrumented
// measurement against real hardware (see BreadcastEnqueueStats in facade.h)
// found the round-trip time to a Chromecast over Wi-Fi sitting at 57-145ms,
// i.e. 2*RTT of 114-290ms: consistently *above* that 133ms ceiling. The
// window was therefore pinned at the ceiling and 3-40% of frames were being
// rejected with MAX_DURATION_IN_FLIGHT every second, each one silently
// breaking the H.264 reference chain and freezing the picture until the next
// key frame.
//
// Raising this to 1200ms lifts the ceiling to 400ms, so 2*RTT lands inside
// the clamp and the window tracks measured network conditions the way
// openscreen intended, instead of being capped below one round trip. The cost
// is ~800ms of additional end-to-end latency, which is unnoticeable for
// screen mirroring to a TV and a straight trade against multi-second freezes.
//
// Note this is only the *sender's* half of the fix: it must stay paired with
// the kMinSenderInFlight patch in vendor/openscreen (see PATCHES.md), which
// raises the floor of that same clamp for the moments when RTT dips.
constexpr std::chrono::milliseconds kTargetPlayoutDelay(1200);
VideoStream BuildVideoStream(const VideoParams& params,
bool use_android_rtp_hack) {
Stream stream;
@ -74,7 +47,7 @@ VideoStream BuildVideoStream(const VideoParams& params,
stream.channels = 1;
stream.rtp_payload_type = GetPayloadType(VideoCodec::kH264, use_android_rtp_hack);
stream.ssrc = GenerateSsrc(/*higher_priority=*/false);
stream.target_delay = kTargetPlayoutDelay;
stream.target_delay = openscreen::cast::kDefaultTargetPlayoutDelay;
stream.aes_key = GenerateRandomBytes16();
stream.aes_iv_mask = GenerateRandomBytes16();
stream.receiver_rtcp_event_log = true;

22
breadcast-caststream-sys/vendor/openscreen/PATCHES.md vendored
View file

@ -62,25 +62,3 @@ There are no release/API-stability guarantees upstream. To update:
(pulled in separately via gclient/DEPS in a full Chromium checkout).
Reimplemented on `EVP_EncodeBlock`/`EVP_DecodeBlock` from system OpenSSL
instead, same public interface.
6. **`cast/streaming/impl/sender_impl.cc`** — raised `kMinSenderInFlight`
from upstream's 66ms to 200ms. This is a behaviour patch, not a
portability one, and is the only one here that changes what goes on the
wire — so unlike the others it should *not* be silently re-applied when
rolling the pin without re-measuring first.
`GetMaxInFlightMediaDuration()` sizes the sender's send window as
`clamp(2*RTT, kMinSenderInFlight, target_playout_delay/3)`. Upstream's
66ms floor assumes the RTT is negligible, which holds for Chrome's own
usage but not for breadcast's measured case: instrumenting real
`Sender::EnqueueFrame` result codes (see `BreadcastEnqueueStats` in
`../../src/facade.h`) against a Chromecast over Wi-Fi showed RTT of
57-145ms and a steady 3-40% of frames per second rejected with
`MAX_DURATION_IN_FLIGHT`. Because breadcast enqueues already-encoded
frames, each such rejection silently breaks the H.264 reference chain
rather than merely dropping a frame, which is what the user-visible
multi-second picture freezes turned out to be.
Pairs with `kTargetPlayoutDelay` in `../../src/session.cc`, which raises
the *ceiling* of that same clamp (the ceiling, not this floor, was the
binding constraint at 400ms playout delay). Both are needed: this floor
covers RTT dips, that ceiling covers the normal case.

14
breadcast-caststream-sys/vendor/openscreen/cast/streaming/impl/sender_impl.cc vendored
View file

@ -24,18 +24,10 @@ namespace {
// The minimum amount of media the Sender keeps in-flight, regardless of the
// measured network round-trip time. This keeps the encoder pipeline flowing on
// low-latency networks. See crbug.com/498035450.
//
// LOCAL PATCH (breadcast): upstream is 66ms, roughly two video frames at
// 30 FPS. That is only enough when the round-trip time is genuinely
// negligible. Instrumented measurement against real hardware (a Chromecast
// over Wi-Fi) showed RTT swinging between 57ms and 145ms, so a 66ms floor
// leaves the send window narrower than a single round trip -- frames are
// rejected with MAX_DURATION_IN_FLIGHT faster than acknowledgements can
// free the window back up. A 200ms floor is ~6 frames at 30 FPS, enough to
// cover one round trip at the worst observed RTT. See PATCHES.md.
// low-latency networks (roughly two video frames at 30 FPS). See
// crbug.com/498035450.
constexpr Clock::duration kMinSenderInFlight =
Clock::to_duration(milliseconds(200));
Clock::to_duration(milliseconds(66));
} // namespace

14
breadcast-core/src/cast_sender.rs
View file

@ -326,19 +326,9 @@ fn run_io_loop(
// doc comment for why that's fine for this project's usage.
}
Ok(_) => {}
// Only end the loop on an error that means the connection
// itself is gone -- a single malformed/unexpected message (e.g.
// a MEDIA_STATUS missing a field this crate's struct treats as
// required) is a `Serialization`/`Parsing` error, not a dead
// socket, and used to take the whole receiver-status/control
// channel down with it for the rest of the session even though
// the RTP stream itself was unaffected.
Err(e @ (rust_cast::errors::Error::Io(_) | rust_cast::errors::Error::Tls(_) | rust_cast::errors::Error::Dns(_))) => {
tracing::debug!(error = %e, "cast session io loop ending: connection error");
return;
}
Err(e) => {
tracing::warn!(error = %e, "cast session: ignoring unparseable/unexpected message");
tracing::debug!(error = %e, "cast session io loop ending: receive error");
return;
}
}
}

56
breadcast-core/src/caststream.rs
View file

@ -24,7 +24,7 @@ use breadcast_caststream_sys::{
self as sys, breadcast_caststream_sender_create, breadcast_caststream_sender_destroy,
breadcast_caststream_sender_enqueue_frame, breadcast_caststream_sender_estimated_bandwidth_bps,
breadcast_caststream_sender_needs_key_frame, breadcast_caststream_sender_negotiate,
breadcast_caststream_sender_on_message, breadcast_caststream_sender_take_stats,
breadcast_caststream_sender_on_message,
};
/// The Cast Streaming ("Mirroring") receiver app id, pre-installed on every
@ -48,44 +48,10 @@ pub struct VideoParams {
impl Default for VideoParams {
fn default() -> Self {
Self {
// NOT the value that goes on the wire. `build_video_pipeline_for_streaming`
// returns the geometry it will really encode, and both
// `breadcastd::cast_mirror` and `cast_stream_test` pass *that*
// to `CastStreamSender::start` -- this default only supplies the
// fields that don't vary with the capture path (bitrate, frame
// rate denominator).
//
// It has to work that way because the pipeline picks its capture
// backend at runtime, and the two backends differ in resolution
// and frame rate. This OFFER's resolution is what the receiver
// allocates its decoder/output surface for, so advertising
// anything other than what's actually sent is a real protocol
// mismatch -- one that plausibly explains a receiver decoder
// corrupting/freezing rather than just looking soft. Keeping the
// two in sync by hand across two files is exactly how that got
// out of step before; returning it from the pipeline builder
// makes the mismatch unrepresentable. The values below are the
// DMA-BUF path's, kept only as a sane standalone default.
width: 1920,
height: 1080,
// Kept equal to `breadcastd::cast_mirror::MAX_BITRATE_KBPS *
// 1000` -- see that constant's doc comment for why 8 Mbps
// (this struct's previous value) isn't used here: real hardware
// testing showed the AIMD probe pinning to whatever this
// ceiling is for the entire session once the estimator reports
// (unreliably) that there's room, and 8 Mbps sustained was more
// than the previous network+receiver could actually hold,
// producing repeated multi-second freezes rather than just
// softer video.
max_bitrate_bps: 6_000_000,
// A *ceiling*, not a promise -- which is what makes it safe for
// the DMA-BUF path, where nothing caps the rate (`vapostproc` is
// a per-frame transform and can't do temporal conversion, see
// `build_video_pipeline_for_streaming`) and frames arrive at
// whatever the compositor delivers, up to this machine's 60Hz
// refresh. The wl_shm path does have a `videorate` capping it
// hard, and overrides this with the rate it actually enforces.
max_frame_rate_numerator: 60,
max_bitrate_bps: 8_000_000,
max_frame_rate_numerator: 30,
max_frame_rate_denominator: 1,
}
}
@ -251,22 +217,6 @@ impl CastStreamSender {
Ok(())
}
/// A snapshot of how the *underlying* `Sender::EnqueueFrame` has been
/// answering, plus its in-flight window gauges. Reading resets the
/// counters, so polling once a second yields per-second rates.
///
/// [`Self::enqueue_frame`] deliberately cannot report any of this: it
/// returns as soon as the frame is posted to openscreen's TaskRunner,
/// before the real accept/reject happens. Any "frames enqueued per
/// second" figure derived from its return value is therefore a count of
/// *attempts*, and stays pinned at the capture rate even while every
/// frame is being rejected downstream and the picture is frozen.
pub fn enqueue_stats(&self) -> sys::EnqueueStats {
let mut stats = sys::EnqueueStats::default();
unsafe { breadcast_caststream_sender_take_stats(self.raw, &mut stats) };
stats
}
/// True if the receiver wants a key frame as soon as possible. Cheap to
/// poll frequently (e.g. once per captured frame, before encoding it).
pub fn needs_key_frame(&self) -> bool {

6
breadcast-core/src/examples/cast_stream_test.rs
View file

@ -10,7 +10,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use breadcast_core::caststream::{CastStreamEvent, WEBRTC_NAMESPACE};
use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
use breadcast_core::net::local_lan_ip;
use breadcast_core::pipeline::{
build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
@ -58,7 +58,7 @@ async fn main() -> anyhow::Result<()> {
let capture = breadcast_core::CaptureSession::start().await?;
println!("Got PipeWire video node id: {}", capture.video_node_id());
let (pipeline, appsink, encoder, video_params) = build_video_pipeline_for_streaming(capture.video_node_id())?;
let (pipeline, appsink, encoder) = build_video_pipeline_for_streaming(capture.video_node_id())?;
// Watch the encode pipeline's own bus in the background -- see
// mirror_test.rs's identical block for why this matters.
@ -81,7 +81,7 @@ async fn main() -> anyhow::Result<()> {
&device.host,
"sender-0",
session.transport_id(),
video_params,
VideoParams::default(),
)?;
let sender = Arc::new(sender);

407
breadcast-core/src/pipeline/mod.rs
View file

@ -1,6 +1,4 @@
use std::io::{Read as _, Write as _};
use std::os::unix::fs::DirBuilderExt;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::time::Duration;
@ -10,8 +8,6 @@ use gstreamer::prelude::*;
use gstreamer_app as gst_app;
use gstreamer_video as gst_video;
use crate::caststream::VideoParams;
/// Builds (but doesn't start) the capture → encode → mux → HLS pipeline for
/// a single video source. `output_dir` is created if it doesn't exist;
/// `hlssink3` writes `segment%05d.ts` files and `playlist.m3u8` there.
@ -66,37 +62,15 @@ pub fn build_video_pipeline(video_node_id: u32, output_dir: &Path) -> Result<gst
// (stretches to 16:9) — correctness/compatibility first, an
// aspect-preserving scale (letterbox via `videoscale
// add-borders=true`) is a follow-up, not a blocker.
// HLS segment sizing is *the* dominant term in this path's end-to-end
// latency, and the two knobs are coupled: a segment can only be cut on a
// key frame, so the real segment duration is `max(target-duration,
// GOP length)` no matter what `target-duration` says. With the previous
// `target-duration=2` + `key-int-max=60` (60 frames / 30fps = a 2s GOP),
// segments were 2s, and a renderer that buffers the customary three of
// them before starting playback sits ~6s behind live -- on top of
// however much of the playlist it decides to start from. `dlna_mirror.rs`
// then waited for 3 segments to exist before even handing over the URL,
// adding another ~6s of already-stale content.
//
// 1s segments (GOP dropped to 30 frames to make that actually
// achievable) roughly halve that. Going below 1s is not worth it here:
// MPEG-TS + a per-segment key frame means shorter segments cost real
// bitrate, and classic (non-LL) HLS clients don't reliably honour
// sub-second target durations anyway. Genuinely low latency on this path
// needs LL-HLS, which `hlssink3` does not implement -- the Cast
// Streaming path (`build_video_pipeline_for_streaming`) is the
// low-latency answer, and this one is the compatibility answer.
//
// `playlist-length`/`max-files` shrink to match so the playlist doesn't
// advertise a long backlog of stale segments for a client to start from.
let pipeline_str = format!(
"pipewiresrc path={video_node_id} do-timestamp=true ! \
videoconvert ! videoscale ! videorate ! \
video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
vah264enc bitrate=4000 key-int-max=30 rate-control=cbr ! \
vah264enc bitrate=4000 key-int-max=60 rate-control=cbr ! \
video/x-h264,profile=main ! \
h264parse config-interval=1 ! \
hlssink.video \
hlssink3 name=hlssink target-duration=1 playlist-length=3 max-files=6"
hlssink3 name=hlssink target-duration=2 playlist-length=6 max-files=10"
);
let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
@ -149,243 +123,23 @@ pub fn build_video_pipeline(video_node_id: u32, output_dir: &Path) -> Result<gst
/// `CastStreamSender::estimated_bandwidth_bps()` — see
/// [`request_key_frame`]/`set_video_bitrate_kbps` for the two knobs a
/// congestion-control loop needs).
/// Which capture-side memory path [`build_video_pipeline_for_streaming`] asks
/// PipeWire (and through it, the compositor's portal implementation) to hand
/// this pipeline. The two are not interchangeable: they have *different*
/// known failure modes on different compositor versions, which is why the
/// choice is made at runtime rather than baked into one pipeline string.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CaptureBackend {
/// Zero-copy: `pipewiresrc` hands DMA-BUFs straight to `vapostproc`,
/// which hands VA-memory straight to `vah264enc`. Nothing ever touches
/// system memory, and the portal never falls back to `wl_shm`.
Dmabuf,
/// `videoconvert ! videoscale ! videorate` on plain system memory, which
/// makes PipeWire request `wl_shm` buffers from the portal. CPU-costly
/// and stall-prone (see [`choose_capture_backend`]), but it does not go
/// anywhere near a compositor's DMA-BUF render-into-client-buffer path.
Shm,
}
/// First Hyprland release containing `renderer/rbo: avoid nullptr deref`
/// (upstream commit `ae1690c2`, PR #15167, 2026-06-18), tagged in v0.56.0 on
/// 2026-07-20.
///
/// Below this, asking Hyprland for a DMA-BUF screencast can **kill the
/// user's entire compositor session**, and there is nothing this pipeline
/// can do about it from the client side. The mechanism, confirmed against a
/// real coredump on v0.55.4 plus upstream's own source:
/// `Screenshare::CScreenshareFrame::copyDmabuf()` renders the monitor into
/// the client-supplied DMA-BUF via `IHyprRenderer::beginRender` ->
/// `getOrCreateRenderbuffer`. If `CGLRenderbuffer`'s constructor fails to
/// import the buffer (`createEGLImage` returns `EGL_NO_IMAGE_KHR`) it
/// early-returns leaving `m_framebuffer` null -- and v0.55.4's destructor
/// then unconditionally does `unbind(); m_framebuffer->release();` on that
/// null pointer while the failed renderbuffer is being torn down. The
/// resulting abort takes down Hyprland, every window, and (separately,
/// same instant) `xdg-desktop-portal-hyprland`. Upstream's fix is a one-line
/// `if (m_framebuffer)` guard; it converts the abort into a dropped frame.
///
/// Reported upstream at least three times (hyprwm/Hyprland #13487, #13543,
/// #13653, all v0.54.x, all on AMD) and auto-closed unread by the
/// issues-are-disabled bot rather than triaged, so the "is it fixed?"
/// question can only be answered from the commit log, not the tracker. The
/// reported triggers (window-group tab switching, touchpad gestures, an
/// emulator in a Discord stream) have nothing in common with each other or
/// with resolution -- treat the import failure as intermittent, not as
/// something a particular capture geometry or DRM modifier provokes.
///
/// Note in particular that this is *not* avoidable by requesting a
/// "simpler" buffer layout. `vapostproc` advertises exactly one AMD DRM
/// modifier on its `video/x-raw(memory:DMABuf)` pads -- `0x0200000008401b04`
/// = GFX11, 64K_R_X tiling, `DCC=0` (verified with `gst-inspect-1.0
/// vapostproc` and `drm_fourcc.h`'s field shifts). It offers no LINEAR
/// alternative, and the one modifier it does offer is already uncompressed,
/// so there is no tiling/compression hazard left to negotiate away.
const HYPRLAND_MIN_SAFE_DMABUF: (u32, u32, u32) = (0, 56, 0);
/// Reads the running Hyprland's version over its own IPC socket (the same
/// `j/version` request `hyprctl version -j` makes) without spawning
/// `hyprctl`, which needn't be installed. `None` if this isn't a Hyprland
/// session at all, or if the version can't be determined.
fn hyprland_version() -> Option<(u32, u32, u32)> {
let signature = std::env::var("HYPRLAND_INSTANCE_SIGNATURE").ok()?;
let runtime_dir = std::env::var("XDG_RUNTIME_DIR").ok()?;
let mut socket = UnixStream::connect(format!("{runtime_dir}/hypr/{signature}/.socket.sock")).ok()?;
// Bounded on both halves: this runs on the way into starting a mirror
// session, and a wedged compositor must not be able to hang that.
socket.set_write_timeout(Some(Duration::from_secs(1))).ok()?;
socket.set_read_timeout(Some(Duration::from_secs(1))).ok()?;
socket.write_all(b"j/version").ok()?;
let mut response = String::new();
socket.read_to_string(&mut response).ok()?;
let parsed: serde_json::Value = serde_json::from_str(&response).ok()?;
// `version` is the plain "0.55.4"; `tag` is "v0.55.4" and is what older
// Hyprlands report, so accept either.
let raw = parsed.get("version").or_else(|| parsed.get("tag"))?.as_str()?;
parse_hyprland_version(raw)
}
/// Splits a Hyprland version string into comparable components. Strips a
/// leading `v` (`tag` carries one, `version` doesn't) and anything from the
/// first `-` (a git build's tag looks like `v0.55.4-123-gdeadbee`).
fn parse_hyprland_version(raw: &str) -> Option<(u32, u32, u32)> {
let mut parts = raw.trim().trim_start_matches('v').split('-').next()?.split('.');
let major = parts.next()?.parse().ok()?;
let minor = parts.next()?.parse().ok()?;
// A two-component "0.56" is treated as 0.56.0 rather than rejected --
// erring toward *parsing* here is safe, since the comparison against
// `HYPRLAND_MIN_SAFE_DMABUF` is what decides anything.
let patch = parts.next().unwrap_or("0").parse().ok()?;
Some((major, minor, patch))
}
/// Picks the capture path, trading two *different* real bugs off against
/// each other rather than pretending either one is hypothetical.
///
/// [`CaptureBackend::Dmabuf`] is the better path and the default: it is
/// genuinely zero-copy, and it sidesteps `xdg-desktop-portal-hyprland`'s
/// `wl_shm` stall entirely. That stall is not a hiccup -- it is terminal.
/// In xdpw's `src/portals/Screencopy.cpp`, when the PipeWire consumer is
/// holding every buffer, the portal logs "Out of buffers" and re-queues a
/// frame only while `copyRetries++ < MAX_RETRIES` (10); `copyRetries` is
/// reset to 0 *only* on a successful copy. So ten consecutive misses and
/// the portal stops requesting frames forever, without sending an error to
/// PipeWire -- which is exactly why a 45-second freeze showed up in
/// `journalctl` and nowhere on this pipeline's own GStreamer bus. (Worth
/// keeping in mind that "the consumer is holding every buffer" means the
/// stall can *originate* downstream: a brief encoder or RTP-send stall stops
/// buffers being recycled, and the portal's give-up logic then makes it
/// permanent. [`pull_encoded_frame`]'s watchdog is the backstop for both.)
///
/// But on Hyprland older than [`HYPRLAND_MIN_SAFE_DMABUF`] the DMA-BUF path
/// can abort the compositor outright, which is a categorically worse outcome
/// than a stalled cast -- so there, fall back to `wl_shm` and let the
/// watchdog bound the damage. Non-Hyprland sessions are unaffected by that
/// bug and keep DMA-BUF.
fn choose_capture_backend() -> CaptureBackend {
let Some(version) = hyprland_version() else {
// Either not Hyprland (so the Hyprland-specific crash can't apply),
// or Hyprland with an unreadable version. The latter is the
// ambiguous case; prefer the path that cannot take the desktop down.
if std::env::var_os("HYPRLAND_INSTANCE_SIGNATURE").is_some() {
tracing::warn!(
"running under Hyprland but could not read its version; using the slower wl_shm \
capture path, since DMA-BUF screencast aborts the compositor before v{}.{}.{}",
HYPRLAND_MIN_SAFE_DMABUF.0,
HYPRLAND_MIN_SAFE_DMABUF.1,
HYPRLAND_MIN_SAFE_DMABUF.2,
);
return CaptureBackend::Shm;
}
return CaptureBackend::Dmabuf;
};
if version < HYPRLAND_MIN_SAFE_DMABUF {
tracing::warn!(
hyprland = format!("{}.{}.{}", version.0, version.1, version.2),
"this Hyprland predates the fix for the DMA-BUF screencast compositor crash \
(upstream PR #15167, released in v0.56.0) -- falling back to the slower, \
stall-prone wl_shm capture path. Updating Hyprland restores zero-copy capture."
);
return CaptureBackend::Shm;
}
CaptureBackend::Dmabuf
}
pub fn build_video_pipeline_for_streaming(
video_node_id: u32,
) -> Result<(gst::Pipeline, gst_app::AppSink, gst::Element, VideoParams)> {
) -> Result<(gst::Pipeline, gst_app::AppSink, gst::Element)> {
gst::init().context("failed to initialize GStreamer")?;
// 1920x1080@native-rate Main profile. Went 1080p -> 720p -> 1080p again
// tonight: the first 1080p attempt froze near-instantly, but that had
// nothing to do with resolution -- it was the `xdg-desktop-portal-hyprland`
// wl_shm buffer-exhaustion bug below (real, structural, whatever the
// resolution) compounded by openscreen's in-flight RTP budget being too
// tight for this receiver's actual RTT (see `frame_chain_broken` in
// `breadcast-caststream-sys/src/facade.cc`, and the playout-delay tuning
// in `breadcast-caststream-sys/src/session.cc`). With both of those
// fixed -- confirmed via a real, freeze-free 720p session -- the
// packet-count increase 1080p brings back is no longer landing on an
// already-struggling budget, so it's worth trying again on its own
// merits.
//
// On the DMA-BUF path `pipewiresrc` deliberately does *not* go through
// `videoconvert ! videoscale ! videorate ! video/x-raw,...` (plain
// system-memory caps) -- doing so forces PipeWire to hand the
// compositor's portal implementation a `wl_shm` (shared-memory) buffer
// request, and on this system (`xdg-desktop-portal-hyprland`) that path
// is real-world buggy: journalctl during a live freeze showed it
// repeatedly logging "Asked for a wl_shm buffer which is legacy" / "Out
// of buffers" / "Retrying screencopy" in a tight loop that never
// actually delivered a frame -- multi-second (once 45+ second) stalls
// with *zero* signal on breadcast's own GStreamer bus, since nothing
// here was erroring, it was just starved waiting on a buffer the
// portal's legacy path never produced. See `choose_capture_backend` for
// why that stall is permanent rather than transient, and for the one
// case where it's still the lesser evil.
//
// `vapostproc` (VA-API postprocessor -- confirmed present via
// `gst-inspect-1.0 vapostproc`, ships in `gst-plugins-bad`'s `va`
// plugin) accepts `video/x-raw(memory:DMABuf)` directly from
// `pipewiresrc` and outputs `video/x-raw(memory:VAMemory)`, which
// `vah264enc` also accepts natively -- a fully zero-copy DMA-BUF path
// from portal to hardware encoder that never touches the legacy wl_shm
// fallback at all. No `videorate` in this path: `vapostproc` is a
// per-frame transform (scale/convert), not a temporal one, so it can't
// do frame-rate reduction the way `videorate` does on raw memory --
// frames flow at whatever rate PipeWire actually delivers rather than a
// forced 30fps. This is fine for RTP: `facade.cc` derives RTP timestamps
// from each frame's real capture time regardless of the nominal rate,
// and openscreen's own frame pacing doesn't assume a fixed source rate
// either. The returned `VideoParams` advertises 60 as a *ceiling*
// (`max_frame_rate_*`), which stays truthful whether the compositor
// actually delivers 60 or fewer; the wl_shm path's `videorate` does cap
// hard, so it advertises the rate it really enforces.
//
// Both branches return the geometry they actually encode, and the caller
// hands that straight to the Cast OFFER. That coupling is deliberate:
// advertising a resolution other than what's really sent is a genuine
// protocol mismatch this project has already been bitten by once, and
// keeping two constants manually in sync across two files is how that
// happened. Returning it makes the mismatch unrepresentable.
let (pipeline_str, params) = match choose_capture_backend() {
CaptureBackend::Dmabuf => (
"pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
video/x-raw(memory:DMABuf),format=DMA_DRM ! \
vapostproc ! \
video/x-raw(memory:VAMemory),format=NV12,width=1920,height=1080 ! \
vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
video/x-h264,profile=main ! \
h264parse name=h264parse config-interval=-1 ! \
video/x-h264,stream-format=byte-stream,alignment=au ! \
appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true",
VideoParams { width: 1920, height: 1080, max_frame_rate_numerator: 60, ..VideoParams::default() },
),
// 720p30 rather than 1080p60 on this path on purpose: every frame is
// a CPU convert + scale here, and CPU cost is precisely what makes
// the portal's "Out of buffers" give-up more likely, since the
// portal runs out exactly when the consumer is slow to recycle
// buffers. The lighter shape is also what was last known to work on
// real hardware before the DMA-BUF switch.
CaptureBackend::Shm => (
"pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
videoconvert ! videoscale ! videorate ! \
video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
video/x-h264,profile=main ! \
h264parse name=h264parse config-interval=-1 ! \
video/x-h264,stream-format=byte-stream,alignment=au ! \
appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true",
VideoParams { width: 1280, height: 720, max_frame_rate_numerator: 30, ..VideoParams::default() },
),
};
let pipeline_str = pipeline_str.replace("%VIDEO_NODE_ID%", &video_node_id.to_string());
// Same 1280x720@30 Main-profile baseline as build_video_pipeline, for
// the same reason (see its doc comment) -- broad decoder compatibility
// first, revisit upward once a specific device's real ceiling is known.
let pipeline_str = "pipewiresrc path=%VIDEO_NODE_ID% do-timestamp=true ! \
videoconvert ! videoscale ! videorate ! \
video/x-raw,format=NV12,width=1280,height=720,framerate=30/1 ! \
vah264enc name=venc bitrate=4000 key-int-max=60 rate-control=cbr ! \
video/x-h264,profile=main ! \
h264parse name=h264parse config-interval=-1 ! \
video/x-h264,stream-format=byte-stream,alignment=au ! \
appsink name=appsink emit-signals=false sync=false max-buffers=4 drop=true"
.replace("%VIDEO_NODE_ID%", &video_node_id.to_string());
let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
let Ok(pipeline) = element.downcast::<gst::Pipeline>() else {
@ -400,95 +154,28 @@ pub fn build_video_pipeline_for_streaming(
let encoder = pipeline.by_name("venc").context("parsed pipeline has no element named 'venc'")?;
Ok((pipeline, appsink, encoder, params))
Ok((pipeline, appsink, encoder))
}
/// How long [`pull_encoded_frame`] waits per `try_pull_sample` call. Short
/// enough that a teardown from another thread is noticed promptly, long
/// enough not to spin.
const CAPTURE_STALL_POLL: Duration = Duration::from_millis(250);
/// How long [`pull_encoded_frame`] tolerates a *playing* pipeline producing
/// no frames at all before declaring the capture dead.
///
/// This exists because the failure it catches is otherwise completely
/// silent. `xdg-desktop-portal-hyprland` stops requesting frames after ten
/// consecutive "Out of buffers" misses and never sends an error to PipeWire
/// (see [`choose_capture_backend`]); a compositor that fails to import a
/// capture buffer likewise just drops the frame. In both cases GStreamer has
/// nothing to report -- no bus error, no EOS, no flow-return failure -- so
/// without a timeout here the frame pump blocks in `pull_sample` forever and
/// the mirror session appears frozen with nothing anywhere saying why. That
/// is precisely the 45-second freeze that took a `journalctl` dig to
/// explain.
///
/// Bailing propagates out of `breadcastd`'s frame-pump thread, which already
/// reports `DaemonCommand::SessionEnded` on exit, so the session tears down
/// and the failure surfaces as a real event instead of a hang. Generous
/// enough (10s) that a merely slow moment -- a heavy compositor frame, a
/// bitrate renegotiation -- doesn't trip it; anything longer than this is
/// not a hiccup, since neither of the known failure modes recovers.
const CAPTURE_STALL_TIMEOUT: Duration = Duration::from_secs(10);
/// Pulls one complete Annex-B H.264 access unit from `appsink`, blocking
/// until one is available. Returns `None` once the pipeline reaches EOS or
/// the sink otherwise stops (e.g. pipeline torn down from another thread),
/// and errors if the pipeline is still playing but has gone
/// [`CAPTURE_STALL_TIMEOUT`] without producing a frame.
///
/// A buffer with no PTS is skipped (this pulls the next one instead) rather
/// than reported with a substituted timestamp, as an earlier version did
/// with `.unwrap_or(0)`. That substitution was actively dangerous rather
/// than merely imprecise: openscreen derives the frame's RTP timestamp from
/// this value and enforces strict monotonicity with a *fatal* `OSP_CHECK`
/// (`sender_impl.cc`'s `OSP_CHECK_GT(frame.rtp_timestamp, ...)`), not an
/// error return -- so a single PTS-less buffer part-way into a session would
/// abort the whole daemon. `facade.cc` independently drops non-monotonic
/// capture times as a second line of defence; neither guard makes the other
/// redundant, since `enqueue_frame` is a public FFI entry point that has to
/// hold up against any caller.
/// the sink otherwise stops (e.g. pipeline torn down from another thread).
pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>, bool, i64)>> {
let poll = gst::ClockTime::from_mseconds(CAPTURE_STALL_POLL.as_millis() as u64);
let mut stalled_for = Duration::ZERO;
loop {
let Some(sample) = appsink.try_pull_sample(Some(poll)) else {
// EOS is the ordinary end: the user hit "Stop sharing" in the
// portal, or the source went away.
if appsink.is_eos() {
return Ok(None);
}
// Teardown from another thread (`CastMirrorSession::stop` sets
// the pipeline to Null) makes the sink flush, and a flushing
// sink returns `None` *immediately* rather than after the
// timeout. Treat that as a clean end too -- otherwise this would
// busy-spin for the whole stall budget and then report a
// spurious "capture stalled" on every normal stop.
if !matches!(appsink.current_state(), gst::State::Playing | gst::State::Paused) {
return Ok(None);
}
stalled_for += CAPTURE_STALL_POLL;
if stalled_for < CAPTURE_STALL_TIMEOUT {
continue;
}
bail!(
"capture stalled: no encoded frame for {}s while the pipeline was still \
playing (no GStreamer error, no EOS). This is the shape of a portal-side \
give-up -- see `choose_capture_backend` -- rather than a pipeline fault, \
and it will not recover on its own",
CAPTURE_STALL_TIMEOUT.as_secs()
);
};
stalled_for = Duration::ZERO;
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)));
}
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)))
}
/// Sends an upstream "force key unit" event from `appsink`, propagating to
@ -623,33 +310,3 @@ pub async fn wait_for_playlist_segments(playlist_path: &Path, min_segments: usiz
tokio::time::sleep(Duration::from_millis(250)).await;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_both_shapes_hyprland_reports() {
// `version` (plain) and `tag` (v-prefixed) from the same running
// compositor, plus the `-N-gSHA` suffix a git build's tag carries.
assert_eq!(parse_hyprland_version("0.55.4"), Some((0, 55, 4)));
assert_eq!(parse_hyprland_version("v0.55.4"), Some((0, 55, 4)));
assert_eq!(parse_hyprland_version("v0.56.0-123-gdeadbee"), Some((0, 56, 0)));
assert_eq!(parse_hyprland_version("0.56"), Some((0, 56, 0)));
assert_eq!(parse_hyprland_version(" v0.56.2\n"), Some((0, 56, 2)));
assert_eq!(parse_hyprland_version(""), None);
assert_eq!(parse_hyprland_version("unknown"), None);
}
#[test]
fn straddles_the_dmabuf_crash_fix_correctly() {
// The whole point of the constant: v0.55.4 aborts the compositor on
// a DMA-BUF screencast, v0.56.0 is the first release with the fix.
assert!(parse_hyprland_version("0.55.4").unwrap() < HYPRLAND_MIN_SAFE_DMABUF);
assert!(parse_hyprland_version("0.55.99").unwrap() < HYPRLAND_MIN_SAFE_DMABUF);
assert!(parse_hyprland_version("0.56.0").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
assert!(parse_hyprland_version("0.56.1").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
assert!(parse_hyprland_version("1.0.0").unwrap() >= HYPRLAND_MIN_SAFE_DMABUF);
}
}

323
breadcastd/src/cast_mirror.rs
View file

@ -13,7 +13,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use anyhow::{Context, Result};
use breadcast_core::caststream::{CastStreamEvent, WEBRTC_NAMESPACE};
use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
use breadcast_core::pipeline::{
build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
};
@ -24,53 +24,11 @@ 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.5 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 = 1500;
/// 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 -- see that constant's doc comment for
/// why this is 6 Mbps and not higher: real hardware testing showed the AIMD
/// probe below pinning to whatever this ceiling is for the *entire* session
/// (the estimator it trusts read a suspiciously flat ~20 Mbps almost the
/// whole time), and 8 Mbps sustained was more than the previous
/// network+receiver could actually hold without repeated multi-second
/// freezes -- though the deeper cause of those freezes turned out to be
/// `frame_chain_broken` in `facade.cc`, not bitrate on its own. 6 Mbps is
/// still a very generous ceiling for 720p30; revisit only with real
/// evidence this specific link+receiver can sustain more, not just because
/// the estimator claims there's headroom.
const MAX_BITRATE_KBPS: u32 = 6000;
pub struct CastMirrorSession {
pipeline: gst::Pipeline,
session: CastSession,
capture: Option<CaptureSession>,
/// 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<Arc<CastStreamSender>>,
/// 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<CastStreamSender>`.
message_pump: Option<std::thread::JoinHandle<()>>,
/// 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<std::thread::JoinHandle<()>>,
/// Pulls encoded frames from the appsink into the FFI session. Ends on
/// pipeline EOS/flush. Holds an `Arc<CastStreamSender>`.
frame_pump: Option<std::thread::JoinHandle<()>>,
threads: Vec<std::thread::JoinHandle<()>>,
}
impl CastMirrorSession {
@ -87,13 +45,7 @@ impl CastMirrorSession {
let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
let video_node_id = capture.video_node_id();
// `video_params` describes what this pipeline will *actually* encode
// -- it isn't a constant, because the pipeline picks its capture path
// at runtime (see `build_video_pipeline_for_streaming`) and the two
// paths differ in resolution and frame rate. It's threaded into the
// OFFER below rather than re-derived there, so the advertised stream
// and the encoded stream cannot drift apart.
let (pipeline, appsink, encoder, video_params) =
let (pipeline, appsink, encoder) =
build_video_pipeline_for_streaming(video_node_id).context("failed to build the encode pipeline")?;
{
@ -122,11 +74,13 @@ impl CastMirrorSession {
.context("failed to connect and launch the Mirroring receiver")?;
let (sender, stream_events) =
CastStreamSender::start(&device.host, "sender-0", session.transport_id(), video_params)
CastStreamSender::start(&device.host, "sender-0", session.transport_id(), VideoParams::default())
.context("failed to start the Cast Streaming session")?;
let sender = Arc::new(sender);
let message_pump = {
let mut threads = Vec::new();
threads.push({
let sender = sender.clone();
std::thread::spawn(move || {
while let Some(msg) = raw_messages.recv() {
@ -135,10 +89,10 @@ impl CastMirrorSession {
}
}
})
};
});
let negotiated = Arc::new(AtomicBool::new(false));
let event_pump = {
threads.push({
let session = session.clone();
let negotiated = negotiated.clone();
std::thread::spawn(move || {
@ -155,7 +109,7 @@ impl CastMirrorSession {
}
}
})
};
});
tracing::info!(device = %device.name, "sending Cast Streaming OFFER");
sender.negotiate();
@ -165,22 +119,16 @@ impl CastMirrorSession {
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
}
if !negotiated.load(Ordering::Acquire) {
// Bounded for the same reason `Self::stop`'s calls are -- an
// unresponsive receiver (which is exactly what "negotiation
// timed out" implies) can wedge either of these forever
// otherwise, taking the whole single-threaded daemon actor
// down with it before this even gets to return an error.
stop_session_bounded(&session).await;
close_capture_bounded(capture).await;
let _ = session.stop();
capture.close().await.ok();
anyhow::bail!("never received an ANSWER from {} (negotiation timed out)", device.name);
}
pipeline.set_state(gst::State::Playing).context("failed to start the encode pipeline")?;
tracing::info!(device = %device.name, "mirroring started");
let frame_pump = {
threads.push({
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 {
@ -192,256 +140,63 @@ impl CastMirrorSession {
// thread than drop the notification.
let _ = daemon_tx.blocking_send(DaemonCommand::SessionEnded);
})
};
});
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),
})
Ok(Self { pipeline, session, capture: Some(capture), threads })
}
/// 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<CastStreamSender>`.
/// 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.
/// 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.
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");
}
stop_session_bounded(&self.session).await;
if let Err(e) = self.session.stop() {
tracing::warn!(error = ?e, "failed to cleanly stop the cast session");
}
if let Some(capture) = self.capture.take() {
close_capture_bounded(capture).await;
if let Err(e) = capture.close().await {
tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
}
}
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;
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.event_pump.take(), "event").await;
}
}
/// `CastSession::stop` blocks on a round trip the receiver has to answer,
/// over a `rust_cast` connection that offers no read timeout -- if the
/// receiver has gone unresponsive (wedged decoder, dropped off the
/// network, etc.) that round trip never returns. Both callers run inside
/// the single-threaded daemon actor, so an unbounded wait here doesn't
/// just fail one stop -- it permanently freezes the entire daemon (every
/// future IPC request hangs too), recoverable only by killing the
/// process. Bound it: if the receiver hasn't answered in 5s, give up on a
/// graceful stop and let teardown continue anyway. The io thread may leak
/// (still blocked in that same call), but a single leaked thread beats an
/// unrecoverable daemon.
async fn stop_session_bounded(session: &CastSession) {
let session = session.clone();
match tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || session.stop())).await {
Ok(Ok(Err(e))) => tracing::warn!(error = ?e, "failed to cleanly stop the cast session"),
Ok(Err(panic)) => tracing::warn!(error = ?panic, "cast session stop task panicked"),
Err(_) => tracing::warn!("cast session did not acknowledge stop within 5s (receiver unresponsive?) -- tearing down anyway"),
Ok(Ok(Ok(()))) => {}
}
}
/// Same reasoning as [`stop_session_bounded`]: `capture.close()` is a
/// `Session::close()` D-Bus call to the xdg-desktop-portal backend, which
/// this project has observed to be flaky (the "Failed to populate
/// properties cache... UnknownMethod" warnings logged on every portal
/// session) -- an unbounded `.await` here is just as capable of freezing
/// the whole daemon actor if that call never gets a reply.
async fn close_capture_bounded(capture: CaptureSession) {
match tokio::time::timeout(std::time::Duration::from_secs(5), capture.close()).await {
Ok(Err(e)) => tracing::warn!(error = ?e, "failed to cleanly close the portal capture session"),
Err(_) => tracing::warn!("portal capture session did not close within 5s -- abandoning it"),
Ok(Ok(())) => {}
}
}
/// Bounded to 5s for the same reason [`CastMirrorSession::stop`]'s own wait
/// on `CastSession::stop` is: `message_pump` blocks on a channel only the
/// (possibly wedged, per that comment) cast session io thread ever closes,
/// so an unbounded join here is exactly as capable of freezing the whole
/// daemon actor forever. A timed-out thread is abandoned rather than
/// joined -- it may still be running, but nothing here waits on it again.
async fn join_pump(handle: Option<std::thread::JoinHandle<()>>, what: &str) {
let Some(handle) = handle else { return };
match tokio::time::timeout(std::time::Duration::from_secs(5), tokio::task::spawn_blocking(move || handle.join())).await {
Ok(Err(panic)) => tracing::warn!(pump = what, error = ?panic, "mirror session pump thread join task panicked"),
Err(_) => tracing::warn!(pump = what, "mirror session pump thread did not exit within 5s -- abandoning it"),
Ok(Ok(_)) => {}
}
}
/// 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<std::time::Instant> = None;
// Pulled-frame and successful-enqueue counters, logged once/sec
// alongside the bitrate step below -- otherwise a stalled pipeline
// (upstream not producing samples) and a stalled sender (producing
// samples nobody can get rid of) are both silent: this is a per-frame
// hot loop, so anything more than a periodic summary would flood the
// log rather than help debug either case.
let mut pulled_since_log: u32 = 0;
let mut enqueued_since_log: u32 = 0;
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
};
pulled_since_log += 1;
if sender.needs_key_frame()
&& !is_key_frame
&& last_key_frame_request.is_none_or(|t| t.elapsed() >= std::time::Duration::from_millis(250))
{
if sender.needs_key_frame() && !is_key_frame {
request_key_frame(appsink);
last_key_frame_request = Some(std::time::Instant::now());
}
if is_key_frame {
last_key_frame_request = None;
}
match sender.enqueue_frame(&data, is_key_frame, capture_time_us) {
Ok(()) => enqueued_since_log += 1,
Err(e) => tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)"),
if let Err(e) = sender.enqueue_frame(&data, is_key_frame, capture_time_us) {
tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)");
}
if last_bitrate_update.elapsed() >= std::time::Duration::from_secs(1) {
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);
}
// `enqueued_fps` counts *posted* frames, not accepted ones (see
// `CastStreamSender::enqueue_stats`) -- it is the `accepted_fps`
// and `rejected_*` fields below that say whether video is
// actually reaching the receiver. A run where `enqueued_fps`
// holds at 30 while `accepted_fps` drops to 0 is a frozen
// picture, and nothing else logged here would show it.
let stats = sender.enqueue_stats();
tracing::debug!(
pulled_fps = pulled_since_log,
enqueued_fps = enqueued_since_log,
accepted_fps = stats.enqueue_ok,
rejected_in_flight = stats.enqueue_max_duration_in_flight,
rejected_id_span = stats.enqueue_id_span_limit,
rejected_too_large = stats.enqueue_payload_too_large,
dropped_non_monotonic = stats.dropped_non_monotonic,
in_flight_frames = stats.in_flight_frames,
in_flight_ms = stats.in_flight_ms,
max_in_flight_ms = stats.max_in_flight_ms,
rtt_ms = stats.round_trip_time_ms,
bitrate_kbps = current_kbps,
estimated_bandwidth_bps = sender.estimated_bandwidth_bps(),
"frame pump rate"
);
pulled_since_log = 0;
enqueued_since_log = 0;
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);
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");
}
}
}

32
breadcastd/src/daemon.rs
View file

@ -73,12 +73,6 @@ impl ActiveSession {
}
}
/// `host` strings come straight from mDNS resolution, so this just checks
/// the address, not whether a zone id is attached (mDNS never gives us one).
fn is_link_local_v6(host: &str) -> bool {
matches!(host.parse::<std::net::IpAddr>(), Ok(std::net::IpAddr::V6(v6)) if (v6.segments()[0] & 0xffc0) == 0xfe80)
}
struct Daemon {
cast_devices: HashMap<String, CastDevice>,
/// Keyed by `DlnaDevice::url`, the closest thing DLNA has to a stable
@ -134,20 +128,7 @@ impl Daemon {
}
}
DaemonCommand::CastDeviceFound(device) => {
// mDNS resolves one physical device on every local address
// it has -- typically a private IPv4 and a link-local IPv6
// -- as separate events carrying the same id. A bare
// `fe80::` address has no interface scope attached, so
// `TcpStream::connect`ing to it fails outright; don't let
// one clobber an already-usable host just because it
// happened to resolve more recently.
let should_replace = !matches!(
self.cast_devices.get(&device.id),
Some(existing) if is_link_local_v6(&device.host) && !is_link_local_v6(&existing.host)
);
if should_replace {
self.cast_devices.insert(device.id.clone(), device);
}
self.cast_devices.insert(device.id.clone(), device);
self.broadcast_devices();
}
DaemonCommand::CastDeviceLost(id) => {
@ -182,11 +163,8 @@ impl Daemon {
let _ = reply.send(Ok(()));
}
Err(e) => {
// `{e:#}` (not `{e}`/`to_string()`) so the full anyhow
// context chain reaches the caller/GUI instead of just
// the outermost ".context()" message.
bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &format!("{e:#}"));
let _ = reply.send(Err(format!("{e:#}")));
bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &e.to_string());
let _ = reply.send(Err(e.to_string()));
}
}
return;
@ -206,8 +184,8 @@ impl Daemon {
let _ = reply.send(Ok(()));
}
Err(e) => {
bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &format!("{e:#}"));
let _ = reply.send(Err(format!("{e:#}")));
bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &e.to_string());
let _ = reply.send(Err(e.to_string()));
}
}
return;

8
breadcastd/src/dlna_mirror.rs
View file

@ -81,13 +81,7 @@ impl DlnaMirrorSession {
.context("failed to start the HLS HTTP server")?;
let stream_url = http.url(lan_ip, "playlist.m3u8");
// 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))
wait_for_playlist_segments(&output_dir.join("playlist.m3u8"), 3, Duration::from_secs(20))
.await
.context("encode pipeline never produced playable HLS segments")?;