perf(host): session-transition trace + Welcome-time display prep (native path)

Latency plan P0.1 + P1.1/P1.2 (design/first-frame-and-resize-latency.md):

P0.1 — every native session runs a bringup::Trace (hello -> welcome -> start
-> punch_done -> display_acquired -> capture_attached -> first_frame ->
encoder_open -> first_au -> first_packet), one summary info! line when the
first video packet leaves; each accepted resize runs its own trace
(reconfigure -> pipeline_rebuilt). Totals surface per session as
time_to_first_frame_ms / last_resize_ms in session_status -> mgmt /status,
so every subsequent latency change is measured, not vibed. (The Windows
manager logs its own activation/settle deltas — correlate by wall clock.)

P1.1/P1.2 — on the Windows native path the display bring-up no longer
serializes behind the Start round-trip and the up-to-2.5 s hole-punch wait:
a prep thread kicks off at Welcome (mode is final there) and runs monitor
create -> activation -> verified settle -> capture attach -> first frame ->
encoder open while the network waits are in flight; the data plane hands it
the post-punch SessionContext and it becomes the stream thread on a warm
pipeline. Abort between Welcome and Start drops the hand-off channel and the
prep result releases into the keep-alive machinery (stop/quit + watcher are
created pre-handshake so a vanished client also aborts the build retries).
Same slot-scoped begin_idd_setup serialization as the inline path. Linux
keeps the inline bring-up (launch semantics bind before create); GameStream
untouched.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-16 16:45:34 +02:00
parent e62cd5448e
commit 8374dfedf3
5 changed files with 451 additions and 74 deletions
+89
View File
@@ -0,0 +1,89 @@
//! Session-transition latency trace (design/first-frame-and-resize-latency.md P0.1).
//!
//! One [`Trace`] per transition — session bring-up (`hello → … → first_packet`) or a mid-stream
//! resize (`reconfigure_received → … → pipeline_rebuilt`) — collects millisecond stage stamps
//! across the threads a transition crosses (handshake task, display-prep/encode thread, send
//! thread) and emits ONE summary `info!` line when the transition completes, so every landed
//! latency change is measured against a number instead of vibes. The completed total also lands
//! in a shared slot [`crate::session_status`] exposes (`time_to_first_frame_ms` /
//! `last_resize_ms`), so the web-console Dashboard and future regressions can read it per session.
//!
//! Deliberately coarse: stages are stamped where the session layer can see them; layers the trace
//! doesn't reach (the Windows display manager's activation ladder / settle waits) log their own
//! per-stage deltas and correlate by wall clock.
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
/// A single transition's stage trace. Cheap and thread-safe: `mark` is a mutex push, `finish`
/// emits the one summary line (exactly once — later calls no-op, so an abandoned trace stays
/// silent).
pub(crate) struct Trace {
/// Which transition this traces (`"bringup"` / `"resize"`) — the summary line's `kind`.
kind: &'static str,
origin: Instant,
/// `(stage, ms since origin)` in stamp order.
stages: Mutex<Vec<(&'static str, u32)>>,
finished: AtomicBool,
/// Where the completed total lands — shared with [`crate::session_status`].
total_ms: Arc<AtomicU32>,
}
impl Trace {
/// Start a trace at "now" (= the first stage's zero point). `total_ms` is the shared slot the
/// completed total is stored into (0 until the transition finishes).
pub(crate) fn start(kind: &'static str, total_ms: Arc<AtomicU32>) -> Arc<Self> {
Arc::new(Self {
kind,
origin: Instant::now(),
stages: Mutex::new(Vec::new()),
finished: AtomicBool::new(false),
total_ms,
})
}
/// The shared slot the completed total is stored into (for `session_status::register`).
pub(crate) fn total_slot(&self) -> Arc<AtomicU32> {
self.total_ms.clone()
}
/// Stamp a stage at "now" — first occurrence only (a retried build re-crosses its stamp
/// points; the first crossing is the one the transition timeline wants). No-op after
/// [`finish`](Self::finish), so steady-state paths that also cross a stamped point stay free.
pub(crate) fn mark(&self, stage: &'static str) {
if self.finished.load(Ordering::Relaxed) {
return;
}
let ms = self.origin.elapsed().as_millis().min(u32::MAX as u128) as u32;
let mut stages = self.stages.lock().unwrap();
if stages.iter().any(|(s, _)| *s == stage) {
return;
}
stages.push((stage, ms));
}
/// Stamp the final stage and emit the one-line summary (first call only). The final stage's
/// offset is the transition total, stored into the shared slot.
pub(crate) fn finish(&self, stage: &'static str) {
if self.finished.swap(true, Ordering::Relaxed) {
return;
}
let total = self.origin.elapsed().as_millis().min(u32::MAX as u128) as u32;
let mut stages = self.stages.lock().unwrap();
stages.push((stage, total));
let line = stages
.iter()
.map(|(s, ms)| format!("{s}+{ms}"))
.collect::<Vec<_>>()
.join(" ");
drop(stages);
self.total_ms.store(total.max(1), Ordering::Relaxed);
tracing::info!(
kind = self.kind,
total_ms = total,
stages = %line,
"session-transition trace"
);
}
}
+1
View File
@@ -19,6 +19,7 @@
#![deny(clippy::undocumented_unsafe_blocks)] #![deny(clippy::undocumented_unsafe_blocks)]
mod audio; mod audio;
mod bringup;
mod capture; mod capture;
mod config; mod config;
mod detect; mod detect;
+12
View File
@@ -391,6 +391,12 @@ struct StreamInfo {
/// Client's parity floor per FEC block (`minRequiredFecPackets`). /// Client's parity floor per FEC block (`minRequiredFecPackets`).
min_fec: u8, min_fec: u8,
codec: ApiCodec, codec: ApiCodec,
/// Session bring-up total, hello → first video packet, in ms (native sessions; `null` on the
/// GameStream plane or while the session is still bringing up).
time_to_first_frame_ms: Option<u32>,
/// Most recent mid-stream resize total, reconfigure → pipeline rebuilt, in ms (native sessions;
/// `null` when no resize happened / GameStream).
last_resize_ms: Option<u32>,
} }
/// Non-sensitive host status for the local tray icon: counts and booleans only — no PIN values, /// Non-sensitive host status for the local tray icon: counts and booleans only — no PIN values,
@@ -1480,6 +1486,9 @@ async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> {
packet_size: c.packet_size as u32, packet_size: c.packet_size as u32,
min_fec: c.min_fec, min_fec: c.min_fec,
codec: c.codec.into(), codec: c.codec.into(),
// Transition latencies are traced on the native plane only (latency plan P0.1).
time_to_first_frame_ms: None,
last_resize_ms: None,
}) })
.or_else(|| { .or_else(|| {
native.first().map(|s| StreamInfo { native.first().map(|s| StreamInfo {
@@ -1492,6 +1501,9 @@ async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> {
packet_size: 0, packet_size: 0,
min_fec: 0, min_fec: 0,
codec: s.codec.into(), codec: s.codec.into(),
time_to_first_frame_ms: (s.time_to_first_frame_ms > 0)
.then_some(s.time_to_first_frame_ms),
last_resize_ms: (s.last_resize_ms > 0).then_some(s.last_resize_ms),
}) })
}); });
Json(RuntimeStatus { Json(RuntimeStatus {
+321 -61
View File
@@ -815,6 +815,38 @@ async fn serve_session(
let source = opts.source; let source = opts.source;
let frames = opts.frames; let frames = opts.frames;
let data_port = opts.data_port; let data_port = opts.data_port;
// Session-transition trace (latency plan P0.1): zeroed here — the Hello is in hand, pairing
// gates are behind us — and finished by the send thread when the FIRST video packet leaves.
// The completed totals surface per session in `session_status` (→ mgmt `/status`).
let bringup = crate::bringup::Trace::start("bringup", Arc::new(AtomicU32::new(0)));
// The mid-stream resize counterpart: each accepted Reconfigure runs its own trace into this
// shared slot (latest wins), registered alongside the bring-up total.
let resize_ms: Arc<AtomicU32> = Arc::new(AtomicU32::new(0));
// Stop signal: stream duration elapsed or the client went away. Created (with its watcher)
// BEFORE the handshake so the Welcome-time display prep below can already observe a client
// that vanished mid-handshake (its build-retry loop aborts on `stop`).
let stop = Arc::new(AtomicBool::new(false));
// Deliberate-quit signal: set (before `stop`, so the display lease reads it on teardown) when
// the client closed the connection with `QUIT_CODE` — a user "stop", which skips the
// keep-alive linger. A bare disconnect / idle timeout leaves it false → the display lingers
// for a reconnect.
let quit = Arc::new(AtomicBool::new(false));
{
let stop = stop.clone();
let quit = quit.clone();
let conn = conn.clone();
tokio::spawn(async move {
let reason = conn.closed().await;
if matches!(&reason, quinn::ConnectionError::ApplicationClosed(ac)
if ac.error_code == quinn::VarInt::from_u32(QUIT_CODE))
{
quit.store(true, Ordering::SeqCst);
}
stop.store(true, Ordering::SeqCst);
});
}
let handshake = async { let handshake = async {
let mut hello = Hello::decode(&first).map_err(|e| anyhow!("Hello decode: {e:?}"))?; let mut hello = Hello::decode(&first).map_err(|e| anyhow!("Hello decode: {e:?}"))?;
if hello.abi_version != punktfunk_core::WIRE_VERSION { if hello.abi_version != punktfunk_core::WIRE_VERSION {
@@ -1149,14 +1181,76 @@ async fn serve_session(
host_caps: punktfunk_core::quic::HOST_CAP_GAMEPAD_STATE, host_caps: punktfunk_core::quic::HOST_CAP_GAMEPAD_STATE,
}; };
io::write_msg(&mut send, &welcome.encode()).await?; io::write_msg(&mut send, &welcome.encode()).await?;
bringup.mark("welcome");
// P1.1/P1.2 (latency plan): kick the display prep NOW — the negotiated mode is final in
// the Welcome just sent, and nothing in monitor create → activation → settle → capture
// attach → encoder open needs the client's Start or the punched socket. The prep thread
// BECOMES the stream thread: the data plane hands it the post-punch SessionContext and it
// runs `virtual_stream` on the warm pipeline, so the whole display bring-up hides behind
// the Start RTT + the (up to 2.5 s) hole-punch wait. If the session dies before its data
// plane comes up (handshake timeout, client vanished), the channel drops and the prep
// result is released — the monitor lands in the keep-alive machinery exactly like a
// normal session end (and `stop`, watched above, aborts a still-running build retry).
// Windows native path only: the Linux backends bind launch semantics before create
// (gamescope nests the launch command), which must not run for a client that never
// sends Start; GameStream has neither a Start gate nor a punch.
#[cfg(target_os = "windows")]
let prep: Option<PrepHandle> = match (source, compositor) {
(Punktfunk1Source::Virtual, Some(comp)) => {
let (ctx_tx, ctx_rx) = std::sync::mpsc::sync_channel::<SessionContext>(1);
let client_identity = endpoint::peer_fingerprint(&conn);
let client_hdr = hello.display_hdr;
let (mode, shard_payload) = (hello.mode, welcome.shard_payload);
let (quit, stop, trace) = (quit.clone(), stop.clone(), bringup.clone());
std::thread::Builder::new()
.name("punktfunk1-stream".into())
.spawn(move || -> Result<()> {
let prepared = prepare_display(
comp,
mode,
client_identity,
client_hdr,
bitrate_kbps,
bit_depth,
chroma,
codec,
shard_payload,
&quit,
&stop,
&trace,
);
let Ok(ctx) = ctx_rx.recv() else {
// No data plane ever came (handshake abort / punch failure): drop
// `prepared` — its lease release hands the monitor to keep-alive
// policy, exactly like a normal session end.
return Ok(());
};
match prepared {
Ok(p) => virtual_stream(ctx, Some(p)),
Err(e) => Err(e),
}
})
.map(|handle| (ctx_tx, handle))
.map_err(|e| {
tracing::warn!(error = %e,
"display-prep thread spawn failed — falling back to inline bring-up")
})
.ok()
}
_ => None,
};
#[cfg(not(target_os = "windows"))]
let prep: Option<PrepHandle> = None;
let start = Start::decode(&io::read_msg(&mut recv).await?) let start = Start::decode(&io::read_msg(&mut recv).await?)
.map_err(|e| anyhow!("Start decode: {e:?}"))?; .map_err(|e| anyhow!("Start decode: {e:?}"))?;
bringup.mark("start");
Ok::<_, anyhow::Error>(( Ok::<_, anyhow::Error>((
hello, welcome, udp_port, data_sock, direct, start, compositor, hello, welcome, udp_port, data_sock, direct, start, compositor, prep,
)) ))
}; };
let (hello, welcome, udp_port, data_sock, direct, start, compositor) = let (hello, welcome, udp_port, data_sock, direct, start, compositor, prep) =
tokio::time::timeout(HANDSHAKE_TIMEOUT, handshake) tokio::time::timeout(HANDSHAKE_TIMEOUT, handshake)
.await .await
.map_err(|_| anyhow!("handshake timed out after {HANDSHAKE_TIMEOUT:?}"))??; .map_err(|_| anyhow!("handshake timed out after {HANDSHAKE_TIMEOUT:?}"))??;
@@ -1467,26 +1561,8 @@ async fn serve_session(
); );
}); });
// Stop signal: stream duration elapsed or the client went away. // (The stop/quit flags + their disconnect watcher are created above, before the handshake, so
let stop = Arc::new(AtomicBool::new(false)); // the Welcome-time display prep can observe a mid-handshake disconnect.)
// Deliberate-quit signal: set (before `stop`, so the display lease reads it on teardown) when the
// client closed the connection with `QUIT_CODE` — a user "stop", which skips the keep-alive linger.
// A bare disconnect / idle timeout leaves it false → the display lingers for a reconnect.
let quit = Arc::new(AtomicBool::new(false));
{
let stop = stop.clone();
let quit = quit.clone();
let conn = conn.clone();
tokio::spawn(async move {
let reason = conn.closed().await;
if matches!(&reason, quinn::ConnectionError::ApplicationClosed(ac)
if ac.error_code == quinn::VarInt::from_u32(QUIT_CODE))
{
quit.store(true, Ordering::SeqCst);
}
stop.store(true, Ordering::SeqCst);
});
}
// Register this now-live session for mode-conflict admission (Stage 4): carry its identity, the // Register this now-live session for mode-conflict admission (Stage 4): carry its identity, the
// negotiated mode, and its stop flag so a LATER connecting client's admission can see it and // negotiated mode, and its stop flag so a LATER connecting client's admission can see it and
@@ -1655,6 +1731,10 @@ async fn serve_session(
let client_label = endpoint::peer_fingerprint(&conn) let client_label = endpoint::peer_fingerprint(&conn)
.map(|fp| fingerprint_hex(&fp)[..12].to_string()) .map(|fp| fingerprint_hex(&fp)[..12].to_string())
.unwrap_or_else(|| conn.remote_address().ip().to_string()); .unwrap_or_else(|| conn.remote_address().ip().to_string());
// Transition-trace handles for the data plane (P0.1): the punch stamp + the virtual-stream
// stages ride the same per-session trace; resizes write their totals into the shared slot.
let bringup_dp = bringup.clone();
let resize_ms_dp = resize_ms.clone();
let result: Result<()> = async { let result: Result<()> = async {
tokio::task::spawn_blocking(move || -> Result<()> { tokio::task::spawn_blocking(move || -> Result<()> {
// Bring up the (already-bound) data-plane socket. Default: hole-punch — wait briefly // Bring up the (already-bound) data-plane socket. Default: hole-punch — wait briefly
@@ -1684,6 +1764,7 @@ async fn serve_session(
return Err(anyhow::Error::new(e)).context("bind data plane"); return Err(anyhow::Error::new(e)).context("bind data plane");
} }
}; };
bringup_dp.mark("punch_done");
tracing::info!( tracing::info!(
%client_udp, %client_udp,
udp_port, udp_port,
@@ -1709,7 +1790,7 @@ async fn serve_session(
Punktfunk1Source::Virtual => { Punktfunk1Source::Virtual => {
let compositor = compositor let compositor = compositor
.expect("the Virtual source resolves a compositor during the handshake"); .expect("the Virtual source resolves a compositor during the handshake");
virtual_stream(SessionContext { let ctx = SessionContext {
session, session,
mode, mode,
seconds, seconds,
@@ -1736,7 +1817,29 @@ async fn serve_session(
client_label, client_label,
launch: launch_for_dp, launch: launch_for_dp,
client_hdr, client_hdr,
}) bringup: bringup_dp,
resize_ms: resize_ms_dp,
};
match prep {
// P1.1: the display prep started at Welcome on its own thread — hand it
// the post-punch context and adopt its result as the stream result (that
// thread runs `virtual_stream` on the pipeline it already built).
Some((ctx_tx, prep_thread)) => match ctx_tx.send(ctx) {
Ok(()) => match prep_thread.join() {
Ok(r) => r,
Err(_) => Err(anyhow!("prepared stream thread panicked")),
},
// The prep thread died before the hand-off (panicked during prep —
// its guard/lease unwound): run the stream inline instead.
Err(std::sync::mpsc::SendError(ctx)) => {
tracing::warn!(
"display-prep thread gone before hand-off — building inline"
);
virtual_stream(ctx, None)
}
},
None => virtual_stream(ctx, None),
}
} }
} }
}) })
@@ -3588,6 +3691,10 @@ struct SendStats {
/// Live encoder bitrate (kbps) — the capture thread updates it on a mid-stream adaptive /// Live encoder bitrate (kbps) — the capture thread updates it on a mid-stream adaptive
/// bitrate change, so the web-console sample reports what the encoder is ACTUALLY targeting. /// bitrate change, so the web-console sample reports what the encoder is ACTUALLY targeting.
bitrate_kbps: Arc<AtomicU32>, bitrate_kbps: Arc<AtomicU32>,
/// The session's bring-up trace (P0.1): the send thread FINISHES it — `first_packet` — the
/// moment the first video AU's packets have fully left the socket (finish is once-only, so
/// the per-frame call is a cheap no-op afterwards).
bringup: Arc<crate::bringup::Trace>,
} }
/// Pack a `(width, height, refresh_hz)` mode into one atomic word (w:16|h:16|hz:16) for the live /// Pack a `(width, height, refresh_hz)` mode into one atomic word (w:16|h:16|hz:16) for the live
@@ -3714,6 +3821,11 @@ fn send_loop(
burst_cap, burst_cap,
) { ) {
Ok(stat) => { Ok(stat) => {
// First VIDEO packets are on the wire — complete the bring-up trace (P0.1;
// once-only, no-op on every later frame). Speed-test filler isn't video.
if msg.flags & FLAG_PROBE as u32 == 0 {
stats.bringup.finish("first_packet");
}
// Host timing (0xCF): stamped now — the AU's packets have fully left the // Host timing (0xCF): stamped now — the AU's packets have fully left the
// socket — against the same capture anchor the wire pts carries, so the // socket — against the same capture anchor the wire pts carries, so the
// client's per-frame math tiles exactly (network = its host+network this). // client's per-frame math tiles exactly (network = its host+network this).
@@ -4068,9 +4180,15 @@ struct SessionContext {
/// so host apps tone-map to the client's real panel) and preferred over the generic baseline /// so host apps tone-map to the client's real panel) and preferred over the generic baseline
/// for the 0xCE mastering metadata. /// for the 0xCE mastering metadata.
client_hdr: Option<punktfunk_core::quic::HdrMeta>, client_hdr: Option<punktfunk_core::quic::HdrMeta>,
/// The session's bring-up trace (latency plan P0.1): the pipeline-build stages stamp into it
/// and the send thread finishes it when the first video packet leaves.
bringup: Arc<crate::bringup::Trace>,
/// Shared slot the latest completed mid-stream resize total (ms) lands in — registered with
/// `session_status` so the Dashboard shows it.
resize_ms: Arc<AtomicU32>,
} }
fn virtual_stream(ctx: SessionContext) -> Result<()> { fn virtual_stream(ctx: SessionContext, prepared: Option<PreparedDisplay>) -> Result<()> {
// This thread runs the capture+encode loop (single-process — the only topology: Linux portal / // This thread runs the capture+encode loop (single-process — the only topology: Linux portal /
// synthetic, Windows in-process IDD-push). Elevate it so a CPU-heavy game can't deschedule our GPU // synthetic, Windows in-process IDD-push). Elevate it so a CPU-heavy game can't deschedule our GPU
// submission. // submission.
@@ -4117,6 +4235,8 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
client_label, client_label,
launch, launch,
client_hdr, client_hdr,
bringup,
resize_ms,
} = ctx; } = ctx;
tracing::info!( tracing::info!(
compositor = compositor.id(), compositor = compositor.id(),
@@ -4125,54 +4245,79 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
bit_depth, bit_depth,
"punktfunk/1 virtual display" "punktfunk/1 virtual display"
); );
// Open the backend FIRST — on Windows this constructs the vdisplay backend, which initialises the // The vdisplay backend + built pipeline: either PREPARED at Welcome time on this very thread
// host-lifetime VirtualDisplayManager (§2.5). It does NO monitor work, so it must precede the IDD-push // (P1.1/P1.2 — the display bring-up already overlapped the Start RTT + hole-punch), or built
// preempt below (which reaches the manager) — otherwise `vdm()` is called before init and panics. // inline now (Linux, synthetic-adjacent paths, prep fallback).
let (mut vd, pipe) = match prepared {
Some(p) => (p.vd, p.pipeline),
None => {
// Open the backend FIRST — on Windows this constructs the vdisplay backend, which
// initialises the host-lifetime VirtualDisplayManager (§2.5). It does NO monitor work,
// so it must precede the IDD-push preempt below (which reaches the manager) —
// otherwise `vdm()` is called before init and panics.
let mut vd = crate::vdisplay::open(compositor)?; let mut vd = crate::vdisplay::open(compositor)?;
// Per-client STABLE monitor identity (Phase 2): hand the backend the connecting client's cert // Per-client STABLE monitor identity (Phase 2): hand the backend the connecting
// fingerprint so a freshly CREATED virtual monitor gets this client's persistent id — Windows then // client's cert fingerprint so a freshly CREATED virtual monitor gets this client's
// reapplies the client's saved per-monitor config (DPI scaling) on reconnect. No-op on Linux backends // persistent id — Windows then reapplies the client's saved per-monitor config (DPI
// and for anonymous/GameStream clients (no fingerprint → the driver auto-allocates). // scaling) on reconnect. No-op on Linux backends and for anonymous/GameStream clients
// (no fingerprint → the driver auto-allocates).
vd.set_client_identity(endpoint::peer_fingerprint(&conn)); vd.set_client_identity(endpoint::peer_fingerprint(&conn));
// The client display's HDR volume (Hello) → a freshly created virtual monitor's EDID CTA HDR // The client display's HDR volume (Hello) → a freshly created virtual monitor's EDID
// block (pf-vdisplay), so host apps + the OS tone-map to the client's real panel instead of the // CTA HDR block (pf-vdisplay), so host apps + the OS tone-map to the client's real
// driver's built-in ~1000-nit placeholder. No-op on Linux backends and for older/SDR clients. // panel instead of the driver's built-in ~1000-nit placeholder. No-op on Linux
// backends and for older/SDR clients.
vd.set_client_hdr(client_hdr); vd.set_client_hdr(client_hdr);
// Deliberate-quit wiring (Windows pf-vdisplay; no-op elsewhere): every lease the backend mints — // Deliberate-quit wiring (Windows pf-vdisplay; no-op elsewhere): every lease the
// the retry-hold below AND the capturer's — carries the session's quit flag, so a user "stop" // backend mints — the retry-hold below AND the capturer's — carries the session's quit
// (⌘D → the QUIT close code) tears the virtual monitor down the moment the pipeline drops instead // flag, so a user "stop" (⌘D → the QUIT close code) tears the virtual monitor down the
// of lingering 10 s. The reconnect then finds the manager Idle and does a clean fresh ADD (with // moment the pipeline drops instead of lingering 10 s. The reconnect then finds the
// the user's think-time as driver settle) rather than the Lingering-preempt's REMOVE→ADD churn. // manager Idle and does a clean fresh ADD (with the user's think-time as driver
// `keep_alive = forever` (gaming-rig) outranks the quit — the monitor pins as before. // settle) rather than the Lingering-preempt's REMOVE→ADD churn. `keep_alive = forever`
// (gaming-rig) outranks the quit — the monitor pins as before.
vd.set_quit_flag(quit.clone()); vd.set_quit_flag(quit.clone());
// Per-session launch (non-Windows): hand the resolved command to the backend instance so // Per-session launch (non-Windows): hand the resolved command to the backend instance
// gamescope's bare spawn nests it — per-instance, no process-global env, so concurrent sessions // so gamescope's bare spawn nests it — per-instance, no process-global env, so
// can't stomp each other's launch target. The other backends' default `set_launch_command` is a // concurrent sessions can't stomp each other's launch target. The other backends'
// no-op; they get the command spawned into the live session after capture is up (below). // default `set_launch_command` is a no-op; they get the command spawned into the live
// session after capture is up (below).
#[cfg(not(target_os = "windows"))] #[cfg(not(target_os = "windows"))]
vd.set_launch_command(launch.clone()); vd.set_launch_command(launch.clone());
// IDD-push reconnect preempt (the dance now lives in the manager, Goal-1 §2.5): serialize setup so a // IDD-push reconnect preempt (the dance now lives in the manager, Goal-1 §2.5):
// reconnect FLOOD can't run concurrent monitor create/teardown, STOP the prior session + WAIT for it // serialize setup so a reconnect FLOOD can't run concurrent monitor create/teardown,
// to release its monitor (instead of tearing a monitor out from under a still-live session), and // STOP the prior session + WAIT for it to release its monitor (instead of tearing a
// register THIS session's stop. The returned guard holds the setup lock across the pipeline build; // monitor out from under a still-live session), and register THIS session's stop. The
// dropping it lets the next reconnect begin (and preempt us). Held BEFORE the monitor is created // returned guard holds the setup lock across the pipeline build; dropping it (end of
// (build_pipeline → vd.create), so the preempt still precedes this session's monitor creation. // this arm) lets the next reconnect begin (and preempt us). Held BEFORE the monitor is
// SLOT-scoped (Stage W1): the preempt targets only a prior session holding THIS client's slot — // created (build_pipeline → vd.create), so the preempt still precedes this session's
// a different identity's session is an admission question, never a preempt. // monitor creation. SLOT-scoped (Stage W1): the preempt targets only a prior session
// holding THIS client's slot — a different identity's session is an admission
// question, never a preempt.
#[cfg(target_os = "windows")] #[cfg(target_os = "windows")]
let _idd_setup_guard = let _idd_setup_guard = (plan.capture == crate::session_plan::CaptureBackend::IddPush)
(plan.capture == crate::session_plan::CaptureBackend::IddPush).then(|| { .then(|| {
let slot = crate::vdisplay::manager::slot_id_for( let slot = crate::vdisplay::manager::slot_id_for(
endpoint::peer_fingerprint(&conn), endpoint::peer_fingerprint(&conn),
(mode.width, mode.height), (mode.width, mode.height),
); );
crate::vdisplay::manager::vdm().begin_idd_setup(slot, stop.clone()) crate::vdisplay::manager::vdm().begin_idd_setup(slot, stop.clone())
}); });
let pipe = build_pipeline_with_retry(
&mut vd,
mode,
bitrate_kbps,
bit_depth,
plan,
&quit,
&stop,
Some(bringup.as_ref()),
)?;
// Setup done — the IDD-push setup lock releases as the guard leaves this arm's scope,
// so the next reconnect can begin (and preempt us).
(vd, pipe)
}
};
let (mut capturer, mut enc, mut frame, mut interval, mut cur_node_id, mut cur_display_gen) = let (mut capturer, mut enc, mut frame, mut interval, mut cur_node_id, mut cur_display_gen) =
build_pipeline_with_retry(&mut vd, mode, bitrate_kbps, bit_depth, plan, &quit, &stop)?; pipe;
// Setup done — release the IDD-push setup lock so the next reconnect can begin (and preempt us).
#[cfg(target_os = "windows")]
drop(_idd_setup_guard);
// Capture is live — launch the requested title so it renders onto the streamed output and // Capture is live — launch the requested title so it renders onto the streamed output and
// grabs focus. Windows spawns the library id into the interactive user session; Linux spawns // grabs focus. Windows spawns the library id into the interactive user session; Linux spawns
@@ -4239,6 +4384,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
codec: plan.codec.label(), codec: plan.codec.label(),
client: client_label, client: client_label,
bitrate_kbps: live_bitrate.clone(), bitrate_kbps: live_bitrate.clone(),
bringup: bringup.clone(),
}; };
let send_thread = std::thread::Builder::new() let send_thread = std::thread::Builder::new()
.name("punktfunk-send".into()) .name("punktfunk-send".into())
@@ -4272,6 +4418,8 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
plan.codec, plan.codec,
stop.clone(), stop.clone(),
force_idr.clone(), force_idr.clone(),
bringup.total_slot(),
resize_ms.clone(),
); );
// Mid-stream session-switch watcher (opt-in via PUNKTFUNK_SESSION_WATCH; never under an explicit // Mid-stream session-switch watcher (opt-in via PUNKTFUNK_SESSION_WATCH; never under an explicit
@@ -4404,6 +4552,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
plan, plan,
&quit, &quit,
&stop, &stop,
None,
)?; )?;
Ok((new_vd, pipe)) Ok((new_vd, pipe))
})(); })();
@@ -4454,6 +4603,10 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
} }
if let Some(new_mode) = want { if let Some(new_mode) = want {
tracing::info!(?new_mode, "rebuilding pipeline for mode switch"); tracing::info!(?new_mode, "rebuilding pipeline for mode switch");
// Resize trace (P0.1): reconfigure-received → pipeline rebuilt (incl. the first
// new-mode frame — `build_pipeline` waits for it). Total lands in the shared
// `resize_ms` slot (→ `session_status`); a failed rebuild abandons it silently.
let resize_trace = crate::bringup::Trace::start("resize", resize_ms.clone());
// PyroWave's Automatic bitrate is a per-mode ~1.6 bpp pin (resolve_bitrate_kbps_for) — // PyroWave's Automatic bitrate is a per-mode ~1.6 bpp pin (resolve_bitrate_kbps_for) —
// a resolution change moves the operating point (1080p→4K quadruples the pixel rate), // a resolution change moves the operating point (1080p→4K quadruples the pixel rate),
// so re-resolve it for the new mode. Explicit client rates stay put (the operator knows // so re-resolve it for the new mode. Explicit client rates stay put (the operator knows
@@ -4466,7 +4619,15 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
// Build the new pipeline BEFORE dropping the old one: the host already acked // Build the new pipeline BEFORE dropping the old one: the host already acked
// the switch as accepted, so a rebuild failure must not kill an otherwise // the switch as accepted, so a rebuild failure must not kill an otherwise
// healthy session — keep streaming the current mode and log instead. // healthy session — keep streaming the current mode and log instead.
match build_pipeline(&mut vd, new_mode, mode_bitrate, bit_depth, plan, &quit) { match build_pipeline(
&mut vd,
new_mode,
mode_bitrate,
bit_depth,
plan,
&quit,
Some(resize_trace.as_ref()),
) {
Ok(next_pipe) => { Ok(next_pipe) => {
if mode_bitrate != bitrate_kbps { if mode_bitrate != bitrate_kbps {
tracing::info!( tracing::info!(
@@ -4514,6 +4675,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
last_au_at = std::time::Instant::now(); last_au_at = std::time::Instant::now();
encoder_resets = 0; encoder_resets = 0;
last_forced_idr = Some(std::time::Instant::now()); // fresh encoder opens on an IDR — anchor the cooldown last_forced_idr = Some(std::time::Instant::now()); // fresh encoder opens on an IDR — anchor the cooldown
resize_trace.finish("pipeline_rebuilt");
} }
Err(e) => { Err(e) => {
tracing::warn!(error = %format!("{e:#}"), ?new_mode, tracing::warn!(error = %format!("{e:#}"), ?new_mode,
@@ -4822,6 +4984,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
plan, plan,
&quit, &quit,
&stop, &stop,
None,
) { ) {
Ok(p) => break p, Ok(p) => break p,
Err(e2) => { Err(e2) => {
@@ -5066,6 +5229,7 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
}; };
// Hand to the send thread; this blocks (backpressure) if it's behind. An Err means it // Hand to the send thread; this blocks (backpressure) if it's behind. An Err means it
// exited (send failure / stop) — end the encode loop too. // exited (send failure / stop) — end the encode loop too.
bringup.mark("first_au"); // P0.1 (first-crossing only; free afterwards)
if frame_tx.send(msg).is_err() { if frame_tx.send(msg).is_err() {
send_gone = true; send_gone = true;
break; break;
@@ -5182,6 +5346,83 @@ type Pipeline = (
/// error chain is classified and permanent ones short-circuit. Each failed attempt drops its /// error chain is classified and permanent ones short-circuit. Each failed attempt drops its
/// capturer, which (via `PortalCapturer::Drop`) tears the PipeWire thread + virtual output down /// capturer, which (via `PortalCapturer::Drop`) tears the PipeWire thread + virtual output down
/// before the next attempt — no leak across retries. /// before the next attempt — no leak across retries.
/// The Welcome-time display-prep hand-off (latency plan P1.1/P1.2): the opened vdisplay backend +
/// the fully built pipeline — monitor create, activation, settle, capture attach, first frame,
/// encoder open — produced on the prep/stream thread while the client's Start round-trip and the
/// UDP hole-punch are still in flight, so the entire display bring-up hides behind the network
/// waits. Constructed on the Windows native path only today: the Linux backends bind launch
/// semantics before create (gamescope nests the launch command), which must not run for a client
/// that never sends Start.
struct PreparedDisplay {
vd: Box<dyn crate::vdisplay::VirtualDisplay>,
pipeline: Pipeline,
}
/// The prep thread's hand-off pair: the sender delivers the post-punch [`SessionContext`] to the
/// thread (which then runs [`virtual_stream`] on its prepared display); the join handle returns
/// the stream result. Dropping the sender un-received aborts the prep cleanly (the prepared
/// display's lease releases into keep-alive policy).
type PrepHandle = (
std::sync::mpsc::SyncSender<SessionContext>,
std::thread::JoinHandle<Result<()>>,
);
/// Build the session's display + pipeline at Welcome time (latency plan P1.1/P1.2), before the
/// client's `Start` and the hole-punch — the negotiated mode is final once the Welcome is built,
/// and nothing in monitor create → activation → settle → capture attach → encoder open needs the
/// punched socket. Mirrors `virtual_stream`'s inline bring-up exactly: same backend setters, same
/// slot-scoped `begin_idd_setup` serialization (the guard releases when this returns), same
/// retry-wrapped build. The caller threads the SAME values the Welcome committed, so the prepared
/// pipeline and the later `SessionContext` can never disagree.
#[cfg(target_os = "windows")]
#[allow(clippy::too_many_arguments)]
fn prepare_display(
compositor: crate::vdisplay::Compositor,
mode: punktfunk_core::Mode,
client_identity: Option<[u8; 32]>,
client_hdr: Option<punktfunk_core::quic::HdrMeta>,
bitrate_kbps: u32,
bit_depth: u8,
chroma: crate::encode::ChromaFormat,
codec: crate::encode::Codec,
shard_payload: u16,
quit: &Arc<AtomicBool>,
stop: &Arc<AtomicBool>,
trace: &crate::bringup::Trace,
) -> Result<PreparedDisplay> {
// Same plan resolution as `virtual_stream` (pure in these inputs + host config), including
// PyroWave's datagram-aligned wire mode — `Session::shard_payload()` echoes the negotiated
// Welcome value passed here.
let mut plan = crate::session_plan::SessionPlan::resolve(bit_depth, chroma, codec);
if codec == crate::encode::Codec::PyroWave {
plan.wire_chunk = Some(shard_payload as usize);
}
let mut vd = crate::vdisplay::open(compositor)?;
vd.set_client_identity(client_identity);
vd.set_client_hdr(client_hdr);
vd.set_quit_flag(quit.clone());
// Slot-scoped setup serialization + reconnect preempt — see the inline arm in
// `virtual_stream` for the full rationale; released when this fn returns.
let _idd_setup_guard =
(plan.capture == crate::session_plan::CaptureBackend::IddPush).then(|| {
let slot =
crate::vdisplay::manager::slot_id_for(client_identity, (mode.width, mode.height));
crate::vdisplay::manager::vdm().begin_idd_setup(slot, stop.clone())
});
let pipeline = build_pipeline_with_retry(
&mut vd,
mode,
bitrate_kbps,
bit_depth,
plan,
quit,
stop,
Some(trace),
)?;
Ok(PreparedDisplay { vd, pipeline })
}
#[allow(clippy::too_many_arguments)]
fn build_pipeline_with_retry( fn build_pipeline_with_retry(
vd: &mut Box<dyn crate::vdisplay::VirtualDisplay>, vd: &mut Box<dyn crate::vdisplay::VirtualDisplay>,
mode: punktfunk_core::Mode, mode: punktfunk_core::Mode,
@@ -5190,6 +5431,9 @@ fn build_pipeline_with_retry(
plan: crate::session_plan::SessionPlan, plan: crate::session_plan::SessionPlan,
quit: &Arc<AtomicBool>, quit: &Arc<AtomicBool>,
stop: &Arc<AtomicBool>, stop: &Arc<AtomicBool>,
// Transition trace (P0.1): `Some` for the traced builds (bring-up, resize); each stage stamps
// once (first crossing) so the retry loop can pass it through unconditionally.
trace: Option<&crate::bringup::Trace>,
) -> Result<Pipeline> { ) -> Result<Pipeline> {
// ~10s first-frame wait per attempt. 8 gives a ~90s budget for the SLOW case: a host-managed // ~10s first-frame wait per attempt. 8 gives a ~90s budget for the SLOW case: a host-managed
// gamescope session cold-starting Steam Big Picture (the SteamOS/Bazzite takeover) can take // gamescope session cold-starting Steam Big Picture (the SteamOS/Bazzite takeover) can take
@@ -5227,7 +5471,7 @@ fn build_pipeline_with_retry(
attempt - 1 attempt - 1
); );
} }
match build_pipeline(vd, mode, bitrate_kbps, bit_depth, plan, quit) { match build_pipeline(vd, mode, bitrate_kbps, bit_depth, plan, quit, trace) {
Ok(pipe) => { Ok(pipe) => {
if attempt > 1 { if attempt > 1 {
tracing::info!(attempt, "pipeline up after retry"); tracing::info!(attempt, "pipeline up after retry");
@@ -5302,6 +5546,7 @@ fn reset_stalled_encoder(
true true
} }
#[allow(clippy::too_many_arguments)]
fn build_pipeline( fn build_pipeline(
vd: &mut Box<dyn crate::vdisplay::VirtualDisplay>, vd: &mut Box<dyn crate::vdisplay::VirtualDisplay>,
mode: punktfunk_core::Mode, mode: punktfunk_core::Mode,
@@ -5309,6 +5554,9 @@ fn build_pipeline(
bit_depth: u8, bit_depth: u8,
plan: crate::session_plan::SessionPlan, plan: crate::session_plan::SessionPlan,
quit: &Arc<AtomicBool>, quit: &Arc<AtomicBool>,
// Transition trace (P0.1): stamps the build's stages (display acquire, capture attach, first
// frame, encoder open) into the bring-up/resize timeline. `None` on untraced rebuilds.
trace: Option<&crate::bringup::Trace>,
) -> Result<Pipeline> { ) -> Result<Pipeline> {
// Acquire through the registry (design/display-management.md): on Linux this pools the display // Acquire through the registry (design/display-management.md): on Linux this pools the display
// for keep-alive (reuse a kept one, or create + keep the backend's keepalive so it outlives the // for keep-alive (reuse a kept one, or create + keep the backend's keepalive so it outlives the
@@ -5317,6 +5565,9 @@ fn build_pipeline(
// `quit` flag rides into the lease so a deliberate-quit teardown skips the keep-alive linger. // `quit` flag rides into the lease so a deliberate-quit teardown skips the keep-alive linger.
let vout = crate::vdisplay::registry::acquire(vd, mode, quit.clone()) let vout = crate::vdisplay::registry::acquire(vd, mode, quit.clone())
.context("create virtual output")?; .context("create virtual output")?;
if let Some(t) = trace {
t.mark("display_acquired");
}
// A2: if this was a REUSED kept display and its first frame fails, tear the (dead) pool entry down // A2: if this was a REUSED kept display and its first frame fails, tear the (dead) pool entry down
// so the retry loop's next acquire creates fresh instead of re-wedging on the same corpse. Read the // so the retry loop's next acquire creates fresh instead of re-wedging on the same corpse. Read the
// gen BEFORE `capture_virtual_output` consumes `vout`. (Linux-only — the pool is Linux.) // gen BEFORE `capture_virtual_output` consumes `vout`. (Linux-only — the pool is Linux.)
@@ -5354,6 +5605,9 @@ fn build_pipeline(
let mut capturer = let mut capturer =
crate::capture::capture_virtual_output(vout, plan.output_format(), plan.capture) crate::capture::capture_virtual_output(vout, plan.output_format(), plan.capture)
.context("capture virtual output")?; .context("capture virtual output")?;
if let Some(t) = trace {
t.mark("capture_attached");
}
capturer.set_active(true); capturer.set_active(true);
let frame = match capturer.next_frame().context("first frame") { let frame = match capturer.next_frame().context("first frame") {
Ok(f) => f, Ok(f) => f,
@@ -5366,6 +5620,9 @@ fn build_pipeline(
return Err(e); return Err(e);
} }
}; };
if let Some(t) = trace {
t.mark("first_frame");
}
// `bit_depth` is the handshake-negotiated value (8, or 10 = HEVC Main10 when the client // `bit_depth` is the handshake-negotiated value (8, or 10 = HEVC Main10 when the client
// advertised VIDEO_CAP_10BIT and the host opted in). Threaded down from the Welcome. // advertised VIDEO_CAP_10BIT and the host opted in). Threaded down from the Welcome.
let mut enc = crate::encode::open_video( let mut enc = crate::encode::open_video(
@@ -5380,6 +5637,9 @@ fn build_pipeline(
plan.chroma, plan.chroma,
) )
.context("open video encoder")?; .context("open video encoder")?;
if let Some(t) = trace {
t.mark("encoder_open");
}
if let Some(c) = plan.wire_chunk { if let Some(c) = plan.wire_chunk {
enc.set_wire_chunking(c); enc.set_wire_chunking(c);
} }
@@ -36,6 +36,11 @@ struct LiveSession {
/// One-shot force-keyframe flag ([`force_idr_all`] → mgmt `POST /session/idr`); the encode loop /// One-shot force-keyframe flag ([`force_idr_all`] → mgmt `POST /session/idr`); the encode loop
/// drains it alongside a client's decode-recovery keyframe request. /// drains it alongside a client's decode-recovery keyframe request.
force_idr: Arc<AtomicBool>, force_idr: Arc<AtomicBool>,
/// Completed bring-up total (hello → first packet), ms; 0 until the first packet left. Written
/// once by the session's [`crate::bringup::Trace`] (latency plan P0.1).
ttff_ms: Arc<AtomicU32>,
/// Most recent completed mid-stream resize (reconfigure → pipeline rebuilt), ms; 0 = none yet.
last_resize_ms: Arc<AtomicU32>,
} }
/// A resolved read of one live session, for the `/status` view. /// A resolved read of one live session, for the `/status` view.
@@ -46,6 +51,10 @@ pub struct SessionSnapshot {
pub fps: u32, pub fps: u32,
pub bitrate_kbps: u32, pub bitrate_kbps: u32,
pub codec: Codec, pub codec: Codec,
/// Bring-up total (hello → first packet), ms; 0 while still bringing up (latency plan P0.1).
pub time_to_first_frame_ms: u32,
/// Most recent mid-stream resize total, ms; 0 = no resize this session.
pub last_resize_ms: u32,
} }
fn registry() -> &'static Mutex<Vec<LiveSession>> { fn registry() -> &'static Mutex<Vec<LiveSession>> {
@@ -65,6 +74,8 @@ pub fn register(
codec: Codec, codec: Codec,
stop: Arc<AtomicBool>, stop: Arc<AtomicBool>,
force_idr: Arc<AtomicBool>, force_idr: Arc<AtomicBool>,
ttff_ms: Arc<AtomicU32>,
last_resize_ms: Arc<AtomicU32>,
) -> LiveSessionGuard { ) -> LiveSessionGuard {
let id = next_id(); let id = next_id();
registry().lock().unwrap().push(LiveSession { registry().lock().unwrap().push(LiveSession {
@@ -74,6 +85,8 @@ pub fn register(
codec, codec,
stop, stop,
force_idr, force_idr,
ttff_ms,
last_resize_ms,
}); });
LiveSessionGuard { id } LiveSessionGuard { id }
} }
@@ -109,6 +122,8 @@ pub fn snapshot() -> Vec<SessionSnapshot> {
fps, fps,
bitrate_kbps: s.bitrate_kbps.load(Ordering::Relaxed), bitrate_kbps: s.bitrate_kbps.load(Ordering::Relaxed),
codec: s.codec, codec: s.codec,
time_to_first_frame_ms: s.ttff_ms.load(Ordering::Relaxed),
last_resize_ms: s.last_resize_ms.load(Ordering::Relaxed),
} }
}) })
.collect() .collect()