Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 enricobuehler
|
a784682d4c | ||
|
enricobuehler
|
134874b3c8 | ||
|
enricobuehler
|
b22d0da75b | ||
|
enricobuehler
|
d398e2296f | ||
|
enricobuehler
|
f9668b16a1 |
Generated
+27
-27
@@ -2159,7 +2159,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "latency-probe"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
|
||||
[[package]]
|
||||
name = "lazy_static"
|
||||
@@ -2264,7 +2264,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "libvpl-sys"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"bindgen",
|
||||
"cmake",
|
||||
@@ -2299,7 +2299,7 @@ checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
|
||||
|
||||
[[package]]
|
||||
name = "loss-harness"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"punktfunk-core",
|
||||
]
|
||||
@@ -2788,7 +2788,7 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
|
||||
|
||||
[[package]]
|
||||
name = "pf-capture"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ashpd",
|
||||
@@ -2808,7 +2808,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-client-core"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ash",
|
||||
@@ -2832,7 +2832,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-clipboard"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ashpd",
|
||||
@@ -2850,7 +2850,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-console-ui"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ash",
|
||||
@@ -2871,7 +2871,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-encode"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ash",
|
||||
@@ -2894,7 +2894,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-ffvk"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"ash",
|
||||
"bindgen",
|
||||
@@ -2903,7 +2903,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-frame"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"libc",
|
||||
@@ -2915,7 +2915,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-gpu"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"pf-host-config",
|
||||
@@ -2929,11 +2929,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-host-config"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
|
||||
[[package]]
|
||||
name = "pf-inject"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ashpd",
|
||||
@@ -2961,14 +2961,14 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-paths"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pf-presenter"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ash",
|
||||
@@ -2983,7 +2983,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-vdisplay"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ashpd",
|
||||
@@ -3013,7 +3013,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-win-display"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"pf-paths",
|
||||
@@ -3025,7 +3025,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "pf-zerocopy"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ash",
|
||||
@@ -3221,7 +3221,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-client-android"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"android_logger",
|
||||
"jni",
|
||||
@@ -3237,7 +3237,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-client-linux"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-channel",
|
||||
@@ -3253,7 +3253,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-client-session"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"pf-client-core",
|
||||
@@ -3268,7 +3268,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-client-windows"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"async-channel",
|
||||
"ffmpeg-next",
|
||||
@@ -3287,7 +3287,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-core"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"bytes",
|
||||
@@ -3318,7 +3318,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-host"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"aes",
|
||||
"aes-gcm",
|
||||
@@ -3400,7 +3400,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-probe"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"mdns-sd",
|
||||
@@ -3414,7 +3414,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "punktfunk-tray"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"ksni",
|
||||
@@ -3437,7 +3437,7 @@ checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea"
|
||||
|
||||
[[package]]
|
||||
name = "pyrowave-sys"
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
dependencies = [
|
||||
"bindgen",
|
||||
"cmake",
|
||||
|
||||
+1
-1
@@ -48,7 +48,7 @@ exclude = [
|
||||
ndk = { path = "clients/android/native/vendor/ndk" }
|
||||
|
||||
[workspace.package]
|
||||
version = "0.14.0"
|
||||
version = "0.15.0"
|
||||
edition = "2021"
|
||||
rust-version = "1.82"
|
||||
license = "MIT OR Apache-2.0"
|
||||
|
||||
+1
-1
@@ -10,7 +10,7 @@
|
||||
"name": "MIT OR Apache-2.0",
|
||||
"identifier": "MIT OR Apache-2.0"
|
||||
},
|
||||
"version": "0.14.0"
|
||||
"version": "0.15.0"
|
||||
},
|
||||
"paths": {
|
||||
"/api/v1/clients": {
|
||||
|
||||
@@ -404,7 +404,14 @@ fn feeder_loop(
|
||||
// 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();
|
||||
// Core reassembly-completion stamp (ABI v9), NOT the pull instant: stamping
|
||||
// here would fold the hand-off queue wait into the network latency figure
|
||||
// (a client-side standing backlog masquerading as network). 0 = older core.
|
||||
let received_ns = if frame.received_ns > 0 {
|
||||
frame.received_ns as i128
|
||||
} else {
|
||||
now_realtime_ns()
|
||||
};
|
||||
{
|
||||
let mut g = in_flight
|
||||
.lock()
|
||||
|
||||
@@ -221,7 +221,13 @@ pub(super) fn run_sync(
|
||||
// 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();
|
||||
// Core reassembly-completion stamp (ABI v9), not the pull instant — see
|
||||
// async_loop: a pull stamp folds hand-off queue wait into "network".
|
||||
let received_ns = if frame.received_ns > 0 {
|
||||
frame.received_ns as i128
|
||||
} else {
|
||||
now_realtime_ns()
|
||||
};
|
||||
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
|
||||
|
||||
@@ -579,13 +579,21 @@ struct ContentView: View {
|
||||
model?.disconnect() // the captured-state ⌃⌥⇧D combo
|
||||
},
|
||||
onFrame: { [meter = model.meter, latency = model.latency,
|
||||
split = model.latencySplit, offset = conn.clockOffsetNs] au in
|
||||
split = model.latencySplit, queue = model.clientQueue,
|
||||
offset = conn.clockOffsetNs] au in
|
||||
meter.note(byteCount: au.data.count)
|
||||
latency.record(ptsNs: au.ptsNs, offsetNs: offset)
|
||||
// The same receipt, keyed by pts, awaiting its 0xCF host timing (the
|
||||
// host/network split — drained by the 1 s stats tick).
|
||||
// host/network split — drained by the 1 s stats tick). receivedNs is
|
||||
// the core's reassembly stamp (ABI v9), so the split's network term no
|
||||
// longer contains the client-queue wait...
|
||||
split.recordReceipt(
|
||||
ptsNs: au.ptsNs, receivedNs: au.receivedNs, offsetNs: offset)
|
||||
// ...which is measured as its own term instead (receipt→pull, both
|
||||
// client-local).
|
||||
queue.record(
|
||||
ptsNs: UInt64(bitPattern: au.receivedNs), atNs: au.pulledNs,
|
||||
offsetNs: 0)
|
||||
},
|
||||
onSessionEnd: { [weak model] in
|
||||
Task { @MainActor in model?.sessionEnded() }
|
||||
|
||||
@@ -102,6 +102,12 @@ final class SessionModel: ObservableObject {
|
||||
@Published var decodeValid = false
|
||||
@Published var displayP50Ms = 0.0
|
||||
@Published var displayValid = false
|
||||
/// Client-queue wait: core reassembly receipt → the pump's pull (`AccessUnit.pulledNs −
|
||||
/// receivedNs`, ABI v9 receipt split — the 2026-07 two-pair investigation). ~0 on a healthy
|
||||
/// stream; a persistent value is a client-side standing backlog that used to hide inside
|
||||
/// "network". Shown in the detailed tier only when it says something (≥ ~2 ms).
|
||||
@Published var clientQueueP50Ms = 0.0
|
||||
@Published var clientQueueValid = false
|
||||
/// The measured OS present floor (design/apple-presentation-rebuild.md): the deadline
|
||||
/// engine's vend→glass pipeline depth — an OS property no client can pace under (~2 refresh
|
||||
/// intervals composited; would read ~1 under direct-to-display). The HUD subtracts it from
|
||||
@@ -147,6 +153,9 @@ final class SessionModel: ObservableObject {
|
||||
let endToEnd = LatencyMeter()
|
||||
let decodeStage = LatencyMeter()
|
||||
let displayStage = LatencyMeter()
|
||||
/// Client-queue sampler (see `clientQueueP50Ms`) — fed per AU by the stream view's onFrame,
|
||||
/// drained by the same 1 s tick as the stage meters.
|
||||
let clientQueue = LatencyMeter()
|
||||
/// The OS present floor sampler (see `osFloorP50Ms`) — fed one sample per display-link
|
||||
/// update by the deadline engine, drained by the same 1 s tick as the stage meters.
|
||||
let presentFloor = LatencyMeter()
|
||||
@@ -489,6 +498,7 @@ final class SessionModel: ObservableObject {
|
||||
endToEndValid = false
|
||||
decodeValid = false
|
||||
displayValid = false
|
||||
clientQueueValid = false
|
||||
osFloorValid = false
|
||||
lostFrames = 0
|
||||
lostPct = 0
|
||||
@@ -679,6 +689,12 @@ final class SessionModel: ObservableObject {
|
||||
} else {
|
||||
self.osFloorValid = false
|
||||
}
|
||||
if let q = self.clientQueue.drain() {
|
||||
self.clientQueueP50Ms = q.p50Ms
|
||||
self.clientQueueValid = true
|
||||
} else {
|
||||
self.clientQueueValid = false
|
||||
}
|
||||
// Mirror the window to the unified log (see statsLog) — one line per second,
|
||||
// stages in ms, only while frames actually flowed. `fps` counts RECEIVED AUs;
|
||||
// `presents` counts frames that reached glass (the display meter's sample count)
|
||||
@@ -691,7 +707,7 @@ final class SessionModel: ObservableObject {
|
||||
let line = String(
|
||||
format: "fps=%d presents=%d e2e_p50=%.1f e2e_p95=%.1f hostnet_p50=%.1f "
|
||||
+ "decode_p50=%.1f display_p50=%.1f lost=%d "
|
||||
+ "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f",
|
||||
+ "floor_p50=%.1f display_adj=%.1f e2e_adj=%.1f queue_p50=%.1f",
|
||||
frames,
|
||||
displayWindow?.count ?? 0,
|
||||
self.endToEndValid ? self.endToEndP50Ms : -1,
|
||||
@@ -702,7 +718,8 @@ final class SessionModel: ObservableObject {
|
||||
lost,
|
||||
self.osFloorValid ? self.osFloorP50Ms : -1,
|
||||
self.displayValid ? self.displayAdjP50Ms : -1,
|
||||
self.endToEndValid ? self.endToEndAdjP50Ms : -1)
|
||||
self.endToEndValid ? self.endToEndAdjP50Ms : -1,
|
||||
self.clientQueueValid ? self.clientQueueP50Ms : -1)
|
||||
statsLog.info("\(line, privacy: .public)")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,6 +118,16 @@ struct StreamHUDView: View {
|
||||
.font(.system(.caption2, design: .monospaced))
|
||||
.foregroundStyle(.tertiary)
|
||||
}
|
||||
// Client-queue wait (reassembly receipt → decode pull, ABI v9 split): ~0 on
|
||||
// a healthy stream and hidden as noise; shown from 2 ms — a persistent value
|
||||
// is a client-side standing backlog that pre-split builds displayed as
|
||||
// "network" (the 2026-07 two-pair plateau). The core's standing-latency
|
||||
// bleed logs alongside when it acts on the same state.
|
||||
if model.clientQueueValid && model.clientQueueP50Ms >= 2 {
|
||||
Text("client queue +\(model.clientQueueP50Ms, specifier: "%.1f") (receive backlog — standing if it persists)")
|
||||
.font(.system(.caption2, design: .monospaced))
|
||||
.foregroundStyle(.tertiary)
|
||||
}
|
||||
}
|
||||
} else if model.hostNetworkValid {
|
||||
// Stage-1 fallback presenter: the layer decodes + presents internally with no
|
||||
|
||||
@@ -35,10 +35,31 @@ public struct AccessUnit: Sendable {
|
||||
public let ptsNs: UInt64
|
||||
public let frameIndex: UInt32
|
||||
public let flags: UInt32
|
||||
/// Client `CLOCK_REALTIME` instant the AU was handed over by the core (post-FEC, decrypted)
|
||||
/// — the **received** measurement point of design/stats-unification.md. The decode stage is
|
||||
/// `decodedNs - receivedNs`, both client-local (no skew offset applies).
|
||||
/// Client `CLOCK_REALTIME` instant the AU finished reassembly in the core (post-FEC,
|
||||
/// decrypted — `PunktfunkFrame.received_ns`, ABI v9) — the **received** measurement point of
|
||||
/// design/stats-unification.md. NOT the pull instant: stamping at the pull folded the
|
||||
/// pre-decode hand-off wait into the network term, which is how the 2026-07 two-pair
|
||||
/// standing-latency plateau hid as "network". The decode stage is `decodedNs - receivedNs`,
|
||||
/// both client-local (no skew offset applies).
|
||||
public let receivedNs: Int64
|
||||
/// Client `CLOCK_REALTIME` instant this pull returned. `pulledNs - receivedNs` is the
|
||||
/// client-queue wait (kernel hand-off + FrameChannel dwell) — the term the HUD splits out
|
||||
/// so a client-side standing backlog can never masquerade as network latency again.
|
||||
public let pulledNs: Int64
|
||||
|
||||
/// `pulledNs` defaults to `receivedNs` (zero queue wait) for callers with no pull instant —
|
||||
/// the synthetic probe AUs and decode tests, where the split is meaningless.
|
||||
public init(
|
||||
data: Data, ptsNs: UInt64, frameIndex: UInt32, flags: UInt32,
|
||||
receivedNs: Int64, pulledNs: Int64? = nil
|
||||
) {
|
||||
self.data = data
|
||||
self.ptsNs = ptsNs
|
||||
self.frameIndex = frameIndex
|
||||
self.flags = flags
|
||||
self.receivedNs = receivedNs
|
||||
self.pulledNs = pulledNs ?? receivedNs
|
||||
}
|
||||
}
|
||||
|
||||
/// One Opus audio packet (48 kHz stereo, 5 ms frames) — decode with AVAudioConverter
|
||||
@@ -662,11 +683,16 @@ public final class PunktfunkConnection {
|
||||
let data = Data(bytes: base, count: Int(frame.len)) // copy: ptr valid only until next call
|
||||
var ts = timespec()
|
||||
clock_gettime(CLOCK_REALTIME, &ts)
|
||||
let receivedNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
|
||||
let pulledNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
|
||||
// Receipt = the core's reassembly-completion stamp (ABI v9); the pull instant is
|
||||
// kept separately so the client-queue wait is its own measured term. 0 would mean a
|
||||
// pre-v9 core — impossible here (core and Kit ship in one binary), but fall back to
|
||||
// the pull instant rather than record a 1970 receipt.
|
||||
let receivedNs = frame.received_ns > 0 ? Int64(frame.received_ns) : pulledNs
|
||||
return AccessUnit(
|
||||
data: data, ptsNs: frame.pts_ns,
|
||||
frameIndex: frame.frame_index, flags: frame.flags,
|
||||
receivedNs: receivedNs)
|
||||
receivedNs: receivedNs, pulledNs: pulledNs)
|
||||
case statusNoFrame:
|
||||
return nil
|
||||
case statusClosed:
|
||||
|
||||
@@ -1213,7 +1213,9 @@ public final class Stage2Pipeline {
|
||||
let chunkAligned =
|
||||
au.flags & PunktfunkConnection.userFlagChunkAligned != 0
|
||||
let ptsNs = au.ptsNs
|
||||
let receivedNs = au.receivedNs
|
||||
// Decode stage starts at the PULL (matching the VT path's FrameContext —
|
||||
// receipt→pull is the HUD's separate client-queue term, ABI v9 split).
|
||||
let receivedNs = au.pulledNs
|
||||
let flags = au.flags
|
||||
let submitted = decoder.decode(
|
||||
au: au.data, chunkAligned: chunkAligned, windowSize: windowSize
|
||||
|
||||
@@ -32,9 +32,12 @@ public enum ReadyImage: @unchecked Sendable {
|
||||
public struct ReadyFrame: @unchecked Sendable {
|
||||
/// Host capture clock (the AU's pts), in nanoseconds.
|
||||
public let ptsNs: UInt64
|
||||
/// Client `CLOCK_REALTIME` instant the AU was received (`AccessUnit.receivedNs`, threaded
|
||||
/// through the decode via the frame refcon), in nanoseconds. 0 when unknown (a caller that
|
||||
/// didn't stamp receipt) — the decode-stage meter then drops the sample via its sanity guard.
|
||||
/// Client `CLOCK_REALTIME` instant the AU left `nextAU` (`AccessUnit.pulledNs`, threaded
|
||||
/// through the decode via the frame refcon), in nanoseconds — the decode stage's start
|
||||
/// point. (Named for its historical role; since the ABI v9 receipt split the true
|
||||
/// reassembly receipt lives on `AccessUnit.receivedNs`, and receipt→pull is the HUD's own
|
||||
/// client-queue term.) 0 when unknown (a caller that didn't stamp) — the decode-stage meter
|
||||
/// then drops the sample via its sanity guard.
|
||||
public let receivedNs: Int64
|
||||
/// Client `CLOCK_REALTIME` instant decode completed, in nanoseconds.
|
||||
public let decodedNs: Int64
|
||||
@@ -167,7 +170,11 @@ public final class VideoDecoder: @unchecked Sendable {
|
||||
var infoOut = VTDecodeInfoFlags()
|
||||
// The AU's receipt instant + wire flags ride through as a retained context; the output
|
||||
// callback reclaims it. Retain immediately before submit so no early return can leak it.
|
||||
let ctx = FrameContext(receivedNs: au.receivedNs, flags: au.flags)
|
||||
// The decode stage starts at the PULL (the AU leaving nextAU), not the reassembly
|
||||
// receipt: both consumers — the decode-stage meter and the ABR decode signal — are
|
||||
// specified from the pull, and the receipt→pull wait is the HUD's separate client-queue
|
||||
// term (see AccessUnit.pulledNs).
|
||||
let ctx = FrameContext(receivedNs: au.pulledNs, flags: au.flags)
|
||||
let refcon = Unmanaged.passRetained(ctx).toOpaque()
|
||||
let status = VTDecompressionSessionDecodeFrame(
|
||||
session,
|
||||
|
||||
@@ -27,9 +27,13 @@ ui = ["dep:pf-console-ui", "dep:serde_json"]
|
||||
# Same Linux+Windows gating as the rest of the client stack; elsewhere this is a stub
|
||||
# binary.
|
||||
[target.'cfg(any(target_os = "linux", windows))'.dependencies]
|
||||
pf-presenter = { path = "../../crates/pf-presenter" }
|
||||
# `default-features = false` on both: THIS crate's `pyrowave` feature (above) is the single
|
||||
# switch that turns the wavelet codec on, and it enables it explicitly on each. Inheriting their
|
||||
# defaults instead would make `--no-default-features` a lie — the Windows ARM64 leg builds that
|
||||
# way precisely to skip the vendored PyroWave C++, which has no ARM64 SIMD path.
|
||||
pf-presenter = { path = "../../crates/pf-presenter", default-features = false }
|
||||
pf-console-ui = { path = "../../crates/pf-console-ui", optional = true }
|
||||
pf-client-core = { path = "../../crates/pf-client-core" }
|
||||
pf-client-core = { path = "../../crates/pf-client-core", default-features = false }
|
||||
punktfunk-core = { path = "../../crates/punktfunk-core", features = ["quic"] }
|
||||
# The fake-library dev hook (`PUNKTFUNK_FAKE_LIBRARY`, browse mode) parses GameEntry JSON.
|
||||
serde_json = { version = "1", optional = true }
|
||||
|
||||
@@ -29,7 +29,17 @@ punktfunk-core = { path = "../../crates/punktfunk-core", features = ["quic"] }
|
||||
# The shared client service layer: the trust/settings stores (ONE `Settings` struct for the
|
||||
# shell and the spawned session binary — src/trust.rs re-exports it) and the game-library
|
||||
# data model (fetch + art pipeline) behind the library page.
|
||||
pf-client-core = { path = "../../crates/pf-client-core" }
|
||||
#
|
||||
# `default-features = false` drops pf-client-core's default `pyrowave`, which would otherwise
|
||||
# build the vendored PyroWave C++ INTO THE SHELL — dead weight here (the shell never decodes;
|
||||
# it only offers "pyrowave" as a codec preference string the session binary acts on) and fatal
|
||||
# on ARM64, where Granite's math falls back to x86 SSE intrinsics and stops at
|
||||
# `simd.hpp: #error "Implement me."`. This does NOT drop PyroWave from the Windows client:
|
||||
# decode lives in the spawned punktfunk-session binary, whose own default enables the feature,
|
||||
# and cargo's feature unification turns it back on for the shared pf-client-core whenever that
|
||||
# binary is in the same build (x64). On the ARM64 leg both are built --no-default-features, so
|
||||
# nothing enables it and the C++ is never compiled.
|
||||
pf-client-core = { path = "../../crates/pf-client-core", default-features = false }
|
||||
|
||||
# WinUI 3 UI via windows-reactor (a declarative React-like framework backed by WinUI). Its
|
||||
# `build.rs` downloads the Windows App SDK NuGets and stages the bootstrap DLL + resources.pri
|
||||
|
||||
@@ -59,9 +59,14 @@
|
||||
</Application>
|
||||
<!--
|
||||
Second entry point: the couch/console UI, for an HTPC or a TV-attached box where the
|
||||
desktop shell is the wrong first screen. Same full-trust executable, launched with
|
||||
`--console`, which hands straight off to the session binary's controller-driven
|
||||
browse mode (host list, pairing, settings, library) fullscreen.
|
||||
desktop shell is the wrong first screen. Its own executable (punktfunk-console.exe)
|
||||
because an MSIX Application entry cannot pass arguments to a full-trust exe; it hands
|
||||
straight off to the session binary's controller-driven browse mode (host list,
|
||||
pairing, settings, library) fullscreen.
|
||||
|
||||
NOTE: never write a double hyphen in this file. XML forbids it inside a comment, and
|
||||
makepri rejects the whole manifest ("Appx manifest not found or is invalid") — which
|
||||
is exactly how the console flag spelled out here broke the v0.15.0 MSIX build.
|
||||
-->
|
||||
<Application Id="PunktfunkConsole" Executable="punktfunk-console.exe"
|
||||
EntryPoint="Windows.FullTrustApplication">
|
||||
|
||||
@@ -447,8 +447,16 @@ fn pump(
|
||||
// every ~8–16 ms at 60–120 Hz anyway, so this rarely times out mid-stream).
|
||||
match connector.next_frame(Duration::from_millis(20)) {
|
||||
Ok(frame) => {
|
||||
// The `received` point: AU fully reassembled, in hand, before decode.
|
||||
let received_ns = now_ns();
|
||||
// The `received` point: reassembly COMPLETION, stamped by the core session as
|
||||
// the AU crossed poll_frame (ABI v9). Stamping here at the hand-off pull instead
|
||||
// would fold the pre-decode queue wait into `host+network` — a client-side
|
||||
// standing backlog masquerading as network latency (the 2026-07 two-pair
|
||||
// investigation). 0 = a core predating the stamp; fall back to the pull instant.
|
||||
let received_ns = if frame.received_ns > 0 {
|
||||
frame.received_ns
|
||||
} else {
|
||||
now_ns()
|
||||
};
|
||||
// fps / goodput count every received AU (spec), decoded or not.
|
||||
frames_n += 1;
|
||||
bytes_n += frame.data.len() as u64;
|
||||
|
||||
@@ -1078,7 +1078,18 @@ impl Encoder for NvencCudaEncoder {
|
||||
// 4:4:4 honesty: engage FREXT only on a genuine YUV444 input; a subsampled NV12/RGB input
|
||||
// can't reconstruct full chroma, so clear the flag so `caps().chroma_444` is truthful.
|
||||
self.chroma_444 = self.chroma_444 && buf.yuv444;
|
||||
self.init_session()?;
|
||||
// `init_session` publishes `self.encoder` before its remaining fallible steps (bitstream
|
||||
// buffers, input-surface alloc, `register_resource`), so a failure there leaves a live
|
||||
// session with `inited == false`. Every guard on the re-init path keys off `inited`, so
|
||||
// without this the next submit would skip teardown and overwrite `self.encoder`, leaking
|
||||
// the session and its registered input surfaces permanently. `teardown` keys off
|
||||
// `encoder.is_null()`, not `inited`, so it cleans up exactly this half-built state.
|
||||
if let Err(e) = self.init_session() {
|
||||
// SAFETY: the encode thread owns the session and a failed init leaves nothing
|
||||
// mid-encode to race with.
|
||||
unsafe { self.teardown() };
|
||||
return Err(e);
|
||||
}
|
||||
} else {
|
||||
// Steady state: the copy helpers need the shared context current on this thread.
|
||||
cuda::make_current().context("cuCtxSetCurrent (encode thread)")?;
|
||||
|
||||
@@ -1851,7 +1851,20 @@ impl VulkanVideoEncoder {
|
||||
let f = &self.frames[slot];
|
||||
let mut fb = [[0u32; 2]; 1];
|
||||
dev.get_query_pool_results(f.query_pool, 0, &mut fb, vk::QueryResultFlags::WAIT)?;
|
||||
let (off, len) = (fb[0][0] as usize, fb[0][1] as usize);
|
||||
// The (offset, bytes-written) pair is driver-reported: validate it against the bitstream
|
||||
// allocation BEFORE mapping, or the `from_raw_parts` below reads outside the buffer and
|
||||
// ships whatever it finds straight onto the wire. Checked in u64 so the add cannot wrap,
|
||||
// and before `map_memory` so there is no unmap to unwind on the error path.
|
||||
let (off64, len64) = (fb[0][0] as u64, fb[0][1] as u64);
|
||||
if off64.saturating_add(len64) > self.bs_size {
|
||||
anyhow::bail!(
|
||||
"vulkan-encode: driver reported bitstream feedback offset={off64} \
|
||||
bytes_written={len64}, outside the {} byte bitstream buffer — the encode likely \
|
||||
overflowed its destination range",
|
||||
self.bs_size
|
||||
);
|
||||
}
|
||||
let (off, len) = (off64 as usize, len64 as usize);
|
||||
let p =
|
||||
dev.map_memory(f.bs_mem, 0, vk::WHOLE_SIZE, vk::MemoryMapFlags::empty())? as *const u8;
|
||||
let prefix: &[u8] = if f.keyframe {
|
||||
@@ -1923,7 +1936,28 @@ impl Encoder for VulkanVideoEncoder {
|
||||
if first_frame < 0 || first_frame > last_frame {
|
||||
return false;
|
||||
}
|
||||
// Taint sweep BEFORE picking the anchor (the fecbec2d fix AMF and QSV got; this backend was
|
||||
// carved out one commit later and never received it). "Resident and older than THIS loss" is
|
||||
// not the same as "the client decoded it": after an earlier loss [a,b] was recovered at wire
|
||||
// r, everything in [a, r-1] is undecodable at the client — the lost frames plus every frame
|
||||
// that predicted through the gap. Those wires stay valid anchor candidates here until the
|
||||
// 8-slot ring rolls them out, so a LATER loss can anchor on one and ship corruption tagged
|
||||
// `recovery_anchor` — which is the client's definitive re-anchor signal (reanchor.rs), so it
|
||||
// lifts the post-loss freeze onto a picture built from a reference it never had.
|
||||
//
|
||||
// Blank `slot_wire` ONLY. `slot_poc` must keep naming every physically-resident DPB picture
|
||||
// for `build_h265_rps_s0`, or a conforming decoder evicts them and the anchor references a
|
||||
// picture the client already dropped. `slot_wire` is the RFI/loss domain; `slot_poc` is the
|
||||
// reference-delta domain. `prev_slot` and the normal P-frame path are indices, not wires, so
|
||||
// ordinary prediction is unaffected.
|
||||
for w in self.slot_wire.iter_mut() {
|
||||
if *w >= first_frame {
|
||||
*w = -1;
|
||||
}
|
||||
}
|
||||
// Can we anchor a clean P-frame to a resident slot strictly older than the loss?
|
||||
// (A sweep that empties every candidate yields `None` here and declines the RFI, matching
|
||||
// `qsv_live_ltr_rfi_taint_sweep_declines`.)
|
||||
match pick_recovery_slot(&self.slot_wire, first_frame) {
|
||||
Some(_) => {
|
||||
self.pending_loss = Some(first_frame);
|
||||
@@ -2697,6 +2731,62 @@ mod tests {
|
||||
assert_eq!(pick_recovery_slot(&[-1; 8], 5), None);
|
||||
}
|
||||
|
||||
/// The taint sweep (fecbec2d's fix, ported here): a slot encoded inside an EARLIER, still
|
||||
/// unrepaired loss window must not become the "known-good" anchor of a LATER loss. Without the
|
||||
/// sweep, `pick_recovery_slot` accepts it — it is resident and its wire is below the second
|
||||
/// loss start — and the frame ships tagged `recovery_anchor`, lifting the client's freeze onto
|
||||
/// a reference it never decoded.
|
||||
#[test]
|
||||
fn taint_sweep_excludes_slots_from_an_earlier_loss() {
|
||||
// Apply the sweep exactly as `invalidate_ref_frames` does.
|
||||
fn sweep(wires: &mut [i64], loss_first: i64) {
|
||||
for w in wires.iter_mut() {
|
||||
if *w >= loss_first {
|
||||
*w = -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Slots hold wires 0..7. Loss 1 starts at wire 4, so wires 4..7 are undecodable at the
|
||||
// client. A second loss report arrives at wire 6 while they are all still resident.
|
||||
let tainted = [4i64, 5, 6, 7];
|
||||
|
||||
// WITHOUT the sweep this is the bug: the newest wire below 6 is wire 5 — squarely inside
|
||||
// loss 1's unrepaired window — and it would be served as the "known-good" anchor.
|
||||
let unswept = [0i64, 1, 2, 3, 4, 5, 6, 7];
|
||||
let picked = pick_recovery_slot(&unswept, 6).expect("unswept picks something");
|
||||
assert!(
|
||||
tainted.contains(&unswept[picked]),
|
||||
"precondition: without the sweep the anchor comes from the earlier loss window"
|
||||
);
|
||||
|
||||
// WITH the sweep, loss 1 blanks 4..7, so loss 2 can only reach genuinely clean wires.
|
||||
let mut wires = unswept;
|
||||
sweep(&mut wires, 4);
|
||||
assert_eq!(wires, [0, 1, 2, 3, -1, -1, -1, -1]);
|
||||
let picked = pick_recovery_slot(&wires, 6).expect("clean wires remain");
|
||||
assert_eq!(picked, 3, "newest clean survivor is wire 3");
|
||||
assert!(!tainted.contains(&wires[picked]));
|
||||
|
||||
// Encoding resumes after recovery; wires 8..11 refill the swept slots and are clean. A
|
||||
// later loss at wire 10 legitimately anchors on wire 9 — the sweep must not over-reject.
|
||||
wires[4] = 8;
|
||||
wires[5] = 9;
|
||||
wires[6] = 10;
|
||||
wires[7] = 11;
|
||||
sweep(&mut wires, 10);
|
||||
assert_eq!(
|
||||
pick_recovery_slot(&wires, 10),
|
||||
Some(5),
|
||||
"wire 9 is post-recovery, clean"
|
||||
);
|
||||
|
||||
// A loss covering every live wire leaves nothing clean → decline, caller serves an IDR.
|
||||
let mut all = [5i64, 6, 7, 8, 9, 10, 11, 12];
|
||||
sweep(&mut all, 5);
|
||||
assert_eq!(pick_recovery_slot(&all, 5), None);
|
||||
}
|
||||
|
||||
/// The full-retention RPS: every resident picture is listed (so the decoder keeps it), the
|
||||
/// setup slot's dying occupant is not, and `used_by_curr_pic` marks exactly the real reference.
|
||||
#[test]
|
||||
|
||||
@@ -1135,7 +1135,19 @@ impl Encoder for NvencD3d11Encoder {
|
||||
self.chroma_444 = false;
|
||||
}
|
||||
let device = frame.device.clone();
|
||||
self.init_session(&device)?;
|
||||
// `init_session` publishes `self.encoder` (and charges LIVE_SESSION_UNITS) BEFORE its
|
||||
// last fallible steps, so a failure there leaves a live session with `inited == false`.
|
||||
// Every guard on the re-init path keys off `inited`, so without this the next submit
|
||||
// would skip teardown and overwrite `self.encoder` — leaking the session permanently
|
||||
// (toward the driver's per-process cap) along with its session-budget units.
|
||||
// `teardown` keys off `encoder.is_null()`, not `inited`, so it cleans up exactly this
|
||||
// half-built state and is a no-op when nothing was opened.
|
||||
if let Err(e) = self.init_session(&device) {
|
||||
// SAFETY: same contract as the teardown above — the encode thread owns the session,
|
||||
// and a failed init leaves nothing mid-encode to race with.
|
||||
unsafe { self.teardown() };
|
||||
return Err(e);
|
||||
}
|
||||
self.init_device = dev_raw;
|
||||
}
|
||||
// The session's opening frame — NVENC emits it as an IDR regardless of pic flags, so the
|
||||
|
||||
@@ -146,7 +146,12 @@ const NUM_LTR_SLOTS: usize = 2;
|
||||
/// `PUNKTFUNK_NO_QSV_LTR` — defeat switch for the LTR-RFI path (parity with
|
||||
/// `PUNKTFUNK_NO_AMF_LTR`); loss recovery then always falls back to IDR.
|
||||
fn ltr_disabled() -> bool {
|
||||
std::env::var("PUNKTFUNK_NO_QSV_LTR").is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
|
||||
// Same accepted spellings as AMF's `ltr_disabled` — this had dropped the `trim()` and the
|
||||
// `yes`/`on` forms, so a value with stray whitespace (easy to produce with `set VAR=1 `)
|
||||
// silently left LTR enabled on Intel while the identical value worked on AMD.
|
||||
std::env::var("PUNKTFUNK_NO_QSV_LTR")
|
||||
.map(|v| matches!(v.trim(), "1" | "true" | "yes" | "on"))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Frames between LTR marks (`PUNKTFUNK_LTR_INTERVAL_FRAMES`, shared with AMF); default ~1/4 s
|
||||
@@ -171,13 +176,22 @@ fn ltr_test_force_at() -> Option<i64> {
|
||||
/// Mirrors [`super::amf`]'s `PUNKTFUNK_INTRA_REFRESH` opt-in: request the intra-refresh wave
|
||||
/// instead of LTR (mutually exclusive — the wave sweeps the whole picture, LTR pins references).
|
||||
fn intra_refresh_requested() -> bool {
|
||||
// Spelling parity with AMF (see `ltr_disabled` above).
|
||||
std::env::var("PUNKTFUNK_INTRA_REFRESH")
|
||||
.is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
|
||||
.map(|v| matches!(v.trim(), "1" | "true" | "yes" | "on"))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// The wave period in frames (~0.5 s), the same shape as Linux NVENC / AMF.
|
||||
/// The wave period in frames (~0.5 s), `PUNKTFUNK_IR_PERIOD_FRAMES` overrides — the same knob and
|
||||
/// default as AMF / Linux NVENC. (This claimed parity while ignoring the env var entirely, so the
|
||||
/// knob silently did nothing on Intel; the clamp is kept because `mfxU16` bounds the field.)
|
||||
fn intra_refresh_period(fps: u32) -> u16 {
|
||||
(fps / 2).clamp(8, 240) as u16
|
||||
std::env::var("PUNKTFUNK_IR_PERIOD_FRAMES")
|
||||
.ok()
|
||||
.and_then(|s| s.trim().parse::<u32>().ok())
|
||||
.filter(|v| *v >= 2)
|
||||
.unwrap_or(fps / 2)
|
||||
.clamp(8, 240) as u16
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------
|
||||
@@ -700,7 +714,16 @@ pub struct QsvEncoder {
|
||||
/// `EncoderCaps::supports_rfi` and all per-frame marking/forcing below.
|
||||
ltr_active: bool,
|
||||
/// The wire frame index stored in each LTR slot (`None` = never marked).
|
||||
///
|
||||
/// This mirrors the HARDWARE DPB, so an entry must not be cleared merely because we distrust
|
||||
/// it: nulling issues no VPL call, and the encoder keeps the frame marked long-term until that
|
||||
/// `LongTermIdx` is re-marked or an IDR flushes it. Distrust is recorded in `ltr_tainted`
|
||||
/// instead, so the rejection list can still NAME the entry the hardware is holding.
|
||||
ltr_slots: [Option<i64>; NUM_LTR_SLOTS],
|
||||
/// Per-slot taint from `invalidate_ref_frames`' sweep: the mark is still live in the hardware
|
||||
/// DPB but was encoded inside the client's corrupt window, so it may not anchor a recovery —
|
||||
/// it must be REJECTED instead. Cleared wherever the slot is re-marked or the DPB is flushed.
|
||||
ltr_tainted: [bool; NUM_LTR_SLOTS],
|
||||
next_ltr_slot: usize,
|
||||
ltr_mark_interval: i64,
|
||||
/// Set by `invalidate_ref_frames`: the slot the next submitted frame force-references.
|
||||
@@ -775,6 +798,7 @@ impl QsvEncoder {
|
||||
ir_active: false,
|
||||
ltr_active: false,
|
||||
ltr_slots: [None; NUM_LTR_SLOTS],
|
||||
ltr_tainted: [false; NUM_LTR_SLOTS],
|
||||
next_ltr_slot: 0,
|
||||
ltr_mark_interval: ltr_mark_interval(fps),
|
||||
pending_force: None,
|
||||
@@ -899,6 +923,7 @@ impl QsvEncoder {
|
||||
self.ltr_active = ltr_active;
|
||||
self.ir_active = ir_active;
|
||||
self.ltr_slots = [None; NUM_LTR_SLOTS];
|
||||
self.ltr_tainted = [false; NUM_LTR_SLOTS];
|
||||
self.next_ltr_slot = 0;
|
||||
self.pending_force = None;
|
||||
self.hdr_applied = self.hdr_meta;
|
||||
@@ -1024,6 +1049,7 @@ impl Encoder for QsvEncoder {
|
||||
// An IDR voids the decoder's reference buffers — drop stale slots and any
|
||||
// queued force; the mark cadence below re-anchors on the IDR itself.
|
||||
self.ltr_slots = [None; NUM_LTR_SLOTS];
|
||||
self.ltr_tainted = [false; NUM_LTR_SLOTS]; // the IDR flushed the DPB with them
|
||||
self.next_ltr_slot = 0;
|
||||
self.pending_force = None;
|
||||
} else if self.ltr_test_force_at == Some(cur_idx) {
|
||||
@@ -1039,7 +1065,9 @@ impl Encoder for QsvEncoder {
|
||||
// emptied the slot since the force was queued. An empty slot means there is
|
||||
// nothing clean to re-reference — the frame must ship as a plain P WITHOUT the
|
||||
// `recovery_anchor` tag (the client lifts its post-loss freeze on that tag).
|
||||
if let Some(idx) = self.ltr_slots[slot] {
|
||||
// The slot is no longer emptied by the sweep, so test the taint flag too — a
|
||||
// tainted slot is exactly the "nothing clean to re-reference" case.
|
||||
if let Some(idx) = self.ltr_slots[slot].filter(|_| !self.ltr_tainted[slot]) {
|
||||
force_ltr = Some((slot, idx));
|
||||
recovery_anchor = true;
|
||||
}
|
||||
@@ -1047,6 +1075,9 @@ impl Encoder for QsvEncoder {
|
||||
if force_ltr.is_none() && (forced || cur_idx % self.ltr_mark_interval == 0) {
|
||||
let slot = self.next_ltr_slot;
|
||||
self.ltr_slots[slot] = Some(cur_idx);
|
||||
// Re-marking replaces the hardware's LongTermIdx: the tainted frame is gone from
|
||||
// the DPB and this slot is clean again.
|
||||
self.ltr_tainted[slot] = false;
|
||||
self.next_ltr_slot = (self.next_ltr_slot + 1) % NUM_LTR_SLOTS;
|
||||
mark_slot = Some(slot);
|
||||
}
|
||||
@@ -1309,13 +1340,23 @@ impl Encoder for QsvEncoder {
|
||||
// loss ships corruption as the recovery anchor, and every subsequent mark re-samples
|
||||
// the soup — the sustained-loss field failure where the picture never healed. Dropped
|
||||
// slots stay dropped; the cadence re-marks a clean frame within ~1/4 s.
|
||||
for marked in self.ltr_slots.iter_mut() {
|
||||
//
|
||||
// Mark tainted rather than clearing: `ltr_slots` mirrors the HARDWARE DPB, and nulling an
|
||||
// entry issues no VPL call — the frame stays marked long-term in the encoder. Clearing it
|
||||
// made the rejection list below (which iterates the post-sweep mirror and only names `Some`
|
||||
// slots) silently SKIP the one entry the sweep exists to distrust, so the recovery frame
|
||||
// could still predict from it. With two slots the "exactly one swept" case is the modal
|
||||
// one, and it was the broken one.
|
||||
for (slot, marked) in self.ltr_slots.iter().enumerate() {
|
||||
if marked.is_some_and(|idx| idx >= first) {
|
||||
*marked = None;
|
||||
self.ltr_tainted[slot] = true;
|
||||
}
|
||||
}
|
||||
let mut best: Option<(usize, i64)> = None;
|
||||
for (slot, marked) in self.ltr_slots.iter().enumerate() {
|
||||
if self.ltr_tainted[slot] {
|
||||
continue; // still in the DPB, but encoded inside the corrupt window
|
||||
}
|
||||
if let Some(idx) = *marked {
|
||||
if idx < first && best.is_none_or(|(_, b)| idx > b) {
|
||||
best = Some((slot, idx));
|
||||
@@ -1440,6 +1481,7 @@ impl Encoder for QsvEncoder {
|
||||
self.ltr_active = ltr;
|
||||
self.ir_active = ir;
|
||||
self.ltr_slots = [None; NUM_LTR_SLOTS];
|
||||
self.ltr_tainted = [false; NUM_LTR_SLOTS];
|
||||
self.next_ltr_slot = 0;
|
||||
self.pending_force = None;
|
||||
if let Some(inner) = self.inner.as_mut() {
|
||||
|
||||
@@ -223,6 +223,13 @@ impl Encoder for TrackedEncoder {
|
||||
}
|
||||
}
|
||||
|
||||
/// Ceiling applied to the negotiated bitrate before it reaches openh264: software H.264 realistically
|
||||
/// caps far below the rates a hardware session negotiates, and handing it the full figure just
|
||||
/// misconfigures its rate control. Module-scope so BOTH software arms share one value — the Linux
|
||||
/// arm was missing the clamp the Windows arm applied.
|
||||
#[cfg(any(target_os = "linux", target_os = "windows"))]
|
||||
const SW_BITRATE_CEIL: u64 = 100_000_000;
|
||||
|
||||
/// Open the platform encoder backend. Returns the encoder together with the display label of the
|
||||
/// branch that ACTUALLY opened (`nvenc`/`vaapi`/`vulkan`/`amf`/`qsv`/`software`) — the label feeds
|
||||
/// the mgmt API's live-session record, and only the open site knows which internal fallback won
|
||||
@@ -407,7 +414,13 @@ fn open_video_backend(
|
||||
);
|
||||
}
|
||||
let _ = (cuda, bit_depth); // software path is CPU + 8-bit only
|
||||
sw::OpenH264Encoder::open(format, width, height, fps, bitrate_bps)
|
||||
sw::OpenH264Encoder::open(
|
||||
format,
|
||||
width,
|
||||
height,
|
||||
fps,
|
||||
bitrate_bps.min(SW_BITRATE_CEIL),
|
||||
)
|
||||
.map(|e| (Box::new(e) as Box<dyn Encoder>, "software"))
|
||||
}
|
||||
"auto" | "" => {
|
||||
@@ -610,8 +623,6 @@ fn open_video_backend(
|
||||
(build a GPU backend: --features nvenc or amf-qsv, or request H264)"
|
||||
);
|
||||
let _ = (bit_depth, chroma); // the software H.264 path is 8-bit 4:2:0 only
|
||||
// Software H.264 realistically caps far below the negotiated hardware rates.
|
||||
const SW_BITRATE_CEIL: u64 = 100_000_000;
|
||||
sw::OpenH264Encoder::open(
|
||||
format,
|
||||
width,
|
||||
@@ -784,6 +795,12 @@ fn nvidia_present() -> bool {
|
||||
/// picks its vendor's backend — AMD/Intel → VAAPI on that GPU's render node, NVIDIA → NVENC (still
|
||||
/// requiring the proprietary driver's device nodes; a nouveau NVIDIA GPU can't NVENC) — otherwise
|
||||
/// today's NVIDIA-presence probe, unchanged.
|
||||
///
|
||||
/// ⚠ This resolves the **`auto` case only** — it deliberately ignores `encoder_pref`. It is NOT a
|
||||
/// mirror of [`open_video`]'s dispatch and must not be used to decide which backend a capability
|
||||
/// probe should ask: use [`linux_zero_copy_is_vaapi`], which layers `encoder_pref` on top of this.
|
||||
/// (`can_encode_10bit` used this directly and answered for the wrong backend whenever a host
|
||||
/// forced one.)
|
||||
#[cfg(target_os = "linux")]
|
||||
fn linux_auto_is_vaapi() -> bool {
|
||||
if let Some(g) = pf_gpu::manual_selection() {
|
||||
@@ -997,7 +1014,13 @@ pub fn can_encode_10bit(codec: Codec) -> bool {
|
||||
// only half the Linux gate — the capture side (GNOME 50+ portal monitor in HDR mode)
|
||||
// is resolved separately by the host (`capturer_supports_hdr` / the GameStream RTSP
|
||||
// honor), since this probe can't know what the compositor will negotiate.
|
||||
if linux_auto_is_vaapi() {
|
||||
// Resolve through the SAME helper `can_encode_444` uses (and which mirrors
|
||||
// `open_video`'s dispatch): `linux_auto_is_vaapi` ignores `encoder_pref`, so on a box
|
||||
// that forces a backend — e.g. `encoder_pref = "vaapi"` on an NVIDIA host — this probe
|
||||
// would answer for NVENC while the session actually opens VAAPI, and the negotiated bit
|
||||
// depth (plus the HDR/SDR colour label derived from it) would describe a backend that
|
||||
// never runs. That is exactly the dishonesty this probe exists to prevent.
|
||||
if linux_zero_copy_is_vaapi() {
|
||||
vaapi::probe_can_encode_10bit(codec)
|
||||
} else {
|
||||
linux::probe_can_encode_10bit(codec)
|
||||
|
||||
@@ -11,7 +11,11 @@ repository.workspace = true
|
||||
# Same Linux+Windows gating as the rest of the client stack (dmabuf import is the one
|
||||
# Linux-only module — see lib.rs).
|
||||
[target.'cfg(any(target_os = "linux", windows))'.dependencies]
|
||||
pf-client-core = { path = "../pf-client-core" }
|
||||
# `default-features = false`: the PyroWave decode backend is turned on through THIS crate's own
|
||||
# `pyrowave` feature (which re-exports it below), never by inheriting the dependency's default.
|
||||
# Otherwise a consumer that deliberately builds us without `pyrowave` still drags the vendored
|
||||
# C++ in — fatal on Windows ARM64, where Granite has no SIMD path.
|
||||
pf-client-core = { path = "../pf-client-core", default-features = false }
|
||||
# AVVkFrame access (Vulkan Video frames: live sync state under the frames lock).
|
||||
pf-ffvk = { path = "../pf-ffvk" }
|
||||
punktfunk-core = { path = "../punktfunk-core", features = ["quic"] }
|
||||
|
||||
@@ -125,6 +125,12 @@ pub struct PunktfunkFrame {
|
||||
pub frame_index: u32,
|
||||
pub pts_ns: u64,
|
||||
pub flags: u32,
|
||||
/// Wall-clock reassembly-completion instant (ns since the Unix epoch, CLOCK_REALTIME — the
|
||||
/// clock `pts_ns` and the skew handshake use). THIS is the receipt stamp for latency math:
|
||||
/// a stamp the embedder takes itself at the poll return additionally contains the
|
||||
/// pre-decode hand-off queue wait, so a client-side standing backlog would masquerade as
|
||||
/// network latency (ABI v9 — the 2026-07 two-pair standing-latency investigation).
|
||||
pub received_ns: u64,
|
||||
}
|
||||
|
||||
/// Snapshot of session counters.
|
||||
@@ -391,6 +397,7 @@ pub unsafe extern "C" fn punktfunk_client_poll_frame(
|
||||
frame_index: f.frame_index,
|
||||
pts_ns: f.pts_ns,
|
||||
flags: f.flags,
|
||||
received_ns: f.received_ns,
|
||||
};
|
||||
}
|
||||
PunktfunkStatus::Ok
|
||||
@@ -1744,6 +1751,7 @@ pub unsafe extern "C" fn punktfunk_connection_next_au(
|
||||
frame_index: f.frame_index,
|
||||
pts_ns: f.pts_ns,
|
||||
flags: f.flags,
|
||||
received_ns: f.received_ns,
|
||||
};
|
||||
}
|
||||
PunktfunkStatus::Ok
|
||||
|
||||
@@ -73,6 +73,142 @@ pub(crate) const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2;
|
||||
/// FIRST no-op clock flush — the moment a step is actually suspected.
|
||||
pub(crate) const CLOCK_RESYNC_INTERVAL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Standing-latency bleed (the 2026-07 two-pair investigation): how far above the session's own
|
||||
/// one-way-delay floor a report window's MINIMUM must sit to count as a standing elevation. The
|
||||
/// jump-to-live detectors above deliberately ignore anything below ~6 frames / 400 ms, so a
|
||||
/// small standing state — a sub-frame kernel/reassembly backlog, or a stale clock offset after a
|
||||
/// wall-clock step — is carried forever and reads as permanent extra "network" latency. 10 ms
|
||||
/// sits above skew-handshake error + normal LAN jitter, and below a single 60 fps frame period,
|
||||
/// so the observed one-frame plateau (~17 ms) trips it while a healthy stream cannot.
|
||||
pub(crate) const STANDING_LAT_THRESH_NS: i128 = 10_000_000;
|
||||
|
||||
/// Consecutive elevated report windows (~750 ms each) before the bleed escalates — ~4.5 s of a
|
||||
/// continuously standing, loss-free elevation. Windows with any loss reset the run: loss means
|
||||
/// genuine congestion, which the FEC/ABR machinery owns, not this detector.
|
||||
pub(crate) const STANDING_LAT_WINDOWS: u32 = 6;
|
||||
|
||||
/// Per-session cap on flush+keyframe bleeds. A standing state that survives a clock re-sync AND
|
||||
/// this many local flushes is not local and not clock — the path latency itself changed; the
|
||||
/// detector disarms with a warning instead of paying a recovery keyframe every few seconds.
|
||||
pub(crate) const STANDING_LAT_MAX_BLEEDS: u32 = 3;
|
||||
|
||||
/// What the standing-latency detector asks the pump to do this window (see [`StandingLatency`]).
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub(crate) enum StandingLatAction {
|
||||
None,
|
||||
/// First escalation: ask for a mid-stream clock re-sync — free, and a stale offset from a
|
||||
/// stepped/slewed wall clock produces exactly this signature (an applied re-sync re-bases
|
||||
/// the floor via the pump's `clock_gen` watch, clearing the elevation if that was the cause).
|
||||
Resync {
|
||||
above_ms: i64,
|
||||
},
|
||||
/// The elevation survived a re-sync attempt: flush the local receive backlog + request a
|
||||
/// keyframe (the jump-to-live action), draining a real sub-threshold standing queue. The
|
||||
/// pump reports execution back via [`StandingLatency::bled`]; an unexecuted action simply
|
||||
/// re-arms next window.
|
||||
Bleed {
|
||||
above_ms: i64,
|
||||
},
|
||||
/// Bleed cap reached and the elevation is back: give up and say so.
|
||||
Disarm {
|
||||
above_ms: i64,
|
||||
},
|
||||
}
|
||||
|
||||
/// Detector for a small, constant, loss-free one-way-delay elevation — the standing state the
|
||||
/// jump-to-live thresholds deliberately tolerate. Tracks the session's OWD floor (minimum of
|
||||
/// report-window minimums since start / last re-base) and escalates when windows sit
|
||||
/// persistently above it: re-sync first, then a bounded number of flush+keyframe bleeds, then
|
||||
/// disarm. Pure state machine (no clocks, no I/O) so the escalation ladder is unit-testable.
|
||||
pub(crate) struct StandingLatency {
|
||||
/// Lowest window-minimum OWD seen since session start / last [`rebase`](Self::rebase).
|
||||
floor_ns: Option<i128>,
|
||||
/// Minimum per-frame OWD this report window; `None` = no frames yet.
|
||||
window_min_ns: Option<i128>,
|
||||
/// Consecutive elevated windows.
|
||||
run: u32,
|
||||
/// The current elevation already got its re-sync request — next escalation is a bleed.
|
||||
resync_tried: bool,
|
||||
bleeds: u32,
|
||||
disarmed: bool,
|
||||
}
|
||||
|
||||
impl StandingLatency {
|
||||
pub(crate) fn new() -> Self {
|
||||
StandingLatency {
|
||||
floor_ns: None,
|
||||
window_min_ns: None,
|
||||
run: 0,
|
||||
resync_tried: false,
|
||||
bleeds: 0,
|
||||
disarmed: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Feed one frame's skew-corrected OWD (capture→reassembly-complete, ns). Caller gates on a
|
||||
/// live clock offset and plausibility (0 < owd < 10 s), like the ABR OWD signal.
|
||||
pub(crate) fn note_frame(&mut self, owd_ns: i128) {
|
||||
self.window_min_ns = Some(match self.window_min_ns {
|
||||
Some(m) => m.min(owd_ns),
|
||||
None => owd_ns,
|
||||
});
|
||||
}
|
||||
|
||||
/// Close a report window. `loss_free` = the window carried zero loss (loss resets the run —
|
||||
/// congestion is the FEC/ABR machinery's problem, and queues under loss are not "standing").
|
||||
pub(crate) fn on_window(&mut self, loss_free: bool) -> StandingLatAction {
|
||||
let Some(wmin) = self.window_min_ns.take() else {
|
||||
return StandingLatAction::None; // no frames this window — no evidence either way
|
||||
};
|
||||
let floor = *self.floor_ns.get_or_insert(wmin);
|
||||
self.floor_ns = Some(floor.min(wmin));
|
||||
let above_ns = wmin - floor;
|
||||
if self.disarmed {
|
||||
return StandingLatAction::None;
|
||||
}
|
||||
if !loss_free || above_ns < STANDING_LAT_THRESH_NS {
|
||||
self.run = 0;
|
||||
if above_ns < STANDING_LAT_THRESH_NS {
|
||||
self.resync_tried = false; // elevation cleared — a future one re-syncs first again
|
||||
}
|
||||
return StandingLatAction::None;
|
||||
}
|
||||
self.run += 1;
|
||||
if self.run < STANDING_LAT_WINDOWS {
|
||||
return StandingLatAction::None;
|
||||
}
|
||||
self.run = 0; // each escalation gets a fresh observation run
|
||||
let above_ms = (above_ns / 1_000_000) as i64;
|
||||
if !self.resync_tried {
|
||||
self.resync_tried = true;
|
||||
StandingLatAction::Resync { above_ms }
|
||||
} else if self.bleeds < STANDING_LAT_MAX_BLEEDS {
|
||||
StandingLatAction::Bleed { above_ms }
|
||||
} else {
|
||||
self.disarmed = true;
|
||||
StandingLatAction::Disarm { above_ms }
|
||||
}
|
||||
}
|
||||
|
||||
/// The pump executed a [`StandingLatAction::Bleed`] (flush + keyframe). The floor is KEPT: a
|
||||
/// successful bleed brings OWD back down to it (elevation clears naturally); an unsuccessful
|
||||
/// one leaves the elevation visible so the ladder continues toward the cap.
|
||||
pub(crate) fn bled(&mut self) {
|
||||
self.bleeds += 1;
|
||||
self.window_min_ns = None;
|
||||
}
|
||||
|
||||
/// A mid-stream clock re-sync was APPLIED (the pump's `clock_gen` watch): every OWD reading
|
||||
/// shifted, so the floor and any elevation measured under the old offset are meaningless —
|
||||
/// re-learn from scratch. The bleed budget survives (it caps keyframes per session).
|
||||
pub(crate) fn rebase(&mut self) {
|
||||
self.floor_ns = None;
|
||||
self.window_min_ns = None;
|
||||
self.run = 0;
|
||||
self.resync_tried = false;
|
||||
}
|
||||
}
|
||||
|
||||
/// 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
|
||||
@@ -191,6 +327,7 @@ mod frame_channel_tests {
|
||||
pts_ns: i as u64,
|
||||
flags: 0,
|
||||
complete: true,
|
||||
received_ns: 0,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -258,3 +395,143 @@ mod frame_channel_tests {
|
||||
assert_eq!(popped(&ch), Some(total - FRAME_QUEUE_HARD_CAP as u32));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod standing_latency_tests {
|
||||
use super::{
|
||||
StandingLatAction, StandingLatency, STANDING_LAT_MAX_BLEEDS, STANDING_LAT_THRESH_NS,
|
||||
STANDING_LAT_WINDOWS,
|
||||
};
|
||||
|
||||
const FLOOR: i128 = 2_000_000; // a healthy 2 ms LAN OWD
|
||||
const ELEVATED: i128 = FLOOR + STANDING_LAT_THRESH_NS + 7_000_000; // ~one 60fps frame above
|
||||
|
||||
/// Run `n` windows at `owd`, asserting every window but the last returns None; returns the
|
||||
/// last window's action.
|
||||
fn run_windows(d: &mut StandingLatency, owd: i128, n: u32) -> StandingLatAction {
|
||||
for i in 0..n {
|
||||
d.note_frame(owd);
|
||||
let a = d.on_window(true);
|
||||
if i + 1 < n {
|
||||
assert_eq!(a, StandingLatAction::None, "window {i} escalated early");
|
||||
} else {
|
||||
return a;
|
||||
}
|
||||
}
|
||||
unreachable!("n > 0 by construction");
|
||||
}
|
||||
|
||||
/// Learn a clean floor: one window at the healthy OWD.
|
||||
fn learned(d: &mut StandingLatency) {
|
||||
d.note_frame(FLOOR);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn healthy_stream_never_escalates() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
// Jitter riding above the floor but under the threshold: never a run.
|
||||
for _ in 0..(STANDING_LAT_WINDOWS * 4) {
|
||||
d.note_frame(FLOOR + STANDING_LAT_THRESH_NS - 1);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn escalation_ladder_resync_then_bleeds_then_disarm() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
// First full elevated run asks for the free fix: a clock re-sync.
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
// Re-sync didn't help (no rebase came) — each further run is a bleed, up to the cap...
|
||||
for _ in 0..STANDING_LAT_MAX_BLEEDS {
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Bleed { .. }
|
||||
));
|
||||
d.bled();
|
||||
}
|
||||
// ...then the detector gives up loudly, once, and stays quiet.
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Disarm { .. }
|
||||
));
|
||||
d.note_frame(ELEVATED);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn loss_windows_reset_the_run() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
for _ in 0..(STANDING_LAT_WINDOWS - 1) {
|
||||
d.note_frame(ELEVATED);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
// A lossy window means congestion, not a standing state: run resets...
|
||||
d.note_frame(ELEVATED);
|
||||
assert_eq!(d.on_window(false), StandingLatAction::None);
|
||||
// ...so the ladder needs the full run again before acting.
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_resets_the_ladder_to_resync_first() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
// The elevation clears on its own (e.g. the successful bleed case, or transient): the
|
||||
// next episode starts back at the free escalation, not at a bleed.
|
||||
d.note_frame(FLOOR);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn applied_resync_rebases_and_clears_a_stale_offset_elevation() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
assert!(matches!(
|
||||
run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
// The re-sync APPLIES (pump sees clock_gen move) → rebase. The corrected offset brings
|
||||
// OWD readings back to truth; the floor re-learns and nothing ever escalates to a bleed.
|
||||
d.rebase();
|
||||
for _ in 0..(STANDING_LAT_WINDOWS * 2) {
|
||||
d.note_frame(FLOOR);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_windows_are_no_evidence() {
|
||||
let mut d = StandingLatency::new();
|
||||
learned(&mut d);
|
||||
for _ in 0..(STANDING_LAT_WINDOWS - 1) {
|
||||
d.note_frame(ELEVATED);
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
}
|
||||
// A frameless window (paused stream) neither advances nor resets the run...
|
||||
assert_eq!(d.on_window(true), StandingLatAction::None);
|
||||
// ...so one more elevated window completes it.
|
||||
d.note_frame(ELEVATED);
|
||||
assert!(matches!(
|
||||
d.on_window(true),
|
||||
StandingLatAction::Resync { .. }
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
//! The client worker: QUIC handshake + control/input/datagram tasks + the blocking data-plane pump.
|
||||
|
||||
use super::frame_channel::{
|
||||
CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, FLUSH_LATENCY,
|
||||
NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_TIME,
|
||||
StandingLatAction, StandingLatency, CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN,
|
||||
FLUSH_LATENCY, NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW,
|
||||
STANDING_TIME,
|
||||
};
|
||||
use super::worker::reject_from_close;
|
||||
use super::*;
|
||||
@@ -514,15 +515,19 @@ pub(super) async fn run_pump(args: WorkerArgs) {
|
||||
// late exactly then) — keep the old estimate and let the next
|
||||
// periodic batch try again.
|
||||
if accept_resync(rtt_ns, clock_rtt_ns.unwrap_or(0)) {
|
||||
clock_offset.store(offset_ns, Ordering::Relaxed);
|
||||
clock_gen.fetch_add(1, Ordering::Relaxed);
|
||||
tracing::debug!(
|
||||
// info, not debug: ≤1/min, and it is THE forensic
|
||||
// trail for a stale-offset (stepped/slewed wall clock)
|
||||
// latency plateau — the 2026-07 two-pair investigation
|
||||
// had to reconstruct this blind.
|
||||
tracing::info!(
|
||||
offset_ns,
|
||||
rtt_us = rtt_ns / 1000,
|
||||
"mid-stream clock re-sync applied"
|
||||
);
|
||||
clock_offset.store(offset_ns, Ordering::Relaxed);
|
||||
clock_gen.fetch_add(1, Ordering::Relaxed);
|
||||
} else {
|
||||
tracing::debug!(
|
||||
tracing::info!(
|
||||
rtt_us = rtt_ns / 1000,
|
||||
"clock re-sync batch discarded — RTT above the \
|
||||
connect-time baseline (congested window)"
|
||||
@@ -729,6 +734,12 @@ pub(super) async fn run_pump(args: WorkerArgs) {
|
||||
let mut clock_detector_armed = true;
|
||||
let mut resync_wanted = false;
|
||||
let mut seen_clock_gen = pump_clock_gen.load(Ordering::Relaxed);
|
||||
// Standing-latency bleed (see StandingLatency): the third detector, for the small,
|
||||
// constant, loss-free OWD elevation the two jump-to-live detectors deliberately
|
||||
// tolerate (< QUEUE_HIGH frames, < FLUSH_LATENCY behind) — a sub-frame standing
|
||||
// backlog, or a stale clock offset after a wall-clock step, either of which otherwise
|
||||
// reads as permanent extra "network" latency for the rest of the session.
|
||||
let mut standing_lat = StandingLatency::new();
|
||||
while !pump_shutdown.load(Ordering::SeqCst) {
|
||||
// The live host↔client offset: re-loaded every iteration so an applied mid-stream
|
||||
// re-sync takes effect on the very next frame's latency math.
|
||||
@@ -740,6 +751,10 @@ pub(super) async fn run_pump(args: WorkerArgs) {
|
||||
seen_clock_gen = gen;
|
||||
stale_since = None;
|
||||
noop_clock_flushes = 0;
|
||||
// Every OWD reading shifted with the offset — the standing-latency floor and
|
||||
// any elevation measured under the old one are meaningless now. If a stale
|
||||
// offset WAS the elevation, this is also the moment it gets fixed.
|
||||
standing_lat.rebase();
|
||||
if !clock_detector_armed {
|
||||
clock_detector_armed = true;
|
||||
tracing::info!(
|
||||
@@ -843,6 +858,51 @@ pub(super) async fn run_pump(args: WorkerArgs) {
|
||||
window_dropped,
|
||||
);
|
||||
let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm }));
|
||||
// Standing-latency bleed: close the detector's window with this report's loss
|
||||
// verdict and run its escalation ladder — re-sync first (free; a stale offset
|
||||
// from a stepped wall clock produces exactly this signature and the applied
|
||||
// re-sync rebases the floor), then a bounded flush+keyframe (drains a real
|
||||
// sub-threshold standing backlog the jump-to-live thresholds tolerate), then a
|
||||
// loud disarm (the path latency itself changed; nothing local fixes that).
|
||||
match standing_lat.on_window(loss_ppm == 0 && window_dropped == 0) {
|
||||
StandingLatAction::None => {}
|
||||
StandingLatAction::Resync { above_ms } => {
|
||||
tracing::info!(
|
||||
above_ms,
|
||||
"standing latency above the session floor with zero loss — \
|
||||
requesting a clock re-sync first (a stale offset reads exactly \
|
||||
like this)"
|
||||
);
|
||||
let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
|
||||
}
|
||||
StandingLatAction::Bleed { above_ms } => {
|
||||
// Shares the jump-to-live cooldown: an unexecuted bleed simply re-arms
|
||||
// over the next windows (the detector's run rebuilds).
|
||||
if last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) {
|
||||
last_flush = Some(Instant::now());
|
||||
flush_in_window = true;
|
||||
let flushed = session.flush_backlog().unwrap_or(0);
|
||||
let dropped = frames.clear();
|
||||
let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
|
||||
standing_lat.bled();
|
||||
tracing::warn!(
|
||||
above_ms,
|
||||
flushed_datagrams = flushed,
|
||||
dropped_frames = dropped,
|
||||
"standing latency survived a clock re-sync — bled the local \
|
||||
backlog (flush + keyframe)"
|
||||
);
|
||||
}
|
||||
}
|
||||
StandingLatAction::Disarm { above_ms } => {
|
||||
tracing::warn!(
|
||||
above_ms,
|
||||
"standing latency persists after a re-sync and every bleed — not \
|
||||
local, not clock; the path latency changed. Leaving it be \
|
||||
(reconnect re-baselines)"
|
||||
);
|
||||
}
|
||||
}
|
||||
// Adaptive bitrate: drain any host ack first (its clamp is authoritative), then
|
||||
// feed the controller this window's congestion signals; a decision becomes a
|
||||
// SetBitrate on the control stream.
|
||||
@@ -981,6 +1041,13 @@ pub(super) async fn run_pump(args: WorkerArgs) {
|
||||
if clock_offset_ns != 0 && lat_ns > 0 {
|
||||
owd_sum_ns += lat_ns;
|
||||
owd_frames += 1;
|
||||
// The standing-latency detector rides the same signal, but off the
|
||||
// window MINIMUM (robust against jitter/burst spikes — a standing
|
||||
// state elevates the floor itself). Same 10 s plausibility clamp as
|
||||
// the hn stats use.
|
||||
if lat_ns < 10_000_000_000 {
|
||||
standing_lat.note_frame(lat_ns);
|
||||
}
|
||||
}
|
||||
if clock_detector_armed
|
||||
&& clock_offset_ns != 0
|
||||
|
||||
@@ -83,7 +83,12 @@ pub use stats::Stats;
|
||||
/// `punktfunk_connection_clipboard_{control,offer,fetch,serve,cancel}` +
|
||||
/// `punktfunk_connection_next_clipboard`. Additive; the wire grows only backward-compatible control
|
||||
/// messages (0x40-0x44) and a new `Welcome::host_caps` bit, so [`WIRE_VERSION`] is unchanged.
|
||||
pub const ABI_VERSION: u32 = 8;
|
||||
/// v9: `PunktfunkFrame` grew `received_ns` — the reassembly-completion receipt stamp, so
|
||||
/// embedders stop stamping receipt at the hand-off pull (which folds the pre-decode queue wait
|
||||
/// into apparent network latency). Struct-size change on the frame poll surface = a hard ABI
|
||||
/// break for embedders reading `PunktfunkFrame`; nothing on the wire moved, so [`WIRE_VERSION`]
|
||||
/// is unchanged.
|
||||
pub const ABI_VERSION: u32 = 9;
|
||||
|
||||
/// The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check.
|
||||
/// Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface**
|
||||
|
||||
@@ -489,6 +489,7 @@ impl Reassembler {
|
||||
pts_ns: done.pts_ns,
|
||||
flags: done.user_flags,
|
||||
complete: true,
|
||||
received_ns: 0, // stamped by Session::poll_frame at the session boundary
|
||||
}));
|
||||
}
|
||||
Ok(None)
|
||||
@@ -592,6 +593,7 @@ impl ReassemblyWindow {
|
||||
pts_ns: f.pts_ns,
|
||||
flags: f.user_flags,
|
||||
complete: false,
|
||||
received_ns: 0, // stamped by Session::poll_frame at the session boundary
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -30,6 +30,14 @@ pub struct Frame {
|
||||
/// ([`crate::packet::USER_FLAG_CHUNK_ALIGNED`]) are ever delivered partial; missing
|
||||
/// shard ranges are zero-filled at their exact offsets.
|
||||
pub complete: bool,
|
||||
/// Wall-clock instant (ns since the Unix epoch, CLOCK_REALTIME basis — the same clock the
|
||||
/// skew handshake compares and the host stamps `pts_ns` with) at which this AU finished
|
||||
/// reassembly, stamped by [`Session::poll_frame`] as the frame leaves the session. Embedders
|
||||
/// that previously stamped receipt themselves at the hand-off pull should use this instead:
|
||||
/// the pull stamp additionally contains the pre-decode queue wait, silently folding any
|
||||
/// client-side standing backlog into the apparent NETWORK latency. The reassembler itself
|
||||
/// leaves this 0 (it owns no clock — the stamp is the session boundary's job).
|
||||
pub received_ns: u64,
|
||||
}
|
||||
|
||||
/// One end of a stream. Constructed for a single [`Role`]; calling the other role's
|
||||
@@ -82,6 +90,18 @@ pub struct Session {
|
||||
lane_scratch: Vec<Vec<u8>>,
|
||||
}
|
||||
|
||||
/// Stamp [`Frame::received_ns`] as the frame crosses the session boundary in
|
||||
/// [`Session::poll_frame`] — completed frames return the moment their last shard lands, so
|
||||
/// stamping at return IS stamping at reassembly completion (µs apart). CLOCK_REALTIME to match
|
||||
/// `pts_ns` / the skew handshake (deliberately not monotonic — cross-machine latency math).
|
||||
fn stamp_received(mut f: Frame) -> Frame {
|
||||
f.received_ns = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_nanos() as u64)
|
||||
.unwrap_or(0);
|
||||
f
|
||||
}
|
||||
|
||||
/// Wire-packet count at which a frame's sealing splits across two lanes (plan Phase 1.5):
|
||||
/// below it the channel rendezvous (~µs) isn't worth it; at it the halved AES-GCM span
|
||||
/// (≥ ~125 µs of ~1 µs/packet work) dwarfs the hand-off. ≈300 KB of wire, i.e. ≥150 Mbps
|
||||
@@ -684,7 +704,7 @@ impl Session {
|
||||
// Nothing new on the wire — hand over an aged-out partial if one is
|
||||
// waiting (it can only get staler).
|
||||
if let Some(p) = self.reassembler.take_partial() {
|
||||
return Ok(p);
|
||||
return Ok(stamp_received(p));
|
||||
}
|
||||
return Err(PunktfunkError::NoFrame);
|
||||
}
|
||||
@@ -748,12 +768,12 @@ impl Session {
|
||||
}
|
||||
if let Some(frame) = pushed {
|
||||
StatsCounters::add(&self.stats.frames_completed, 1);
|
||||
return Ok(frame);
|
||||
return Ok(stamp_received(frame));
|
||||
}
|
||||
// A push that completed nothing may still have aged a partial out — deliver it
|
||||
// ahead of further draining (its successors are already arriving).
|
||||
if let Some(p) = self.reassembler.take_partial() {
|
||||
return Ok(p);
|
||||
return Ok(stamp_received(p));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,7 +37,12 @@
|
||||
// `punktfunk_connection_clipboard_{control,offer,fetch,serve,cancel}` +
|
||||
// `punktfunk_connection_next_clipboard`. Additive; the wire grows only backward-compatible control
|
||||
// messages (0x40-0x44) and a new `Welcome::host_caps` bit, so [`WIRE_VERSION`] is unchanged.
|
||||
#define ABI_VERSION 8
|
||||
// v9: `PunktfunkFrame` grew `received_ns` — the reassembly-completion receipt stamp, so
|
||||
// embedders stop stamping receipt at the hand-off pull (which folds the pre-decode queue wait
|
||||
// into apparent network latency). Struct-size change on the frame poll surface = a hard ABI
|
||||
// break for embedders reading `PunktfunkFrame`; nothing on the wire moved, so [`WIRE_VERSION`]
|
||||
// is unchanged.
|
||||
#define ABI_VERSION 9
|
||||
|
||||
// The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check.
|
||||
// Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface**
|
||||
@@ -1112,6 +1117,12 @@ typedef struct {
|
||||
uint32_t frame_index;
|
||||
uint64_t pts_ns;
|
||||
uint32_t flags;
|
||||
// Wall-clock reassembly-completion instant (ns since the Unix epoch, CLOCK_REALTIME — the
|
||||
// clock `pts_ns` and the skew handshake use). THIS is the receipt stamp for latency math:
|
||||
// a stamp the embedder takes itself at the poll return additionally contains the
|
||||
// pre-decode hand-off queue wait, so a client-side standing backlog would masquerade as
|
||||
// network latency (ABI v9 — the 2026-07 two-pair standing-latency investigation).
|
||||
uint64_t received_ns;
|
||||
} PunktfunkFrame;
|
||||
|
||||
// A single input event. `#[repr(C)]` — shared verbatim with the C ABI as
|
||||
|
||||
Reference in New Issue
Block a user