Compare commits
11 commits
9c6fc61f53
...
2cc1310752
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2cc1310752 | ||
|
|
bd511fea33 | ||
|
|
22a18eee1b | ||
|
|
a058482b39 | ||
|
|
1ee607b5e9 | ||
|
|
5e587ce342 | ||
|
|
b792743626 | ||
|
|
7fb1934d52 | ||
|
|
696e3f540f | ||
|
|
14274856a3 | ||
|
|
5b49955e33 |
14 changed files with 1076 additions and 95 deletions
4
.gitignore
vendored
4
.gitignore
vendored
|
|
@ -31,3 +31,7 @@ logs/
|
|||
*.pid
|
||||
|
||||
# Local hygiene notes (not for commit)
|
||||
CLAUDE.md
|
||||
|
||||
# graphify knowledge-graph output (local tool cache, not for commit)
|
||||
graphify-out/
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
#include "facade.h"
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <future>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
|
|
@ -44,6 +45,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<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
|
||||
|
|
@ -52,6 +81,41 @@ 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
|
||||
|
|
@ -62,8 +126,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 +256,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 +276,10 @@ 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);
|
||||
});
|
||||
}
|
||||
|
|
@ -225,15 +303,32 @@ 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 =
|
||||
|
|
@ -254,17 +349,74 @@ 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)
|
||||
// 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);
|
||||
// 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;
|
||||
}
|
||||
});
|
||||
|
||||
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;
|
||||
}
|
||||
|
|
@ -277,6 +429,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.
|
||||
|
|
|
|||
|
|
@ -103,6 +103,59 @@ 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);
|
||||
|
|
|
|||
|
|
@ -45,6 +45,29 @@ 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).
|
||||
|
|
@ -113,6 +136,16 @@ 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;
|
||||
|
|
|
|||
|
|
@ -39,6 +39,33 @@ 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;
|
||||
|
|
@ -47,7 +74,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 = openscreen::cast::kDefaultTargetPlayoutDelay;
|
||||
stream.target_delay = kTargetPlayoutDelay;
|
||||
stream.aes_key = GenerateRandomBytes16();
|
||||
stream.aes_iv_mask = GenerateRandomBytes16();
|
||||
stream.receiver_rtcp_event_log = true;
|
||||
|
|
|
|||
|
|
@ -62,3 +62,25 @@ 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.
|
||||
|
|
|
|||
|
|
@ -24,10 +24,18 @@ 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 (roughly two video frames at 30 FPS). See
|
||||
// crbug.com/498035450.
|
||||
// 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.
|
||||
constexpr Clock::duration kMinSenderInFlight =
|
||||
Clock::to_duration(milliseconds(66));
|
||||
Clock::to_duration(milliseconds(200));
|
||||
|
||||
} // namespace
|
||||
|
||||
|
|
|
|||
|
|
@ -326,10 +326,20 @@ fn run_io_loop(
|
|||
// doc comment for why that's fine for this project's usage.
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
tracing::debug!(error = %e, "cast session io loop ending: receive error");
|
||||
// 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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_on_message, breadcast_caststream_sender_take_stats,
|
||||
};
|
||||
|
||||
/// The Cast Streaming ("Mirroring") receiver app id, pre-installed on every
|
||||
|
|
@ -48,10 +48,44 @@ 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,
|
||||
max_bitrate_bps: 8_000_000,
|
||||
max_frame_rate_numerator: 30,
|
||||
// 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_frame_rate_denominator: 1,
|
||||
}
|
||||
}
|
||||
|
|
@ -217,6 +251,22 @@ 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 {
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ use std::sync::Arc;
|
|||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
|
||||
use breadcast_core::caststream::{CastStreamEvent, 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) = build_video_pipeline_for_streaming(capture.video_node_id())?;
|
||||
let (pipeline, appsink, encoder, video_params) = 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(),
|
||||
VideoParams::default(),
|
||||
video_params,
|
||||
)?;
|
||||
let sender = Arc::new(sender);
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
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;
|
||||
|
||||
|
|
@ -8,6 +10,8 @@ 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.
|
||||
|
|
@ -62,15 +66,37 @@ 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=60 rate-control=cbr ! \
|
||||
vah264enc bitrate=4000 key-int-max=30 rate-control=cbr ! \
|
||||
video/x-h264,profile=main ! \
|
||||
h264parse config-interval=1 ! \
|
||||
hlssink.video \
|
||||
hlssink3 name=hlssink target-duration=2 playlist-length=6 max-files=10"
|
||||
hlssink3 name=hlssink target-duration=1 playlist-length=3 max-files=6"
|
||||
);
|
||||
|
||||
let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
|
||||
|
|
@ -123,23 +149,243 @@ 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)> {
|
||||
) -> Result<(gst::Pipeline, gst_app::AppSink, gst::Element, VideoParams)> {
|
||||
gst::init().context("failed to initialize GStreamer")?;
|
||||
|
||||
// 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());
|
||||
// 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());
|
||||
|
||||
let element = gst::parse::launch(&pipeline_str).context("failed to parse GStreamer pipeline")?;
|
||||
let Ok(pipeline) = element.downcast::<gst::Pipeline>() else {
|
||||
|
|
@ -154,28 +400,95 @@ 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))
|
||||
Ok((pipeline, appsink, encoder, params))
|
||||
}
|
||||
|
||||
/// 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).
|
||||
/// 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.
|
||||
pub fn pull_encoded_frame(appsink: &gst_app::AppSink) -> Result<Option<(Vec<u8>, 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)))
|
||||
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)));
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends an upstream "force key unit" event from `appsink`, propagating to
|
||||
|
|
@ -310,3 +623,33 @@ 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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ use std::sync::Arc;
|
|||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use breadcast_core::caststream::{CastStreamEvent, VideoParams, WEBRTC_NAMESPACE};
|
||||
use breadcast_core::caststream::{CastStreamEvent, WEBRTC_NAMESPACE};
|
||||
use breadcast_core::pipeline::{
|
||||
build_video_pipeline_for_streaming, pull_encoded_frame, request_key_frame, set_video_bitrate_kbps,
|
||||
};
|
||||
|
|
@ -24,11 +24,53 @@ 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>,
|
||||
threads: Vec<std::thread::JoinHandle<()>>,
|
||||
/// 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<()>>,
|
||||
}
|
||||
|
||||
impl CastMirrorSession {
|
||||
|
|
@ -45,7 +87,13 @@ impl CastMirrorSession {
|
|||
let capture = CaptureSession::start().await.context("failed to start portal screen capture")?;
|
||||
let video_node_id = capture.video_node_id();
|
||||
|
||||
let (pipeline, appsink, encoder) =
|
||||
// `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) =
|
||||
build_video_pipeline_for_streaming(video_node_id).context("failed to build the encode pipeline")?;
|
||||
|
||||
{
|
||||
|
|
@ -74,13 +122,11 @@ impl CastMirrorSession {
|
|||
.context("failed to connect and launch the Mirroring receiver")?;
|
||||
|
||||
let (sender, stream_events) =
|
||||
CastStreamSender::start(&device.host, "sender-0", session.transport_id(), VideoParams::default())
|
||||
CastStreamSender::start(&device.host, "sender-0", session.transport_id(), video_params)
|
||||
.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 +135,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 +155,7 @@ impl CastMirrorSession {
|
|||
}
|
||||
}
|
||||
})
|
||||
});
|
||||
};
|
||||
|
||||
tracing::info!(device = %device.name, "sending Cast Streaming OFFER");
|
||||
sender.negotiate();
|
||||
|
|
@ -119,16 +165,22 @@ impl CastMirrorSession {
|
|||
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
||||
}
|
||||
if !negotiated.load(Ordering::Acquire) {
|
||||
let _ = session.stop();
|
||||
capture.close().await.ok();
|
||||
// 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;
|
||||
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");
|
||||
|
||||
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,63 +192,256 @@ 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<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.
|
||||
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");
|
||||
}
|
||||
if let Err(e) = self.session.stop() {
|
||||
tracing::warn!(error = ?e, "failed to cleanly stop the cast session");
|
||||
}
|
||||
stop_session_bounded(&self.session).await;
|
||||
if let Some(capture) = self.capture.take() {
|
||||
if let Err(e) = capture.close().await {
|
||||
tracing::warn!(error = ?e, "failed to cleanly close the portal capture session");
|
||||
}
|
||||
close_capture_bounded(capture).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.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;
|
||||
}
|
||||
}
|
||||
|
||||
/// `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 {
|
||||
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) {
|
||||
tracing::debug!(error = ?e, "dropped a frame (not negotiated yet or backpressure)");
|
||||
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 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);
|
||||
}
|
||||
// `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;
|
||||
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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -73,6 +73,12 @@ 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
|
||||
|
|
@ -128,7 +134,20 @@ impl Daemon {
|
|||
}
|
||||
}
|
||||
DaemonCommand::CastDeviceFound(device) => {
|
||||
self.cast_devices.insert(device.id.clone(), 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.broadcast_devices();
|
||||
}
|
||||
DaemonCommand::CastDeviceLost(id) => {
|
||||
|
|
@ -163,8 +182,11 @@ impl Daemon {
|
|||
let _ = reply.send(Ok(()));
|
||||
}
|
||||
Err(e) => {
|
||||
bread_events::emit_mirroring_failed(&self.bread_client, &device.id, &e.to_string());
|
||||
let _ = reply.send(Err(e.to_string()));
|
||||
// `{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:#}")));
|
||||
}
|
||||
}
|
||||
return;
|
||||
|
|
@ -184,8 +206,8 @@ impl Daemon {
|
|||
let _ = reply.send(Ok(()));
|
||||
}
|
||||
Err(e) => {
|
||||
bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &e.to_string());
|
||||
let _ = reply.send(Err(e.to_string()));
|
||||
bread_events::emit_mirroring_failed(&self.bread_client, &device.url, &format!("{e:#}"));
|
||||
let _ = reply.send(Err(format!("{e:#}")));
|
||||
}
|
||||
}
|
||||
return;
|
||||
|
|
|
|||
|
|
@ -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")?;
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue