From db9d79554a80c424dd0e68c20c11b85c593c9aba Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Sat, 5 Sep 2026 20:55:30 +0800 Subject: [PATCH] =?UTF-8?q?perf(video):=20=E6=96=B0=E5=A2=9E=20RKMPP=20DMA?= =?UTF-8?q?=20=E9=87=87=E9=9B=86=E7=BC=96=E7=A0=81=E9=80=9A=E8=B7=AF?= =?UTF-8?q?=E5=B9=B6=E5=AE=8C=E5=96=84=E6=81=A2=E5=A4=8D=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 支持原生 HDMI 和 UVC 缓冲区导出，增加同步 RKMPP 编码及可选 MJPEG 硬件转码。 校验帧布局和缓冲区租约，在 DMA 不可用或编码失败时回退到复制通路。 保留自定义码率和 GOP 策略，重开采集时同步 HDMI 源帧率，并区分 UVC 超时状态。 验证：88 个视频测试通过（含 4 个新增回归测试）；ARM64 cargo check --tests 通过。 --- libs/hwcodec/build.rs | 1 + libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dma_jpeg.h | 38 ++ libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf.cpp | 329 +++++++++++++++ libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf_ffi.h | 31 ++ libs/hwcodec/src/lib.rs | 5 + libs/hwcodec/src/rkmpp_dmabuf.rs | 193 +++++++++ src/video/capture/dmabuf_layout.rs | 201 +++++++++ src/video/capture/linux.rs | 230 +++++++++- src/video/capture/status.rs | 51 +++ src/video/pipeline/dmabuf.rs | 392 ++++++++++++++++++ src/video/pipeline/shared.rs | 247 +++++++++-- 11 files changed, 1680 insertions(+), 38 deletions(-) create mode 100644 libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dma_jpeg.h create mode 100644 libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf.cpp create mode 100644 libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf_ffi.h create mode 100644 libs/hwcodec/src/rkmpp_dmabuf.rs create mode 100644 src/video/capture/dmabuf_layout.rs create mode 100644 src/video/pipeline/dmabuf.rs diff --git a/libs/hwcodec/build.rs b/libs/hwcodec/build.rs index 26aad7f3..08441f97 100644 --- a/libs/hwcodec/build.rs +++ b/libs/hwcodec/build.rs @@ -488,6 +488,7 @@ mod ffmpeg { } } builder.file(ffmpeg_hw_dir.join("ffmpeg_hw_mjpeg_h26x.cpp")); + builder.file(ffmpeg_hw_dir.join("rkmpp_dmabuf.cpp")); } else { println!( "cargo:info=Skipping ffmpeg_hw_mjpeg_h26x.cpp (RKMPP) for arch {}", diff --git a/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dma_jpeg.h b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dma_jpeg.h new file mode 100644 index 00000000..c7f4ad56 --- /dev/null +++ b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dma_jpeg.h @@ -0,0 +1,38 @@ +#pragma once +#include +#include + +// Validate bounded JPEG headers before giving hardware a fixed-size output +// buffer. Only baseline 8-bit JPEG is accepted; other streams use copy fallback. +// No scan-data traversal or full-packet copy is needed. +inline bool rkmpp_dma_jpeg_header(const uint8_t *data, size_t size, int width, int height) { + if (!data || size < 4 || data[0] != 0xff || data[1] != 0xd8) return false; + size_t pos = 2; + bool sof = false; + while (pos < size) { + if (data[pos++] != 0xff) return false; + while (pos < size && data[pos] == 0xff) ++pos; + if (pos == size) return false; + const unsigned marker = data[pos++]; + if (!marker || marker == 0xd8 || marker == 0xd9 || marker == 1 || + (marker >= 0xd0 && marker <= 0xd7)) return false; + if (size - pos < 2) return false; + const size_t length = (size_t(data[pos]) << 8) | data[pos + 1]; + if (length < 2 || length > size - pos) return false; + if (marker == 0xc0) { + if (sof || length < 8 || data[pos + 2] != 8) return false; + const unsigned h = (unsigned(data[pos + 3]) << 8) | data[pos + 4]; + const unsigned w = (unsigned(data[pos + 5]) << 8) | data[pos + 6]; + const unsigned components = data[pos + 7]; + if (w != unsigned(width) || h != unsigned(height) || + (components != 1 && components != 3) || length != 8 + 3 * components) return false; + sof = true; + } else if (marker >= 0xc0 && marker <= 0xcf && marker != 0xc4 && marker != 0xcc) { + return false; // Progressive, lossless, extended or differential SOF. + } else if (marker == 0xda) { + return sof && length >= 6 && size - pos > length; + } + pos += length; + } + return false; +} diff --git a/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf.cpp b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf.cpp new file mode 100644 index 00000000..74c13621 --- /dev/null +++ b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf.cpp @@ -0,0 +1,329 @@ +#include "rkmpp_dmabuf_ffi.h" +#include +#include +#include +#include +#include "rkmpp_dma_jpeg.h" + +// Native MPP is already linked by the ARM FFmpeg/RKMPP build. Keep this +// optional for toolchains which only supply the FFmpeg headers. +#if defined(__linux__) && __has_include() +#define HAVE_MPP_DMA 1 +#include +#include +#include +extern "C" { +#include +#include +#include +#include +#include +#include +#include +} +#endif + +static thread_local char dma_error[192] = {}; +static int fail(const char *operation, int code) { + std::snprintf(dma_error, sizeof(dma_error), "%s (ret=%d)", operation, code); + return -1; +} + +#ifdef HAVE_MPP_DMA +struct RkmppDmaEncoder { + MppCtx ctx = nullptr; + MppApi *api = nullptr; + MppEncCfg cfg = nullptr; + MppPacket packet = nullptr; + std::array buffers{}; + std::array capacities{}; + size_t count = 0; + size_t minimum = 0; + int width = 0, height = 0, stride = 0; + MppFrameFormat format = MPP_FMT_YUV420SP; + bool jpeg = false; + MppCtx decoder = nullptr; + MppApi *dec_api = nullptr; + MppBufferGroup decoded_group = nullptr; + MppBuffer decoded_buffer = nullptr; + MppFrame decoded_frame = nullptr; + MppPacket input_packet = nullptr; + bool decoded_layout_set = false; + int decoded_stride = 0, decoded_vstride = 0; + + void close() { + if (packet) mpp_packet_deinit(&packet); + if (ctx) { + // A timeout must not expose still-in-use input to V4L2 QBUF. + api->reset(ctx); + mpp_destroy(ctx); + ctx = nullptr; + } + // On any decoder failure, end hardware access before releasing the + // input packet, exported buffers, or allowing the caller's QBUF. + if (decoder) { + dec_api->reset(decoder); + mpp_destroy(decoder); + decoder = nullptr; + } + if (input_packet) mpp_packet_deinit(&input_packet); + if (decoded_frame) mpp_frame_deinit(&decoded_frame); + if (decoded_buffer) { mpp_buffer_put(decoded_buffer); decoded_buffer = nullptr; } + if (decoded_group) { mpp_buffer_group_put(decoded_group); decoded_group = nullptr; } + for (auto &buffer : buffers) { + if (buffer) { mpp_buffer_put(buffer); buffer = nullptr; } + } + if (cfg) { mpp_enc_cfg_deinit(cfg); cfg = nullptr; } + } + ~RkmppDmaEncoder() { close(); } +}; + +static bool set_cfg(RkmppDmaEncoder *e, const char *key, int value) { + int ret = mpp_enc_cfg_set_s32(e->cfg, key, value); + if (ret) fail(key, ret); + return ret == 0; +} + +extern "C" int rkmpp_dma_reconfigure(RkmppDmaEncoder *e, int kbps, int gop) { + if (!e || !e->ctx || kbps <= 0 || kbps > 1000000 || gop <= 0) + return fail("invalid DMA encoder configuration", -1); + const int bps = kbps * 1000; + if (!set_cfg(e, "rc:bps_target", bps) || + !set_cfg(e, "rc:bps_max", bps + bps / 16) || + !set_cfg(e, "rc:bps_min", bps - bps / 16) || + !set_cfg(e, "rc:gop", gop)) return -1; + int ret = e->api->control(e->ctx, MPP_ENC_SET_CFG, e->cfg); + return ret ? fail("MPP_ENC_SET_CFG", ret) : 0; +} + +extern "C" RkmppDmaEncoder *rkmpp_dma_new( + int width, int height, int stride, int format, int codec, int fps, + int kbps, int gop, const int *fds, const size_t *sizes, size_t count) { + if (width <= 0 || height <= 0 || width > 8192 || height > 8192 || + (width & 1) || (height & 1) || format < 0 || format > 4 || + codec < 0 || codec > 1 || fps <= 0 || fps > 240 || + (format != 4 && stride < width * (format == 1 || format == 3 ? 3 : format == 2 ? 2 : 1)) || + (format == 2 && stride % 16 != 0) || !fds || !sizes || !count || count > 16) { + fail("invalid DMA frame layout", -1); return nullptr; + } + auto *e = new (std::nothrow) RkmppDmaEncoder; + if (!e) { fail("allocate DMA encoder", -1); return nullptr; } + e->width = width; e->height = height; e->stride = stride; e->count = count; + e->jpeg = format == 4; + if (e->jpeg) e->stride = stride = (width + 15) & ~15; + const int vstride = e->jpeg ? (height + 15) & ~15 : height; + e->format = format == 3 ? MPP_FMT_RGB888 : format == 2 ? MPP_FMT_YUV422_YUYV : format == 1 ? MPP_FMT_BGR888 : MPP_FMT_YUV420SP; + auto abort_init = [e](const char *op, int ret) -> RkmppDmaEncoder * { + fail(op, ret); delete e; return nullptr; + }; + int ret = mpp_create(&e->ctx, &e->api); + if (ret) return abort_init("mpp_create", ret); + RK_S64 timeout = 2000; + ret = e->api->control(e->ctx, MPP_SET_OUTPUT_TIMEOUT, &timeout); + if (ret) return abort_init("MPP_SET_OUTPUT_TIMEOUT", ret); + ret = e->api->control(e->ctx, MPP_SET_INPUT_TIMEOUT, &timeout); + if (ret) return abort_init("MPP_SET_INPUT_TIMEOUT", ret); + ret = mpp_init(e->ctx, MPP_CTX_ENC, codec ? MPP_VIDEO_CodingHEVC : MPP_VIDEO_CodingAVC); + if (ret) return abort_init("mpp_init", ret); + ret = mpp_enc_cfg_init(&e->cfg); + if (ret) return abort_init("mpp_enc_cfg_init", ret); + ret = e->api->control(e->ctx, MPP_ENC_GET_CFG, e->cfg); + if (ret) return abort_init("MPP_ENC_GET_CFG", ret); + if (!set_cfg(e, "prep:width", width) || !set_cfg(e, "prep:height", height) || + !set_cfg(e, "prep:hor_stride", stride) || !set_cfg(e, "prep:ver_stride", vstride) || + !set_cfg(e, "prep:format", e->format) || !set_cfg(e, "rc:mode", MPP_ENC_RC_MODE_CBR) || + !set_cfg(e, "rc:fps_in_flex", 0) || !set_cfg(e, "rc:fps_in_num", fps) || + !set_cfg(e, "rc:fps_in_denorm", 1) || !set_cfg(e, "rc:fps_out_flex", 0) || + !set_cfg(e, "rc:fps_out_num", fps) || !set_cfg(e, "rc:fps_out_denorm", 1) || + !set_cfg(e, "codec:type", codec ? MPP_VIDEO_CodingHEVC : MPP_VIDEO_CodingAVC)) { + delete e; return nullptr; + } + // Match the browser-friendly baseline profile used by the existing RKMPP + // byte encoder, rather than inheriting MPP's High-profile default. + const int level = int64_t(width) * height * fps <= int64_t(1920) * 1080 * 60 ? 42 : 52; + if (!codec && (!set_cfg(e, "h264:profile", 66) || !set_cfg(e, "h264:level", level) || + !set_cfg(e, "h264:cabac_en", 0) || !set_cfg(e, "h264:trans8x8", 0))) { + delete e; return nullptr; + } + if (rkmpp_dma_reconfigure(e, kbps, gop)) { delete e; return nullptr; } + MppEncHeaderMode mode = MPP_ENC_HEADER_MODE_EACH_IDR; + ret = e->api->control(e->ctx, MPP_ENC_SET_HEADER_MODE, &mode); + if (ret) return abort_init("MPP_ENC_SET_HEADER_MODE", ret); + // Reject arithmetic overflow even on 32-bit ARM; stride is supplied by a driver. + if (size_t(stride) > std::numeric_limits::max() / size_t(height)) + return abort_init("DMA buffer size overflow", -1); + size_t minimum = size_t(stride) * height; + if (format == 0) { + if (minimum > std::numeric_limits::max() / 3) + return abort_init("DMA buffer size overflow", -1); + minimum = minimum * 3 / 2; + } + if (e->jpeg) minimum = 68; // SOI/payload plus bounded hardware read headroom. + e->minimum = minimum; + for (size_t i = 0; i < count; ++i) { + if (fds[i] < 0 || sizes[i] < minimum) return abort_init("short DMA buffer", -1); + MppBufferInfo info{}; + info.type = MPP_BUFFER_TYPE_EXT_DMA; info.fd = fds[i]; + info.size = sizes[i]; info.index = static_cast(i); + ret = mpp_buffer_import(&e->buffers[i], &info); + if (ret) return abort_init("mpp_buffer_import", ret); + e->capacities[i] = sizes[i]; + } + if (e->jpeg) { + ret = mpp_create(&e->decoder, &e->dec_api); + if (ret) return abort_init("mpp_create JPEG decoder", ret); + ret = mpp_init(e->decoder, MPP_CTX_DEC, MPP_VIDEO_CodingMJPEG); + if (ret) return abort_init("mpp_init JPEG decoder", ret); + MppFrameFormat output = MPP_FMT_YUV420SP; + ret = e->dec_api->control(e->decoder, MPP_DEC_SET_OUTPUT_FORMAT, &output); + if (ret) return abort_init("JPEG NV12 output", ret); + ret = mpp_buffer_group_get_internal(&e->decoded_group, MPP_BUFFER_TYPE_DRM); + if (ret) return abort_init("JPEG output buffer group", ret); + // MPP JPEG requires aligned storage; reserve the conservative size used + // by its advanced-task decoder demo. One output reused after encode. + ret = mpp_buffer_get(e->decoded_group, &e->decoded_buffer, size_t(stride) * vstride * 4); + if (ret) return abort_init("JPEG output buffer", ret); + ret = mpp_frame_init(&e->decoded_frame); + if (ret) return abort_init("JPEG output frame", ret); + mpp_frame_set_buffer(e->decoded_frame, e->decoded_buffer); + } + return e; +} + +static int dma_read_sync(MppBuffer buffer, bool start) { + dma_buf_sync sync{}; + sync.flags = DMA_BUF_SYNC_READ | (start ? DMA_BUF_SYNC_START : DMA_BUF_SYNC_END); + int ret; + do { ret = ioctl(mpp_buffer_get_fd(buffer), DMA_BUF_IOCTL_SYNC, &sync); } + while (ret < 0 && errno == EINTR); + return ret; +} + +static int decode_jpeg(RkmppDmaEncoder *e, size_t index, size_t bytes_used) { + MppBuffer input = e->buffers[index]; + // The parser reads only header bytes with explicit DMA CPU-read ownership. + if (dma_read_sync(input, true)) return fail("JPEG DMA read sync start", errno); + const auto *data = static_cast(mpp_buffer_get_ptr(input)); + const bool valid = rkmpp_dma_jpeg_header(data, bytes_used, e->width, e->height); + if (dma_read_sync(input, false)) return fail("JPEG DMA read sync end", errno); + if (!valid) return fail("unsupported/mismatched JPEG header", -1); + + int ret = mpp_packet_init_with_buffer(&e->input_packet, input); + if (ret) return fail("JPEG input packet", ret); + mpp_packet_set_length(e->input_packet, bytes_used); + MppTask task = nullptr; + ret = e->dec_api->poll(e->decoder, MPP_PORT_INPUT, static_cast(2000)); + if (ret) return fail("JPEG input poll", ret); + ret = e->dec_api->dequeue(e->decoder, MPP_PORT_INPUT, &task); + if (ret || !task) return fail("JPEG input task", ret); + ret = mpp_task_meta_set_packet(task, KEY_INPUT_PACKET, e->input_packet); + if (ret) return fail("JPEG input metadata", ret); + ret = mpp_task_meta_set_frame(task, KEY_OUTPUT_FRAME, e->decoded_frame); + if (ret) return fail("JPEG output metadata", ret); + ret = e->dec_api->enqueue(e->decoder, MPP_PORT_INPUT, task); + if (ret) return fail("JPEG submit", ret); + task = nullptr; + ret = e->dec_api->poll(e->decoder, MPP_PORT_OUTPUT, static_cast(2000)); + if (ret) return fail("JPEG output poll", ret); + ret = e->dec_api->dequeue(e->decoder, MPP_PORT_OUTPUT, &task); + if (ret || !task) return fail("JPEG output task", ret); + MppFrame result = nullptr; + ret = mpp_task_meta_get_frame(task, KEY_OUTPUT_FRAME, &result); + if (ret || result != e->decoded_frame) return fail("JPEG output frame mismatch", ret); + ret = e->dec_api->enqueue(e->decoder, MPP_PORT_OUTPUT, task); + if (ret) return fail("JPEG return output task", ret); + ret = e->dec_api->poll(e->decoder, MPP_PORT_INPUT, static_cast(2000)); + if (ret) return fail("JPEG input completion", ret); + mpp_packet_deinit(&e->input_packet); + if (mpp_frame_get_errinfo(result) || mpp_frame_get_discard(result) || + mpp_frame_get_info_change(result) || + mpp_frame_get_width(result) != unsigned(e->width) || + mpp_frame_get_height(result) != unsigned(e->height) || + mpp_frame_get_fmt(result) != MPP_FMT_YUV420SP || + mpp_frame_get_buffer(result) != e->decoded_buffer) { + std::snprintf(dma_error, sizeof(dma_error), + "invalid JPEG decoded frame: err=%u discard=%u info_change=%u size=%ux%u fmt=%x buffer_match=%d", + mpp_frame_get_errinfo(result), mpp_frame_get_discard(result), mpp_frame_get_info_change(result), + mpp_frame_get_width(result), mpp_frame_get_height(result), unsigned(mpp_frame_get_fmt(result)), + int(mpp_frame_get_buffer(result) == e->decoded_buffer)); + return -1; + } + const int hs = mpp_frame_get_hor_stride(result), vs = mpp_frame_get_ver_stride(result); + if (hs < e->width || vs < e->height || hs > 8192 || vs > 8192 || (hs & 15) || (vs & 15) || + size_t(hs) * vs * 3 / 2 > mpp_buffer_get_size(e->decoded_buffer)) + return fail("invalid JPEG decoded stride", -1); + if (!e->decoded_layout_set) { + if (!set_cfg(e, "prep:hor_stride", hs) || !set_cfg(e, "prep:ver_stride", vs)) return -1; + ret = e->api->control(e->ctx, MPP_ENC_SET_CFG, e->cfg); + if (ret) return fail("JPEG encoder layout", ret); + e->decoded_stride = hs; e->decoded_vstride = vs; e->decoded_layout_set = true; + } else if (hs != e->decoded_stride || vs != e->decoded_vstride) { + return fail("JPEG decoded layout changed", -1); + } + return 0; +} + +extern "C" int rkmpp_dma_encode(RkmppDmaEncoder *e, size_t index, size_t bytes_used, int fresh_fd, int64_t pts_us, + int force_idr, const uint8_t **data, size_t *size) { + if (!e || !e->ctx || index >= e->count || !data || !size) + return fail("invalid DMA encode call", -1); + *data = nullptr; *size = 0; + if (e->packet) mpp_packet_deinit(&e->packet); + auto abort_encode = [e](const char *op, int ret) { + fail(op, ret); e->close(); return -1; + }; + if (bytes_used > e->capacities[index] || + (e->jpeg ? (bytes_used < 4 || e->capacities[index] - bytes_used < 64) : bytes_used != e->minimum)) + return abort_encode("invalid DMA payload length", -1); + if (fresh_fd >= 0) { + MppBuffer replacement = nullptr; + MppBufferInfo info{}; + info.type = MPP_BUFFER_TYPE_EXT_DMA; info.fd = fresh_fd; + info.size = e->capacities[index]; info.index = static_cast(index); + const int ret = mpp_buffer_import(&replacement, &info); + if (ret) return abort_encode("refresh USB DMA import", ret); + if (e->buffers[index]) mpp_buffer_put(e->buffers[index]); + e->buffers[index] = replacement; + } + if (e->jpeg && decode_jpeg(e, index, bytes_used)) { + // Preserve the detailed decoder failure while ending BOTH engines. + e->close(); return -1; + } + if (force_idr) { + int ret = e->api->control(e->ctx, MPP_ENC_SET_IDR_FRAME, nullptr); + if (ret) return abort_encode("MPP_ENC_SET_IDR_FRAME", ret); + } + MppFrame frame = e->jpeg ? e->decoded_frame : nullptr; + int ret = 0; + if (!e->jpeg) { + ret = mpp_frame_init(&frame); + if (ret) return abort_encode("mpp_frame_init", ret); + mpp_frame_set_width(frame, e->width); mpp_frame_set_height(frame, e->height); + mpp_frame_set_hor_stride(frame, e->stride); mpp_frame_set_ver_stride(frame, e->height); + mpp_frame_set_fmt(frame, e->format); + mpp_frame_set_buffer(frame, e->buffers[index]); + } + mpp_frame_set_pts(frame, pts_us); + ret = e->api->encode_put_frame(e->ctx, frame); + if (!e->jpeg) mpp_frame_deinit(&frame); + if (ret) return abort_encode("encode_put_frame", ret); + ret = e->api->encode_get_packet(e->ctx, &e->packet); + if (ret || !e->packet) return abort_encode("encode_get_packet", ret); + if (mpp_packet_is_partition(e->packet) || !mpp_packet_get_length(e->packet)) + return abort_encode("incomplete DMA encoder output", -1); + // One synchronous input, no temporal scalability/reordering/split output. + // A completed packet is the input-consumption barrier for this mode. + *data = static_cast(mpp_packet_get_pos(e->packet)); + *size = mpp_packet_get_length(e->packet); + return 0; +} +extern "C" void rkmpp_dma_free(RkmppDmaEncoder *e) { delete e; } +#else +extern "C" RkmppDmaEncoder *rkmpp_dma_new(int,int,int,int,int,int,int,int,const int*,const size_t*,size_t) { + fail("RKMPP DMA support not built", -1); return nullptr; +} +extern "C" int rkmpp_dma_encode(RkmppDmaEncoder*,size_t,size_t,int,int64_t,int,const uint8_t**,size_t*) { return -1; } +extern "C" int rkmpp_dma_reconfigure(RkmppDmaEncoder*,int,int) { return -1; } +extern "C" void rkmpp_dma_free(RkmppDmaEncoder*) {} +#endif +extern "C" const char *rkmpp_dma_error(void) { return dma_error; } diff --git a/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf_ffi.h b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf_ffi.h new file mode 100644 index 00000000..a72ef612 --- /dev/null +++ b/libs/hwcodec/cpp/ffmpeg_hw/rkmpp_dmabuf_ffi.h @@ -0,0 +1,31 @@ +#pragma once +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct RkmppDmaEncoder RkmppDmaEncoder; +// format: 0=NV12, 1=BGR24, 2=YUYV, 3=RGB24 (byte stride), 4=MJPEG (stride ignored). +// codec: 0=H264, 1=HEVC. MJPEG is decoded to hardware NV12, then encoded. +RkmppDmaEncoder *rkmpp_dma_new(int width, int height, int stride, int format, + int codec, int fps, int kbps, int gop, + const int *fds, const size_t *sizes, size_t count); +// Synchronous input-completion boundary. On failure the encoder is destroyed +// internally BEFORE returning, so it cannot keep reading the capture buffer. +// Output is borrowed until the next call or destruction; copy before reusing it. +// bytes_used must be the actual captured payload. MJPEG needs 64 bytes of +// readable allocation headroom; never pass sizeimage as the compressed length. +// fresh_fd=-1 reuses the original import (native HDMI). UVC supplies a new +// export of this dequeued slot. Keep it open until replacement/free; the old +// export can be closed AFTER this call, including on failure. +int rkmpp_dma_encode(RkmppDmaEncoder *encoder, size_t index, size_t bytes_used, int fresh_fd, int64_t pts_us, + int force_idr, const uint8_t **data, size_t *size); +int rkmpp_dma_reconfigure(RkmppDmaEncoder *encoder, int kbps, int gop); +void rkmpp_dma_free(RkmppDmaEncoder *encoder); +const char *rkmpp_dma_error(void); + +#ifdef __cplusplus +} +#endif diff --git a/libs/hwcodec/src/lib.rs b/libs/hwcodec/src/lib.rs index 9a57fa75..b1e01c37 100644 --- a/libs/hwcodec/src/lib.rs +++ b/libs/hwcodec/src/lib.rs @@ -5,6 +5,11 @@ pub mod ffmpeg; #[cfg(any(target_arch = "aarch64", target_arch = "arm", feature = "rkmpp"))] pub mod ffmpeg_hw; pub mod ffmpeg_ram; +#[cfg(all( + target_os = "linux", + any(target_arch = "aarch64", target_arch = "arm", feature = "rkmpp") +))] +pub mod rkmpp_dmabuf; #[no_mangle] pub extern "C" fn hwcodec_log(level: i32, message: *const std::os::raw::c_char) { diff --git a/libs/hwcodec/src/rkmpp_dmabuf.rs b/libs/hwcodec/src/rkmpp_dmabuf.rs new file mode 100644 index 00000000..58a02394 --- /dev/null +++ b/libs/hwcodec/src/rkmpp_dmabuf.rs @@ -0,0 +1,193 @@ +//! Synchronous RKMPP encoder for pre-exported V4L2 DMA buffers. +//! Unlike the byte-slice encoder this never reads raw pixels on the CPU. + +use std::ffi::{c_char, c_int, c_void, CStr}; +use std::os::fd::{AsRawFd, OwnedFd}; +use std::ptr::NonNull; + +unsafe extern "C" { + fn rkmpp_dma_new( + width: c_int, + height: c_int, + stride: c_int, + format: c_int, + codec: c_int, + fps: c_int, + kbps: c_int, + gop: c_int, + fds: *const c_int, + sizes: *const usize, + count: usize, + ) -> *mut c_void; + fn rkmpp_dma_encode( + encoder: *mut c_void, + index: usize, + bytes_used: usize, + fresh_fd: c_int, + pts_us: i64, + force_idr: c_int, + data: *mut *const u8, + size: *mut usize, + ) -> c_int; + fn rkmpp_dma_reconfigure(encoder: *mut c_void, kbps: c_int, gop: c_int) -> c_int; + fn rkmpp_dma_free(encoder: *mut c_void); + fn rkmpp_dma_error() -> *const c_char; +} + +#[derive(Debug, Clone, Copy)] +#[repr(i32)] +pub enum DmaFormat { + Nv12 = 0, + Bgr24 = 1, + Yuyv = 2, + Rgb24 = 3, + Mjpeg = 4, +} + +pub struct DmaEncoderConfig { + pub width: u32, + pub height: u32, + pub stride: u32, + pub format: DmaFormat, + pub hevc: bool, + pub fps: u32, + pub bitrate_kbps: u32, + pub gop: u32, +} + +pub struct DmaEncoder { + ctx: NonNull, + // Export FDs remain open until AFTER mpp_destroy and imported buffer release. + _buffers: Vec<(OwnedFd, usize)>, + // Keep refreshed exports alive until native replacement/release has ended + // all references to the previous import. At most one FD per capture slot. + fresh_buffers: Vec>, +} + +// Exclusive ownership: the context can move between threads but all calls are sequential. +unsafe impl Send for DmaEncoder {} + +fn last_error() -> String { + unsafe { + CStr::from_ptr(rkmpp_dma_error()) + .to_string_lossy() + .into_owned() + } +} + +impl DmaEncoder { + pub fn new(config: DmaEncoderConfig, buffers: Vec<(OwnedFd, usize)>) -> Result { + let fds: Vec<_> = buffers.iter().map(|(fd, _)| fd.as_raw_fd()).collect(); + let sizes: Vec<_> = buffers.iter().map(|(_, size)| *size).collect(); + for value in [ + config.width, + config.height, + config.stride, + config.fps, + config.bitrate_kbps, + config.gop, + ] { + if value > c_int::MAX as u32 { + return Err("DMA encoder parameter overflow".into()); + } + } + let ptr = unsafe { + rkmpp_dma_new( + config.width as _, + config.height as _, + config.stride as _, + config.format as c_int, + config.hevc as _, + config.fps as _, + config.bitrate_kbps as _, + config.gop as _, + fds.as_ptr(), + sizes.as_ptr(), + buffers.len(), + ) + }; + Ok(Self { + ctx: NonNull::new(ptr).ok_or_else(last_error)?, + fresh_buffers: (0..buffers.len()).map(|_| None).collect(), + _buffers: buffers, + }) + } + + /// # Safety + /// The indexed buffer must be dequeued and exclusively leased to this call. + /// Do not requeue/write it until this function returns. On failure native MPP + /// is synchronously destroyed before returning, ending all input access. + /// `bytes_used` is the actual DQBUF payload length, not the buffer capacity. + /// A refreshed FD, if supplied, must refer to the same leased capture slot + /// with the capacity registered at construction. Ownership is retained here. + pub unsafe fn encode( + &mut self, + index: usize, + bytes_used: usize, + fresh_fd: Option, + pts_ms: i64, + force_idr: bool, + ) -> Result, String> { + let mut data = std::ptr::null(); + let mut size = 0; + if index >= self.fresh_buffers.len() { + return Err("Invalid DMA capture index".into()); + } + let ret = unsafe { + rkmpp_dma_encode( + self.ctx.as_ptr(), + index, + bytes_used, + fresh_fd.as_ref().map_or(-1, AsRawFd::as_raw_fd), + pts_ms.saturating_mul(1000), + force_idr as _, + &mut data, + &mut size, + ) + }; + if fresh_fd.is_some() { + // Native has now released the previous import, or destroyed both + // hardware contexts on error. Only now may its old FD be closed. + self.fresh_buffers[index] = fresh_fd; + } + if ret != 0 { + return Err(last_error()); + } + if data.is_null() || size == 0 { + return Err("Empty RKMPP DMA packet".into()); + } + // Copy only the compressed output, releasing the driver's packet promptly + // regardless of how long a network subscriber retains its Bytes. + Ok(unsafe { std::slice::from_raw_parts(data, size) }.to_vec()) + } + + pub fn reconfigure(&mut self, kbps: u32, gop: u32) -> Result<(), String> { + if kbps > c_int::MAX as u32 || gop > c_int::MAX as u32 { + return Err("DMA encoder parameter overflow".into()); + } + if unsafe { rkmpp_dma_reconfigure(self.ctx.as_ptr(), kbps as _, gop as _) } != 0 { + return Err(last_error()); + } + Ok(()) + } +} + +impl Drop for DmaEncoder { + fn drop(&mut self) { + unsafe { rkmpp_dma_free(self.ctx.as_ptr()) }; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn format_discriminants_match_native_abi() { + assert_eq!(DmaFormat::Nv12 as c_int, 0); + assert_eq!(DmaFormat::Bgr24 as c_int, 1); + assert_eq!(DmaFormat::Yuyv as c_int, 2); + assert_eq!(DmaFormat::Rgb24 as c_int, 3); + assert_eq!(DmaFormat::Mjpeg as c_int, 4); + } +} diff --git a/src/video/capture/dmabuf_layout.rs b/src/video/capture/dmabuf_layout.rs new file mode 100644 index 00000000..2a2caceb --- /dev/null +++ b/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 { + 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, +) -> 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); + } +} diff --git a/src/video/capture/linux.rs b/src/video/capture/linux.rs index 6ae0cc00..ad6e7d5b 100644 --- a/src/video/capture/linux.rs +++ b/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, native_hdmirx_state: Option, native_hdmirx_next_state_check: Option, + #[cfg(any(target_arch = "aarch64", target_arch = "arm"))] + dma_layout_bytes: Option, } fn open_capture_device(path: &Path) -> io::Result { @@ -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, - ) -> std::result::Result { + fn dequeue_buffer(&mut self) -> std::result::Result { 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> { + 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( + &mut self, + consume: impl FnOnce(usize, usize, Option) -> 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, + ) -> std::result::Result { + 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 io::Result<()>>(Option); + +#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))] +impl io::Result<()>> BufferReturn { + fn finish(mut self) -> io::Result<()> { + self.0.take().expect("capture lease already returned")() + } +} + +#[cfg(any(test, target_arch = "aarch64", target_arch = "arm"))] +impl io::Result<()>> Drop for BufferReturn { + 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 diff --git a/src/video/capture/status.rs b/src/video/capture/status.rs index c6b91939..0238a8b2 100644 --- a/src/video/capture/status.rs +++ b/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!( diff --git a/src/video/pipeline/dmabuf.rs b/src/video/pipeline/dmabuf.rs new file mode 100644 index 00000000..0fc9fba7 --- /dev/null +++ b/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 { + 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), +} + +// Field order is intentional, including during unwinding: destroy the encoder +// and its imported FDs before STREAMOFF/unmap/REQBUFS(0). +struct ActiveCapture { + encoder: Option, + 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); +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, + 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, + initial: ActiveCapture, + mut config: SharedVideoPipelineConfig, + device: std::path::PathBuf, + buffer_count: u32, + bridge: BridgeContext, + mut commands: mpsc::UnboundedReceiver, +) -> 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 = 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(()) +} diff --git a/src/video/pipeline/shared.rs b/src/video/pipeline/shared.rs index 22d4a70e..1269b43d 100644 --- a/src/video/pipeline/shared.rs +++ b/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, + 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) { + 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 = match open_capture_stream( + #[allow(unused_mut)] + let mut preopened: Option = 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);