fix: bound gRPC camera stream latency
This commit is contained in:
parent
d40c92b9bf
commit
1181b2d541
11
README.md
11
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 代码；
|
||||
|
||||
@ -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<std::mutex> 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<uint64_t>(maximum_pending_frames)) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
cursor.next_sequence = next_sequence_;
|
||||
cursor.dropped_count += pending;
|
||||
return pending;
|
||||
}
|
||||
|
||||
template<class Rep, class Period>
|
||||
std::optional<ReadResult> waitRead(
|
||||
Cursor& cursor,
|
||||
|
||||
@ -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<const Payload>(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;
|
||||
|
||||
@ -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 文件做语法和字段验证。例如：
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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 完整音频包
|
||||
|
||||
应正确填写：
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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_;
|
||||
|
||||
@ -28,6 +28,7 @@ int g_sdk_ref_count = 0;
|
||||
bool g_sdk_initialized = false;
|
||||
std::atomic<int> g_ignored_data_type_log_count{0};
|
||||
std::atomic<int> g_es_video_log_count{0};
|
||||
std::atomic<int> 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<uint64_t>(packet_info->dwTimeStampHigh) << 32U) |
|
||||
static_cast<uint64_t>(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<int>(source_frame_rate)
|
||||
: fps_;
|
||||
frame_data.width = packet_width > 0 ? static_cast<int>(packet_width) : width_;
|
||||
frame_data.height = packet_height > 0 ? static_cast<int>(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<std::chrono::nanoseconds>(
|
||||
capture_monotonic.time_since_epoch()).count();
|
||||
frame_data.capture_utc_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
|
||||
@ -764,13 +793,21 @@ void HikvisionCamera::pushEncodedFrame_(
|
||||
frame_data.pts = static_cast<int64_t>(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);
|
||||
}
|
||||
|
||||
|
||||
@ -49,7 +49,12 @@ void emitEsPacket(
|
||||
const LONG real_handle,
|
||||
std::vector<BYTE> 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<DWORD>(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<std::thread> producers;
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -71,6 +71,7 @@ public:
|
||||
|
||||
std::optional<FrameReadResult> tryRead();
|
||||
std::optional<FrameReadResult> waitRead(std::chrono::milliseconds timeout);
|
||||
uint64_t discardPendingIfExceeds(size_t maximum_pending_frames);
|
||||
uint64_t droppedCount() const noexcept;
|
||||
void reset();
|
||||
|
||||
|
||||
@ -423,6 +423,8 @@ struct CameraPump final : PumpState<device::AbstractCamera> {
|
||||
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;
|
||||
|
||||
@ -405,6 +405,14 @@ std::optional<MediaSourceHub::FrameReadResult> 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;
|
||||
}
|
||||
|
||||
@ -188,6 +188,130 @@ void testBroadcastFrameRing() {
|
||||
CHECK_TRUE(!ring.publish(makeFrame(descriptor, 5, 60)).has_value());
|
||||
}
|
||||
|
||||
void testBroadcastDiscardPending() {
|
||||
using Ring = BroadcastFrameRing<MediaFrame>;
|
||||
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<uint8_t>(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<int>;
|
||||
constexpr uint64_t frame_count = 4000;
|
||||
Ring ring(64);
|
||||
auto cursor = ring.makeCursor(Ring::StartPosition::NEXT_PUBLISHED);
|
||||
std::atomic<bool> start{false};
|
||||
std::atomic<bool> 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<const int>(static_cast<int>(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<MediaFrame>;
|
||||
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<std::mutex> lock(sink_mutex);
|
||||
sink = callback_sink;
|
||||
return true;
|
||||
};
|
||||
callbacks.stop = [&] {
|
||||
std::lock_guard<std::mutex> 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<std::mutex> lock(sink_mutex);
|
||||
producer = sink;
|
||||
}
|
||||
CHECK_TRUE(static_cast<bool>(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();
|
||||
|
||||
@ -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
|
||||
# --------------------------------------------------------
|
||||
|
||||
@ -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。
|
||||
|
||||
@ -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<cmvr::api::GetRGBImageStreamCommand_Feedback, cmvr::api::GetRGBImageStreamCommand_Request>* stream) override;
|
||||
private:
|
||||
device::DeviceManager& dmgr_;
|
||||
CameraStreamLowLatencyConfig stream_config_;
|
||||
|
||||
//双向流读写线程
|
||||
std::shared_ptr<std::thread> read_thread_ = nullptr;
|
||||
|
||||
60
cmvr-es/service/grpc/include/grpc_camera_stream_policy.h
Normal file
60
cmvr-es/service/grpc/include/grpc_camera_stream_policy.h
Normal file
@ -0,0 +1,60 @@
|
||||
#ifndef CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H
|
||||
#define CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <chrono>
|
||||
#include <cstddef>
|
||||
#include <cstdint>
|
||||
#include <optional>
|
||||
|
||||
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<size_t>(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<uint64_t> 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<std::chrono::nanoseconds>(
|
||||
max_frame_age).count();
|
||||
return *age_ns > static_cast<uint64_t>(max_age_ns);
|
||||
}
|
||||
|
||||
} // namespace cmvr::service
|
||||
|
||||
#endif // CMVR_ES_GRPC_CAMERA_STREAM_POLICY_H
|
||||
@ -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<AbstractCamera>(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<double>(frame_age_ns) / 1'000'000.0;
|
||||
const double write_ms = static_cast<double>(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<uint64_t>(
|
||||
std::chrono::duration_cast<std::chrono::nanoseconds>(
|
||||
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<int32_t>(descriptor->width));
|
||||
response.mutable_color_frame()->set_height(static_cast<int32_t>(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<uint64_t>(std::numeric_limits<int32_t>::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::microseconds>(
|
||||
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;
|
||||
|
||||
@ -0,0 +1,54 @@
|
||||
#include "service/grpc/include/grpc_camera_stream_policy.h"
|
||||
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <iostream>
|
||||
|
||||
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<bool>(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;
|
||||
}
|
||||
@ -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<service::gRPCCameraServiceImpl>();
|
||||
camera_service_ = std::make_unique<service::gRPCCameraServiceImpl>(
|
||||
service::makeCameraStreamLowLatencyConfig(
|
||||
cfg_.camera_stream_max_pending_frames(),
|
||||
cfg_.camera_stream_max_frame_age_ms()));
|
||||
system_service_ = std::make_unique<service::gRPCSystemServiceImpl>();
|
||||
speaker_service_ = std::make_unique<service::gRPCSpeakerServiceImpl>();
|
||||
microphone_service_ = std::make_unique<service::gRPCMicroPhoneServiceImpl>();
|
||||
|
||||
@ -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 与验证
|
||||
|
||||
@ -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 {
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@ -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;
|
||||
|
||||
Loading…
Reference in New Issue
Block a user