GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2026-09-12 10:34:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: 修复无信号状态同步与采集管线阻塞

Browse Source 
- 使用非阻塞 V4L2 句柄并类型化采集错误
- 完善视频管线生命周期，避免复用停止中的管线
- 区分音视频事件并支持 REST/WebSocket 状态快照
- 修复信号丢失后保留最后一帧及刷新后黑屏问题
- 优化无信号提示、状态展示优先级及相关界面细节
- 降低重复 OTG 错误日志级别
This commit is contained in:
mofeng-git
2026-07-30 22:43:59 +08:00
parent ce1712ff2e
commit 6fdcf5c7c9
24 changed files with 406 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/audio/recovery.rs
View File

@@ -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(),
});

61
src/events/mod.rs
View File

@@ -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<SystemEvent>>,
prefix_topics: std::collections::HashMap<String, broadcast::Sender<SystemEvent>>,
device_info_dirty_tx: broadcast::Sender<()>,
latest_video_stream_state: RwLock<Option<SystemEvent>>,
}
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<SystemEvent> {
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")
));
}
}

15
src/events/types.rs
View File

@@ -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<String>,
#[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(),
},

4
src/hid/otg.rs
View File

@@ -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 {

100
src/video/capture/linux.rs
View File

@@ -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<Instant>,
}
fn open_capture_device(path: &Path) -> io::Result<File> {
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<u8>) -> io::Result<CaptureMeta> {
pub fn next_into(
&mut self,
dst: &mut Vec<u8>,
) -> std::result::Result<CaptureMeta, CaptureReadError> {
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::<V4l2Event>(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);
}
}

41
src/video/capture/mod.rs
View File

@@ -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<io::Error> 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)]

16
src/video/capture/windows.rs
View File

@@ -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<u8>) -> io::Result<CaptureMeta> {
pub fn next_into(
&mut self,
dst: &mut Vec<u8>,
) -> std::result::Result<CaptureMeta, CaptureReadError> {
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)))
}
}
}

53
src/video/device/bridge.rs
View File

@@ -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<String> {
}
pub fn open_subdev(path: &Path) -> io::Result<File> {
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::<V4l2Event>(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<boo
}
}
let mut drained = 0u32;
while let Ok(_ev) = ioctl::dqevent::<V4l2Event>(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<boo
mod tests {
use super::*;
#[test]
fn subdevice_handles_are_non_blocking() {
let file = tempfile::NamedTempFile::new().unwrap();
let opened = open_subdev(file.path()).unwrap();
let flags = unsafe { libc::fcntl(opened.as_raw_fd(), libc::F_GETFL) };
assert!(flags >= 0);
assert_ne!(flags & libc::O_NONBLOCK, 0);
}
#[test]
fn rk628_fingerprint_matches_vga() {
let mut bt: v4l2_bt_timings = unsafe { std::mem::zeroed() };

4
src/video/pipeline/mod.rs
View File

@@ -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,
};

57
src/video/pipeline/shared.rs
View File

@@ -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]

38
src/video/streamer.rs
View File

@@ -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<String>,
pub device: Option<String>,
pub format: Option<String>,
pub resolution: Option<(u32, u32)>,

4
src/video/types.rs
View File

@@ -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,
};

13
src/web/handlers/stream.rs
View File

@@ -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<Arc<AppState>>) -> Json<StreamerStats> {
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

31
src/web/ws.rs
View File

@@ -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<AppState>,
topics: &[String],
replay_stream_state: bool,
event_tx: &mpsc::UnboundedSender<BusMessage>,
event_tasks: &mut Vec<JoinHandle<()>>,
) {
@@ -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<AppState>) {
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"));
}
}

26
src/webrtc/webrtc_streamer.rs
View File

@@ -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<Arc<SharedVideoPipeline>> {
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;
}
}
}