perf(video): 新增 RKMPP DMA 采集编码通路并完善恢复逻辑

支持原生 HDMI 和 UVC 缓冲区导出，增加同步 RKMPP 编码及可选 MJPEG 硬件转码。
校验帧布局和缓冲区租约，在 DMA 不可用或编码失败时回退到复制通路。
保留自定义码率和 GOP 策略，重开采集时同步 HDMI 源帧率，并区分 UVC 超时状态。

验证：88 个视频测试通过（含 4 个新增回归测试）；ARM64 cargo check --tests 通过。

src/video/capture/dmabuf_layout.rs

@@ -0,0 +1,201 @@
+//! Conservative, dependency-free eligibility checks for linear RKMPP input.
+
+pub struct DmaCaptureLayout<'a> {
+    pub native_hdmi: bool,
+    pub driver: &'a str,
+    pub bus_info: &'a str,
+    pub configurable_usb: bool,
+    pub single_planar: bool,
+    pub fourcc: [u8; 4],
+    pub width: u32,
+    pub height: u32,
+    pub stride: u32,
+}
+
+impl DmaCaptureLayout<'_> {
+    /// Minimum readable bytes, not the driver's page-aligned allocation size.
+    pub fn minimum_bytes(&self) -> Option<usize> {
+        if self.width == 0
+            || self.height == 0
+            || self.width > 8192
+            || self.height > 8192
+            || self.width % 2 != 0
+            || self.height % 2 != 0
+        {
+            return None;
+        }
+        let bytes_per_row = if self.native_hdmi {
+            match &self.fourcc {
+                b"NV12" => self.width,
+                b"BGR3" => self.width.checked_mul(3)?,
+                _ => return None,
+            }
+        } else if self.configurable_usb
+            && self.single_planar
+            && self.driver == "uvcvideo"
+            && self.bus_info.starts_with("usb-")
+        {
+            match &self.fourcc {
+                // Compressed frames have variable bytesused and no byte stride.
+                b"MJPG" => return Some(4),
+                b"YUYV" if self.stride % 16 == 0 => self.width.checked_mul(2)?,
+                b"NV12" if self.stride % 16 == 0 => self.width,
+                b"RGB3" if self.stride % 16 == 0 => self.width.checked_mul(3)?,
+                _ => return None,
+            }
+        } else {
+            return None;
+        };
+        if self.stride < bytes_per_row {
+            return None;
+        }
+        let size = (self.stride as usize).checked_mul(self.height as usize)?;
+        if self.fourcc == *b"NV12" {
+            size.checked_mul(3)?.checked_div(2)
+        } else {
+            Some(size)
+        }
+    }
+}
+
+/// An MJPEG DMA packet has a bounded payload, not stride * height bytes.
+/// Reserve readable headroom for the MPP bitstream reader without modifying
+/// capture memory. The decoder receives only `used`, never the allocation size.
+pub fn valid_payload(
+    compressed: bool,
+    used: usize,
+    capacity: usize,
+    expected: Option<usize>,
+) -> bool {
+    if compressed {
+        used >= 4 && used.checked_add(64).is_some_and(|end| end <= capacity)
+    } else {
+        used > 0 && Some(used) == expected && used <= capacity
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn usb() -> DmaCaptureLayout<'static> {
+        DmaCaptureLayout {
+            native_hdmi: false,
+            driver: "uvcvideo",
+            bus_info: "usb-fc880000.usb-1.1",
+            configurable_usb: true,
+            single_planar: true,
+            fourcc: *b"YUYV",
+            width: 1920,
+            height: 1080,
+            stride: 3840,
+        }
+    }
+
+    #[test]
+    fn usb_yuyv_uses_byte_stride_and_supports_padding() {
+        let mut layout = usb();
+        assert_eq!(layout.minimum_bytes(), Some(4_147_200));
+        layout.width = 640;
+        layout.height = 480;
+        layout.stride = 1280;
+        assert_eq!(layout.minimum_bytes(), Some(614_400));
+        layout.stride = 1296;
+        assert_eq!(layout.minimum_bytes(), Some(622_080));
+    }
+
+    #[test]
+    fn usb_requires_correct_driver_bus_queue_and_control_mode() {
+        let mut layout = usb();
+        layout.driver = "rkcif";
+        assert_eq!(layout.minimum_bytes(), None);
+        layout = usb();
+        layout.bus_info = "platform:hdmi";
+        assert_eq!(layout.minimum_bytes(), None);
+        layout = usb();
+        layout.single_planar = false;
+        assert_eq!(layout.minimum_bytes(), None);
+        layout = usb();
+        layout.configurable_usb = false;
+        assert_eq!(layout.minimum_bytes(), None);
+    }
+
+    #[test]
+    fn unverified_usb_formats_stay_on_copy_path() {
+        for fourcc in [
+            *b"H264", *b"NV21", *b"NV16", *b"NV24", *b"BGR3", *b"YU12", *b"UYVY", *b"YVYU",
+            *b"BAD!",
+        ] {
+            let mut layout = usb();
+            layout.fourcc = fourcc;
+            assert_eq!(layout.minimum_bytes(), None, "{fourcc:?}");
+        }
+    }
+
+    #[test]
+    fn usb_nv12_rgb_and_mjpeg_layouts() {
+        let mut layout = usb();
+        layout.fourcc = *b"NV12";
+        layout.stride = 1920;
+        assert_eq!(layout.minimum_bytes(), Some(3_110_400));
+        layout.fourcc = *b"RGB3";
+        layout.stride = 5760;
+        assert_eq!(layout.minimum_bytes(), Some(6_220_800));
+        layout.stride = 1920;
+        assert_eq!(layout.minimum_bytes(), None);
+        layout.fourcc = *b"MJPG";
+        layout.stride = 0;
+        assert_eq!(layout.minimum_bytes(), Some(4));
+    }
+
+    #[test]
+    fn compressed_payload_is_bounded_and_not_allocation_size() {
+        assert!(valid_payload(true, 63163, 4147200, None));
+        for used in [0, 3, 4147200, usize::MAX] {
+            assert!(!valid_payload(true, used, 4147200, None));
+        }
+        assert!(valid_payload(true, 4, 68, None));
+        assert!(!valid_payload(true, 4, 67, None));
+        assert!(valid_payload(false, 614400, 614400, Some(614400)));
+        assert!(!valid_payload(false, 614399, 614400, Some(614400)));
+        assert!(!valid_payload(false, 614400, 614399, Some(614400)));
+    }
+
+    #[test]
+    fn malformed_geometry_or_stride_is_rejected() {
+        for (w, h, stride) in [
+            (0, 1080, 3840),
+            (1920, 0, 3840),
+            (1919, 1080, 3840),
+            (1920, 1079, 3840),
+            (8194, 1080, 16384),
+            (1920, 8194, 3840),
+            (1920, 1080, 0),
+            (1920, 1080, 1920),
+            (1920, 1080, 3841),
+            (u32::MAX, u32::MAX, u32::MAX),
+        ] {
+            let mut layout = usb();
+            layout.width = w;
+            layout.height = h;
+            layout.stride = stride;
+            assert_eq!(layout.minimum_bytes(), None);
+        }
+    }
+
+    #[test]
+    fn native_hdmi_formats_are_preserved_but_not_expanded() {
+        let mut layout = usb();
+        layout.native_hdmi = true;
+        layout.single_planar = false;
+        layout.fourcc = *b"BGR3";
+        layout.stride = 5760;
+        assert_eq!(layout.minimum_bytes(), Some(6_220_800));
+        layout.fourcc = *b"NV12";
+        layout.stride = 1920;
+        assert_eq!(layout.minimum_bytes(), Some(3_110_400));
+        layout.fourcc = *b"YUYV";
+        layout.stride = 3840;
+        assert_eq!(layout.minimum_bytes(), None);
+    }
+}

