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

feat: 新增 Amlogic AMLENC 硬件编码支持

Browse Source 
- 新增动态加载及 ABI 校验的 AMLENC H.264/H.265 编码器
- 注册 Amlogic 后端并扩展编码器自检
- 并行执行 MJPEG 解码与 AMLENC 编码
- 复用 NV12 缓冲区并移除冗余帧初始化
- 优化低延迟帧分发及 RTCP 关键帧恢复
- 动态调整 AMLENC 码率尚未完成，效果不佳
This commit is contained in:
mofeng-git
2026-08-02 16:23:28 +08:00
parent 814f23a27c
commit 47af17bebc
18 changed files with 1683 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<SharedVideoPipelineStats>,
running: watch::Sender<bool>,
running_rx: watch::Receiver<bool>,
/// 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<bool>,
encoder_done_rx: watch::Receiver<bool>,
h264_profile_level_id: watch::Sender<Option<String>>,
h264_profile_level_id_rx: watch::Receiver<Option<String>>,
cmd_tx: ParkingRwLock<Option<tokio::sync::mpsc::UnboundedSender<PipelineCmd>>>,
@@ -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<Arc<EncodedVideoFrame>> {
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<EncodedVideoFrame>) {
fn broadcast_encoded(&self, frame: Arc<EncodedVideoFrame>) {
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<Vec<EncodedVideoFrame>> {
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());
}
}