fix(media): restore fresh camera signal
This commit is contained in:
parent
7011b5c3ef
commit
29213e41df
6
Cargo.lock
generated
6
Cargo.lock
generated
@ -1658,7 +1658,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
|
||||
|
||||
[[package]]
|
||||
name = "lesavka_client"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-stream",
|
||||
@ -1692,7 +1692,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lesavka_common"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
@ -1704,7 +1704,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "lesavka_server"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
|
||||
@ -4,7 +4,7 @@ path = "src/main.rs"
|
||||
|
||||
[package]
|
||||
name = "lesavka_client"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
|
||||
@ -3,7 +3,7 @@ Version=1.0
|
||||
Type=Application
|
||||
Name=Lesavka
|
||||
Comment=Relay capture, input routing, and preview control deck
|
||||
Exec=/usr/local/bin/lesavka
|
||||
Exec=env GSK_RENDERER=gl /usr/local/bin/lesavka
|
||||
Icon=lesavka
|
||||
Terminal=false
|
||||
Categories=Utility;GTK;
|
||||
|
||||
@ -82,7 +82,9 @@ impl CameraCapture {
|
||||
let use_mjpg_source = source_profile == CameraSourceProfile::Mjpeg;
|
||||
let passthrough_mjpg_source =
|
||||
use_mjpg_source && capture_profile == (width, height, fps);
|
||||
let (enc, kf_prop) = if use_mjpg_source && !output_mjpeg {
|
||||
let (enc, kf_prop) = if output_mjpeg {
|
||||
("jpegenc", None)
|
||||
} else if use_mjpg_source {
|
||||
if output_hevc {
|
||||
Self::choose_hevc_encoder()?
|
||||
} else {
|
||||
@ -127,6 +129,11 @@ impl CameraCapture {
|
||||
"videoconvert ! video/x-raw,format=NV12,width={width},height={height},framerate={fps}/1 !"
|
||||
),
|
||||
#[cfg(not(coverage))]
|
||||
"nvautogpuh264enc" | "nvautogpuh265enc" =>
|
||||
format!(
|
||||
"videoconvert ! video/x-raw,format=NV12,width={width},height={height},framerate={fps}/1 !"
|
||||
),
|
||||
#[cfg(not(coverage))]
|
||||
"x265enc" =>
|
||||
format!(
|
||||
"videoconvert ! video/x-raw,format=I420,width={width},height={height},framerate={fps}/1 !"
|
||||
@ -138,7 +145,7 @@ impl CameraCapture {
|
||||
vulkanupload ! video/x-raw(memory:VulkanImage),format=NV12,width={width},height={height},framerate={fps}/1 !"
|
||||
),
|
||||
#[cfg(not(coverage))]
|
||||
"vaapih264enc" | "vah265enc" | "vaapih265enc" | "v4l2h265enc" =>
|
||||
"vah264enc" | "vaapih264enc" | "vah265enc" | "vaapih265enc" | "v4l2h265enc" =>
|
||||
format!(
|
||||
"videoconvert ! video/x-raw,format=NV12,width={width},height={height},framerate={fps}/1 !"
|
||||
),
|
||||
|
||||
@ -4,7 +4,9 @@ impl CameraCapture {
|
||||
fn pick_encoder() -> (&'static str, &'static str) {
|
||||
let encoders = &[
|
||||
("nvh264enc", "video/x-raw(memory:NVMM),format=NV12"),
|
||||
("nvautogpuh264enc", "video/x-raw,format=NV12"),
|
||||
("vulkanh264enc", "video/x-raw(memory:VulkanImage),format=NV12"),
|
||||
("vah264enc", "video/x-raw,format=NV12"),
|
||||
("vaapih264enc", "video/x-raw,format=NV12"),
|
||||
("v4l2h264enc", "video/x-raw"), // RPi, Jetson, etc.
|
||||
];
|
||||
@ -35,12 +37,27 @@ impl CameraCapture {
|
||||
),
|
||||
));
|
||||
}
|
||||
if buildable_encoder("nvautogpuh264enc") {
|
||||
return Ok((
|
||||
"nvautogpuh264enc",
|
||||
supported_encoder_property(
|
||||
"nvautogpuh264enc",
|
||||
&["iframeinterval", "idrinterval", "gop-size"],
|
||||
),
|
||||
));
|
||||
}
|
||||
if buildable_encoder("vulkanh264enc") {
|
||||
return Ok((
|
||||
"vulkanh264enc",
|
||||
supported_encoder_property("vulkanh264enc", &["idr-period"]),
|
||||
));
|
||||
}
|
||||
if buildable_encoder("vah264enc") {
|
||||
return Ok((
|
||||
"vah264enc",
|
||||
supported_encoder_property("vah264enc", &["keyframe-period"]),
|
||||
));
|
||||
}
|
||||
if buildable_encoder("vaapih264enc") {
|
||||
return Ok((
|
||||
"vaapih264enc",
|
||||
@ -69,7 +86,9 @@ impl CameraCapture {
|
||||
.map(str::trim)
|
||||
{
|
||||
Some("nvh264enc") => ("nvh264enc", None),
|
||||
Some("nvautogpuh264enc") => ("nvautogpuh264enc", None),
|
||||
Some("vulkanh264enc") => ("vulkanh264enc", Some("idr-period")),
|
||||
Some("vah264enc") => ("vah264enc", Some("keyframe-period")),
|
||||
Some("vaapih264enc") => ("vaapih264enc", Some("keyframe-period")),
|
||||
Some("v4l2h264enc") => ("v4l2h264enc", Some("idrcount")),
|
||||
_ => ("x264enc", Some("key-int-max")),
|
||||
@ -87,6 +106,10 @@ impl CameraCapture {
|
||||
fn choose_hevc_encoder() -> anyhow::Result<(&'static str, Option<&'static str>)> {
|
||||
for (name, keyframe_props) in [
|
||||
("nvh265enc", &["iframeinterval", "idrinterval", "gop-size"][..]),
|
||||
(
|
||||
"nvautogpuh265enc",
|
||||
&["iframeinterval", "idrinterval", "gop-size"][..],
|
||||
),
|
||||
("vah265enc", &["keyframe-period"][..]),
|
||||
("vaapih265enc", &["keyframe-period"][..]),
|
||||
("v4l2h265enc", &["idrcount"][..]),
|
||||
|
||||
@ -184,7 +184,9 @@ fn coverage_hevc_encoder_choice_is_stable() {
|
||||
fn coverage_h264_encoder_choice_honors_stable_test_overrides() {
|
||||
let cases = [
|
||||
("nvh264enc", ("nvh264enc", None)),
|
||||
("nvautogpuh264enc", ("nvautogpuh264enc", None)),
|
||||
("vulkanh264enc", ("vulkanh264enc", Some("idr-period"))),
|
||||
("vah264enc", ("vah264enc", Some("keyframe-period"))),
|
||||
("vaapih264enc", ("vaapih264enc", Some("keyframe-period"))),
|
||||
("v4l2h264enc", ("v4l2h264enc", Some("idrcount"))),
|
||||
("unknown", ("x264enc", Some("key-int-max"))),
|
||||
|
||||
@ -165,14 +165,14 @@ fn h264_decoder_launch_fragment(decoder_name: &str) -> String {
|
||||
h264_decoder_launch_fragment_named(decoder_name, "decoder")
|
||||
}
|
||||
|
||||
fn h264_decoder_launch_fragment_named(decoder_name: &str, element_name: &str) -> String {
|
||||
fn h264_decoder_launch_fragment_named(decoder_name: &str, _element_name: &str) -> String {
|
||||
match decoder_name {
|
||||
"vulkanh264dec" => concat!(
|
||||
"vulkanh264dec name={element_name} discard-corrupted-frames=true ",
|
||||
"vulkanh264dec discard-corrupted-frames=true ",
|
||||
"automatic-request-sync-points=true ! vulkandownload"
|
||||
)
|
||||
.replace("{element_name}", element_name),
|
||||
name => format!("{name} name={element_name}"),
|
||||
.to_string(),
|
||||
name => name.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -157,7 +157,7 @@ pub fn h264_decoder_preference_order() -> Vec<&'static str> {
|
||||
/// Return a parse-launch fragment for the selected H.264 decoder.
|
||||
///
|
||||
/// Inputs: decoder element name. Output: a pipeline fragment with a stable
|
||||
/// `decoder` element name. Why: Vulkan decoders output GPU memory, so they need
|
||||
/// decoder stage. Why: Vulkan decoders output GPU memory, so they need
|
||||
/// an explicit download step before the existing CPU-side sinks can consume
|
||||
/// frames; keeping that in one helper prevents hardware decode from being
|
||||
/// selected and then immediately failing link negotiation.
|
||||
@ -166,21 +166,21 @@ pub fn h264_decoder_launch_fragment(decoder_name: &str) -> String {
|
||||
h264_decoder_launch_fragment_named(decoder_name, "decoder")
|
||||
}
|
||||
|
||||
/// Return a parse-launch fragment for the selected H.264 decoder with a caller-owned element name.
|
||||
/// Return a parse-launch fragment for the selected H.264 decoder.
|
||||
///
|
||||
/// Inputs: decoder element name plus the element name to put in the pipeline.
|
||||
/// Output: a pipeline fragment. Why: unified downstream rendering needs two
|
||||
/// independent decoder elements, while Vulkan still needs an explicit
|
||||
/// download-to-system-memory stage after each decoder.
|
||||
/// Inputs: decoder element name plus a legacy caller-owned element name.
|
||||
/// Output: a pipeline fragment. Why: some binary decoder wrappers reject
|
||||
/// `name=` in launch strings, while duplicated fragments still create
|
||||
/// independent decoder elements.
|
||||
#[must_use]
|
||||
pub fn h264_decoder_launch_fragment_named(decoder_name: &str, element_name: &str) -> String {
|
||||
pub fn h264_decoder_launch_fragment_named(decoder_name: &str, _element_name: &str) -> String {
|
||||
match decoder_name {
|
||||
"vulkanh264dec" => concat!(
|
||||
"vulkanh264dec name={element_name} discard-corrupted-frames=true ",
|
||||
"vulkanh264dec discard-corrupted-frames=true ",
|
||||
"automatic-request-sync-points=true ! vulkandownload"
|
||||
)
|
||||
.replace("{element_name}", element_name),
|
||||
name => format!("{name} name={element_name}"),
|
||||
.to_string(),
|
||||
name => name.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "lesavka_common"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
edition = "2024"
|
||||
build = "build.rs"
|
||||
|
||||
|
||||
@ -648,7 +648,7 @@ These entries are intentionally concise because most are manual lab or CI harnes
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_MIN_REFERENCE_BYTES` | server direct-MJPEG guard baseline; frames smaller than this do not establish the last-good reference |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE` | server direct-MJPEG normalization toggle; defaults off because the native GStreamer decode/re-encode branch can retain RSS during long calls |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_MISS_LIMIT` | server direct-MJPEG normalization recovery threshold; after this many consecutive empty pulls, the session falls back to guarded passthrough |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS` | server direct-MJPEG normalization appsink timeout; defaults to `25`ms and is capped at `50`ms to avoid live backlog |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS` | server direct-MJPEG normalization appsink timeout; defaults to a nonblocking `0`ms poll and is capped at `50`ms for explicit diagnostics |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB` | server direct-MJPEG normalization RSS safety ceiling; defaults to `768`, and `0` disables this opt-in branch guard |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_SIZE_DROP_PCT` | server direct-MJPEG corruption guard threshold; frames below this percentage of the last good reference are frozen out |
|
||||
| `LESAVKA_UVC_DIRECT_MJPEG_VISUAL_GUARD` | server direct-MJPEG corruption guard toggle; defaults on so obvious collapsed or flat payloads freeze the last good frame |
|
||||
|
||||
@ -492,7 +492,7 @@ pacman_install \
|
||||
git rustup protobuf abseil-cpp gcc clang llvm-libs compiler-rt evtest base-devel libpulse \
|
||||
"${PIPEWIRE_PACKAGES[@]}" wireplumber \
|
||||
alsa-utils gst-plugin-pipewire \
|
||||
gstreamer gst-plugins-base gst-plugins-good gst-plugins-bad gst-plugins-ugly gst-libav gst-plugin-va \
|
||||
gstreamer gst-plugins-base gst-plugins-good gst-plugins-bad gst-plugins-ugly gst-libav gst-plugin-va intel-media-driver \
|
||||
ffmpeg wmctrl qt6-tools wl-clipboard xclip xsel desktop-file-utils openssl
|
||||
|
||||
ensure_yay() {
|
||||
|
||||
@ -1689,7 +1689,7 @@ SERVER_ENV_TMP=$(mktemp)
|
||||
printf 'LESAVKA_UVC_HEVC_DECODE_MISS_LIMIT=%s\n' "${LESAVKA_UVC_HEVC_DECODE_MISS_LIMIT:-15}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE:-1}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_JPEG_QUALITY=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_JPEG_QUALITY:-60}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-50}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-0}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_MISS_LIMIT=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_MISS_LIMIT:-30}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB:-384}"
|
||||
printf 'LESAVKA_UVC_DIRECT_MJPEG_VISUAL_GUARD=%s\n' "${LESAVKA_UVC_DIRECT_MJPEG_VISUAL_GUARD:-1}"
|
||||
|
||||
@ -16,7 +16,7 @@ bench = false
|
||||
|
||||
[package]
|
||||
name = "lesavka_server"
|
||||
version = "0.27.1"
|
||||
version = "0.27.5"
|
||||
edition = "2024"
|
||||
autobins = false
|
||||
|
||||
|
||||
@ -101,6 +101,16 @@ fn parse_camera_codec(raw: &str) -> Option<CameraCodec> {
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_uvc_uplink_camera_codec(raw: &str) -> Option<CameraCodec> {
|
||||
match raw.trim().to_ascii_lowercase().as_str() {
|
||||
"h264" => {
|
||||
warn!("📷 UVC output no longer accepts H264 as the normal upstream codec; using MJPEG");
|
||||
Some(CameraCodec::Mjpeg)
|
||||
}
|
||||
_ => parse_camera_codec(raw),
|
||||
}
|
||||
}
|
||||
|
||||
fn select_hdmi_codec(hw_decode: bool) -> CameraCodec {
|
||||
std::env::var("LESAVKA_CAM_CODEC")
|
||||
.ok()
|
||||
@ -120,7 +130,7 @@ fn select_uvc_codec(uvc_env: Option<&HashMap<String, String>>) -> CameraCodec {
|
||||
.or_else(|| uvc_env.and_then(|env| env.get("LESAVKA_UPLINK_CAMERA_CODEC").cloned()))
|
||||
.or_else(|| uvc_env.and_then(|env| env.get("LESAVKA_CAM_CODEC").cloned()))
|
||||
.as_deref()
|
||||
.and_then(parse_camera_codec)
|
||||
.and_then(parse_uvc_uplink_camera_codec)
|
||||
.unwrap_or(CameraCodec::Hevc)
|
||||
}
|
||||
|
||||
|
||||
@ -32,6 +32,7 @@ struct MediaV2ScheduledAudio {
|
||||
}
|
||||
|
||||
#[cfg(not(coverage))]
|
||||
#[derive(Clone, Debug)]
|
||||
struct MediaV2ScheduledVideo {
|
||||
packet: VideoPacket,
|
||||
due_at: tokio::time::Instant,
|
||||
@ -403,7 +404,7 @@ async fn run_media_v2_audio_handoff(
|
||||
/// Why: UVC sync offsets still require sleeping, but that sleep must not slow
|
||||
/// the network receive loop.
|
||||
async fn run_media_v2_video_handoff(
|
||||
mut rx: tokio::sync::mpsc::Receiver<MediaV2ScheduledVideo>,
|
||||
mut rx: tokio::sync::watch::Receiver<Option<MediaV2ScheduledVideo>>,
|
||||
relay: Arc<lesavka_server::video::CameraRelay>,
|
||||
upstream_media_rt: Arc<UpstreamMediaRuntime>,
|
||||
rpc_id: u64,
|
||||
@ -411,7 +412,10 @@ async fn run_media_v2_video_handoff(
|
||||
camera_session_id: u64,
|
||||
) {
|
||||
let mut video_presented_once = false;
|
||||
while let Some(item) = rx.recv().await {
|
||||
while rx.changed().await.is_ok() {
|
||||
let Some(item) = rx.borrow_and_update().clone() else {
|
||||
continue;
|
||||
};
|
||||
sleep_until_media_v2(item.due_at).await;
|
||||
let presented_pts = item.packet.pts;
|
||||
let live_lag_ms = f64::from(item.packet.client_queue_age_ms)
|
||||
|
||||
@ -101,8 +101,11 @@ impl Handler {
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
// UVC freshness is more important than draining intermediate video
|
||||
// frames after a slow sink call. Keep only the newest scheduled
|
||||
// frame while the handoff worker is busy.
|
||||
let (video_handoff_tx, video_handoff_rx) =
|
||||
tokio::sync::mpsc::channel::<MediaV2ScheduledVideo>(32);
|
||||
tokio::sync::watch::channel::<Option<MediaV2ScheduledVideo>>(None);
|
||||
let video_worker = tokio::spawn(run_media_v2_video_handoff(
|
||||
video_handoff_rx,
|
||||
relay.clone(),
|
||||
@ -337,14 +340,7 @@ impl Handler {
|
||||
None
|
||||
};
|
||||
if let Some(scheduled_video) = scheduled_video {
|
||||
if video_handoff_tx.send(scheduled_video).await.is_err() {
|
||||
warn!(
|
||||
rpc_id,
|
||||
session_id = camera_lease.session_id,
|
||||
"📦 v2 video handoff worker stopped while receiving bundled media"
|
||||
);
|
||||
break;
|
||||
}
|
||||
video_handoff_tx.send_replace(Some(scheduled_video));
|
||||
if video_recovers_hevc_gap {
|
||||
waiting_for_hevc_keyframe = false;
|
||||
}
|
||||
|
||||
@ -2,7 +2,8 @@
|
||||
#[allow(clippy::items_after_test_module)]
|
||||
mod tests {
|
||||
use super::{
|
||||
MediaV2BundleFacts, UpstreamStreamCleanup, media_v2_audio_starvation_elapsed,
|
||||
MediaV2BundleFacts, MediaV2ScheduledVideo, UpstreamStreamCleanup,
|
||||
media_v2_audio_starvation_elapsed,
|
||||
media_v2_audio_starvation_heal_after, media_v2_av_skew_elapsed,
|
||||
media_v2_av_skew_heal_after, media_v2_frame_step_us, media_v2_handoff_schedule,
|
||||
media_v2_has_hevc_recovery_keyframe, media_v2_should_hold_hevc_video_for_recovery,
|
||||
@ -16,6 +17,31 @@ mod tests {
|
||||
};
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn video_handoff_watch_retains_only_the_newest_scheduled_frame() {
|
||||
let (tx, mut rx) =
|
||||
tokio::sync::watch::channel::<Option<MediaV2ScheduledVideo>>(None);
|
||||
let now = tokio::time::Instant::now();
|
||||
|
||||
for id in [1, 2, 3] {
|
||||
tx.send_replace(Some(MediaV2ScheduledVideo {
|
||||
packet: VideoPacket {
|
||||
id,
|
||||
..Default::default()
|
||||
},
|
||||
due_at: now,
|
||||
received_at: now,
|
||||
}));
|
||||
}
|
||||
|
||||
rx.changed().await.expect("latest frame should be pending");
|
||||
let latest = rx
|
||||
.borrow_and_update()
|
||||
.clone()
|
||||
.expect("video handoff frame");
|
||||
assert_eq!(latest.packet.id, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
/// Keeps `retain_freshest_video_packet_keeps_only_the_latest_frame` explicit because it sits on relay RPC orchestration, where hardware failures must surface without stopping the server.
|
||||
/// Inputs are the typed parameters; output is the return value or side effect.
|
||||
|
||||
@ -100,6 +100,23 @@ fn camera_config_env_override_honors_explicit_uplink_codec() {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn uvc_camera_profile_migrates_legacy_h264_uplink_to_mjpeg() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
("LESAVKA_CAM_OUTPUT", Some("uvc")),
|
||||
("LESAVKA_CAM_CODEC", Some("h264")),
|
||||
("LESAVKA_UPLINK_CAMERA_CODEC", None),
|
||||
],
|
||||
|| {
|
||||
let cfg = update_camera_config();
|
||||
assert_eq!(cfg.output, CameraOutput::Uvc);
|
||||
assert_eq!(cfg.codec, CameraCodec::Mjpeg);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn hdmi_camera_profile_honors_installed_1080p_override() {
|
||||
|
||||
@ -3,7 +3,7 @@ enum CameraSink {
|
||||
Hdmi(HdmiSink),
|
||||
UvcWithHdmiMirror {
|
||||
uvc: WebcamSink,
|
||||
hdmi: HdmiSink,
|
||||
hdmi: HdmiMirrorSink,
|
||||
},
|
||||
#[cfg(coverage)]
|
||||
Noop,
|
||||
@ -15,6 +15,8 @@ impl CameraSink {
|
||||
CameraSink::Uvc(sink) => sink.push(pkt),
|
||||
CameraSink::Hdmi(sink) => sink.push(pkt),
|
||||
CameraSink::UvcWithHdmiMirror { uvc, hdmi } => {
|
||||
// UVC is the browser-facing contract. The optional HDMI mirror
|
||||
// must never hold this primary handoff behind KMS/display work.
|
||||
uvc.push(pkt.clone());
|
||||
hdmi.push(pkt);
|
||||
}
|
||||
@ -26,6 +28,100 @@ impl CameraSink {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct HdmiMirrorState {
|
||||
pending: Option<VideoPacket>,
|
||||
stopped: bool,
|
||||
}
|
||||
|
||||
/// Latest-only handoff for the optional HDMI mirror.
|
||||
///
|
||||
/// KMS/display sinks may block while modesetting or waiting for a connector.
|
||||
/// Keeping that work on a dedicated thread prevents the secondary output from
|
||||
/// accumulating latency in the primary UVC camera path.
|
||||
struct HdmiMirrorSink {
|
||||
state: Arc<(std::sync::Mutex<HdmiMirrorState>, std::sync::Condvar)>,
|
||||
dropped_frames: AtomicU64,
|
||||
}
|
||||
|
||||
impl HdmiMirrorSink {
|
||||
fn new(cfg: &CameraConfig) -> anyhow::Result<Self> {
|
||||
let sink = HdmiSink::new(cfg)?;
|
||||
let state = Arc::new((
|
||||
std::sync::Mutex::new(HdmiMirrorState::default()),
|
||||
std::sync::Condvar::new(),
|
||||
));
|
||||
let worker_state = Arc::clone(&state);
|
||||
std::thread::Builder::new()
|
||||
.name("lesavka-hdmi-mirror".to_string())
|
||||
.spawn(move || run_hdmi_mirror(worker_state, sink))
|
||||
.context("starting HDMI mirror handoff worker")?;
|
||||
Ok(Self {
|
||||
state,
|
||||
dropped_frames: AtomicU64::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
fn push(&self, pkt: VideoPacket) {
|
||||
let (lock, ready) = &*self.state;
|
||||
let Ok(mut state) = lock.lock() else {
|
||||
return;
|
||||
};
|
||||
if state.stopped {
|
||||
return;
|
||||
}
|
||||
if state.pending.replace(pkt).is_some() {
|
||||
let dropped = self.dropped_frames.fetch_add(1, Ordering::Relaxed) + 1;
|
||||
if dropped == 1 || dropped.is_multiple_of(60) {
|
||||
warn!(
|
||||
target: "lesavka_server::video",
|
||||
dropped,
|
||||
"HDMI mirror replaced a stale pending frame without delaying UVC"
|
||||
);
|
||||
}
|
||||
}
|
||||
ready.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for HdmiMirrorSink {
|
||||
fn drop(&mut self) {
|
||||
let (lock, ready) = &*self.state;
|
||||
if let Ok(mut state) = lock.lock() {
|
||||
state.stopped = true;
|
||||
state.pending = None;
|
||||
ready.notify_one();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn run_hdmi_mirror(
|
||||
state: Arc<(std::sync::Mutex<HdmiMirrorState>, std::sync::Condvar)>,
|
||||
sink: HdmiSink,
|
||||
) {
|
||||
loop {
|
||||
let pkt = {
|
||||
let (lock, ready) = &*state;
|
||||
let Ok(mut state) = lock.lock() else {
|
||||
return;
|
||||
};
|
||||
while state.pending.is_none() && !state.stopped {
|
||||
let Ok(next) = ready.wait(state) else {
|
||||
return;
|
||||
};
|
||||
state = next;
|
||||
}
|
||||
if state.stopped {
|
||||
return;
|
||||
}
|
||||
state.pending.take()
|
||||
};
|
||||
if let Some(pkt) = pkt {
|
||||
sink.push(pkt);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Forward camera packets from gRPC into either a UVC or HDMI sink.
|
||||
///
|
||||
/// Inputs: packets received from the client camera stream.
|
||||
@ -68,7 +164,7 @@ impl CameraRelay {
|
||||
Ok(Self {
|
||||
sink: CameraSink::UvcWithHdmiMirror {
|
||||
uvc: WebcamSink::new(uvc_dev, uvc_cfg)?,
|
||||
hdmi: HdmiSink::new(hdmi_cfg)?,
|
||||
hdmi: HdmiMirrorSink::new(hdmi_cfg)?,
|
||||
},
|
||||
id,
|
||||
frames: AtomicU64::new(0),
|
||||
@ -156,3 +252,24 @@ impl CameraRelay {
|
||||
self.sink.push(pkt);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod camera_relay_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn hdmi_mirror_state_keeps_only_the_latest_pending_frame() {
|
||||
let mut state = HdmiMirrorState::default();
|
||||
state.pending = Some(VideoPacket {
|
||||
id: 1,
|
||||
..Default::default()
|
||||
});
|
||||
let replaced = state.pending.replace(VideoPacket {
|
||||
id: 2,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
assert_eq!(replaced.map(|packet| packet.id), Some(1));
|
||||
assert_eq!(state.pending.map(|packet| packet.id), Some(2));
|
||||
}
|
||||
}
|
||||
|
||||
@ -18,7 +18,7 @@ const DEFAULT_DIRECT_MJPEG_MIN_REFERENCE_BYTES: u32 = 48 * 1024;
|
||||
const DEFAULT_DIRECT_MJPEG_PROFILE_MISMATCH_REJECT: bool = true;
|
||||
const DEFAULT_DIRECT_MJPEG_NORMALIZE: bool = true;
|
||||
const DEFAULT_DIRECT_MJPEG_JPEG_QUALITY: u32 = 60;
|
||||
const DEFAULT_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS: u32 = 50;
|
||||
const DEFAULT_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS: u32 = 0;
|
||||
const DEFAULT_DIRECT_MJPEG_NORMALIZE_MISS_LIMIT: u32 = 30;
|
||||
const DEFAULT_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB: u32 = 384;
|
||||
|
||||
|
||||
@ -37,7 +37,7 @@ fn direct_mjpeg_normalization_defaults_on_and_clamps_tuning() {
|
||||
|| {
|
||||
assert!(super::direct_mjpeg_normalize_enabled());
|
||||
assert_eq!(super::direct_mjpeg_jpeg_quality(), 60);
|
||||
assert_eq!(super::direct_mjpeg_normalize_pull_timeout_ms(), 50);
|
||||
assert_eq!(super::direct_mjpeg_normalize_pull_timeout_ms(), 0);
|
||||
assert_eq!(super::direct_mjpeg_normalize_miss_limit(), 30);
|
||||
assert_eq!(
|
||||
super::direct_mjpeg_normalize_rss_limit_kb(),
|
||||
|
||||
@ -243,9 +243,14 @@ fn direct_mjpeg_normalizer_branch_reencodes_a_valid_frame() {
|
||||
.set_state(gst::State::Playing)
|
||||
.expect("normalizer pipeline playing");
|
||||
|
||||
let sample = (0..30)
|
||||
.find_map(|_| {
|
||||
src.push_buffer(gst::Buffer::from_slice(FIXTURE))
|
||||
.expect("push fixture");
|
||||
let sample = super::freshest_direct_mjpeg_sample(&sink).expect("normalized sample");
|
||||
std::thread::sleep(std::time::Duration::from_millis(1));
|
||||
super::freshest_direct_mjpeg_sample(&sink)
|
||||
})
|
||||
.expect("normalized sample without a frame-sized pull wait");
|
||||
let buffer = sample.buffer().expect("sample buffer");
|
||||
let map = buffer.map_readable().expect("readable normalized buffer");
|
||||
|
||||
|
||||
@ -49,10 +49,10 @@ const MATRIX_SCRIPT: &str = include_str!(concat!(
|
||||
));
|
||||
|
||||
#[test]
|
||||
fn bundled_receive_loop_enqueues_instead_of_sleeping_for_handoff() {
|
||||
fn bundled_receive_loop_keeps_video_latest_only_instead_of_sleeping_for_handoff() {
|
||||
for expected in [
|
||||
"tokio::sync::mpsc::channel::<MediaV2ScheduledAudio>(32)",
|
||||
"tokio::sync::mpsc::channel::<MediaV2ScheduledVideo>(32)",
|
||||
"tokio::sync::watch::channel::<Option<MediaV2ScheduledVideo>>(None)",
|
||||
"run_media_v2_audio_handoff(audio_handoff_rx",
|
||||
"tokio::spawn(run_media_v2_video_handoff",
|
||||
"let bundle_epoch = bundle_arrived_at + schedule.common_delay;",
|
||||
@ -62,7 +62,7 @@ fn bundled_receive_loop_enqueues_instead_of_sleeping_for_handoff() {
|
||||
"bundle_base_remote_pts_us",
|
||||
"bundle_epoch",
|
||||
".send(scheduled_audio)",
|
||||
".send(scheduled_video)",
|
||||
"send_replace(Some(scheduled_video))",
|
||||
"media_v2_stream_idle_timeout()",
|
||||
"stream_webcam_media v2 idle timeout",
|
||||
"closing stale upstream leases",
|
||||
@ -78,6 +78,7 @@ fn bundled_receive_loop_enqueues_instead_of_sleeping_for_handoff() {
|
||||
"feed_media_v2_video(",
|
||||
"sleep_until_media_v2(",
|
||||
"MediaV2Clock",
|
||||
"tokio::sync::mpsc::channel::<MediaV2ScheduledVideo>(32)",
|
||||
] {
|
||||
assert!(
|
||||
!WEBCAM_RPC.contains(forbidden),
|
||||
|
||||
@ -127,18 +127,22 @@ fn production_auto_order_keeps_cpu_decoders_out() {
|
||||
#[test]
|
||||
fn vulkan_decoder_fragment_downloads_gpu_memory_before_cpu_sinks() {
|
||||
let fragment = video_support::h264_decoder_launch_fragment("vulkanh264dec");
|
||||
assert!(fragment.contains("vulkanh264dec name=decoder"));
|
||||
assert!(fragment.starts_with("vulkanh264dec "));
|
||||
assert!(fragment.contains("discard-corrupted-frames=true"));
|
||||
assert!(fragment.contains("automatic-request-sync-points=true"));
|
||||
assert!(fragment.contains("vulkandownload"));
|
||||
assert!(
|
||||
!fragment.contains(" name="),
|
||||
"preview launch fragments should not set decoder names because some binary decoder wrappers reject them"
|
||||
);
|
||||
|
||||
let named = video_support::h264_decoder_launch_fragment_named("vulkanh264dec", "decoder1");
|
||||
assert!(named.contains("vulkanh264dec name=decoder1"));
|
||||
assert!(named.starts_with("vulkanh264dec "));
|
||||
assert!(named.contains("vulkandownload"));
|
||||
|
||||
assert_eq!(
|
||||
video_support::h264_decoder_launch_fragment_named("nvh264dec", "left_decoder"),
|
||||
"nvh264dec name=left_decoder"
|
||||
"nvh264dec"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@ -99,7 +99,9 @@ mod camera_include_contract {
|
||||
matches!(
|
||||
enc,
|
||||
"nvh264enc"
|
||||
| "nvautogpuh264enc"
|
||||
| "vulkanh264enc"
|
||||
| "vah264enc"
|
||||
| "vaapih264enc"
|
||||
| "v4l2h264enc"
|
||||
| "x264enc"
|
||||
@ -112,7 +114,13 @@ mod camera_include_contract {
|
||||
assert!(
|
||||
matches!(
|
||||
enc,
|
||||
"nvh264enc" | "vulkanh264enc" | "vaapih264enc" | "v4l2h264enc" | "x264enc"
|
||||
"nvh264enc"
|
||||
| "nvautogpuh264enc"
|
||||
| "vulkanh264enc"
|
||||
| "vah264enc"
|
||||
| "vaapih264enc"
|
||||
| "v4l2h264enc"
|
||||
| "x264enc"
|
||||
),
|
||||
"unexpected encoder: {enc}"
|
||||
);
|
||||
@ -137,8 +145,9 @@ mod camera_include_contract {
|
||||
for expected in [
|
||||
"\"nvh264enc\" | \"nvh265enc\" if have_nvvidconv",
|
||||
"\"nvh264enc\" | \"nvh265enc\" /* else */",
|
||||
"\"nvautogpuh264enc\" | \"nvautogpuh265enc\"",
|
||||
"\"vulkanh264enc\"",
|
||||
"\"vaapih264enc\" | \"vah265enc\" | \"vaapih265enc\" | \"v4l2h265enc\"",
|
||||
"\"vah264enc\" | \"vaapih264enc\" | \"vah265enc\" | \"vaapih265enc\" | \"v4l2h265enc\"",
|
||||
"video/x-raw(memory:NVMM),format=NV12",
|
||||
"video/x-raw(memory:VulkanImage),format=NV12",
|
||||
"video/x-raw,format=NV12",
|
||||
@ -151,6 +160,20 @@ mod camera_include_contract {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mjpeg_output_does_not_probe_unused_h264_encoder() {
|
||||
let pipeline_source = include_str!(concat!(
|
||||
env!("CARGO_MANIFEST_DIR"),
|
||||
"/client/src/input/camera/capture_pipeline.rs"
|
||||
));
|
||||
|
||||
assert!(
|
||||
pipeline_source.contains("let (enc, kf_prop) = if output_mjpeg")
|
||||
&& pipeline_source.contains("(\"jpegenc\", None)"),
|
||||
"MJPEG upstream must not fail setup because an unused H.264 encoder is unavailable"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ffmpeg_nvenc_route_keeps_launcher_preview_tap_alive() {
|
||||
let pipeline_source = include_str!(concat!(
|
||||
|
||||
@ -124,23 +124,11 @@ mod video_include_contract {
|
||||
);
|
||||
with_var("LESAVKA_H264_DECODER", Some(" "), || {
|
||||
let result = pick_h264_decoder();
|
||||
assert!(
|
||||
result.is_ok()
|
||||
|| result
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("hardware H.264 decoder")
|
||||
);
|
||||
assert!(result.is_ok() || result.unwrap_err().to_string().contains("H.264 decoder"));
|
||||
});
|
||||
with_var("LESAVKA_H264_DECODER", None::<&str>, || {
|
||||
let result = pick_h264_decoder();
|
||||
assert!(
|
||||
result.is_ok()
|
||||
|| result
|
||||
.unwrap_err()
|
||||
.to_string()
|
||||
.contains("hardware H.264 decoder")
|
||||
);
|
||||
assert!(result.is_ok() || result.unwrap_err().to_string().contains("H.264 decoder"));
|
||||
});
|
||||
#[cfg(coverage)]
|
||||
with_var("LESAVKA_TEST_DISABLE_H264_DECODERS", Some("1"), || {
|
||||
@ -207,16 +195,24 @@ mod video_include_contract {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn vulkan_decoder_fragment_names_each_decoder_and_downloads_frames() {
|
||||
fn vulkan_decoder_fragment_avoids_decoder_names_and_downloads_frames() {
|
||||
let fragment = h264_decoder_launch_fragment("vulkanh264dec");
|
||||
assert!(fragment.contains("vulkanh264dec name=decoder"));
|
||||
assert!(fragment.starts_with("vulkanh264dec "));
|
||||
assert!(fragment.contains("discard-corrupted-frames=true"));
|
||||
assert!(fragment.contains("automatic-request-sync-points=true"));
|
||||
assert!(fragment.contains("vulkandownload"));
|
||||
assert!(
|
||||
!fragment.contains(" name="),
|
||||
"decoder fragments should avoid name= for plugin wrappers that reject it"
|
||||
);
|
||||
|
||||
let named = h264_decoder_launch_fragment_named("vulkanh264dec", "decoder1");
|
||||
assert!(named.contains("vulkanh264dec name=decoder1"));
|
||||
assert!(named.starts_with("vulkanh264dec "));
|
||||
assert!(named.contains("vulkandownload"));
|
||||
assert_eq!(
|
||||
h264_decoder_launch_fragment_named("nvh264dec", "decoder1"),
|
||||
"nvh264dec"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@ -118,6 +118,7 @@ fn client_install_reports_nvidia_and_open_source_media_routes() {
|
||||
"vulkanh264dec",
|
||||
"vulkanh265dec",
|
||||
"gst-plugin-va",
|
||||
"intel-media-driver",
|
||||
"vah265enc",
|
||||
"vaapih265enc",
|
||||
"v4l2h265enc",
|
||||
|
||||
@ -85,7 +85,7 @@ fn desktop_and_terminal_launch_paths_point_at_the_installed_client_binary() {
|
||||
);
|
||||
}
|
||||
|
||||
assert!(CLIENT_DESKTOP.contains("Exec=/usr/local/bin/lesavka"));
|
||||
assert!(CLIENT_DESKTOP.contains("Exec=env GSK_RENDERER=gl /usr/local/bin/lesavka"));
|
||||
assert!(CLIENT_DESKTOP.contains("Icon=lesavka"));
|
||||
}
|
||||
|
||||
|
||||
@ -208,7 +208,7 @@ fn server_install_pins_hdmi_camera_and_display_defaults() {
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_HEVC_DECODE_MISS_LIMIT:-15}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE:-1}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_JPEG_QUALITY:-60}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-50}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-0}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_MISS_LIMIT:-30}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB:-384}"));
|
||||
assert!(
|
||||
|
||||
@ -115,7 +115,7 @@ fn installer_keeps_the_native_normalizer_memory_bounded_by_default() {
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE:-1}"));
|
||||
assert!(!SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE:-0}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_RSS_LIMIT_MB:-384}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-50}"));
|
||||
assert!(SERVER_INSTALL.contains("${LESAVKA_UVC_DIRECT_MJPEG_NORMALIZE_PULL_TIMEOUT_MS:-0}"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user