src/video/capture/linux.rs

@@ -3,6 +3,8 @@
 use std::fs::File;
 use std::io;
 use std::os::fd::AsFd;
+#[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+use std::os::fd::OwnedFd;
 use std::os::unix::fs::OpenOptionsExt;
 use std::path::{Path, PathBuf};
 use std::time::{Duration, Instant};
@@ -29,6 +31,10 @@ use crate::video::device::VideoControlMode;
 use crate::video::format::{PixelFormat, Resolution};
 use crate::video::signal::SignalStatus;
 
+#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))]
+#[path = "dmabuf_layout.rs"]
+mod dmabuf_layout;
+
 /// Metadata for a captured frame.
 #[derive(Debug, Clone, Copy)]
 pub struct CaptureMeta {
@@ -67,6 +73,8 @@ pub struct CaptureStream {
     bridge_kind: Option<CsiBridgeKind>,
     native_hdmirx_state: Option<NativeHdmirxState>,
     native_hdmirx_next_state_check: Option<Instant>,
+    #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+    dma_layout_bytes: Option<usize>,
 }
 
 fn open_capture_device(path: &Path) -> io::Result<File> {
@@ -319,6 +327,24 @@ impl CaptureStream {
             mappings.push(plane_maps);
         }
 
+        #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+        let dma_layout_bytes = PixelFormat::from_v4l2r(actual_fmt.pixelformat).and_then(|format| {
+            dmabuf_layout::DmaCaptureLayout {
+                native_hdmi: is_native_hdmirx,
+                driver: &caps.driver,
+                bus_info: &caps.bus_info,
+                configurable_usb: !is_source_following
+                    && bridge.kind.is_none()
+                    && !bridge.has_subdev(),
+                single_planar: queue == QueueType::VideoCapture,
+                fourcc: format.to_fourcc(),
+                width: actual_resolution.width,
+                height: actual_resolution.height,
+                stride,
+            }
+            .minimum_bytes()
+        });
+
         let mut stream = Self {
             fd,
             queue,
@@ -332,6 +358,8 @@ impl CaptureStream {
             bridge_kind: bridge.kind,
             native_hdmirx_state,
             native_hdmirx_next_state_check,
+            #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+            dma_layout_bytes,
         };
 
         stream.queue_all_buffers()?;
@@ -421,10 +449,7 @@ impl CaptureStream {
         }
     }
 
-    pub fn next_into(
-        &mut self,
-        dst: &mut Vec<u8>,
-    ) -> std::result::Result<CaptureMeta, CaptureReadError> {
+    fn dequeue_buffer(&mut self) -> std::result::Result<V4l2Buffer, CaptureReadError> {
         self.wait_ready()?;
 
         // Several vendor BSPs update G_FMT/DV timings without making the
@@ -455,6 +480,143 @@ impl CaptureStream {
                 };
                 CaptureReadError::Io(error)
             })?;
