diff --git a/README.md b/README.md index ec1a4999..9c7680d7 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,8 @@ flowchart LR - 描述信息、帧及其 payload 均为不可变对象，可以安全地跨协议共享； - 环形队列有界，慢消费者不会阻塞生产者； - 被覆盖的帧会形成精确的 dropped count，协议层据此标记 discontinuity； +- gRPC RGB 实时流会在积压或帧龄超过配置阈值时主动清空旧帧，避免慢客户端形成 + 数秒 FIFO 延迟； - H.264/H.265 出现丢帧或 generation 变化后，协议层可通过 Hub 请求新的关键帧； - source `start` 必须响应 cancellation，source `stop` 必须同步解除回调并回收生产线程。 @@ -418,7 +420,7 @@ cmvr_es.pb.txt | [`cmvr_es.pb.txt`](cmvr-es/config/cmvr_es.pb.txt) | 根配置，引用日志、设备管理和任务管理配置 | | [`device_manager.pb.txt`](cmvr-es/config/manager/device_manager.pb.txt) | 设备实例、类别、配置文件和启用状态 | | [`task_manager.pb.txt`](cmvr-es/config/manager/task_manager.pb.txt) | gRPC、QUIC、触屏等任务及运行模式 | -| [`grpc_server_task.pb.txt`](cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt) | gRPC 地址、端口和 reflection | +| [`grpc_server_task.pb.txt`](cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt) | gRPC 地址、端口、reflection，以及相机流积压/帧龄上限 | | [`quic_edge_task.pb.txt`](cmvr-es/config/tasks/quic_edge_task/quic_edge_task.pb.txt) | QUIC 网关、TLS、重连、心跳和媒体轨道 | | [`logger.pb.txt`](cmvr-es/config/logger/logger.pb.txt) | 日志级别、路由、目录和轮转参数 | @@ -430,6 +432,10 @@ cmvr_es.pb.txt 日志配置中的相对 `directory` 是相对于可执行文件目录解析，而不是配置根目录。默认值 `../log` 对应 `output/log/` 或部署后的 `/opt/cmvr-es/log/`。 +配置 TextFormat 不具备“新配置给旧二进制读取”的前向兼容性。本次新增的 gRPC +相机流参数需要与新二进制一起部署；不要只更新 +`grpc_server_task.pb.txt` 而继续运行旧的 `output/bin/cmvr_es`。 + ### 6.2 默认无硬件开发配置 仓库默认配置适用于当前主机没有连接实际设备的场景： @@ -580,6 +586,7 @@ ctest --test-dir build --output-on-failure | --- | --- | | `hikvision_camera_callback_test` | Hikvision 回调生命周期和并发安全 | | `media_source_hub_test` | 多订阅者、环形队列、丢帧和停止语义 | +| `grpc_camera_stream_policy_test` | gRPC 相机流积压阈值、帧龄判断和默认值 | | `quic_edge_protocol_test` | QUIC 控制帧、DATAGRAM 和 fake transport | | `quic_edge_task_test` | QUIC 任务配置和生命周期 | @@ -972,6 +979,8 @@ H.265、AAC 等格式的浏览器支持度并不统一。平台不能假设收 - 不复用已经发布字段的 tag； - 删除字段时使用 `reserved`； - 优先增加可选字段，不随意改变现有语义； +- `FrameData` 新增的采集时间、源序列、PTS/DTS、SDK 帧率/时间戳/帧号字段保持 + wire compatibility；Java 平台需重新生成 protobuf/gRPC 代码后才能读取新字段； - gRPC API 和 QUIC v1 线协议分别维护版本兼容； - 修改 QUIC DATAGRAM 固定头时必须提升协议版本并同步平台网关； - 提交前重新生成并验证 C++、Java 代码； diff --git a/cmvr-es/common/base/ring_buffer.h b/cmvr-es/common/base/ring_buffer.h index 32e0479c..6aa393b5 100644 --- a/cmvr-es/common/base/ring_buffer.h +++ b/cmvr-es/common/base/ring_buffer.h @@ -276,6 +276,40 @@ public: return tryReadLocked_(cursor); } + // Low-latency consumers can use this before reading to abandon an excessive + // backlog atomically. When the number of currently readable entries exceeds + // maximum_pending_frames, every pending entry is discarded and the cursor is + // advanced to the next sequence that will be published. Frames already + // overwritten by the ring and frames actively discarded here are both + // reflected in Cursor::dropped_count; the next successful read reports their + // sum through ReadResult::dropped_since_last_read. + uint64_t discardPendingIfExceeds( + Cursor& cursor, + const size_t maximum_pending_frames) const { + std::lock_guard lock(mutex_); + synchronizeCursorGenerationLocked_(cursor); + + if (!entries_.empty()) { + const uint64_t oldest_sequence = entries_.front().sequence; + if (cursor.next_sequence < oldest_sequence) { + cursor.dropped_count += oldest_sequence - cursor.next_sequence; + cursor.next_sequence = oldest_sequence; + } + } + + const uint64_t pending = + cursor.next_sequence < next_sequence_ + ? next_sequence_ - cursor.next_sequence + : 0; + if (pending <= static_cast(maximum_pending_frames)) { + return 0; + } + + cursor.next_sequence = next_sequence_; + cursor.dropped_count += pending; + return pending; + } + template std::optional waitRead( Cursor& cursor, diff --git a/cmvr-es/common/media/media_frame.h b/cmvr-es/common/media/media_frame.h index a5896b04..b81e8345 100644 --- a/cmvr-es/common/media/media_frame.h +++ b/cmvr-es/common/media/media_frame.h @@ -172,6 +172,10 @@ public: TrackDescriptorPtr descriptor; Payload payload; uint64_t sequence{0}; + // Opaque producer-native timing/counter values. Their units and epoch + // are source-defined; zero means that the source did not provide them. + uint64_t source_timestamp{0}; + uint64_t source_frame_number{0}; int64_t pts{0}; int64_t dts{0}; int64_t duration{0}; @@ -185,6 +189,8 @@ public: : descriptor(std::move(config.descriptor)), payload(std::make_shared(std::move(config.payload))), sequence(config.sequence), + source_timestamp(config.source_timestamp), + source_frame_number(config.source_frame_number), pts(config.pts), dts(config.dts), duration(config.duration), @@ -212,6 +218,8 @@ public: const TrackDescriptorPtr descriptor; const PayloadPtr payload; const uint64_t sequence; + const uint64_t source_timestamp; + const uint64_t source_frame_number; const int64_t pts; const int64_t dts; const int64_t duration; diff --git a/cmvr-es/config/README.md b/cmvr-es/config/README.md index 778d3c2f..c1414757 100644 --- a/cmvr-es/config/README.md +++ b/cmvr-es/config/README.md @@ -76,6 +76,24 @@ cmvr_es.pb.txt - QUIC 需要 TaskManager 与 `QuicEdgeConfig.enable` 同时开启； - QUIC 零媒体轨道是合法配置。 +### gRPC 相机实时流 + +[`tasks/grpc_server_task/grpc_server_task.pb.txt`](tasks/grpc_server_task/grpc_server_task.pb.txt) +中的两个低延迟参数仅作用于 gRPC RGB 编码流，不改变机械臂、AGV 等控制 RPC： + +- `camera_stream_max_pending_frames`：单个客户端允许的待发送帧数，超过后清空该 + 客户端积压；默认 2； +- `camera_stream_max_frame_age_ms`：从设备回调进入边缘系统起计算的最大帧龄， + 超过后不再发送；默认 250 ms。 + +两个字段填 0 或旧配置未包含字段时使用默认值。丢弃 H.264/H.265 帧后服务会请求 +IDR 并等待关键帧恢复。如果现场采集、编码本身稳定超过 250 ms，应根据日志中的 +`age_ms` 调高帧龄阈值，而不是增大环形队列。 + +新二进制可以读取未包含这两个字段的旧配置；旧二进制不能解析包含新字段的 +TextFormat。部署时必须同步更新程序与配置，不能只把新版 +`grpc_server_task.pb.txt` 复制给旧的 `output/bin/cmvr_es`。 + ## 配置验证 构建后可以用 `protoc --encode` 对单个 TextFormat 文件做语法和字段验证。例如： diff --git a/cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt b/cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt index 86a70d57..9001afb0 100644 --- a/cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt +++ b/cmvr-es/config/tasks/grpc_server_task/grpc_server_task.pb.txt @@ -3,4 +3,6 @@ grpc_server { host: "0.0.0.0" port: "50052" enable_reflection: true + camera_stream_max_pending_frames: 2 + camera_stream_max_frame_age_ms: 250 } diff --git a/cmvr-es/devices/README.md b/cmvr-es/devices/README.md index 646eaa18..23b14e70 100644 --- a/cmvr-es/devices/README.md +++ b/cmvr-es/devices/README.md @@ -238,6 +238,9 @@ CameraDeviceConfig / AGVDeviceConfig / ... 的外层 id - `sequence` - `capture_monotonic_ns` - `capture_utc_ns` +- `source_timestamp`：设备 SDK 提供的原始时间戳；单位和时钟域由设备定义，未知 + 时保持 0，不能直接当成 Unix 时间； +- `source_frame_number`：设备 SDK 提供的原始帧号，未知时保持 0； - `pts`、`dts` - `time_base_num`、`time_base_den` - `duration` @@ -247,6 +250,10 @@ CameraDeviceConfig / AGVDeviceConfig / ... 的外层 id H.264/H.265 后端必须识别关键帧，并尽量实现 `requestKeyFrame()`。 +Hikvision 后端优先采用 SDK 回调中的有效帧率，并保留 SDK 的 64 位原始时间戳和帧号； +SDK 帧率无效时才回退到配置的 `fps`。这些字段用于跨层诊断，设备层不应在不了解 +SDK 时钟语义时擅自换算。 + ### Microphone 完整音频包 应正确填写： diff --git a/cmvr-es/devices/camera/abstract_camera.h b/cmvr-es/devices/camera/abstract_camera.h index 8b5b8a36..dfde90cc 100644 --- a/cmvr-es/devices/camera/abstract_camera.h +++ b/cmvr-es/devices/camera/abstract_camera.h @@ -45,6 +45,10 @@ namespace cmvr::device { // when a complete encoded access unit is published. uint64_t stream_epoch = 0; uint64_t sequence = 0; + // Opaque producer-native timing/counter values. Their units and epoch + // are source-defined; zero means that the source did not provide them. + uint64_t source_timestamp = 0; + uint64_t source_frame_number = 0; int64_t capture_monotonic_ns = 0; int64_t capture_utc_ns = 0; int64_t pts = 0; diff --git a/cmvr-es/devices/camera/hikvision_camera/include/hikvision_camera.h b/cmvr-es/devices/camera/hikvision_camera/include/hikvision_camera.h index ff438cd6..0fd6e27f 100644 --- a/cmvr-es/devices/camera/hikvision_camera/include/hikvision_camera.h +++ b/cmvr-es/devices/camera/hikvision_camera/include/hikvision_camera.h @@ -45,7 +45,11 @@ public: unsigned char* buffer, unsigned int buffer_size, unsigned int width, - unsigned int height); + unsigned int height, + uint64_t source_timestamp, + uint64_t source_frame_number, + unsigned int source_frame_rate, + unsigned int source_packet_mode); private: bool initSdk_(); @@ -62,7 +66,11 @@ private: unsigned int buffer_size, bool is_key_frame, unsigned int width, - unsigned int height); + unsigned int height, + uint64_t source_timestamp, + uint64_t source_frame_number, + unsigned int source_frame_rate, + unsigned int source_packet_mode); void resetStreamState_(); config::HikvisionCameraConfig camera_; diff --git a/cmvr-es/devices/camera/hikvision_camera/src/hikvision_camera.cpp b/cmvr-es/devices/camera/hikvision_camera/src/hikvision_camera.cpp index 1645a602..e2d92821 100644 --- a/cmvr-es/devices/camera/hikvision_camera/src/hikvision_camera.cpp +++ b/cmvr-es/devices/camera/hikvision_camera/src/hikvision_camera.cpp @@ -28,6 +28,7 @@ int g_sdk_ref_count = 0; bool g_sdk_initialized = false; std::atomic g_ignored_data_type_log_count{0}; std::atomic g_es_video_log_count{0}; +std::atomic g_unexpected_packet_mode_log_count{0}; constexpr DWORD kHikvisionPacketFileHeader = 0; constexpr DWORD kHikvisionPacketVideoIFrame = 1; @@ -224,12 +225,19 @@ void CALLBACK hikvisionEsRealPlayCallback( if (!camera || !packet_info) { return; } + const uint64_t source_timestamp = + (static_cast(packet_info->dwTimeStampHigh) << 32U) | + static_cast(packet_info->dwTimeStamp); camera->onEsData(real_handle, packet_info->dwPacketType, packet_info->pPacketBuffer, packet_info->dwPacketSize, packet_info->wWidth, - packet_info->wHeight); + packet_info->wHeight, + source_timestamp, + packet_info->dwFrameNum, + packet_info->dwFrameRate, + packet_info->dwPacketMode); } } // namespace @@ -665,7 +673,11 @@ void HikvisionCamera::onEsData( unsigned char* buffer, const unsigned int buffer_size, const unsigned int packet_width, - const unsigned int packet_height) + const unsigned int packet_height, + const uint64_t source_timestamp, + const uint64_t source_frame_number, + const unsigned int source_frame_rate, + const unsigned int source_packet_mode) { if (!buffer || buffer_size == 0) { return; @@ -712,7 +724,11 @@ void HikvisionCamera::onEsData( const int video_log_count = g_es_video_log_count.fetch_add(1); if (video_log_count < 10) { CMVR_LOG(INFO) << "[HikvisionCamera] ES video packet_type=" - << packet_type << ", size=" << buffer_size; + << packet_type << ", size=" << buffer_size + << ", source_timestamp=" << source_timestamp + << ", source_frame_number=" << source_frame_number + << ", source_frame_rate=" << source_frame_rate + << ", source_packet_mode=" << source_packet_mode; } if (!callback_publishing_enabled_) { @@ -732,7 +748,11 @@ void HikvisionCamera::onEsData( buffer_size, is_key_frame, packet_width, - packet_height); + packet_height, + source_timestamp, + source_frame_number, + source_frame_rate, + source_packet_mode); } void HikvisionCamera::pushEncodedFrame_( @@ -740,7 +760,11 @@ void HikvisionCamera::pushEncodedFrame_( unsigned int buffer_size, bool is_key_frame, unsigned int packet_width, - unsigned int packet_height) + unsigned int packet_height, + uint64_t source_timestamp, + uint64_t source_frame_number, + unsigned int source_frame_rate, + unsigned int source_packet_mode) { if (!buffer || buffer_size == 0 || !stream_frame_buffer_) { return; @@ -751,12 +775,17 @@ void HikvisionCamera::pushEncodedFrame_( const auto capture_utc = std::chrono::system_clock::now(); frame_data.rgbFrame.assign(buffer, buffer + buffer_size); frame_data.codec = codec_; - frame_data.fps = fps_; + frame_data.fps = + source_frame_rate >= 1U && source_frame_rate <= 1000U + ? static_cast(source_frame_rate) + : fps_; frame_data.width = packet_width > 0 ? static_cast(packet_width) : width_; frame_data.height = packet_height > 0 ? static_cast(packet_height) : height_; frame_data.bKey = is_key_frame; frame_data.stream_epoch = stream_epoch_; frame_data.sequence = stream_sequence_++; + frame_data.source_timestamp = source_timestamp; + frame_data.source_frame_number = source_frame_number; frame_data.capture_monotonic_ns = std::chrono::duration_cast( capture_monotonic.time_since_epoch()).count(); frame_data.capture_utc_ns = std::chrono::duration_cast( @@ -764,13 +793,21 @@ void HikvisionCamera::pushEncodedFrame_( frame_data.pts = static_cast(frame_data.sequence); frame_data.dts = frame_data.pts; frame_data.time_base_num = 1; - frame_data.time_base_den = std::max(1, fps_); + frame_data.time_base_den = std::max(1, frame_data.fps); frame_data.duration = 1; frame_data.codec_config_generation = codec_config_generation_; // Hikvision labels packet type 0 as a "file header", but the bundled SDK // does not guarantee that it is a decoder-ready VPS/SPS/PPS blob. Keep it // only for change detection until its format is verified on real hardware. fillIntrinsics_(frame_data.intrinsics); + if (source_packet_mode > 1U) { + const int log_count = g_unexpected_packet_mode_log_count.fetch_add(1); + if (log_count < 5) { + CMVR_LOG(WARNING) << "[HikvisionCamera] unexpected ES source_packet_mode=" + << source_packet_mode + << ", source_frame_number=" << source_frame_number; + } + } stream_frame_buffer_->push(frame_data); } diff --git a/cmvr-es/devices/camera/hikvision_camera/tests/hikvision_camera_callback_test.cpp b/cmvr-es/devices/camera/hikvision_camera/tests/hikvision_camera_callback_test.cpp index 07ba6524..1aa72441 100644 --- a/cmvr-es/devices/camera/hikvision_camera/tests/hikvision_camera_callback_test.cpp +++ b/cmvr-es/devices/camera/hikvision_camera/tests/hikvision_camera_callback_test.cpp @@ -49,7 +49,12 @@ void emitEsPacket( const LONG real_handle, std::vector payload, const WORD width = 640, - const WORD height = 360) + const WORD height = 360, + const DWORD timestamp_low = 0, + const DWORD timestamp_high = 0, + const DWORD frame_number = 0, + const DWORD frame_rate = 0, + const DWORD packet_mode = 0) { EsDataCallback callback = nullptr; void* user = nullptr; @@ -66,18 +71,36 @@ void emitEsPacket( NET_DVR_PACKET_INFO_EX packet{}; packet.wWidth = width; packet.wHeight = height; + packet.dwTimeStamp = timestamp_low; + packet.dwTimeStampHigh = timestamp_high; + packet.dwFrameNum = frame_number; + packet.dwFrameRate = frame_rate; packet.dwPacketType = packet_type; packet.dwPacketSize = static_cast(payload.size()); packet.pPacketBuffer = payload.data(); + packet.dwPacketMode = packet_mode; callback(real_handle, &packet, user); } -void emitIFrame(const LONG real_handle = g_real_handle) +void emitIFrame( + const LONG real_handle = g_real_handle, + const DWORD timestamp_low = 0, + const DWORD timestamp_high = 0, + const DWORD frame_number = 0, + const DWORD frame_rate = 0, + const DWORD packet_mode = 0) { emitEsPacket( kVideoIFrame, real_handle, - {0x00, 0x00, 0x00, 0x01, 0x65, 0x88, 0x84, 0x21}); + {0x00, 0x00, 0x00, 0x01, 0x65, 0x88, 0x84, 0x21}, + 640, + 360, + timestamp_low, + timestamp_high, + frame_number, + frame_rate, + packet_mode); } void emitPFrame(const LONG real_handle = g_real_handle) @@ -142,7 +165,13 @@ bool testCallbackPublicationLifecycle() CHECK_TRUE(!camera.waitEncodedFrame( frame, cursor, std::chrono::milliseconds(10))); - emitIFrame(); + emitIFrame( + g_real_handle, + 0x89ABCDEFU, + 0x01234567U, + 42U, + 30U, + 1U); CHECK_TRUE(camera.waitEncodedFrame( frame, cursor, std::chrono::milliseconds(50))); CHECK_TRUE(frame.stream_epoch == 1); @@ -151,6 +180,10 @@ bool testCallbackPublicationLifecycle() CHECK_TRUE(frame.bKey); CHECK_TRUE(frame.width == 640); CHECK_TRUE(frame.height == 360); + CHECK_TRUE(frame.source_timestamp == 0x0123456789ABCDEFULL); + CHECK_TRUE(frame.source_frame_number == 42U); + CHECK_TRUE(frame.fps == 30); + CHECK_TRUE(frame.time_base_den == 30); CHECK_TRUE(camera.requestKeyFrame()); CHECK_TRUE(g_key_frame_request_count.load() == 2); @@ -177,12 +210,22 @@ bool testCallbackPublicationLifecycle() emitPFrame(); CHECK_TRUE(!camera.waitEncodedFrame( frame, cursor, std::chrono::milliseconds(10))); - emitIFrame(); + emitIFrame( + g_real_handle, + 0x76543210U, + 0xFEDCBA98U, + 99U, + 1001U, + 0U); CHECK_TRUE(camera.waitEncodedFrame( frame, cursor, std::chrono::milliseconds(50))); CHECK_TRUE(frame.sequence == 1); CHECK_TRUE(frame.codec_config_generation == 2); CHECK_TRUE(frame.codec_config.empty()); + CHECK_TRUE(frame.source_timestamp == 0xFEDCBA9876543210ULL); + CHECK_TRUE(frame.source_frame_number == 99U); + CHECK_TRUE(frame.fps == 25); + CHECK_TRUE(frame.time_base_den == 25); constexpr int kConcurrentCallbacks = 8; std::vector producers; diff --git a/cmvr-es/manager/README.md b/cmvr-es/manager/README.md index f261195d..21e3ac31 100644 --- a/cmvr-es/manager/README.md +++ b/cmvr-es/manager/README.md @@ -177,12 +177,16 @@ Descriptor 最低要求： - `NEXT_PUBLISHED` 忽略已有帧； - `OLDEST_AVAILABLE` 从最旧保留帧开始； - `LATEST_AVAILABLE` 读取当前最新帧； +- `discardPendingIfExceeds(limit)` 在积压超过阈值时原子地把该消费者游标推进到 + 当前发布末尾，并把主动丢弃数量计入下一次读取的 `dropped_since_last_read`； +- `discardPendingIfExceeds()` 与 `read()` 一样只能由该 Subscription 的单一消费 + 线程调用，不能用它替代 Subscription 的线程所有权约束； - source 重启会 reset ring、提升 `BroadcastFrameRing` 内部 generation，并把 ring 的 `ReadResult.sequence` 从 0 重新计数； - close 唤醒等待者并拒绝新 publish。 ring generation 不等于 `TrackDescriptor::generation`，ring 的 `ReadResult.sequence` 也不等于 `MediaFrame::sequence`。Descriptor generation 和媒体帧 sequence 仍由 producer/adapter 维护。 -协议消费者需要分别处理 ring `generation_changed`、descriptor generation、消费者 drop 和 frame discontinuity；任一不连续发生时都应传播状态，帧间编码还应请求关键帧。 +协议消费者需要分别处理 ring `generation_changed`、descriptor generation、消费者 drop 和 frame discontinuity；任一不连续发生时都应传播状态，帧间编码还应请求关键帧。主动丢弃 H.264/H.265 积压后不得直接发送 P/B 帧，必须等新的关键帧恢复。 ## 新增第四种 Manager diff --git a/cmvr-es/manager/media_source_hub/include/media_source_hub.h b/cmvr-es/manager/media_source_hub/include/media_source_hub.h index 21dfc8f3..c7dcb553 100644 --- a/cmvr-es/manager/media_source_hub/include/media_source_hub.h +++ b/cmvr-es/manager/media_source_hub/include/media_source_hub.h @@ -71,6 +71,7 @@ public: std::optional tryRead(); std::optional waitRead(std::chrono::milliseconds timeout); + uint64_t discardPendingIfExceeds(size_t maximum_pending_frames); uint64_t droppedCount() const noexcept; void reset(); diff --git a/cmvr-es/manager/media_source_hub/src/device_media_source_adapter.cpp b/cmvr-es/manager/media_source_hub/src/device_media_source_adapter.cpp index b1c4de7d..36b21653 100644 --- a/cmvr-es/manager/media_source_hub/src/device_media_source_adapter.cpp +++ b/cmvr-es/manager/media_source_hub/src/device_media_source_adapter.cpp @@ -423,6 +423,8 @@ struct CameraPump final : PumpState { frame.descriptor = std::move(descriptor); frame.payload = std::move(source.rgbFrame); frame.sequence = source.sequence; + frame.source_timestamp = source.source_timestamp; + frame.source_frame_number = source.source_frame_number; frame.pts = source.pts; frame.dts = source.dts; frame.duration = source.duration; diff --git a/cmvr-es/manager/media_source_hub/src/media_source_hub.cpp b/cmvr-es/manager/media_source_hub/src/media_source_hub.cpp index 6eebab28..cace2727 100644 --- a/cmvr-es/manager/media_source_hub/src/media_source_hub.cpp +++ b/cmvr-es/manager/media_source_hub/src/media_source_hub.cpp @@ -405,6 +405,14 @@ std::optional MediaSourceHub::Subscription::wai return source_->ring.waitRead(cursor_, timeout); } +uint64_t MediaSourceHub::Subscription::discardPendingIfExceeds( + const size_t maximum_pending_frames) { + if (!active_ || !source_) { + return 0; + } + return source_->ring.discardPendingIfExceeds(cursor_, maximum_pending_frames); +} + uint64_t MediaSourceHub::Subscription::droppedCount() const noexcept { return cursor_.dropped_count; } diff --git a/cmvr-es/manager/media_source_hub/tests/media_source_hub_test.cpp b/cmvr-es/manager/media_source_hub/tests/media_source_hub_test.cpp index fdb60f73..b96cea9b 100644 --- a/cmvr-es/manager/media_source_hub/tests/media_source_hub_test.cpp +++ b/cmvr-es/manager/media_source_hub/tests/media_source_hub_test.cpp @@ -188,6 +188,130 @@ void testBroadcastFrameRing() { CHECK_TRUE(!ring.publish(makeFrame(descriptor, 5, 60)).has_value()); } +void testBroadcastDiscardPending() { + using Ring = BroadcastFrameRing; + const auto descriptor = makeVideoDescriptor(Codec::H264, 1, {1, 2, 3}); + + { + Ring ring(8); + auto cursor = ring.makeCursor(Ring::StartPosition::OLDEST_AVAILABLE); + ring.publish(makeFrame(descriptor, 0, 10)); + ring.publish(makeFrame(descriptor, 1, 20)); + + CHECK_TRUE(ring.discardPendingIfExceeds(cursor, 2) == 0); + const auto first = ring.tryRead(cursor); + CHECK_TRUE(first.has_value()); + CHECK_TRUE(first->sequence == 0); + CHECK_TRUE(first->dropped_since_last_read == 0); + } + + { + Ring ring(8); + auto cursor = ring.makeCursor(Ring::StartPosition::OLDEST_AVAILABLE); + ring.publish(makeFrame(descriptor, 0, 10)); + ring.publish(makeFrame(descriptor, 1, 20)); + ring.publish(makeFrame(descriptor, 2, 30)); + + CHECK_TRUE(ring.discardPendingIfExceeds(cursor, 2) == 3); + CHECK_TRUE(!ring.tryRead(cursor).has_value()); + + ring.publish(makeFrame(descriptor, 3, 40)); + const auto after_discard = ring.tryRead(cursor); + CHECK_TRUE(after_discard.has_value()); + CHECK_TRUE(after_discard->sequence == 3); + CHECK_TRUE(after_discard->dropped_count == 3); + CHECK_TRUE(after_discard->dropped_since_last_read == 3); + } + + { + // Two frames are overwritten before the explicit three-frame discard. + // Both kinds of loss must be reported by the next successful read. + Ring ring(3); + auto cursor = ring.makeCursor(Ring::StartPosition::OLDEST_AVAILABLE); + for (uint64_t sequence = 0; sequence < 5; ++sequence) { + ring.publish(makeFrame( + descriptor, + sequence, + static_cast(sequence))); + } + + CHECK_TRUE(ring.discardPendingIfExceeds(cursor, 2) == 3); + CHECK_TRUE(cursor.dropped_count == 5); + ring.publish(makeFrame(descriptor, 5, 50)); + const auto after_overwrite_and_discard = ring.tryRead(cursor); + CHECK_TRUE(after_overwrite_and_discard.has_value()); + CHECK_TRUE(after_overwrite_and_discard->sequence == 5); + CHECK_TRUE(after_overwrite_and_discard->dropped_count == 5); + CHECK_TRUE(after_overwrite_and_discard->dropped_since_last_read == 5); + } + + { + // An old-generation OLDEST_AVAILABLE cursor adopts the reset generation + // before deciding whether that generation's pending frames are excessive. + Ring ring(4); + auto cursor = ring.makeCursor(Ring::StartPosition::OLDEST_AVAILABLE); + ring.publish(makeFrame(descriptor, 0, 10)); + ring.reset(); + ring.publish(makeFrame(descriptor, 1, 20)); + ring.publish(makeFrame(descriptor, 2, 30)); + + CHECK_TRUE(ring.discardPendingIfExceeds(cursor, 1) == 2); + ring.publish(makeFrame(descriptor, 3, 40)); + const auto after_reset = ring.tryRead(cursor); + CHECK_TRUE(after_reset.has_value()); + CHECK_TRUE(after_reset->generation == 2); + CHECK_TRUE(after_reset->sequence == 2); + CHECK_TRUE(after_reset->generation_changed); + CHECK_TRUE(after_reset->dropped_since_last_read == 2); + } +} + +void testBroadcastDiscardConcurrentPublish() { + using Ring = BroadcastFrameRing; + constexpr uint64_t frame_count = 4000; + Ring ring(64); + auto cursor = ring.makeCursor(Ring::StartPosition::NEXT_PUBLISHED); + std::atomic start{false}; + std::atomic publisher_done{false}; + + std::thread publisher([&] { + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + for (uint64_t sequence = 0; sequence < frame_count; ++sequence) { + ring.publish(std::make_shared(static_cast(sequence))); + if ((sequence & 7U) == 0U) { + std::this_thread::yield(); + } + } + publisher_done.store(true, std::memory_order_release); + }); + + uint64_t read_count = 0; + uint64_t actively_discarded = 0; + start.store(true, std::memory_order_release); + while (true) { + actively_discarded += ring.discardPendingIfExceeds(cursor, 8); + if (ring.tryRead(cursor)) { + ++read_count; + continue; + } + if (publisher_done.load(std::memory_order_acquire)) { + actively_discarded += ring.discardPendingIfExceeds(cursor, 8); + if (ring.tryRead(cursor)) { + ++read_count; + continue; + } + break; + } + std::this_thread::yield(); + } + publisher.join(); + + CHECK_TRUE(read_count + cursor.dropped_count == frame_count); + CHECK_TRUE(actively_discarded <= cursor.dropped_count); +} + void testBroadcastConcurrency() { using Ring = BroadcastFrameRing; constexpr uint64_t frame_count = 500; @@ -301,6 +425,48 @@ void testHubLifecycleAndDescriptorRefresh() { CHECK_TRUE(!hub.hasSource(initial_descriptor->id)); } +void testSubscriptionDiscardPending() { + MediaSourceHub hub; + const auto descriptor = makeVideoDescriptor(Codec::H264, 1); + std::mutex sink_mutex; + MediaSourceHub::FrameSink sink; + + MediaSourceHub::SourceCallbacks callbacks; + callbacks.start = [&](const MediaSourceHub::FrameSink& callback_sink, + const MediaSourceHub::CancelPredicate&) { + std::lock_guard lock(sink_mutex); + sink = callback_sink; + return true; + }; + callbacks.stop = [&] { + std::lock_guard lock(sink_mutex); + sink = {}; + }; + + CHECK_TRUE(hub.registerSource(descriptor, std::move(callbacks), 8)); + auto subscription = hub.subscribe(descriptor->id); + CHECK_TRUE(subscription.valid()); + + MediaSourceHub::FrameSink producer; + { + std::lock_guard lock(sink_mutex); + producer = sink; + } + CHECK_TRUE(static_cast(producer)); + producer(makeFrame(descriptor, 0, 10)); + producer(makeFrame(descriptor, 1, 20)); + producer(makeFrame(descriptor, 2, 30)); + + CHECK_TRUE(subscription.discardPendingIfExceeds(2) == 3); + CHECK_TRUE(!subscription.tryRead().has_value()); + producer(makeFrame(descriptor, 3, 40)); + const auto next = subscription.tryRead(); + CHECK_TRUE(next.has_value()); + CHECK_TRUE(next->value->sequence == 3); + CHECK_TRUE(next->dropped_since_last_read == 3); + CHECK_TRUE(subscription.droppedCount() == 3); +} + void testHubFailedStartAndShutdown() { MediaSourceHub hub; const auto descriptor = makeVideoDescriptor(Codec::UNKNOWN, 1); @@ -498,8 +664,11 @@ int main() { testMediaMetadataValidation(); testLegacySpmcCompatibility(); testBroadcastFrameRing(); + testBroadcastDiscardPending(); + testBroadcastDiscardConcurrentPublish(); testBroadcastConcurrency(); testHubLifecycleAndDescriptorRefresh(); + testSubscriptionDiscardPending(); testHubFailedStartAndShutdown(); testKeyFrameRequestIsOrderedBeforeStop(); testHubCancelsBlockedStartWithoutBlockingShutdown(); diff --git a/cmvr-es/service/CMakeLists.txt b/cmvr-es/service/CMakeLists.txt index 940f7f3d..6f6862af 100644 --- a/cmvr-es/service/CMakeLists.txt +++ b/cmvr-es/service/CMakeLists.txt @@ -29,6 +29,21 @@ target_link_libraries(service PRIVATE add_library(cmvr_es::service ALIAS service) install(TARGETS service LIBRARY DESTINATION lib) +if(BUILD_TESTING) + add_executable(grpc_camera_stream_policy_test + grpc/tests/grpc_camera_stream_policy_test.cpp + ) + target_include_directories(grpc_camera_stream_policy_test + PRIVATE + ${CMAKE_SOURCE_DIR}/cmvr-es + ) + add_test( + NAME grpc_camera_stream_policy_test + COMMAND grpc_camera_stream_policy_test + ) + set_tests_properties(grpc_camera_stream_policy_test PROPERTIES TIMEOUT 10) +endif() + # -------------------------------------------------------- # Unit test # -------------------------------------------------------- diff --git a/cmvr-es/service/README.md b/cmvr-es/service/README.md index 6802b950..6c6800a2 100644 --- a/cmvr-es/service/README.md +++ b/cmvr-es/service/README.md @@ -107,10 +107,22 @@ cmvr::api::ExampleService::Service - 不持有设备状态锁进行网络写； - 为 wait/read 使用有限 timeout； - 慢客户端不能阻塞设备生产线程； -- H.264/H.265 丢帧后等待关键帧恢复。 +- H.264/H.265 丢帧后等待关键帧恢复； +- gRPC RGB 流在积压超过 `camera_stream_max_pending_frames` 或帧龄超过 + `camera_stream_max_frame_age_ms` 时主动丢弃旧帧，请求 IDR，并从下一个关键帧恢复。 当前仅 gRPC RGB 和麦克风流使用 MediaSourceHub；Depth/RGBD 仍直接读取设备帧。 +gRPC 相机实时流默认最多保留 2 帧积压、最大允许 250 ms 帧龄。两个配置项填 0 +时使用上述默认值。该策略以低延迟为目标，不保证每个视频帧都到达客户端；控制命令 +仍由普通 gRPC RPC 承担。 + +`FrameData` 附带 `capture_utc_ns`、`source_sequence`、`pts`、`dts`、 +`source_fps`、`source_timestamp` 和 `source_frame_number`。平台端可用采集时间 +与接收时间的差值区分设备、网络、服务端写阻塞和客户端解码/渲染队列延迟。新增字段 +保持 protobuf wire compatibility，旧客户端可以继续连接，但需要重新生成代码后才能 +读取这些诊断字段。 + ### 当前安全状态 GrpcServerTask 使用同步 `grpc::ServerBuilder` 和 `grpc::InsecureServerCredentials()`。reflection 由配置控制。当前没有 gRPC TLS、认证、授权或标准 health service。 diff --git a/cmvr-es/service/grpc/include/grpc_camera_service.h b/cmvr-es/service/grpc/include/grpc_camera_service.h index b9be651d..62e72277 100644 --- a/cmvr-es/service/grpc/include/grpc_camera_service.h +++ b/cmvr-es/service/grpc/include/grpc_camera_service.h @@ -9,12 +9,14 @@ #include "common/base/grpc_utils.h" #include "manager/device_manager/include/device_manager.h" #include "devices/camera/abstract_camera.h" +#include "service/grpc/include/grpc_camera_stream_policy.h" namespace cmvr::service { class gRPCCameraServiceImpl final: public api::CameraService::Service { public: - gRPCCameraServiceImpl(); + explicit gRPCCameraServiceImpl( + CameraStreamLowLatencyConfig stream_config = {}); ~gRPCCameraServiceImpl() override = default; grpc::Status GetStatus(grpc::ServerContext* context, const api::GetCameraStateCommand_Request* request, api::GetCameraStateCommand_Feedback* response) override; grpc::Status StartCamera(grpc::ServerContext* context, const api::StartCameraCommand_Request* request, api::StartCameraCommand_Feedback* response) override; @@ -30,6 +32,7 @@ namespace cmvr::service { grpc::Status GetRGBImageStream(grpc::ServerContext* context, grpc::ServerReaderWriter* stream) override; private: device::DeviceManager& dmgr_; + CameraStreamLowLatencyConfig stream_config_; //双向流读写线程 std::shared_ptr read_thread_ = nullptr; diff --git a/cmvr-es/service/grpc/include/grpc_camera_stream_policy.h b/cmvr-es/service/grpc/include/grpc_camera_stream_policy.h new file mode 100644 index 00000000..cf5a1f16 --- /dev/null +++ b/cmvr-es/service/grpc/include/grpc_camera_stream_policy.h @@ -0,0 +1,60 @@ +#ifndef CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H +#define CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H + +#pragma once + +#include +#include +#include +#include + +namespace cmvr::service { + +inline constexpr size_t kDefaultCameraStreamMaxPendingFrames = 2; +inline constexpr uint32_t kDefaultCameraStreamMaxFrameAgeMs = 250; + +struct CameraStreamLowLatencyConfig { + size_t max_pending_frames{kDefaultCameraStreamMaxPendingFrames}; + std::chrono::milliseconds max_frame_age{ + kDefaultCameraStreamMaxFrameAgeMs}; +}; + +inline CameraStreamLowLatencyConfig makeCameraStreamLowLatencyConfig( + const uint32_t max_pending_frames, + const uint32_t max_frame_age_ms) noexcept { + CameraStreamLowLatencyConfig config; + config.max_pending_frames = max_pending_frames == 0 + ? kDefaultCameraStreamMaxPendingFrames + : static_cast(max_pending_frames); + config.max_frame_age = std::chrono::milliseconds( + max_frame_age_ms == 0 + ? kDefaultCameraStreamMaxFrameAgeMs + : max_frame_age_ms); + return config; +} + +inline std::optional cameraFrameAgeNs( + const uint64_t capture_time_ns, + const uint64_t now_ns) noexcept { + if (capture_time_ns == 0 || now_ns < capture_time_ns) { + return std::nullopt; + } + return now_ns - capture_time_ns; +} + +inline bool cameraFrameExceedsAgeLimit( + const uint64_t capture_time_ns, + const uint64_t now_ns, + const std::chrono::milliseconds max_frame_age) noexcept { + const auto age_ns = cameraFrameAgeNs(capture_time_ns, now_ns); + if (!age_ns || max_frame_age.count() <= 0) { + return false; + } + const auto max_age_ns = std::chrono::duration_cast( + max_frame_age).count(); + return *age_ns > static_cast(max_age_ns); +} + +} // namespace cmvr::service + +#endif // CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H diff --git a/cmvr-es/service/grpc/src/grpc_camera_service.cpp b/cmvr-es/service/grpc/src/grpc_camera_service.cpp index 15cd5871..5ca371ef 100644 --- a/cmvr-es/service/grpc/src/grpc_camera_service.cpp +++ b/cmvr-es/service/grpc/src/grpc_camera_service.cpp @@ -102,7 +102,10 @@ private: }; } -gRPCCameraServiceImpl::gRPCCameraServiceImpl(): dmgr_(DeviceManager::getInstance()) {} +gRPCCameraServiceImpl::gRPCCameraServiceImpl( + CameraStreamLowLatencyConfig stream_config) + : dmgr_(DeviceManager::getInstance()), + stream_config_(stream_config) {} grpc::Status gRPCCameraServiceImpl::GetStatus(grpc::ServerContext* context, const api::GetCameraStateCommand_Request* request, api::GetCameraStateCommand_Feedback* response) @@ -629,7 +632,9 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte return grpc::Status::OK; } string dev_id = request.header().device_id(); - CMVR_LOG(INFO) << "[gRPCCameraServiceImpl] (GetRGBImageStream): start,id=" << dev_id; + CMVR_LOG(INFO) << "[gRPCCameraServiceImpl] (GetRGBImageStream): start,id=" << dev_id + << ", max_pending_frames=" << stream_config_.max_pending_frames + << ", max_frame_age_ms=" << stream_config_.max_frame_age.count(); const auto dev = dmgr_.getDevice(dev_id); if (!dev) { api::GetRGBImageStreamCommand_Feedback response; @@ -664,6 +669,9 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte bool waiting_for_key_frame = true; auto last_key_frame_request = std::chrono::steady_clock::now(); + auto last_latency_log = std::chrono::steady_clock::time_point{}; + uint64_t discarded_since_log = 0; + std::chrono::microseconds last_write_duration{0}; media_hub.requestKeyFrame(track_id); const auto request_key_frame_if_due = [&] { const auto now = std::chrono::steady_clock::now(); @@ -672,6 +680,37 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte last_key_frame_request = now; } }; + const auto request_key_frame_now = [&] { + waiting_for_key_frame = true; + media_hub.requestKeyFrame(track_id); + last_key_frame_request = std::chrono::steady_clock::now(); + }; + const auto log_latency_event = [&]( + const char* reason, + const uint64_t discarded, + const uint64_t frame_age_ns, + const std::chrono::microseconds write_duration) { + discarded_since_log += discarded; + const auto now = std::chrono::steady_clock::now(); + if (last_latency_log != std::chrono::steady_clock::time_point{} && + now - last_latency_log < std::chrono::seconds(1)) { + return; + } + const double frame_age_ms = static_cast(frame_age_ns) / 1'000'000.0; + const double write_ms = static_cast(write_duration.count()) / 1'000.0; + CMVR_LOG(WARNING) << "[gRPCCameraServiceImpl] low-latency camera stream event" + << ", id=" << dev_id + << ", reason=" << reason + << ", discarded=" << discarded_since_log + << ", age_ms=" << frame_age_ms + << ", write_ms=" << write_ms + << ", max_pending_frames=" + << stream_config_.max_pending_frames + << ", max_frame_age_ms=" + << stream_config_.max_frame_age.count(); + discarded_since_log = 0; + last_latency_log = now; + }; while (true) { if (context->IsCancelled()) @@ -695,6 +734,28 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte if (!descriptor) { continue; } + const uint64_t now_ns = static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count()); + const auto frame_age = cameraFrameAgeNs(frame.capture_time_ns, now_ns); + if (frame_age && + cameraFrameExceedsAgeLimit( + frame.capture_time_ns, + now_ns, + stream_config_.max_frame_age)) { + // This frame is already outside the latency budget. Flush all + // currently queued frames and wait for a fresh IDR; sending any + // P/B frame after an intentional gap would break decoder continuity. + const uint64_t discarded = + 1 + subscription.discardPendingIfExceeds(0); + request_key_frame_now(); + log_latency_event( + "stale_frame", + discarded, + *frame_age, + last_write_duration); + continue; + } const bool inter_frame_codec = descriptor->codec == cmvr::media::Codec::H264 || descriptor->codec == cmvr::media::Codec::H265; if (!inter_frame_codec || descriptor->payload_format != cmvr::media::PayloadFormat::ANNEX_B) { @@ -707,9 +768,7 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte break; } if (read->dropped_since_last_read > 0 || read->generation_changed || frame.discontinuity) { - waiting_for_key_frame = true; - media_hub.requestKeyFrame(track_id); - last_key_frame_request = std::chrono::steady_clock::now(); + request_key_frame_now(); } if (waiting_for_key_frame && !frame.key_frame) { request_key_frame_if_due(); @@ -727,6 +786,13 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte descriptor->codec == cmvr::media::Codec::H265 ? "h265" : "unknown"); response.mutable_color_frame()->set_width(static_cast(descriptor->width)); response.mutable_color_frame()->set_height(static_cast(descriptor->height)); + response.mutable_color_frame()->set_capture_utc_ns(frame.capture_utc_ns); + response.mutable_color_frame()->set_source_sequence(frame.sequence); + response.mutable_color_frame()->set_pts(frame.pts); + response.mutable_color_frame()->set_dts(frame.dts); + response.mutable_color_frame()->set_source_fps(descriptor->nominal_rate); + response.mutable_color_frame()->set_source_timestamp(frame.source_timestamp); + response.mutable_color_frame()->set_source_frame_number(frame.source_frame_number); response.mutable_intrinsics()->set_fx(descriptor->fx); response.mutable_intrinsics()->set_fy(descriptor->fy); @@ -740,10 +806,28 @@ grpc::Status gRPCCameraServiceImpl::GetRGBImageStream(grpc::ServerContext* conte frame.sequence, static_cast(std::numeric_limits::max())))); + const auto write_started = std::chrono::steady_clock::now(); if (!stream->Write(response)) { CMVR_LOG(INFO) << "[gRPCCameraServiceImpl] (stream->Write) failed,id=" << dev_id; break; } + last_write_duration = std::chrono::duration_cast( + std::chrono::steady_clock::now() - write_started); + + // A successful synchronous Write may have been flow-controlled long + // enough for the source to outpace this consumer. Once the pending + // count crosses the configured trigger, discard the whole pending + // batch and require a fresh key frame before resuming. + const uint64_t discarded = subscription.discardPendingIfExceeds( + stream_config_.max_pending_frames); + if (discarded > 0) { + request_key_frame_now(); + log_latency_event( + "write_backpressure", + discarded, + frame_age.value_or(0), + last_write_duration); + } } CMVR_LOG(INFO) << "[gRPCCameraServiceImpl] (GetRGBImageStream): end,id=" << dev_id; return grpc::Status::OK; diff --git a/cmvr-es/service/grpc/tests/grpc_camera_stream_policy_test.cpp b/cmvr-es/service/grpc/tests/grpc_camera_stream_policy_test.cpp new file mode 100644 index 00000000..1d84b942 --- /dev/null +++ b/cmvr-es/service/grpc/tests/grpc_camera_stream_policy_test.cpp @@ -0,0 +1,54 @@ +#include "service/grpc/include/grpc_camera_stream_policy.h" + +#include +#include +#include + +namespace { + +bool check(const bool condition, const char* expression, const int line) { + if (condition) { + return true; + } + std::cerr << "CHECK failed at line " << line << ": " << expression << '\n'; + return false; +} + +#define CHECK_TRUE(expression) \ + do { \ + if (!check(static_cast(expression), #expression, __LINE__)) { \ + return 1; \ + } \ + } while (false) + +} // namespace + +int main() { + using namespace std::chrono_literals; + using cmvr::service::cameraFrameAgeNs; + using cmvr::service::cameraFrameExceedsAgeLimit; + using cmvr::service::makeCameraStreamLowLatencyConfig; + + const auto defaults = makeCameraStreamLowLatencyConfig(0, 0); + CHECK_TRUE(defaults.max_pending_frames == 2); + CHECK_TRUE(defaults.max_frame_age == 250ms); + + const auto configured = makeCameraStreamLowLatencyConfig(7, 900); + CHECK_TRUE(configured.max_pending_frames == 7); + CHECK_TRUE(configured.max_frame_age == 900ms); + + CHECK_TRUE(!cameraFrameAgeNs(0, 1'000'000'000ULL).has_value()); + CHECK_TRUE(!cameraFrameAgeNs(2'000'000'000ULL, 1'000'000'000ULL).has_value()); + CHECK_TRUE(cameraFrameAgeNs(1'000'000'000ULL, 1'250'000'000ULL).value() == + 250'000'000ULL); + + CHECK_TRUE(!cameraFrameExceedsAgeLimit( + 1'000'000'000ULL, 1'250'000'000ULL, 250ms)); + CHECK_TRUE(cameraFrameExceedsAgeLimit( + 1'000'000'000ULL, 1'250'000'001ULL, 250ms)); + CHECK_TRUE(!cameraFrameExceedsAgeLimit( + 1'000'000'000ULL, 2'000'000'000ULL, 0ms)); + + std::cout << "grpc_camera_stream_policy_test: PASS\n"; + return 0; +} diff --git a/cmvr-es/task/grpc_server_task/src/grpc_server_task.cpp b/cmvr-es/task/grpc_server_task/src/grpc_server_task.cpp index dd48753a..60d8c697 100644 --- a/cmvr-es/task/grpc_server_task/src/grpc_server_task.cpp +++ b/cmvr-es/task/grpc_server_task/src/grpc_server_task.cpp @@ -95,7 +95,10 @@ bool GrpcServerTask::start() const std::string port = cfg_.port().empty() ? "50051" : cfg_.port(); const std::string local_address = host + ":" + port; - camera_service_ = std::make_unique(); + camera_service_ = std::make_unique( + service::makeCameraStreamLowLatencyConfig( + cfg_.camera_stream_max_pending_frames(), + cfg_.camera_stream_max_frame_age_ms())); system_service_ = std::make_unique(); speaker_service_ = std::make_unique(); microphone_service_ = std::make_unique(); diff --git a/protos/README.md b/protos/README.md index 2e75b513..fef77e4e 100644 --- a/protos/README.md +++ b/protos/README.md @@ -154,6 +154,11 @@ output/bin/protoc \ gRPC Java client 还需要 grpc-java plugin。当前 Proto 没有统一 `java_package` 和 `java_multiple_files`，增加这些 option 时应同步两端生成结果和平台 import。 +`cmvr.api.FrameData` 的字段 7–13 是可选的实时流诊断信息，包括边缘端采集 UTC +时间、源序列、PTS/DTS、SDK 帧率、SDK 原始时间戳和帧号。旧客户端会安全忽略 +这些字段；Java 平台需要重新生成 message 和 stub 才能读取它们。SDK 原始时间戳 +的单位和时钟域由设备定义，不能按 Unix 时间直接解释。 + 浏览器不能直接消费自定义 QUIC ALPN，不应从这些 Proto 推导“浏览器可直连 Edge”。 ## Review 与验证 diff --git a/protos/cmvr/api/camera_command.proto b/protos/cmvr/api/camera_command.proto index 487bac96..7305955a 100644 --- a/protos/cmvr/api/camera_command.proto +++ b/protos/cmvr/api/camera_command.proto @@ -19,6 +19,15 @@ message FrameData { FrameType type = 4; string codec = 5; bool is_key_frame = 6; + // Optional capture and source metadata. Fields 1-6 remain wire-compatible + // with existing clients; older clients safely ignore these additions. + int64 capture_utc_ns = 7; + uint64 source_sequence = 8; + int64 pts = 9; + int64 dts = 10; + uint32 source_fps = 11; + uint64 source_timestamp = 12; + uint64 source_frame_number = 13; } message CameraIntrinsics { @@ -205,4 +214,3 @@ message GetRGBDImagesStreamCommand { } - diff --git a/protos/cmvr/config/grpc_server_config/grpc_server_config.proto b/protos/cmvr/config/grpc_server_config/grpc_server_config.proto index 11ad8018..0650c735 100644 --- a/protos/cmvr/config/grpc_server_config/grpc_server_config.proto +++ b/protos/cmvr/config/grpc_server_config/grpc_server_config.proto @@ -6,6 +6,12 @@ message GRPCServerConfig { string port = 2; bool enable_reflection = 3; string id = 4; + // Bounds the continuity-oriented media queue before the gRPC stream skips + // ahead and waits for a new key frame. Zero uses the service default. + uint32 camera_stream_max_pending_frames = 5; + // Frames older than this monotonic age are not sent. Zero uses the service + // default so configurations written before these fields remain low-latency. + uint32 camera_stream_max_frame_age_ms = 6; } message GRPCServerRootConfig { GRPCServerConfig grpc_server = 1;