From e2838640784709d508c3b52dfc1cc4754de353c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 13:38:41 +0000 Subject: [PATCH 1/5] fix(video): rebuild the pipeline when the HDMI source changes under it v4l2src negotiates the capture format once. When the HDMI source re-locks with different timings or a different colour format (host display sleep, GPU retrain, re-plug), the receiver starts writing a new frame layout while the pipeline keeps the old caps, so the encoder reads the planes wrong and viewers get green and magenta stripes with no error on the bus. The supervisor only restarted pipelines that had died, so this never healed. The supervisor now also: - polls the capture node (VIDIOC_G_FMT + VIDIOC_QUERY_DV_TIMINGS, read-only, one fd kept open) every 2s and rebuilds the pipeline when the format or timings differ from the start-of-pipeline baseline on two polls in a row; signal loss alone is not a change; - rebuilds at once when a capture buffer's size differs from the first one; - rebuilds when no raw frame arrives for 10s while a signal is present, backing off on repeats. Each rebuild logs the old and new source, and the negotiated capture caps are logged once per pipeline. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GRSn3XsjYLc494XwPez2mF --- AGENTS.md | 6 +- app/src/hardware/mod.rs | 1 + app/src/hardware/v4l2.rs | 235 ++++++++++++++++++++++++++++++++++++ app/src/pipeline.rs | 58 ++++++++- app/src/video.rs | 130 ++++++++++++++++++-- app/tests/capture_source.rs | 101 ++++++++++++++++ 6 files changed, 521 insertions(+), 10 deletions(-) create mode 100644 app/src/hardware/v4l2.rs create mode 100644 app/tests/capture_source.rs diff --git a/AGENTS.md b/AGENTS.md index 6c1a5d8..f35d845 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -111,7 +111,8 @@ lib.rs # module roots (see below) ├── webrtc.rs # PeerConnection factory, ICE, data channels (incl. cdcacm route) ├── video.rs # GStreamer + MPP pipeline; zero-copy `bytes::Bytes::from_owner(MappedBuffer)` # frames into WebRTC RTP — no per-frame memcpy - # supervisor restarts a dead pipeline (5s→300s backoff); `restartVideoPipeline` RPC + # supervisor rebuilds the pipeline when it dies, when the HDMI source's format/timings + # change under it, or when capture stalls (5s→300s backoff); `restartVideoPipeline` RPC ├── hardware/ │ ├── usb/ # HID (keyboard/mouse), storage (virtual media), gadget, descriptors │ ├── native/ # Ctrl socket (SEQPACKET), JSON-RPC bridge @@ -120,7 +121,8 @@ lib.rs # module roots (see below) │ ├── edid.rs # EDID parsing + persistence │ ├── hw.rs # Hardware probe / capability table, watchdog │ ├── jiggler.rs # Mouse jiggler -│ └── tuning.rs # RK3588 big-core CPU governor / IRQ affinity +│ ├── tuning.rs # RK3588 big-core CPU governor / IRQ affinity +│ └── v4l2.rs # Read-only G_FMT / QUERY_DV_TIMINGS on the capture node (source-change watch) ├── cloud/ # CloudManager singleton via `get_cloud_manager()`; OIDC, WebSocket relay ├── middleware.rs # Auth hoop (noPassword/password/cookie), dev mode ├── state.rs # AppState (sessions, sockets, ICE queue) diff --git a/app/src/hardware/mod.rs b/app/src/hardware/mod.rs index 9804452..e15bc92 100644 --- a/app/src/hardware/mod.rs +++ b/app/src/hardware/mod.rs @@ -6,3 +6,4 @@ pub mod jiggler; pub mod native; pub mod tuning; pub mod usb; +pub mod v4l2; diff --git a/app/src/hardware/v4l2.rs b/app/src/hardware/v4l2.rs new file mode 100644 index 0000000..bd1c8d4 --- /dev/null +++ b/app/src/hardware/v4l2.rs @@ -0,0 +1,235 @@ +//! Read-only queries of the HDMI capture node's live state, used to notice a +//! source change (resolution, timings or pixel layout) while the GStreamer +//! pipeline keeps streaming with the format it negotiated at start. + +use std::os::fd::OwnedFd; + +use anyhow::{Context, Result}; +use rustix::fs::{Mode, OFlags, open}; +use rustix::ioctl::{Getter, Opcode, Updater, ioctl, opcode}; + +const V4L2_BUF_TYPE_VIDEO_CAPTURE: u32 = 1; +const V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE: u32 = 9; + +/// `struct v4l2_format`: the union holds pointers, so it starts at offset 8 +/// and the struct is 208 bytes on 64-bit targets. +#[repr(C, align(8))] +struct V4l2Format { + buf_type: u32, + _pad: u32, + fmt: [u8; 200], +} + +/// `struct v4l2_dv_timings` (packed): a type and a 128-byte union whose +/// `v4l2_bt_timings` member is what HDMI receivers report. +#[repr(C)] +struct V4l2DvTimings { + timing_type: u32, + data: [u8; 128], +} + +const _: () = assert!(std::mem::size_of::() == 208); +const _: () = assert!(std::mem::size_of::() == 132); + +const VIDIOC_G_FMT: Opcode = opcode::read_write::(b'V', 4); +const VIDIOC_QUERY_DV_TIMINGS: Opcode = opcode::read::(b'V', 99); + +/// The pixel layout the driver currently writes into capture buffers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CaptureFormat { + pub width: u32, + pub height: u32, + pub fourcc: u32, + pub bytesperline: u32, + pub sizeimage: u32, +} + +impl CaptureFormat { + pub fn fourcc_str(&self) -> String { + self.fourcc + .to_le_bytes() + .iter() + .map(|&b| if b.is_ascii_graphic() { b as char } else { '?' }) + .collect() + } +} + +/// The signal timings the HDMI receiver currently detects. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DvTimings { + pub width: u32, + pub height: u32, + pub interlaced: bool, + pub pixelclock: u64, +} + +/// What the capture node reports at one point in time. `timings` is `None` +/// when no stable signal is present. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CaptureSignal { + pub format: Option, + pub timings: Option, +} + +impl std::fmt::Display for CaptureSignal { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self.format { + Some(c) => write!( + f, + "format {} {}x{} stride {} size {}", + c.fourcc_str(), + c.width, + c.height, + c.bytesperline, + c.sizeimage + )?, + None => write!(f, "format unknown")?, + } + match self.timings { + Some(t) => write!( + f, + ", signal {}x{}{} {:.2} MHz", + t.width, + t.height, + if t.interlaced { "i" } else { "p" }, + t.pixelclock as f64 / 1e6 + ), + None => write!(f, ", no signal"), + } + } +} + +fn u32_at(buf: &[u8], offset: usize) -> u32 { + let mut b = [0u8; 4]; + b.copy_from_slice(&buf[offset..offset + 4]); + u32::from_ne_bytes(b) +} + +fn u64_at(buf: &[u8], offset: usize) -> u64 { + let mut b = [0u8; 8]; + b.copy_from_slice(&buf[offset..offset + 8]); + u64::from_ne_bytes(b) +} + +/// Parses the `fmt` union of `struct v4l2_format` for a capture buffer type: +/// `v4l2_pix_format_mplane` (first plane) or `v4l2_pix_format`. +pub fn parse_pix_format(mplane: bool, fmt: &[u8; 200]) -> CaptureFormat { + let (bytesperline, sizeimage) = if mplane { + (u32_at(fmt, 24), u32_at(fmt, 20)) + } else { + (u32_at(fmt, 16), u32_at(fmt, 20)) + }; + CaptureFormat { + width: u32_at(fmt, 0), + height: u32_at(fmt, 4), + fourcc: u32_at(fmt, 8), + bytesperline, + sizeimage, + } +} + +/// Parses the `v4l2_bt_timings` member of `struct v4l2_dv_timings`. +pub fn parse_bt_timings(data: &[u8; 128]) -> DvTimings { + DvTimings { + width: u32_at(data, 0), + height: u32_at(data, 4), + interlaced: u32_at(data, 8) != 0, + pixelclock: u64_at(data, 16), + } +} + +/// A non-streaming handle on the capture node, kept open across polls. +pub struct CaptureProbe { + fd: OwnedFd, +} + +impl CaptureProbe { + pub fn open(device: &str) -> Result { + let fd = open(device, OFlags::RDWR | OFlags::NONBLOCK | OFlags::CLOEXEC, Mode::empty()) + .with_context(|| format!("open {device}"))?; + Ok(Self { fd }) + } + + pub fn query(&self) -> CaptureSignal { + CaptureSignal { format: self.format().ok(), timings: self.timings().ok() } + } + + fn format(&self) -> Result { + for (buf_type, mplane) in + [(V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE, true), (V4L2_BUF_TYPE_VIDEO_CAPTURE, false)] + { + let mut req = V4l2Format { buf_type, _pad: 0, fmt: [0; 200] }; + // SAFETY: VIDIOC_G_FMT's opcode is built from V4l2Format, whose + // layout matches the kernel's 208-byte struct v4l2_format; the + // kernel only writes within it. + let rc = unsafe { ioctl(&self.fd, Updater::::new(&mut req)) }; + if rc.is_ok() { + return Ok(parse_pix_format(mplane, &req.fmt)); + } + } + anyhow::bail!("VIDIOC_G_FMT failed for both capture buffer types") + } + + fn timings(&self) -> Result { + // SAFETY: VIDIOC_QUERY_DV_TIMINGS's opcode is built from + // V4l2DvTimings, which matches the kernel's packed 132-byte struct. + let t = unsafe { ioctl(&self.fd, Getter::::new()) } + .context("VIDIOC_QUERY_DV_TIMINGS")?; + if t.timing_type != 0 { + anyhow::bail!("unsupported DV timing type {}", t.timing_type); + } + Ok(parse_bt_timings(&t.data)) + } +} + +/// What one poll of the capture node means for the running pipeline. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SourceObservation { + /// First reading with a signal; it becomes the reference. + Baseline, + Steady, + NoSignal, + /// Differs from the reference once; confirmed on the next poll. + Pending, + /// Differed on two polls in a row: the pipeline no longer matches the + /// source and must be rebuilt. Carries the old reference. + Changed { + from: CaptureSignal, + }, +} + +/// Tracks the capture node's reported state against the state it had when +/// the pipeline started. Losing the signal is not a change by itself; the +/// signal coming back different is. +#[derive(Debug, Default)] +pub struct SourceWatch { + baseline: Option, + pending: Option, +} + +impl SourceWatch { + pub fn reset(&mut self) { + *self = Self::default(); + } + + pub fn observe(&mut self, now: CaptureSignal) -> SourceObservation { + if now.timings.is_none() { + self.pending = None; + return SourceObservation::NoSignal; + } + let Some(baseline) = self.baseline else { + self.baseline = Some(now); + return SourceObservation::Baseline; + }; + if now == baseline { + self.pending = None; + return SourceObservation::Steady; + } + if self.pending == Some(now) { + self.reset(); + return SourceObservation::Changed { from: baseline }; + } + self.pending = Some(now); + SourceObservation::Pending + } +} diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index 96fe1ba..acf4408 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, Ordering}; use anyhow::{Context, Result}; use gstreamer as gst; @@ -127,6 +127,12 @@ pub struct VideoPipeline { /// When the oldest unanswered keyframe request was made; the encoder /// output probe logs how long it took to produce the keyframe. keyframe_requested_at: Arc>>, + capture_src: gst::Element, + /// Raw frames the capture source has pushed, paused encoding included. + raw_frames: Arc, + /// Set when a capture buffer's size differs from the first one: the driver + /// now writes a different layout than the caps the pipeline negotiated. + capture_size_changed: Arc, } impl VideoPipeline { @@ -145,6 +151,7 @@ impl VideoPipeline { let pipeline = gst::Pipeline::new(); let v4l2src = gst::ElementFactory::make("v4l2src") + .name("capture_src") .property("device", &config.device) .build() .context("Failed to create v4l2src")?; @@ -293,6 +300,36 @@ impl VideoPipeline { ]) .context("Failed to link video pipeline elements")?; + // Counts raw frames for the stall check, and catches the driver changing + // the frame layout under negotiated caps (an HDMI source switching + // colour format or timings), which otherwise reaches the encoder as + // misread planes: green and magenta stripes rather than an error. + let raw_frames = Arc::new(AtomicU64::new(0)); + let capture_size_changed = Arc::new(AtomicBool::new(false)); + let probe_frames = raw_frames.clone(); + let probe_size_changed = capture_size_changed.clone(); + let first_size = AtomicUsize::new(0); + v4l2src.static_pad("src").context("v4l2src has no src pad")?.add_probe( + gst::PadProbeType::BUFFER, + move |_pad, info| { + probe_frames.fetch_add(1, Ordering::Relaxed); + if let Some(buffer) = info.buffer() { + let size = buffer.size(); + let first = first_size.load(Ordering::Relaxed); + if first == 0 { + first_size.store(size, Ordering::Relaxed); + } else if first != size && !probe_size_changed.swap(true, Ordering::AcqRel) { + warn!( + expected = first, + got = size, + "capture buffer size changed under negotiated caps" + ); + } + } + gst::PadProbeReturn::Ok + }, + ); + let raw_frame_waiters: Arc = Arc::default(); let raw_frame_requested = Arc::new(AtomicBool::new(false)); // The probe only takes a buffer ref when a snapshot is pending, so the @@ -358,6 +395,9 @@ impl VideoPipeline { raw_frame_waiters, raw_frame_requested, keyframe_requested_at, + capture_src: v4l2src, + raw_frames, + capture_size_changed, }) } @@ -886,6 +926,22 @@ impl PipelineManager { self.video.request_raw_frame() } + /// Raw frames captured since this pipeline was built. + pub fn raw_frames(&self) -> u64 { + self.video.raw_frames.load(Ordering::Relaxed) + } + + /// Whether the capture buffer size changed since start; see the probe in + /// [`VideoPipeline::new`]. + pub fn capture_size_changed(&self) -> bool { + self.video.capture_size_changed.load(Ordering::Acquire) + } + + /// The caps the capture source negotiated, once it has. + pub fn capture_caps(&self) -> Option { + self.video.capture_src.static_pad("src")?.current_caps().map(|c| c.to_string()) + } + /// Requests an IDR from the encoder. /// /// The request carries the current running time. An "as soon as possible" diff --git a/app/src/video.rs b/app/src/video.rs index f17a02e..05d3cb6 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -4,12 +4,13 @@ use std::sync::{Arc, OnceLock}; use prometheus::Encoder; use serde_json as serde_json_crate; use tokio::sync::{RwLock, mpsc}; -use tokio::time::{Duration, interval}; +use tokio::time::{Duration, Instant, interval}; use tracing::{debug, info, warn}; use webrtc::media::Sample; use webrtc::track::track_local::track_local_static_sample::TrackLocalStaticSample; use crate::api::{JsonRpcProcessor, default_registry}; +use crate::hardware::v4l2::{CaptureProbe, CaptureSignal, SourceObservation, SourceWatch}; use crate::pipeline::{AudioConfig, H264Profile, PipelineManager, VideoConfig}; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -62,7 +63,10 @@ static H264_PROFILE: parking_lot::Mutex> = parking_lot::Mute static PIPELINE_CONFIG: OnceLock<(VideoConfig, Option)> = OnceLock::new(); static SUPERVISOR_STOPPED: AtomicBool = AtomicBool::new(false); -const SUPERVISOR_PERIOD: Duration = Duration::from_secs(5); +const SUPERVISOR_PERIOD: Duration = Duration::from_secs(2); +/// No raw frame for this long while the receiver reports a signal means the +/// capture stream is wedged; repeat stalls back off like failed restarts. +const STALL_LIMIT: Duration = Duration::from_secs(10); const RESTART_BACKOFF_BASE: Duration = Duration::from_secs(5); const RESTART_BACKOFF_MAX: Duration = Duration::from_secs(300); @@ -286,23 +290,135 @@ pub fn restart_backoff(consecutive_failures: u32) -> Duration { .min(RESTART_BACKOFF_MAX) } -/// Restarts the pipeline whenever its bus loop has died (GStreamer error, EOS, or a -/// failed start at boot), backing off exponentially while restarts keep failing. +/// Per-pipeline state the supervisor keeps between ticks. +struct LiveCheck { + probe: Option>, + probe_error_logged: bool, + source: SourceWatch, + last_frames: u64, + last_progress: Instant, + stall_restarts: u32, + caps_logged: bool, +} + +impl LiveCheck { + fn new() -> Self { + Self { + probe: None, + probe_error_logged: false, + source: SourceWatch::default(), + last_frames: 0, + last_progress: Instant::now(), + stall_restarts: 0, + caps_logged: false, + } + } + + /// Forgets everything tied to the pipeline that was just replaced. + fn pipeline_replaced(&mut self) { + self.source.reset(); + self.last_frames = 0; + self.last_progress = Instant::now(); + self.caps_logged = false; + } + + async fn query_source(&mut self, device: &str) -> Option { + if self.probe.is_none() { + match CaptureProbe::open(device) { + Ok(p) => self.probe = Some(Arc::new(p)), + Err(e) => { + if !self.probe_error_logged { + warn!(error = %format!("{e:#}"), "cannot query capture node; source change detection off"); + self.probe_error_logged = true; + } + return None; + } + } + } + let probe = self.probe.clone()?; + tokio::task::spawn_blocking(move || probe.query()).await.ok() + } + + /// Returns why the running pipeline must be rebuilt, if it must. + async fn check(&mut self, device: &str) -> Option { + let (frames, size_changed, caps) = { + let guard = PIPELINE_MANAGER.lock().await; + let m = guard.as_ref()?; + (m.raw_frames(), m.capture_size_changed(), m.capture_caps()) + }; + if size_changed { + return Some("capture buffer size changed".to_owned()); + } + if !self.caps_logged + && let Some(caps) = caps + { + info!(%caps, "capture caps negotiated"); + if !caps.contains("NV12") { + warn!("capture source is not NV12; frames are converted in software"); + } + self.caps_logged = true; + } + + let signal = self.query_source(device).await; + if let Some(now) = signal { + match self.source.observe(now) { + SourceObservation::Baseline => info!(source = %now, "capture source baseline"), + SourceObservation::Pending => { + info!(source = %now, "capture source changed, confirming") + } + SourceObservation::Changed { from } => { + return Some(format!("capture source changed from [{from}] to [{now}]")); + } + SourceObservation::Steady | SourceObservation::NoSignal => {} + } + } + + if frames != self.last_frames { + self.last_frames = frames; + self.last_progress = Instant::now(); + self.stall_restarts = 0; + return None; + } + // Without a signal there is nothing to capture; wait for it to return. + if signal.is_some_and(|s| s.timings.is_none()) { + self.last_progress = Instant::now(); + return None; + } + let limit = STALL_LIMIT.max(restart_backoff(self.stall_restarts)); + let stalled = self.last_progress.elapsed(); + if stalled >= limit { + self.stall_restarts = self.stall_restarts.saturating_add(1); + return Some(format!("no raw frame for {}s with a signal present", stalled.as_secs())); + } + None + } +} + +/// Keeps the pipeline matched to the HDMI source. It rebuilds the pipeline +/// when its bus loop has died (GStreamer error, EOS, or a failed start at +/// boot), when the receiver's format or timings change under it (the encoder +/// would otherwise keep reading frames with the old layout), and when capture +/// stalls while a signal is present. Failed restarts back off exponentially. pub fn spawn_video_supervisor() { tokio::spawn(async move { let mut tick = interval(SUPERVISOR_PERIOD); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut failures = 0u32; + let mut live = LiveCheck::new(); loop { tick.tick().await; if SUPERVISOR_STOPPED.load(Ordering::SeqCst) { return; } - if PIPELINE_CONFIG.get().is_none() || is_video_pipeline_running().await { + let Some((video_config, _)) = PIPELINE_CONFIG.get() else { continue }; + if is_video_pipeline_running().await { failures = 0; - continue; + let Some(reason) = live.check(&video_config.device).await else { continue }; + warn!(%reason, "video pipeline no longer matches the source, rebuilding"); + } else { + warn!(failures, "video pipeline is down, restarting"); } - warn!(failures, "video pipeline is down, restarting"); + live.pipeline_replaced(); match restart_video_pipeline().await { Ok(()) => failures = 0, Err(e) => { diff --git a/app/tests/capture_source.rs b/app/tests/capture_source.rs new file mode 100644 index 0000000..4e1db83 --- /dev/null +++ b/app/tests/capture_source.rs @@ -0,0 +1,101 @@ +use rustkvm::hardware::v4l2::{ + CaptureFormat, CaptureSignal, DvTimings, SourceObservation, SourceWatch, parse_bt_timings, + parse_pix_format, +}; + +const NV12: u32 = u32::from_le_bytes(*b"NV12"); +const NV24: u32 = u32::from_le_bytes(*b"NV24"); + +fn put(buf: &mut [u8], offset: usize, value: u32) { + buf[offset..offset + 4].copy_from_slice(&value.to_ne_bytes()); +} + +fn signal(fourcc: u32, width: u32, height: u32) -> CaptureSignal { + CaptureSignal { + format: Some(CaptureFormat { + width, + height, + fourcc, + bytesperline: width, + sizeimage: width * height * 3 / 2, + }), + timings: Some(DvTimings { width, height, interlaced: false, pixelclock: 594_000_000 }), + } +} + +#[test] +fn parses_mplane_and_single_plane_formats() { + let mut fmt = [0u8; 200]; + put(&mut fmt, 0, 3840); + put(&mut fmt, 4, 2160); + put(&mut fmt, 8, NV12); + put(&mut fmt, 20, 12_441_600); + put(&mut fmt, 24, 3840); + let mp = parse_pix_format(true, &fmt); + assert_eq!( + (mp.width, mp.height, mp.bytesperline, mp.sizeimage), + (3840, 2160, 3840, 12_441_600) + ); + assert_eq!(mp.fourcc_str(), "NV12"); + + let mut fmt = [0u8; 200]; + put(&mut fmt, 0, 1920); + put(&mut fmt, 4, 1080); + put(&mut fmt, 8, NV12); + put(&mut fmt, 16, 1920); + put(&mut fmt, 20, 3_110_400); + let sp = parse_pix_format(false, &fmt); + assert_eq!((sp.width, sp.bytesperline, sp.sizeimage), (1920, 1920, 3_110_400)); +} + +#[test] +fn parses_bt_timings() { + let mut data = [0u8; 128]; + put(&mut data, 0, 3840); + put(&mut data, 4, 2160); + data[16..24].copy_from_slice(&594_000_000u64.to_ne_bytes()); + let t = parse_bt_timings(&data); + assert_eq!( + t, + DvTimings { width: 3840, height: 2160, interlaced: false, pixelclock: 594_000_000 } + ); +} + +#[test] +fn source_change_needs_two_matching_polls() { + let mut watch = SourceWatch::default(); + let before = signal(NV12, 3840, 2160); + let after = signal(NV24, 3840, 2160); + assert_eq!(watch.observe(before), SourceObservation::Baseline); + assert_eq!(watch.observe(before), SourceObservation::Steady); + assert_eq!(watch.observe(after), SourceObservation::Pending); + assert_eq!(watch.observe(after), SourceObservation::Changed { from: before }); + // The next reading after a change becomes the new reference. + assert_eq!(watch.observe(after), SourceObservation::Baseline); +} + +#[test] +fn a_single_odd_reading_is_not_a_change() { + let mut watch = SourceWatch::default(); + let normal = signal(NV12, 3840, 2160); + watch.observe(normal); + assert_eq!(watch.observe(signal(NV12, 1920, 1080)), SourceObservation::Pending); + assert_eq!(watch.observe(normal), SourceObservation::Steady); + assert_eq!(watch.observe(signal(NV12, 1920, 1080)), SourceObservation::Pending); +} + +#[test] +fn signal_loss_alone_does_not_trigger_but_a_different_return_does() { + let mut watch = SourceWatch::default(); + let before = signal(NV12, 3840, 2160); + let lost = CaptureSignal { format: before.format, timings: None }; + watch.observe(before); + assert_eq!(watch.observe(lost), SourceObservation::NoSignal); + assert_eq!(watch.observe(lost), SourceObservation::NoSignal); + assert_eq!(watch.observe(before), SourceObservation::Steady); + + let after = signal(NV24, 3840, 2160); + assert_eq!(watch.observe(lost), SourceObservation::NoSignal); + assert_eq!(watch.observe(after), SourceObservation::Pending); + assert_eq!(watch.observe(after), SourceObservation::Changed { from: before }); +} From 5ccc618fe82ca0726f148544892f1141e368a8e9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 14:19:10 +0000 Subject: [PATCH 2/5] fix(video): ignore pixel-clock jitter when comparing the capture source On hardware the receiver reports 593.95 and 593.96 MHz alternately on a steady 4K60 signal, so the source watch rebuilt the pipeline every 30 seconds. Compare only the frame layout and video mode. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GRSn3XsjYLc494XwPez2mF --- app/src/hardware/v4l2.rs | 17 +++++++++++++++-- app/tests/capture_source.rs | 29 +++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/app/src/hardware/v4l2.rs b/app/src/hardware/v4l2.rs index bd1c8d4..3ee4963 100644 --- a/app/src/hardware/v4l2.rs +++ b/app/src/hardware/v4l2.rs @@ -71,6 +71,19 @@ pub struct CaptureSignal { pub timings: Option, } +impl CaptureSignal { + /// Whether both readings describe the same frame layout and video mode. + /// The pixel clock is left out: the receiver's measurement wobbles by a + /// few kHz on a steady signal (593.95 vs 593.96 MHz on the T6), and + /// counting that as a change rebuilt the pipeline every 30 seconds. + pub fn same_source(&self, other: &Self) -> bool { + let layout = |f: &CaptureFormat| (f.fourcc, f.width, f.height, f.bytesperline); + let mode = |t: &DvTimings| (t.width, t.height, t.interlaced); + self.format.as_ref().map(layout) == other.format.as_ref().map(layout) + && self.timings.as_ref().map(mode) == other.timings.as_ref().map(mode) + } +} + impl std::fmt::Display for CaptureSignal { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self.format { @@ -221,11 +234,11 @@ impl SourceWatch { self.baseline = Some(now); return SourceObservation::Baseline; }; - if now == baseline { + if now.same_source(&baseline) { self.pending = None; return SourceObservation::Steady; } - if self.pending == Some(now) { + if self.pending.is_some_and(|p| p.same_source(&now)) { self.reset(); return SourceObservation::Changed { from: baseline }; } diff --git a/app/tests/capture_source.rs b/app/tests/capture_source.rs index 4e1db83..bcdf087 100644 --- a/app/tests/capture_source.rs +++ b/app/tests/capture_source.rs @@ -99,3 +99,32 @@ fn signal_loss_alone_does_not_trigger_but_a_different_return_does() { assert_eq!(watch.observe(after), SourceObservation::Pending); assert_eq!(watch.observe(after), SourceObservation::Changed { from: before }); } + +#[test] +fn pixel_clock_wobble_is_not_a_change() { + let mut watch = SourceWatch::default(); + let mut a = signal(NV12, 3840, 2160); + let mut b = a; + if let (Some(ta), Some(tb)) = (a.timings.as_mut(), b.timings.as_mut()) { + ta.pixelclock = 593_950_000; + tb.pixelclock = 593_960_000; + } + assert_eq!(watch.observe(a), SourceObservation::Baseline); + for _ in 0..4 { + assert_eq!(watch.observe(b), SourceObservation::Steady); + assert_eq!(watch.observe(a), SourceObservation::Steady); + } +} + +#[test] +fn a_one_off_bogus_mode_is_not_a_change() { + let mut watch = SourceWatch::default(); + let normal = signal(NV12, 3840, 2160); + let mut bogus = normal; + if let Some(t) = bogus.timings.as_mut() { + t.height = 547; + } + watch.observe(normal); + assert_eq!(watch.observe(bogus), SourceObservation::Pending); + assert_eq!(watch.observe(normal), SourceObservation::Steady); +} From f92cfc07367de695dc5b07b6b83c07245edfd0f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 14:35:23 +0000 Subject: [PATCH 3/5] fix(video): replug HDMI in software when the receiver is stuck relocking Hardware showed the real failure behind the striped picture: after the host woke from a 15-hour sleep, the rockchip_hdmirx receiver looped once a second. Every lock ends in hdmirx_format_change, which queues a source-change event even when the mode is unchanged; v4l2src restarts the stream on each event, and the restart relocks and queues the next one. Frames are cut mid-DMA, and only a physical replug ended it. Format and timings never changed, so the source-change rebuild from the previous commits never fired, and on hardware it misfired on pixel-clock jitter. Replace it with what actually recovers the input: - subscribe to V4L2 source-change events on a monitoring fd, counted on a small thread (the kernel keeps only the newest pending event); - 6+ events within 10 s is a relock storm: pulse hot-plug detect through /sys/class/hdmirx/hdmirx/status (off, 2 s, on) so the source renegotiates, backing off 30 s, 60 s, ... while the storm persists; - drop the format/timing comparison and the buffer-size check; keep the stall rebuild and the negotiated-caps log. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GRSn3XsjYLc494XwPez2mF --- AGENTS.md | 6 +- app/src/hardware/v4l2.rs | 218 +++++++++++++++++++++++++++--------- app/src/pipeline.rs | 35 +----- app/src/video.rs | 125 +++++++++++++++------ app/tests/capture_source.rs | 113 +++++++------------ 5 files changed, 296 insertions(+), 201 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index f35d845..31e8a9a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -111,8 +111,8 @@ lib.rs # module roots (see below) ├── webrtc.rs # PeerConnection factory, ICE, data channels (incl. cdcacm route) ├── video.rs # GStreamer + MPP pipeline; zero-copy `bytes::Bytes::from_owner(MappedBuffer)` # frames into WebRTC RTP — no per-frame memcpy - # supervisor rebuilds the pipeline when it dies, when the HDMI source's format/timings - # change under it, or when capture stalls (5s→300s backoff); `restartVideoPipeline` RPC + # supervisor rebuilds the pipeline when it dies or capture stalls (5s→300s backoff), + # and replugs HDMI (HPD pulse) on a receiver relock storm; `restartVideoPipeline` RPC ├── hardware/ │ ├── usb/ # HID (keyboard/mouse), storage (virtual media), gadget, descriptors │ ├── native/ # Ctrl socket (SEQPACKET), JSON-RPC bridge @@ -122,7 +122,7 @@ lib.rs # module roots (see below) │ ├── hw.rs # Hardware probe / capability table, watchdog │ ├── jiggler.rs # Mouse jiggler │ ├── tuning.rs # RK3588 big-core CPU governor / IRQ affinity -│ └── v4l2.rs # Read-only G_FMT / QUERY_DV_TIMINGS on the capture node (source-change watch) +│ └── v4l2.rs # Capture-node G_FMT/QUERY_DV_TIMINGS, source-change events, hdmirx HPD pulse ├── cloud/ # CloudManager singleton via `get_cloud_manager()`; OIDC, WebSocket relay ├── middleware.rs # Auth hoop (noPassword/password/cookie), dev mode ├── state.rs # AppState (sessions, sockets, ICE queue) diff --git a/app/src/hardware/v4l2.rs b/app/src/hardware/v4l2.rs index 3ee4963..b77ce3a 100644 --- a/app/src/hardware/v4l2.rs +++ b/app/src/hardware/v4l2.rs @@ -1,12 +1,20 @@ -//! Read-only queries of the HDMI capture node's live state, used to notice a -//! source change (resolution, timings or pixel layout) while the GStreamer -//! pipeline keeps streaming with the format it negotiated at start. +//! The HDMI receiver's live state and its one recovery lever. +//! +//! After the host wakes from a long sleep the RK3588 receiver can fall into a +//! relock loop: every lock ends in `hdmirx_format_change`, which queues a +//! V4L2 source-change event even when nothing changed; v4l2src answers each +//! event by restarting the stream; the restart relocks and queues the next +//! event. The loop runs about once a second, frames are cut mid-DMA (green +//! and magenta stripes), and it does not end on its own. A replug ends it, so +//! the recovery is a software replug: drop hot-plug detect and raise it again. +use std::collections::VecDeque; use std::os::fd::OwnedFd; +use std::time::{Duration, Instant}; use anyhow::{Context, Result}; use rustix::fs::{Mode, OFlags, open}; -use rustix::ioctl::{Getter, Opcode, Updater, ioctl, opcode}; +use rustix::ioctl::{Getter, Opcode, Setter, Updater, ioctl, opcode}; const V4L2_BUF_TYPE_VIDEO_CAPTURE: u32 = 1; const V4L2_BUF_TYPE_VIDEO_CAPTURE_MPLANE: u32 = 9; @@ -28,8 +36,41 @@ struct V4l2DvTimings { data: [u8; 128], } +/// `struct v4l2_event_subscription`. +#[repr(C)] +struct V4l2EventSubscription { + ev_type: u32, + id: u32, + flags: u32, + reserved: [u32; 5], +} + +/// `struct v4l2_event`: the union holds an `s64`, so it starts at offset 8 +/// and the struct is 136 bytes on 64-bit targets. +#[repr(C, align(8))] +struct V4l2Event { + ev_type: u32, + _pad: u32, + u: [u8; 64], + pending: u32, + sequence: u32, + timestamp: [i64; 2], + id: u32, + reserved: [u32; 8], +} + const _: () = assert!(std::mem::size_of::() == 208); const _: () = assert!(std::mem::size_of::() == 132); +const _: () = assert!(std::mem::size_of::() == 32); +const _: () = assert!(std::mem::size_of::() == 136); + +const V4L2_EVENT_SOURCE_CHANGE: u32 = 5; +const VIDIOC_DQEVENT: Opcode = opcode::read::(b'V', 89); +const VIDIOC_SUBSCRIBE_EVENT: Opcode = opcode::write::(b'V', 90); + +/// The rockchip_hdmirx sysfs node: writing `off` drops hot-plug detect and +/// `on` raises it again. +const HDMIRX_HPD_NODE: &str = "/sys/class/hdmirx/hdmirx/status"; const VIDIOC_G_FMT: Opcode = opcode::read_write::(b'V', 4); const VIDIOC_QUERY_DV_TIMINGS: Opcode = opcode::read::(b'V', 99); @@ -71,19 +112,6 @@ pub struct CaptureSignal { pub timings: Option, } -impl CaptureSignal { - /// Whether both readings describe the same frame layout and video mode. - /// The pixel clock is left out: the receiver's measurement wobbles by a - /// few kHz on a steady signal (593.95 vs 593.96 MHz on the T6), and - /// counting that as a change rebuilt the pipeline every 30 seconds. - pub fn same_source(&self, other: &Self) -> bool { - let layout = |f: &CaptureFormat| (f.fourcc, f.width, f.height, f.bytesperline); - let mode = |t: &DvTimings| (t.width, t.height, t.interlaced); - self.format.as_ref().map(layout) == other.format.as_ref().map(layout) - && self.timings.as_ref().map(mode) == other.timings.as_ref().map(mode) - } -} - impl std::fmt::Display for CaptureSignal { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self.format { @@ -151,16 +179,57 @@ pub fn parse_bt_timings(data: &[u8; 128]) -> DvTimings { } } -/// A non-streaming handle on the capture node, kept open across polls. +/// A non-streaming handle on the capture node, kept open across polls. It +/// subscribes to source-change events so it sees every relock the driver +/// reports, alongside whatever is streaming. pub struct CaptureProbe { fd: OwnedFd, + events: bool, } impl CaptureProbe { pub fn open(device: &str) -> Result { let fd = open(device, OFlags::RDWR | OFlags::NONBLOCK | OFlags::CLOEXEC, Mode::empty()) .with_context(|| format!("open {device}"))?; - Ok(Self { fd }) + let sub = V4l2EventSubscription { + ev_type: V4L2_EVENT_SOURCE_CHANGE, + id: 0, + flags: 0, + reserved: [0; 5], + }; + // SAFETY: VIDIOC_SUBSCRIBE_EVENT's opcode is built from + // V4l2EventSubscription, which matches the kernel's 32-byte struct; + // the kernel only reads it. + let events = unsafe { + ioctl(&fd, Setter::::new(sub)) + } + .is_ok(); + Ok(Self { fd, events }) + } + + /// Whether the driver accepted the source-change subscription. + pub fn has_events(&self) -> bool { + self.events + } + + /// Dequeues and counts pending source-change events. The kernel keeps + /// only the newest one per subscription, so this counts at most one per + /// call unless several arrived in between; callers poll often enough. + pub fn take_source_changes(&self) -> u32 { + if !self.events { + return 0; + } + let mut n = 0; + while n < 1024 { + // SAFETY: VIDIOC_DQEVENT's opcode is built from V4l2Event, which + // matches the kernel's 136-byte struct v4l2_event. + match unsafe { ioctl(&self.fd, Getter::::new()) } { + Ok(ev) if ev.ev_type == V4L2_EVENT_SOURCE_CHANGE => n += 1, + Ok(_) => {} + Err(_) => break, + } + } + n } pub fn query(&self) -> CaptureSignal { @@ -195,54 +264,91 @@ impl CaptureProbe { } } -/// What one poll of the capture node means for the running pipeline. +/// Drops hot-plug detect for `low_for`, then raises it, so the source sees an +/// unplug and renegotiates the link from scratch. HPD is always raised again, +/// with retries: leaving it low would blank the input for good. +pub fn pulse_hotplug(low_for: Duration) -> Result<()> { + std::fs::write(HDMIRX_HPD_NODE, "off") + .with_context(|| format!("write off to {HDMIRX_HPD_NODE}"))?; + std::thread::sleep(low_for); + let mut last = None; + for _ in 0..5 { + match std::fs::write(HDMIRX_HPD_NODE, "on") { + Ok(()) => return Ok(()), + Err(e) => last = Some(e), + } + std::thread::sleep(Duration::from_millis(200)); + } + match last { + Some(e) => Err(e).with_context(|| format!("write on to {HDMIRX_HPD_NODE}")), + None => Ok(()), + } +} + +/// What the latest source-change events mean. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SourceObservation { - /// First reading with a signal; it becomes the reference. - Baseline, - Steady, - NoSignal, - /// Differs from the reference once; confirmed on the next poll. - Pending, - /// Differed on two polls in a row: the pipeline no longer matches the - /// source and must be rebuilt. Carries the old reference. - Changed { - from: CaptureSignal, +pub enum RelockVerdict { + Quiet, + /// A relock storm, but the last replug was too recent to try again. + Storm { + events: u32, + }, + /// A relock storm: replug now. `attempt` counts replugs in this episode. + Replug { + events: u32, + attempt: u32, }, } -/// Tracks the capture node's reported state against the state it had when -/// the pipeline started. Losing the signal is not a change by itself; the -/// signal coming back different is. +/// Recognises the relock loop from source-change events. A healthy input +/// reports none, a real mode change or a replug a few; the loop reports one +/// a second without end. Replugs back off while the loop persists. #[derive(Debug, Default)] -pub struct SourceWatch { - baseline: Option, - pending: Option, +pub struct RelockWatch { + recent: VecDeque<(Instant, u32)>, + last_replug: Option, + attempts: u32, } -impl SourceWatch { - pub fn reset(&mut self) { - *self = Self::default(); +impl RelockWatch { + /// Window the events are counted over. + pub const WINDOW: Duration = Duration::from_secs(10); + /// Events within the window that make a storm. + pub const THRESHOLD: u32 = 6; + /// Quiet time after which an episode is over and backoff resets. + pub const SETTLE: Duration = Duration::from_secs(300); + + fn backoff(attempts: u32) -> Duration { + Duration::from_secs(30) + .saturating_mul(1u32 << attempts.min(5)) + .min(Duration::from_secs(600)) } - pub fn observe(&mut self, now: CaptureSignal) -> SourceObservation { - if now.timings.is_none() { - self.pending = None; - return SourceObservation::NoSignal; + pub fn observe(&mut self, now: Instant, events: u32) -> RelockVerdict { + if events > 0 { + self.recent.push_back((now, events)); } - let Some(baseline) = self.baseline else { - self.baseline = Some(now); - return SourceObservation::Baseline; - }; - if now.same_source(&baseline) { - self.pending = None; - return SourceObservation::Steady; + while self.recent.front().is_some_and(|&(t, _)| now.duration_since(t) > Self::WINDOW) { + self.recent.pop_front(); + } + let total: u32 = self.recent.iter().map(|&(_, n)| n).sum(); + if total < Self::THRESHOLD { + if self.last_replug.is_some_and(|t| now.duration_since(t) >= Self::SETTLE) { + self.attempts = 0; + self.last_replug = None; + } + return RelockVerdict::Quiet; } - if self.pending.is_some_and(|p| p.same_source(&now)) { - self.reset(); - return SourceObservation::Changed { from: baseline }; + let due = self.attempts == 0 + || self + .last_replug + .is_none_or(|t| now.duration_since(t) >= Self::backoff(self.attempts - 1)); + if !due { + return RelockVerdict::Storm { events: total }; } - self.pending = Some(now); - SourceObservation::Pending + self.attempts += 1; + self.last_replug = Some(now); + self.recent.clear(); + RelockVerdict::Replug { events: total, attempt: self.attempts } } } diff --git a/app/src/pipeline.rs b/app/src/pipeline.rs index acf4408..78a470c 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -1,5 +1,5 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering}; use anyhow::{Context, Result}; use gstreamer as gst; @@ -130,9 +130,6 @@ pub struct VideoPipeline { capture_src: gst::Element, /// Raw frames the capture source has pushed, paused encoding included. raw_frames: Arc, - /// Set when a capture buffer's size differs from the first one: the driver - /// now writes a different layout than the caps the pipeline negotiated. - capture_size_changed: Arc, } impl VideoPipeline { @@ -300,32 +297,13 @@ impl VideoPipeline { ]) .context("Failed to link video pipeline elements")?; - // Counts raw frames for the stall check, and catches the driver changing - // the frame layout under negotiated caps (an HDMI source switching - // colour format or timings), which otherwise reaches the encoder as - // misread planes: green and magenta stripes rather than an error. + // Counts raw frames for the supervisor's stall check. let raw_frames = Arc::new(AtomicU64::new(0)); - let capture_size_changed = Arc::new(AtomicBool::new(false)); let probe_frames = raw_frames.clone(); - let probe_size_changed = capture_size_changed.clone(); - let first_size = AtomicUsize::new(0); v4l2src.static_pad("src").context("v4l2src has no src pad")?.add_probe( gst::PadProbeType::BUFFER, - move |_pad, info| { + move |_pad, _info| { probe_frames.fetch_add(1, Ordering::Relaxed); - if let Some(buffer) = info.buffer() { - let size = buffer.size(); - let first = first_size.load(Ordering::Relaxed); - if first == 0 { - first_size.store(size, Ordering::Relaxed); - } else if first != size && !probe_size_changed.swap(true, Ordering::AcqRel) { - warn!( - expected = first, - got = size, - "capture buffer size changed under negotiated caps" - ); - } - } gst::PadProbeReturn::Ok }, ); @@ -397,7 +375,6 @@ impl VideoPipeline { keyframe_requested_at, capture_src: v4l2src, raw_frames, - capture_size_changed, }) } @@ -931,12 +908,6 @@ impl PipelineManager { self.video.raw_frames.load(Ordering::Relaxed) } - /// Whether the capture buffer size changed since start; see the probe in - /// [`VideoPipeline::new`]. - pub fn capture_size_changed(&self) -> bool { - self.video.capture_size_changed.load(Ordering::Acquire) - } - /// The caps the capture source negotiated, once it has. pub fn capture_caps(&self) -> Option { self.video.capture_src.static_pad("src")?.current_caps().map(|c| c.to_string()) diff --git a/app/src/video.rs b/app/src/video.rs index 05d3cb6..612d3c0 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -1,4 +1,4 @@ -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, OnceLock}; use prometheus::Encoder; @@ -10,7 +10,7 @@ use webrtc::media::Sample; use webrtc::track::track_local::track_local_static_sample::TrackLocalStaticSample; use crate::api::{JsonRpcProcessor, default_registry}; -use crate::hardware::v4l2::{CaptureProbe, CaptureSignal, SourceObservation, SourceWatch}; +use crate::hardware::v4l2::{CaptureProbe, RelockVerdict, RelockWatch, pulse_hotplug}; use crate::pipeline::{AudioConfig, H264Profile, PipelineManager, VideoConfig}; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -290,11 +290,16 @@ pub fn restart_backoff(consecutive_failures: u32) -> Duration { .min(RESTART_BACKOFF_MAX) } -/// Per-pipeline state the supervisor keeps between ticks. +/// How long hot-plug detect stays low on a software replug. Long enough for +/// the source to register an unplug. +const REPLUG_LOW: Duration = Duration::from_secs(2); + +/// State the supervisor keeps between ticks. struct LiveCheck { probe: Option>, probe_error_logged: bool, - source: SourceWatch, + relock_events: Option>, + relock: RelockWatch, last_frames: u64, last_progress: Instant, stall_restarts: u32, @@ -306,7 +311,8 @@ impl LiveCheck { Self { probe: None, probe_error_logged: false, - source: SourceWatch::default(), + relock_events: None, + relock: RelockWatch::default(), last_frames: 0, last_progress: Instant::now(), stall_restarts: 0, @@ -316,39 +322,69 @@ impl LiveCheck { /// Forgets everything tied to the pipeline that was just replaced. fn pipeline_replaced(&mut self) { - self.source.reset(); self.last_frames = 0; self.last_progress = Instant::now(); self.caps_logged = false; } - async fn query_source(&mut self, device: &str) -> Option { + fn probe(&mut self, device: &str) -> Option> { if self.probe.is_none() { match CaptureProbe::open(device) { - Ok(p) => self.probe = Some(Arc::new(p)), + Ok(p) => { + let p = Arc::new(p); + if p.has_events() { + info!("watching the HDMI receiver for relock storms"); + self.relock_events = Some(spawn_relock_counter(p.clone())); + } else { + warn!( + "capture node has no source-change events; relock storms go undetected" + ); + } + self.probe = Some(p); + } Err(e) => { if !self.probe_error_logged { - warn!(error = %format!("{e:#}"), "cannot query capture node; source change detection off"); + warn!(error = %format!("{e:#}"), "cannot open capture node for monitoring"); self.probe_error_logged = true; } - return None; } } } - let probe = self.probe.clone()?; - tokio::task::spawn_blocking(move || probe.query()).await.ok() + self.probe.clone() + } + + /// Replugs the source when the receiver is stuck relocking. + async fn check_relock(&mut self) { + let Some(counter) = &self.relock_events else { return }; + let events = counter.swap(0, Ordering::Relaxed); + if events > 0 { + info!(events, "HDMI receiver reported a source change"); + } + match self.relock.observe(std::time::Instant::now(), events) { + RelockVerdict::Quiet => {} + RelockVerdict::Storm { events } => { + debug!(events, "HDMI receiver still relocking; waiting before the next replug"); + } + RelockVerdict::Replug { events, attempt } => { + warn!(events, attempt, "HDMI receiver is stuck relocking, replugging the source"); + match tokio::task::spawn_blocking(|| pulse_hotplug(REPLUG_LOW)).await { + Ok(Ok(())) => info!("software replug done"), + Ok(Err(e)) => warn!(error = %format!("{e:#}"), "software replug failed"), + Err(e) => warn!(error = %e, "software replug task failed"), + } + // The signal drops while HPD is low; don't count that as a stall. + self.last_progress = Instant::now(); + } + } } /// Returns why the running pipeline must be rebuilt, if it must. async fn check(&mut self, device: &str) -> Option { - let (frames, size_changed, caps) = { + let (frames, caps) = { let guard = PIPELINE_MANAGER.lock().await; let m = guard.as_ref()?; - (m.raw_frames(), m.capture_size_changed(), m.capture_caps()) + (m.raw_frames(), m.capture_caps()) }; - if size_changed { - return Some("capture buffer size changed".to_owned()); - } if !self.caps_logged && let Some(caps) = caps { @@ -359,19 +395,8 @@ impl LiveCheck { self.caps_logged = true; } - let signal = self.query_source(device).await; - if let Some(now) = signal { - match self.source.observe(now) { - SourceObservation::Baseline => info!(source = %now, "capture source baseline"), - SourceObservation::Pending => { - info!(source = %now, "capture source changed, confirming") - } - SourceObservation::Changed { from } => { - return Some(format!("capture source changed from [{from}] to [{now}]")); - } - SourceObservation::Steady | SourceObservation::NoSignal => {} - } - } + let probe = self.probe(device); + self.check_relock().await; if frames != self.last_frames { self.last_frames = frames; @@ -379,6 +404,10 @@ impl LiveCheck { self.stall_restarts = 0; return None; } + let signal = match probe { + Some(p) => tokio::task::spawn_blocking(move || p.query()).await.ok(), + None => None, + }; // Without a signal there is nothing to capture; wait for it to return. if signal.is_some_and(|s| s.timings.is_none()) { self.last_progress = Instant::now(); @@ -388,17 +417,39 @@ impl LiveCheck { let stalled = self.last_progress.elapsed(); if stalled >= limit { self.stall_restarts = self.stall_restarts.saturating_add(1); - return Some(format!("no raw frame for {}s with a signal present", stalled.as_secs())); + let source = signal.map_or_else(|| "unknown".to_owned(), |s| s.to_string()); + return Some(format!("no raw frame for {}s, source {source}", stalled.as_secs())); } None } } -/// Keeps the pipeline matched to the HDMI source. It rebuilds the pipeline -/// when its bus loop has died (GStreamer error, EOS, or a failed start at -/// boot), when the receiver's format or timings change under it (the encoder -/// would otherwise keep reading frames with the old layout), and when capture -/// stalls while a signal is present. Failed restarts back off exponentially. +/// Counts the receiver's source-change events on a thread of its own: the +/// kernel keeps only the newest pending event, so a 2 s supervisor tick +/// would see at most one of a once-a-second storm. +fn spawn_relock_counter(probe: Arc) -> Arc { + let count = Arc::new(AtomicU32::new(0)); + let out = count.clone(); + let spawned = std::thread::Builder::new().name("hdmirx-events".into()).spawn(move || { + while !SUPERVISOR_STOPPED.load(Ordering::Relaxed) { + let n = probe.take_source_changes(); + if n > 0 { + count.fetch_add(n, Ordering::Relaxed); + } + std::thread::sleep(Duration::from_millis(200)); + } + }); + if let Err(e) = spawned { + warn!(error = %e, "cannot start the HDMI event thread"); + } + out +} + +/// Keeps the capture pipeline healthy. It rebuilds the pipeline when its bus +/// loop has died (GStreamer error, EOS, or a failed start at boot) or when +/// capture stalls while a signal is present, and replugs the HDMI source in +/// software when the receiver is stuck in a relock loop. Failed restarts back +/// off exponentially. pub fn spawn_video_supervisor() { tokio::spawn(async move { let mut tick = interval(SUPERVISOR_PERIOD); @@ -414,7 +465,7 @@ pub fn spawn_video_supervisor() { if is_video_pipeline_running().await { failures = 0; let Some(reason) = live.check(&video_config.device).await else { continue }; - warn!(%reason, "video pipeline no longer matches the source, rebuilding"); + warn!(%reason, "video capture stalled, rebuilding the pipeline"); } else { warn!(failures, "video pipeline is down, restarting"); } diff --git a/app/tests/capture_source.rs b/app/tests/capture_source.rs index bcdf087..3e1389d 100644 --- a/app/tests/capture_source.rs +++ b/app/tests/capture_source.rs @@ -1,28 +1,15 @@ +use std::time::{Duration, Instant}; + use rustkvm::hardware::v4l2::{ - CaptureFormat, CaptureSignal, DvTimings, SourceObservation, SourceWatch, parse_bt_timings, - parse_pix_format, + DvTimings, RelockVerdict, RelockWatch, parse_bt_timings, parse_pix_format, }; const NV12: u32 = u32::from_le_bytes(*b"NV12"); -const NV24: u32 = u32::from_le_bytes(*b"NV24"); fn put(buf: &mut [u8], offset: usize, value: u32) { buf[offset..offset + 4].copy_from_slice(&value.to_ne_bytes()); } -fn signal(fourcc: u32, width: u32, height: u32) -> CaptureSignal { - CaptureSignal { - format: Some(CaptureFormat { - width, - height, - fourcc, - bytesperline: width, - sizeimage: width * height * 3 / 2, - }), - timings: Some(DvTimings { width, height, interlaced: false, pixelclock: 594_000_000 }), - } -} - #[test] fn parses_mplane_and_single_plane_formats() { let mut fmt = [0u8; 200]; @@ -61,70 +48,50 @@ fn parses_bt_timings() { ); } -#[test] -fn source_change_needs_two_matching_polls() { - let mut watch = SourceWatch::default(); - let before = signal(NV12, 3840, 2160); - let after = signal(NV24, 3840, 2160); - assert_eq!(watch.observe(before), SourceObservation::Baseline); - assert_eq!(watch.observe(before), SourceObservation::Steady); - assert_eq!(watch.observe(after), SourceObservation::Pending); - assert_eq!(watch.observe(after), SourceObservation::Changed { from: before }); - // The next reading after a change becomes the new reference. - assert_eq!(watch.observe(after), SourceObservation::Baseline); -} - -#[test] -fn a_single_odd_reading_is_not_a_change() { - let mut watch = SourceWatch::default(); - let normal = signal(NV12, 3840, 2160); - watch.observe(normal); - assert_eq!(watch.observe(signal(NV12, 1920, 1080)), SourceObservation::Pending); - assert_eq!(watch.observe(normal), SourceObservation::Steady); - assert_eq!(watch.observe(signal(NV12, 1920, 1080)), SourceObservation::Pending); +/// Feeds one event per second, like the relock loop seen on the T6, and +/// returns the verdict of each second. +fn storm(watch: &mut RelockWatch, start: Instant, secs: u64) -> Vec { + (0..secs).map(|s| watch.observe(start + Duration::from_secs(s), 1)).collect() } #[test] -fn signal_loss_alone_does_not_trigger_but_a_different_return_does() { - let mut watch = SourceWatch::default(); - let before = signal(NV12, 3840, 2160); - let lost = CaptureSignal { format: before.format, timings: None }; - watch.observe(before); - assert_eq!(watch.observe(lost), SourceObservation::NoSignal); - assert_eq!(watch.observe(lost), SourceObservation::NoSignal); - assert_eq!(watch.observe(before), SourceObservation::Steady); - - let after = signal(NV24, 3840, 2160); - assert_eq!(watch.observe(lost), SourceObservation::NoSignal); - assert_eq!(watch.observe(after), SourceObservation::Pending); - assert_eq!(watch.observe(after), SourceObservation::Changed { from: before }); +fn a_replug_or_mode_change_is_not_a_storm() { + let mut watch = RelockWatch::default(); + let t = Instant::now(); + for s in 0..3 { + assert_eq!(watch.observe(t + Duration::from_secs(s), 1), RelockVerdict::Quiet); + } + for s in 3..60 { + assert_eq!(watch.observe(t + Duration::from_secs(s), 0), RelockVerdict::Quiet); + } } #[test] -fn pixel_clock_wobble_is_not_a_change() { - let mut watch = SourceWatch::default(); - let mut a = signal(NV12, 3840, 2160); - let mut b = a; - if let (Some(ta), Some(tb)) = (a.timings.as_mut(), b.timings.as_mut()) { - ta.pixelclock = 593_950_000; - tb.pixelclock = 593_960_000; - } - assert_eq!(watch.observe(a), SourceObservation::Baseline); - for _ in 0..4 { - assert_eq!(watch.observe(b), SourceObservation::Steady); - assert_eq!(watch.observe(a), SourceObservation::Steady); - } +fn a_relock_loop_triggers_a_replug_then_backs_off() { + let mut watch = RelockWatch::default(); + let t = Instant::now(); + let verdicts = storm(&mut watch, t, 120); + let replugs: Vec<_> = verdicts + .iter() + .enumerate() + .filter_map(|(s, v)| match v { + RelockVerdict::Replug { attempt, .. } => Some((s, *attempt)), + _ => None, + }) + .collect(); + // First replug once the window holds THRESHOLD events, the next ones + // 30 s and then 60 s later. + assert_eq!(replugs, vec![(5, 1), (35, 2), (95, 3)]); + assert!(matches!(verdicts[20], RelockVerdict::Storm { .. })); } #[test] -fn a_one_off_bogus_mode_is_not_a_change() { - let mut watch = SourceWatch::default(); - let normal = signal(NV12, 3840, 2160); - let mut bogus = normal; - if let Some(t) = bogus.timings.as_mut() { - t.height = 547; - } - watch.observe(normal); - assert_eq!(watch.observe(bogus), SourceObservation::Pending); - assert_eq!(watch.observe(normal), SourceObservation::Steady); +fn backoff_resets_after_the_input_settles() { + let mut watch = RelockWatch::default(); + let t = Instant::now(); + storm(&mut watch, t, 6); + let later = t + Duration::from_secs(6) + RelockWatch::SETTLE; + assert_eq!(watch.observe(later, 0), RelockVerdict::Quiet); + let verdicts = storm(&mut watch, later + Duration::from_secs(1), 6); + assert_eq!(verdicts[5], RelockVerdict::Replug { events: 6, attempt: 1 }); } From e7e1916d00dd2af8ce613e711f18a500c169ae7e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 15:25:41 +0000 Subject: [PATCH 4/5] fix(hid): coalesce pointer backlog and resync key state after a host stall Device logs from a host sleep/wake showed the HID writer timing out on the absolute mouse while the host stopped polling, then the host re-enumerating USB. Two things made input feel laggy and drifting around that: - While the host polls slowly, every queued pointer move was replayed one by one, so the cursor trailed seconds behind and then raced. The writer now takes everything queued at once and keeps only the last position of consecutive absolute moves (button changes and wheel steps keep their place, so presses and releases land where they happened), and sums consecutive relative moves. - A timeout on one device dropped every queued report, key-ups included, which can leave a key held on the host. Dropped reports now leave behind the latest keyboard and pointer state, which is resent (every 500 ms until the host reads again) unless newer input supersedes it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GRSn3XsjYLc494XwPez2mF --- app/src/hardware/usb/hid.rs | 182 ++++++++++++++++++++++++++++++++---- app/tests/hid_coalesce.rs | 65 +++++++++++++ 2 files changed, 229 insertions(+), 18 deletions(-) create mode 100644 app/tests/hid_coalesce.rs diff --git a/app/src/hardware/usb/hid.rs b/app/src/hardware/usb/hid.rs index 332bc37..5c9a35d 100644 --- a/app/src/hardware/usb/hid.rs +++ b/app/src/hardware/usb/hid.rs @@ -9,7 +9,7 @@ use std::time::{Duration, Instant}; use parking_lot::{Mutex, RwLock}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; -use tracing::{trace, warn}; +use tracing::{debug, info, trace, warn}; pub const KEYBOARD_LED_MASK_NUM_LOCK: u8 = 1 << 0; pub const KEYBOARD_LED_MASK_CAPS_LOCK: u8 = 1 << 1; @@ -110,14 +110,16 @@ const HID_DEV_KEYBOARD: usize = 0; const HID_DEV_ABS_MOUSE: usize = 1; const HID_DEV_REL_MOUSE: usize = 2; -enum HidWriteCmd { +/// One report bound for a gadget device. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HidReport { Keyboard([u8; 8]), AbsMouse([u8; 6]), AbsWheel([u8; 2]), RelMouse([u8; 4]), } -impl HidWriteCmd { +impl HidReport { fn device_and_payload(&self) -> (usize, &[u8]) { match self { Self::Keyboard(b) => (HID_DEV_KEYBOARD, b.as_slice()), @@ -128,6 +130,111 @@ impl HidWriteCmd { } } +/// Merges queued pointer reports that only matter as a whole: consecutive +/// absolute moves with the same buttons keep only the last position, and +/// consecutive relative moves with the same buttons are summed while they fit +/// a report. Keyboard reports, button changes and wheel steps stay as they +/// are and in order. After a host stall this turns a backlog of stale moves +/// into one jump to where the pointer is now, instead of a slow replay. +/// +/// `abs_buttons` carries the absolute pointer's buttons from one batch to the +/// next, so a batch's first report counts as a move when they did not change. +pub fn coalesce_reports(batch: &mut Vec, abs_buttons: &mut Option) { + let mut out: Vec = Vec::with_capacity(batch.len()); + // Index in `out` of the last absolute move that later moves may replace. + // A report that changes the buttons is never replaced: a press or release + // has to land where it happened. + let mut abs_tail: Option = None; + for report in batch.drain(..) { + match report { + HidReport::AbsMouse(new) => { + if let Some(i) = abs_tail + && let HidReport::AbsMouse(old) = &mut out[i] + && old[1] == new[1] + { + *old = new; + continue; + } + let is_move = *abs_buttons == Some(new[1]); + *abs_buttons = Some(new[1]); + out.push(report); + abs_tail = is_move.then_some(out.len() - 1); + continue; + } + HidReport::AbsWheel(_) => abs_tail = None, + HidReport::RelMouse(new) => { + if let Some(HidReport::RelMouse(old)) = out.last_mut() + && old[0] == new[0] + { + let dx = i16::from(old[1] as i8) + i16::from(new[1] as i8); + let dy = i16::from(old[2] as i8) + i16::from(new[2] as i8); + if let (Ok(dx), Ok(dy)) = (i8::try_from(dx), i8::try_from(dy)) { + old[1] = dx as u8; + old[2] = dy as u8; + continue; + } + } + } + HidReport::Keyboard(_) => {} + } + out.push(report); + } + *batch = out; +} + +/// Reports that restore the host's view of key and button state after +/// reports were dropped: the latest keyboard report (full state, so a lost +/// key-up cannot leave a key held), the latest absolute report, and the +/// latest relative buttons with no motion. +#[derive(Debug, Default)] +struct Resync { + keyboard: Option<[u8; 8]>, + abs: Option<[u8; 6]>, + rel_buttons: Option, +} + +impl Resync { + fn record(&mut self, report: &HidReport) { + match *report { + HidReport::Keyboard(r) => self.keyboard = Some(r), + HidReport::AbsMouse(r) => self.abs = Some(r), + HidReport::RelMouse(r) => self.rel_buttons = Some(r[0]), + HidReport::AbsWheel(_) => {} + } + } + + fn is_empty(&self) -> bool { + self.keyboard.is_none() && self.abs.is_none() && self.rel_buttons.is_none() + } + + /// Prepends the state reports a fresh batch does not already supersede. + fn apply(&mut self, batch: &mut Vec) -> usize { + let has = |f: fn(&HidReport) -> bool| batch.iter().any(f); + let mut lead = Vec::new(); + if let Some(r) = self.keyboard.take() + && !has(|r| matches!(r, HidReport::Keyboard(_))) + { + lead.push(HidReport::Keyboard(r)); + } + if let Some(r) = self.abs.take() + && !has(|r| matches!(r, HidReport::AbsMouse(_))) + { + lead.push(HidReport::AbsMouse(r)); + } + if let Some(b) = self.rel_buttons.take() + && !has(|r| matches!(r, HidReport::RelMouse(_))) + { + lead.push(HidReport::RelMouse([b, 0, 0, 0])); + } + let n = lead.len(); + batch.splice(0..0, lead); + n + } +} + +/// How often a pending resync is retried while no new input arrives. +const HID_RESYNC_RETRY: Duration = Duration::from_millis(500); + struct HidWriter { paths: [String; 3], files: [Option; 3], @@ -136,16 +243,55 @@ struct HidWriter { } impl HidWriter { - fn run(mut self, rx: mpsc::Receiver) { - while let Ok(cmd) = rx.recv() { - let (dev, payload) = cmd.device_and_payload(); - if let Err(e) = self.write_report(dev, payload) { - self.files[dev] = None; - let mut dropped = 0usize; - while rx.try_recv().is_ok() { - dropped += 1; + fn run(mut self, rx: mpsc::Receiver) { + let mut resync = Resync::default(); + let mut abs_buttons = None; + loop { + let first = if resync.is_empty() { + match rx.recv() { + Ok(r) => Some(r), + Err(_) => break, + } + } else { + match rx.recv_timeout(HID_RESYNC_RETRY) { + Ok(r) => Some(r), + Err(mpsc::RecvTimeoutError::Timeout) => None, + Err(mpsc::RecvTimeoutError::Disconnected) => break, + } + }; + let mut batch: Vec = first.into_iter().collect(); + while batch.len() < HID_WRITE_QUEUE_DEPTH { + match rx.try_recv() { + Ok(r) => batch.push(r), + Err(_) => break, } - self.warn_rate_limited(dev, &e, dropped); + } + let queued = batch.len(); + coalesce_reports(&mut batch, &mut abs_buttons); + if queued > batch.len() { + debug!(queued, sent = batch.len(), "coalesced HID pointer backlog"); + } + let resynced = resync.apply(&mut batch); + + let mut reports = batch.into_iter(); + while let Some(report) = reports.next() { + let (dev, payload) = report.device_and_payload(); + if let Err(e) = self.write_report(dev, payload) { + self.files[dev] = None; + // Keep only the latest state of what is dropped, and + // send it once the host reads reports again. + resync.record(&report); + let mut dropped = 0usize; + for r in reports.by_ref().chain(std::iter::from_fn(|| rx.try_recv().ok())) { + resync.record(&r); + dropped += 1; + } + self.warn_rate_limited(dev, &e, dropped); + break; + } + } + if resynced > 0 && resync.is_empty() { + info!(reports = resynced, "HID state resent after host stall"); } } trace!("hid writer thread exiting"); @@ -218,7 +364,7 @@ fn poll_writable(fd: std::os::fd::RawFd, timeout: Duration) -> std::io::Result, + write_tx: mpsc::SyncSender, last_full_warn: Mutex>, keyboard_reader_cancel: Mutex, @@ -265,7 +411,7 @@ impl Hid { } } - fn enqueue(&self, cmd: HidWriteCmd) -> anyhow::Result<()> { + fn enqueue(&self, cmd: HidReport) -> anyhow::Result<()> { self.reset_user_input_time(); match self.write_tx.try_send(cmd) { Ok(()) => Ok(()), @@ -386,7 +532,7 @@ impl Hid { for (i, k) in keys.iter().take(HID_KEY_BUFFER_SIZE).enumerate() { report[2 + i] = *k; } - self.enqueue(HidWriteCmd::Keyboard(report)) + self.enqueue(HidReport::Keyboard(report)) } pub fn abs_mouse_report(&self, x: i32, y: i32, buttons: u8) -> anyhow::Result<()> { @@ -397,18 +543,18 @@ impl Hid { data[3] = ((x as u16 >> 8) & 0x00FF) as u8; data[4] = (y as u16 & 0x00FF) as u8; data[5] = ((y as u16 >> 8) & 0x00FF) as u8; - self.enqueue(HidWriteCmd::AbsMouse(data)) + self.enqueue(HidReport::AbsMouse(data)) } pub fn abs_mouse_wheel_report(&self, wheel_y: i8) -> anyhow::Result<()> { if wheel_y == 0 { return Ok(()); } - self.enqueue(HidWriteCmd::AbsWheel([2u8, wheel_y as u8])) + self.enqueue(HidReport::AbsWheel([2u8, wheel_y as u8])) } pub fn rel_mouse_report(&self, dx: i8, dy: i8, buttons: u8) -> anyhow::Result<()> { - self.enqueue(HidWriteCmd::RelMouse([buttons, dx as u8, dy as u8, 0u8])) + self.enqueue(HidReport::RelMouse([buttons, dx as u8, dy as u8, 0u8])) } pub fn get_last_user_input_time(&self) -> Instant { diff --git a/app/tests/hid_coalesce.rs b/app/tests/hid_coalesce.rs new file mode 100644 index 0000000..d7eea61 --- /dev/null +++ b/app/tests/hid_coalesce.rs @@ -0,0 +1,65 @@ +use rustkvm::hardware::usb::hid::{HidReport, coalesce_reports}; + +fn abs(buttons: u8, x: u16, y: u16) -> HidReport { + let [x0, x1] = x.to_le_bytes(); + let [y0, y1] = y.to_le_bytes(); + HidReport::AbsMouse([1, buttons, x0, x1, y0, y1]) +} + +fn rel(buttons: u8, dx: i8, dy: i8) -> HidReport { + HidReport::RelMouse([buttons, dx as u8, dy as u8, 0]) +} + +fn key(code: u8) -> HidReport { + HidReport::Keyboard([0, 0, code, 0, 0, 0, 0, 0]) +} + +#[test] +fn a_backlog_of_moves_becomes_one_jump() { + let mut buttons = Some(0); + let mut batch: Vec<_> = (0..64).map(|i| abs(0, i * 10, i * 5)).collect(); + coalesce_reports(&mut batch, &mut buttons); + assert_eq!(batch, vec![abs(0, 630, 315)]); +} + +#[test] +fn button_changes_and_keys_keep_their_order() { + let mut buttons = Some(0); + let mut batch = vec![ + abs(0, 1, 1), + abs(0, 2, 2), + key(4), + abs(1, 2, 2), // press + abs(1, 3, 3), + abs(1, 4, 4), // drag + abs(0, 4, 4), // release + key(0), + ]; + coalesce_reports(&mut batch, &mut buttons); + // The press stays at (2, 2); the drag moves to (4, 4) before the release. + assert_eq!(batch, vec![abs(0, 2, 2), key(4), abs(1, 2, 2), abs(1, 4, 4), abs(0, 4, 4), key(0)]); + assert_eq!(buttons, Some(0)); +} + +#[test] +fn a_wheel_step_splits_moves() { + let mut buttons = Some(0); + let mut batch = vec![abs(0, 1, 1), HidReport::AbsWheel([2, 1]), abs(0, 2, 2), abs(0, 3, 3)]; + coalesce_reports(&mut batch, &mut buttons); + assert_eq!(batch, vec![abs(0, 1, 1), HidReport::AbsWheel([2, 1]), abs(0, 3, 3)]); +} + +#[test] +fn relative_moves_are_summed_within_range() { + let mut batch = vec![rel(0, 10, -5), rel(0, 20, -5), rel(0, 100, 0), rel(1, 1, 1)]; + coalesce_reports(&mut batch, &mut None); + assert_eq!(batch, vec![rel(0, 30, -10), rel(0, 100, 0), rel(1, 1, 1)]); +} + +#[test] +fn the_first_report_after_a_button_change_is_kept() { + let mut buttons = Some(1); + let mut batch = vec![abs(0, 5, 5), abs(0, 6, 6), abs(0, 7, 7)]; + coalesce_reports(&mut batch, &mut buttons); + assert_eq!(batch, vec![abs(0, 5, 5), abs(0, 7, 7)]); +} From 84710f1e391a4d74e6bf35ad089acc5a26f1fe0d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 15:55:15 +0000 Subject: [PATCH 5/5] Revert "fix(hid): coalesce pointer backlog and resync key state after a host stall" This reverts commit e7e1916. On hardware the mouse stuttered with it deployed and was smooth again after rolling the device back to f92cfc0, so the change goes until the cause is understood. The HDMI relock fix stays. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01GRSn3XsjYLc494XwPez2mF --- app/src/hardware/usb/hid.rs | 182 ++++-------------------------------- app/tests/hid_coalesce.rs | 65 ------------- 2 files changed, 18 insertions(+), 229 deletions(-) delete mode 100644 app/tests/hid_coalesce.rs diff --git a/app/src/hardware/usb/hid.rs b/app/src/hardware/usb/hid.rs index 5c9a35d..332bc37 100644 --- a/app/src/hardware/usb/hid.rs +++ b/app/src/hardware/usb/hid.rs @@ -9,7 +9,7 @@ use std::time::{Duration, Instant}; use parking_lot::{Mutex, RwLock}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, trace, warn}; +use tracing::{trace, warn}; pub const KEYBOARD_LED_MASK_NUM_LOCK: u8 = 1 << 0; pub const KEYBOARD_LED_MASK_CAPS_LOCK: u8 = 1 << 1; @@ -110,16 +110,14 @@ const HID_DEV_KEYBOARD: usize = 0; const HID_DEV_ABS_MOUSE: usize = 1; const HID_DEV_REL_MOUSE: usize = 2; -/// One report bound for a gadget device. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum HidReport { +enum HidWriteCmd { Keyboard([u8; 8]), AbsMouse([u8; 6]), AbsWheel([u8; 2]), RelMouse([u8; 4]), } -impl HidReport { +impl HidWriteCmd { fn device_and_payload(&self) -> (usize, &[u8]) { match self { Self::Keyboard(b) => (HID_DEV_KEYBOARD, b.as_slice()), @@ -130,111 +128,6 @@ impl HidReport { } } -/// Merges queued pointer reports that only matter as a whole: consecutive -/// absolute moves with the same buttons keep only the last position, and -/// consecutive relative moves with the same buttons are summed while they fit -/// a report. Keyboard reports, button changes and wheel steps stay as they -/// are and in order. After a host stall this turns a backlog of stale moves -/// into one jump to where the pointer is now, instead of a slow replay. -/// -/// `abs_buttons` carries the absolute pointer's buttons from one batch to the -/// next, so a batch's first report counts as a move when they did not change. -pub fn coalesce_reports(batch: &mut Vec, abs_buttons: &mut Option) { - let mut out: Vec = Vec::with_capacity(batch.len()); - // Index in `out` of the last absolute move that later moves may replace. - // A report that changes the buttons is never replaced: a press or release - // has to land where it happened. - let mut abs_tail: Option = None; - for report in batch.drain(..) { - match report { - HidReport::AbsMouse(new) => { - if let Some(i) = abs_tail - && let HidReport::AbsMouse(old) = &mut out[i] - && old[1] == new[1] - { - *old = new; - continue; - } - let is_move = *abs_buttons == Some(new[1]); - *abs_buttons = Some(new[1]); - out.push(report); - abs_tail = is_move.then_some(out.len() - 1); - continue; - } - HidReport::AbsWheel(_) => abs_tail = None, - HidReport::RelMouse(new) => { - if let Some(HidReport::RelMouse(old)) = out.last_mut() - && old[0] == new[0] - { - let dx = i16::from(old[1] as i8) + i16::from(new[1] as i8); - let dy = i16::from(old[2] as i8) + i16::from(new[2] as i8); - if let (Ok(dx), Ok(dy)) = (i8::try_from(dx), i8::try_from(dy)) { - old[1] = dx as u8; - old[2] = dy as u8; - continue; - } - } - } - HidReport::Keyboard(_) => {} - } - out.push(report); - } - *batch = out; -} - -/// Reports that restore the host's view of key and button state after -/// reports were dropped: the latest keyboard report (full state, so a lost -/// key-up cannot leave a key held), the latest absolute report, and the -/// latest relative buttons with no motion. -#[derive(Debug, Default)] -struct Resync { - keyboard: Option<[u8; 8]>, - abs: Option<[u8; 6]>, - rel_buttons: Option, -} - -impl Resync { - fn record(&mut self, report: &HidReport) { - match *report { - HidReport::Keyboard(r) => self.keyboard = Some(r), - HidReport::AbsMouse(r) => self.abs = Some(r), - HidReport::RelMouse(r) => self.rel_buttons = Some(r[0]), - HidReport::AbsWheel(_) => {} - } - } - - fn is_empty(&self) -> bool { - self.keyboard.is_none() && self.abs.is_none() && self.rel_buttons.is_none() - } - - /// Prepends the state reports a fresh batch does not already supersede. - fn apply(&mut self, batch: &mut Vec) -> usize { - let has = |f: fn(&HidReport) -> bool| batch.iter().any(f); - let mut lead = Vec::new(); - if let Some(r) = self.keyboard.take() - && !has(|r| matches!(r, HidReport::Keyboard(_))) - { - lead.push(HidReport::Keyboard(r)); - } - if let Some(r) = self.abs.take() - && !has(|r| matches!(r, HidReport::AbsMouse(_))) - { - lead.push(HidReport::AbsMouse(r)); - } - if let Some(b) = self.rel_buttons.take() - && !has(|r| matches!(r, HidReport::RelMouse(_))) - { - lead.push(HidReport::RelMouse([b, 0, 0, 0])); - } - let n = lead.len(); - batch.splice(0..0, lead); - n - } -} - -/// How often a pending resync is retried while no new input arrives. -const HID_RESYNC_RETRY: Duration = Duration::from_millis(500); - struct HidWriter { paths: [String; 3], files: [Option; 3], @@ -243,55 +136,16 @@ struct HidWriter { } impl HidWriter { - fn run(mut self, rx: mpsc::Receiver) { - let mut resync = Resync::default(); - let mut abs_buttons = None; - loop { - let first = if resync.is_empty() { - match rx.recv() { - Ok(r) => Some(r), - Err(_) => break, - } - } else { - match rx.recv_timeout(HID_RESYNC_RETRY) { - Ok(r) => Some(r), - Err(mpsc::RecvTimeoutError::Timeout) => None, - Err(mpsc::RecvTimeoutError::Disconnected) => break, - } - }; - let mut batch: Vec = first.into_iter().collect(); - while batch.len() < HID_WRITE_QUEUE_DEPTH { - match rx.try_recv() { - Ok(r) => batch.push(r), - Err(_) => break, + fn run(mut self, rx: mpsc::Receiver) { + while let Ok(cmd) = rx.recv() { + let (dev, payload) = cmd.device_and_payload(); + if let Err(e) = self.write_report(dev, payload) { + self.files[dev] = None; + let mut dropped = 0usize; + while rx.try_recv().is_ok() { + dropped += 1; } - } - let queued = batch.len(); - coalesce_reports(&mut batch, &mut abs_buttons); - if queued > batch.len() { - debug!(queued, sent = batch.len(), "coalesced HID pointer backlog"); - } - let resynced = resync.apply(&mut batch); - - let mut reports = batch.into_iter(); - while let Some(report) = reports.next() { - let (dev, payload) = report.device_and_payload(); - if let Err(e) = self.write_report(dev, payload) { - self.files[dev] = None; - // Keep only the latest state of what is dropped, and - // send it once the host reads reports again. - resync.record(&report); - let mut dropped = 0usize; - for r in reports.by_ref().chain(std::iter::from_fn(|| rx.try_recv().ok())) { - resync.record(&r); - dropped += 1; - } - self.warn_rate_limited(dev, &e, dropped); - break; - } - } - if resynced > 0 && resync.is_empty() { - info!(reports = resynced, "HID state resent after host stall"); + self.warn_rate_limited(dev, &e, dropped); } } trace!("hid writer thread exiting"); @@ -364,7 +218,7 @@ fn poll_writable(fd: std::os::fd::RawFd, timeout: Duration) -> std::io::Result, + write_tx: mpsc::SyncSender, last_full_warn: Mutex>, keyboard_reader_cancel: Mutex, @@ -411,7 +265,7 @@ impl Hid { } } - fn enqueue(&self, cmd: HidReport) -> anyhow::Result<()> { + fn enqueue(&self, cmd: HidWriteCmd) -> anyhow::Result<()> { self.reset_user_input_time(); match self.write_tx.try_send(cmd) { Ok(()) => Ok(()), @@ -532,7 +386,7 @@ impl Hid { for (i, k) in keys.iter().take(HID_KEY_BUFFER_SIZE).enumerate() { report[2 + i] = *k; } - self.enqueue(HidReport::Keyboard(report)) + self.enqueue(HidWriteCmd::Keyboard(report)) } pub fn abs_mouse_report(&self, x: i32, y: i32, buttons: u8) -> anyhow::Result<()> { @@ -543,18 +397,18 @@ impl Hid { data[3] = ((x as u16 >> 8) & 0x00FF) as u8; data[4] = (y as u16 & 0x00FF) as u8; data[5] = ((y as u16 >> 8) & 0x00FF) as u8; - self.enqueue(HidReport::AbsMouse(data)) + self.enqueue(HidWriteCmd::AbsMouse(data)) } pub fn abs_mouse_wheel_report(&self, wheel_y: i8) -> anyhow::Result<()> { if wheel_y == 0 { return Ok(()); } - self.enqueue(HidReport::AbsWheel([2u8, wheel_y as u8])) + self.enqueue(HidWriteCmd::AbsWheel([2u8, wheel_y as u8])) } pub fn rel_mouse_report(&self, dx: i8, dy: i8, buttons: u8) -> anyhow::Result<()> { - self.enqueue(HidReport::RelMouse([buttons, dx as u8, dy as u8, 0u8])) + self.enqueue(HidWriteCmd::RelMouse([buttons, dx as u8, dy as u8, 0u8])) } pub fn get_last_user_input_time(&self) -> Instant { diff --git a/app/tests/hid_coalesce.rs b/app/tests/hid_coalesce.rs deleted file mode 100644 index d7eea61..0000000 --- a/app/tests/hid_coalesce.rs +++ /dev/null @@ -1,65 +0,0 @@ -use rustkvm::hardware::usb::hid::{HidReport, coalesce_reports}; - -fn abs(buttons: u8, x: u16, y: u16) -> HidReport { - let [x0, x1] = x.to_le_bytes(); - let [y0, y1] = y.to_le_bytes(); - HidReport::AbsMouse([1, buttons, x0, x1, y0, y1]) -} - -fn rel(buttons: u8, dx: i8, dy: i8) -> HidReport { - HidReport::RelMouse([buttons, dx as u8, dy as u8, 0]) -} - -fn key(code: u8) -> HidReport { - HidReport::Keyboard([0, 0, code, 0, 0, 0, 0, 0]) -} - -#[test] -fn a_backlog_of_moves_becomes_one_jump() { - let mut buttons = Some(0); - let mut batch: Vec<_> = (0..64).map(|i| abs(0, i * 10, i * 5)).collect(); - coalesce_reports(&mut batch, &mut buttons); - assert_eq!(batch, vec![abs(0, 630, 315)]); -} - -#[test] -fn button_changes_and_keys_keep_their_order() { - let mut buttons = Some(0); - let mut batch = vec![ - abs(0, 1, 1), - abs(0, 2, 2), - key(4), - abs(1, 2, 2), // press - abs(1, 3, 3), - abs(1, 4, 4), // drag - abs(0, 4, 4), // release - key(0), - ]; - coalesce_reports(&mut batch, &mut buttons); - // The press stays at (2, 2); the drag moves to (4, 4) before the release. - assert_eq!(batch, vec![abs(0, 2, 2), key(4), abs(1, 2, 2), abs(1, 4, 4), abs(0, 4, 4), key(0)]); - assert_eq!(buttons, Some(0)); -} - -#[test] -fn a_wheel_step_splits_moves() { - let mut buttons = Some(0); - let mut batch = vec![abs(0, 1, 1), HidReport::AbsWheel([2, 1]), abs(0, 2, 2), abs(0, 3, 3)]; - coalesce_reports(&mut batch, &mut buttons); - assert_eq!(batch, vec![abs(0, 1, 1), HidReport::AbsWheel([2, 1]), abs(0, 3, 3)]); -} - -#[test] -fn relative_moves_are_summed_within_range() { - let mut batch = vec![rel(0, 10, -5), rel(0, 20, -5), rel(0, 100, 0), rel(1, 1, 1)]; - coalesce_reports(&mut batch, &mut None); - assert_eq!(batch, vec![rel(0, 30, -10), rel(0, 100, 0), rel(1, 1, 1)]); -} - -#[test] -fn the_first_report_after_a_button_change_is_kept() { - let mut buttons = Some(1); - let mut batch = vec![abs(0, 5, 5), abs(0, 6, 6), abs(0, 7, 7)]; - coalesce_reports(&mut batch, &mut buttons); - assert_eq!(batch, vec![abs(0, 5, 5), abs(0, 7, 7)]); -}