From 6fdcf5c7c9ba941b514e72559b9f140887e755e6 Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Thu, 30 Jul 2026 22:43:59 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=97=A0=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7=E7=8A=B6=E6=80=81=E5=90=8C=E6=AD=A5=E4=B8=8E=E9=87=87?= =?UTF-8?q?=E9=9B=86=E7=AE=A1=E7=BA=BF=E9=98=BB=E5=A1=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用非阻塞 V4L2 句柄并类型化采集错误 - 完善视频管线生命周期，避免复用停止中的管线 - 区分音视频事件并支持 REST/WebSocket 状态快照 - 修复信号丢失后保留最后一帧及刷新后黑屏问题 - 优化无信号提示、状态展示优先级及相关界面细节 - 降低重复 OTG 错误日志级别 --- src/audio/recovery.rs | 5 +- src/events/mod.rs | 61 ++++++++++++++- src/events/types.rs | 15 ++-- src/hid/otg.rs | 4 +- src/video/capture/linux.rs | 100 +++++++++++++----------- src/video/capture/mod.rs | 41 ++++++++++ src/video/capture/windows.rs | 16 ++-- src/video/device/bridge.rs | 53 ++++++++++--- src/video/pipeline/mod.rs | 4 +- src/video/pipeline/shared.rs | 57 ++++++++++---- src/video/streamer.rs | 38 +++++---- src/video/types.rs | 4 +- src/web/handlers/stream.rs | 13 ++- src/web/ws.rs | 31 ++++++++ src/webrtc/webrtc_streamer.rs | 26 ++++-- web/src/api/index.ts | 15 +--- web/src/components/ComputerUseSheet.vue | 4 +- web/src/components/VideoInputFields.vue | 2 +- web/src/composables/useConsoleEvents.ts | 16 +--- web/src/i18n/en-US.ts | 7 +- web/src/i18n/zh-CN.ts | 7 +- web/src/types/websocket.ts | 35 ++++++++- web/src/views/ConsoleView.vue | 16 ++-- web/src/views/SettingsView.vue | 1 - 24 files changed, 406 insertions(+), 165 deletions(-) diff --git a/src/audio/recovery.rs b/src/audio/recovery.rs index 2ec3e5c2..59acc797 100644 --- a/src/audio/recovery.rs +++ b/src/audio/recovery.rs @@ -12,7 +12,7 @@ use super::device::{enumerate_audio_devices, AudioDeviceInfo}; use super::monitor::AudioHealthMonitor; use super::streamer::{AudioStreamState, AudioStreamer, AudioStreamerConfig}; use super::types::AudioControllerConfig; -use crate::events::{EventBus, StreamDeviceLostKind, SystemEvent}; +use crate::events::{EventBus, StreamKind, SystemEvent}; const RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); @@ -286,6 +286,7 @@ impl AudioRecovery { ) { if let Some(bus) = self.inner.event_bus.read().await.as_ref() { bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Audio, state: state.to_string(), device, reason: reason.map(str::to_string), @@ -298,7 +299,7 @@ impl AudioRecovery { async fn publish_device_lost(&self, device: &str, reason: &str) { if let Some(bus) = self.inner.event_bus.read().await.as_ref() { bus.publish(SystemEvent::StreamDeviceLost { - kind: StreamDeviceLostKind::Audio, + kind: StreamKind::Audio, device: device.to_string(), reason: reason.to_string(), }); diff --git a/src/events/mod.rs b/src/events/mod.rs index b657bdd2..ecd2c84f 100644 --- a/src/events/mod.rs +++ b/src/events/mod.rs @@ -6,9 +6,10 @@ use self::types::EXACT_EVENT_TOPICS; pub use types::{ AtxDeviceInfo, AudioDeviceInfo, ClientStats, HidDeviceInfo, LedState, MsdDeviceInfo, - MsdDeviceMediaInfo, StreamDeviceLostKind, SystemEvent, TtydDeviceInfo, VideoDeviceInfo, + MsdDeviceMediaInfo, StreamKind, SystemEvent, TtydDeviceInfo, VideoDeviceInfo, }; +use std::sync::RwLock; use tokio::sync::broadcast; const EVENT_CHANNEL_CAPACITY: usize = 256; @@ -40,6 +41,7 @@ pub struct EventBus { exact_topics: std::collections::HashMap<&'static str, broadcast::Sender>, prefix_topics: std::collections::HashMap>, device_info_dirty_tx: broadcast::Sender<()>, + latest_video_stream_state: RwLock>, } impl EventBus { @@ -60,12 +62,26 @@ impl EventBus { exact_topics, prefix_topics, device_info_dirty_tx, + latest_video_stream_state: RwLock::new(None), } } pub fn publish(&self, event: SystemEvent) { let event_name = event.event_name(); + if matches!( + event, + SystemEvent::StreamStateChanged { + kind: StreamKind::Video, + .. + } + ) { + *self + .latest_video_stream_state + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(event.clone()); + } + if let Some(tx) = self.exact_topics.get(event_name) { let _ = tx.send(event.clone()); } @@ -103,6 +119,15 @@ impl EventBus { self.device_info_dirty_tx.subscribe() } + /// Stateful video status topics replay this value to new WebSocket + /// subscribers so a page refresh cannot miss an earlier signal-loss edge. + pub fn latest_video_stream_state(&self) -> Option { + self.latest_video_stream_state + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } + pub fn subscriber_count(&self) -> usize { self.tx.receiver_count() } @@ -124,6 +149,7 @@ mod tests { let mut rx = bus.subscribe(); bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "streaming".to_string(), device: Some("/dev/video0".to_string()), reason: None, @@ -132,6 +158,10 @@ mod tests { let event = rx.recv().await.unwrap(); assert!(matches!(event, SystemEvent::StreamStateChanged { .. })); + assert!(matches!( + bus.latest_video_stream_state(), + Some(SystemEvent::StreamStateChanged { state, .. }) if state == "streaming" + )); } #[tokio::test] @@ -143,6 +173,7 @@ mod tests { assert_eq!(bus.subscriber_count(), 2); bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "ready".to_string(), device: Some("/dev/video0".to_string()), reason: None, @@ -162,6 +193,7 @@ mod tests { let mut rx = bus.subscribe_topic("stream.state_changed").unwrap(); bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "ready".to_string(), device: None, reason: None, @@ -178,6 +210,7 @@ mod tests { let mut rx = bus.subscribe_topic("stream.*").unwrap(); bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "ready".to_string(), device: None, reason: None, @@ -200,10 +233,36 @@ mod tests { assert_eq!(bus.subscriber_count(), 0); bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "ready".to_string(), device: None, reason: None, next_retry_ms: None, }); } + + #[test] + fn audio_state_does_not_replace_latest_video_state() { + let bus = EventBus::new(); + bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, + state: "no_signal".to_string(), + device: Some("/dev/video0".to_string()), + reason: Some("no_sync".to_string()), + next_retry_ms: Some(500), + }); + bus.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Audio, + state: "streaming".to_string(), + device: Some("hw:0,0".to_string()), + reason: None, + next_retry_ms: None, + }); + + assert!(matches!( + bus.latest_video_stream_state(), + Some(SystemEvent::StreamStateChanged { state, reason, .. }) + if state == "no_signal" && reason.as_deref() == Some("no_sync") + )); + } } diff --git a/src/events/types.rs b/src/events/types.rs index 0edcd47e..7a438354 100644 --- a/src/events/types.rs +++ b/src/events/types.rs @@ -92,10 +92,10 @@ pub struct ClientStats { pub connected_secs: u64, } -/// Video vs audio source for [`SystemEvent::StreamDeviceLost`] (WebSocket `stream.device_lost`). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +/// Media subsystem that owns a stream state or device event. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] -pub enum StreamDeviceLostKind { +pub enum StreamKind { Video, Audio, } @@ -114,6 +114,7 @@ pub enum SystemEvent { #[serde(rename = "stream.state_changed")] StreamStateChanged { + kind: StreamKind, state: String, device: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -141,7 +142,7 @@ pub enum SystemEvent { #[serde(rename = "stream.device_lost")] StreamDeviceLost { - kind: StreamDeviceLostKind, + kind: StreamKind, device: String, reason: String, }, @@ -274,6 +275,7 @@ mod tests { #[test] fn test_event_name() { let event = SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "streaming".to_string(), device: Some("/dev/video0".to_string()), reason: None, @@ -285,7 +287,7 @@ mod tests { #[test] fn stream_device_lost_json_snake_case_kind() { let event = SystemEvent::StreamDeviceLost { - kind: StreamDeviceLostKind::Audio, + kind: StreamKind::Audio, device: "hw:0,0".to_string(), reason: "test".to_string(), }; @@ -306,6 +308,7 @@ mod tests { from_mode: String::new(), }, SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: String::new(), device: None, reason: None, @@ -323,7 +326,7 @@ mod tests { fps: 0, }, SystemEvent::StreamDeviceLost { - kind: StreamDeviceLostKind::Video, + kind: StreamKind::Video, device: String::new(), reason: String::new(), }, diff --git a/src/hid/otg.rs b/src/hid/otg.rs index fa4fba77..23443ea1 100644 --- a/src/hid/otg.rs +++ b/src/hid/otg.rs @@ -167,9 +167,9 @@ impl OtgBackend { if now.duration_since(*last_log).as_secs() >= 1 { let count = self.error_count.swap(0, Ordering::Relaxed); if count > 1 { - warn!("{} (repeated {} times)", msg, count); + debug!("{} (repeated {} times)", msg, count); } else { - warn!("{}", msg); + debug!("{}", msg); } *last_log = now; } else { diff --git a/src/video/capture/linux.rs b/src/video/capture/linux.rs index 3c1a5161..6ae0cc00 100644 --- a/src/video/capture/linux.rs +++ b/src/video/capture/linux.rs @@ -3,6 +3,7 @@ use std::fs::File; use std::io; use std::os::fd::AsFd; +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; use std::time::{Duration, Instant}; @@ -13,29 +14,21 @@ use v4l2r::bindings::{ V4L2_DV_BT_656_1120, }; use v4l2r::ioctl::{ - self, Capabilities, Capability as V4l2rCapability, Event as V4l2Event, EventType, - MemoryConsistency, PlaneMapping, QBufPlane, QBuffer, QueryBuffer, QueryDvTimingsError, - SubscribeEventFlags, V4l2Buffer, + self, Capabilities, Capability as V4l2rCapability, EventType, IntoErrno, MemoryConsistency, + PlaneMapping, QBufPlane, QBuffer, QueryBuffer, QueryDvTimingsError, SubscribeEventFlags, + V4l2Buffer, }; use v4l2r::memory::{MemoryType, MmapHandle}; use v4l2r::nix::errno::Errno; use v4l2r::{Format as V4l2rFormat, PixelFormat as V4l2rPixelFormat, QueueType}; +use super::CaptureReadError; use crate::error::{AppError, Result}; use crate::video::device::bridge::{self as csi_bridge, CsiBridgeKind, ProbeResult}; use crate::video::device::VideoControlMode; use crate::video::format::{PixelFormat, Resolution}; use crate::video::signal::SignalStatus; -/// `io::Error` payload when the driver posts `V4L2_EVENT_SOURCE_CHANGE`. -pub const SOURCE_CHANGED_MARKER: &str = "v4l2_source_changed"; - -pub fn is_source_changed_error(err: &io::Error) -> bool { - err.get_ref() - .map(|inner| inner.to_string() == SOURCE_CHANGED_MARKER) - .unwrap_or(false) -} - /// Metadata for a captured frame. #[derive(Debug, Clone, Copy)] pub struct CaptureMeta { @@ -76,6 +69,14 @@ pub struct CaptureStream { native_hdmirx_next_state_check: Option, } +fn open_capture_device(path: &Path) -> io::Result { + File::options() + .read(true) + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(path) +} + impl CaptureStream { /// UVC: uses `resolution`. CSI bridges: DV-probe first; may return `CaptureNoSignal`. pub fn open( @@ -149,10 +150,7 @@ impl CaptureStream { } // ── Phase 1: open the capture (video) node ───────────────────── - let mut fd = File::options() - .read(true) - .write(true) - .open(device_path.as_ref()) + let mut fd = open_capture_device(device_path.as_ref()) .map_err(|e| AppError::VideoError(format!("Failed to open device: {}", e)))?; let caps: V4l2rCapability = ioctl::querycap(&fd) @@ -423,7 +421,10 @@ impl CaptureStream { } } - pub fn next_into(&mut self, dst: &mut Vec) -> io::Result { + pub fn next_into( + &mut self, + dst: &mut Vec, + ) -> std::result::Result { self.wait_ready()?; // Several vendor BSPs update G_FMT/DV timings without making the @@ -440,12 +441,20 @@ impl CaptureStream { info!( "Native HDMI RX active format/timings changed without a usable event; requesting stream re-open" ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } } - let dqbuf: V4l2Buffer = ioctl::dqbuf(&self.fd, self.queue, MemoryType::Mmap) - .map_err(|e| io::Error::other(format!("dqbuf failed: {}", e)))?; + let dqbuf: V4l2Buffer = + ioctl::dqbuf(&self.fd, self.queue, MemoryType::Mmap).map_err(|error| { + let message = error.to_string(); + let error = if error.into_errno() == Errno::EAGAIN as i32 { + io::Error::from(io::ErrorKind::WouldBlock) + } else { + io::Error::other(format!("dqbuf failed: {}", message)) + }; + CaptureReadError::Io(error) + })?; let index = dqbuf.as_v4l2_buffer().index as usize; let sequence = dqbuf.as_v4l2_buffer().sequence as u64; @@ -495,7 +504,7 @@ impl CaptureStream { self.resolution.height, self.stride ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } } @@ -505,7 +514,7 @@ impl CaptureStream { }) } - fn wait_ready(&self) -> io::Result<()> { + fn wait_ready(&self) -> std::result::Result<(), CaptureReadError> { if self.timeout.is_zero() { return Ok(()); } @@ -524,16 +533,17 @@ impl CaptureStream { } let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { - return Err(io::Error::new(io::ErrorKind::TimedOut, "capture timeout")); + return Err(io::Error::new(io::ErrorKind::TimedOut, "capture timeout").into()); } // `nix::poll` accepts a u16 millisecond timeout. Round sub-ms // durations up, and preserve the original deadline if a very long // timeout needs more than one poll call. let timeout_ms = remaining.as_millis().clamp(1, u16::MAX as u128) as u16; - let ready = poll(&mut poll_fds, PollTimeout::from(timeout_ms))?; + let ready = poll(&mut poll_fds, PollTimeout::from(timeout_ms)) + .map_err(|error| CaptureReadError::Io(error.into()))?; if ready == 0 { if Instant::now() >= deadline { - return Err(io::Error::new(io::ErrorKind::TimedOut, "capture timeout")); + return Err(io::Error::new(io::ErrorKind::TimedOut, "capture timeout").into()); } continue; } @@ -544,13 +554,13 @@ impl CaptureStream { if let Some(subdev_fd) = self.subdev_fd.as_ref() { if let Some(revents) = poll_fds.get(1).and_then(|f| f.revents()) { if revents.contains(PollFlags::POLLPRI) { - let drained = drain_events(subdev_fd); + let drained = csi_bridge::drain_v4l2_events(subdev_fd); info!( "Subdev SOURCE_CHANGE detected (drained {} event(s)), \ requesting stream re-open", drained ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } } } @@ -561,10 +571,10 @@ impl CaptureStream { "capture poll: video revents={:?} (ERR/HUP) — requesting stream re-open", revents ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } if revents.contains(PollFlags::POLLPRI) { - let drained = drain_events(&self.fd); + let drained = csi_bridge::drain_v4l2_events(&self.fd); if self.native_hdmirx_state_unchanged() { debug!( "Ignoring {} spurious native HDMI RX SOURCE_CHANGE event(s): active format/timings are unchanged", @@ -580,7 +590,7 @@ impl CaptureStream { requesting stream re-open", drained ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } if !revents.contains(PollFlags::POLLIN) { // rkcif + RK628: the driver may wake `poll` after internally @@ -590,7 +600,7 @@ impl CaptureStream { "capture poll: ready={} video revents={:?} (no POLLIN) — requesting stream re-open", ready, revents ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } return Ok(()); } @@ -599,7 +609,7 @@ impl CaptureStream { "capture poll: ready={} but video revents unavailable — requesting stream re-open", ready ); - return Err(io::Error::other(SOURCE_CHANGED_MARKER)); + return Err(CaptureReadError::SourceChanged); } } @@ -704,20 +714,6 @@ impl Drop for CaptureStream { } } -/// Drain any pending `V4L2_EVENT_*` events on `fd`. Used after POLLPRI to -/// clear the queue so the next poll doesn't immediately wake up on stale -/// state. Capped at 16 events per call. -fn drain_events(fd: &File) -> u32 { - let mut drained = 0u32; - while let Ok(_ev) = ioctl::dqevent::(fd) { - drained = drained.saturating_add(1); - if drained >= 16 { - break; - } - } - drained -} - /// Result of a successful `VIDIOC_QUERY_DV_TIMINGS` + `VIDIOC_S_DV_TIMINGS` /// probe. Used by the CSI bridge path to override the requested resolution /// with the source-reported geometry before `S_FMT`. @@ -937,7 +933,7 @@ fn set_fps(fd: &File, queue: QueueType, fps: u32) -> std::result::Result<(), ioc #[cfg(test)] mod tests { - use super::{DvTimingsSignature, NativeHdmirxState}; + use super::{open_capture_device, DvTimingsSignature, NativeHdmirxState}; use crate::video::format::PixelFormat; fn timing() -> DvTimingsSignature { @@ -986,4 +982,14 @@ mod tests { }; assert_eq!(no_timing_state.timings_match(None), Some(true)); } + + #[test] + fn capture_device_handles_are_non_blocking() { + let temp = tempfile::NamedTempFile::new().expect("create temporary device file"); + let opened = open_capture_device(temp.path()).expect("open capture device"); + let flags = + nix::fcntl::fcntl(&opened, nix::fcntl::FcntlArg::F_GETFL).expect("read file flags"); + + assert_ne!(flags & libc::O_NONBLOCK, 0); + } } diff --git a/src/video/capture/mod.rs b/src/video/capture/mod.rs index 82e0a7dc..106d8a1a 100644 --- a/src/video/capture/mod.rs +++ b/src/video/capture/mod.rs @@ -1,10 +1,51 @@ //! Video capture implementations and capture-state helpers. +use std::fmt; +use std::io; + pub(crate) mod runtime; pub(crate) mod status; pub const DEFAULT_CAPTURE_BUFFER_COUNT: u32 = 4; +/// Expected source changes are control flow, not stringly typed I/O errors. +#[derive(Debug)] +pub enum CaptureReadError { + SourceChanged, + Io(io::Error), +} + +impl CaptureReadError { + pub fn as_io_error(&self) -> Option<&io::Error> { + match self { + Self::SourceChanged => None, + Self::Io(error) => Some(error), + } + } +} + +impl From for CaptureReadError { + fn from(error: io::Error) -> Self { + Self::Io(error) + } +} + +impl fmt::Display for CaptureReadError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::SourceChanged => formatter.write_str("capture source changed"), + Self::Io(error) => error.fmt(formatter), + } + } +} + +impl std::error::Error for CaptureReadError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.as_io_error() + .map(|error| error as &(dyn std::error::Error + 'static)) + } +} + #[cfg(unix)] mod linux; #[cfg(windows)] diff --git a/src/video/capture/windows.rs b/src/video/capture/windows.rs index 1e8df15e..b7c4b11b 100644 --- a/src/video/capture/windows.rs +++ b/src/video/capture/windows.rs @@ -2,6 +2,7 @@ use std::io; use std::path::{Path, PathBuf}; use std::time::Duration; +use super::CaptureReadError; use crate::error::{AppError, Result}; use crate::video::device::bridge::{CsiBridgeKind, ProbeResult}; use crate::video::device::{ @@ -9,14 +10,6 @@ use crate::video::device::{ }; use crate::video::format::{PixelFormat, Resolution}; -pub const SOURCE_CHANGED_MARKER: &str = "dshow_source_changed"; - -pub fn is_source_changed_error(err: &io::Error) -> bool { - err.get_ref() - .map(|inner| inner.to_string() == SOURCE_CHANGED_MARKER) - .unwrap_or(false) -} - #[derive(Debug, Clone, Copy)] pub struct CaptureMeta { pub bytes_used: usize, @@ -119,7 +112,10 @@ impl CaptureStream { self.stride } - pub fn next_into(&mut self, dst: &mut Vec) -> io::Result { + pub fn next_into( + &mut self, + dst: &mut Vec, + ) -> std::result::Result { match self.capture.read_packet() { Ok((packet, sequence)) => { dst.clear(); @@ -135,7 +131,7 @@ impl CaptureStream { } else { io::ErrorKind::Other }; - Err(io::Error::new(kind, err.message)) + Err(CaptureReadError::Io(io::Error::new(kind, err.message))) } } } diff --git a/src/video/device/bridge.rs b/src/video/device/bridge.rs index 96665fcb..c855b245 100644 --- a/src/video/device/bridge.rs +++ b/src/video/device/bridge.rs @@ -3,6 +3,7 @@ use std::fs::File; use std::io; use std::os::fd::{AsFd, AsRawFd, FromRawFd}; +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; use std::sync::mpsc; use std::thread; @@ -14,7 +15,9 @@ use tracing::{debug, info, warn}; use v4l2r::bindings::{ v4l2_bt_timings, v4l2_dv_timings, V4L2_DV_BT_656_1120, V4L2_DV_FL_HAS_CEA861_VIC, }; -use v4l2r::ioctl::{self, Event as V4l2Event, EventType, QueryDvTimingsError, SubscribeEventFlags}; +use v4l2r::ioctl::{ + self, Event as V4l2Event, EventType, IntoErrno, QueryDvTimingsError, SubscribeEventFlags, +}; use v4l2r::nix::errno::Errno; use crate::video::signal::SignalStatus; @@ -134,7 +137,36 @@ fn read_sysfs_name(subdev_sysfs: &Path) -> Option { } pub fn open_subdev(path: &Path) -> io::Result { - File::options().read(true).write(true).open(path) + File::options() + .read(true) + .write(true) + .custom_flags(libc::O_NONBLOCK) + .open(path) +} + +/// Drain a non-blocking V4L2 event queue until the driver reports that it is +/// empty. Both video nodes and subdevices use the same event ioctl contract. +pub fn drain_v4l2_events(fd: &File) -> u32 { + let mut drained = 0u32; + loop { + match ioctl::dqevent::(fd) { + Ok(_event) => { + drained = drained.saturating_add(1); + if drained >= 16 { + break; + } + } + Err(error) => { + let message = error.to_string(); + let errno = error.into_errno(); + if errno != Errno::EAGAIN as i32 && errno != Errno::ENOENT as i32 { + debug!("Failed to drain V4L2 event queue: {}", message); + } + break; + } + } + } + drained } pub fn probe_signal(subdev_fd: &impl AsRawFd, kind: CsiBridgeKind) -> ProbeResult { @@ -298,13 +330,7 @@ pub fn wait_source_change(subdev_fd: &File, timeout: Duration) -> io::Result(subdev_fd) { - drained = drained.saturating_add(1); - if drained >= 16 { - break; - } - } + let drained = drain_v4l2_events(subdev_fd); debug!("subdev source_change drained {} event(s)", drained); Ok(true) } @@ -313,6 +339,15 @@ pub fn wait_source_change(subdev_fd: &File, timeout: Duration) -> io::Result= 0); + assert_ne!(flags & libc::O_NONBLOCK, 0); + } + #[test] fn rk628_fingerprint_matches_vga() { let mut bt: v4l2_bt_timings = unsafe { std::mem::zeroed() }; diff --git a/src/video/pipeline/mod.rs b/src/video/pipeline/mod.rs index 2349b672..e12e6e51 100644 --- a/src/video/pipeline/mod.rs +++ b/src/video/pipeline/mod.rs @@ -4,6 +4,6 @@ mod encoder_state; mod shared; pub use shared::{ - EncodedVideoFrame, PipelineAppliedConfig, PipelineStateNotification, SharedVideoPipeline, - SharedVideoPipelineConfig, SharedVideoPipelineStats, + EncodedVideoFrame, PipelineAppliedConfig, PipelineLifecycle, PipelineStateNotification, + SharedVideoPipeline, SharedVideoPipelineConfig, SharedVideoPipelineStats, }; diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 26c9bd1d..1a625162 100644 --- a/src/video/pipeline/shared.rs +++ b/src/video/pipeline/shared.rs @@ -47,7 +47,7 @@ use crate::video::capture::status::{ capture_error_log_key, classify_capture_io_error, is_device_lost_message, signal_status_from_capture_kind, CaptureIoErrorKind, }; -use crate::video::capture::{is_source_changed_error, BridgeContext, CaptureStream}; +use crate::video::capture::{BridgeContext, CaptureReadError, CaptureStream}; use crate::video::codec::h264_bitstream; use crate::video::codec::registry::{EncoderBackend, VideoEncoderType}; use crate::video::device::parse_bridge_kind; @@ -98,6 +98,13 @@ pub struct PipelineAppliedConfig { pub fps: u32, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PipelineLifecycle { + Running, + Stopping, + Stopped, +} + impl PipelineStateNotification { fn streaming(resolution: Resolution, format: PixelFormat, fps: u32) -> Self { Self { @@ -453,6 +460,19 @@ impl SharedVideoPipeline { *self.running_rx.borrow() } + /// Lifecycle state derived from the stop-request flag and the capture + /// thread's completion signal. A stopping pipeline must never receive a + /// new subscriber or be replaced before it releases the V4L2 device. + pub fn lifecycle(&self) -> PipelineLifecycle { + if self.running_flag.load(Ordering::Acquire) { + PipelineLifecycle::Running + } else if *self.running_rx.borrow() { + PipelineLifecycle::Stopping + } else { + PipelineLifecycle::Stopped + } + } + /// Subscribe to running state changes /// /// Returns a watch receiver that can be used to detect when the pipeline stops. @@ -909,7 +929,7 @@ impl SharedVideoPipeline { consecutive_timeouts = 0; meta } - Err(e) => { + Err(CaptureReadError::SourceChanged) => { // V4L2 driver reported V4L2_EVENT_SOURCE_CHANGE. // The current capture is effectively invalidated: // drop the stream so the next iteration re-opens @@ -917,21 +937,22 @@ impl SharedVideoPipeline { // path for source-side resolution switches on // RK628 / rkcif; the retry policy is only a fallback // when a driver does not provide usable events. - if is_source_changed_error(&e) { - info!( - "Capture reported SOURCE_CHANGE — \ - dropping stream for immediate re-open" - ); - if recovery_policy.control_mode() - == VideoControlMode::SourceFollowing - { - pipeline.notify_state(PipelineStateNotification::no_signal( - SignalStatus::NoSignal, - Some(recovery_policy.retry_delay(1).as_millis() as u64), - )); - } - consecutive_timeouts = 0; - stream = None; + info!( + "Capture reported SOURCE_CHANGE — \ + dropping stream for immediate re-open" + ); + if recovery_policy.control_mode() == VideoControlMode::SourceFollowing { + pipeline.notify_state(PipelineStateNotification::no_signal( + SignalStatus::NoSignal, + Some(recovery_policy.retry_delay(1).as_millis() as u64), + )); + } + consecutive_timeouts = 0; + stream = None; + continue; + } + Err(CaptureReadError::Io(e)) => { + if e.kind() == std::io::ErrorKind::WouldBlock { continue; } if e.kind() == std::io::ErrorKind::TimedOut { @@ -1607,10 +1628,12 @@ mod tests { assert!(!pipeline.running_flag.load(Ordering::Acquire)); assert!(pipeline.is_running()); + assert_eq!(pipeline.lifecycle(), PipelineLifecycle::Stopping); // Simulate the capture thread's common cleanup tail. let _ = pipeline.running.send(false); assert!(!pipeline.is_running()); + assert_eq!(pipeline.lifecycle(), PipelineLifecycle::Stopped); } #[tokio::test] diff --git a/src/video/streamer.rs b/src/video/streamer.rs index b7d352bc..14e144d4 100644 --- a/src/video/streamer.rs +++ b/src/video/streamer.rs @@ -19,7 +19,7 @@ use super::device::{ use super::format::{PixelFormat, Resolution}; use super::frame::{FrameBuffer, FrameBufferPool, VideoFrame}; use crate::error::{AppError, Result}; -use crate::events::{EventBus, StreamDeviceLostKind, SystemEvent}; +use crate::events::{EventBus, StreamKind, SystemEvent}; use crate::stream::MjpegStreamHandler; use crate::utils::LogThrottler; use crate::video::capture::runtime::open_capture_stream; @@ -28,7 +28,7 @@ use crate::video::capture::status::{ CaptureIoErrorKind, }; use crate::video::capture::{ - is_source_changed_error, BridgeContext, CaptureStream, DEFAULT_CAPTURE_BUFFER_COUNT, + BridgeContext, CaptureReadError, CaptureStream, DEFAULT_CAPTURE_BUFFER_COUNT, }; use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy}; @@ -252,6 +252,7 @@ impl Streamer { let next = self.next_retry_ms.load(Ordering::Relaxed); SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: external.to_string(), device, reason: reason.map(|s| s.to_string()), @@ -1067,16 +1068,19 @@ impl Streamer { let mut owned = buffer_pool.take(MIN_CAPTURE_FRAME_SIZE); let meta = match stream.next_into(&mut owned) { Ok(meta) => meta, - Err(e) => { - if is_source_changed_error(&e) { - info!("Capture SOURCE_CHANGE — soft-restart for DV re-probe"); - let delay = recovery_policy - .retry_delay(no_signal_restart_count.saturating_add(1)); - set_retry(delay.as_millis() as u64); - go_offline(); - set_state(StreamerState::NoSignal); - need_soft_restart = true; - break 'capture; + Err(CaptureReadError::SourceChanged) => { + info!("Capture SOURCE_CHANGE — soft-restart for DV re-probe"); + let delay = + recovery_policy.retry_delay(no_signal_restart_count.saturating_add(1)); + set_retry(delay.as_millis() as u64); + go_offline(); + set_state(StreamerState::NoSignal); + need_soft_restart = true; + break 'capture; + } + Err(CaptureReadError::Io(e)) => { + if e.kind() == std::io::ErrorKind::WouldBlock { + continue 'capture; } if e.kind() == std::io::ErrorKind::TimedOut { if signal_present { @@ -1308,9 +1312,11 @@ impl Streamer { pub async fn stats(&self) -> StreamerStats { let config = self.config.read().await; let fps = self.current_fps.load(Ordering::Relaxed) as f32 / 100.0; + let (state, reason) = self.state().await.external_state(); StreamerStats { - state: self.state().await, + state: state.to_string(), + reason: reason.map(str::to_string), device: self.current_device().await.map(|d| d.name), format: Some(config.format.to_string()), resolution: Some((config.resolution.width, config.resolution.height)), @@ -1402,7 +1408,7 @@ impl Streamer { // Publish device lost event self.publish_event(SystemEvent::StreamDeviceLost { - kind: StreamDeviceLostKind::Video, + kind: StreamKind::Video, device: device.clone(), reason: reason.clone(), }) @@ -1551,7 +1557,9 @@ impl Default for Streamer { /// Streamer statistics #[derive(Debug, Clone, serde::Serialize)] pub struct StreamerStats { - pub state: StreamerState, + pub state: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option, pub device: Option, pub format: Option, pub resolution: Option<(u32, u32)>, diff --git a/src/video/types.rs b/src/video/types.rs index 02845759..31ba49f8 100644 --- a/src/video/types.rs +++ b/src/video/types.rs @@ -17,6 +17,6 @@ pub use super::codec::registry::{EncoderBackend, VideoEncoderType}; // From video::pipeline pub use super::pipeline::{ - EncodedVideoFrame, PipelineAppliedConfig, PipelineStateNotification, SharedVideoPipeline, - SharedVideoPipelineConfig, SharedVideoPipelineStats, + EncodedVideoFrame, PipelineAppliedConfig, PipelineLifecycle, PipelineStateNotification, + SharedVideoPipeline, SharedVideoPipelineConfig, SharedVideoPipelineStats, }; diff --git a/src/web/handlers/stream.rs b/src/web/handlers/stream.rs index 44b85a08..6f1743de 100644 --- a/src/web/handlers/stream.rs +++ b/src/web/handlers/stream.rs @@ -1,5 +1,6 @@ use super::*; +use crate::events::SystemEvent; use crate::video::streamer::StreamerStats; use axum::{ body::Body, @@ -16,7 +17,17 @@ fn stream_mode_label(mode: StreamMode, codec: crate::video::codec::VideoCodecTyp /// Get stream state pub async fn stream_state(State(state): State>) -> Json { - Json(state.stream_manager.stats().await) + let mut stats = state.stream_manager.stats().await; + if let Some(SystemEvent::StreamStateChanged { + state: event_state, + reason, + .. + }) = state.events.latest_video_stream_state() + { + stats.state = event_state; + stats.reason = reason; + } + Json(stats) } /// Start streaming diff --git a/src/web/ws.rs b/src/web/ws.rs index 3dd8e4b4..cc51d520 100644 --- a/src/web/ws.rs +++ b/src/web/ws.rs @@ -49,9 +49,14 @@ fn is_device_info_topic(topic: &str) -> bool { matches!(topic, "*" | "system.*" | "system.device_info") } +fn is_stream_state_topic(topic: &str) -> bool { + matches!(topic, "*" | "stream.*" | "stream.state_changed") +} + fn rebuild_event_tasks( state: &Arc, topics: &[String], + replay_stream_state: bool, event_tx: &mpsc::UnboundedSender, event_tasks: &mut Vec>, ) { @@ -61,7 +66,17 @@ fn rebuild_event_tasks( let topics = normalize_topics(topics); let mut device_info_task_added = false; + let mut stream_state_snapshot_added = false; for topic in topics { + if replay_stream_state && is_stream_state_topic(&topic) && !stream_state_snapshot_added { + if let Some(snapshot) = state.events.latest_video_stream_state() { + if event_tx.send(BusMessage::Event(snapshot)).is_err() { + return; + } + } + stream_state_snapshot_added = true; + } + if is_device_info_topic(&topic) && !device_info_task_added { let state = state.clone(); let mut rx = state.subscribe_device_info(); @@ -157,12 +172,19 @@ async fn handle_socket(socket: WebSocket, state: Arc) { msg = receiver.next() => { match msg { Some(Ok(Message::Text(text))) => { + let had_stream_state = normalize_topics(&subscribed_topics) + .iter() + .any(|topic| is_stream_state_topic(topic)); if let Err(e) = handle_client_message(&text, &mut subscribed_topics).await { warn!("Failed to handle client message: {}", e); } else { + let has_stream_state = normalize_topics(&subscribed_topics) + .iter() + .any(|topic| is_stream_state_topic(topic)); rebuild_event_tasks( &state, &subscribed_topics, + !had_stream_state && has_stream_state, &event_tx, &mut event_tasks, ); @@ -308,4 +330,13 @@ mod tests { assert!(is_device_info_topic("*")); assert!(!is_device_info_topic("stream.*")); } + + #[test] + fn test_is_stream_state_topic_matches_stateful_subscriptions() { + assert!(is_stream_state_topic("*")); + assert!(is_stream_state_topic("stream.*")); + assert!(is_stream_state_topic("stream.state_changed")); + assert!(!is_stream_state_topic("stream.stats_update")); + assert!(!is_stream_state_topic("system.device_info")); + } } diff --git a/src/webrtc/webrtc_streamer.rs b/src/webrtc/webrtc_streamer.rs index 3a8588db..852c403d 100644 --- a/src/webrtc/webrtc_streamer.rs +++ b/src/webrtc/webrtc_streamer.rs @@ -10,7 +10,7 @@ use tracing::{debug, info, trace, warn}; use crate::audio::{AudioController, OpusFrame}; use crate::error::{AppError, Result}; -use crate::events::{EventBus, StreamDeviceLostKind, SystemEvent}; +use crate::events::{EventBus, StreamKind, SystemEvent}; use crate::hid::HidController; use crate::video::capture::DEFAULT_CAPTURE_BUFFER_COUNT; use crate::video::codec::h264_bitstream; @@ -19,9 +19,9 @@ use crate::video::device::{ VideoDeviceRecoveryHint, }; use crate::video::types::{ - BitratePreset, EncoderBackend, PipelineStateNotification, PixelFormat, Resolution, - SharedVideoPipeline, SharedVideoPipelineConfig, SharedVideoPipelineStats, VideoCodecType, - VideoEncoderType, + BitratePreset, EncoderBackend, PipelineLifecycle, PipelineStateNotification, PixelFormat, + Resolution, SharedVideoPipeline, SharedVideoPipelineConfig, SharedVideoPipelineStats, + VideoCodecType, VideoEncoderType, }; use super::config::{TurnServer, WebRtcConfig}; @@ -263,6 +263,7 @@ impl WebRtcStreamer { Arc::new(move |notification: PipelineStateNotification| { let recovered = update_signal_recovery_edge(&recovery_pending, notification.state); events.publish(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: notification.state.to_string(), device: Some(device.clone()), reason: notification.reason.map(|reason| reason.to_string()), @@ -398,7 +399,7 @@ impl WebRtcStreamer { ); streamer .publish_stream_event(SystemEvent::StreamDeviceLost { - kind: StreamDeviceLostKind::Video, + kind: StreamKind::Video, device: original_device.clone(), reason: reason.clone(), }) @@ -415,6 +416,7 @@ impl WebRtcStreamer { .await; streamer .publish_stream_event(SystemEvent::StreamStateChanged { + kind: StreamKind::Video, state: "device_lost".to_string(), device: Some(original_device.clone()), reason: Some("recovering".to_string()), @@ -478,9 +480,17 @@ impl WebRtcStreamer { async fn ensure_video_pipeline(&self) -> Result> { let mut pipeline_guard = self.video_pipeline.write().await; - if let Some(ref pipeline) = *pipeline_guard { - if pipeline.is_running() { - return Ok(pipeline.clone()); + if let Some(pipeline) = pipeline_guard.as_ref().cloned() { + match pipeline.lifecycle() { + PipelineLifecycle::Running => return Ok(pipeline), + PipelineLifecycle::Stopping => { + info!("Waiting for stopping video pipeline to release capture device"); + pipeline.stop_and_wait(PIPELINE_RELEASE_TIMEOUT).await?; + *pipeline_guard = None; + } + PipelineLifecycle::Stopped => { + *pipeline_guard = None; + } } } diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 39319c9b..a9d70165 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -10,6 +10,7 @@ import type { ComputerUseStartRequest, } from '@/types/generated' import { useHidWebSocket, type HidKeyboardEvent, type HidMouseEvent } from '@/composables/useHidWebSocket' +import type { StreamSignalReason, StreamState } from '@/types/websocket' const API_BASE = '/api' @@ -299,18 +300,8 @@ export interface VideoEncoderSelfCheckResponse { export const streamApi = { status: () => request<{ - state: - | 'uninitialized' - | 'ready' - | 'streaming' - | 'no_signal' - | 'no_cable' - | 'no_sync' - | 'out_of_range' - | 'device_lost' - | 'recovering' - | 'device_busy' - | 'error' + state: StreamState + reason?: StreamSignalReason | null device: string | null format: string | null resolution: [number, number] | null diff --git a/web/src/components/ComputerUseSheet.vue b/web/src/components/ComputerUseSheet.vue index d60a9858..c808a7d4 100644 --- a/web/src/components/ComputerUseSheet.vue +++ b/web/src/components/ComputerUseSheet.vue @@ -3,7 +3,7 @@ import { computed, nextTick, onMounted, ref, watch } from 'vue' import { useI18n } from 'vue-i18n' import { Bot, - BrainCircuit, + Brain, ChevronDown, ChevronRight, Image, @@ -293,7 +293,7 @@ onMounted(loadConfig) > - + {{ item.failed ? t('computerUse.reasoning.failed') diff --git a/web/src/components/VideoInputFields.vue b/web/src/components/VideoInputFields.vue index c5b9193d..80a895fc 100644 --- a/web/src/components/VideoInputFields.vue +++ b/web/src/components/VideoInputFields.vue @@ -32,7 +32,7 @@ const { t } = useI18n()