feat(core,android): Automatic bitrate caps at the client decode limit, not the link ceiling
windows-host / package (push) Successful in 2m7s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 34s
ci / web (push) Successful in 53s
ci / docs-site (push) Successful in 1m3s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 4m31s
ci / rust (push) Failing after 6m39s
decky / build-publish (push) Successful in 30s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 9s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 8s
ci / bench (push) Successful in 5m17s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 5m0s
arch / build-publish (push) Successful in 16m33s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 9s
apple / swift (push) Successful in 5m13s
android / android (push) Successful in 17m48s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 6m16s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 10m5s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 9m0s
deb / build-publish (push) Successful in 12m2s
flatpak / build-publish (push) Failing after 8m6s
docker / deploy-docs (push) Successful in 23s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Successful in 18m18s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Successful in 18m22s
release / apple (push) Successful in 27m42s
apple / screenshots (push) Successful in 18m40s
windows-host / package (push) Successful in 2m7s
windows-msix / package (arm64, C:\Users\Public\ffmpeg-arm64, --no-default-features, aarch64-pc-windows-msvc, C:\t-a64) (push) Successful in 34s
ci / web (push) Successful in 53s
ci / docs-site (push) Successful in 1m3s
windows-msix / package (x64, C:\Users\Public\ffmpeg, , x86_64-pc-windows-msvc, C:\t) (push) Successful in 4m31s
ci / rust (push) Failing after 6m39s
decky / build-publish (push) Successful in 30s
docker / build-push (--build-arg FEDORA_VERSION=44, ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora44-rpm) (push) Successful in 9s
docker / build-push (., web/Dockerfile, punktfunk-web) (push) Successful in 8s
ci / bench (push) Successful in 5m17s
windows / build (aarch64-pc-windows-msvc) (push) Successful in 5m0s
arch / build-publish (push) Successful in 16m33s
docker / build-push (docs-site, docs-site/Dockerfile, punktfunk-docs) (push) Successful in 9s
apple / swift (push) Successful in 5m13s
android / android (push) Successful in 17m48s
windows / build (x86_64-pc-windows-msvc) (push) Successful in 6m16s
docker / build-push (ci, ci/fedora-rpm.Dockerfile, punktfunk-fedora-rpm) (push) Successful in 10m5s
docker / build-push (ci, ci/rust-ci.Dockerfile, punktfunk-rust-ci) (push) Successful in 9m0s
deb / build-publish (push) Successful in 12m2s
flatpak / build-publish (push) Failing after 8m6s
docker / deploy-docs (push) Successful in 23s
rpm / build-publish (44, fedora-44, punktfunk-fedora44-rpm) (push) Successful in 18m18s
rpm / build-publish (43, bazzite, punktfunk-fedora-rpm) (push) Successful in 18m22s
release / apple (push) Successful in 27m42s
apple / screenshots (push) Successful in 18m40s
The Automatic bitrate controller only reacted to network signals (loss, capture→received OWD, FEC-unrecoverable frames, jump-to-live flush), so on a fast LAN feeding a slower mobile HW decoder it slow-started straight to the link-probe ceiling and parked there — backlogging frames inside the decoder, where those signals never register, and choking it. Reported on a Snapdragon 8 Gen 1: Automatic pinned ~500 Mbps with unusable latency. Feed the client's decode-stage latency (received→decoded) into the controller as a first-class signal, symmetric with the existing OWD one: a rise over its rolling-min baseline ends the slow-start climb and, sustained over two windows, backs the rate ×0.7 down to the real decode limit — so Automatic settles where the decoder keeps up. - core/abr: on_window gains decode_mean_us; a decode_means rolling-min baseline + DECODE_RISE_US (15 ms) fold a decode rise into the bad-window logic. - core/client: per-frame report_decode_us accumulator, drained to a window mean by the data-plane pump; wants_decode_latency() gate (Automatic, non-PyroWave) lets embedders skip the measurement where it's ignored. Re-target log prints the driving signals. - android/decode: report the decode stage on both the sync and async decode paths, HUD-independent, measured from the AU leaving next_frame (so codec-input backpressure is included) and excluding the vsync present wait. Apple/Windows report_decode_us calls to follow. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -229,9 +229,11 @@ fn run_sync(
|
||||
// reclaimed after the codec is dropped below.
|
||||
let tracker = DisplayTracker::new(stats.clone(), clock_offset.clone());
|
||||
let render_cb = install_render_callback(&codec, &tracker);
|
||||
// HUD stage split: receipt timestamps keyed by the pts we queue into the codec, so the decoded
|
||||
// point (output-buffer dequeue — MediaCodec round-trips presentationTimeUs) can be paired back
|
||||
// to its receipt for the `decode` stage. Only fed while the HUD is visible.
|
||||
// Receipt timestamps keyed by the pts we queue into the codec, so the decoded point (output-
|
||||
// buffer dequeue — MediaCodec round-trips presentationTimeUs) can be paired back to its receipt
|
||||
// for the `decode` stage. Fed while the HUD is visible OR the adaptive-bitrate controller wants
|
||||
// the decode signal (`measure_decode`) — the decoder-backlog bottleneck the network can't see.
|
||||
let measure_decode = client.wants_decode_latency();
|
||||
let mut in_flight: VecDeque<(u64, i128)> = VecDeque::new();
|
||||
// Phase-2 host/network split (design/stats-unification.md): received AUs awaiting their 0xCF
|
||||
// host timing, as (pts_ns, capture→received µs). The timings are drained non-blockingly right
|
||||
@@ -272,40 +274,45 @@ fn run_sync(
|
||||
&p[..p.len().min(6)]
|
||||
);
|
||||
}
|
||||
// HUD stat, `received` point: host+network = client_now + (host−client) −
|
||||
// capture_pts. Gated on the HUD being visible — `enabled` first so the hidden
|
||||
// steady state skips the wall-clock read and the lock entirely. The receipt
|
||||
// stamp is also parked in `in_flight` (keyed by the pts the codec will echo on
|
||||
// the output buffer) for the decoded-point pairing in `drain`.
|
||||
if stats.enabled() {
|
||||
// Receipt stamp for the `decode` stage pairing, parked in `in_flight` (keyed by
|
||||
// the pts the codec echoes on its output buffer) whenever it's needed: the HUD
|
||||
// being visible, or the ABR decode signal (`measure_decode`). The HUD-only
|
||||
// samplers (`received` point, host/network split) stay gated on the overlay so
|
||||
// the hidden steady state adds only a wall-clock read + the receipt push.
|
||||
if stats.enabled() || measure_decode {
|
||||
let received_ns = now_realtime_ns();
|
||||
let clock_offset = clock_offset.load(Ordering::Relaxed);
|
||||
let lat_ns = received_ns + clock_offset as i128 - frame.pts_ns as i128;
|
||||
let lat_us = (lat_ns > 0 && lat_ns < 10_000_000_000)
|
||||
.then_some((lat_ns / 1000) as u64);
|
||||
stats.note_received(frame.data.len(), lat_us, clock_offset != 0);
|
||||
in_flight.push_back((frame.pts_ns / 1000, received_ns));
|
||||
if in_flight.len() > IN_FLIGHT_CAP {
|
||||
in_flight.pop_front(); // stale — codec never echoed it back
|
||||
}
|
||||
// Phase-2 split: park this AU's capture→received sample, then match any
|
||||
// 0xCF host timings that have arrived — host = the host's own
|
||||
// capture→sent, network = our capture→received minus it (per-frame
|
||||
// tiling; saturating in case of clock jitter).
|
||||
if let Some(hostnet_us) = lat_us {
|
||||
pending_split.push_back((frame.pts_ns, hostnet_us));
|
||||
if pending_split.len() > PENDING_SPLIT_CAP {
|
||||
pending_split.pop_front(); // 0xCF lost / old host — evict
|
||||
// HUD stat, `received` point: host+network = client_now + (host−client) −
|
||||
// capture_pts.
|
||||
if stats.enabled() {
|
||||
let clock_offset = clock_offset.load(Ordering::Relaxed);
|
||||
let lat_ns = received_ns + clock_offset as i128 - frame.pts_ns as i128;
|
||||
let lat_us = (lat_ns > 0 && lat_ns < 10_000_000_000)
|
||||
.then_some((lat_ns / 1000) as u64);
|
||||
stats.note_received(frame.data.len(), lat_us, clock_offset != 0);
|
||||
// Phase-2 split: park this AU's capture→received sample, then match any
|
||||
// 0xCF host timings that have arrived — host = the host's own
|
||||
// capture→sent, network = our capture→received minus it (per-frame
|
||||
// tiling; saturating in case of clock jitter).
|
||||
if let Some(hostnet_us) = lat_us {
|
||||
pending_split.push_back((frame.pts_ns, hostnet_us));
|
||||
if pending_split.len() > PENDING_SPLIT_CAP {
|
||||
pending_split.pop_front(); // 0xCF lost / old host — evict
|
||||
}
|
||||
}
|
||||
}
|
||||
while let Ok(t) = client.next_host_timing(Duration::ZERO) {
|
||||
if let Some(i) = pending_split.iter().position(|&(p, _)| p == t.pts_ns)
|
||||
{
|
||||
let (_, hostnet_us) = pending_split.remove(i).unwrap();
|
||||
stats.note_host_split(
|
||||
t.host_us as u64,
|
||||
hostnet_us.saturating_sub(t.host_us as u64),
|
||||
);
|
||||
while let Ok(t) = client.next_host_timing(Duration::ZERO) {
|
||||
if let Some(i) =
|
||||
pending_split.iter().position(|&(p, _)| p == t.pts_ns)
|
||||
{
|
||||
let (_, hostnet_us) = pending_split.remove(i).unwrap();
|
||||
stats.note_host_split(
|
||||
t.host_us as u64,
|
||||
hostnet_us.saturating_sub(t.host_us as u64),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -345,6 +352,8 @@ fn run_sync(
|
||||
};
|
||||
let (r, d) = drain(
|
||||
&codec,
|
||||
&client,
|
||||
measure_decode,
|
||||
&window,
|
||||
&mut applied_ds,
|
||||
wait,
|
||||
@@ -866,6 +875,9 @@ fn run_async(
|
||||
// output back to them. Behind a `Mutex` since two threads touch it — only ever locked while the
|
||||
// HUD is visible.
|
||||
let clock_offset = client.clock_offset_shared();
|
||||
// Whether the adaptive-bitrate controller wants the `decode` stage as its decoder-backlog
|
||||
// signal (Automatic, non-PyroWave): then `in_flight` is fed regardless of the HUD.
|
||||
let measure_decode = client.wants_decode_latency();
|
||||
let in_flight = Arc::new(Mutex::new(VecDeque::<(u64, i128)>::new()));
|
||||
// Display stage (spec `display` + the capture→displayed headline): the rendered frame is
|
||||
// parked in the tracker at release; the OnFrameRendered callback pairs it with
|
||||
@@ -886,7 +898,15 @@ fn run_async(
|
||||
std::thread::Builder::new()
|
||||
.name("pf-decode-feed".into())
|
||||
.spawn(move || {
|
||||
feeder_loop(client, stats, in_flight, clock_offset, shutdown, ev_tx);
|
||||
feeder_loop(
|
||||
client,
|
||||
stats,
|
||||
measure_decode,
|
||||
in_flight,
|
||||
clock_offset,
|
||||
shutdown,
|
||||
ev_tx,
|
||||
);
|
||||
})
|
||||
.ok()
|
||||
};
|
||||
@@ -976,6 +996,8 @@ fn run_async(
|
||||
let had_output = !ready.is_empty();
|
||||
present_ready(
|
||||
&codec,
|
||||
&client,
|
||||
measure_decode,
|
||||
&mut ready,
|
||||
&stats,
|
||||
&in_flight,
|
||||
@@ -1052,6 +1074,7 @@ fn run_async(
|
||||
fn feeder_loop(
|
||||
client: Arc<NativeClient>,
|
||||
stats: Arc<crate::stats::VideoStats>,
|
||||
measure_decode: bool,
|
||||
in_flight: Arc<Mutex<VecDeque<(u64, i128)>>>,
|
||||
clock_offset: Arc<AtomicI64>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
@@ -1067,13 +1090,11 @@ fn feeder_loop(
|
||||
// instead of a full IDR (the frames_dropped keyframe path is the backstop). The gap
|
||||
// verdict rides the Au event so the decode loop arms its freeze gate on the same signal.
|
||||
let gap = client.note_frame_index(frame.frame_index);
|
||||
if stats.enabled() {
|
||||
// Park the receipt stamp (keyed by the pts the codec echoes) whenever the `decode`
|
||||
// stage is consumed: the HUD, or the ABR decode signal (`measure_decode`). The
|
||||
// HUD-only `received` point + host/network split stay gated on the overlay.
|
||||
if stats.enabled() || measure_decode {
|
||||
let received_ns = now_realtime_ns();
|
||||
let clock_offset = clock_offset.load(Ordering::Relaxed) as i128;
|
||||
let lat_ns = received_ns + clock_offset - frame.pts_ns as i128;
|
||||
let lat_us =
|
||||
(lat_ns > 0 && lat_ns < 10_000_000_000).then_some((lat_ns / 1000) as u64);
|
||||
stats.note_received(frame.data.len(), lat_us, clock_offset != 0);
|
||||
{
|
||||
let mut g = in_flight
|
||||
.lock()
|
||||
@@ -1083,19 +1104,27 @@ fn feeder_loop(
|
||||
g.pop_front(); // stale — codec never echoed it back
|
||||
}
|
||||
}
|
||||
if let Some(hostnet_us) = lat_us {
|
||||
pending_split.push_back((frame.pts_ns, hostnet_us));
|
||||
if pending_split.len() > PENDING_SPLIT_CAP {
|
||||
pending_split.pop_front();
|
||||
if stats.enabled() {
|
||||
let clock_offset = clock_offset.load(Ordering::Relaxed) as i128;
|
||||
let lat_ns = received_ns + clock_offset - frame.pts_ns as i128;
|
||||
let lat_us = (lat_ns > 0 && lat_ns < 10_000_000_000)
|
||||
.then_some((lat_ns / 1000) as u64);
|
||||
stats.note_received(frame.data.len(), lat_us, clock_offset != 0);
|
||||
if let Some(hostnet_us) = lat_us {
|
||||
pending_split.push_back((frame.pts_ns, hostnet_us));
|
||||
if pending_split.len() > PENDING_SPLIT_CAP {
|
||||
pending_split.pop_front();
|
||||
}
|
||||
}
|
||||
}
|
||||
while let Ok(t) = client.next_host_timing(Duration::ZERO) {
|
||||
if let Some(i) = pending_split.iter().position(|&(p, _)| p == t.pts_ns) {
|
||||
let (_, hostnet_us) = pending_split.remove(i).unwrap();
|
||||
stats.note_host_split(
|
||||
t.host_us as u64,
|
||||
hostnet_us.saturating_sub(t.host_us as u64),
|
||||
);
|
||||
while let Ok(t) = client.next_host_timing(Duration::ZERO) {
|
||||
if let Some(i) = pending_split.iter().position(|&(p, _)| p == t.pts_ns)
|
||||
{
|
||||
let (_, hostnet_us) = pending_split.remove(i).unwrap();
|
||||
stats.note_host_split(
|
||||
t.host_us as u64,
|
||||
hostnet_us.saturating_sub(t.host_us as u64),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1221,6 +1250,8 @@ fn feed_ready(
|
||||
#[allow(clippy::too_many_arguments)] // one call site; mirrors the sync loop's drain
|
||||
fn present_ready(
|
||||
codec: &MediaCodec,
|
||||
client: &NativeClient,
|
||||
measure_decode: bool,
|
||||
ready: &mut Vec<OutputReady>,
|
||||
stats: &crate::stats::VideoStats,
|
||||
in_flight: &Mutex<VecDeque<(u64, i128)>>,
|
||||
@@ -1234,12 +1265,22 @@ fn present_ready(
|
||||
if ready.is_empty() {
|
||||
return;
|
||||
}
|
||||
if stats.enabled() {
|
||||
// Pair each output's decode stage (feeds the ABR decode signal always; the HUD histogram only
|
||||
// while visible) — both consume the receipt map, so enter for either.
|
||||
if stats.enabled() || measure_decode {
|
||||
let mut g = in_flight
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
for o in ready.iter() {
|
||||
note_decoded_pts(stats, &mut g, clock_offset, o.pts_us, o.decoded_ns);
|
||||
note_decoded_pts(
|
||||
client,
|
||||
measure_decode,
|
||||
stats,
|
||||
&mut g,
|
||||
clock_offset,
|
||||
o.pts_us,
|
||||
o.decoded_ns,
|
||||
);
|
||||
}
|
||||
}
|
||||
// Fold EVERY output through the gate in pts (== decode) order — even the ones newest-wins discards —
|
||||
@@ -1460,6 +1501,8 @@ fn feed(
|
||||
#[allow(clippy::too_many_arguments)] // one call site; mirrors the async loop's present_ready
|
||||
fn drain(
|
||||
codec: &MediaCodec,
|
||||
client: &NativeClient,
|
||||
measure_decode: bool,
|
||||
window: &NativeWindow,
|
||||
applied_ds: &mut Option<DataSpace>,
|
||||
first_wait: Duration,
|
||||
@@ -1489,11 +1532,20 @@ fn drain(
|
||||
let flags = take_flags(recovery_flags, pts_us);
|
||||
held_present =
|
||||
gate.on_decoded(flags, false, Instant::now()) == GateVerdict::Present;
|
||||
let meta = if stats.enabled() {
|
||||
let meta = if stats.enabled() || measure_decode {
|
||||
// The dequeue IS the sync loop's decoded-availability instant.
|
||||
let decoded_ns = now_realtime_ns();
|
||||
note_decoded_pts(stats, in_flight, clock_offset, pts_us, decoded_ns);
|
||||
Some((pts_us, decoded_ns))
|
||||
note_decoded_pts(
|
||||
client,
|
||||
measure_decode,
|
||||
stats,
|
||||
in_flight,
|
||||
clock_offset,
|
||||
pts_us,
|
||||
decoded_ns,
|
||||
);
|
||||
// The tracker's `display` stage is a HUD concern — park only when visible.
|
||||
stats.enabled().then_some((pts_us, decoded_ns))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
@@ -1564,6 +1616,8 @@ fn drain(
|
||||
/// `decoded_ns` is the availability instant: the dequeue (sync loop) or the output callback's
|
||||
/// stamp (async loop).
|
||||
fn note_decoded_pts(
|
||||
client: &NativeClient,
|
||||
measure_decode: bool,
|
||||
stats: &crate::stats::VideoStats,
|
||||
in_flight: &mut VecDeque<(u64, i128)>,
|
||||
clock_offset: i64,
|
||||
@@ -1582,12 +1636,25 @@ fn note_decoded_pts(
|
||||
break;
|
||||
}
|
||||
}
|
||||
// pts_us is the truncated frame.pts_ns/1000 we queued, so ×1000 re-approximates capture time
|
||||
// to < 1 µs — negligible against the ms-scale figures shown.
|
||||
let e2e_ns = decoded_ns + clock_offset as i128 - pts_us as i128 * 1000;
|
||||
let e2e_us = (e2e_ns > 0 && e2e_ns < 10_000_000_000).then_some((e2e_ns / 1000) as u64);
|
||||
let decode_us = received_ns.map(|r| ((decoded_ns - r).max(0) / 1000) as u64);
|
||||
stats.note_decoded(e2e_us, decode_us);
|
||||
// Adaptive bitrate: the `decode` stage (received→decoded, single-clock local) IS the decoder-
|
||||
// backlog signal — the only bottleneck the host-side network signals can't see (a fast LAN
|
||||
// feeding a slower mobile decoder). Report it whenever the controller is armed, regardless of
|
||||
// the HUD; `report_decode_us` is a cheap accumulate the pump windows.
|
||||
if measure_decode {
|
||||
if let Some(us) = decode_us {
|
||||
client.report_decode_us(us.min(u32::MAX as u64) as u32);
|
||||
}
|
||||
}
|
||||
// HUD histogram: only while the overlay is visible (a measure-only caller enters here for the
|
||||
// ABR report alone). `end-to-end` = capture→decoded (skew-corrected) tiles the `decode` stage.
|
||||
// pts_us is the truncated frame.pts_ns/1000 we queued, so ×1000 re-approximates capture time to
|
||||
// < 1 µs — negligible against the ms-scale figures shown.
|
||||
if stats.enabled() {
|
||||
let e2e_ns = decoded_ns + clock_offset as i128 - pts_us as i128 * 1000;
|
||||
let e2e_us = (e2e_ns > 0 && e2e_ns < 10_000_000_000).then_some((e2e_ns / 1000) as u64);
|
||||
stats.note_decoded(e2e_us, decode_us);
|
||||
}
|
||||
}
|
||||
|
||||
/// The AU `user_flags` for a decoded output, keyed by the echoed `presentationTimeUs`. Recovery
|
||||
|
||||
@@ -53,6 +53,15 @@ const HEAVY_LOSS_PPM: u32 = 20_000;
|
||||
/// How far the window's mean one-way delay may sit above the rolling baseline before it counts
|
||||
/// as queue growth. 25 ms is far beyond jitter at any streamable frame rate.
|
||||
const OWD_RISE_US: i64 = 25_000;
|
||||
/// How far the window's mean *decode-stage* latency (client hand-off → decoder output, reported by
|
||||
/// the embedder) may sit above its rolling baseline before it counts as the decoder falling behind.
|
||||
/// This is the signal the network-side ones can't see: on a fast LAN a mobile HW decoder saturates
|
||||
/// long before the link does, backlogging frames INSIDE the decoder where loss/OWD never register —
|
||||
/// so without this the controller slow-starts straight to the link ceiling and parks there, choking
|
||||
/// the decoder. A rising decode latency ends the climb and (sustained) backs the rate off to the
|
||||
/// real decode limit. Local, low-noise signal (no network jitter), so a tighter threshold than OWD:
|
||||
/// 15 ms of standing decode queue is unambiguous backlog at any streamable frame rate.
|
||||
const DECODE_RISE_US: i64 = 15_000;
|
||||
/// Rolling window (in 750 ms report windows, ~30 s) whose minimum mean is the OWD baseline.
|
||||
/// Long enough to remember the uncongested floor, short enough to follow genuine path changes.
|
||||
const BASELINE_WINDOWS: usize = 40;
|
||||
@@ -76,6 +85,10 @@ pub(crate) struct BitrateController {
|
||||
probing: bool,
|
||||
/// Recent window mean OWDs (µs); the rolling min is the uncongested baseline.
|
||||
owd_means: VecDeque<i64>,
|
||||
/// Recent window mean decode-stage latencies (µs); the rolling min is the decoder's
|
||||
/// keeping-up baseline. Empty on embedders that don't report decode latency (the decode
|
||||
/// signal is then simply absent — identical to the pre-decode-signal behavior).
|
||||
decode_means: VecDeque<i64>,
|
||||
bad_windows: u32,
|
||||
clean_windows: u32,
|
||||
last_change: Option<Instant>,
|
||||
@@ -95,6 +108,7 @@ impl BitrateController {
|
||||
floor_kbps: FLOOR_KBPS.min(start_kbps.max(1)),
|
||||
probing: true,
|
||||
owd_means: VecDeque::with_capacity(BASELINE_WINDOWS),
|
||||
decode_means: VecDeque::with_capacity(BASELINE_WINDOWS),
|
||||
bad_windows: 0,
|
||||
clean_windows: 0,
|
||||
last_change: None,
|
||||
@@ -125,13 +139,16 @@ impl BitrateController {
|
||||
/// Feed one report window; returns the rate to request now, if any. `dropped` = frames that
|
||||
/// went FEC-unrecoverable in the window, `loss_ppm` the window's [`crate::quic::LossReport`]
|
||||
/// figure, `owd_mean_us` the window's mean skew-corrected capture→received latency (`None`
|
||||
/// without a clock handshake), `flushed` = the pump's jump-to-live fired in the window.
|
||||
/// without a clock handshake), `decode_mean_us` the window's mean client decode-stage latency
|
||||
/// (`None` on an embedder that doesn't report it — the signal is then absent), `flushed` = the
|
||||
/// pump's jump-to-live fired in the window.
|
||||
pub(crate) fn on_window(
|
||||
&mut self,
|
||||
now: Instant,
|
||||
dropped: u64,
|
||||
loss_ppm: u32,
|
||||
owd_mean_us: Option<i64>,
|
||||
decode_mean_us: Option<i64>,
|
||||
flushed: bool,
|
||||
) -> Option<u32> {
|
||||
if !self.enabled {
|
||||
@@ -161,11 +178,31 @@ impl BitrateController {
|
||||
}
|
||||
None => false,
|
||||
};
|
||||
// Decode-stage latency: same rolling-min-baseline treatment as OWD, but measuring the
|
||||
// CLIENT'S decoder rather than the link. A rise means the decoder is backlogging frames —
|
||||
// the bottleneck the network signals are blind to. Marking the window bad both ends slow
|
||||
// start (so the climb stops the moment decode latency lifts, instead of doubling on into
|
||||
// the link ceiling) and, sustained, drives the ×0.7 backoff down to the real decode limit.
|
||||
let decode_bad = match decode_mean_us {
|
||||
Some(mean) => {
|
||||
let bad = self
|
||||
.decode_means
|
||||
.iter()
|
||||
.min()
|
||||
.is_some_and(|&base| mean > base + DECODE_RISE_US);
|
||||
if self.decode_means.len() == BASELINE_WINDOWS {
|
||||
self.decode_means.pop_front();
|
||||
}
|
||||
self.decode_means.push_back(mean);
|
||||
bad
|
||||
}
|
||||
None => false,
|
||||
};
|
||||
// SEVERE = the user already saw damage (an unrecoverable frame, a jump-to-live flush) or
|
||||
// loss far past any blip — one window is enough. Ordinary congestion (heavy-but-
|
||||
// recoverable loss, an OWD rise) still needs two consecutive windows.
|
||||
// recoverable loss, an OWD rise, a decode-latency rise) still needs two consecutive windows.
|
||||
let severe = dropped > 0 || flushed || loss_ppm >= SEVERE_LOSS_PPM;
|
||||
let bad = severe || loss_ppm >= HEAVY_LOSS_PPM || owd_bad;
|
||||
let bad = severe || loss_ppm >= HEAVY_LOSS_PPM || owd_bad || decode_bad;
|
||||
if bad {
|
||||
self.bad_windows += 1;
|
||||
self.clean_windows = 0;
|
||||
@@ -231,7 +268,7 @@ mod tests {
|
||||
fn run_clean(c: &mut BitrateController, start: Instant, from: u32, n: u32) -> Option<u32> {
|
||||
let mut out = None;
|
||||
for i in from..from + n {
|
||||
out = c.on_window(ticks(start, i), 0, 0, Some(10_000), false);
|
||||
out = c.on_window(ticks(start, i), 0, 0, Some(10_000), None, false);
|
||||
if out.is_some() {
|
||||
return out;
|
||||
}
|
||||
@@ -244,7 +281,10 @@ mod tests {
|
||||
// start 0 = explicit user bitrate or a host that didn't echo one → permanently off.
|
||||
let mut c = BitrateController::new(0);
|
||||
let now = Instant::now();
|
||||
assert_eq!(c.on_window(now, 5, 900_000, Some(500_000), true), None);
|
||||
assert_eq!(
|
||||
c.on_window(now, 5, 900_000, Some(500_000), None, true),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -252,17 +292,23 @@ mod tests {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
let start = Instant::now();
|
||||
// Heavy-but-recoverable loss (2–6 %) is ORDINARY: one window is a blip — no reaction.
|
||||
assert_eq!(c.on_window(ticks(start, 0), 0, 25_000, None, false), None);
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 25_000, None, None, false),
|
||||
None
|
||||
);
|
||||
// The second consecutive bad window backs off ×0.7.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 1), 0, 25_000, None, false),
|
||||
c.on_window(ticks(start, 1), 0, 25_000, None, None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
c.on_ack(14_000);
|
||||
// Still bad after the cooldown → another ×0.7 step from the ACKED rate.
|
||||
assert_eq!(c.on_window(ticks(start, 6), 0, 25_000, None, false), None); // bad #1 again
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 7), 0, 25_000, None, false),
|
||||
c.on_window(ticks(start, 6), 0, 25_000, None, None, false),
|
||||
None
|
||||
); // bad #1 again
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 7), 0, 25_000, None, None, false),
|
||||
Some(9_800)
|
||||
);
|
||||
}
|
||||
@@ -273,16 +319,19 @@ mod tests {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
let start = Instant::now();
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 1, 0, None, false),
|
||||
c.on_window(ticks(start, 0), 1, 0, None, None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
// …and so does a jump-to-live flush.
|
||||
let mut c = BitrateController::new(20_000);
|
||||
assert_eq!(c.on_window(ticks(start, 0), 0, 0, None, true), Some(14_000));
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 0, None, None, true),
|
||||
Some(14_000)
|
||||
);
|
||||
// …and ≥6 % window loss.
|
||||
let mut c = BitrateController::new(20_000);
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 80_000, None, false),
|
||||
c.on_window(ticks(start, 0), 0, 80_000, None, None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
}
|
||||
@@ -292,14 +341,17 @@ mod tests {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
let start = Instant::now();
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 1, 0, None, false),
|
||||
c.on_window(ticks(start, 0), 1, 0, None, None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
c.on_ack(14_000);
|
||||
// A severe window INSIDE the 1.5 s cooldown (tick 1 = 750 ms) → held; at the cooldown
|
||||
// boundary (tick 2 = 1.5 s) it fires.
|
||||
assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, false), None);
|
||||
assert_eq!(c.on_window(ticks(start, 2), 1, 0, None, false), Some(9_800));
|
||||
assert_eq!(c.on_window(ticks(start, 1), 1, 0, None, None, false), None);
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 2), 1, 0, None, None, false),
|
||||
Some(9_800)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -307,11 +359,14 @@ mod tests {
|
||||
let mut c = BitrateController::new(6_000);
|
||||
let start = Instant::now();
|
||||
// ×0.7 of 6000 = 4200 < floor → clamped to 5000.
|
||||
assert_eq!(c.on_window(ticks(start, 0), 1, 0, None, false), Some(5_000));
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 1, 0, None, None, false),
|
||||
Some(5_000)
|
||||
);
|
||||
c.on_ack(5_000);
|
||||
// At the floor, further bad windows request nothing.
|
||||
assert_eq!(c.on_window(ticks(start, 6), 1, 0, None, false), None);
|
||||
assert_eq!(c.on_window(ticks(start, 7), 1, 0, None, false), None);
|
||||
assert_eq!(c.on_window(ticks(start, 6), 1, 0, None, None, false), None);
|
||||
assert_eq!(c.on_window(ticks(start, 7), 1, 0, None, None, false), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -319,7 +374,7 @@ mod tests {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
let start = Instant::now();
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 1, 0, None, false),
|
||||
c.on_window(ticks(start, 0), 1, 0, None, None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
c.on_ack(14_000);
|
||||
@@ -342,7 +397,7 @@ mod tests {
|
||||
// Every cooled clean window doubles until the ceiling caps the climb, then quiet.
|
||||
let mut got = Vec::new();
|
||||
for i in 0..14 {
|
||||
if let Some(k) = c.on_window(ticks(start, i), 0, 0, Some(10_000), false) {
|
||||
if let Some(k) = c.on_window(ticks(start, i), 0, 0, Some(10_000), None, false) {
|
||||
c.on_ack(k);
|
||||
got.push(k);
|
||||
}
|
||||
@@ -356,20 +411,20 @@ mod tests {
|
||||
c.set_ceiling(300_000);
|
||||
let start = Instant::now();
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 0, Some(10_000), false),
|
||||
c.on_window(ticks(start, 0), 0, 0, Some(10_000), None, false),
|
||||
Some(40_000)
|
||||
);
|
||||
c.on_ack(40_000);
|
||||
// Severe window → immediate ×0.7, and slow start is over.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 2), 1, 0, Some(10_000), false),
|
||||
c.on_window(ticks(start, 2), 1, 0, Some(10_000), None, false),
|
||||
Some(28_000)
|
||||
);
|
||||
c.on_ack(28_000);
|
||||
// Clean again — but the next climb is additive, after the 6-window clean run.
|
||||
let mut next = None;
|
||||
for i in 3..12 {
|
||||
next = c.on_window(ticks(start, i), 0, 0, Some(10_000), false);
|
||||
next = c.on_window(ticks(start, i), 0, 0, Some(10_000), None, false);
|
||||
if next.is_some() {
|
||||
assert!(i >= 8, "additive climb must wait for the clean run");
|
||||
break;
|
||||
@@ -382,7 +437,7 @@ mod tests {
|
||||
fn set_ceiling_is_ignored_when_disabled_and_never_lowers() {
|
||||
let mut c = BitrateController::new(0);
|
||||
c.set_ceiling(1_000_000);
|
||||
assert_eq!(c.on_window(Instant::now(), 0, 0, None, false), None);
|
||||
assert_eq!(c.on_window(Instant::now(), 0, 0, None, None, false), None);
|
||||
let mut c = BitrateController::new(20_000);
|
||||
c.set_ceiling(10_000); // below the negotiated start → ignored
|
||||
assert_eq!(c.ceiling_kbps, 20_000);
|
||||
@@ -395,21 +450,72 @@ mod tests {
|
||||
// Establish a ~10 ms baseline over a few clean windows.
|
||||
for i in 0..4 {
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, i), 0, 0, Some(10_000), false),
|
||||
c.on_window(ticks(start, i), 0, 0, Some(10_000), None, false),
|
||||
None
|
||||
);
|
||||
}
|
||||
// Delay climbs 40 ms above baseline with ZERO loss — bufferbloat. Two windows → back off.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 4), 0, 0, Some(50_000), false),
|
||||
c.on_window(ticks(start, 4), 0, 0, Some(50_000), None, false),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 5), 0, 0, Some(52_000), false),
|
||||
c.on_window(ticks(start, 5), 0, 0, Some(52_000), None, false),
|
||||
Some(14_000)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_latency_rise_alone_is_a_congestion_signal() {
|
||||
// The link is pristine (zero loss, flat OWD) but the client's decoder is falling behind —
|
||||
// the LAN-vs-mobile-decoder case. Only the decode signal can catch it.
|
||||
let mut c = BitrateController::new(20_000);
|
||||
let start = Instant::now();
|
||||
// A ~8 ms decode baseline over a few clean windows.
|
||||
for i in 0..4 {
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, i), 0, 0, Some(10_000), Some(8_000), false),
|
||||
None
|
||||
);
|
||||
}
|
||||
// Decode latency climbs 30 ms above baseline with ZERO loss and flat OWD: the decoder is
|
||||
// backlogging. Two windows → back off ×0.7, exactly like an OWD rise.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 4), 0, 0, Some(10_000), Some(38_000), false),
|
||||
None
|
||||
);
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 5), 0, 0, Some(10_000), Some(40_000), false),
|
||||
Some(14_000)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_latency_caps_the_slow_start_climb() {
|
||||
// A fat link (probe measured ~300 Mbps) but a decoder that saturates around the start rate.
|
||||
let mut c = BitrateController::new(20_000);
|
||||
c.set_ceiling(300_000);
|
||||
let start = Instant::now();
|
||||
// First clean window (decoder fine at 20 Mbps) → slow start doubles to 40.
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 0), 0, 0, Some(10_000), Some(8_000), false),
|
||||
Some(40_000)
|
||||
);
|
||||
c.on_ack(40_000);
|
||||
// At 40 Mbps the decoder starts backing up (30 ms over baseline): the window is bad, so the
|
||||
// climb stops here instead of doubling on toward the 300 Mbps link ceiling…
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 2), 0, 0, Some(10_000), Some(38_000), false),
|
||||
None
|
||||
);
|
||||
// …and a second backed-up window backs the rate off, settling at the decode limit rather
|
||||
// than choking the decoder at the link ceiling (the reported bug).
|
||||
assert_eq!(
|
||||
c.on_window(ticks(start, 4), 0, 0, Some(10_000), Some(40_000), false),
|
||||
Some(28_000)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ack_silence_disables_the_controller() {
|
||||
let mut c = BitrateController::new(20_000);
|
||||
@@ -418,7 +524,9 @@ mod tests {
|
||||
let mut i = 0;
|
||||
// Keep every window bad and never ack: exactly MAX_UNACKED requests, then silence.
|
||||
while i < 60 {
|
||||
if c.on_window(ticks(start, i), 1, 0, None, false).is_some() {
|
||||
if c.on_window(ticks(start, i), 1, 0, None, None, false)
|
||||
.is_some()
|
||||
{
|
||||
sent += 1;
|
||||
}
|
||||
i += 1;
|
||||
|
||||
@@ -234,6 +234,21 @@ const MIC_QUEUE: usize = 64;
|
||||
/// the control task is wedged, which callers treat as a closed session.
|
||||
const CTRL_QUEUE: usize = 32;
|
||||
|
||||
/// Client decode-stage latency accumulator for the adaptive-bitrate controller's decode signal.
|
||||
/// The embedder adds one sample per decoded frame ([`NativeClient::report_decode_us`], µs from the
|
||||
/// AU leaving [`NativeClient::next_frame`] to its decoded output) and the data-plane pump drains a
|
||||
/// window mean once per report window to feed [`crate::abr::BitrateController::on_window`]. This is
|
||||
/// the only signal that sees the CLIENT'S decoder: on a fast LAN a mobile HW decoder saturates long
|
||||
/// before the link, backlogging frames inside the decoder where loss/OWD never register. Sum+count
|
||||
/// (not a running mean) so the pump takes an unweighted window mean and resets. Always accumulated —
|
||||
/// the controller ignores it when Automatic is off, and the pump drains it every window regardless,
|
||||
/// so it stays bounded (a full window at 240 fps is ~180 samples).
|
||||
#[derive(Default)]
|
||||
struct DecodeLatAcc {
|
||||
sum_us: u64,
|
||||
count: u32,
|
||||
}
|
||||
|
||||
/// The pre-decode video hand-off from the data-plane pump to the embedder. Unlike the side planes
|
||||
/// (self-contained samples that drop the newest on overflow), video AUs are reference-chained under the
|
||||
/// host's infinite GOP: dropping ANY frame mid-stream corrupts every dependent frame until the next
|
||||
@@ -497,6 +512,14 @@ pub struct NativeClient {
|
||||
/// the pump's first no-op clock flush). Shared with the pump and, via
|
||||
/// [`clock_offset_shared`](Self::clock_offset_shared), with embedder latency-math threads.
|
||||
clock_offset: Arc<AtomicI64>,
|
||||
/// Decode-stage latency samples from the embedder ([`report_decode_us`](Self::report_decode_us)),
|
||||
/// drained per window by the data-plane pump to feed the adaptive-bitrate controller's decode
|
||||
/// signal. Shared with the pump; see [`DecodeLatAcc`].
|
||||
decode_lat: Arc<Mutex<DecodeLatAcc>>,
|
||||
/// Whether the adaptive-bitrate controller is armed for this session (Automatic bitrate and not
|
||||
/// a rate-pinned PyroWave stream) — exposed via [`wants_decode_latency`](Self::wants_decode_latency)
|
||||
/// so an embedder skips the per-frame decode measurement when the controller wouldn't use it.
|
||||
wants_decode: bool,
|
||||
worker: Option<std::thread::JoinHandle<()>>,
|
||||
/// The currently active session mode (the Welcome's, then updated by every accepted
|
||||
/// [`NativeClient::request_mode`]).
|
||||
@@ -680,6 +703,7 @@ impl NativeClient {
|
||||
let fec_recovered = Arc::new(AtomicU64::new(0));
|
||||
let hot_tids = Arc::new(Mutex::new(Vec::new()));
|
||||
let clock_offset = Arc::new(AtomicI64::new(0));
|
||||
let decode_lat = Arc::new(Mutex::new(DecodeLatAcc::default()));
|
||||
|
||||
let host = host.to_string();
|
||||
let frame_chan_w = frame_chan.clone();
|
||||
@@ -691,6 +715,7 @@ impl NativeClient {
|
||||
let fec_recovered_w = fec_recovered.clone();
|
||||
let hot_tids_w = hot_tids.clone();
|
||||
let clock_offset_w = clock_offset.clone();
|
||||
let decode_lat_w = decode_lat.clone();
|
||||
let ctrl_tx_pump = ctrl_tx.clone(); // the data-plane pump sends adaptive-FEC LossReports
|
||||
let worker = std::thread::Builder::new()
|
||||
.name("punktfunk-client".into())
|
||||
@@ -745,6 +770,7 @@ impl NativeClient {
|
||||
fec_recovered: fec_recovered_w,
|
||||
hot_tids: hot_tids_w,
|
||||
clock_offset: clock_offset_w,
|
||||
decode_lat: decode_lat_w,
|
||||
}));
|
||||
})
|
||||
.map_err(PunktfunkError::Io)?;
|
||||
@@ -778,6 +804,10 @@ impl NativeClient {
|
||||
rfi: Mutex::new(RfiRecovery::default()),
|
||||
hot_tids,
|
||||
clock_offset,
|
||||
decode_lat,
|
||||
// The controller arms exactly when the pump does (see `abr::BitrateController::new`
|
||||
// below): Automatic (the user asked for bitrate 0) and not a rate-pinned PyroWave stream.
|
||||
wants_decode: bitrate_kbps == 0 && negotiated.codec != crate::quic::CODEC_PYROWAVE,
|
||||
mode: mode_slot,
|
||||
host_fingerprint: negotiated.host_fingerprint,
|
||||
resolved_compositor: negotiated.compositor,
|
||||
@@ -1090,6 +1120,30 @@ impl NativeClient {
|
||||
self.clock_offset.clone()
|
||||
}
|
||||
|
||||
/// Report one decoded frame's decode-stage latency, in microseconds: the wall-clock elapsed from
|
||||
/// the access unit leaving [`next_frame`](Self::next_frame) to its decoded output becoming
|
||||
/// available (dequeued from the decoder). This feeds the "Automatic" bitrate controller's decode
|
||||
/// signal — the only one that sees the client's own decoder, so the rate can be capped at the
|
||||
/// real decode limit instead of climbing to the network link ceiling and choking a slower HW
|
||||
/// decoder (the LAN-vs-mobile-decoder case). Measure from the AU handoff, NOT from the codec-queue
|
||||
/// call, so decoder-input backpressure (the backlog) is included; exclude the presenter's vsync
|
||||
/// wait so a paced/capped frame rate doesn't read as decode latency. Cheap and lock-brief — the
|
||||
/// embedder may call it every frame unconditionally; the controller ignores it when Automatic is
|
||||
/// off and the pump drains it every window regardless, so the accumulator stays bounded.
|
||||
pub fn report_decode_us(&self, us: u32) {
|
||||
let mut acc = self.decode_lat.lock().unwrap();
|
||||
acc.sum_us += us as u64;
|
||||
acc.count += 1;
|
||||
}
|
||||
|
||||
/// Whether [`report_decode_us`](Self::report_decode_us) is worth calling this session: `true`
|
||||
/// only when the adaptive-bitrate controller is armed (Automatic bitrate, non-PyroWave), so an
|
||||
/// embedder can skip the per-frame decode-latency measurement entirely for explicit-bitrate and
|
||||
/// PyroWave sessions (where the signal is ignored). Constant for the session — check once.
|
||||
pub fn wants_decode_latency(&self) -> bool {
|
||||
self.wants_decode
|
||||
}
|
||||
|
||||
/// Start a bandwidth speed test: ask the host to burst filler over the data plane at
|
||||
/// `target_kbps` of goodput for `duration_ms`, *briefly pausing video*. Non-blocking — the
|
||||
/// measurement accumulates in the background; poll [`NativeClient::probe_result`] until its
|
||||
@@ -1366,6 +1420,9 @@ struct WorkerArgs {
|
||||
/// The live clock offset (see [`NativeClient::clock_offset`]): the worker seeds it with the
|
||||
/// connect-time estimate; the control task's mid-stream re-syncs update it.
|
||||
clock_offset: Arc<AtomicI64>,
|
||||
/// Decode-stage latency samples from the embedder (see [`NativeClient::decode_lat`]): the pump
|
||||
/// drains a window mean into the adaptive-bitrate controller's decode signal.
|
||||
decode_lat: Arc<Mutex<DecodeLatAcc>>,
|
||||
}
|
||||
|
||||
/// The worker: QUIC handshake, then the input/datagram/control tasks + the blocking
|
||||
@@ -1418,6 +1475,7 @@ async fn worker_main(args: WorkerArgs) {
|
||||
fec_recovered,
|
||||
hot_tids,
|
||||
clock_offset,
|
||||
decode_lat,
|
||||
} = args;
|
||||
let setup = async {
|
||||
let remote: std::net::SocketAddr = join_host_port(&host, port)
|
||||
@@ -1977,6 +2035,7 @@ async fn worker_main(args: WorkerArgs) {
|
||||
let pump_hot_tids = hot_tids.clone();
|
||||
let pump_clock_offset = clock_offset.clone();
|
||||
let pump_clock_gen = clock_gen.clone();
|
||||
let pump_decode_lat = decode_lat.clone();
|
||||
let _ = tokio::task::spawn_blocking(move || {
|
||||
pin_thread_user_interactive(); // feeds the frame channel → the user-interactive video pump
|
||||
register_hot_tid(&pump_hot_tids); // this thread does UDP receive + FEC reassembly — hint it
|
||||
@@ -2162,14 +2221,34 @@ async fn worker_main(args: WorkerArgs) {
|
||||
let owd_mean_us =
|
||||
(owd_frames > 0).then(|| (owd_sum_ns / owd_frames as i128 / 1000) as i64);
|
||||
(owd_sum_ns, owd_frames) = (0, 0);
|
||||
// Drain the embedder's decode-latency window (always, so it stays bounded even when
|
||||
// the controller is disabled) → the mean feeds the decode signal; `None` when the
|
||||
// embedder reported nothing this window (old embedder / no decoded frames).
|
||||
let decode_mean_us = {
|
||||
let mut acc = pump_decode_lat.lock().unwrap();
|
||||
let (sum, count) = (acc.sum_us, acc.count);
|
||||
*acc = DecodeLatAcc::default();
|
||||
(count > 0).then(|| (sum / count as u64) as i64)
|
||||
};
|
||||
if let Some(kbps) = abr.on_window(
|
||||
Instant::now(),
|
||||
window_dropped,
|
||||
loss_ppm,
|
||||
owd_mean_us,
|
||||
decode_mean_us,
|
||||
flush_in_window,
|
||||
) {
|
||||
tracing::info!(kbps, "adaptive bitrate: requesting encoder re-target");
|
||||
// Log the window's signals alongside the decision so an on-glass session can
|
||||
// tell a decode-driven re-target (the new signal — decode_mean_us elevated with
|
||||
// loss/OWD flat) from a network-driven one.
|
||||
tracing::info!(
|
||||
kbps,
|
||||
loss_ppm,
|
||||
owd_mean_us = owd_mean_us.unwrap_or(-1),
|
||||
decode_mean_us = decode_mean_us.unwrap_or(-1),
|
||||
flushed = flush_in_window,
|
||||
"adaptive bitrate: requesting encoder re-target"
|
||||
);
|
||||
let _ = ctrl_tx.try_send(CtrlRequest::SetBitrate(kbps));
|
||||
}
|
||||
flush_in_window = false;
|
||||
|
||||
Reference in New Issue
Block a user