From 268e6e19a4c81536f9a37ab5e4c0e065c7579e6d Mon Sep 17 00:00:00 2001 From: whoisaldo Date: Thu, 27 Aug 2026 13:22:05 -0400 Subject: [PATCH] Fix four host faults the synthetic tests could never see MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From the code-review pass over the merged revamp. Each is invisible to CI because the synthetic capture backend starts instantly, never changes resolution, and never touches the virtual display. The extended display could never come up. spawn_generation re-armed the running flag but left the watchdog heartbeat atomics holding the previous generation's timestamps, and the wedge check has no warm-up. Attaching the virtual display polls for up to 10 seconds before the capture loop stamps its first heartbeat, so at 3 seconds the watchdog judged the new generation by the old one's clock, killed it, and did the same to the next five until the storm brake parked the supervisor in Failed. The atomics are now stamped when a generation is spawned. A superseded capture thread kept running. Stage threads are detached, so one blocked in the virtual-display poll or the ACCESS_LOST retry ladder outlives its generation; it then re-read the process-wide running flag, found it true again for the NEXT generation, and carried on. Two capture loops then fight over a DXGI duplication only one can hold, and the zombie's heartbeats mask the watchdog that would have caught it. Stage loops now stop when their own generation is superseded, which is what CURRENT_GENERATION was there for. Lowering the bitrate slider did nothing. set_ceiling clamps the current rung and reports that it changed, but the caller discarded the decision and used only on_report's, which returns unchanged for a clean report. So the label moved and the encoder did not, for the life of the pipeline. Both decisions now count. A typed-in address counted as a viewer. client_connected only checked target_addr, which the GUI's Target IP field and the persisted target_ip both wrote — so with nobody connected the capture loop skipped its idle throttle and ran flat out, and with Extended display selected the virtual monitor came up for an audience of none. In v2 the session is the only truth, so client_connected asks it. The Target IP control is gone with it: media carries a session id from the handshake, so an address typed there could never have received a stream. --- host/src/capture/dxgi.rs | 13 ++++-- host/src/capture/mod.rs | 14 +++--- host/src/capture/synthetic.rs | 7 ++- host/src/control.rs | 28 +++++++++++ host/src/encoder/mod.rs | 10 ++-- host/src/gui.rs | 88 +++++++---------------------------- host/src/pipeline.rs | 15 ++++-- host/src/supervisor.rs | 31 ++++++++++-- host/src/transport/mod.rs | 14 +++++- 9 files changed, 125 insertions(+), 95 deletions(-) diff --git a/host/src/capture/dxgi.rs b/host/src/capture/dxgi.rs index 1b63ea1..62d2d5a 100644 --- a/host/src/capture/dxgi.rs +++ b/host/src/capture/dxgi.rs @@ -100,6 +100,7 @@ pub fn run_capture_loop( slot: &FrameSlot, shared: SharedControl, adapter_index: u32, + generation: u64, ) -> Result<(), Box> { // --- Create DXGI factory and select the output to capture --- let factory: IDXGIFactory1 = unsafe { CreateDXGIFactory1()? }; @@ -272,7 +273,9 @@ pub fn run_capture_loop( let mut cursor_y: i32 = 0; loop { - if !shared.running.load(Ordering::SeqCst) { + if !shared.running.load(Ordering::SeqCst) + || !crate::supervisor::generation_is_current(generation) + { info!("Capture loop stopping on running=false"); break; } @@ -445,7 +448,9 @@ pub fn run_capture_loop( } Err(e) if e.code() == DXGI_ERROR_WAIT_TIMEOUT => { // Desktop unchanged — not an error. - if !shared.running.load(Ordering::SeqCst) { + if !shared.running.load(Ordering::SeqCst) + || !crate::supervisor::generation_is_current(generation) + { info!("Capture loop stopping after timeout on running=false"); break; } @@ -489,7 +494,9 @@ pub fn run_capture_loop( let mut reinit = None; let delays_ms = [100u64, 250, 500, 1000, 2000]; for attempt in 0..15 { - if !shared.running.load(Ordering::SeqCst) { + if !shared.running.load(Ordering::SeqCst) + || !crate::supervisor::generation_is_current(generation) + { break; } // Keep the watchdog quiet: this wait IS the recovery. diff --git a/host/src/capture/mod.rs b/host/src/capture/mod.rs index 956165c..05596d0 100644 --- a/host/src/capture/mod.rs +++ b/host/src/capture/mod.rs @@ -205,8 +205,7 @@ fn vdd_attach_timeout() -> Duration { /// a client is connected, so an idle PC never shows a phantom second monitor. #[cfg(windows)] pub(crate) fn client_connected(shared: &SharedControl) -> bool { - let addr = *shared.target_addr.lock(); - !addr.ip().is_unspecified() && addr.port() != 0 + shared.client_connected() } /// Reconcile the virtual display device to the requested target and return the concrete @@ -310,6 +309,7 @@ pub fn run_capture_stage( slot: FrameSlot, shared: SharedControl, adapter_index: u32, + generation: u64, ) -> Result<(), Box> { struct CloseOnExit(FrameSlot); impl Drop for CloseOnExit { @@ -318,7 +318,7 @@ pub fn run_capture_stage( } } let guard = CloseOnExit(slot.handle()); - let result = run_platform_loop(&guard.0, shared, adapter_index); + let result = run_platform_loop(&guard.0, shared, adapter_index, generation); if let Err(ref e) = result { error!(error = %e, "Capture loop exited with error"); } @@ -337,13 +337,14 @@ fn run_platform_loop( slot: &FrameSlot, shared: SharedControl, adapter_index: u32, + generation: u64, ) -> Result<(), Box> { let synthetic_requested = std::env::var("ETERNAL_CAPTURE").is_ok_and(|v| v.trim().eq_ignore_ascii_case("synthetic")); if synthetic_requested { - synthetic::run_capture_loop(slot, shared) + synthetic::run_capture_loop(slot, shared, generation) } else { - dxgi::run_capture_loop(slot, shared, adapter_index) + dxgi::run_capture_loop(slot, shared, adapter_index, generation) } } @@ -352,8 +353,9 @@ fn run_platform_loop( slot: &FrameSlot, shared: SharedControl, _adapter_index: u32, + generation: u64, ) -> Result<(), Box> { - synthetic::run_capture_loop(slot, shared) + synthetic::run_capture_loop(slot, shared, generation) } /// Choose the freshly-attached virtual output: the first non-primary output whose device name was diff --git a/host/src/capture/synthetic.rs b/host/src/capture/synthetic.rs index 03da19f..40cd4c3 100644 --- a/host/src/capture/synthetic.rs +++ b/host/src/capture/synthetic.rs @@ -50,6 +50,7 @@ fn synthetic_size() -> (u32, u32) { pub fn run_capture_loop( slot: &FrameSlot, shared: SharedControl, + generation: u64, ) -> Result<(), Box> { let (width, height) = synthetic_size(); let row_bytes = (width * 4) as usize; @@ -78,8 +79,10 @@ pub fn run_capture_loop( let mut next_deadline = Instant::now(); loop { - if !shared.running.load(Ordering::SeqCst) { - info!("Capture loop stopping on running=false"); + if !shared.running.load(Ordering::SeqCst) + || !crate::supervisor::generation_is_current(generation) + { + info!("Capture loop stopping (stop requested or generation superseded)"); break; } diff --git a/host/src/control.rs b/host/src/control.rs index 85500c2..45d558a 100644 --- a/host/src/control.rs +++ b/host/src/control.rs @@ -131,6 +131,16 @@ impl SharedControl { pub fn stop(&self) { self.running.store(false, Ordering::SeqCst); } + + /// Is a client actually watching? Protocol v2 makes the session the only + /// truth: media needs a session id, so a target address on its own (a + /// persisted `target_ip`, a stale value from a previous client) must never + /// read as "someone is connected" — that would keep the capture loop at + /// full rate with nobody watching and bring the virtual display up with no + /// viewer. + pub fn client_connected(&self) -> bool { + self.session.lock().is_active() + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -175,3 +185,21 @@ impl GuiControl { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn client_connected_requires_a_session_not_just_an_address() { + let shared = SharedControl::new(9876, 15_000_000); + assert!(!shared.client_connected(), "no session, no client"); + + // A target address on its own — a persisted setting, or a leftover + // from a client that has since gone — must not read as "connected". + // It used to, which held the virtual display up and kept the capture + // loop at full rate with nobody watching. + *shared.target_addr.lock() = "192.168.1.50:9876".parse().unwrap(); + assert!(!shared.client_connected()); + } +} diff --git a/host/src/encoder/mod.rs b/host/src/encoder/mod.rs index 5e1f1d9..8eb839c 100644 --- a/host/src/encoder/mod.rs +++ b/host/src/encoder/mod.rs @@ -41,8 +41,9 @@ pub fn run_encode_stage( tx: mpsc::Sender, shared: SharedControl, gpu: GpuInfo, + generation: u64, ) -> Result<(), Box> { - let result = run_encode_loop(frames, tx, shared, gpu); + let result = run_encode_loop(frames, tx, shared, gpu, generation); if let Err(ref e) = result { error!(error = %e, "Encode loop exited with error"); } @@ -58,6 +59,7 @@ fn run_encode_loop( tx: mpsc::Sender, shared: SharedControl, gpu: GpuInfo, + generation: u64, ) -> Result<(), Box> { // Apply GUI encoder override at pipeline start. The override is consulted only here; // mid-stream changes don't take effect until the user requests a Restart. @@ -130,8 +132,10 @@ fn run_encode_loop( .and_then(|v| v.trim().parse().ok()); while let Some(raw_frame) = frames.blocking_take() { - if !shared.running.load(Ordering::SeqCst) { - info!("Encoder loop stopping on running=false"); + if !shared.running.load(Ordering::SeqCst) + || !crate::supervisor::generation_is_current(generation) + { + info!("Encoder loop stopping (stop requested or generation superseded)"); break; } diff --git a/host/src/gui.rs b/host/src/gui.rs index 4d01ea0..d2f53a9 100644 --- a/host/src/gui.rs +++ b/host/src/gui.rs @@ -1,5 +1,3 @@ -use std::net::SocketAddr; - use eframe::egui; use qrcode::{Color as QrModuleColor, QrCode}; use tracing::{info, warn}; @@ -145,8 +143,9 @@ pub struct AnalyzerApp { current_tab: AppTab, settings_bitrate_mbps: f32, settings_fps_target: u32, - settings_target_ip: String, - settings_target_error: Option, + /// Last settings error worth showing (currently only the autostart + /// registry write). + settings_error: Option, settings_start_on_boot: bool, settings_hevc_enabled: bool, settings_vdd_match: bool, @@ -199,21 +198,6 @@ impl AnalyzerApp { .target_fps .store(fps_target, std::sync::atomic::Ordering::SeqCst); - let settings_target_ip = if let Some(ip) = persisted.target_ip.clone() { - if let Ok(addr) = ip.parse::() { - *control.shared.target_addr.lock() = addr; - PIPELINE_STATS.lock().set_target_addr(addr.to_string()); - } - ip - } else { - let target_addr = *control.shared.target_addr.lock(); - if target_addr.ip().is_unspecified() || target_addr.port() == 0 { - String::new() - } else { - target_addr.to_string() - } - }; - let encoder_choice = if let Some(name) = persisted.encoder_override.as_deref() { *control.shared.encoder_override.lock() = Some(name.to_string()); ENCODER_CHOICES @@ -271,8 +255,7 @@ impl AnalyzerApp { current_tab: AppTab::Stream, settings_bitrate_mbps: bitrate_mbps, settings_fps_target: fps_target, - settings_target_ip, - settings_target_error: None, + settings_error: None, settings_start_on_boot: start_on_boot, settings_hevc_enabled: persisted.hevc_enabled, settings_vdd_match: persisted.vdd_match_resolution, @@ -285,24 +268,6 @@ impl AnalyzerApp { } } - fn apply_target_addr(&mut self) { - match self.settings_target_ip.trim().parse::() { - Ok(target_addr) => { - *self.control.shared.target_addr.lock() = target_addr; - PIPELINE_STATS - .lock() - .set_target_addr(target_addr.to_string()); - self.settings_target_error = None; - info!(target = %target_addr, "Transport target updated from GUI"); - self.persist_settings(); - } - Err(error) => { - self.settings_target_error = Some("Enter host:port".to_string()); - warn!(error = %error, target = %self.settings_target_ip, "Invalid target address"); - } - } - } - fn persist_settings(&self) { let encoder_override = ENCODER_CHOICES .iter() @@ -317,11 +282,9 @@ impl AnalyzerApp { let file = SettingsFile { bitrate_mbps: self.settings_bitrate_mbps, target_fps: self.settings_fps_target, - target_ip: if self.settings_target_ip.trim().is_empty() { - None - } else { - Some(self.settings_target_ip.trim().to_string()) - }, + // v2 has no manual target; the field stays in the file only so + // older settings.json still parse. + target_ip: None, encoder_override, capture_display: if self.settings_capture_display.is_empty() { None @@ -381,13 +344,6 @@ impl eframe::App for AnalyzerApp { ctx.set_visuals(visuals); let snap = StatsSnapshot::take(); - if self.settings_target_ip.is_empty() - && !snap.target_addr.is_empty() - && snap.target_addr != "0.0.0.0:9876" - { - self.settings_target_ip = snap.target_addr.clone(); - } - self.draw_sidebar(ctx, &snap); egui::CentralPanel::default() @@ -762,23 +718,12 @@ impl AnalyzerApp { ui.add_space(12.0); - // --- Target IP -------------------------------------------------------- - ui.horizontal(|ui| { - ui.label(egui::RichText::new("Target IP").color(TEXT).size(13.0)); - let response = ui.add( - egui::TextEdit::singleline(&mut self.settings_target_ip).desired_width(220.0), - ); - let enter_pressed = - response.lost_focus() && ui.input(|i| i.key_pressed(egui::Key::Enter)); - if amber_button(ui, "Apply").clicked() || enter_pressed { - self.apply_target_addr(); - } - }); - if let Some(error) = &self.settings_target_error { - ui.label(egui::RichText::new(error).color(RED).size(11.0)); - } - - ui.add_space(12.0); + // Protocol v2 has no manual target: media carries the session id + // minted by the HELLO2 handshake, so an address typed here could + // never receive a stream. The old field also wrote target_addr, + // which the capture loop read as "a client is watching" — enough + // to hold the virtual display up and keep capture at full rate + // with nobody connected. // --- Encoder override dropdown --------------------------------------- let detected = PIPELINE_STATS.lock().codec_name.clone(); @@ -962,12 +907,15 @@ impl AnalyzerApp { if self.settings_start_on_boot != prev_boot { if let Err(error) = set_startup_registry(self.settings_start_on_boot) { self.settings_start_on_boot = prev_boot; - self.settings_target_error = Some(error); + self.settings_error = Some(error); } else { - self.settings_target_error = None; + self.settings_error = None; self.persist_settings(); } } + if let Some(error) = &self.settings_error { + ui.label(egui::RichText::new(error).color(RED).size(11.0)); + } } }); diff --git a/host/src/pipeline.rs b/host/src/pipeline.rs index 73d3439..8c8cf0a 100644 --- a/host/src/pipeline.rs +++ b/host/src/pipeline.rs @@ -46,11 +46,15 @@ pub async fn run_pipeline_supervised( let capture_thread = std::thread::Builder::new() .name(format!("capture-g{generation}")) .spawn(move || { - let outcome = - match capture::run_capture_stage(frame_producer, capture_shared, adapter_index) { - Ok(()) => StageOutcome::Completed, - Err(e) => StageOutcome::Failed(e.to_string()), - }; + let outcome = match capture::run_capture_stage( + frame_producer, + capture_shared, + adapter_index, + generation, + ) { + Ok(()) => StageOutcome::Completed, + Err(e) => StageOutcome::Failed(e.to_string()), + }; capture_reporter.stage_exited(Stage::Capture, outcome); }); if let Err(e) = capture_thread { @@ -69,6 +73,7 @@ pub async fn run_pipeline_supervised( nal_tx, encoder_shared, encoder_gpu, + generation, ) { Ok(()) => StageOutcome::Completed, Err(e) => StageOutcome::Failed(e.to_string()), diff --git a/host/src/supervisor.rs b/host/src/supervisor.rs index 313e477..cf5b6bb 100644 --- a/host/src/supervisor.rs +++ b/host/src/supervisor.rs @@ -21,6 +21,19 @@ use crate::stats::PIPELINE_STATS; /// zombie thread can never confuse the supervisor about the live pipeline). pub static CURRENT_GENERATION: AtomicU64 = AtomicU64::new(0); +/// True while `generation` is still the live one. +/// +/// Stage threads are detached, so one blocked in a long driver call (the +/// virtual-display attach poll, the ACCESS_LOST retry ladder) can outlive its +/// generation. `shared.running` is process-wide and gets re-armed for the NEXT +/// generation, so a straggler that only checked that flag would wake up and +/// keep running forever — two capture loops fighting over a DXGI duplication +/// only one can own, while its heartbeats masked the watchdog that would have +/// noticed. +pub fn generation_is_current(generation: u64) -> bool { + CURRENT_GENERATION.load(Ordering::SeqCst) == generation +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Stage { Capture, @@ -365,10 +378,7 @@ fn wedge_reason(shared: &SharedControl, machine: &Machine) -> Option { // A client is connected but no frame has been produced for 5s. Sound // because the idle keepalive guarantees a frame at least every ~750ms // while a client is registered. - let client_connected = { - let addr = *shared.target_addr.lock(); - !addr.ip().is_unspecified() && addr.port() != 0 - }; + let client_connected = shared.client_connected(); if client_connected && frame_ms > 0 && now_ms.saturating_sub(frame_ms) > 5_000 { return Some(format!( "no captured frame for {}ms with a client connected", @@ -399,6 +409,19 @@ fn spawn_generation( command_tx: mpsc::Sender, ) -> std::thread::JoinHandle<()> { shared.running.store(true, Ordering::SeqCst); + // Give the new generation a full watchdog window. These atomics carry the + // PREVIOUS generation's timestamps, so without this the wedge check judges + // a thread that hasn't started yet: capture needing more than 3s to reach + // its first heartbeat (attaching the virtual display polls for up to 10s, + // and DXGI init is not instant on every driver) was declared wedged, + // restarted, and declared wedged again until the storm brake parked the + // supervisor in Failed — so the extended display could never come up. + { + let now_ms = crate::clock::host_now_us() / 1000; + shared.hb_capture_loop_ms.store(now_ms, Ordering::Relaxed); + shared.hb_capture_frame_ms.store(now_ms, Ordering::Relaxed); + shared.hb_encode_frame_ms.store(now_ms, Ordering::Relaxed); + } { let mut stats = PIPELINE_STATS.lock(); stats.mark_pipeline_started(); diff --git a/host/src/transport/mod.rs b/host/src/transport/mod.rs index cfdb07d..66cf0d9 100644 --- a/host/src/transport/mod.rs +++ b/host/src/transport/mod.rs @@ -163,8 +163,18 @@ pub async fn start_sender( if let Some(report) = actions.report.take() { let ceiling = shared.bitrate_bps.load(Ordering::SeqCst); - let _ = abr.set_ceiling(ceiling); - let decision = abr.on_report(&report, Instant::now()); + // A lowered ceiling clamps the rung + // immediately; that decision must be + // honored, or dragging the slider + // down changes the label and nothing + // else for the life of the pipeline. + let clamped = abr.set_ceiling(ceiling); + let reported = + abr.on_report(&report, Instant::now()); + let decision = abr::AbrDecision { + target_bps: reported.target_bps, + changed: clamped.changed || reported.changed, + }; if decision.changed { shared .abr_current_bps