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