+        Ok(dqbuf)
+    }
+
+    /// Native HDMI NV12/BGR24 and single-planar USB UVC YUYV/NV12/RGB24/MJPEG.
+    /// Actual EXPBUF/import support is probed separately; failure retains copy.
+    #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+    pub(crate) fn supports_rkmpp_dmabuf(&self) -> bool {
+        self.dma_layout_bytes.is_some_and(|minimum| {
+            (2..=16).contains(&self.mappings.len())
+                && self
+                    .mappings
+                    .iter()
+                    .all(|planes| planes.len() == 1 && planes[0].len() >= minimum)
+        })
+    }
+
+    #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+    pub(crate) fn export_dmabufs(&self) -> io::Result<Vec<(OwnedFd, usize)>> {
+        if !self.supports_rkmpp_dmabuf() {
+            return Err(io::Error::new(
+                io::ErrorKind::Unsupported,
+                "Unsupported RKMPP DMA capture layout",
+            ));
+        }
+        self.mappings
+            .iter()
+            .enumerate()
+            .map(|(index, planes)| {
+                let fd = ioctl::expbuf(&self.fd, self.queue, index, 0, ioctl::ExpbufFlags::CLOEXEC)
+                    .map_err(|error| io::Error::other(error.to_string()))?;
+                Ok((fd, planes[0].len()))
+            })
+            .collect()
+    }
+
+    /// Run a synchronous consumer while a buffer is dequeued. QBUF occurs only
+    /// after the callback returns, including its error path. Consumers must end
+    /// hardware access before returning; see hwcodec::rkmpp_dmabuf::encode.
+    #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
+    pub(crate) fn with_next_dmabuf<T>(
+        &mut self,
+        consume: impl FnOnce(usize, usize, Option<OwnedFd>) -> T,
+    ) -> std::result::Result<(CaptureMeta, T), CaptureReadError> {
+        let buffer = self.dequeue_buffer()?;
+        let index = buffer.as_v4l2_buffer().index as usize;
+        let sequence = buffer.as_v4l2_buffer().sequence as u64;
+        if index >= self.mappings.len() {
+            return Err(
+                io::Error::new(io::ErrorKind::InvalidData, "Invalid capture buffer index").into(),
+            );
+        }
+        let expected = self.expected_capture_bytes();
+        let mapped_size = self.mappings[index][0].len();
+        let native_hdmi = self.native_hdmirx_state.is_some();
+        let compressed = self.format == PixelFormat::Mjpeg;
+        let lease = BufferReturn(Some(|| {
+            self.queue_buffer(index as u32)
+                .map_err(|e| io::Error::other(e.to_string()))
+        }));
+        if buffer.as_v4l2_buffer().flags & v4l2r::bindings::V4L2_BUF_FLAG_ERROR != 0 {
+            // A corrupt UVC frame is not a source change or a DMA failure.
+            // Return it without ever letting the encoder read its payload.
+            lease.finish()?;
+            return Err(io::Error::from(io::ErrorKind::WouldBlock).into());
+        }
+        if !native_hdmi
+            && buffer.as_v4l2_buffer().field != v4l2r::bindings::v4l2_field_V4L2_FIELD_NONE
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "Interlaced USB DMA frames are not supported",
+            )
+            .into());
+        }
+        let mut planes = buffer.planes_iter();
+        let plane = planes
+            .next()
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "Missing DMA plane"))?;
+        if planes.next().is_some() || plane.data_offset.copied().unwrap_or(0) != 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "Unsupported DMA plane offset/layout",
+            )
+            .into());
+        }
+        let bytes_used = *plane.bytesused as usize;
+        if !dmabuf_layout::valid_payload(compressed, bytes_used, mapped_size, expected) {
+            if !native_hdmi {
+                // An unexpected UVC payload is not evidence of a source mode
+                // change. Disable DMA instead of reopening it indefinitely.
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    "Unexpected USB DMA payload length",
+                )
+                .into());
+            }
+            return Err(CaptureReadError::SourceChanged);
+        }
+        // UVC commonly fills vmalloc memory on the CPU. Older BSP exporters
+        // cache DMA attachments without usable per-frame CPU-access sync hooks.
+        // A fresh export object forces a fresh device mapping of this completed
+        // frame. Reuse the actual capture allocation, not a stale attachment.
+        let fresh_fd = if !native_hdmi {
+            Some(
+                ioctl::expbuf(
+                    &self.fd,
+                    self.queue,
+                    index,
+                    0,
+                    ioctl::ExpbufFlags::CLOEXEC | ioctl::ExpbufFlags::RDWR,
+                )
+                .map_err(|error| {
+                    io::Error::new(
+                        io::ErrorKind::InvalidData,
+                        format!("USB DMA re-export failed: {error}"),
+                    )
+                })?,
+            )
+        } else {
+            None
+        };
+        let output = consume(index, bytes_used, fresh_fd);
+        lease.finish()?;
+        Ok((
+            CaptureMeta {
+                bytes_used,
+                sequence,
+            },
+            output,
+        ))
+    }
+
+    pub fn next_into(
+        &mut self,
+        dst: &mut Vec<u8>,
+    ) -> std::result::Result<CaptureMeta, CaptureReadError> {
+        let dqbuf = self.dequeue_buffer()?;
         let index = dqbuf.as_v4l2_buffer().index as usize;
         let sequence = dqbuf.as_v4l2_buffer().sequence as u64;
 
@@ -664,7 +826,7 @@ impl CaptureStream {
         Ok(())
     }
 
-    fn queue_buffer(&mut self, index: u32) -> Result<()> {
+    fn queue_buffer(&self, index: u32) -> Result<()> {
         let handle = MmapHandle;
         let planes = self.mappings[index as usize]
             .iter()
@@ -682,6 +844,64 @@ impl CaptureStream {
     }
 }
 
+#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))]
+struct BufferReturn<F: FnOnce() -> io::Result<()>>(Option<F>);
+
+#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))]
+impl<F: FnOnce() -> io::Result<()>> BufferReturn<F> {
+    fn finish(mut self) -> io::Result<()> {
+        self.0.take().expect("capture lease already returned")()
+    }
+}
+
+#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))]
+impl<F: FnOnce() -> io::Result<()>> Drop for BufferReturn<F> {
+    fn drop(&mut self) {
+        if let Some(return_buffer) = self.0.take() {
+            if let Err(error) = return_buffer() {
+                warn!("Failed to return leased capture buffer: {}", error);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod dma_lease_tests {
+    use super::*;
+    use std::cell::RefCell;
+
+    #[test]
+    fn returns_buffer_once_after_consumer_and_does_not_retry_failed_qbuf() {
+        let operations = RefCell::new(Vec::new());
+        let lease = BufferReturn(Some(|| {
+            operations.borrow_mut().push("qbuf");
+            Err(io::Error::other("device lost"))
+        }));
+        operations.borrow_mut().push("encode completed");
+        assert!(lease.finish().is_err());
+        assert_eq!(*operations.borrow(), ["encode completed", "qbuf"]);
+    }
+
+    #[test]
+    fn returns_buffer_on_validation_error_or_unwind() {
+        let returns = std::cell::Cell::new(0);
+        {
+            let _lease = BufferReturn(Some(|| {
+                returns.set(returns.get() + 1);
+                Ok(())
+            }));
+        }
+        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _lease = BufferReturn(Some(|| {
+                returns.set(returns.get() + 1);
+                Ok(())
+            }));
+            panic!("consumer panic");
+        }));
+        assert_eq!(returns.get(), 2);
+    }
+}
+
 impl Drop for CaptureStream {
     fn drop(&mut self) {
         // Release ordering matters on rkcif: a subsequent open()/S_FMT from a

src/video/capture/status.rs

@@ -2,8 +2,32 @@
 
 use std::io;
 
+#[cfg(any(
+    test,
+    all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm"))
+))]
+use crate::video::device::VideoControlMode;
 use crate::video::signal::SignalStatus;
 
+#[cfg(any(
+    test,
+    all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm"))
+))]
+pub(crate) fn capture_recovery_status(
+    control_mode: VideoControlMode,
+    error: &io::Error,
+) -> SignalStatus {
+    if control_mode == VideoControlMode::Configurable && error.kind() == io::ErrorKind::TimedOut {
+        return SignalStatus::UvcCaptureStall;
+    }
+    match classify_capture_io_error(error) {
+        CaptureIoErrorKind::TransientSignal {
+            status: Some(status),
+        } => status,
+        _ => SignalStatus::NoSignal,
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum CaptureIoErrorKind {
     DeviceLost,
@@ -52,6 +76,33 @@ pub fn capture_error_log_key(err: &io::Error) -> String {
 mod tests {
     use super::*;
 
+    #[test]
+    fn recovery_distinguishes_uvc_stalls_from_hdmi_signal_loss() {
+        let timeout = io::Error::from(io::ErrorKind::TimedOut);
+        assert_eq!(
+            capture_recovery_status(VideoControlMode::Configurable, &timeout),
+            SignalStatus::UvcCaptureStall
+        );
+        assert_eq!(
+            capture_recovery_status(VideoControlMode::SourceFollowing, &timeout),
+            SignalStatus::NoSignal
+        );
+        assert_eq!(
+            capture_recovery_status(
+                VideoControlMode::Configurable,
+                &io::Error::from_raw_os_error(71)
+            ),
+            SignalStatus::UvcUsbError
+        );
+        assert_eq!(
+            capture_recovery_status(
+                VideoControlMode::SourceFollowing,
+                &io::Error::from_raw_os_error(5)
+            ),
+            SignalStatus::NoSignal
+        );
+    }
+
     #[test]
     fn maps_known_signal_status_strings() {
         assert_eq!(

src/video/pipeline/dmabuf.rs

@@ -0,0 +1,392 @@
+//! RKMPP-only capture/encode worker. Raw buffer ownership never crosses into a
+//! latest-frame slot or a network subscriber. Other encoders use shared.rs.
+use super::*;
+use crate::video::capture::status::capture_recovery_status;
+use crate::video::codec::registry::EncoderRegistry;
+use hwcodec::rkmpp_dmabuf::{DmaEncoder, DmaEncoderConfig, DmaFormat};
+
+pub(super) fn eligible(config: &SharedVideoPipelineConfig) -> bool {
+    if std::env::var("ONE_KVM_RKMPP_DMABUF").as_deref() == Ok("0") {
+        return false;
+    }
+    // The UVC per-frame mapping needed by older BSPs costs more than copying
+    // compressed packets in our current tests. Keep JPEG DMA opt-in; raw DMA
+    // remains automatic. The existing JPEG hardware transcode is the default.
+    if config.input_format == PixelFormat::Mjpeg
+        && std::env::var("ONE_KVM_RKMPP_MJPEG_DMABUF").as_deref() != Ok("1")
+    {
+        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),
+    };
+    rkmpp_dma_eligible(
+        selected.map(|e| e.backend),
+        config.output_codec,
+        config.input_format,
+    )
+}
+
+pub(super) fn prepare(
+    stream: &CaptureStream,
+    config: &SharedVideoPipelineConfig,
+) -> Result<DmaEncoder> {
+    let buffers = stream
+        .export_dmabufs()
+        .map_err(|e| AppError::VideoError(e.to_string()))?;
+    DmaEncoder::new(
+        DmaEncoderConfig {
+            width: config.resolution.width,
+            height: config.resolution.height,
+            stride: stream.stride(),
+            format: match stream.format() {
+                PixelFormat::Nv12 => DmaFormat::Nv12,
+                PixelFormat::Bgr24 => DmaFormat::Bgr24,
+                PixelFormat::Yuyv => DmaFormat::Yuyv,
+                PixelFormat::Rgb24 => DmaFormat::Rgb24,
+                PixelFormat::Mjpeg => DmaFormat::Mjpeg,
+                _ => return Err(AppError::VideoError("Unsupported DMA format".into())),
+            },
+            hevc: config.output_codec == VideoEncoderType::H265,
+            fps: config.fps,
+            bitrate_kbps: config.bitrate_kbps(),
+            gop: config.gop_size().max(1),
+        },
+        buffers,
+    )
+    .map_err(AppError::VideoError)
+}
+
+enum CaptureEncoder {
+    Dma(DmaEncoder),
+    Copy(Box<EncoderThreadState>),
+}
+
+// Field order is intentional, including during unwinding: destroy the encoder
+// and its imported FDs before STREAMOFF/unmap/REQBUFS(0).
+struct ActiveCapture {
+    encoder: Option<CaptureEncoder>,
+    stream: CaptureStream,
+}
+
+impl ActiveCapture {
+    fn fallback(&mut self, config: &SharedVideoPipelineConfig) -> Result<()> {
+        drop(self.encoder.take());
+        self.encoder = Some(CaptureEncoder::Copy(Box::new(build_encoder_state(config)?)));
+        Ok(())
+    }
+}
+
+struct Completion(Arc<SharedVideoPipeline>);
+impl Drop for Completion {
+    fn drop(&mut self) {
+        self.0.running_flag.store(false, Ordering::Release);
+        self.0.clear_cmd_tx();
+        let _ = self.0.encoder_done.send(true);
+        let _ = self.0.running.send(false);
+        info!("RKMPP capture/encode worker stopped and device resources released");
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(super) fn start(
+    pipeline: Arc<SharedVideoPipeline>,
+    stream: CaptureStream,
+    encoder: DmaEncoder,
+    config: SharedVideoPipelineConfig,
+    device: std::path::PathBuf,
+    buffer_count: u32,
+    bridge: BridgeContext,
+) -> Result<()> {
+    let (tx, rx) = mpsc::unbounded_channel();
+    *pipeline.cmd_tx.write() = Some(tx);
+    pipeline.running_flag.store(true, Ordering::Release);
+    let _ = pipeline.encoder_done.send(false);
+    let _ = pipeline.running.send(true);
+    let worker = pipeline.clone();
+    info!(
+        "RKMPP DMA candidate: device={} format={:?} resolution={:?} stride={}",
+        device.display(),
+        stream.format(),
+        stream.resolution(),
+        stream.stride()
+    );
+    let active = ActiveCapture {
+        encoder: Some(CaptureEncoder::Dma(encoder)),
+        stream,
+    };
+    let result = std::thread::Builder::new()
+        .name("rkmpp-dmabuf".into())
+        .spawn(move || {
+            let _completion = Completion(worker.clone());
+            if let Err(error) = run(&worker, active, config, device, buffer_count, bridge, rx) {
+                error!("RKMPP DMA worker failed: {}", error);
+            }
+        });
+    if let Err(error) = result {
+        drop(Completion(pipeline));
+        return Err(AppError::VideoError(format!(
+            "Failed to start RKMPP DMA worker: {error}"
+        )));
+    }
+    info!("RKMPP DMA capture path active: no CPU raw-frame copies (ONE_KVM_RKMPP_DMABUF=0 disables it)");
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+fn run(
+    pipeline: &Arc<SharedVideoPipeline>,
+    initial: ActiveCapture,
+    mut config: SharedVideoPipelineConfig,
+    device: std::path::PathBuf,
+    buffer_count: u32,
+    bridge: BridgeContext,
+    mut commands: mpsc::UnboundedReceiver<PipelineCmd>,
+) -> Result<()> {
+    let policy = CaptureRecoveryPolicy::new(config.control_mode);
+    let mut active = Some(initial);
+    let mut allow_dma = true;
+    let mut failures = 0u32;
+    let mut idle_since: Option<Instant> = None;
+    let buffer_pool = Arc::new(FrameBufferPool::new(2)); // allocated only on fallback
+    let mut fps_frames = 0u32;
+    let mut fps_start = Instant::now();
+    let errors = LogThrottler::with_secs(5);
+
+    while pipeline.running_flag.load(Ordering::Acquire) {
+        if pipeline.subscriber_count() == 0 {
+            if idle_since.get_or_insert_with(Instant::now).elapsed()
+                >= Duration::from_secs(AUTO_STOP_GRACE_PERIOD_SECS)
+            {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(50));
+            continue;
+        }
+        idle_since = None;
+
+        while let Ok(command) = commands.try_recv() {
+            let PipelineCmd::SetBitrate { preset } = command;
+            // Preserve Custom values and preset-specific GOPs across fallback/reopen.
+            config.bitrate_preset = preset;
+            if let Some(capture) = active.as_mut() {
+                match capture.encoder.as_mut().expect("active encoder") {
+                    CaptureEncoder::Dma(encoder) => {
+                        if let Err(error) =
+                            encoder.reconfigure(config.bitrate_kbps(), config.gop_size().max(1))
+                        {
+                            warn!(
+                                "RKMPP DMA reconfigure failed, using copy encoder: {}",
+                                error
+                            );
+                            capture.fallback(&config)?;
+                            allow_dma = false;
+                            pipeline.keyframe_requested.store(true, Ordering::Release);
+                        }
+                    }
+                    CaptureEncoder::Copy(encoder) => {
+                        pipeline.apply_cmd(encoder, PipelineCmd::SetBitrate { preset })?
+                    }
+                }
+            }
+        }
+
+        if active.is_none() {
+            match open_capture_stream_for_retry(
+                &device,
+                config.resolution,
+                config.input_format,
+                config.fps,
+                buffer_count.max(2),
+                Duration::from_secs(2),
+                bridge.clone(),
+                config.control_mode,
+                is_device_lost_message,
+            ) {
+                CaptureOpenResult::Opened(stream) => {
+                    if stream.resolution() != config.resolution
+                        || stream.format() != config.input_format
+                    {
+                        *pipeline.pending_sync_geometry.lock() =
+                            Some((stream.resolution(), stream.format()));
+                        break;
+                    }
+                    config.align_source_fps(stream.source_fps());
+                    // Update only timing: a concurrently queued bitrate command
+                    // must retain the user's latest preset in the shared config.
+                    pipeline.config.blocking_write().fps = config.fps;
+                    let encoder = if allow_dma && stream.supports_rkmpp_dmabuf() {
+                        match prepare(&stream, &config) {
+                            Ok(encoder) => CaptureEncoder::Dma(encoder),
+                            Err(error) => {
+                                warn!("RKMPP DMA reopen failed, using copy encoder: {}", error);
+                                allow_dma = false;
+                                CaptureEncoder::Copy(Box::new(build_encoder_state(&config)?))
+                            }
+                        }
+                    } else {
+                        CaptureEncoder::Copy(Box::new(build_encoder_state(&config)?))
+                    };
+                    active = Some(ActiveCapture {
+                        encoder: Some(encoder),
+                        stream,
+                    });
+                    pipeline.keyframe_requested.store(true, Ordering::Release);
+                }
+                CaptureOpenResult::NoSignal(status) => {
+                    failures = failures.saturating_add(1);
+                    let delay = policy.retry_delay(failures);
+                    pipeline.notify_state(PipelineStateNotification::no_signal(
+                        status,
+                        Some(delay.as_millis() as u64),
+                    ));
+                    wait_for_source_change(&bridge, delay, || {
+                        pipeline.running_flag.load(Ordering::Acquire)
+                    });
+                    continue;
+                }
+                CaptureOpenResult::DeviceLost(reason) => {
+                    pipeline.mark_device_lost(reason);
+                    break;
+                }
+                CaptureOpenResult::Fatal => break,
+            }
+        }
+
+        let capture = active.as_mut().expect("opened capture");
+        let result = match capture.encoder.as_mut().expect("active encoder") {
+            CaptureEncoder::Dma(encoder) => {
+                let pts = pipeline.pts_ms();
+                capture
+                    .stream
+                    .with_next_dmabuf(|index, bytes_used, fresh_fd| {
+                        let keyframe = pipeline.keyframe_requested.swap(false, Ordering::AcqRel);
+                        // The callback holds the dequeue lease until native encode
+                        // completes (or destroys MPP on error), before QBUF.
+                        unsafe { encoder.encode(index, bytes_used, fresh_fd, pts, keyframe) }
+                    })
+                    .map(|(_, packet)| {
+                        packet
+                            .map(|packet| {
+                                let data = Bytes::from(packet);
+                                let is_keyframe = match config.output_codec {
+                                    VideoEncoderType::H264 => h264_bitstream::is_keyframe(&data),
+                                    VideoEncoderType::H265 => h265_bitstream::is_keyframe(&data),
+                                    _ => false,
+                                };
+                                let (data, is_keyframe) = pipeline.inspect_and_parameterize_packet(
+                                    config.output_codec,
+                                    data,
+                                    is_keyframe,
+                                );
+                                if config.output_codec == VideoEncoderType::H264 {
+                                    pipeline.update_h264_profile_level_id(&data);
+                                }
+                                vec![EncodedVideoFrame {
+                                    data,
+                                    pts_ms: pts,
+                                    is_keyframe,
+                                    sequence: pipeline.sequence.fetch_add(1, Ordering::Relaxed) + 1,
+                                    duration: Duration::from_micros(
+                                        1_000_000 / config.fps.max(1) as u64,
+                                    ),
+                                    codec: config.output_codec,
+                                }]
+                            })
+                            .map_err(AppError::VideoError)
+                    })
+            }
+            CaptureEncoder::Copy(encoder) => {
+                let mut raw = buffer_pool.take(0);
+                capture.stream.next_into(&mut raw).map(|meta| {
+                    let frame = VideoFrame::from_pooled(
+                        Arc::new(FrameBuffer::new(raw, Some(buffer_pool.clone()))),
+                        config.resolution,
+                        config.input_format,
+                        capture.stream.stride(),
+                        meta.sequence,
+                    );
+                    pipeline.encode_frame_sync(encoder, &frame)
+                })
+            }
+        };
+
+        match result {
+            Ok(Ok(frames)) => {
+                failures = 0;
+                pipeline.notify_state(PipelineStateNotification::streaming(
+                    config.resolution,
+                    config.input_format,
+                    config.fps,
+                ));
+                for frame in frames {
+                    pipeline.broadcast_encoded(Arc::new(frame));
+                    fps_frames += 1;
+                }
+            }
+            Ok(Err(error)) => {
+                if matches!(capture.encoder, Some(CaptureEncoder::Dma(_))) {
+                    warn!("RKMPP DMA encode failed; disabling DMA for this pipeline and using copy encoder: {}", error);
+                    capture.fallback(&config)?;
+                    allow_dma = false;
+                    pipeline.keyframe_requested.store(true, Ordering::Release);
+                } else if errors.should_log("copy_encode") {
+                    error!("RKMPP copy encode failed: {}", error);
+                }
+            }
+            Err(CaptureReadError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                continue
+            }
+            Err(CaptureReadError::Io(error))
+                if error.kind() == std::io::ErrorKind::InvalidData && allow_dma =>
+            {
+                warn!(
+                    "Unsupported RKMPP DMA frame layout, using copy encoder: {}",
+                    error
+                );
+                capture.fallback(&config)?;
+                allow_dma = false;
+                pipeline.keyframe_requested.store(true, Ordering::Release);
+            }
+            Err(error) => {
+                let mut status = SignalStatus::NoSignal;
+                if let CaptureReadError::Io(ref io) = error {
+                    if classify_capture_io_error(io) == CaptureIoErrorKind::DeviceLost
+                        || is_device_lost_message(&io.to_string())
+                    {
+                        pipeline.mark_device_lost(io.to_string());
+                        break;
+                    }
+                    if errors.should_log("capture") {
+                        warn!("RKMPP DMA capture recovery: {}", io);
+                    }
+                    status = capture_recovery_status(config.control_mode, io);
+                }
+                // ActiveCapture drops encoder/imports before the V4L2 stream.
+                drop(active.take());
+                failures = failures.saturating_add(1);
+                let delay = policy.retry_delay(failures);
+                pipeline.notify_state(PipelineStateNotification::no_signal(
+                    status,
+                    Some(delay.as_millis() as u64),
+                ));
+                if !matches!(error, CaptureReadError::SourceChanged) {
+                    wait_for_source_change(&bridge, delay, || {
+                        pipeline.running_flag.load(Ordering::Acquire)
+                    });
+                }
+            }
+        }
+        if fps_start.elapsed() >= Duration::from_secs(1) {
+            pipeline.stats.blocking_lock().current_fps =
+                fps_frames as f32 / fps_start.elapsed().as_secs_f32();
+            fps_frames = 0;
+            fps_start = Instant::now();
+        }
+    }
+    // Explicitly release in the worker before Completion publishes stopped.
+    drop(active);
+    Ok(())
+}

src/video/pipeline/shared.rs

@@ -29,6 +29,96 @@ use tracing::{debug, error, info, trace, warn};
 
 use super::encoder_state::{build_encoder_state, should_parallel_decode_mjpeg, EncoderThreadState};
 
+#[cfg(all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm")))]
+#[path = "dmabuf.rs"]
+mod dmabuf;
+
+#[cfg(any(
+    test,
+    all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm"))
+))]
+fn rkmpp_dma_eligible(
+    backend: Option<EncoderBackend>,
+    codec: VideoEncoderType,
+    format: PixelFormat,
+) -> bool {
+    backend == Some(EncoderBackend::Rkmpp)
+        && matches!(codec, VideoEncoderType::H264 | VideoEncoderType::H265)
+        && matches!(
+            format,
+            PixelFormat::Bgr24
+                | PixelFormat::Nv12
+                | PixelFormat::Yuyv
+                | PixelFormat::Rgb24
+                | PixelFormat::Mjpeg
+        )
+}
+
+#[cfg(test)]
+mod dma_selection_tests {
+    use super::*;
+    #[test]
+    fn only_selected_rkmpp_uses_dma() {
+        for backend in [
+            EncoderBackend::Software,
+            EncoderBackend::Vaapi,
+            EncoderBackend::Nvenc,
+            EncoderBackend::Qsv,
+            EncoderBackend::Amf,
+            EncoderBackend::V4l2m2m,
+        ] {
+            for codec in [VideoEncoderType::H264, VideoEncoderType::H265] {
+                for format in [
+                    PixelFormat::Bgr24,
+                    PixelFormat::Nv12,
+                    PixelFormat::Yuyv,
+                    PixelFormat::Rgb24,
+                    PixelFormat::Mjpeg,
+                ] {
+                    assert!(!rkmpp_dma_eligible(Some(backend), codec, format));
+                }
+            }
+        }
+        assert!(!rkmpp_dma_eligible(
+            None,
+            VideoEncoderType::H264,
+            PixelFormat::Nv12
+        ));
+        for codec in [VideoEncoderType::H264, VideoEncoderType::H265] {
+            for format in [
+                PixelFormat::Bgr24,
+                PixelFormat::Nv12,
+                PixelFormat::Yuyv,
+                PixelFormat::Rgb24,
+                PixelFormat::Mjpeg,
+            ] {
+                assert!(rkmpp_dma_eligible(
+                    Some(EncoderBackend::Rkmpp),
+                    codec,
+                    format
+                ));
+            }
+        }
+        for format in [
+            PixelFormat::Nv16,
+            PixelFormat::Nv21,
+            PixelFormat::Nv24,
+            PixelFormat::Yuv420,
+        ] {
+            assert!(!rkmpp_dma_eligible(
+                Some(EncoderBackend::Rkmpp),
+                VideoEncoderType::H264,
+                format
+            ));
+        }
+        assert!(!rkmpp_dma_eligible(
+            Some(EncoderBackend::Rkmpp),
+            VideoEncoderType::VP9,
+            PixelFormat::Nv12
+        ));
+    }
+}
+
 /// Grace period before auto-stopping pipeline when no subscribers (in seconds)
 const AUTO_STOP_GRACE_PERIOD_SECS: u64 = 3;
 /// After this many consecutive timeouts, log a prominent warning.
@@ -172,7 +262,9 @@ pub struct EncodedVideoFrame {
 }
 
 enum PipelineCmd {
-    SetBitrate { bitrate_kbps: u32, gop: u32 },
+    SetBitrate {
+        preset: crate::video::codec::BitratePreset,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -255,6 +347,15 @@ impl Default for SharedVideoPipelineConfig {
 }
 
 impl SharedVideoPipelineConfig {
+    /// Keep encoder timing aligned with the negotiated HDMI source on every open.
+    fn align_source_fps(&mut self, source_fps: Option<f64>) {
+        if self.control_mode == VideoControlMode::SourceFollowing {
+            if let Some(fps) = source_fps {
+                self.fps = fps.round().clamp(1.0, 120.0) as u32;
+            }
+        }
+    }
+
     /// Get effective bitrate in kbps
     pub fn bitrate_kbps(&self) -> u32 {
         self.bitrate_preset.bitrate_kbps()
@@ -538,14 +639,13 @@ impl SharedVideoPipeline {
 
     fn apply_cmd(&self, state: &mut EncoderThreadState, cmd: PipelineCmd) -> Result<()> {
         match cmd {
-            PipelineCmd::SetBitrate { bitrate_kbps, gop } => {
-                #[cfg(not(any(target_arch = "aarch64", target_arch = "arm")))]
-                let _ = gop;
+            PipelineCmd::SetBitrate { preset } => {
+                let bitrate_kbps = preset.bitrate_kbps();
                 #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
                 if state.ffmpeg_hw_enabled {
                     if let Some(ref mut pipeline) = state.ffmpeg_hw_pipeline {
                         pipeline
-                            .reconfigure(bitrate_kbps as i32, gop as i32)
+                            .reconfigure(bitrate_kbps as i32, preset.gop_size(state.fps) as i32)
                             .map_err(|e| {
                                 let detail = if e.is_empty() {
                                     ffmpeg_hw_last_error()
@@ -767,7 +867,8 @@ impl SharedVideoPipeline {
             subdev_path.clone(),
             parse_bridge_kind(bridge_kind.as_deref()),
         );
-        let preopened: Option<CaptureStream> = match open_capture_stream(
+        #[allow(unused_mut)]
+        let mut preopened: Option<CaptureStream> = match open_capture_stream(
             &device_path,
             config.resolution,
             config.input_format,
@@ -781,11 +882,7 @@ impl SharedVideoPipeline {
                 let negotiated_res = s.resolution();
                 let negotiated_fmt = s.format();
                 let previous = (config.resolution, config.input_format, config.fps);
-                if config.control_mode == VideoControlMode::SourceFollowing {
-                    if let Some(source_fps) = s.source_fps() {
-                        config.fps = source_fps.round().clamp(1.0, 120.0) as u32;
-                    }
-                }
+                config.align_source_fps(s.source_fps());
                 config.resolution = negotiated_res;
                 config.input_format = negotiated_fmt;
                 if previous != (config.resolution, config.input_format, config.fps) {
@@ -822,6 +919,32 @@ impl SharedVideoPipeline {
             Err(e) => return Err(e),
         };
 
+        #[cfg(all(target_os = "linux", any(target_arch = "aarch64", target_arch = "arm")))]
+        if dmabuf::eligible(&config) {
+            if let Some(stream) = preopened.as_ref().filter(|s| s.supports_rkmpp_dmabuf()) {
+                match dmabuf::prepare(stream, &config) {
+                    Ok(encoder) => {
+                        return dmabuf::start(
+                            self.clone(),
+                            preopened.take().expect("preopened DMA capture"),
+                            encoder,
+                            config,
+                            device_path,
+                            buffer_count,
+                            BridgeContext::from_parts(
+                                subdev_path,
+                                parse_bridge_kind(bridge_kind.as_deref()),
+                            ),
+                        );
+                    }
+                    Err(error) => warn!(
+                        "RKMPP DMA unavailable; using existing copy pipeline: {}",
+                        error
+                    ),
+                }
+            }
+        }
+
         let mut encoder_config = config.clone();
         if parallel_mjpeg_decode {
             encoder_config.input_format = PixelFormat::Nv12;
@@ -1400,23 +1523,7 @@ impl SharedVideoPipeline {
         let input_format = state.input_format;
         let raw_frame = frame.data();
 
-        let process_start = PROCESS_START.get_or_init(Instant::now);
-        let current_ts_us = process_start.elapsed().as_micros() as i64;
-        let start_ts_us = self.pipeline_start_time_us.load(Ordering::Acquire);
-        let pts_ms = if start_ts_us == 0 {
-            let start_ts_us = match self.pipeline_start_time_us.compare_exchange(
-                0,
-                current_ts_us,
-                Ordering::AcqRel,
-                Ordering::Acquire,
-            ) {
-                Ok(_) => current_ts_us,
-                Err(existing) => existing,
-            };
-            current_ts_us.saturating_sub(start_ts_us) / 1000
-        } else {
-            current_ts_us.saturating_sub(start_ts_us) / 1000
-        };
+        let pts_ms = self.pts_ms();
 
         #[cfg(any(target_arch = "aarch64", target_arch = "arm"))]
         if state.ffmpeg_hw_enabled {
@@ -1551,6 +1658,28 @@ impl SharedVideoPipeline {
         }
     }
 
+    fn pts_ms(&self) -> i64 {
+        let current_ts_us = PROCESS_START
+            .get_or_init(Instant::now)
+            .elapsed()
+            .as_micros() as i64;
+        let start_ts_us = self.pipeline_start_time_us.load(Ordering::Acquire);
+        let start_ts_us = if start_ts_us == 0 {
+            match self.pipeline_start_time_us.compare_exchange(
+                0,
+                current_ts_us,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => current_ts_us,
+                Err(existing) => existing,
+            }
+        } else {
+            start_ts_us
+        };
+        current_ts_us.saturating_sub(start_ts_us) / 1000
+    }
+
     /// Stop the pipeline (non-blocking, does not wait for capture thread to exit)
     pub fn stop(&self) {
         if self.running_flag.swap(false, Ordering::AcqRel) {
@@ -1630,13 +1759,11 @@ impl SharedVideoPipeline {
         &self,
         preset: crate::video::codec::BitratePreset,
     ) -> Result<()> {
-        let bitrate_kbps = preset.bitrate_kbps();
-        let gop = {
+        {
             let mut config = self.config.write().await;
             config.bitrate_preset = preset;
-            config.gop_size()
-        };
-        self.send_cmd(PipelineCmd::SetBitrate { bitrate_kbps, gop });
+        }
+        self.send_cmd(PipelineCmd::SetBitrate { preset });
         Ok(())
     }
 
@@ -1831,6 +1958,60 @@ mod tests {
     use super::*;
     use crate::video::codec::BitratePreset;
 
+    #[tokio::test]
+    async fn bitrate_commands_preserve_custom_values_and_gop_policy() {
+        let pipeline = SharedVideoPipeline::new(SharedVideoPipelineConfig::default()).unwrap();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        *pipeline.cmd_tx.write() = Some(tx);
+        for preset in [
+            BitratePreset::Custom(2500),
+            BitratePreset::Custom(1000),
+            BitratePreset::Speed,
+            BitratePreset::Quality,
+        ] {
+            pipeline.set_bitrate_preset(preset).await.unwrap();
+            let PipelineCmd::SetBitrate { preset: received } = rx.try_recv().unwrap();
+            assert_eq!(received, preset);
+            assert_eq!(pipeline.config().await.bitrate_preset, preset);
+            // Rebuilt encoders must retain the preset's policy at the new FPS.
+            let restored = SharedVideoPipelineConfig {
+                bitrate_preset: received,
+                fps: 60,
+                ..Default::default()
+            };
+            assert_eq!(restored.bitrate_kbps(), preset.bitrate_kbps());
+            assert_eq!(restored.gop_size(), preset.gop_size(60));
+        }
+    }
+
+    #[test]
+    fn source_reopen_updates_fps_and_gop_without_changing_geometry_or_bitrate() {
+        let mut config = SharedVideoPipelineConfig {
+            control_mode: VideoControlMode::SourceFollowing,
+            resolution: Resolution::HD1080,
+            fps: 60,
+            bitrate_preset: BitratePreset::Quality,
+            ..Default::default()
+        };
+        config.align_source_fps(Some(29.97));
+        assert_eq!(config.fps, 30);
+        assert_eq!(config.gop_size(), 60);
+        assert_eq!(config.resolution, Resolution::HD1080);
+        assert_eq!(config.bitrate_kbps(), 8000);
+        config.align_source_fps(None);
+        assert_eq!(config.fps, 30);
+        config.align_source_fps(Some(59.94));
+        assert_eq!(config.fps, 60);
+        assert_eq!(config.gop_size(), 120);
+    }
+
+    #[test]
+    fn configurable_capture_keeps_requested_fps() {
+        let mut config = SharedVideoPipelineConfig::default();
+        config.align_source_fps(Some(60.0));
+        assert_eq!(config.fps, 30);
+    }
+
     #[test]
     fn test_pipeline_config() {
         let h264 = SharedVideoPipelineConfig::h264(Resolution::HD1080, BitratePreset::Balanced);