fix(audio/windows): stop the mint retry path from broadcasting PnP device changes at the whole box, and revert session tuning when streaming ends
Field report (2026-08-12): Punktfunk's audio devices tank Helldivers 2 to 1% lows of 2-5 FPS; uninstalling restores performance. Two host-side mechanisms can plausibly do that, both fixed here. The mint retry storm: minted::ensure_blocking() ran a FULL provisioning pass on every mic-pump open with no cooldown, no in-flight guard, and no give-up - and ensure_role() reached UpdateDriverForPlugAndPlayDevicesW even when the devnode already existed. On a box where minting never latches, the pump's reopen backoff (capped 60 s) turned that into a PnP driver rebind + system-wide device-change broadcast roughly once a minute, forever - and games rebuild their audio graph on each broadcast. Now: * ensure_role() gets a steady-state fast path: a marker devnode whose endpoints are all live resolves without touching PnP or the default-device policy. * ensure_blocking() waits on an in-flight pass instead of racing a second SetupAPI sweep against it (the dead-mic-air deploy race), honours RETRY_COOLDOWN after a failed pass (first-ever resolve still blocks, per the cold-boot mint contract), and * five unlatched passes stop minting for the host lifetime (a service restart re-arms) - counted across the worker and the blocking path. The never-reverted session tuning: pf-frame's tune_process_once() put the whole host at HIGH_PRIORITY_CLASS with timeBeginPeriod(1) and DWM MMCSS on the first hot stream thread and documented 'reverts at process exit' - but the host is a 24/7 service, so after one stream it competed at HIGH class with a 1 ms global timer against whatever the user played locally, forever. The process-wide tuning is now refcounted across the hot threads via a TLS guard: the first hot thread applies it, the last one's exit reverts it (timeEndPeriod, DwmEnableMMCSS(0), NORMAL class) - the same thread-exit lifetime the MMCSS and execution-state effects already ride. Every on_hot_thread() call site is a session-scoped thread (capture/encode, packetizer, send, NVENC retrieve), so the revert lands at session teardown.
This commit is contained in:
@@ -8,14 +8,16 @@
|
||||
//!
|
||||
//! Raw C-ABI FFI (winmm/kernel32/dwmapi/avrt) rather than the `windows` crate so it builds without
|
||||
//! pulling new windows-rs features. No-op on non-Windows. Per-thread effects (MMCSS, execution
|
||||
//! state) auto-revert at thread exit (= session end); the process-wide bits revert at process exit.
|
||||
//! state) auto-revert at thread exit (= session end); the process-wide bits are refcounted over
|
||||
//! the hot threads and revert when the LAST one exits — the host must not keep HIGH priority and
|
||||
//! a 1 ms global timer while a local game runs and nobody streams (2026-08-12 field report).
|
||||
//! See `design/host-latency-plan.md` Tier 3A.
|
||||
|
||||
#[cfg(target_os = "windows")]
|
||||
mod imp {
|
||||
#![allow(non_snake_case)]
|
||||
use std::ffi::c_void;
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::Mutex;
|
||||
|
||||
type Handle = *mut c_void;
|
||||
type Bool = i32;
|
||||
@@ -23,6 +25,7 @@ mod imp {
|
||||
#[link(name = "winmm")]
|
||||
unsafe extern "system" {
|
||||
fn timeBeginPeriod(uPeriod: u32) -> u32;
|
||||
fn timeEndPeriod(uPeriod: u32) -> u32;
|
||||
}
|
||||
#[link(name = "kernel32")]
|
||||
unsafe extern "system" {
|
||||
@@ -55,6 +58,7 @@ mod imp {
|
||||
}
|
||||
|
||||
const HIGH_PRIORITY_CLASS: u32 = 0x0000_0080;
|
||||
const NORMAL_PRIORITY_CLASS: u32 = 0x0000_0020;
|
||||
const ES_CONTINUOUS: u32 = 0x8000_0000;
|
||||
const ES_SYSTEM_REQUIRED: u32 = 0x0000_0001;
|
||||
const ES_DISPLAY_REQUIRED: u32 = 0x0000_0002;
|
||||
@@ -114,16 +118,19 @@ mod imp {
|
||||
}
|
||||
}
|
||||
|
||||
static PROCESS_TUNED: OnceLock<()> = OnceLock::new();
|
||||
/// Live hot (session) threads. A Mutex, not an atomic: the 0↔1 transitions carry the
|
||||
/// apply/revert side effects, and an interleaved fetch_add/fetch_sub pair could otherwise
|
||||
/// finish with a running session untuned (transitions are rare — thread start/exit only).
|
||||
static HOT_THREADS: Mutex<usize> = Mutex::new(0);
|
||||
|
||||
/// Process-wide tuning, applied exactly once. Reverts at process exit. Best-effort: each call is
|
||||
/// independent and a failure is ignored (e.g. a non-elevated host may not get HIGH class).
|
||||
fn tune_process_once() {
|
||||
/// Process-wide tuning, applied when the FIRST hot thread registers. Best-effort: each call
|
||||
/// is independent and a failure is ignored (e.g. a non-elevated host may not get HIGH class).
|
||||
fn tune_process() {
|
||||
// SAFETY: each call is a C-ABI FFI into winmm/kernel32/dwmapi declared with a matching
|
||||
// `extern "system"` signature; every argument is a plain integer (no pointers/buffers escape),
|
||||
// and `GetCurrentProcess()` returns the current-process pseudo-handle (a constant, always valid,
|
||||
// never closed). The body runs inside `get_or_init`, so it executes exactly once per process.
|
||||
PROCESS_TUNED.get_or_init(|| unsafe {
|
||||
// never closed).
|
||||
unsafe {
|
||||
// 1 ms timer granularity (default ~15.6 ms) — the floor for precise frame pacing and the
|
||||
// encode|send split's sub-ms sleeps.
|
||||
timeBeginPeriod(1);
|
||||
@@ -134,16 +141,66 @@ mod imp {
|
||||
// control/capture/encode/send threads on the CPU (Apollo does the same).
|
||||
SetPriorityClass(GetCurrentProcess(), HIGH_PRIORITY_CLASS);
|
||||
tracing::info!("windows session tuning applied (timer 1ms, DWM MMCSS, HIGH priority)");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// Call at the start of each capture/encode/send (hot stream) thread. Applies the process-wide
|
||||
/// tuning once, registers the calling thread with MMCSS ("Games"), and asserts the display/system
|
||||
/// must stay awake for as long as this thread lives. The MMCSS handle is intentionally leaked and
|
||||
/// the execution-state assertion is bound to this thread — both are reverted by the OS when the
|
||||
/// thread exits, so a session that ends tears them down without explicit bookkeeping.
|
||||
/// The mirror of [`tune_process`], run when the LAST hot thread exits. Leaving the tuning in
|
||||
/// place used to be the design ("reverts at process exit") — but the host is a 24/7 service,
|
||||
/// so after one stream it competed at HIGH class with a 1 ms global timer against whatever
|
||||
/// the user played locally, forever.
|
||||
fn untune_process() {
|
||||
// SAFETY: same FFI surface as `tune_process` — plain-integer arguments, constant
|
||||
// pseudo-handle, no pointers or buffers.
|
||||
unsafe {
|
||||
timeEndPeriod(1); // pairs the timeBeginPeriod(1)
|
||||
DwmEnableMMCSS(0);
|
||||
SetPriorityClass(GetCurrentProcess(), NORMAL_PRIORITY_CLASS);
|
||||
tracing::info!("windows session tuning reverted (timer, DWM MMCSS, NORMAL priority)");
|
||||
}
|
||||
}
|
||||
|
||||
/// One per hot thread, parked in TLS by [`on_hot_thread`]; its Drop runs at thread exit
|
||||
/// (= session teardown), the same lifetime the MMCSS/execution-state effects already ride.
|
||||
struct HotThreadGuard;
|
||||
|
||||
impl Drop for HotThreadGuard {
|
||||
fn drop(&mut self) {
|
||||
// A poisoned lock skips the revert (best-effort, like every call here) instead of
|
||||
// panicking inside a TLS destructor.
|
||||
if let Ok(mut n) = HOT_THREADS.lock() {
|
||||
*n -= 1;
|
||||
if *n == 0 {
|
||||
untune_process();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
thread_local! {
|
||||
static HOT_THREAD: std::cell::OnceCell<HotThreadGuard> =
|
||||
const { std::cell::OnceCell::new() };
|
||||
}
|
||||
|
||||
/// Call at the start of each capture/encode/send (hot stream) thread. Registers the thread in
|
||||
/// the process-tuning refcount (first in applies, last out reverts), registers it with MMCSS
|
||||
/// ("Games"), and asserts the display/system must stay awake for as long as this thread lives.
|
||||
/// The MMCSS handle is intentionally leaked and the execution-state assertion is bound to this
|
||||
/// thread — both are reverted by the OS when the thread exits, and the refcount guard's TLS
|
||||
/// Drop runs there too, so a session that ends tears everything down without explicit
|
||||
/// bookkeeping.
|
||||
pub fn on_hot_thread() {
|
||||
tune_process_once();
|
||||
HOT_THREAD.with(|slot| {
|
||||
if slot.get().is_none() {
|
||||
{
|
||||
let mut n = HOT_THREADS.lock().unwrap();
|
||||
*n += 1;
|
||||
if *n == 1 {
|
||||
tune_process();
|
||||
}
|
||||
}
|
||||
let _ = slot.set(HotThreadGuard);
|
||||
}
|
||||
});
|
||||
// SAFETY: C-ABI FFI declared with matching `extern "system"` signatures. SetThreadExecutionState
|
||||
// takes only flag bits. `task` is a local NUL-terminated UTF-16 buffer ("Games\0") alive for the
|
||||
// whole block, so `task.as_ptr()` is a valid LPCWSTR for the call, and `&mut idx` is a live local
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
use super::pad_endpoint as pe;
|
||||
use super::{audio_control, wiring_plan};
|
||||
use anyhow::{bail, Context, Result};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
@@ -40,6 +40,17 @@ const ENDPOINT_WAIT: Duration = Duration::from_secs(15);
|
||||
/// Minimum spacing between provisioning retries once the startup attempt failed
|
||||
/// ([`ensure_provisioned`] is called from wiring passes, which recur freely).
|
||||
const RETRY_COOLDOWN: Duration = Duration::from_secs(60);
|
||||
/// Full passes that ended unlatched before minting gives up for this host lifetime (a service
|
||||
/// restart re-arms). An unlatched pass that reaches the PnP surface costs the whole BOX, not
|
||||
/// just us: the driver (re)bind raises a device-change broadcast every running app services,
|
||||
/// and games rebuild their audio graph on it — a box that cannot mint must not pay that on
|
||||
/// every retry forever (field-measured 2026-08-12 as Helldivers 2 hitching to 2–5 FPS 1% lows,
|
||||
/// one hitch per mic-pump reopen).
|
||||
const MAX_UNLATCHED_ATTEMPTS: u32 = 5;
|
||||
/// How long [`ensure_blocking`] waits on a pass another thread already runs before giving the
|
||||
/// wiring plan the unlatched answer (a full cold-boot pass worst-cases around two
|
||||
/// [`ENDPOINT_WAIT`]s plus the stamp settles).
|
||||
const BLOCKING_WAIT: Duration = Duration::from_secs(90);
|
||||
|
||||
/// The two minted roles. `value` is the persisted marker; the needles drive
|
||||
/// [`discover_driver`].
|
||||
@@ -107,6 +118,26 @@ static PROVISIONED: OnceLock<Arc<MintedAudio>> = OnceLock::new();
|
||||
static PROVISIONING: AtomicBool = AtomicBool::new(false);
|
||||
/// When the last attempt STARTED — the [`RETRY_COOLDOWN`] anchor.
|
||||
static LAST_ATTEMPT: Mutex<Option<Instant>> = Mutex::new(None);
|
||||
/// Completed passes that did not latch, across the worker and the blocking path — the
|
||||
/// [`MAX_UNLATCHED_ATTEMPTS`] give-up counter.
|
||||
static UNLATCHED_ATTEMPTS: AtomicU32 = AtomicU32::new(0);
|
||||
|
||||
/// Count one finished-but-unlatched pass; the crossing attempt logs the give-up exactly once.
|
||||
fn record_unlatched_attempt() {
|
||||
let n = UNLATCHED_ATTEMPTS.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
if n == MAX_UNLATCHED_ATTEMPTS {
|
||||
tracing::warn!(
|
||||
attempts = n,
|
||||
"minted-audio provisioning keeps failing — giving up for this host lifetime so \
|
||||
retries stop broadcasting device changes at the whole box; the wiring plan keeps \
|
||||
the name-based ladder, a service restart re-arms minting"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn gave_up() -> bool {
|
||||
UNLATCHED_ATTEMPTS.load(Ordering::SeqCst) >= MAX_UNLATCHED_ATTEMPTS
|
||||
}
|
||||
|
||||
/// The wiring plan's tier-0 input: the minted ids, or all-empty while nothing is provisioned.
|
||||
///
|
||||
@@ -135,7 +166,7 @@ pub(crate) fn provisioned() -> Option<Arc<MintedAudio>> {
|
||||
/// Spawn the provisioning worker (idempotent; returns immediately). Called at host start next
|
||||
/// to the pad provider, and again from [`ensure_provisioned`] on the retry path.
|
||||
pub(crate) fn provision_at_startup() {
|
||||
if std::env::var_os("PUNKTFUNK_NO_AUDIO_MINT").is_some() {
|
||||
if std::env::var_os("PUNKTFUNK_NO_AUDIO_MINT").is_some() || gave_up() {
|
||||
return;
|
||||
}
|
||||
if PROVISIONED.get().is_some() || PROVISIONING.swap(true, Ordering::SeqCst) {
|
||||
@@ -155,13 +186,19 @@ pub(crate) fn provision_at_startup() {
|
||||
);
|
||||
let _ = PROVISIONED.set(Arc::new(m));
|
||||
}
|
||||
Ok(_) => tracing::info!(
|
||||
"no minted audio endpoints (Steam's streaming drivers absent?) — the \
|
||||
wiring plan keeps the name-based ladder"
|
||||
),
|
||||
Err(e) => tracing::warn!(error = %format!("{e:#}"),
|
||||
"minted-audio provisioning failed — the wiring plan keeps the name-based \
|
||||
ladder and a later wiring pass retries"),
|
||||
Ok(_) => {
|
||||
tracing::info!(
|
||||
"no minted audio endpoints (Steam's streaming drivers absent?) — the \
|
||||
wiring plan keeps the name-based ladder"
|
||||
);
|
||||
record_unlatched_attempt();
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"),
|
||||
"minted-audio provisioning failed — the wiring plan keeps the name-based \
|
||||
ladder and a later wiring pass retries");
|
||||
record_unlatched_attempt();
|
||||
}
|
||||
}
|
||||
PROVISIONING.store(false, Ordering::SeqCst);
|
||||
});
|
||||
@@ -175,7 +212,7 @@ pub(crate) fn provision_at_startup() {
|
||||
/// [`RETRY_COOLDOWN`] — a box where Steam arrives later mints on a later pass instead of at
|
||||
/// the next reboot.
|
||||
pub(crate) fn ensure_provisioned() {
|
||||
if PROVISIONED.get().is_some() {
|
||||
if PROVISIONED.get().is_some() || gave_up() {
|
||||
return;
|
||||
}
|
||||
{
|
||||
@@ -219,6 +256,19 @@ fn ensure_all() -> Result<MintedAudio> {
|
||||
/// back any default device the fresh endpoint grabbed (measured on the pad program: a newly
|
||||
/// registered endpoint can take either default).
|
||||
fn ensure_role(role: Role) -> Result<(String, String, Option<String>)> {
|
||||
// Steady state: a previous run's devnode with all endpoints live — resolve by marker and
|
||||
// return without touching PnP or the default-device policy. The full pass below (re)binds
|
||||
// the driver even over an existing devnode, and that bind raises a device-change broadcast
|
||||
// every running app services — right at first mint, ruinous from a retry path (each
|
||||
// broadcast makes games rebuild their audio graph; see [`MAX_UNLATCHED_ATTEMPTS`]).
|
||||
if let Some((devnode, render, capture)) = find_healthy_role(role)? {
|
||||
stamp_identity(&render, role, false);
|
||||
if let Some(cap) = capture.as_ref() {
|
||||
stamp_identity(cap, role, true);
|
||||
}
|
||||
return Ok((devnode, render, capture));
|
||||
}
|
||||
|
||||
let prev_render = audio_control::default_render_id();
|
||||
let prev_capture = audio_control::default_capture_id();
|
||||
|
||||
@@ -284,6 +334,27 @@ fn ensure_role(role: Role) -> Result<(String, String, Option<String>)> {
|
||||
Ok((devnode, render, capture))
|
||||
}
|
||||
|
||||
/// The role's marker devnode with EVERY endpoint the role owes already registered, or `None`
|
||||
/// (missing devnode, missing endpoint, or an enumeration error → the caller runs the full
|
||||
/// pass). Same endpoint resolvers [`wait_for`] polls, so "healthy" here is exactly the state
|
||||
/// the full pass would declare ready.
|
||||
fn find_healthy_role(role: Role) -> Result<Option<(String, String, Option<String>)>> {
|
||||
let Some(devnode) = find_role_devnode(role)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let Some(render) = pe::find_endpoint_for_devnode(&devnode)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let capture = match role {
|
||||
Role::Mic => match pe::find_capture_endpoint_for_devnode(&devnode)? {
|
||||
Some(cap) => Some(cap),
|
||||
None => return Ok(None),
|
||||
},
|
||||
Role::Speakers => None,
|
||||
};
|
||||
Ok(Some((devnode, render, capture)))
|
||||
}
|
||||
|
||||
/// How many stamp/settle passes a name gets before we accept "stored but not yet served"
|
||||
/// (a settled endpoint takes the stamp on the first pass; a freshly minted one may need the
|
||||
/// audio stack to notice — it serves after the next Audiosrv restart/reboot at the latest).
|
||||
@@ -510,7 +581,6 @@ pub(crate) fn discover_driver(needle: &str, inf_name: &str) -> Result<(String, S
|
||||
)
|
||||
}
|
||||
|
||||
/// `audio-probe mint` devtest body: one synchronous provisioning pass, results printed.
|
||||
/// Synchronous provisioning — for the mic pump's resolve and the devtests.
|
||||
///
|
||||
/// The pump's FIRST open must not race the startup worker: measured on the target box, the
|
||||
@@ -520,15 +590,50 @@ pub(crate) fn discover_driver(needle: &str, inf_name: &str) -> Result<(String, S
|
||||
/// (existing marker devnodes re-resolve in milliseconds; a cold boot pays the one-time mint)
|
||||
/// keeps the pump's target and the plan's verdict the same thing. Latched calls return
|
||||
/// immediately; the opt-out env is honoured like everywhere else.
|
||||
///
|
||||
/// While UNLATCHED this is where the pump's reopen backoff (capped at 60 s) used to meet an
|
||||
/// unguarded full pass: one PnP rebind + device-change broadcast roughly every minute, forever,
|
||||
/// on any box where minting cannot converge (the 2026-08-12 Helldivers 2 field report). Now a
|
||||
/// pass someone else already runs is WAITED for instead of raced, a failed pass repeats at most
|
||||
/// every [`RETRY_COOLDOWN`], and [`MAX_UNLATCHED_ATTEMPTS`] failures stop retrying for the
|
||||
/// host lifetime.
|
||||
pub(crate) fn ensure_blocking() {
|
||||
if std::env::var_os("PUNKTFUNK_NO_AUDIO_MINT").is_some() || PROVISIONED.get().is_some() {
|
||||
if std::env::var_os("PUNKTFUNK_NO_AUDIO_MINT").is_some()
|
||||
|| PROVISIONED.get().is_some()
|
||||
|| gave_up()
|
||||
{
|
||||
return;
|
||||
}
|
||||
if let Ok(m) = ensure_all() {
|
||||
if m.any() {
|
||||
let _ = PROVISIONED.set(Arc::new(m));
|
||||
// A pass is in flight (the startup worker, or a concurrent resolve): wait for its verdict
|
||||
// rather than racing a second SetupAPI/PnP sweep against it — that race is how the pump
|
||||
// once ended up wired to the cable while the worker minted (the dead-mic-air deploy race).
|
||||
if PROVISIONING.swap(true, Ordering::SeqCst) {
|
||||
let deadline = Instant::now() + BLOCKING_WAIT;
|
||||
while PROVISIONING.load(Ordering::SeqCst) && Instant::now() < deadline {
|
||||
thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
return;
|
||||
}
|
||||
// We own the slot. First-ever resolve runs unconditionally (the cold-boot mint the doc
|
||||
// above insists on); after a failed pass the cooldown answers instead of a re-run.
|
||||
let run = {
|
||||
let mut last = LAST_ATTEMPT.lock().unwrap();
|
||||
if last.is_some_and(|t| t.elapsed() < RETRY_COOLDOWN) {
|
||||
false
|
||||
} else {
|
||||
*last = Some(Instant::now());
|
||||
true
|
||||
}
|
||||
};
|
||||
if run {
|
||||
match ensure_all() {
|
||||
Ok(m) if m.any() => {
|
||||
let _ = PROVISIONED.set(Arc::new(m));
|
||||
}
|
||||
_ => record_unlatched_attempt(),
|
||||
}
|
||||
}
|
||||
PROVISIONING.store(false, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
pub(crate) fn devtest_mint() -> Result<()> {
|
||||
|
||||
Reference in New Issue
Block a user