From 47af17bebcc2d5ed0ff36072f4bbc9eb08dc83c7 Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Sun, 2 Aug 2026 16:23:28 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=20Amlogic=20AMLENC?= =?UTF-8?q?=20=E7=A1=AC=E4=BB=B6=E7=BC=96=E7=A0=81=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增动态加载及 ABI 校验的 AMLENC H.264/H.265 编码器 - 注册 Amlogic 后端并扩展编码器自检 - 并行执行 MJPEG 解码与 AMLENC 编码 - 复用 NV12 缓冲区并移除冗余帧初始化 - 优化低延迟帧分发及 RTCP 关键帧恢复 - 动态调整 AMLENC 码率尚未完成，效果不佳 --- Cargo.toml | 2 + res/vcpkg/libyuv/src/lib.rs | 68 +- src/auth/middleware.rs | 7 +- src/config/schema/stream.rs | 2 + src/stream_encoder.rs | 24 + src/video/codec/amlenc.rs | 985 ++++++++++++++++++++++++++++ src/video/codec/convert.rs | 72 +- src/video/codec/h264.rs | 4 + src/video/codec/h265.rs | 4 + src/video/codec/mod.rs | 2 + src/video/codec/registry.rs | 156 +++++ src/video/codec/self_check.rs | 38 +- src/video/pipeline/encoder_state.rs | 193 ++++-- src/video/pipeline/shared.rs | 230 +++---- src/web/routes.rs | 1 + src/webrtc/universal_session.rs | 98 ++- src/webrtc/webrtc_streamer.rs | 21 + web/src/types/generated.ts | 1 + 18 files changed, 1683 insertions(+), 225 deletions(-) create mode 100644 src/video/codec/amlenc.rs diff --git a/Cargo.toml b/Cargo.toml index 6321eb04..05be9935 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ desktop = [ "dep:serialport", "dep:async-trait", "dep:libc", + "dep:libloading", "dep:ventoy-img", "dep:protobuf", "dep:sodiumoxide", @@ -156,6 +157,7 @@ sdp-types = { version = "0.1", optional = true } serialport = { version = "4", optional = true } async-trait = { version = "0.1", optional = true } libc = { version = "0.2", optional = true } +libloading = { version = "0.8", optional = true } # Ventoy bootable image support ventoy-img = { path = "libs/ventoy-img-rs", optional = true } diff --git a/res/vcpkg/libyuv/src/lib.rs b/res/vcpkg/libyuv/src/lib.rs index 327bb9e3..70d4182d 100644 --- a/res/vcpkg/libyuv/src/lib.rs +++ b/res/vcpkg/libyuv/src/lib.rs @@ -1119,25 +1119,73 @@ pub fn mjpg_size(src: &[u8]) -> Result<(i32, i32)> { /// Decode MJPEG directly to NV12. pub fn mjpg_to_nv12(src: &[u8], dst: &mut [u8], width: i32, height: i32) -> Result<()> { - if width % 2 != 0 || height % 2 != 0 { - return Err(YuvError::InvalidDimensions); - } - - let w = width as usize; - let h = height as usize; - if dst.len() < nv12_size(w, h) { + let (y_size, output_size) = mjpg_nv12_plane_sizes(width, height)?; + if dst.len() < output_size { return Err(YuvError::BufferTooSmall); } - let y_size = w * h; let (dst_y, dst_uv) = dst.split_at_mut(y_size); + // SAFETY: the length check above guarantees writable storage for both planes. + unsafe { mjpg_to_nv12_raw(src, dst_y.as_mut_ptr(), dst_uv.as_mut_ptr(), width, height) } +} +/// Decode MJPEG directly into a reusable `Vec` without zero-filling the output first. +/// +/// `Vec::resize` must initialize every byte before libyuv immediately overwrites the +/// complete NV12 frame. This variant lets libyuv initialize spare capacity directly +/// and publishes the new length only after a successful conversion. +pub fn mjpg_to_nv12_vec(src: &[u8], dst: &mut Vec, width: i32, height: i32) -> Result<()> { + let (y_size, output_size) = mjpg_nv12_plane_sizes(width, height)?; + + dst.clear(); + dst.reserve(output_size); + + // SAFETY: reserve above guarantees writable capacity for the Y and UV planes. + // MJPGToNV12 writes the complete output on success; set_len is deliberately + // delayed until then so callers can never observe partially initialized bytes. + let result = unsafe { + let dst_y = dst.as_mut_ptr(); + mjpg_to_nv12_raw(src, dst_y, dst_y.add(y_size), width, height) + }; + result?; + + // SAFETY: a successful MJPGToNV12 call initialized exactly output_size bytes. + unsafe { dst.set_len(output_size) }; + Ok(()) +} + +#[inline] +fn mjpg_nv12_plane_sizes(width: i32, height: i32) -> Result<(usize, usize)> { + if width % 2 != 0 || height % 2 != 0 || width <= 0 || height <= 0 { + return Err(YuvError::InvalidDimensions); + } + let y_size = (width as usize) + .checked_mul(height as usize) + .ok_or(YuvError::InvalidDimensions)?; + let output_size = y_size + .checked_mul(3) + .map(|size| size / 2) + .ok_or(YuvError::InvalidDimensions)?; + Ok((y_size, output_size)) +} + +/// # Safety +/// +/// `dst_y` and `dst_uv` must point to writable planes sized for `width` x `height` NV12. +#[inline] +unsafe fn mjpg_to_nv12_raw( + src: &[u8], + dst_y: *mut u8, + dst_uv: *mut u8, + width: i32, + height: i32, +) -> Result<()> { call_yuv!(MJPGToNV12( src.as_ptr(), usize_to_size_t(src.len()), - dst_y.as_mut_ptr(), + dst_y, width, - dst_uv.as_mut_ptr(), + dst_uv, width, width, height, diff --git a/src/auth/middleware.rs b/src/auth/middleware.rs index 62c4f807..13885088 100644 --- a/src/auth/middleware.rs +++ b/src/auth/middleware.rs @@ -93,6 +93,11 @@ fn is_public_endpoint(path: &str) -> bool { fn is_setup_public_endpoint(path: &str) -> bool { matches!( path, - "/setup" | "/setup/init" | "/devices" | "/video/input-status" | "/stream/codecs" + "/setup" + | "/setup/init" + | "/devices" + | "/video/input-status" + | "/stream/codecs" + | "/video/codecs" ) } diff --git a/src/config/schema/stream.rs b/src/config/schema/stream.rs index 71920265..25908e4c 100644 --- a/src/config/schema/stream.rs +++ b/src/config/schema/stream.rs @@ -103,6 +103,7 @@ pub enum EncoderType { Amf, Rkmpp, V4l2m2m, + Amlogic, } impl EncoderType { @@ -116,6 +117,7 @@ impl EncoderType { EncoderType::Amf => "AMD AMF", EncoderType::Rkmpp => "Rockchip MPP", EncoderType::V4l2m2m => "V4L2 M2M", + EncoderType::Amlogic => "AMLENC", } } } diff --git a/src/stream_encoder.rs b/src/stream_encoder.rs index ce0360f5..4bae805a 100644 --- a/src/stream_encoder.rs +++ b/src/stream_encoder.rs @@ -14,5 +14,29 @@ pub fn encoder_type_to_backend(encoder: EncoderType) -> Option { EncoderType::Amf => Some(EncoderBackend::Amf), EncoderType::Rkmpp => Some(EncoderBackend::Rkmpp), EncoderType::V4l2m2m => Some(EncoderBackend::V4l2m2m), + EncoderType::Amlogic => Some(EncoderBackend::Amlogic), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn maps_amlogic_config_to_backend() { + assert_eq!( + encoder_type_to_backend(EncoderType::Amlogic), + Some(EncoderBackend::Amlogic) + ); + } + + #[test] + fn amlogic_config_json_round_trip() { + let json = serde_json::to_string(&EncoderType::Amlogic).unwrap(); + assert_eq!(json, "\"amlogic\""); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + EncoderType::Amlogic + ); } } diff --git a/src/video/codec/amlenc.rs b/src/video/codec/amlenc.rs new file mode 100644 index 00000000..99fa06b1 --- /dev/null +++ b/src/video/codec/amlenc.rs @@ -0,0 +1,985 @@ +//! Native Amlogic AMLENC bindings for the S912/GXM vendor Linux 4.9 stack. +//! +//! The vendor libraries are deliberately loaded at runtime. They must be built +//! with the One-KVM ABI v1 patch from the standalone `amlenc` repository; +//! unpatched 0.4 libraries +//! are rejected before any device access is attempted. + +use std::env; +use std::ffi::{c_int, c_long, c_uchar, c_uint, OsStr}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use libloading::Library; +use tracing::{debug, warn}; + +use crate::error::{AppError, Result}; +use crate::video::format::Resolution; + +pub const AMLENC_ABI_VERSION: c_int = 1; +pub const AMLENC_H264_CODEC_NAME: &str = "h264_amlenc"; +pub const AMLENC_H265_CODEC_NAME: &str = "hevc_amlenc"; +pub const AMLENC_H264_DEFAULT_LIBRARY: &str = "libvpcodec.so"; +pub const AMLENC_H265_DEFAULT_LIBRARY: &str = "libvphevcodec.so"; + +const AMLENC_MAX_WIDTH: u32 = 1920; +const AMLENC_MAX_HEIGHT: u32 = 1080; +const AMLENC_MAX_FPS: u32 = 60; +const MIN_OUTPUT_BUFFER_SIZE: usize = 1024 * 1024; +const OUTPUT_STALL_TIMEOUT: Duration = Duration::from_secs(1); +const CODEC_ID_H264: c_int = 4; +const CODEC_ID_H265: c_int = 5; +const IMG_FMT_NV12: c_int = 1; +const FRAME_TYPE_AUTO: c_int = 1; +const FRAME_TYPE_IDR: c_int = 2; +const H264_NV12_FORMAT: c_int = 0; +const H265_NV12_FORMAT: c_int = 1; + +type AbiVersionFn = unsafe extern "C" fn() -> c_int; +type H264InitFn = unsafe extern "C" fn(c_int, c_int, c_int, c_int, c_int, c_int, c_int) -> c_long; +type H265InitFn = unsafe extern "C" fn(c_int, c_int, c_int, c_int, c_int, c_int) -> c_long; +type H264EncodeFn = + unsafe extern "C" fn(c_long, c_int, *mut c_uchar, c_int, *mut c_uchar, c_int) -> c_int; +type H265EncodeFn = + unsafe extern "C" fn(c_long, c_int, *mut c_uchar, c_uint, *mut c_uchar, c_int) -> c_int; +type DestroyFn = unsafe extern "C" fn(c_long) -> c_int; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AmlencCodec { + H264, + H265, +} + +impl AmlencCodec { + pub fn codec_name(self) -> &'static str { + match self { + Self::H264 => AMLENC_H264_CODEC_NAME, + Self::H265 => AMLENC_H265_CODEC_NAME, + } + } + + pub fn default_library(self) -> &'static str { + match self { + Self::H264 => AMLENC_H264_DEFAULT_LIBRARY, + Self::H265 => AMLENC_H265_DEFAULT_LIBRARY, + } + } + + pub fn library_env(self) -> &'static str { + match self { + Self::H264 => "ONE_KVM_AMLENC_H264_LIB", + Self::H265 => "ONE_KVM_AMLENC_H265_LIB", + } + } + + pub fn device_node(self) -> &'static str { + match self { + Self::H264 => "/dev/amvenc_avc", + Self::H265 => "/dev/HevcEnc", + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct AmlencConfig { + pub codec: AmlencCodec, + pub resolution: Resolution, + pub fps: u32, + pub bitrate_kbps: u32, + pub gop: u32, +} + +impl AmlencConfig { + pub fn validate(self) -> Result<()> { + let width = self.resolution.width; + let height = self.resolution.height; + if width == 0 + || height == 0 + || width > AMLENC_MAX_WIDTH + || height > AMLENC_MAX_HEIGHT + || width % 16 != 0 + || height % 2 != 0 + { + return Err(AppError::VideoError(format!( + "AMLENC requires NV12 with 16-aligned width, even height, and at most 1920x1080 (got {}x{})", + width, height + ))); + } + if !(1..=AMLENC_MAX_FPS).contains(&self.fps) { + return Err(AppError::VideoError(format!( + "AMLENC supports 1-60 fps (got {})", + self.fps + ))); + } + if self.bitrate_kbps == 0 || self.bitrate_kbps > (c_int::MAX as u32 / 1000) { + return Err(AppError::VideoError(format!( + "Invalid AMLENC bitrate: {} kbps", + self.bitrate_kbps + ))); + } + if self.gop > c_int::MAX as u32 { + return Err(AppError::VideoError("AMLENC GOP is too large".to_string())); + } + nv12_frame_size(self.resolution)?; + Ok(()) + } + + fn bitrate_bps(self) -> c_int { + (self.bitrate_kbps * 1000) as c_int + } + + fn vendor_gop(self) -> c_int { + match self.codec { + // GXM's H.264 microcode can time out on a later natural IDR for + // complex 1080p pictures. The pinned vendor library defines zero + // as an infinite GOP (one IDR when the instance is created). + AmlencCodec::H264 => 0, + AmlencCodec::H265 => self.gop as c_int, + } + } +} + +pub fn nv12_frame_size(resolution: Resolution) -> Result { + let pixels = (resolution.width as usize) + .checked_mul(resolution.height as usize) + .ok_or_else(|| AppError::VideoError("AMLENC NV12 frame size overflow".to_string()))?; + pixels + .checked_mul(3) + .map(|value| value / 2) + .ok_or_else(|| AppError::VideoError("AMLENC NV12 frame size overflow".to_string())) +} + +fn validate_abi_version(version: c_int, path: &Path) -> Result<()> { + if version != AMLENC_ABI_VERSION { + return Err(AppError::VideoError(format!( + "AMLENC library {} has ABI {}, expected ABI v{}; apply the one-kvm-amlenc-abi-v1.patch from the standalone amlenc repository", + path.display(), + version, + AMLENC_ABI_VERSION + ))); + } + Ok(()) +} + +struct H264Api { + _library: Library, + init: H264InitFn, + encode: H264EncodeFn, + destroy: DestroyFn, +} + +struct H265Api { + _library: Library, + init: H265InitFn, + encode: H265EncodeFn, + destroy: DestroyFn, +} + +enum AmlencApi { + H264(H264Api), + H265(H265Api), +} + +unsafe fn required_symbol(library: &Library, name: &[u8], path: &Path) -> Result { + // SAFETY: the caller supplies the signature from the fixed upstream headers. + unsafe { library.get::(name) } + .map(|symbol| *symbol) + .map_err(|error| { + AppError::VideoError(format!( + "AMLENC library {} is missing {}: {}", + path.display(), + String::from_utf8_lossy(name).trim_end_matches('\0'), + error + )) + }) +} + +impl AmlencApi { + fn load(codec: AmlencCodec, path: &Path) -> Result { + // SAFETY: all calls are made through signatures checked against the pinned headers, + // and the Library remains owned by the API object for the lifetime of the pointers. + let library = unsafe { Library::new(path) }.map_err(|error| { + AppError::VideoError(format!( + "Failed to load AMLENC {} library {}: {}", + codec.codec_name(), + path.display(), + error + )) + })?; + let abi_version: AbiVersionFn = + unsafe { required_symbol(&library, b"one_kvm_amlenc_abi_version\0", path)? }; + // SAFETY: the ABI marker has no arguments or side effects. + validate_abi_version(unsafe { abi_version() }, path)?; + + Ok(match codec { + AmlencCodec::H264 => { + let init: H264InitFn = + unsafe { required_symbol(&library, b"vl_video_encoder_init\0", path)? }; + let encode: H264EncodeFn = + unsafe { required_symbol(&library, b"vl_video_encoder_encode\0", path)? }; + let destroy: DestroyFn = + unsafe { required_symbol(&library, b"vl_video_encoder_destory\0", path)? }; + Self::H264(H264Api { + _library: library, + init, + encode, + destroy, + }) + } + AmlencCodec::H265 => { + let init: H265InitFn = + unsafe { required_symbol(&library, b"vl_video_encoder_init\0", path)? }; + let encode: H265EncodeFn = + unsafe { required_symbol(&library, b"vl_video_encoder_encode\0", path)? }; + let destroy: DestroyFn = + unsafe { required_symbol(&library, b"vl_video_encoder_destory\0", path)? }; + Self::H265(H265Api { + _library: library, + init, + encode, + destroy, + }) + } + }) + } + + unsafe fn init(&self, config: AmlencConfig) -> c_long { + let width = config.resolution.width as c_int; + let height = config.resolution.height as c_int; + match self { + Self::H264(api) => unsafe { + (api.init)( + CODEC_ID_H264, + width, + height, + config.fps as c_int, + config.bitrate_bps(), + config.vendor_gop(), + IMG_FMT_NV12, + ) + }, + Self::H265(api) => unsafe { + (api.init)( + CODEC_ID_H265, + width, + height, + config.fps as c_int, + config.bitrate_bps(), + config.gop as c_int, + ) + }, + } + } + + unsafe fn encode( + &self, + handle: c_long, + frame_type: c_int, + input: *mut c_uchar, + output: *mut c_uchar, + output_len: usize, + ) -> c_int { + match self { + // H.264's fourth argument is documented as input length, but the pinned + // implementation uses it exclusively as output capacity. + Self::H264(api) => unsafe { + (api.encode)( + handle, + frame_type, + input, + output_len as c_int, + output, + H264_NV12_FORMAT, + ) + }, + Self::H265(api) => unsafe { + (api.encode)( + handle, + frame_type, + input, + output_len as c_uint, + output, + H265_NV12_FORMAT, + ) + }, + } + } + + unsafe fn destroy(&self, handle: c_long) { + match self { + Self::H264(api) => { + unsafe { (api.destroy)(handle) }; + } + Self::H265(api) => { + unsafe { (api.destroy)(handle) }; + } + } + } +} + +static AMLENC_INSTANCE_ACTIVE: AtomicBool = AtomicBool::new(false); + +struct ExclusiveInstance; + +impl ExclusiveInstance { + fn acquire() -> Result { + AMLENC_INSTANCE_ACTIVE + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .map_err(|_| { + AppError::VideoError( + "AMLENC hardware is already in use by another encoder or self-check" + .to_string(), + ) + })?; + Ok(Self) + } +} + +impl Drop for ExclusiveInstance { + fn drop(&mut self) { + AMLENC_INSTANCE_ACTIVE.store(false, Ordering::Release); + } +} + +pub struct AmlencEncoder { + api: AmlencApi, + handle: c_long, + config: AmlencConfig, + output: Vec, + force_keyframe: bool, + rebuild_before_next_frame: bool, + expect_parameterized_keyframe: bool, + last_output: Instant, + _exclusive: ExclusiveInstance, +} + +impl AmlencEncoder { + pub fn new(config: AmlencConfig) -> Result { + let path = env::var_os(config.codec.library_env()) + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from(config.codec.default_library())); + Self::with_library(config, path) + } + + pub fn with_library(config: AmlencConfig, path: impl AsRef) -> Result { + config.validate()?; + let exclusive = ExclusiveInstance::acquire()?; + let path = PathBuf::from(path.as_ref()); + let api = AmlencApi::load(config.codec, &path)?; + let frame_size = nv12_frame_size(config.resolution)?; + let output = vec![0; frame_size.max(MIN_OUTPUT_BUFFER_SIZE)]; + let mut encoder = Self { + api, + handle: 0, + config, + output, + force_keyframe: false, + rebuild_before_next_frame: false, + expect_parameterized_keyframe: true, + last_output: Instant::now(), + _exclusive: exclusive, + }; + encoder.create_handle()?; + Ok(encoder) + } + + pub fn codec_name(&self) -> &'static str { + self.config.codec.codec_name() + } + + pub fn config(&self) -> AmlencConfig { + self.config + } + + fn create_handle(&mut self) -> Result<()> { + debug!( + "Creating {} at {}x{} {} fps {} kbps", + self.codec_name(), + self.config.resolution.width, + self.config.resolution.height, + self.config.fps, + self.config.bitrate_kbps + ); + // SAFETY: config validation guarantees values accepted by ABI v1. + self.handle = unsafe { self.api.init(self.config) }; + if self.handle <= 0 { + return Err(AppError::VideoError(format!( + "AMLENC {} initialization failed; check {}, firmware, CMA, and device permissions", + self.codec_name(), + self.config.codec.device_node() + ))); + } + // The first H.264 picture is naturally an IDR. Never pass the + // in-place FORCE_IDR command to the GXM H.264 microcode: later IDRs can + // wedge it. H.265 does not share that observed defect and retains its + // ABI-v1 forced-IRAP behavior. + self.force_keyframe = self.config.codec == AmlencCodec::H265; + self.rebuild_before_next_frame = false; + self.expect_parameterized_keyframe = true; + self.last_output = Instant::now(); + Ok(()) + } + + fn destroy_handle(&mut self) { + if self.handle > 0 { + // SAFETY: the handle was returned by this API instance and is destroyed once. + unsafe { self.api.destroy(self.handle) }; + self.handle = 0; + } + } + + fn rebuild(&mut self, reason: &str) -> Result<()> { + warn!("Rebuilding {} encoder: {}", self.codec_name(), reason); + self.destroy_handle(); + self.create_handle() + } + + pub fn request_keyframe(&mut self) { + if self.config.codec == AmlencCodec::H264 { + // A fresh encoder reliably emits SPS/PPS + IDR on its first AUTO + // frame. Coalesce repeated client requests while a rebuild or + // fresh first frame is already pending. + if !self.expect_parameterized_keyframe { + self.rebuild_before_next_frame = true; + } + } else { + self.force_keyframe = true; + } + } + + pub fn set_bitrate(&mut self, bitrate_kbps: u32) -> Result<()> { + let mut updated = self.config; + updated.bitrate_kbps = bitrate_kbps; + updated.validate()?; + self.config = updated; + self.rebuild("bitrate changed") + } + + pub fn encode_raw(&mut self, data: &[u8]) -> Result> { + let expected = nv12_frame_size(self.config.resolution)?; + if data.len() != expected { + return Err(AppError::VideoError(format!( + "AMLENC requires contiguous NV12 data of exactly {} bytes (got {})", + expected, + data.len() + ))); + } + + if self.rebuild_before_next_frame { + self.rebuild("H.264 keyframe requested")?; + } + + match self.encode_once(data) { + Ok(frame) => Ok(frame), + Err(first_error) => { + self.rebuild(&format!("vendor encode call failed: {first_error}"))?; + self.encode_once(data).map_err(|retry_error| { + AppError::VideoError(format!( + "AMLENC encode failed after one rebuild: {}; retry: {}", + first_error, retry_error + )) + }) + } + } + } + + fn encode_once(&mut self, data: &[u8]) -> Result> { + if self.handle <= 0 { + return Err(AppError::VideoError( + "AMLENC handle is not initialized".to_string(), + )); + } + let forced = self.force_keyframe; + let require_parameterized_keyframe = self.expect_parameterized_keyframe || forced; + let frame_type = if forced { + FRAME_TYPE_IDR + } else { + FRAME_TYPE_AUTO + }; + // The vendor API takes a mutable pointer but does not modify VMALLOC input. + // SAFETY: input/output live for the call, capacities are ABI-sized and the + // output length is validated before any slice is formed. + let length = unsafe { + self.api.encode( + self.handle, + frame_type, + data.as_ptr() as *mut c_uchar, + self.output.as_mut_ptr(), + self.output.len(), + ) + }; + if length < 0 { + return Err(AppError::VideoError(format!( + "{} vendor library returned {}", + self.codec_name(), + length + ))); + } + // A keyframe request applies to one submitted frame. Repeating IDR on + // every zero-output call can trap the S912 driver in its light-reset + // loop; WebRTC will issue another request if this attempt was skipped. + if forced { + self.force_keyframe = false; + } + let length = length as usize; + if length > self.output.len() { + return Err(AppError::VideoError(format!( + "{} returned oversized output {} > {}", + self.codec_name(), + length, + self.output.len() + ))); + } + if length == 0 { + if forced { + return Err(AppError::VideoError(format!( + "{} produced no output for a forced keyframe", + self.codec_name() + ))); + } + // The vendor ABI uses zero for rate-control skips and recoverable + // hardware timeouts. Do not rebuild for a few skipped frames, but + // recover if the vendor stops producing output altogether. + if self.last_output.elapsed() >= OUTPUT_STALL_TIMEOUT { + self.rebuild("no encoded output for one second")?; + } + return Ok(None); + } + + let encoded = &self.output[..length]; + let nal_summary = inspect_annex_b(self.config.codec, encoded); + let keyframe = nal_summary.keyframe; + if require_parameterized_keyframe + && (!keyframe || !nal_summary.has_parameter_sets(self.config.codec)) + { + return Err(AppError::VideoError(format!( + "{} fresh/forced keyframe did not contain an IRAP/IDR and complete parameter sets", + self.codec_name() + ))); + } + self.force_keyframe = false; + self.expect_parameterized_keyframe = false; + self.last_output = Instant::now(); + Ok(Some((Bytes::copy_from_slice(encoded), keyframe))) + } +} + +impl Drop for AmlencEncoder { + fn drop(&mut self) { + self.destroy_handle(); + } +} + +#[derive(Default)] +struct AnnexBNalSummary { + keyframe: bool, + vps: bool, + sps: bool, + pps: bool, +} + +impl AnnexBNalSummary { + fn has_parameter_sets(&self, codec: AmlencCodec) -> bool { + match codec { + AmlencCodec::H264 => self.sps && self.pps, + AmlencCodec::H265 => self.vps && self.sps && self.pps, + } + } +} + +fn inspect_annex_b(codec: AmlencCodec, data: &[u8]) -> AnnexBNalSummary { + let mut summary = AnnexBNalSummary::default(); + let mut index = 0; + while index + 3 <= data.len() { + let start_len = if index + 4 <= data.len() && data[index..index + 4] == [0, 0, 0, 1] { + 4 + } else if data[index..index + 3] == [0, 0, 1] { + 3 + } else { + index += 1; + continue; + }; + let nal = index + start_len; + if nal >= data.len() { + break; + } + let nal_type = match codec { + AmlencCodec::H264 => data[nal] & 0x1f, + AmlencCodec::H265 => (data[nal] >> 1) & 0x3f, + }; + match codec { + AmlencCodec::H264 => match nal_type { + 5 => summary.keyframe = true, + 7 => summary.sps = true, + 8 => summary.pps = true, + _ => {} + }, + AmlencCodec::H265 => match nal_type { + 16..=23 => summary.keyframe = true, + 32 => summary.vps = true, + 33 => summary.sps = true, + 34 => summary.pps = true, + _ => {} + }, + } + index = nal + 1; + } + summary +} + +pub fn is_keyframe(codec: AmlencCodec, data: &[u8]) -> bool { + inspect_annex_b(codec, data).keyframe +} + +pub fn has_parameter_sets(codec: AmlencCodec, data: &[u8]) -> bool { + inspect_annex_b(codec, data).has_parameter_sets(codec) +} + +#[cfg_attr( + not(any(test, all(target_os = "linux", target_arch = "aarch64"))), + allow(dead_code) +)] +fn is_s912_gxm_compatible(compatible: &[u8]) -> bool { + let compatible = String::from_utf8_lossy(compatible).to_ascii_lowercase(); + compatible.contains("amlogic,gxm") + || compatible.contains("amlogic, gxm") + || compatible.contains("amlogic,meson-gxm") + || compatible.contains("amlogic,s912") +} + +pub fn system_is_s912_gxm() -> Result { + #[cfg(all(target_os = "linux", target_arch = "aarch64"))] + { + let compatible = std::fs::read("/proc/device-tree/compatible").map_err(|error| { + AppError::VideoError(format!( + "Cannot read /proc/device-tree/compatible for AMLENC detection: {}", + error + )) + })?; + return Ok(is_s912_gxm_compatible(&compatible)); + } + #[cfg(not(all(target_os = "linux", target_arch = "aarch64")))] + Ok(false) +} + +/// Perform the destructive part of backend detection: initialize and encode one +/// 640x480 NV12 frame. The caller must first check SoC compatibility and node. +pub fn smoke_test(codec: AmlencCodec) -> Result<()> { + let resolution = Resolution::new(640, 480); + let config = AmlencConfig { + codec, + resolution, + fps: 30, + bitrate_kbps: 1_000, + gop: 30, + }; + let mut encoder = AmlencEncoder::new(config)?; + let mut frame = vec![0x80; nv12_frame_size(resolution)?]; + frame[..(resolution.width * resolution.height) as usize].fill(0x10); + for _ in 0..3 { + if encoder.encode_raw(&frame)?.is_some() { + return Ok(()); + } + } + Err(AppError::VideoError(format!( + "{} produced no output during the 640x480 probe", + codec.codec_name() + ))) +} + +#[cfg(test)] +mod tests { + use super::*; + #[cfg(unix)] + use std::process::Command; + #[cfg(unix)] + use std::sync::Mutex; + + #[cfg(unix)] + static TEST_INSTANCE_MUTEX: Mutex<()> = Mutex::new(()); + + #[cfg(unix)] + const H264_FIXTURE: &str = r#" + static int values[16]; + static int mode; + static int fail_pending; + int one_kvm_amlenc_abi_version(void) { return 1; } + long vl_video_encoder_init(int codec, int width, int height, int fps, + int bitrate, int gop, int image_format) { + values[0]++; values[1] = codec; values[2] = width; values[3] = height; + values[4] = fps; values[5] = bitrate; values[6] = gop; + values[7] = image_format; return 1; + } + int vl_video_encoder_encode(long handle, int frame_type, unsigned char *in, + int in_size, unsigned char *out, int format) { + (void)handle; (void)in; values[8]++; values[9] = frame_type; + values[10] = in_size; values[11] = format; + if (fail_pending) { fail_pending = 0; return -9; } + if (mode == 2) return 0; + if (mode == 3) return 2000000; + { unsigned char data[] = {0,0,1,0x67,0,0,1,0x68,0,0,1,0x65}; + for (unsigned long i = 0; i < sizeof(data); i++) out[i] = data[i]; + return sizeof(data); } + } + int vl_video_encoder_destory(long handle) { (void)handle; values[12]++; return 1; } + int test_get(int index) { return values[index]; } + void test_set_mode(int value) { mode = value; } + void test_fail_once(void) { fail_pending = 1; } + "#; + + #[cfg(unix)] + const H265_FIXTURE: &str = r#" + static int values[16]; + static int mode; + int one_kvm_amlenc_abi_version(void) { return 1; } + long vl_video_encoder_init(int codec, int width, int height, int fps, + int bitrate, int gop) { + values[0]++; values[1] = codec; values[2] = width; values[3] = height; + values[4] = fps; values[5] = bitrate; values[6] = gop; return 1; + } + int vl_video_encoder_encode(long handle, int frame_type, unsigned char *in, + unsigned int output_len, unsigned char *out, int format) { + (void)handle; (void)in; values[8]++; values[9] = frame_type; + values[10] = output_len; values[11] = format; + if (mode == 3) return output_len + 1; + { unsigned char data[] = {0,0,1,0x40,1,0,0,1,0x42,1,0,0,1,0x44,1, + 0,0,1,0x26,1}; + for (unsigned long i = 0; i < sizeof(data); i++) out[i] = data[i]; + return sizeof(data); } + } + int vl_video_encoder_destory(long handle) { (void)handle; values[12]++; return 1; } + int test_get(int index) { return values[index]; } + void test_set_mode(int value) { mode = value; } + "#; + + #[cfg(unix)] + fn build_fixture(directory: &Path, name: &str, source: &str) -> PathBuf { + let source_path = directory.join(format!("{name}.c")); + let library_path = directory.join(format!("lib{name}.so")); + std::fs::write(&source_path, source).unwrap(); + let status = Command::new("cc") + .args(["-shared", "-fPIC"]) + .arg(&source_path) + .arg("-o") + .arg(&library_path) + .status() + .unwrap(); + assert!(status.success()); + library_path + } + + #[test] + fn validates_geometry_fps_and_nv12_size() { + let valid = AmlencConfig { + codec: AmlencCodec::H264, + resolution: Resolution::new(1920, 1080), + fps: 60, + bitrate_kbps: 8_000, + gop: 60, + }; + assert!(valid.validate().is_ok()); + assert_eq!(nv12_frame_size(valid.resolution).unwrap(), 3_110_400); + + for invalid in [ + AmlencConfig { + resolution: Resolution::new(1919, 1080), + ..valid + }, + AmlencConfig { + resolution: Resolution::new(1920, 1079), + ..valid + }, + AmlencConfig { + resolution: Resolution::new(2560, 1440), + ..valid + }, + AmlencConfig { fps: 61, ..valid }, + ] { + assert!(invalid.validate().is_err()); + } + } + + #[test] + fn recognizes_vendor_and_mainline_gxm_compatibles() { + assert!(is_s912_gxm_compatible(b"amlogic, Gxm\0khadas,kvim2")); + assert!(is_s912_gxm_compatible( + b"amlogic,q200\0amlogic,s912\0amlogic,meson-gxm" + )); + assert!(!is_s912_gxm_compatible(b"rockchip,rk3588")); + } + + #[test] + fn validates_abi_marker() { + let path = Path::new("libvpcodec.so"); + assert!(validate_abi_version(AMLENC_ABI_VERSION, path).is_ok()); + assert!(validate_abi_version(0, path).is_err()); + } + + #[test] + fn parses_h264_idr_and_parameter_sets() { + let data = [0, 0, 0, 1, 0x67, 1, 0, 0, 1, 0x68, 2, 0, 0, 0, 1, 0x65, 3]; + assert!(is_keyframe(AmlencCodec::H264, &data)); + assert!(has_parameter_sets(AmlencCodec::H264, &data)); + assert!(!is_keyframe(AmlencCodec::H264, &[0, 0, 1, 0x41])); + } + + #[test] + fn parses_h265_irap_and_parameter_sets() { + let data = [ + 0, + 0, + 1, + 32 << 1, + 1, + 0, + 0, + 1, + 33 << 1, + 1, + 0, + 0, + 1, + 34 << 1, + 1, + 0, + 0, + 1, + 19 << 1, + 1, + ]; + assert!(is_keyframe(AmlencCodec::H265, &data)); + assert!(has_parameter_sets(AmlencCodec::H265, &data)); + assert!(!is_keyframe(AmlencCodec::H265, &[0, 0, 1, 1 << 1, 1])); + } + + #[test] + #[cfg(unix)] + fn loads_symbols_maps_both_abis_and_recovers() { + let _test_instance = TEST_INSTANCE_MUTEX.lock().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let h264_path = build_fixture(directory.path(), "amlenc_h264", H264_FIXTURE); + let h265_path = build_fixture(directory.path(), "amlenc_h265", H265_FIXTURE); + + type GetFn = unsafe extern "C" fn(c_int) -> c_int; + type SetModeFn = unsafe extern "C" fn(c_int); + type FailOnceFn = unsafe extern "C" fn(); + + // Keep this second dlopen alive so the fixture's counters remain available. + let h264_control = unsafe { Library::new(&h264_path) }.unwrap(); + let h264_get: GetFn = unsafe { *h264_control.get(b"test_get\0").unwrap() }; + let h264_set_mode: SetModeFn = unsafe { *h264_control.get(b"test_set_mode\0").unwrap() }; + let h264_fail_once: FailOnceFn = unsafe { *h264_control.get(b"test_fail_once\0").unwrap() }; + + let resolution = Resolution::new(640, 480); + let frame = vec![0x80; nv12_frame_size(resolution).unwrap()]; + { + let mut encoder = AmlencEncoder::with_library( + AmlencConfig { + codec: AmlencCodec::H264, + resolution, + fps: 60, + bitrate_kbps: 2_000, + gop: 60, + }, + &h264_path, + ) + .unwrap(); + assert!(encoder.encode_raw(&frame).unwrap().unwrap().1); + // SAFETY: indices and fixture signatures are fixed above. + unsafe { + assert_eq!(h264_get(1), CODEC_ID_H264); + assert_eq!(h264_get(4), 60); + assert_eq!(h264_get(5), 2_000_000); + assert_eq!(h264_get(6), 0); + assert_eq!(h264_get(7), IMG_FMT_NV12); + assert_eq!(h264_get(9), FRAME_TYPE_AUTO); + assert_eq!(h264_get(10), MIN_OUTPUT_BUFFER_SIZE as c_int); + assert_eq!(h264_get(11), H264_NV12_FORMAT); + + h264_fail_once(); + } + assert!(encoder.encode_raw(&frame).unwrap().is_some()); + unsafe { assert_eq!(h264_get(0), 2) }; + + unsafe { h264_set_mode(2) }; + encoder.request_keyframe(); + assert!(encoder.encode_raw(&frame).unwrap().is_none()); + unsafe { assert_eq!(h264_get(9), FRAME_TYPE_AUTO) }; + unsafe { assert_eq!(h264_get(0), 3) }; + assert!(encoder.encode_raw(&frame).unwrap().is_none()); + unsafe { assert_eq!(h264_get(9), FRAME_TYPE_AUTO) }; + assert!(encoder.encode_raw(&frame).unwrap().is_none()); + unsafe { assert_eq!(h264_get(0), 3) }; + + encoder.last_output = Instant::now() - OUTPUT_STALL_TIMEOUT; + assert!(encoder.encode_raw(&frame).unwrap().is_none()); + unsafe { assert_eq!(h264_get(0), 4) }; + + unsafe { h264_set_mode(0) }; + encoder.set_bitrate(3_000).unwrap(); + assert!(encoder.encode_raw(&frame).unwrap().unwrap().1); + unsafe { + assert_eq!(h264_get(5), 3_000_000); + assert_eq!(h264_get(9), FRAME_TYPE_AUTO); + assert_eq!(h264_get(0), 5); + } + } + + let h265_control = unsafe { Library::new(&h265_path) }.unwrap(); + let h265_get: GetFn = unsafe { *h265_control.get(b"test_get\0").unwrap() }; + let h265_set_mode: SetModeFn = unsafe { *h265_control.get(b"test_set_mode\0").unwrap() }; + { + let mut encoder = AmlencEncoder::with_library( + AmlencConfig { + codec: AmlencCodec::H265, + resolution, + fps: 30, + bitrate_kbps: 1_500, + gop: 30, + }, + &h265_path, + ) + .unwrap(); + assert!(encoder.encode_raw(&frame).unwrap().unwrap().1); + unsafe { + assert_eq!(h265_get(1), CODEC_ID_H265); + assert_eq!(h265_get(4), 30); + assert_eq!(h265_get(5), 1_500_000); + assert_eq!(h265_get(9), FRAME_TYPE_IDR); + assert_eq!(h265_get(10), MIN_OUTPUT_BUFFER_SIZE as c_int); + assert_eq!(h265_get(11), H265_NV12_FORMAT); + h265_set_mode(3); + } + let error = encoder.encode_raw(&frame).unwrap_err().to_string(); + assert!(error.contains("oversized output")); + } + } + + #[test] + #[cfg(unix)] + fn rejects_unpatched_library_without_abi_symbol() { + let _test_instance = TEST_INSTANCE_MUTEX.lock().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let path = build_fixture( + directory.path(), + "unpatched_amlenc", + "long vl_video_encoder_init(void) { return 1; }", + ); + let error = AmlencEncoder::with_library( + AmlencConfig { + codec: AmlencCodec::H264, + resolution: Resolution::new(640, 480), + fps: 30, + bitrate_kbps: 1_000, + gop: 30, + }, + path, + ) + .err() + .expect("unpatched library must be rejected") + .to_string(); + assert!(error.contains("one_kvm_amlenc_abi_version")); + } +} diff --git a/src/video/codec/convert.rs b/src/video/codec/convert.rs index f362f064..305edd7e 100644 --- a/src/video/codec/convert.rs +++ b/src/video/codec/convert.rs @@ -558,27 +558,43 @@ impl MjpegToNv12Decoder { } pub fn decode(&mut self, input: &[u8]) -> Result<&[u8]> { + self.check_size(input)?; let width = self.resolution.width as i32; let height = self.resolution.height as i32; - if !self.size_checked { - let (src_width, src_height) = libyuv::mjpg_size(input).map_err(|e| { - AppError::VideoError(format!("libyuv MJPEG header read failed: {}", e)) - })?; - if src_width != width || src_height != height { - return Err(AppError::VideoError(format!( - "libyuv MJPEG size mismatch: {}x{} (expected {}x{})", - src_width, src_height, width, height - ))); - } - self.size_checked = true; - } - libyuv::mjpg_to_nv12(input, self.output_buffer.as_bytes_mut(), width, height) .map_err(|e| AppError::VideoError(format!("libyuv MJPEG->NV12 failed: {}", e)))?; Ok(self.output_buffer.as_bytes()) } + + /// Decode into caller-owned storage so capture and encoding can run on + /// separate threads without copying a full NV12 frame. + pub fn decode_into(&mut self, input: &[u8], output: &mut Vec) -> Result<()> { + self.check_size(input)?; + let width = self.resolution.width as i32; + let height = self.resolution.height as i32; + libyuv::mjpg_to_nv12_vec(input, output, width, height) + .map_err(|e| AppError::VideoError(format!("libyuv MJPEG->NV12 failed: {}", e))) + } + + fn check_size(&mut self, input: &[u8]) -> Result<()> { + if self.size_checked { + return Ok(()); + } + let width = self.resolution.width as i32; + let height = self.resolution.height as i32; + let (src_width, src_height) = libyuv::mjpg_size(input) + .map_err(|e| AppError::VideoError(format!("libyuv MJPEG header read failed: {}", e)))?; + if src_width != width || src_height != height { + return Err(AppError::VideoError(format!( + "libyuv MJPEG size mismatch: {}x{} (expected {}x{})", + src_width, src_height, width, height + ))); + } + self.size_checked = true; + Ok(()) + } } impl Nv12Converter { @@ -862,4 +878,34 @@ mod tests { let result = converter.convert(&yuyv).unwrap(); assert_eq!(result.len(), 24); // 4*4 + 2*2 + 2*2 = 24 bytes } + + #[test] + fn test_mjpeg_decode_into_reuses_output_allocation() { + let resolution = Resolution::new(16, 16); + let pixels = vec![0x80; 16 * 16 * 3]; + let image = turbojpeg::Image { + pixels: pixels.as_slice(), + width: 16, + pitch: 16 * 3, + height: 16, + format: turbojpeg::PixelFormat::RGB, + }; + let mut compressor = turbojpeg::Compressor::new().unwrap(); + compressor.set_quality(80).unwrap(); + compressor.set_subsamp(turbojpeg::Subsamp::Sub2x2).unwrap(); + let jpeg = compressor.compress_to_vec(image).unwrap(); + + let output_size = 16 * 16 * 3 / 2; + let mut output = Vec::with_capacity(output_size); + let allocation = output.as_ptr(); + let mut decoder = MjpegToNv12Decoder::new(resolution); + + decoder.decode_into(&jpeg, &mut output).unwrap(); + assert_eq!(output.len(), output_size); + assert_eq!(output.as_ptr(), allocation); + + decoder.decode_into(&jpeg, &mut output).unwrap(); + assert_eq!(output.len(), output_size); + assert_eq!(output.as_ptr(), allocation); + } } diff --git a/src/video/codec/h264.rs b/src/video/codec/h264.rs index 13d7a3a8..5ebc31e6 100644 --- a/src/video/codec/h264.rs +++ b/src/video/codec/h264.rs @@ -48,6 +48,8 @@ pub enum H264EncoderType { Rkmpp, /// V4L2 M2M (ARM generic) - requires hwcodec extension V4l2M2m, + /// Amlogic S912/GXM AMLENC + Amlogic, /// Software encoding (libx264/openh264) Software, /// No encoder available @@ -64,6 +66,7 @@ impl std::fmt::Display for H264EncoderType { H264EncoderType::Vaapi => write!(f, "VAAPI"), H264EncoderType::Rkmpp => write!(f, "RKMPP"), H264EncoderType::V4l2M2m => write!(f, "V4L2 M2M"), + H264EncoderType::Amlogic => write!(f, "AMLENC"), H264EncoderType::Software => write!(f, "Software"), H264EncoderType::None => write!(f, "None"), } @@ -80,6 +83,7 @@ impl From for H264EncoderType { EncoderBackend::Vaapi => H264EncoderType::Vaapi, EncoderBackend::Rkmpp => H264EncoderType::Rkmpp, EncoderBackend::V4l2m2m => H264EncoderType::V4l2M2m, + EncoderBackend::Amlogic => H264EncoderType::Amlogic, EncoderBackend::Software => H264EncoderType::Software, } } diff --git a/src/video/codec/h265.rs b/src/video/codec/h265.rs index 08f3dbed..dd5df5e7 100644 --- a/src/video/codec/h265.rs +++ b/src/video/codec/h265.rs @@ -45,6 +45,8 @@ pub enum H265EncoderType { Rkmpp, /// V4L2 M2M (ARM generic) V4l2M2m, + /// Amlogic S912/GXM AMLENC + Amlogic, /// Software encoder (libx265) Software, /// No encoder available @@ -61,6 +63,7 @@ impl std::fmt::Display for H265EncoderType { H265EncoderType::Vaapi => write!(f, "VAAPI"), H265EncoderType::Rkmpp => write!(f, "RKMPP"), H265EncoderType::V4l2M2m => write!(f, "V4L2 M2M"), + H265EncoderType::Amlogic => write!(f, "AMLENC"), H265EncoderType::Software => write!(f, "Software"), H265EncoderType::None => write!(f, "None"), } @@ -76,6 +79,7 @@ impl From for H265EncoderType { EncoderBackend::Vaapi => H265EncoderType::Vaapi, EncoderBackend::Rkmpp => H265EncoderType::Rkmpp, EncoderBackend::V4l2m2m => H265EncoderType::V4l2M2m, + EncoderBackend::Amlogic => H265EncoderType::Amlogic, EncoderBackend::Software => H265EncoderType::Software, } } diff --git a/src/video/codec/mod.rs b/src/video/codec/mod.rs index ba0ab378..04d8823f 100644 --- a/src/video/codec/mod.rs +++ b/src/video/codec/mod.rs @@ -3,6 +3,7 @@ use hwcodec::common::DataFormat; use hwcodec::ffmpeg_ram::CodecInfo; +pub mod amlenc; pub mod convert; pub mod h264; @@ -19,6 +20,7 @@ pub mod vp9; #[cfg(all(feature = "desktop", any(target_arch = "aarch64", target_arch = "arm")))] pub mod mjpeg_rkmpp; +pub use amlenc::{AmlencCodec, AmlencConfig, AmlencEncoder}; pub use convert::{MjpegToNv12Decoder, PixelConverter, Yuv420pBuffer}; pub use h264::{H264Config, H264Encoder, H264EncoderType, H264InputFormat}; pub use h265::{H265Config, H265Encoder, H265EncoderType, H265InputFormat}; diff --git a/src/video/codec/registry.rs b/src/video/codec/registry.rs index d12bb663..3baa50c8 100644 --- a/src/video/codec/registry.rs +++ b/src/video/codec/registry.rs @@ -10,11 +10,17 @@ use std::sync::OnceLock; use std::time::Duration; use tracing::{debug, info, warn}; +use super::amlenc::{self, AmlencCodec, AMLENC_H264_CODEC_NAME, AMLENC_H265_CODEC_NAME}; + use hwcodec::common::{DataFormat, Quality, RateControl}; use hwcodec::ffmpeg::{resolve_pixel_format, AVPixelFormat}; use hwcodec::ffmpeg_ram::encode::{EncodeContext, Encoder as HwEncoder}; use hwcodec::ffmpeg_ram::CodecInfo; +// Keep native AMLENC behind the highest-priority desktop GPU backends while +// ensuring it is selected before hwcodec's software priority (3). +const AMLENC_PRIORITY: i32 = 2; + /// Video encoder format type #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum VideoEncoderType { @@ -96,6 +102,8 @@ pub enum EncoderBackend { Rkmpp, /// V4L2 Memory-to-Memory (ARM) V4l2m2m, + /// Amlogic S912/GXM vendor AMLENC + Amlogic, /// Software encoding (libx264, libx265, libvpx) Software, } @@ -115,6 +123,8 @@ impl EncoderBackend { EncoderBackend::Rkmpp } else if name.contains("v4l2m2m") { EncoderBackend::V4l2m2m + } else if name.contains("amlenc") { + EncoderBackend::Amlogic } else { EncoderBackend::Software } @@ -134,6 +144,7 @@ impl EncoderBackend { EncoderBackend::Amf => "AMF", EncoderBackend::Rkmpp => "RKMPP", EncoderBackend::V4l2m2m => "V4L2 M2M", + EncoderBackend::Amlogic => "AMLENC", EncoderBackend::Software => "Software", } } @@ -148,6 +159,7 @@ impl EncoderBackend { "amf" => Some(EncoderBackend::Amf), "rkmpp" => Some(EncoderBackend::Rkmpp), "v4l2m2m" | "v4l2" => Some(EncoderBackend::V4l2m2m), + "amlogic" | "amlenc" => Some(EncoderBackend::Amlogic), "software" | "cpu" => Some(EncoderBackend::Software), _ => None, } @@ -274,6 +286,79 @@ impl EncoderRegistry { } } + fn detect_amlenc(&mut self) { + match amlenc::system_is_s912_gxm() { + Ok(true) => {} + Ok(false) => { + debug!("AMLENC skipped: host is not Linux/aarch64 S912/GXM"); + return; + } + Err(error) => { + warn!("AMLENC skipped: {}", error); + return; + } + } + + self.detect_amlenc_candidates( + true, + |codec| std::path::Path::new(codec.device_node()).exists(), + amlenc::smoke_test, + ); + } + + fn detect_amlenc_candidates( + &mut self, + compatible: bool, + mut node_exists: NodeExists, + mut smoke_test: SmokeTest, + ) where + NodeExists: FnMut(AmlencCodec) -> bool, + SmokeTest: FnMut(AmlencCodec) -> crate::error::Result<()>, + { + if !compatible { + return; + } + + for (codec, format, codec_name) in [ + ( + AmlencCodec::H264, + VideoEncoderType::H264, + AMLENC_H264_CODEC_NAME, + ), + ( + AmlencCodec::H265, + VideoEncoderType::H265, + AMLENC_H265_CODEC_NAME, + ), + ] { + let node = codec.device_node(); + if !node_exists(codec) { + warn!( + "AMLENC {} unavailable: device node {} is missing", + format, node + ); + continue; + } + + match smoke_test(codec) { + Ok(()) => { + self.encoders + .entry(format) + .or_default() + .push(AvailableEncoder { + format, + codec_name: codec_name.to_string(), + backend: EncoderBackend::Amlogic, + priority: AMLENC_PRIORITY, + is_hardware: true, + }); + info!("Registered native AMLENC encoder: {}", codec_name); + } + Err(error) => warn!("AMLENC {} unavailable ({}): {}", format, node, error), + } + } + } + /// Get the global registry instance /// /// The registry is initialized lazily on first access with 1280x720 detection. @@ -341,6 +426,8 @@ impl EncoderRegistry { } } + self.detect_amlenc(); + // Sort encoders by priority (lower is better) for encoders in self.encoders.values_mut() { encoders.sort_by_key(|e| e.priority); @@ -537,6 +624,14 @@ mod tests { EncoderBackend::from_codec_name("libx264"), EncoderBackend::Software ); + assert_eq!( + EncoderBackend::from_codec_name("h264_amlenc"), + EncoderBackend::Amlogic + ); + assert_eq!( + EncoderBackend::from_str("amlogic"), + Some(EncoderBackend::Amlogic) + ); } #[test] @@ -561,4 +656,65 @@ mod tests { println!("Available formats: {:?}", registry.available_formats(false)); println!("Selectable formats: {:?}", registry.selectable_formats()); } + + #[test] + fn test_amlenc_registration_prerequisite_matrix() { + let ok = |_codec| Ok(()); + + let mut incompatible = EncoderRegistry::new(); + incompatible.detect_amlenc_candidates(false, |_| true, ok); + assert!(incompatible.encoders.is_empty()); + + let mut no_nodes = EncoderRegistry::new(); + no_nodes.detect_amlenc_candidates(true, |_| false, ok); + assert!(no_nodes.encoders.is_empty()); + + for reason in ["library missing", "ABI marker missing"] { + let mut rejected = EncoderRegistry::new(); + rejected.detect_amlenc_candidates( + true, + |_| true, + |_| Err(crate::error::AppError::VideoError(reason.to_string())), + ); + assert!(rejected.encoders.is_empty()); + } + + let mut h264_only = EncoderRegistry::new(); + h264_only.detect_amlenc_candidates(true, |codec| codec == AmlencCodec::H264, ok); + assert!(h264_only + .encoder_with_backend(VideoEncoderType::H264, EncoderBackend::Amlogic) + .is_some()); + assert!(h264_only + .encoder_with_backend(VideoEncoderType::H265, EncoderBackend::Amlogic) + .is_none()); + + let mut both = EncoderRegistry::new(); + both.detect_amlenc_candidates(true, |_| true, ok); + assert!(both + .encoder_with_backend(VideoEncoderType::H264, EncoderBackend::Amlogic) + .is_some()); + assert!(both + .encoder_with_backend(VideoEncoderType::H265, EncoderBackend::Amlogic) + .is_some()); + + both.encoders + .entry(VideoEncoderType::H264) + .or_default() + .push(AvailableEncoder { + format: VideoEncoderType::H264, + codec_name: "libx264".to_string(), + backend: EncoderBackend::Software, + priority: 3, + is_hardware: false, + }); + both.encoders + .get_mut(&VideoEncoderType::H264) + .unwrap() + .sort_by_key(|encoder| encoder.priority); + assert_eq!( + both.best_available_encoder(VideoEncoderType::H264) + .map(|encoder| encoder.backend), + Some(EncoderBackend::Amlogic) + ); + } } diff --git a/src/video/codec/self_check.rs b/src/video/codec/self_check.rs index be6eed32..8325e95a 100644 --- a/src/video/codec/self_check.rs +++ b/src/video/codec/self_check.rs @@ -3,8 +3,8 @@ use std::sync::mpsc; use std::time::{Duration, Instant}; use super::{ - EncoderRegistry, H264Config, H264Encoder, H265Config, H265Encoder, VP8Config, VP8Encoder, - VP9Config, VP9Encoder, VideoEncoderType, + AmlencCodec, AmlencConfig, AmlencEncoder, EncoderRegistry, H264Config, H264Encoder, H265Config, + H265Encoder, VP8Config, VP8Encoder, VP9Config, VP9Encoder, VideoEncoderType, }; use crate::error::{AppError, Result}; use crate::video::format::{PixelFormat, Resolution}; @@ -226,6 +226,9 @@ fn run_smoke_test( resolution: Resolution, codec_name_ffmpeg: &str, ) -> Result<()> { + if codec_name_ffmpeg.contains("amlenc") { + return run_amlenc_smoke_test(codec, resolution); + } match codec { VideoEncoderType::H264 => run_h264_smoke_test(resolution, codec_name_ffmpeg), VideoEncoderType::H265 => run_h265_smoke_test(resolution, codec_name_ffmpeg), @@ -234,6 +237,37 @@ fn run_smoke_test( } } +fn run_amlenc_smoke_test(codec: VideoEncoderType, resolution: Resolution) -> Result<()> { + let amlenc_codec = match codec { + VideoEncoderType::H264 => AmlencCodec::H264, + VideoEncoderType::H265 => AmlencCodec::H265, + _ => { + return Err(AppError::VideoError( + "AMLENC only supports H.264 and H.265".to_string(), + )) + } + }; + let mut encoder = AmlencEncoder::new(AmlencConfig { + codec: amlenc_codec, + resolution, + fps: 30, + bitrate_kbps: bitrate_kbps_for_resolution(resolution), + gop: 30, + })?; + let frame_len = PixelFormat::Nv12.frame_size(resolution).ok_or_else(|| { + AppError::VideoError("Cannot calculate AMLENC NV12 self-check size".to_string()) + })?; + let frame = build_nv12_test_frame(resolution, frame_len); + for _ in 0..SELF_CHECK_FRAME_ATTEMPTS { + if encoder.encode_raw(&frame)?.is_some() { + return Ok(()); + } + } + Err(AppError::VideoError( + "AMLENC produced no output after multiple frames".to_string(), + )) +} + fn run_h264_smoke_test(resolution: Resolution, codec_name_ffmpeg: &str) -> Result<()> { let mut encoder = H264Encoder::with_codec( H264Config::low_latency(resolution, bitrate_kbps_for_resolution(resolution)), diff --git a/src/video/pipeline/encoder_state.rs b/src/video/pipeline/encoder_state.rs index 158edf1c..dafbdc53 100644 --- a/src/video/pipeline/encoder_state.rs +++ b/src/video/pipeline/encoder_state.rs @@ -1,4 +1,5 @@ use crate::error::{AppError, Result}; +use crate::video::codec::amlenc::{AmlencCodec, AmlencConfig, AmlencEncoder}; use crate::video::codec::convert::{MjpegToNv12Decoder, Nv12Converter, PixelConverter}; use crate::video::codec::h264::{H264Config, H264Encoder, H264InputFormat}; use crate::video::codec::h265::{H265Config, H265Encoder, H265InputFormat}; @@ -116,6 +117,47 @@ impl VideoEncoderTrait for H265EncoderWrapper { } } +struct AmlencEncoderWrapper(AmlencEncoder); + +impl VideoEncoderTrait for AmlencEncoderWrapper { + fn encode_raw(&mut self, data: &[u8], _pts_ms: i64) -> Result> { + Ok(match self.0.encode_raw(data)? { + Some((data, keyframe)) => vec![EncodedFrame { + data, + key: i32::from(keyframe), + }], + None => Vec::new(), + }) + } + + fn set_bitrate(&mut self, bitrate_kbps: u32) -> Result<()> { + self.0.set_bitrate(bitrate_kbps) + } + + fn codec_name(&self) -> &str { + self.0.codec_name() + } + + fn request_keyframe(&mut self) { + self.0.request_keyframe() + } +} + +fn create_amlenc_encoder( + config: &SharedVideoPipelineConfig, + codec: AmlencCodec, +) -> Result> { + let encoder = AmlencEncoder::new(AmlencConfig { + codec, + resolution: config.resolution, + fps: config.fps, + bitrate_kbps: config.bitrate_kbps(), + gop: config.gop_size(), + })?; + info!("Created native AMLENC encoder: {}", encoder.codec_name()); + Ok(Box::new(AmlencEncoderWrapper(encoder))) +} + struct VP8EncoderWrapper(VP8Encoder); impl VideoEncoderTrait for VP8EncoderWrapper { @@ -189,6 +231,26 @@ fn create_mjpeg_decoder(resolution: Resolution) -> Result<(MjpegDecoderKind, Pix Ok((libyuv_mjpeg_decoder(resolution), PixelFormat::Nv12)) } +/// AMLENC and libjpeg-turbo use independent CPU/hardware resources. Decode +/// MJPEG in the capture worker so encoding the previous NV12 frame can overlap +/// with decoding the next frame. +pub(super) fn should_parallel_decode_mjpeg(config: &SharedVideoPipelineConfig) -> bool { + if !config.input_format.is_compressed() + || !matches!( + config.output_codec, + VideoEncoderType::H264 | VideoEncoderType::H265 + ) + { + return false; + } + let registry = EncoderRegistry::global(); + let selected = match config.encoder_backend { + Some(backend) => registry.encoder_with_backend(config.output_codec, backend), + None => registry.best_available_encoder(config.output_codec), + }; + selected.is_some_and(|encoder| encoder.backend == EncoderBackend::Amlogic) +} + pub(super) fn build_encoder_state( config: &SharedVideoPipelineConfig, ) -> Result { @@ -210,9 +272,14 @@ pub(super) fn build_encoder_state( let is_rkmpp_available = registry .encoder_with_backend(VideoEncoderType::H264, EncoderBackend::Rkmpp) .is_some(); - let use_yuyv_direct = - is_rkmpp_available && !needs_mjpeg_decode && config.input_format == PixelFormat::Yuyv; + let rkmpp_is_allowed = + config.encoder_backend.is_none() || config.encoder_backend == Some(EncoderBackend::Rkmpp); + let use_yuyv_direct = is_rkmpp_available + && rkmpp_is_allowed + && !needs_mjpeg_decode + && config.input_format == PixelFormat::Yuyv; let use_rkmpp_direct = is_rkmpp_available + && rkmpp_is_allowed && !needs_mjpeg_decode && matches!( config.input_format, @@ -348,70 +415,80 @@ pub(super) fn build_encoder_state( let encoder: Box = match config.output_codec { VideoEncoderType::H264 => { let codec_name = selected_codec_name.clone(); - let direct_input_format = h264_direct_input_format(&codec_name, pipeline_input_format); - let input_format = direct_input_format.unwrap_or_else(|| { - if codec_name.contains("libx264") { - H264InputFormat::Yuv420p - } else { - H264InputFormat::Nv12 + if codec_name == crate::video::codec::amlenc::AMLENC_H264_CODEC_NAME { + create_amlenc_encoder(config, AmlencCodec::H264)? + } else { + let direct_input_format = + h264_direct_input_format(&codec_name, pipeline_input_format); + let input_format = direct_input_format.unwrap_or_else(|| { + if codec_name.contains("libx264") { + H264InputFormat::Yuv420p + } else { + H264InputFormat::Nv12 + } + }); + + if use_rkmpp_direct { + info!( + "Creating H264 encoder with RKMPP backend for {} direct input (codec: {})", + config.input_format, codec_name + ); + } else if let Some(ref backend) = config.encoder_backend { + info!( + "Creating H264 encoder with backend {:?} (codec: {})", + backend, codec_name + ); } - }); - if use_rkmpp_direct { - info!( - "Creating H264 encoder with RKMPP backend for {} direct input (codec: {})", - config.input_format, codec_name - ); - } else if let Some(ref backend) = config.encoder_backend { - info!( - "Creating H264 encoder with backend {:?} (codec: {})", - backend, codec_name - ); + create_h264_encoder(config, input_format, &codec_name)? } - - create_h264_encoder(config, input_format, &codec_name)? } VideoEncoderType::H265 => { let codec_name = selected_codec_name.clone(); - let direct_input_format = h265_direct_input_format(&codec_name, pipeline_input_format); - let input_format = direct_input_format.unwrap_or_else(|| { - if codec_name.contains("libx265") { - H265InputFormat::Yuv420p - } else { - H265InputFormat::Nv12 + if codec_name == crate::video::codec::amlenc::AMLENC_H265_CODEC_NAME { + create_amlenc_encoder(config, AmlencCodec::H265)? + } else { + let direct_input_format = + h265_direct_input_format(&codec_name, pipeline_input_format); + let input_format = direct_input_format.unwrap_or_else(|| { + if codec_name.contains("libx265") { + H265InputFormat::Yuv420p + } else { + H265InputFormat::Nv12 + } + }); + + if use_rkmpp_direct { + info!( + "Creating H265 encoder with RKMPP backend for {} direct input (codec: {})", + config.input_format, codec_name + ); + } else if let Some(ref backend) = config.encoder_backend { + info!( + "Creating H265 encoder with backend {:?} (codec: {})", + backend, codec_name + ); } - }); - if use_rkmpp_direct { - info!( - "Creating H265 encoder with RKMPP backend for {} direct input (codec: {})", - config.input_format, codec_name - ); - } else if let Some(ref backend) = config.encoder_backend { - info!( - "Creating H265 encoder with backend {:?} (codec: {})", - backend, codec_name - ); - } - - let encoder = H265Encoder::with_codec( - H265Config { - base: EncoderConfig { - resolution: config.resolution, - input_format: config.input_format, - quality: config.bitrate_kbps(), - fps: config.fps, + let encoder = H265Encoder::with_codec( + H265Config { + base: EncoderConfig { + resolution: config.resolution, + input_format: config.input_format, + quality: config.bitrate_kbps(), + fps: config.fps, + gop_size: config.gop_size(), + }, + bitrate_kbps: config.bitrate_kbps(), gop_size: config.gop_size(), + fps: config.fps, + input_format, }, - bitrate_kbps: config.bitrate_kbps(), - gop_size: config.gop_size(), - fps: config.fps, - input_format, - }, - &codec_name, - )?; - info!("Created H265 encoder: {}", encoder.codec_name()); - Box::new(H265EncoderWrapper(encoder)) + &codec_name, + )?; + info!("Created H265 encoder: {}", encoder.codec_name()); + Box::new(H265EncoderWrapper(encoder)) + } } VideoEncoderType::VP8 => { let codec_name = selected_codec_name.clone(); @@ -446,7 +523,9 @@ pub(super) fn build_encoder_state( }; let codec_name = encoder.codec_name(); - let use_direct_input = if codec_name.contains("rkmpp") { + let use_direct_input = if codec_name.contains("amlenc") { + pipeline_input_format == PixelFormat::Nv12 + } else if codec_name.contains("rkmpp") { matches!( pipeline_input_format, PixelFormat::Yuyv diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 1a625162..81012293 100644 --- a/src/video/pipeline/shared.rs +++ b/src/video/pipeline/shared.rs @@ -26,10 +26,11 @@ use std::time::{Duration, Instant}; use tokio::sync::{mpsc, watch, Mutex, RwLock}; use tracing::{debug, error, info, trace, warn}; -use super::encoder_state::{build_encoder_state, EncoderThreadState}; +use super::encoder_state::{build_encoder_state, should_parallel_decode_mjpeg, EncoderThreadState}; /// Grace period before auto-stopping pipeline when no subscribers (in seconds) const AUTO_STOP_GRACE_PERIOD_SECS: u64 = 3; +const AMLENC_MAX_FPS: u32 = 60; /// After this many consecutive timeouts, log a prominent warning. const CAPTURE_TIMEOUT_RESTART_THRESHOLD: u32 = 5; const CAPTURE_TIMEOUT_SOFT_RESTART_THRESHOLD: u32 = 3; @@ -50,9 +51,14 @@ use crate::video::capture::status::{ use crate::video::capture::{BridgeContext, CaptureReadError, CaptureStream}; use crate::video::codec::h264_bitstream; use crate::video::codec::registry::{EncoderBackend, VideoEncoderType}; +use crate::video::codec::MjpegToNv12Decoder; use crate::video::device::parse_bridge_kind; use crate::video::device::VideoControlMode; use crate::video::format::{PixelFormat, Resolution}; + +fn amlenc_supported_fps(requested_fps: u32) -> u32 { + requested_fps.min(AMLENC_MAX_FPS) +} use crate::video::frame::{FrameBuffer, FrameBufferPool, VideoFrame}; use crate::video::recovery::{wait_for_source_change, CaptureRecoveryPolicy}; use crate::video::signal::SignalStatus; @@ -281,6 +287,11 @@ pub struct SharedVideoPipeline { stats: Mutex, running: watch::Sender, running_rx: watch::Receiver, + /// Becomes true only after the synchronous encoder worker has dropped its + /// vendor handles. Capture teardown alone is not sufficient for AMLENC: + /// a blocked dequeue/encode can otherwise overlap the next pipeline. + encoder_done: watch::Sender, + encoder_done_rx: watch::Receiver, h264_profile_level_id: watch::Sender>, h264_profile_level_id_rx: watch::Receiver>, cmd_tx: ParkingRwLock>>, @@ -312,6 +323,7 @@ impl SharedVideoPipeline { ); let (running_tx, running_rx) = watch::channel(false); + let (encoder_done_tx, encoder_done_rx) = watch::channel(true); let (h264_profile_tx, h264_profile_rx) = watch::channel(None); let pipeline = Arc::new(Self { @@ -320,6 +332,8 @@ impl SharedVideoPipeline { stats: Mutex::new(SharedVideoPipelineStats::default()), running: running_tx, running_rx, + encoder_done: encoder_done_tx, + encoder_done_rx, h264_profile_level_id: h264_profile_tx, h264_profile_level_id_rx: h264_profile_rx, cmd_tx: ParkingRwLock::new(None), @@ -380,7 +394,10 @@ impl SharedVideoPipeline { /// Subscribe to encoded frames pub fn subscribe(&self) -> mpsc::Receiver> { - let (tx, rx) = mpsc::channel(4); + // A queued video frame is already stale when the next frame is ready. + // Keep at most one pending frame so a slow WebRTC writer cannot make + // the encoder wait or accumulate seconds of latency. + let (tx, rx) = mpsc::channel(1); self.subscribers.write().push(tx); rx } @@ -495,7 +512,7 @@ impl SharedVideoPipeline { let _ = self.h264_profile_level_id.send(Some(profile_level_id)); } - async fn broadcast_encoded(&self, frame: Arc) { + fn broadcast_encoded(&self, frame: Arc) { let subscribers = { let guard = self.subscribers.read(); if guard.is_empty() { @@ -505,9 +522,11 @@ impl SharedVideoPipeline { }; for tx in &subscribers { - if tx.send(frame.clone()).await.is_err() { - // Receiver dropped; cleanup happens below. - } + // Never await a consumer. A full one-slot queue means the + // consumer is behind; dropping this frame preserves bounded + // latency and the receiver's sequence-gap logic requests a fresh + // keyframe when necessary. + let _ = tx.try_send(frame.clone()); } if subscribers.iter().any(|tx| tx.is_closed()) { @@ -534,6 +553,18 @@ impl SharedVideoPipeline { } let mut config = self.config.read().await.clone(); + let parallel_mjpeg_decode = should_parallel_decode_mjpeg(&config); + if parallel_mjpeg_decode { + let stable_fps = amlenc_supported_fps(config.fps); + if stable_fps != config.fps { + warn!( + "Limiting S912 AMLENC capture at {}x{} from {} to {} fps (hardware limit)", + config.resolution.width, config.resolution.height, config.fps, stable_fps + ); + config.fps = stable_fps; + *self.config.write().await = config.clone(); + } + } { let mut last = self.last_state_notification.lock(); *last = None; @@ -565,6 +596,9 @@ impl SharedVideoPipeline { } config.resolution = negotiated_res; config.input_format = negotiated_fmt; + if parallel_mjpeg_decode { + config.fps = amlenc_supported_fps(config.fps); + } if previous != (config.resolution, config.input_format, config.fps) { info!( "Negotiated capture {}x{} {:?} @ {} fps (configured {}x{} {:?} @ {} fps) — aligning encoder to source", @@ -599,8 +633,14 @@ impl SharedVideoPipeline { Err(e) => return Err(e), }; - let mut encoder_state = build_encoder_state(&config)?; + let mut encoder_config = config.clone(); + if parallel_mjpeg_decode { + encoder_config.input_format = PixelFormat::Nv12; + info!("Using capture-thread libyuv MJPEG decode with parallel AMLENC encoding"); + } + let mut encoder_state = build_encoder_state(&encoder_config)?; let _ = self.running.send(true); + let _ = self.encoder_done.send(false); self.running_flag.store(true, Ordering::Release); let pipeline = self.clone(); @@ -663,12 +703,11 @@ impl SharedVideoPipeline { input_frame_count = input_frame_count.wrapping_add(1); - match pipeline.encode_frame_sync(&mut encoder_state, &frame, input_frame_count) - { + match pipeline.encode_frame_sync(&mut encoder_state, &frame) { Ok(encoded_frames) => { for encoded_frame in encoded_frames { let encoded_arc = Arc::new(encoded_frame); - handle.block_on(pipeline.broadcast_encoded(encoded_arc)); + pipeline.broadcast_encoded(encoded_arc); encoded_frame_count = encoded_frame_count.wrapping_add(1); fps_frame_count += 1; @@ -702,6 +741,10 @@ impl SharedVideoPipeline { } pipeline.clear_cmd_tx(); + // Dropping encoder_state here releases AMLENC before a caller + // is allowed to construct a replacement pipeline. + drop(encoder_state); + let _ = pipeline.encoder_done.send(true); }); } @@ -720,6 +763,8 @@ impl SharedVideoPipeline { let mut pixel_format = config.input_format; let mut active_fps = config.fps; let mut stride: u32 = 0; + let mut mjpeg_decoder = + parallel_mjpeg_decode.then(|| MjpegToNv12Decoder::new(config.resolution)); if let Some(s) = preopened { resolution = s.resolution(); @@ -1064,11 +1109,30 @@ impl SharedVideoPipeline { pixel_format, active_fps, )); + let (frame_data, frame_format, frame_stride) = + if let Some(decoder) = mjpeg_decoder.as_mut() { + let nv12_size = + resolution.width as usize * resolution.height as usize * 3 / 2; + let mut nv12 = buffer_pool.take(nv12_size); + if let Err(error) = decoder.decode_into(&owned, &mut nv12) { + buffer_pool.put(owned); + buffer_pool.put(nv12); + let key = "capture_mjpeg_decode"; + if capture_error_throttler.should_log(key) { + error!("Dropping undecodable MJPEG frame: {}", error); + } + continue; + } + buffer_pool.put(owned); + (nv12, PixelFormat::Nv12, resolution.width) + } else { + (owned, pixel_format, stride) + }; let frame = Arc::new(VideoFrame::from_pooled( - Arc::new(FrameBuffer::new(owned, Some(buffer_pool.clone()))), + Arc::new(FrameBuffer::new(frame_data, Some(buffer_pool.clone()))), resolution, - pixel_format, - stride, + frame_format, + frame_stride, meta.sequence, )); sequence = meta.sequence.wrapping_add(1); @@ -1099,7 +1163,6 @@ impl SharedVideoPipeline { &self, state: &mut EncoderThreadState, frame: &VideoFrame, - frame_count: u64, ) -> Result> { let fps = state.fps; let codec = state.codec; @@ -1190,16 +1253,6 @@ impl SharedVideoPipeline { .or(compacted_buf.as_deref()) .unwrap_or(raw_frame); - // Debug log for H265 - if codec == VideoEncoderType::H265 && frame_count % 30 == 1 { - debug!( - "[Pipeline-H265] Processing frame #{}: input_size={}, pts_ms={}", - frame_count, - raw_frame.len(), - pts_ms - ); - } - let needs_yuv420p = state.encoder_needs_yuv420p; let encoder = state .encoder @@ -1236,18 +1289,7 @@ impl SharedVideoPipeline { match encode_result { Ok(frames) => { if frames.is_empty() { - if codec == VideoEncoderType::H265 { - warn!( - "[Pipeline-H265] Encoder returned no frames for frame #{}", - frame_count - ); - } else { - trace!( - "Encoder returned no frames for input frame #{} ({})", - frame_count, - codec - ); - } + trace!("Encoder returned no frame ({})", codec); return Ok(Vec::new()); } @@ -1259,23 +1301,6 @@ impl SharedVideoPipeline { self.update_h264_profile_level_id(&encoded.data); } - // Debug log for H265 encoded frame - if codec == VideoEncoderType::H265 && (is_keyframe || frame_count % 30 == 1) { - debug!( - "[Pipeline-H265] Encoded frame #{}: output_size={}, keyframe={}, sequence={}", - frame_count, - encoded.data.len(), - is_keyframe, - sequence - ); - - // Log H265 NAL unit types in the encoded data - if is_keyframe { - let nal_types = parse_h265_nal_types(&encoded.data); - debug!("[Pipeline-H265] Keyframe NAL types: {:?}", nal_types); - } - } - encoded_frames.push(EncodedVideoFrame { data: encoded.data, pts_ms, @@ -1288,15 +1313,7 @@ impl SharedVideoPipeline { Ok(encoded_frames) } - Err(e) => { - if codec == VideoEncoderType::H265 { - error!( - "[Pipeline-H265] Encode error at frame #{}: {}", - frame_count, e - ); - } - Err(e) - } + Err(e) => Err(e), } } @@ -1316,6 +1333,7 @@ impl SharedVideoPipeline { pub async fn stop_and_wait(&self, timeout: std::time::Duration) -> Result<()> { self.stop(); let mut rx = self.running_watch(); + let mut encoder_rx = self.encoder_done_rx.clone(); let deadline = tokio::time::Instant::now() + timeout; while *rx.borrow() { @@ -1344,6 +1362,32 @@ impl SharedVideoPipeline { } } + while !*encoder_rx.borrow() { + let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); + if remaining.is_zero() { + return Err(AppError::VideoError(format!( + "Timed out waiting {:?} for video encoder to release vendor session", + timeout + ))); + } + match tokio::time::timeout(remaining, encoder_rx.changed()).await { + Ok(Ok(())) => {} + Ok(Err(_)) if *encoder_rx.borrow() => break, + Ok(Err(_)) => { + return Err(AppError::VideoError( + "Video encoder lifecycle channel closed before vendor session release" + .to_string(), + )); + } + Err(_) => { + return Err(AppError::VideoError(format!( + "Timed out waiting {:?} for video encoder to release vendor session", + timeout + ))); + } + } + } + Ok(()) } @@ -1548,58 +1592,6 @@ impl Drop for SharedVideoPipeline { } } -/// Parse H265 NAL unit types from Annex B data -fn parse_h265_nal_types(data: &[u8]) -> Vec<(u8, usize)> { - let mut nal_types = Vec::new(); - let mut i = 0; - - while i < data.len() { - // Find start code - let nal_start = if i + 4 <= data.len() - && data[i] == 0 - && data[i + 1] == 0 - && data[i + 2] == 0 - && data[i + 3] == 1 - { - i + 4 - } else if i + 3 <= data.len() && data[i] == 0 && data[i + 1] == 0 && data[i + 2] == 1 { - i + 3 - } else { - i += 1; - continue; - }; - - if nal_start >= data.len() { - break; - } - - // Find next start code to get NAL size - let mut nal_end = data.len(); - let mut j = nal_start + 1; - while j + 3 <= data.len() { - if (data[j] == 0 && data[j + 1] == 0 && data[j + 2] == 1) - || (j + 4 <= data.len() - && data[j] == 0 - && data[j + 1] == 0 - && data[j + 2] == 0 - && data[j + 3] == 1) - { - nal_end = j; - break; - } - j += 1; - } - - // H265 NAL type is in bits 1-6 of first byte - let nal_type = (data[nal_start] >> 1) & 0x3F; - let nal_size = nal_end - nal_start; - nal_types.push((nal_type, nal_size)); - i = nal_end; - } - - nal_types -} - #[cfg(test)] mod tests { use super::*; @@ -1612,6 +1604,11 @@ mod tests { let h265 = SharedVideoPipelineConfig::h265(Resolution::HD720, BitratePreset::Speed); assert_eq!(h265.output_codec, VideoEncoderType::H265); + + assert_eq!(amlenc_supported_fps(30), 30); + assert_eq!(amlenc_supported_fps(50), 50); + assert_eq!(amlenc_supported_fps(60), 60); + assert_eq!(amlenc_supported_fps(120), 60); } #[test] @@ -1644,12 +1641,15 @@ mod tests { )) .unwrap(); let _ = pipeline.running.send(true); + let _ = pipeline.encoder_done.send(false); pipeline.running_flag.store(true, Ordering::Release); let worker = pipeline.clone(); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(30)).await; let _ = worker.running.send(false); + tokio::time::sleep(Duration::from_millis(30)).await; + let _ = worker.encoder_done.send(true); }); let started = Instant::now(); @@ -1657,7 +1657,7 @@ mod tests { .stop_and_wait(Duration::from_secs(1)) .await .unwrap(); - assert!(started.elapsed() >= Duration::from_millis(20)); + assert!(started.elapsed() >= Duration::from_millis(50)); assert!(!pipeline.is_running()); } } diff --git a/src/web/routes.rs b/src/web/routes.rs index e49e1cb2..29f7eb32 100644 --- a/src/web/routes.rs +++ b/src/web/routes.rs @@ -76,6 +76,7 @@ pub fn create_router(state: Arc) -> Router { .route("/stream/mode", post(handlers::stream_mode_set)) .route("/stream/bitrate", post(handlers::stream_set_bitrate)) .route("/stream/codecs", get(handlers::stream_codecs_list)) + .route("/video/codecs", get(handlers::stream_codecs_list)) .route("/stream/constraints", get(handlers::stream_constraints_get)) .route( "/video/encoder/self-check", diff --git a/src/webrtc/universal_session.rs b/src/webrtc/universal_session.rs index 90f9adc9..83b50a3b 100644 --- a/src/webrtc/universal_session.rs +++ b/src/webrtc/universal_session.rs @@ -3,7 +3,7 @@ use std::net::IpAddr; use std::sync::Arc; use std::time::{Duration, Instant}; -use tokio::sync::{watch, Mutex, RwLock}; +use tokio::sync::{broadcast, watch, Mutex, RwLock}; use tracing::{debug, info, warn}; use webrtc::api::interceptor_registry::register_default_interceptors; use webrtc::api::media_engine::MediaEngine; @@ -20,6 +20,8 @@ use webrtc::peer_connection::configuration::RTCConfiguration; use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; use webrtc::peer_connection::RTCPeerConnection; +use webrtc::rtcp::payload_feedbacks::full_intra_request::FullIntraRequest; +use webrtc::rtcp::payload_feedbacks::picture_loss_indication::PictureLossIndication; use webrtc::rtp_transceiver::rtp_codec::{ RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType, }; @@ -38,9 +40,10 @@ use crate::video::codec::h264_bitstream; use crate::video::types::{ BitratePreset, EncodedVideoFrame, PixelFormat, Resolution, VideoEncoderType, }; -use std::sync::atomic::AtomicBool; const MIME_TYPE_H265: &str = "video/H265"; +const KEYFRAME_RETRY_LIMIT: u8 = 3; +const KEYFRAME_RETRY_BASE_DELAY: Duration = Duration::from_secs(1); fn is_allowed_ice_ip(ip: IpAddr) -> bool { match ip { @@ -110,9 +113,9 @@ pub struct UniversalSession { state_rx: watch::Receiver, ice_candidates: Arc>>, hid_controller: Option>, + keyframe_feedback: broadcast::Sender<()>, video_receiver_handle: Mutex>>, audio_receiver_handle: Mutex>>, - fps: u32, } impl UniversalSession { @@ -277,10 +280,45 @@ impl UniversalSession { let pc = Arc::new(pc); - pc.add_track(video_track.as_track_local()) + let video_sender = pc + .add_track(video_track.as_track_local()) .await .map_err(|e| AppError::VideoError(format!("Failed to add video track: {}", e)))?; + // RTCP feedback is advertised in SDP, but it only reaches the + // application while the sender is actively drained. Forward PLI/FIR + // to the shared encoder so a client that missed an IDR can recover. + let (keyframe_feedback, _) = broadcast::channel(8); + let keyframe_feedback_tx = keyframe_feedback.clone(); + let rtcp_session_id = session_id.clone(); + tokio::spawn(async move { + loop { + let (packets, _) = match video_sender.read_rtcp().await { + Ok(value) => value, + Err(error) => { + debug!( + "RTCP reader stopped for session {}: {}", + rtcp_session_id, error + ); + break; + } + }; + if packets.iter().any(|packet| { + packet + .as_any() + .downcast_ref::() + .is_some() + || packet.as_any().downcast_ref::().is_some() + }) { + info!( + "RTCP PLI/FIR requested a keyframe for session {}", + rtcp_session_id + ); + let _ = keyframe_feedback_tx.send(()); + } + } + }); + info!( "{} video track added to peer connection (session {})", config.codec, session_id @@ -309,9 +347,9 @@ impl UniversalSession { state_rx, ice_candidates: Arc::new(Mutex::new(vec![])), hid_controller: None, + keyframe_feedback, video_receiver_handle: Mutex::new(None), audio_receiver_handle: Mutex::new(None), - fps: config.fps, }; session.setup_event_handlers().await; @@ -503,9 +541,8 @@ impl UniversalSession { let video_track = self.video_track.clone(); let mut state_rx = self.state_rx.clone(); let session_id = self.session_id.clone(); - let _fps = self.fps; let expected_codec = self.codec; - let send_in_flight = Arc::new(AtomicBool::new(false)); + let mut keyframe_feedback = self.keyframe_feedback.subscribe(); let handle = tokio::spawn(async move { info!( @@ -538,7 +575,8 @@ impl UniversalSession { request_keyframe(); let mut waiting_for_keyframe = true; let mut last_sequence: Option = None; - let mut last_keyframe_request = Instant::now() - Duration::from_secs(1); + let mut keyframe_requests = 1u8; + let mut next_keyframe_retry = Instant::now() + KEYFRAME_RETRY_BASE_DELAY; let mut frames_sent: u64 = 0; @@ -557,6 +595,16 @@ impl UniversalSession { } } + feedback = keyframe_feedback.recv() => { + if feedback.is_ok() { + request_keyframe(); + waiting_for_keyframe = true; + keyframe_requests = 1; + next_keyframe_retry = + Instant::now() + KEYFRAME_RETRY_BASE_DELAY; + } + } + result = frame_rx.recv() => { let encoded_frame = match result { Some(frame) => frame, @@ -572,17 +620,6 @@ impl UniversalSession { continue; } - if expected_codec == VideoEncoderType::H265 - && (encoded_frame.is_keyframe || frames_sent.is_multiple_of(30)) { - debug!( - "[Session-H265] Received frame #{}: size={}, keyframe={}, seq={}", - frames_sent, - encoded_frame.data.len(), - encoded_frame.is_keyframe, - encoded_frame.sequence - ); - } - let mut gap_detected = false; if let Some(prev) = last_sequence { if encoded_frame.sequence > prev.saturating_add(1) { @@ -593,9 +630,12 @@ impl UniversalSession { if waiting_for_keyframe || gap_detected { if encoded_frame.is_keyframe { waiting_for_keyframe = false; + keyframe_requests = 0; } else { - if gap_detected { + if gap_detected && !waiting_for_keyframe { waiting_for_keyframe = true; + keyframe_requests = 0; + next_keyframe_retry = Instant::now(); } // Some H264 encoders output SPS/PPS in a separate non-keyframe AU @@ -605,11 +645,19 @@ impl UniversalSession { && h264_bitstream::has_sps_pps(encoded_frame.data.as_ref()); let now = Instant::now(); - if now.duration_since(last_keyframe_request) - >= Duration::from_millis(200) + if keyframe_requests < KEYFRAME_RETRY_LIMIT + && now >= next_keyframe_retry { request_keyframe(); - last_keyframe_request = now; + keyframe_requests += 1; + let backoff = 1u32 << (keyframe_requests - 1); + next_keyframe_retry = now + KEYFRAME_RETRY_BASE_DELAY * backoff; + if keyframe_requests == KEYFRAME_RETRY_LIMIT { + warn!( + "Session {} exhausted keyframe retry budget; waiting for the encoder's next natural keyframe", + session_id + ); + } } if !forward_h264_parameter_frame { continue; @@ -617,16 +665,12 @@ impl UniversalSession { } } - let _ = send_in_flight; - let send_result = video_track .write_frame_bytes( encoded_frame.data.clone(), encoded_frame.is_keyframe, ) .await; - let _ = send_in_flight; - match send_result { Ok(()) => { frames_sent += 1; diff --git a/src/webrtc/webrtc_streamer.rs b/src/webrtc/webrtc_streamer.rs index 852c403d..27884dc4 100644 --- a/src/webrtc/webrtc_streamer.rs +++ b/src/webrtc/webrtc_streamer.rs @@ -14,6 +14,7 @@ use crate::events::{EventBus, StreamKind, SystemEvent}; use crate::hid::HidController; use crate::video::capture::DEFAULT_CAPTURE_BUFFER_COUNT; use crate::video::codec::h264_bitstream; +use crate::video::codec::EncoderRegistry; use crate::video::device::{ enumerate_devices, select_recovery_device, VideoControlMode, VideoDevice, VideoDeviceInfo, VideoDeviceRecoveryHint, @@ -1313,6 +1314,26 @@ impl WebRtcStreamer { }; if pipeline_running { + let pipeline = self.video_pipeline.read().await.clone(); + if let Some(pipeline) = pipeline { + let pipeline_config = pipeline.config().await; + let selected_backend = pipeline_config.encoder_backend.or_else(|| { + EncoderRegistry::global() + .best_available_encoder(pipeline_config.output_codec) + .map(|encoder| encoder.backend) + }); + if pipeline_config.input_format == PixelFormat::Mjpeg + && selected_backend == Some(EncoderBackend::Amlogic) + { + info!( + "Applying AMLENC bitrate {} in the encoder worker without restarting MJPEG decode", + preset + ); + pipeline.set_bitrate_preset(preset).await?; + return Ok(()); + } + } + info!("Restarting video pipeline to apply new bitrate: {}", preset); self.stop_video_pipeline_and_release().await?; diff --git a/web/src/types/generated.ts b/web/src/types/generated.ts index 5223b850..890ea13d 100644 --- a/web/src/types/generated.ts +++ b/web/src/types/generated.ts @@ -148,6 +148,7 @@ export enum EncoderType { Amf = "amf", Rkmpp = "rkmpp", V4l2m2m = "v4l2m2m", + Amlogic = "amlogic", } export type BitratePreset =