diff --git a/AGENTS.md b/AGENTS.md index 6c1a5d8..31e8a9a 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 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 @@ -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 # 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/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..b77ce3a --- /dev/null +++ b/app/src/hardware/v4l2.rs @@ -0,0 +1,354 @@ +//! 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, Setter, 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], +} + +/// `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); + +/// 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. 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}"))?; + 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 { + 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)) + } +} + +/// 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 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, + }, +} + +/// 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 RelockWatch { + recent: VecDeque<(Instant, u32)>, + last_replug: Option, + attempts: u32, +} + +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: Instant, events: u32) -> RelockVerdict { + if events > 0 { + self.recent.push_back((now, events)); + } + 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; + } + 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.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 96fe1ba..78a470c 100644 --- a/app/src/pipeline.rs +++ b/app/src/pipeline.rs @@ -127,6 +127,9 @@ 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, } impl VideoPipeline { @@ -145,6 +148,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 +297,17 @@ impl VideoPipeline { ]) .context("Failed to link video pipeline elements")?; + // Counts raw frames for the supervisor's stall check. + let raw_frames = Arc::new(AtomicU64::new(0)); + let probe_frames = raw_frames.clone(); + 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); + 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 +373,8 @@ impl VideoPipeline { raw_frame_waiters, raw_frame_requested, keyframe_requested_at, + capture_src: v4l2src, + raw_frames, }) } @@ -886,6 +903,16 @@ 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) + } + + /// 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..612d3c0 100644 --- a/app/src/video.rs +++ b/app/src/video.rs @@ -1,15 +1,16 @@ -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; 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, RelockVerdict, RelockWatch, pulse_hotplug}; 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,186 @@ 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. +/// 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, + relock_events: Option>, + relock: RelockWatch, + last_frames: u64, + last_progress: Instant, + stall_restarts: u32, + caps_logged: bool, +} + +impl LiveCheck { + fn new() -> Self { + Self { + probe: None, + probe_error_logged: false, + relock_events: None, + relock: RelockWatch::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.last_frames = 0; + self.last_progress = Instant::now(); + self.caps_logged = false; + } + + fn probe(&mut self, device: &str) -> Option> { + if self.probe.is_none() { + match CaptureProbe::open(device) { + 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 open capture node for monitoring"); + self.probe_error_logged = true; + } + } + } + } + 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, caps) = { + let guard = PIPELINE_MANAGER.lock().await; + let m = guard.as_ref()?; + (m.raw_frames(), m.capture_caps()) + }; + 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 probe = self.probe(device); + self.check_relock().await; + + if frames != self.last_frames { + self.last_frames = frames; + self.last_progress = Instant::now(); + 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(); + 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); + 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 + } +} + +/// 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); 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 capture stalled, rebuilding the pipeline"); + } 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..3e1389d --- /dev/null +++ b/app/tests/capture_source.rs @@ -0,0 +1,97 @@ +use std::time::{Duration, Instant}; + +use rustkvm::hardware::v4l2::{ + DvTimings, RelockVerdict, RelockWatch, parse_bt_timings, parse_pix_format, +}; + +const NV12: u32 = u32::from_le_bytes(*b"NV12"); + +fn put(buf: &mut [u8], offset: usize, value: u32) { + buf[offset..offset + 4].copy_from_slice(&value.to_ne_bytes()); +} + +#[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 } + ); +} + +/// 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 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 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 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 }); +}