Conversation
添加共享内存传输层实现，支持Windows命名共享内存和POSIX shm，使用环形缓冲区+原子序列号+内存屏障算法，提供高性能零拷贝帧传输功能。同时更新测试文件和开发路线图标记该功能为已完成。 主要功能包括： - 单生产者-多消费者模式 - 跨进程帧传输延迟<1μs - 吞吐量>10GB/s - 支持阻塞/非阻塞操作 - 内置统计和错误处理
|
✅ Review Complete! The code review has been posted. View Review → |
Reviewer's Guide实现了一个跨平台的零拷贝共享内存传输层，具备环形缓冲区语义、错误处理和统计信息功能，并包含全面的单元测试以及在路线图中的接线/集成。 零拷贝生产者与消费者交互的时序图sequenceDiagram
actor ProducerProcess
participant ProducerShmTransport as ShmTransport_producer
participant ProducerSegment as SharedMemorySegment_producer
participant Control as ShmControlBlock
participant ConsumerShmTransport as ShmTransport_consumer
participant ConsumerSegment as SharedMemorySegment_consumer
actor ConsumerProcess
ProducerProcess->>ProducerShmTransport: InitializeProducer(config)
ProducerShmTransport->>ProducerSegment: SharedMemorySegment::Create(name,size)
ProducerSegment-->>ProducerShmTransport: segment
ProducerShmTransport->>Control: initialize_control_block()
ProducerShmTransport-->>ProducerProcess: initialized
ConsumerProcess->>ConsumerShmTransport: InitializeConsumer(config)
ConsumerShmTransport->>ConsumerSegment: SharedMemorySegment::Open(name,size)
ConsumerSegment-->>ConsumerShmTransport: segment
ConsumerShmTransport->>Control: validate_control_block()
ConsumerShmTransport-->>ConsumerProcess: initialized
loop zero_copy_write_for_each_frame
ProducerProcess->>ProducerShmTransport: AcquireWriteBuffer(timeout)
alt buffer_available
ProducerShmTransport->>Control: is_writable()
ProducerShmTransport-->>ProducerProcess: WriteBuffer(data_ptr,capacity,index)
ProducerProcess->>ProducerShmTransport: CommitWriteBuffer(index,metadata,size)
ProducerShmTransport->>Control: write_sequence++ , stats_update
else timeout
ProducerShmTransport-->>ProducerProcess: nullopt (timeout)
end
end
loop zero_copy_read_for_each_consumer
ConsumerProcess->>ConsumerShmTransport: AcquireReadBuffer(timeout)
alt data_available
ConsumerShmTransport->>Control: has_readable_data()
ConsumerShmTransport-->>ConsumerProcess: ReadBuffer(metadata,span,index)
ConsumerProcess->>ConsumerShmTransport: ReleaseReadBuffer(index)
ConsumerShmTransport->>Control: read_sequence++ , stats_update
else timeout
ConsumerShmTransport-->>ConsumerProcess: nullopt (timeout)
end
end
共享内存传输核心类型的类图classDiagram
class ShmTransportError {
<<enum>>
+Success
+Unknown
+InvalidArgument
+OutOfMemory
+Timeout
+NotInitialized
+AlreadyExists
+NotFound
+PermissionDenied
+ShmCreateFailed
+ShmOpenFailed
+ShmMapFailed
+ShmUnmapFailed
+ShmTooSmall
+ShmCorrupted
+ShmVersionMismatch
+SyncCreateFailed
+SyncWaitFailed
+SyncSignalFailed
+SyncTimeout
+SyncAbandoned
+TransportClosed
+TransportBusy
+BufferOverflow
+BufferUnderflow
+FrameTooLarge
+InvalidFrame
}
class ShmFrameHeader {
<<struct>>
+uint32_t magic
+uint32_t version
+uint32_t sequence_number
+uint32_t frame_number
+uint32_t sequence_id
+uint32_t data_size
+uint64_t capture_timestamp_ns
+uint64_t write_timestamp_ns
+uint32_t width
+uint32_t height
+uint32_t stride
+uint32_t pixel_format
+uint32_t checksum
+uint32_t reserved
+bool is_valid()
+uint32_t calculate_checksum(data)
+bool verify_data(data)
}
class ShmControlBlock {
<<struct>>
+uint32_t magic
+uint32_t version
+uint32_t header_size
+uint32_t flags
+uint32_t buffer_count
+uint32_t max_frame_size
+uint32_t metadata_size
+uint32_t frame_stride
+atomic~uint64_t~ write_sequence
+atomic~uint64_t~ read_sequence
+atomic~uint64_t~ dropped_frames
+atomic~uint32_t~ state
+atomic~uint32_t~ active_readers
+atomic~uint32_t~ active_writers
+atomic~uint64_t~ total_frames_written
+atomic~uint64_t~ total_frames_read
+atomic~uint64_t~ total_bytes_written
+atomic~uint64_t~ total_bytes_read
+bool is_valid()
+uint32_t get_write_index()
+uint32_t get_read_index()
+bool has_readable_data()
+bool is_writable()
+uint32_t available_buffers()
+uint32_t used_buffers()
}
class ShmTransportConfig {
<<struct>>
+string shm_name
+size_t buffer_count
+size_t max_frame_size
+size_t metadata_size
+core::Duration write_timeout
+core::Duration read_timeout
+bool non_blocking
+bool use_cache_line_alignment
+bool prefetch_next_frame
+bool enable_zero_copy
+bool enable_checksum
+bool enable_stats
+bool enable_tracing
+size_t calculate_total_size()
+expected~void,ShmTransportError~ validate()
}
class SharedMemorySegment {
<<abstract>>
+static expected~unique_ptr~Create(name,size)
+static expected~unique_ptr~Open(name,size)
+byte* data()
+size_t size()
+string_view name()
+bool is_valid()
+Result flush()
}
class ShmTransportStats {
<<struct>>
+uint64_t frames_written
+uint64_t frames_read
+uint64_t frames_dropped
+uint64_t bytes_written
+uint64_t bytes_read
+core::Duration min_write_latency
+core::Duration max_write_latency
+core::Duration avg_write_latency
+core::Duration min_read_latency
+core::Duration max_read_latency
+core::Duration avg_read_latency
+uint64_t write_timeouts
+uint64_t read_timeouts
+uint64_t checksum_errors
+core::Timestamp session_start
+core::Timestamp last_write_time
+core::Timestamp last_read_time
+double get_drop_rate()
+uint64_t get_average_frame_size()
+double get_throughput_bytes_per_sec()
+void update_write_latency(latency)
+void update_read_latency(latency)
+void reset()
}
class ShmTransport {
<<class>>
-ShmTransportConfig config_
-unique_ptr~SharedMemorySegment~ segment_
-ShmControlBlock* control_block_
-byte* frame_data_base_
-atomic~bool~ initialized_
-bool is_producer_
-mutex stats_mutex_
-ShmTransportStats stats_
+ShmTransport()
+~ShmTransport()
+Result InitializeProducer(config)
+Result InitializeConsumer(config)
+Result Shutdown()
+bool IsInitialized()
+bool IsProducer()
+bool IsConsumer()
+Result WriteFrame(metadata,data)
+expected~bool,ShmTransportError~ TryWriteFrame(metadata,data)
+expected~bool,ShmTransportError~ WriteFrameWithTimeout(metadata,data,timeout)
+expected~optional~ReadFrame(core::Duration)
+expected~optional~TryReadFrame()
+Result ReadFrameWithCallback(timeout,callback)
+expected~optional~AcquireWriteBuffer(timeout)
+Result CommitWriteBuffer(buffer_index,metadata,actual_size)
+expected~optional~AcquireReadBuffer(timeout)
+Result ReleaseReadBuffer(buffer_index)
+expected~optional~TryAcquireWriteBuffer()
+expected~optional~TryAcquireReadBuffer()
+ShmTransportStats GetStats()
+pair~uint32_t,uint32_t~ GetBufferStatus()
+const ShmTransportConfig& GetConfig()
+const ShmControlBlock* GetControlBlock()
-void update_stats(is_write,bytes,latency)
-ShmFrameHeader* get_frame_header(index)
-byte* get_frame_buffer(index)
}
ShmTransport --> ShmTransportConfig : has
ShmTransport --> SharedMemorySegment : owns
ShmTransport --> ShmControlBlock : points_to
ShmTransport --> ShmFrameHeader : uses
ShmTransport --> ShmTransportStats : maintains
ShmTransportConfig --> ShmControlBlock : defines_layout_for
ShmControlBlock --> ShmFrameHeader : indexes_frames
ShmFrameHeader --> ShmTransportError : uses_for_validation
ShmTransportStats --> ShmControlBlock : aggregates_counters
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your Experience访问你的 dashboard 来：
Getting HelpOriginal review guide in EnglishReviewer's GuideImplements a cross-platform zero-copy shared memory transport layer with ring-buffer semantics, error handling, and statistics, plus comprehensive unit tests and roadmap wiring. Sequence diagram for zero-copy producer and consumer interactionsequenceDiagram
actor ProducerProcess
participant ProducerShmTransport as ShmTransport_producer
participant ProducerSegment as SharedMemorySegment_producer
participant Control as ShmControlBlock
participant ConsumerShmTransport as ShmTransport_consumer
participant ConsumerSegment as SharedMemorySegment_consumer
actor ConsumerProcess
ProducerProcess->>ProducerShmTransport: InitializeProducer(config)
ProducerShmTransport->>ProducerSegment: SharedMemorySegment::Create(name,size)
ProducerSegment-->>ProducerShmTransport: segment
ProducerShmTransport->>Control: initialize_control_block()
ProducerShmTransport-->>ProducerProcess: initialized
ConsumerProcess->>ConsumerShmTransport: InitializeConsumer(config)
ConsumerShmTransport->>ConsumerSegment: SharedMemorySegment::Open(name,size)
ConsumerSegment-->>ConsumerShmTransport: segment
ConsumerShmTransport->>Control: validate_control_block()
ConsumerShmTransport-->>ConsumerProcess: initialized
loop zero_copy_write_for_each_frame
ProducerProcess->>ProducerShmTransport: AcquireWriteBuffer(timeout)
alt buffer_available
ProducerShmTransport->>Control: is_writable()
ProducerShmTransport-->>ProducerProcess: WriteBuffer(data_ptr,capacity,index)
ProducerProcess->>ProducerShmTransport: CommitWriteBuffer(index,metadata,size)
ProducerShmTransport->>Control: write_sequence++ , stats_update
else timeout
ProducerShmTransport-->>ProducerProcess: nullopt (timeout)
end
end
loop zero_copy_read_for_each_consumer
ConsumerProcess->>ConsumerShmTransport: AcquireReadBuffer(timeout)
alt data_available
ConsumerShmTransport->>Control: has_readable_data()
ConsumerShmTransport-->>ConsumerProcess: ReadBuffer(metadata,span,index)
ConsumerProcess->>ConsumerShmTransport: ReleaseReadBuffer(index)
ConsumerShmTransport->>Control: read_sequence++ , stats_update
else timeout
ConsumerShmTransport-->>ConsumerProcess: nullopt (timeout)
end
end
Class diagram for shared memory transport core typesclassDiagram
class ShmTransportError {
<<enum>>
+Success
+Unknown
+InvalidArgument
+OutOfMemory
+Timeout
+NotInitialized
+AlreadyExists
+NotFound
+PermissionDenied
+ShmCreateFailed
+ShmOpenFailed
+ShmMapFailed
+ShmUnmapFailed
+ShmTooSmall
+ShmCorrupted
+ShmVersionMismatch
+SyncCreateFailed
+SyncWaitFailed
+SyncSignalFailed
+SyncTimeout
+SyncAbandoned
+TransportClosed
+TransportBusy
+BufferOverflow
+BufferUnderflow
+FrameTooLarge
+InvalidFrame
}
class ShmFrameHeader {
<<struct>>
+uint32_t magic
+uint32_t version
+uint32_t sequence_number
+uint32_t frame_number
+uint32_t sequence_id
+uint32_t data_size
+uint64_t capture_timestamp_ns
+uint64_t write_timestamp_ns
+uint32_t width
+uint32_t height
+uint32_t stride
+uint32_t pixel_format
+uint32_t checksum
+uint32_t reserved
+bool is_valid()
+uint32_t calculate_checksum(data)
+bool verify_data(data)
}
class ShmControlBlock {
<<struct>>
+uint32_t magic
+uint32_t version
+uint32_t header_size
+uint32_t flags
+uint32_t buffer_count
+uint32_t max_frame_size
+uint32_t metadata_size
+uint32_t frame_stride
+atomic~uint64_t~ write_sequence
+atomic~uint64_t~ read_sequence
+atomic~uint64_t~ dropped_frames
+atomic~uint32_t~ state
+atomic~uint32_t~ active_readers
+atomic~uint32_t~ active_writers
+atomic~uint64_t~ total_frames_written
+atomic~uint64_t~ total_frames_read
+atomic~uint64_t~ total_bytes_written
+atomic~uint64_t~ total_bytes_read
+bool is_valid()
+uint32_t get_write_index()
+uint32_t get_read_index()
+bool has_readable_data()
+bool is_writable()
+uint32_t available_buffers()
+uint32_t used_buffers()
}
class ShmTransportConfig {
<<struct>>
+string shm_name
+size_t buffer_count
+size_t max_frame_size
+size_t metadata_size
+core::Duration write_timeout
+core::Duration read_timeout
+bool non_blocking
+bool use_cache_line_alignment
+bool prefetch_next_frame
+bool enable_zero_copy
+bool enable_checksum
+bool enable_stats
+bool enable_tracing
+size_t calculate_total_size()
+expected~void,ShmTransportError~ validate()
}
class SharedMemorySegment {
<<abstract>>
+static expected~unique_ptr~Create(name,size)
+static expected~unique_ptr~Open(name,size)
+byte* data()
+size_t size()
+string_view name()
+bool is_valid()
+Result flush()
}
class ShmTransportStats {
<<struct>>
+uint64_t frames_written
+uint64_t frames_read
+uint64_t frames_dropped
+uint64_t bytes_written
+uint64_t bytes_read
+core::Duration min_write_latency
+core::Duration max_write_latency
+core::Duration avg_write_latency
+core::Duration min_read_latency
+core::Duration max_read_latency
+core::Duration avg_read_latency
+uint64_t write_timeouts
+uint64_t read_timeouts
+uint64_t checksum_errors
+core::Timestamp session_start
+core::Timestamp last_write_time
+core::Timestamp last_read_time
+double get_drop_rate()
+uint64_t get_average_frame_size()
+double get_throughput_bytes_per_sec()
+void update_write_latency(latency)
+void update_read_latency(latency)
+void reset()
}
class ShmTransport {
<<class>>
-ShmTransportConfig config_
-unique_ptr~SharedMemorySegment~ segment_
-ShmControlBlock* control_block_
-byte* frame_data_base_
-atomic~bool~ initialized_
-bool is_producer_
-mutex stats_mutex_
-ShmTransportStats stats_
+ShmTransport()
+~ShmTransport()
+Result InitializeProducer(config)
+Result InitializeConsumer(config)
+Result Shutdown()
+bool IsInitialized()
+bool IsProducer()
+bool IsConsumer()
+Result WriteFrame(metadata,data)
+expected~bool,ShmTransportError~ TryWriteFrame(metadata,data)
+expected~bool,ShmTransportError~ WriteFrameWithTimeout(metadata,data,timeout)
+expected~optional~ReadFrame(core::Duration)
+expected~optional~TryReadFrame()
+Result ReadFrameWithCallback(timeout,callback)
+expected~optional~AcquireWriteBuffer(timeout)
+Result CommitWriteBuffer(buffer_index,metadata,actual_size)
+expected~optional~AcquireReadBuffer(timeout)
+Result ReleaseReadBuffer(buffer_index)
+expected~optional~TryAcquireWriteBuffer()
+expected~optional~TryAcquireReadBuffer()
+ShmTransportStats GetStats()
+pair~uint32_t,uint32_t~ GetBufferStatus()
+const ShmTransportConfig& GetConfig()
+const ShmControlBlock* GetControlBlock()
-void update_stats(is_write,bytes,latency)
-ShmFrameHeader* get_frame_header(index)
-byte* get_frame_buffer(index)
}
ShmTransport --> ShmTransportConfig : has
ShmTransport --> SharedMemorySegment : owns
ShmTransport --> ShmControlBlock : points_to
ShmTransport --> ShmFrameHeader : uses
ShmTransport --> ShmTransportStats : maintains
ShmTransportConfig --> ShmControlBlock : defines_layout_for
ShmControlBlock --> ShmFrameHeader : indexes_frames
ShmFrameHeader --> ShmTransportError : uses_for_validation
ShmTransportStats --> ShmControlBlock : aggregates_counters
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
There was a problem hiding this comment.
📋 Summary
本 PR 实现了 L0 层的零拷贝共享内存传输层 (shm_transport)，完成了 v0.2.0-alpha.3 的传输层功能。代码质量高，实现了跨平台（Windows/POSIX）的共享内存抽象，使用无锁环形缓冲区算法，包含完整的单元测试和性能基准测试。
| File | Changes | Risk Level | Status |
|---|---|---|---|
shm_transport.hpp |
+942/-0 | 🟡 Medium | Added |
shm_transport.cpp |
+1213/-0 | 🟡 Medium | Added |
test_shm_transport.cpp |
+1188/-0 | 🟢 Low | Added |
CMakeLists.txt |
+1/-0 | 🟢 Low | Modified |
ROADMAP.md |
+1/-1 | 🟢 Low | Modified |
🏗️ Architecture Flow
flowchart LR
subgraph L0["L0 Sensing Layer"]
A[Frame Buffer] -->|Zero-Copy| B[ShmTransport]
B -->|Shared Memory| C[ShmControlBlock]
C -->|Ring Buffer| D[Consumer Process]
end
style B fill:#f9f,stroke:#333,stroke-width:2px
📝 ROADMAP Update Reminder
本 PR 实现了 v0.2.0-alpha.3 的共享内存传输功能。已在 ROADMAP.md 中标记为完成 [x]。请确保在合并前验证 ROADMAP-S.md 中的相应测试标准是否已满足（Valgrind 内存检查、ThreadSanitizer 等）。
⚠️ Warnings (Should Fix)
-
[File:
core/src/l0_sensing/shm_transport.cpp, Line: 46] 使用了using namespace aam::core;，虽然是在 .cpp 文件中，但建议避免使用using namespace以防止命名空间污染。 -
[File:
core/tests/test_shm_transport.cpp, Lines: 863-898] 存在重复的章节注释块（"零拷贝延迟基准测试" 出现了两次），建议清理重复的注释。 -
[File:
core/src/l0_sensing/shm_transport.cpp, End of file] 文件末尾缺少换行符（\n）。
💡 Suggestions
-
性能优化:
WriteFrameWithTimeout和ReadFrame中的忙等待使用了std::this_thread::yield()，在高争用场景下可能退化为自旋锁。考虑未来使用条件变量或事件通知机制降低 CPU 占用。 -
错误处理:
ShmRemove在 Windows 平台上为空操作（仅返回 true），建议在注释中明确说明 Windows 下共享内存的生命周期由最后一个句柄控制。 -
测试覆盖: 建议添加以下边界情况测试：
- 共享内存版本不匹配时的行为
- 进程崩溃后共享内存的清理测试
- 多消费者并发读取的正确性验证
📍 Inline Comments
Open Questions
- Windows 权限: Windows 版本的共享内存使用了默认的 NULL DACL（
CreateFileMappingA第二个参数为nullptr），这意味着任何用户都可以访问。是否需要添加 ACL 配置以限制访问权限？ - 大端序支持: 当前实现假设小端序（x86/x64/ARM），如果未来需要支持大端序架构（如某些嵌入式 MIPS），帧头部的魔数检查和字段布局是否需要调整？
🤖 AI Agent Prompt Generation
🤖 Copy this prompt for AI Agent
Please address the comments from this code review for the shared memory transport implementation:
## Overall Comments
1. **Remove `using namespace` directives**: In `shm_transport.cpp` line 46 and `test_shm_transport.cpp` line 46, remove `using namespace aam::core;` and use explicit namespace prefixes instead.
2. **Add missing newline**: Add a newline at the end of `shm_transport.cpp` (currently missing).
3. **Clean up duplicate comments**: In `test_shm_transport.cpp`, remove the duplicate section comment blocks around lines 863-870 (the "零拷贝延迟基准测试" section appears twice).
## File-specific Changes
### File: core/src/l0_sensing/shm_transport.cpp
- Line 46: Remove `using namespace aam::core;` and prefix `Clock`, `Duration`, `Timestamp` with `aam::core::`
- End of file: Add newline character
### File: core/tests/test_shm_transport.cpp
- Line 38-39: Remove `using namespace aam::core;` and `using namespace aam::l0;` or replace with specific using declarations
- Lines 863-870: Remove duplicate comment block
No functional changes are required - these are all code style and formatting fixes.
Thank you!统计: 0 🚨 | 5
This review was generated by ArknightsAutoMachine AI Reviewer
| { | ||
|
|
||
| // 使用 core 命名空间 | ||
| using namespace aam::core; |
There was a problem hiding this comment.
using namespace aam::core;，建议显式使用命名空间前缀或仅引入需要的符号，以防止命名冲突。
| return {}; | ||
| } | ||
|
|
||
| } // namespace aam::l0 No newline at end of file |
There was a problem hiding this comment.
|
|
||
| // ========================================================================== | ||
| // 零拷贝高性能基准测试 | ||
| // ========================================================================== |
There was a problem hiding this comment.
| // ========================================================================== | ||
| // 测试夹具 | ||
| // ========================================================================== | ||
|
|
There was a problem hiding this comment.
using namespace 使用，建议遵循项目编码规范。
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - 我发现了 9 个问题，并给出了一些总体反馈：
ShmTransportConfig::enable_zero_copy标志目前未被使用，而且普通的WriteFrame路径会无视enable_checksum始终计算校验和；建议把这些标志真正穿透到实现中，使行为与配置一致（例如：在非零拷贝路径上也能跳过校验和，或在所有路径上一致地遵守enable_checksum）。- 关于缓冲区布局目前存在一些重复/未使用的状态：
frame_data_base_、SHM_FRAME_HEADER_SIZE、SHM_CONTROL_BLOCK_SIZE以及MakeShmName中硬编码的前缀逻辑，都与现有字段/常量有重叠；建议要么统一使用它们（例如用 base 指针做偏移计算、统一使用SHM_NAME_PREFIX），要么删掉多余的部分来降低维护成本。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- `ShmTransportConfig::enable_zero_copy` 标志目前未被使用，而且普通的 `WriteFrame` 路径会无视 `enable_checksum` 始终计算校验和；建议把这些标志真正穿透到实现中，使行为与配置一致（例如：在非零拷贝路径上跳过校验和，或者在所有路径上一致地遵守 `enable_checksum`）。
- 关于缓冲区布局目前存在一些重复/未使用的状态：`frame_data_base_`、`SHM_FRAME_HEADER_SIZE`、`SHM_CONTROL_BLOCK_SIZE` 以及 `MakeShmName` 中硬编码的前缀逻辑，都与现有字段/常量有重叠；建议要么统一使用它们（例如用 base 指针做偏移计算、统一使用 `SHM_NAME_PREFIX`），要么删掉多余的部分来降低维护成本。
## Individual Comments
### Comment 1
<location path="core/src/l0_sensing/shm_transport.cpp" line_range="266-275" />
<code_context>
+ segment->name_ = std::string(name);
+ segment->size_ = size;
+
+ segment->handle_ = CreateFileMappingA(
+ INVALID_HANDLE_VALUE,
+ nullptr,
+ PAGE_READWRITE,
+ static_cast<DWORD>((size >> 32) & 0xFFFFFFFF),
+ static_cast<DWORD>(size & 0xFFFFFFFF),
+ segment->name_.c_str()
+ );
+
+ if (segment->handle_ == nullptr) {
+ const DWORD error = GetLastError();
+ if (error == ERROR_ALREADY_EXISTS) {
</code_context>
<issue_to_address>
**issue (bug_risk):** CreateFileMappingA 的错误处理逻辑不正确，既可能漏报 AlreadyExists，也可能导致句柄泄漏。
`CreateFileMappingA` 即便在映射已经存在时也会返回一个有效句柄；当 `segment->handle_ != nullptr` 时，必须在调用之后立刻检查 `GetLastError() == ERROR_ALREADY_EXISTS`。在当前使用 `if (segment->handle_ == nullptr)` 的结构下，`ERROR_ALREADY_EXISTS` 分支永远不会被触发，因此不会报告 AlreadyExists，并且已有的共享内存可能会被错误地清零。
另外，如果 `MapViewOfFile` 失败，代码应显式调用 `CloseHandle(segment->handle_)` 并在返回 `ShmMapFailed` 前重置 `handle_`，以避免句柄泄漏。
</issue_to_address>
### Comment 2
<location path="core/src/l0_sensing/shm_transport.cpp" line_range="823" />
<code_context>
+ }
+
+ control_block_->dropped_frames.fetch_add(1, std::memory_order_relaxed);
+ stats_.write_timeouts++;
+ return false;
+}
</code_context>
<issue_to_address>
**issue (bug_risk):** 统计字段在没有同步保护的情况下被更新，可能导致数据竞争。
`stats_.write_timeouts`（以及类似的 `read_timeouts` / `checksum_errors` 字段）是在没有持有 `stats_mutex_` 的情况下自增的，而其他 `stats_` 字段则是在互斥量保护下更新和读取。在同一个结构体上混用不同的同步策略会导致数据竞争。
请统一策略：要么所有对 `stats_` 的更新都加上 `stats_mutex_` 保护，要么把这些计数器改为 `std::atomic<std::uint64_t>`，并在 `GetStats()` 中保持一致的读取策略，以避免未定义行为。
</issue_to_address>
### Comment 3
<location path="core/tests/test_shm_transport.cpp" line_range="798" />
<code_context>
+// 性能基准测试
+// ==========================================================================
+
+TEST_F(ShmTransportTest, ThroughputBenchmark)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ThroughputBenchmark 使用了硬编码的性能阈值，在不同机器和构建配置下会非常不稳定。
当前测试使用断言 `EXPECT_GT(throughput_mbps, 100.0);`，在较慢或负载较高的 CI 机器、以及 debug 构建中，很容易导致用例不稳定。建议：要么把该测试禁用为性能测试（例如命名为 `DISABLED_ThroughputBenchmark`），要么把它迁移到专门的 benchmark 套件中，或者将其改为仅验证正确性（例如：无错误、吞吐量非零），同时只记录测得的吞吐量。
建议的实现如下：
```cpp
// 性能基准测试
// ==========================================================================
=======
// ==========================================================================
// 性能基准测试（记录吞吐量，仅做正确性校验，不对性能做硬性要求）
// ==========================================================================
```
```cpp
TEST_F(ShmTransportTest, ThroughputBenchmark)
```
```cpp
// 仅校验吞吐量为正，避免在不同机器 / 构建配置下因绝对阈值导致用例不稳定
GTEST_LOG_(INFO) << "ShmTransport throughput: " << throughput_mbps << " MiB/s";
EXPECT_GT(throughput_mbps, 0.0);
```
</issue_to_address>
### Comment 4
<location path="core/tests/test_shm_transport.cpp" line_range="867" />
<code_context>
+// ==========================================================================
+// 零拷贝延迟基准测试 - 使用真正的零拷贝API
+// ==========================================================================
+TEST_F(ShmTransportTest, ZeroCopyLatencyBenchmark)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ZeroCopyLatencyBenchmark 对吞吐量/延迟设置了非常激进的目标，在 CI 环境下很可能不稳定。
当前的断言（`EXPECT_GT(read_mbps, 10000.0);` 和 `EXPECT_LT(read_latency_ns, 50000.0);`）依赖极快且噪声极低的硬件环境，在典型的 CI agent 上很可能出现毛刺。对于自动化运行，建议改为只断言正确性、记录性能，或者把这些严格阈值放在一个 flag/仅 benchmark 模式下，以避免与回归无关的测试失败。
</issue_to_address>
### Comment 5
<location path="core/tests/test_shm_transport.cpp" line_range="1099" />
<code_context>
+// ==========================================================================
+// 微秒级单帧延迟测试（使用零拷贝API）
+// ==========================================================================
+TEST_F(ShmTransportTest, MicrosecondLatencyTest)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** MicrosecondLatencyTest 使用了非常严格的 P99/中位数延迟阈值，测试很可能会抖动。
该测试断言 `EXPECT_LT(p99_latency, 10.0);` 和 `EXPECT_LT(median_latency, 5.0);`，对 CPU 频率调节、虚拟化以及系统负载极为敏感，在 CI 环境中很容易不稳定。建议将其改为不会导致失败的 benchmark（只做日志记录），或者至少放宽阈值，并/或通过 flag 进行控制，使其只在专门的性能环境中运行。
建议实现如下：
```cpp
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
=======
#// ==========================================================================
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
// NOTE:
// This test is extremely sensitive to system load, CPU scaling, and
// virtualization. To avoid CI flakiness, it is gated by an environment
// variable and will be skipped unless explicitly enabled.
const char* perf_env = std::getenv("SHM_TRANSPORT_PERF_TEST");
if (!perf_env || std::strcmp(perf_env, "1") != 0)
{
GTEST_SKIP() << "Skipping MicrosecondLatencyTest; "
<< "enable with SHM_TRANSPORT_PERF_TEST=1 in a dedicated "
<< "performance environment.";
}
```
```cpp
ShmTransportConfig config;
```
目前我只看到了测试体的开头部分。严格的延迟断言很可能位于后面，例如：
- `EXPECT_LT(p99_latency, 10.0);`
- `EXPECT_LT(median_latency, 5.0);`
在引入上述基于环境变量的开关后，这些断言可以保持不变，但只有在环境中设置了 `SHM_TRANSPORT_PERF_TEST=1` 时才会被执行（例如在专门的性能测试 job 中）。如果你更倾向于仅记录日志的 benchmark，还可以在同一个测试体中将这些 `EXPECT_LT` 调用替换为日志输出（例如 `std::cout` 或项目中的日志宏）。
</issue_to_address>
### Comment 6
<location path="core/tests/test_shm_transport.cpp" line_range="320" />
<code_context>
+ }
+}
+
+TEST_F(ShmTransportTest, TryWriteFrameNonBlocking)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** 非阻塞写目前只在 buffer 已满的场景下被测试，写入超时行为没有覆盖到。
当前测试只覆盖了当缓冲区已满时 `TryWriteFrame` 返回 `false` 的情况；并没有直接验证 `WriteFrameWithTimeout` 的超时路径。请增加一个针对性测试：先写满缓冲区，然后以一个非常短的超时时间调用 `WriteFrameWithTimeout`，断言它返回 `false`，并且 `stats_.write_timeouts` / `dropped_frames` 按预期更新。
建议实现如下：
```cpp
EXPECT_EQ(static_cast<int>(data[0]), i);
}
}
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
}
TEST_F(ShmTransportTest, WriteFrameWithTimeout_BufferFullTimesOut)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
auto init_result = producer.InitializeProducer(config);
ASSERT_TRUE(init_result.has_value()) << "Producer initialization failed";
ASSERT_TRUE(producer.IsInitialized());
// Fill the buffer completely using non-blocking writes
for (uint32_t i = 0; i < kTestBufferCount; ++i)
{
FrameMetadata metadata{};
metadata.frame_number = i;
std::vector<uint8_t> frame_data(kTestFrameSize, static_cast<uint8_t>(i));
ASSERT_TRUE(
producer.TryWriteFrame(metadata, frame_data.data(), frame_data.size())
) << "Failed to write frame " << i << " while filling buffer";
}
// At this point the ring buffer should be full; a timed write with a very short
// timeout is expected to fail with a timeout and update stats accordingly.
FrameMetadata timeout_metadata{};
timeout_metadata.frame_number = kTestBufferCount;
std::vector<uint8_t> timeout_frame(kTestFrameSize, 0xFF);
const auto timeout = std::chrono::milliseconds(1);
const bool write_result =
producer.WriteFrameWithTimeout(timeout_metadata,
timeout_frame.data(),
timeout_frame.size(),
timeout);
EXPECT_FALSE(write_result) << "WriteFrameWithTimeout should fail when buffer is full";
const auto stats = producer.GetStats();
EXPECT_EQ(stats.write_timeouts, 1u);
EXPECT_EQ(stats.dropped_frames, 1u);
```
上述测试假设 `ShmTransport` 及相关类已有如下 API 和类型：
1. `bool ShmTransport::TryWriteFrame(const FrameMetadata&, const uint8_t* data, size_t size);`
2. `bool ShmTransport::WriteFrameWithTimeout(const FrameMetadata&, const uint8_t* data, size_t size, std::chrono::milliseconds timeout);`
3. `auto ShmTransport::GetStats() const` 返回一个带有 `write_timeouts` 和 `dropped_frames` 字段的结构体。
4. 一个包含 `frame_number` 字段的 `FrameMetadata` 类型。
如果你的实际签名或统计类型不同，请根据现有实现调整参数顺序、类型名以及统计字段访问方式，但要保持测试逻辑不变：填满缓冲区，用极短的超时时间调用 `WriteFrameWithTimeout`，断言它返回 `false`，并验证与超时和丢帧相关的计数器按预期增加（如果其他测试可能已更新统计数据，可以考虑相对初始值进行增量检查，而不是硬编码为 `1u`）。
</issue_to_address>
### Comment 7
<location path="core/tests/test_shm_transport.cpp" line_range="776" />
<code_context>
+ EXPECT_FALSE(header.is_valid());
+}
+
+TEST_F(ShmTransportTest, FrameHeaderChecksum)
+{
+ ShmFrameHeader header;
</code_context>
<issue_to_address>
**suggestion (testing):** 校验和目前只在头部类型上做了单元测试，没有在传输层上（启用/禁用两种模式）做端到端验证。
当前只在隔离的情况下验证了 `ShmFrameHeader` 的 CRC。考虑到存在 `config.enable_checksum` 以及拷贝与零拷贝 API 在启用/禁用校验和时有不同路径，建议增加以下测试：（1）在 `enable_checksum=true` 时的集成测试——故意破坏共享内存中的 payload，验证 `ReadFrame`/`AcquireReadBuffer` 返回 `ShmTransportError::ShmCorrupted`；（2）在 `enable_checksum=false` 时的测试——确认同样的破坏不会触发校验和错误。这样可以覆盖传输层在损坏检测方面的控制流程。
建议实现如下：
```cpp
// Verify default-constructed header is invalid.
EXPECT_FALSE(header.is_valid());
header.magic = ShmFrameHeader::MAGIC;
header.version = 1;
header.data_size = 100;
EXPECT_TRUE(header.is_valid());
header.magic = 0;
EXPECT_FALSE(header.is_valid());
}
// Integration: checksum enabled should detect corruption in the transport.
TEST_F(ShmTransportTest, ChecksumEnabledDetectsCorruption)
{
// Arrange: create a transport with checksum enabled.
ShmTransportConfig config;
config.enable_checksum = true;
// If there are additional required fields on ShmTransportConfig,
// they should be initialized here consistent with other tests.
auto [writer, reader] = CreateTestTransportPair(config); // uses existing helper in this file
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
// Write one frame.
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the underlying shared-memory payload before reading.
//
// We assume test helpers already expose a way to get at the raw shared-memory
// region for the last written frame, similar to other corruption-related tests.
// For example:
//
// auto* raw_frame = writer.DebugGetLastWrittenFrame();
// ASSERT_NE(raw_frame, nullptr);
// // Flip a byte inside the payload region without touching the header.
// auto* raw_bytes = static_cast<std::uint8_t*>(raw_frame->payload());
// raw_bytes[0] ^= 0xFF;
//
// Replace the following block with the actual helper/API used in this file
// to access and mutate the raw shared-memory frame:
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload()); // payload() should not include header
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF; // corrupt one byte
// Act: attempt to read back the frame via the transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: transport must detect corruption and surface ShmCorrupted.
EXPECT_EQ(read_size, 0u);
EXPECT_EQ(err, ShmTransportError::ShmCorrupted);
}
// Integration: checksum disabled should NOT treat the same corruption as an error.
TEST_F(ShmTransportTest, ChecksumDisabledDoesNotDetectCorruption)
{
// Arrange: create a transport with checksum disabled.
ShmTransportConfig config;
config.enable_checksum = false;
auto [writer, reader] = CreateTestTransportPair(config);
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the same way as in ChecksumEnabledDetectsCorruption.
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload());
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF;
// Act: read via transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: corruption is not flagged via checksum when disabled.
// Depending on implementation, the corrupted payload may be delivered as-is
// or another non-checksum error may occur (e.g., unrelated transport error),
// but it must not be ShmCorrupted due to checksum.
EXPECT_NE(err, ShmTransportError::ShmCorrupted);
// In most implementations, the read should succeed:
EXPECT_EQ(read_size, payload.size());
}
```
上述示例中的若干 helper 和 API（`CreateTestTransportPair`、`WriteFrame`、`ReadFrame`、`DebugGetLastWrittenFrame`、`payload()`、`data_size` 成员，以及 `ShmTransportError::ShmCorrupted`）是根据常见模式推断的，需与实际代码对齐：
1. 将 `CreateTestTransportPair(config)` 替换为当前测试夹具中用于创建 writer/reader 对的实际 helper（例如 `CreateTransportPair`、`CreateClientServer`，或通过 `ShmTransportTest::CreateTransport` 构造）。
2. 根据真实签名调整 `writer.WriteFrame(...)` 和 `reader.ReadFrame(...)` 调用。有的代码会使用 `bool WriteFrame(span<const uint8_t>, ShmTransportError&)`，或者直接返回错误枚举而不用输出参数。
3. 将 `writer.DebugGetLastWrittenFrame()` 与 `payload()` / `data_size` 的用法替换为当前测试中真正用来访问底层共享内存帧的方式。如果尚无此类 helper，可按现有风格新增一个仅用于测试的访问接口（例如通过 `friend` 或 `#ifdef UNIT_TEST` 保护的 debug 方法）。
4. 确认 `ShmTransportError::ShmCorrupted` 名称无误；如果你的代码使用的是其它枚举名（例如 `ShmTransportError::kCorrupted` 或 `Error::kShmCorrupted`），请相应替换。
5. 如果存在单独的零拷贝 API（例如 `AcquireReadBuffer` / `ReleaseReadBuffer`），可以针对零拷贝路径再增加一组类似的测试，或者在上述测试中使用 `AcquireReadBuffer` 替代 `ReadFrame`，视当前测试结构而定。
</issue_to_address>
### Comment 8
<location path="core/tests/test_shm_transport.cpp" line_range="698" />
<code_context>
+ EXPECT_FALSE(block.is_valid());
+}
+
+TEST_F(ShmTransportTest, ControlBlockBufferStatus)
+{
+ ShmControlBlock block;
</code_context>
<issue_to_address>
**suggestion (testing):** ControlBlock 的缓冲区计数逻辑已经做了单元测试，但尚无端到端测试来验证环形缓冲区溢出时的丢帧计数。
为了补充单元测试，建议增加一个集成测试：在没有任何读取的前提下，故意写入超过 ring 容量的帧，随后同时检查 `control_block_->dropped_frames` 以及 `ShmTransportStats` 中对应字段，以确保溢出处理和统计在端到端路径上正确打通。
建议实现如下：
```cpp
// 恢复 magic 但修改 version
block.magic = ShmControlBlock::MAGIC;
block.version = 999; // 错误版本
EXPECT_FALSE(block.is_valid());
}
TEST_F(ShmTransportTest, DroppedFramesOnRingOverflow)
{
// 选择一个很小的 ring 大小，方便在测试中触发溢出
const size_t kRingCapacityFrames = 4;
const size_t kFramesToWrite = 10; // 明显大于 capacity，确保发生溢出
// 初始化一个只写端（producer），不启动 reader，这样不会有消费，ring 会被写满
std::shared_ptr<ShmTransport> writer;
{
ShmTransportOptions opts;
opts.ring_capacity_frames = kRingCapacityFrames;
// NOTE: 这里假设 ShmTransportTest 提供了一个类似 CreateWriterTransport 的辅助方法，
// 或者可以直接构造 ShmTransport。根据现有代码调整下面的初始化方式。
writer = CreateWriterTransport(opts);
}
ASSERT_NE(writer, nullptr);
ASSERT_NE(control_block_, nullptr);
const uint64_t initial_dropped = control_block_->dropped_frames;
// 构造一帧最小有效 payload，用于重复写入
ShmFrame frame{};
frame.timestamp_ns = 1;
frame.data_size = 1;
frame.data[0] = 0xAB;
// 连续写入 kFramesToWrite 帧，不读取
for (size_t i = 0; i < kFramesToWrite; ++i)
{
const bool ok = writer->WriteFrame(frame);
// 写入操作本身不需要全部成功；在 ring 满时，WriteFrame 应触发丢帧计数
(void)ok;
}
// 控制块中的 dropped_frames 应该已经累加（至少大于初始值）
EXPECT_GT(control_block_->dropped_frames, initial_dropped);
// 统计信息中也应该反映同样的 dropped_frames 数量
ShmTransportStats stats{};
writer->GetStats(&stats);
EXPECT_EQ(stats.dropped_frames, control_block_->dropped_frames);
EXPECT_GT(stats.dropped_frames, 0u);
}
```
1. 将 `CreateWriterTransport(opts)` 替换为你在 `ShmTransportTest` 中实际使用的写端初始化方式（例如直接构造或使用现有工厂/辅助函数），并确保它使用 `opts.ring_capacity_frames` 来配置 ring 大小。
2. 确认 `control_block_` 是 `ShmTransportTest` 测试夹具中的成员指针/引用，并且其 `dropped_frames` 字段存在且会在写端发生溢出时递增；如字段名不同，请相应调整断言。
3. 如果 `ShmFrame` 的字段或构造方式不同（例如需要通过工厂或 builder 创建），请相应更新 `ShmFrame frame{}` 部分，但要保证每次写入的帧都被视为“有效帧”。
4. 若 `ShmTransport::WriteFrame` 或 `ShmTransport::GetStats` 的签名不同（例如 `GetStats()` 返回值而不是通过指针输出），请按实际签名调整调用方式，但要保留对 `dropped_frames` 一致性的断言。
5. 如果 `ShmTransportStats` 位于某个命名空间或需要特定头文件，请在文件顶部添加适当的 `#include` 和命名空间限定。
</issue_to_address>
### Comment 9
<location path="core/tests/test_shm_transport.cpp" line_range="617" />
<code_context>
+// 并发测试
+// ==========================================================================
+
+TEST_F(ShmTransportTest, ConcurrentWriteRead)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ConcurrentWriteRead 验证了基本的并发行为，但没有覆盖多消费者语义及 reader 相关计数。
当前只测试了单生产者 + 单消费者。由于实现是 SPMC，并维护了 `active_readers` 计数，建议增加一个测试：让多个 consumer 附着到同一个段，驱动并发 `ReadFrame`/`AcquireReadBuffer`，并验证在 `InitializeConsumer` 和 `Shutdown` 时 `active_readers` 能正确更新，从而完整验证 SPMC 行为。
建议实现如下：
```cpp
// 并发测试
// ==========================================================================
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
=======
// ==========================================================================
// 并发测试
// ==========================================================================
TEST_F(ShmTransportTest, ConcurrentWriteRead)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// TODO: 这里保留原有的并发读写测试逻辑（单生产者 + 单消费者），
// 具体实现依赖于现有的 Write/Read 流程，在本补丁中不做更改。
}
// 多消费者并发测试，验证 SPMC 语义以及 active_readers 统计
TEST_F(ShmTransportTest, MultiConsumerConcurrentReadUpdatesActiveReaders)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
constexpr int kConsumerCount = 4;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// 多个消费者附着到同一个段
std::vector<std::unique_ptr<ShmTransport>> consumers;
consumers.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
auto consumer = std::make_unique<ShmTransport>();
auto init_res = consumer->InitializeConsumer(config);
EXPECT_TRUE(init_res.has_value()) << "Consumer initialization failed for index " << i;
EXPECT_TRUE(consumer->IsInitialized());
consumers.emplace_back(std::move(consumer));
}
// 在此处并发驱动 ReadFrame/AcquireReadBuffer 来验证 SPMC 行为
//
// 为了不假设具体的读 API，这里仅演示多线程并发访问相同的 ShmTransport 实例集合，
// 具体的读调用需要根据现有接口补充（例如 ReadFrame 或 AcquireReadBuffer）。
//
std::atomic<bool> stop{false};
std::vector<std::thread> threads;
threads.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
threads.emplace_back([&, i] {
// 保护性检查，避免空指针
auto* consumer = consumers[i].get();
ASSERT_NE(consumer, nullptr);
// 在 stop 被置为 true 之前，持续执行读取逻辑
while (!stop.load(std::memory_order_acquire)) {
// 在这里调用具体的读取接口，例如：
//
// auto frame_res = consumer->ReadFrame(/*timeout*/);
// if (frame_res.has_value() && frame_res->has_value()) { ... }
//
// 或者：
//
// auto buf_res = consumer->AcquireReadBuffer();
// if (buf_res.has_value() && buf_res->has_value()) { ... }
//
// 为了保持此补丁与现有实现解耦，这里仅做一个短暂 sleep，
// 真正的读取逻辑需要在后续补充。
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
// 让并发读取线程运行一小段时间
std::this_thread::sleep_for(std::chrono::milliseconds(50));
stop.store(true, std::memory_order_release);
for (auto& t : threads) {
if (t.joinable()) {
t.join();
}
}
// 关闭部分消费者，验证 active_readers 统计会递减
// 注意：这里使用占位的 GetActiveReadersForTest()，需要根据实际实现替换。
//
// 期望语义：
// - InitializeConsumer 成功后 active_readers == kConsumerCount
// - 关闭 2 个消费者后 active_readers == kConsumerCount - 2
//
// 如果 ShmTransport 没有对外暴露该计数，需要通过测试辅助接口或友元方式访问共享头。
//
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount);
int shutdown_count = 2;
for (int i = 0; i < shutdown_count; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount - shutdown_count);
// 关闭剩余消费者与生产者，确保不会崩溃且资源全部释放
for (int i = shutdown_count; i < kConsumerCount; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
producer.Shutdown();
EXPECT_FALSE(producer.IsInitialized());
}
```
1. 将关于 `active_readers` 的断言替换为实际可用的访问方式：
- 如果已有类似 `size_t ShmTransport::GetActiveReadersForTest() const;` 的接口，请取消注释并使用真实函数名；
- 如果没有公开接口，可考虑：
- 在共享内存头结构上新增只读测试辅助函数，或
- 将测试类声明为友元，以便直接访问 `active_readers` 字段。
2. 在 `MultiConsumerConcurrentReadUpdatesActiveReaders` 中，用当前实现中的真实读取 API 替换线程循环中的占位逻辑：
- 如果存在 `ReadFrame(...)` 或 `AcquireReadBuffer(...)`，在循环中调用并对返回值做轻量验证（例如是否超时/是否有数据），从而真正驱动 SPMC 读路径。
3. 若测试文件尚未包含 `<thread>`、`<atomic>`、`<vector>` 等头文件，请在文件顶部添加相应的 `#include`。
4. 如果项目中已有自己的线程/同步或测试工具（例如封装好的读写辅助函数），可以用这些工具重写并发部分，以保持与现有测试风格一致。
</issue_to_address>帮我变得更有用！请在每条评论上点击 👍 或 👎，你的反馈会用于改进后续的 Review 质量。
Original comment in English
Hey - I've found 9 issues, and left some high level feedback:
- The
ShmTransportConfig::enable_zero_copyflag is currently unused and the regularWriteFramepath always computes a checksum regardless ofenable_checksum; consider wiring these flags through so behavior matches the configuration (e.g., skip checksum in non-zero-copy or honorenable_checksumconsistently). - There is some duplication/unused state around buffer layout:
frame_data_base_,SHM_FRAME_HEADER_SIZE,SHM_CONTROL_BLOCK_SIZE, and the hardcoded prefix logic inMakeShmNameall overlap with existing fields/constants; consider either using these consistently (e.g., base pointer for offset math, sharedSHM_NAME_PREFIX) or removing them to reduce maintenance overhead.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `ShmTransportConfig::enable_zero_copy` flag is currently unused and the regular `WriteFrame` path always computes a checksum regardless of `enable_checksum`; consider wiring these flags through so behavior matches the configuration (e.g., skip checksum in non-zero-copy or honor `enable_checksum` consistently).
- There is some duplication/unused state around buffer layout: `frame_data_base_`, `SHM_FRAME_HEADER_SIZE`, `SHM_CONTROL_BLOCK_SIZE`, and the hardcoded prefix logic in `MakeShmName` all overlap with existing fields/constants; consider either using these consistently (e.g., base pointer for offset math, shared `SHM_NAME_PREFIX`) or removing them to reduce maintenance overhead.
## Individual Comments
### Comment 1
<location path="core/src/l0_sensing/shm_transport.cpp" line_range="266-275" />
<code_context>
+ segment->name_ = std::string(name);
+ segment->size_ = size;
+
+ segment->handle_ = CreateFileMappingA(
+ INVALID_HANDLE_VALUE,
+ nullptr,
+ PAGE_READWRITE,
+ static_cast<DWORD>((size >> 32) & 0xFFFFFFFF),
+ static_cast<DWORD>(size & 0xFFFFFFFF),
+ segment->name_.c_str()
+ );
+
+ if (segment->handle_ == nullptr) {
+ const DWORD error = GetLastError();
+ if (error == ERROR_ALREADY_EXISTS) {
</code_context>
<issue_to_address>
**issue (bug_risk):** CreateFileMappingA error handling is incorrect and can both miss AlreadyExists and leak handles.
`CreateFileMappingA` returns a valid handle even when the mapping already exists; you must check `GetLastError() == ERROR_ALREADY_EXISTS` immediately after the call when `segment->handle_ != nullptr`. With the current `if (segment->handle_ == nullptr)` structure, the `ERROR_ALREADY_EXISTS` case is never reached, so `AlreadyExists` is never reported and existing shared memory may be incorrectly zeroed.
Also, if `MapViewOfFile` fails, the code should explicitly `CloseHandle(segment->handle_)` and reset `handle_` before returning `ShmMapFailed` to avoid leaking the handle.
</issue_to_address>
### Comment 2
<location path="core/src/l0_sensing/shm_transport.cpp" line_range="823" />
<code_context>
+ }
+
+ control_block_->dropped_frames.fetch_add(1, std::memory_order_relaxed);
+ stats_.write_timeouts++;
+ return false;
+}
</code_context>
<issue_to_address>
**issue (bug_risk):** Stats fields are updated without synchronization, causing potential data races.
`stats_.write_timeouts` (and similar fields like `read_timeouts` / `checksum_errors`) are incremented without `stats_mutex_`, while other `stats_` fields are updated and read under the mutex. This mixed synchronization on the same struct can cause data races.
Either guard all `stats_` updates with `stats_mutex_`, or make these counters `std::atomic<std::uint64_t>` and keep a consistent strategy for reads in `GetStats()` to avoid undefined behavior.
</issue_to_address>
### Comment 3
<location path="core/tests/test_shm_transport.cpp" line_range="798" />
<code_context>
+// 性能基准测试
+// ==========================================================================
+
+TEST_F(ShmTransportTest, ThroughputBenchmark)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ThroughputBenchmark has hardcoded performance thresholds that will be flaky across machines and build configurations.
The test currently enforces `EXPECT_GT(throughput_mbps, 100.0);`, which is likely to be unstable on slower or loaded CI machines and in debug builds. Please either disable it as a performance test (e.g. `DISABLED_ThroughputBenchmark`), move it to a dedicated benchmark suite, or change it to only validate correctness (e.g. no errors, non‑zero throughput) while just logging the measured throughput.
Suggested implementation:
```cpp
// 性能基准测试
// ==========================================================================
=======
// ==========================================================================
// 性能基准测试（记录吞吐量，仅做正确性校验，不对性能做硬性要求）
// ==========================================================================
```
```cpp
TEST_F(ShmTransportTest, ThroughputBenchmark)
```
```cpp
// 仅校验吞吐量为正，避免在不同机器 / 构建配置下因绝对阈值导致用例不稳定
GTEST_LOG_(INFO) << "ShmTransport throughput: " << throughput_mbps << " MiB/s";
EXPECT_GT(throughput_mbps, 0.0);
```
</issue_to_address>
### Comment 4
<location path="core/tests/test_shm_transport.cpp" line_range="867" />
<code_context>
+// ==========================================================================
+// 零拷贝延迟基准测试 - 使用真正的零拷贝API
+// ==========================================================================
+TEST_F(ShmTransportTest, ZeroCopyLatencyBenchmark)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ZeroCopyLatencyBenchmark enforces very high throughput/latency targets which are likely unstable in CI.
These expectations (`EXPECT_GT(read_mbps, 10000.0);` and `EXPECT_LT(read_latency_ns, 50000.0);`) depend on extremely fast, low-noise hardware and are likely to be flaky on typical CI agents. For automated runs, consider instead asserting only correctness, logging performance, or guarding these strict thresholds behind a flag/benchmark-only mode, to avoid non-regression-related test failures.
</issue_to_address>
### Comment 5
<location path="core/tests/test_shm_transport.cpp" line_range="1099" />
<code_context>
+// ==========================================================================
+// 微秒级单帧延迟测试（使用零拷贝API）
+// ==========================================================================
+TEST_F(ShmTransportTest, MicrosecondLatencyTest)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** MicrosecondLatencyTest uses strict P99/median latency thresholds that will likely be flaky.
This test asserts `EXPECT_LT(p99_latency, 10.0);` and `EXPECT_LT(median_latency, 5.0);`, which is extremely sensitive to CPU scaling, virtualization, and general system load, and will likely be unstable in CI. Consider converting it to a non-failing benchmark (log-only), or at least loosening the thresholds and/or gating it by a flag so it only runs in dedicated performance environments.
Suggested implementation:
```cpp
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
=======
#// ==========================================================================
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
// NOTE:
// This test is extremely sensitive to system load, CPU scaling, and
// virtualization. To avoid CI flakiness, it is gated by an environment
// variable and will be skipped unless explicitly enabled.
const char* perf_env = std::getenv("SHM_TRANSPORT_PERF_TEST");
if (!perf_env || std::strcmp(perf_env, "1") != 0)
{
GTEST_SKIP() << "Skipping MicrosecondLatencyTest; "
<< "enable with SHM_TRANSPORT_PERF_TEST=1 in a dedicated "
<< "performance environment.";
}
```
```cpp
ShmTransportConfig config;
```
I only see the beginning of the test body. The strict latency assertions are likely further down, e.g.:
- `EXPECT_LT(p99_latency, 10.0);`
- `EXPECT_LT(median_latency, 5.0);`
With the gating added above, those assertions can remain unchanged but will only be exercised when `SHM_TRANSPORT_PERF_TEST=1` is set in the environment (e.g., in a dedicated performance job). If you prefer a log-only benchmark, you could additionally replace those `EXPECT_LT` calls with logging (e.g., `std::cout` or the project's logging macro) in the same test body.
</issue_to_address>
### Comment 6
<location path="core/tests/test_shm_transport.cpp" line_range="320" />
<code_context>
+ }
+}
+
+TEST_F(ShmTransportTest, TryWriteFrameNonBlocking)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** Non-blocking write is tested only for the buffer-full case; write timeout behaviour is not covered.
Current tests only cover `TryWriteFrame` returning `false` when the buffer is full; they don’t directly verify `WriteFrameWithTimeout`’s timeout path. Please add a focused test that fills the buffer, calls `WriteFrameWithTimeout` with a very short timeout, and asserts it returns `false` and updates `stats_.write_timeouts` / `dropped_frames` as expected.
Suggested implementation:
```cpp
EXPECT_EQ(static_cast<int>(data[0]), i);
}
}
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
}
TEST_F(ShmTransportTest, WriteFrameWithTimeout_BufferFullTimesOut)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
auto init_result = producer.InitializeProducer(config);
ASSERT_TRUE(init_result.has_value()) << "Producer initialization failed";
ASSERT_TRUE(producer.IsInitialized());
// Fill the buffer completely using non-blocking writes
for (uint32_t i = 0; i < kTestBufferCount; ++i)
{
FrameMetadata metadata{};
metadata.frame_number = i;
std::vector<uint8_t> frame_data(kTestFrameSize, static_cast<uint8_t>(i));
ASSERT_TRUE(
producer.TryWriteFrame(metadata, frame_data.data(), frame_data.size())
) << "Failed to write frame " << i << " while filling buffer";
}
// At this point the ring buffer should be full; a timed write with a very short
// timeout is expected to fail with a timeout and update stats accordingly.
FrameMetadata timeout_metadata{};
timeout_metadata.frame_number = kTestBufferCount;
std::vector<uint8_t> timeout_frame(kTestFrameSize, 0xFF);
const auto timeout = std::chrono::milliseconds(1);
const bool write_result =
producer.WriteFrameWithTimeout(timeout_metadata,
timeout_frame.data(),
timeout_frame.size(),
timeout);
EXPECT_FALSE(write_result) << "WriteFrameWithTimeout should fail when buffer is full";
const auto stats = producer.GetStats();
EXPECT_EQ(stats.write_timeouts, 1u);
EXPECT_EQ(stats.dropped_frames, 1u);
```
The above test assumes the following existing APIs and types on `ShmTransport` and related classes:
1. `bool ShmTransport::TryWriteFrame(const FrameMetadata&, const uint8_t* data, size_t size);`
2. `bool ShmTransport::WriteFrameWithTimeout(const FrameMetadata&, const uint8_t* data, size_t size, std::chrono::milliseconds timeout);`
3. `auto ShmTransport::GetStats() const` returning a struct with `write_timeouts` and `dropped_frames` fields.
4. A `FrameMetadata` type with a `frame_number` field.
If your actual signatures or stats type differ, adjust the parameter ordering, type names, and the stats field accesses to match your existing implementation while preserving the test logic: fill the buffer, call `WriteFrameWithTimeout` with a very short timeout, assert it returns `false`, and verify that the timeout- and drop-related counters increase as expected (possibly by checking for increment relative to a baseline snapshot rather than hard-coded `1u` if other tests may have already updated the stats).
</issue_to_address>
### Comment 7
<location path="core/tests/test_shm_transport.cpp" line_range="776" />
<code_context>
+ EXPECT_FALSE(header.is_valid());
+}
+
+TEST_F(ShmTransportTest, FrameHeaderChecksum)
+{
+ ShmFrameHeader header;
</code_context>
<issue_to_address>
**suggestion (testing):** Checksum is tested on the header type but not integrated through the transport in both enabled/disabled modes.
Currently this only validates `ShmFrameHeader`’s CRC in isolation. Given `config.enable_checksum` and the separate paths for checksum-enabled/disabled in copy and zero-copy APIs, please add: (1) an integration test with `enable_checksum=true` where you corrupt the shared-memory payload and verify `ReadFrame`/`AcquireReadBuffer` returns `ShmTransportError::ShmCorrupted`; and (2) a test with `enable_checksum=false` confirming the same corruption does not trigger a checksum error. This will cover the transport-level control flow around corruption detection.
Suggested implementation:
```cpp
// Verify default-constructed header is invalid.
EXPECT_FALSE(header.is_valid());
header.magic = ShmFrameHeader::MAGIC;
header.version = 1;
header.data_size = 100;
EXPECT_TRUE(header.is_valid());
header.magic = 0;
EXPECT_FALSE(header.is_valid());
}
// Integration: checksum enabled should detect corruption in the transport.
TEST_F(ShmTransportTest, ChecksumEnabledDetectsCorruption)
{
// Arrange: create a transport with checksum enabled.
ShmTransportConfig config;
config.enable_checksum = true;
// If there are additional required fields on ShmTransportConfig,
// they should be initialized here consistent with other tests.
auto [writer, reader] = CreateTestTransportPair(config); // uses existing helper in this file
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
// Write one frame.
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the underlying shared-memory payload before reading.
//
// We assume test helpers already expose a way to get at the raw shared-memory
// region for the last written frame, similar to other corruption-related tests.
// For example:
//
// auto* raw_frame = writer.DebugGetLastWrittenFrame();
// ASSERT_NE(raw_frame, nullptr);
// // Flip a byte inside the payload region without touching the header.
// auto* raw_bytes = static_cast<std::uint8_t*>(raw_frame->payload());
// raw_bytes[0] ^= 0xFF;
//
// Replace the following block with the actual helper/API used in this file
// to access and mutate the raw shared-memory frame:
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload()); // payload() should not include header
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF; // corrupt one byte
// Act: attempt to read back the frame via the transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: transport must detect corruption and surface ShmCorrupted.
EXPECT_EQ(read_size, 0u);
EXPECT_EQ(err, ShmTransportError::ShmCorrupted);
}
// Integration: checksum disabled should NOT treat the same corruption as an error.
TEST_F(ShmTransportTest, ChecksumDisabledDoesNotDetectCorruption)
{
// Arrange: create a transport with checksum disabled.
ShmTransportConfig config;
config.enable_checksum = false;
auto [writer, reader] = CreateTestTransportPair(config);
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the same way as in ChecksumEnabledDetectsCorruption.
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload());
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF;
// Act: read via transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: corruption is not flagged via checksum when disabled.
// Depending on implementation, the corrupted payload may be delivered as-is
// or another non-checksum error may occur (e.g., unrelated transport error),
// but it must not be ShmCorrupted due to checksum.
EXPECT_NE(err, ShmTransportError::ShmCorrupted);
// In most implementations, the read should succeed:
EXPECT_EQ(read_size, payload.size());
}
```
The exact helpers and APIs used above (`CreateTestTransportPair`, `WriteFrame`, `ReadFrame`, `DebugGetLastWrittenFrame`, `payload()`, `data_size` member, and `ShmTransportError::ShmCorrupted`) are inferred from common patterns and the comment you provided:
1. Replace `CreateTestTransportPair(config)` with whatever the existing test fixture uses to instantiate a writer/reader pair (e.g. `CreateTransportPair`, `CreateClientServer`, or construction via `ShmTransportTest::CreateTransport`).
2. Adjust `writer.WriteFrame(...)` and `reader.ReadFrame(...)` calls to match the real signatures. Some codebases use `bool WriteFrame(span<const uint8_t>, ShmTransportError&)` or return an error enum directly instead of using an out-parameter.
3. Replace `writer.DebugGetLastWrittenFrame()` and the `payload()`/`data_size` usage with the actual way the tests already access the underlying shared-memory segment/frame. If no such helper exists, add a small test-only accessor consistent with your existing style (e.g., a `friend` test or a debug method guarded by `#ifdef UNIT_TEST`).
4. Ensure `ShmTransportError::ShmCorrupted` is the correct enum name; if your code uses a different identifier (e.g., `ShmTransportError::kCorrupted` or `Error::kShmCorrupted`), substitute it accordingly.
5. If you have separate zero-copy APIs (e.g., `AcquireReadBuffer` / `ReleaseReadBuffer`), mirror the same corruption scenario in a second pair of tests for the zero-copy path, or extend one of the tests above to use `AcquireReadBuffer` instead of `ReadFrame`, depending on how your existing tests are structured.
</issue_to_address>
### Comment 8
<location path="core/tests/test_shm_transport.cpp" line_range="698" />
<code_context>
+ EXPECT_FALSE(block.is_valid());
+}
+
+TEST_F(ShmTransportTest, ControlBlockBufferStatus)
+{
+ ShmControlBlock block;
</code_context>
<issue_to_address>
**suggestion (testing):** ControlBlock buffer accounting is unit-tested, but there is no end-to-end test that asserts dropped frame counting when the ring overflows.
To complement the unit test, please add an integration test that deliberately overruns the ring (producer writes more frames than fit, with no reads) and then asserts both `control_block_->dropped_frames` and the corresponding field in `ShmTransportStats` to ensure overflow handling and stats are correctly wired end-to-end.
Suggested implementation:
```cpp
// 恢复 magic 但修改 version
block.magic = ShmControlBlock::MAGIC;
block.version = 999; // 错误版本
EXPECT_FALSE(block.is_valid());
}
TEST_F(ShmTransportTest, DroppedFramesOnRingOverflow)
{
// 选择一个很小的 ring 大小，方便在测试中触发溢出
const size_t kRingCapacityFrames = 4;
const size_t kFramesToWrite = 10; // 明显大于 capacity，确保发生溢出
// 初始化一个只写端（producer），不启动 reader，这样不会有消费，ring 会被写满
std::shared_ptr<ShmTransport> writer;
{
ShmTransportOptions opts;
opts.ring_capacity_frames = kRingCapacityFrames;
// NOTE: 这里假设 ShmTransportTest 提供了一个类似 CreateWriterTransport 的辅助方法，
// 或者可以直接构造 ShmTransport。根据现有代码调整下面的初始化方式。
writer = CreateWriterTransport(opts);
}
ASSERT_NE(writer, nullptr);
ASSERT_NE(control_block_, nullptr);
const uint64_t initial_dropped = control_block_->dropped_frames;
// 构造一帧最小有效 payload，用于重复写入
ShmFrame frame{};
frame.timestamp_ns = 1;
frame.data_size = 1;
frame.data[0] = 0xAB;
// 连续写入 kFramesToWrite 帧，不读取
for (size_t i = 0; i < kFramesToWrite; ++i)
{
const bool ok = writer->WriteFrame(frame);
// 写入操作本身不需要全部成功；在 ring 满时，WriteFrame 应触发丢帧计数
(void)ok;
}
// 控制块中的 dropped_frames 应该已经累加（至少大于初始值）
EXPECT_GT(control_block_->dropped_frames, initial_dropped);
// 统计信息中也应该反映同样的 dropped_frames 数量
ShmTransportStats stats{};
writer->GetStats(&stats);
EXPECT_EQ(stats.dropped_frames, control_block_->dropped_frames);
EXPECT_GT(stats.dropped_frames, 0u);
}
```
1. 将 `CreateWriterTransport(opts)` 替换为你在 `ShmTransportTest` 中实际使用的 writer 端初始化方式（例如构造函数或现有的工厂/辅助函数），并确保它使用 `opts.ring_capacity_frames` 来配置 ring 大小。
2. 确认 `control_block_` 是 `ShmTransportTest` 测试夹具中的成员指针/引用，并且其 `dropped_frames` 字段存在且在写端溢出时会被递增；如果名字不同，请相应修改断言中的字段名。
3. 如果 `ShmFrame` 的字段或构造方式不同（例如需要通过工厂或 builder 创建），需要将 `ShmFrame frame{}` 部分改为你项目中已有的创建方式，但要保持所有写入帧都被视为“有效帧”。
4. 如果 `ShmTransport::WriteFrame` 或 `ShmTransport::GetStats` 的签名不同（例如 `GetStats()` 返回一个值而不是通过指针输出），请据实际签名调整调用方式，但保留对 `dropped_frames` 一致性的断言。
5. 如果 `ShmTransportStats` 位于命名空间内或需要头文件，请在文件顶部添加相应的 `#include` 和命名空间限定符。
</issue_to_address>
### Comment 9
<location path="core/tests/test_shm_transport.cpp" line_range="617" />
<code_context>
+// 并发测试
+// ==========================================================================
+
+TEST_F(ShmTransportTest, ConcurrentWriteRead)
+{
+ ShmTransport producer;
</code_context>
<issue_to_address>
**suggestion (testing):** ConcurrentWriteRead validates basic concurrency but not multi-consumer semantics or reader accounting.
This currently exercises only a single producer and single consumer. Since the implementation is SPMC and tracks `active_readers`, please add a test that attaches multiple consumers to the same segment, drives concurrent `ReadFrame`/`AcquireReadBuffer`, and verifies that `active_readers` is updated correctly on `InitializeConsumer` and `Shutdown` to fully validate the SPMC behavior.
Suggested implementation:
```cpp
// 并发测试
// ==========================================================================
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
=======
// ==========================================================================
// 并发测试
// ==========================================================================
TEST_F(ShmTransportTest, ConcurrentWriteRead)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// TODO: 这里保留原有的并发读写测试逻辑（单生产者 + 单消费者），
// 具体实现依赖于现有的 Write/Read 流程，在本补丁中不做更改。
}
// 多消费者并发测试，验证 SPMC 语义以及 active_readers 统计
TEST_F(ShmTransportTest, MultiConsumerConcurrentReadUpdatesActiveReaders)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
constexpr int kConsumerCount = 4;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// 多个消费者附着到同一个段
std::vector<std::unique_ptr<ShmTransport>> consumers;
consumers.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
auto consumer = std::make_unique<ShmTransport>();
auto init_res = consumer->InitializeConsumer(config);
EXPECT_TRUE(init_res.has_value()) << "Consumer initialization failed for index " << i;
EXPECT_TRUE(consumer->IsInitialized());
consumers.emplace_back(std::move(consumer));
}
// 在此处并发驱动 ReadFrame/AcquireReadBuffer 来验证 SPMC 行为
//
// 为了不假设具体的读 API，这里仅演示多线程并发访问相同的 ShmTransport 实例集合，
// 具体的读调用需要根据现有接口补充（例如 ReadFrame 或 AcquireReadBuffer）。
//
std::atomic<bool> stop{false};
std::vector<std::thread> threads;
threads.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
threads.emplace_back([&, i] {
// 保护性检查，避免空指针
auto* consumer = consumers[i].get();
ASSERT_NE(consumer, nullptr);
// 在 stop 被置为 true 之前，持续执行读取逻辑
while (!stop.load(std::memory_order_acquire)) {
// 在这里调用具体的读取接口，例如：
//
// auto frame_res = consumer->ReadFrame(/*timeout*/);
// if (frame_res.has_value() && frame_res->has_value()) { ... }
//
// 或者：
//
// auto buf_res = consumer->AcquireReadBuffer();
// if (buf_res.has_value() && buf_res->has_value()) { ... }
//
// 为了保持此补丁与现有实现解耦，这里仅做一个短暂 sleep，
// 真正的读取逻辑需要在后续补充。
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
// 让并发读取线程运行一小段时间
std::this_thread::sleep_for(std::chrono::milliseconds(50));
stop.store(true, std::memory_order_release);
for (auto& t : threads) {
if (t.joinable()) {
t.join();
}
}
// 关闭部分消费者，验证 active_readers 统计会递减
// 注意：这里使用占位的 GetActiveReadersForTest()，需要根据实际实现替换。
//
// 期望语义：
// - InitializeConsumer 成功后 active_readers == kConsumerCount
// - 关闭 2 个消费者后 active_readers == kConsumerCount - 2
//
// 如果 ShmTransport 没有对外暴露该计数，需要通过测试辅助接口或友元方式访问共享头。
//
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount);
int shutdown_count = 2;
for (int i = 0; i < shutdown_count; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount - shutdown_count);
// 关闭剩余消费者与生产者，确保不会崩溃且资源全部释放
for (int i = shutdown_count; i < kConsumerCount; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
producer.Shutdown();
EXPECT_FALSE(producer.IsInitialized());
}
```
1. 将测试中关于 active_readers 的断言替换为实际可用的访问方式：
- 如果有类似 `size_t ShmTransport::GetActiveReadersForTest() const;` 的接口，请取消注释并使用真实函数名。
- 如果没有公开接口，请考虑：
- 在共享内存头结构上新增只读测试辅助函数，或
- 将测试类声明为友元，以便直接访问 `active_readers` 字段。
2. 在 `MultiConsumerConcurrentReadUpdatesActiveReaders` 中，用当前实现中的真实读取 API 替换线程循环中的占位部分：
- 若有 `ReadFrame(...)` 或 `AcquireReadBuffer(...)`，在循环中调用并对返回值做轻量验证（例如是否超时/有无数据），以真实驱动 SPMC 读路径。
3. 若测试文件尚未包含 `<thread>`, `<atomic>`, `<vector>` 等头文件，请在文件顶部适当位置添加：
`#include <thread>`, `#include <atomic>`, `#include <vector>`.
4. 如果已有自己的线程/同步工具或测试工具（例如封装好的读写辅助函数），可将并发部分重写为调用这些工具，以保持与现有测试风格一致。
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+266
to
+275
| segment->handle_ = CreateFileMappingA( | ||
| INVALID_HANDLE_VALUE, | ||
| nullptr, | ||
| PAGE_READWRITE, | ||
| static_cast<DWORD>((size >> 32) & 0xFFFFFFFF), | ||
| static_cast<DWORD>(size & 0xFFFFFFFF), | ||
| segment->name_.c_str() | ||
| ); | ||
|
|
||
| if (segment->handle_ == nullptr) { |
There was a problem hiding this comment.
issue (bug_risk): CreateFileMappingA 的错误处理逻辑不正确，既可能漏报 AlreadyExists，也可能导致句柄泄漏。
CreateFileMappingA 即便在映射已经存在时也会返回一个有效句柄；当 segment->handle_ != nullptr 时，必须在调用之后立刻检查 GetLastError() == ERROR_ALREADY_EXISTS。在当前使用 if (segment->handle_ == nullptr) 的结构下，ERROR_ALREADY_EXISTS 分支永远不会被触发，因此不会报告 AlreadyExists，并且已有的共享内存可能会被错误地清零。
另外，如果 MapViewOfFile 失败，代码应显式调用 CloseHandle(segment->handle_) 并在返回 ShmMapFailed 前重置 handle_，以避免句柄泄漏。
Original comment in English
issue (bug_risk): CreateFileMappingA error handling is incorrect and can both miss AlreadyExists and leak handles.
CreateFileMappingA returns a valid handle even when the mapping already exists; you must check GetLastError() == ERROR_ALREADY_EXISTS immediately after the call when segment->handle_ != nullptr. With the current if (segment->handle_ == nullptr) structure, the ERROR_ALREADY_EXISTS case is never reached, so AlreadyExists is never reported and existing shared memory may be incorrectly zeroed.
Also, if MapViewOfFile fails, the code should explicitly CloseHandle(segment->handle_) and reset handle_ before returning ShmMapFailed to avoid leaking the handle.
| } | ||
|
|
||
| control_block_->dropped_frames.fetch_add(1, std::memory_order_relaxed); | ||
| stats_.write_timeouts++; |
There was a problem hiding this comment.
issue (bug_risk): 统计字段在没有同步保护的情况下被更新，可能导致数据竞争。
stats_.write_timeouts（以及类似的 read_timeouts / checksum_errors 字段）是在没有持有 stats_mutex_ 的情况下自增的，而其他 stats_ 字段则是在互斥量保护下更新和读取。在同一个结构体上混用不同的同步策略会导致数据竞争。
请统一策略：要么所有对 stats_ 的更新都加上 stats_mutex_ 保护，要么把这些计数器改为 std::atomic<std::uint64_t>，并在 GetStats() 中保持一致的读取策略，以避免未定义行为。
Original comment in English
issue (bug_risk): Stats fields are updated without synchronization, causing potential data races.
stats_.write_timeouts (and similar fields like read_timeouts / checksum_errors) are incremented without stats_mutex_, while other stats_ fields are updated and read under the mutex. This mixed synchronization on the same struct can cause data races.
Either guard all stats_ updates with stats_mutex_, or make these counters std::atomic<std::uint64_t> and keep a consistent strategy for reads in GetStats() to avoid undefined behavior.
| // 性能基准测试 | ||
| // ========================================================================== | ||
|
|
||
| TEST_F(ShmTransportTest, ThroughputBenchmark) |
There was a problem hiding this comment.
suggestion (testing): ThroughputBenchmark 使用了硬编码的性能阈值，在不同机器和构建配置下会非常不稳定。
当前测试使用断言 EXPECT_GT(throughput_mbps, 100.0);，在较慢或负载较高的 CI 机器、以及 debug 构建中，很容易导致用例不稳定。请考虑：要么把该测试禁用为性能测试（例如命名为 DISABLED_ThroughputBenchmark），要么把它迁移到专门的 benchmark 套件中，或者将其改为仅验证正确性（例如：无错误、吞吐量非零），同时只记录测得的吞吐量。
建议的实现如下：
// 性能基准测试
// ==========================================================================
=======
// ==========================================================================
// 性能基准测试（记录吞吐量，仅做正确性校验，不对性能做硬性要求）
// ==========================================================================
TEST_F(ShmTransportTest, ThroughputBenchmark)
// 仅校验吞吐量为正，避免在不同机器 / 构建配置下因绝对阈值导致用例不稳定
GTEST_LOG_(INFO) << "ShmTransport throughput: " << throughput_mbps << " MiB/s";
EXPECT_GT(throughput_mbps, 0.0);
Original comment in English
suggestion (testing): ThroughputBenchmark has hardcoded performance thresholds that will be flaky across machines and build configurations.
The test currently enforces EXPECT_GT(throughput_mbps, 100.0);, which is likely to be unstable on slower or loaded CI machines and in debug builds. Please either disable it as a performance test (e.g. DISABLED_ThroughputBenchmark), move it to a dedicated benchmark suite, or change it to only validate correctness (e.g. no errors, non‑zero throughput) while just logging the measured throughput.
Suggested implementation:
// 性能基准测试
// ==========================================================================
=======
// ==========================================================================
// 性能基准测试（记录吞吐量，仅做正确性校验，不对性能做硬性要求）
// ==========================================================================
TEST_F(ShmTransportTest, ThroughputBenchmark)
// 仅校验吞吐量为正，避免在不同机器 / 构建配置下因绝对阈值导致用例不稳定
GTEST_LOG_(INFO) << "ShmTransport throughput: " << throughput_mbps << " MiB/s";
EXPECT_GT(throughput_mbps, 0.0);
| // ========================================================================== | ||
| // 零拷贝延迟基准测试 - 使用真正的零拷贝API | ||
| // ========================================================================== | ||
| TEST_F(ShmTransportTest, ZeroCopyLatencyBenchmark) |
There was a problem hiding this comment.
suggestion (testing): ZeroCopyLatencyBenchmark 对吞吐量/延迟设置了非常激进的目标，在 CI 环境下很可能不稳定。
当前的断言（EXPECT_GT(read_mbps, 10000.0); 和 EXPECT_LT(read_latency_ns, 50000.0);）依赖极快且噪声极低的硬件环境，在典型的 CI agent 上很可能出现毛刺。对于自动化运行，建议改为只断言正确性、记录性能，或者把这些严格阈值放在一个 flag/仅 benchmark 模式下，以避免与回归无关的测试失败。
Original comment in English
suggestion (testing): ZeroCopyLatencyBenchmark enforces very high throughput/latency targets which are likely unstable in CI.
These expectations (EXPECT_GT(read_mbps, 10000.0); and EXPECT_LT(read_latency_ns, 50000.0);) depend on extremely fast, low-noise hardware and are likely to be flaky on typical CI agents. For automated runs, consider instead asserting only correctness, logging performance, or guarding these strict thresholds behind a flag/benchmark-only mode, to avoid non-regression-related test failures.
| // ========================================================================== | ||
| // 微秒级单帧延迟测试（使用零拷贝API） | ||
| // ========================================================================== | ||
| TEST_F(ShmTransportTest, MicrosecondLatencyTest) |
There was a problem hiding this comment.
suggestion (testing): MicrosecondLatencyTest 使用了非常严格的 P99/中位数延迟阈值，测试很可能会抖动。
该测试断言 EXPECT_LT(p99_latency, 10.0); 和 EXPECT_LT(median_latency, 5.0);，对 CPU 频率调节、虚拟化以及系统负载极为敏感，在 CI 环境中很容易不稳定。建议将其改为不会导致失败的 benchmark（只做日志记录），或者至少放宽阈值，并/或通过 flag 进行控制，使其只在专门的性能环境中运行。
建议实现如下：
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
=======
#// ==========================================================================
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
// NOTE:
// This test is extremely sensitive to system load, CPU scaling, and
// virtualization. To avoid CI flakiness, it is gated by an environment
// variable and will be skipped unless explicitly enabled.
const char* perf_env = std::getenv("SHM_TRANSPORT_PERF_TEST");
if (!perf_env || std::strcmp(perf_env, "1") != 0)
{
GTEST_SKIP() << "Skipping MicrosecondLatencyTest; "
<< "enable with SHM_TRANSPORT_PERF_TEST=1 in a dedicated "
<< "performance environment.";
}
ShmTransportConfig config;
目前我只看到了测试体的开头部分。严格的延迟断言很可能位于后面，例如：
EXPECT_LT(p99_latency, 10.0);EXPECT_LT(median_latency, 5.0);
在引入上述基于环境变量的开关后，这些断言可以保持不变，但只有在环境中设置了 SHM_TRANSPORT_PERF_TEST=1 时才会被执行（例如在专门的性能测试 job 中）。如果你更倾向于仅记录日志的 benchmark，还可以在同一个测试体中将这些 EXPECT_LT 调用替换为日志输出（例如 std::cout 或项目中的日志宏）。
Original comment in English
suggestion (testing): MicrosecondLatencyTest uses strict P99/median latency thresholds that will likely be flaky.
This test asserts EXPECT_LT(p99_latency, 10.0); and EXPECT_LT(median_latency, 5.0);, which is extremely sensitive to CPU scaling, virtualization, and general system load, and will likely be unstable in CI. Consider converting it to a non-failing benchmark (log-only), or at least loosening the thresholds and/or gating it by a flag so it only runs in dedicated performance environments.
Suggested implementation:
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
=======
#// ==========================================================================
// 微秒级单帧延迟测试（使用零拷贝API）
// ==========================================================================
TEST_F(ShmTransportTest, MicrosecondLatencyTest)
{
// NOTE:
// This test is extremely sensitive to system load, CPU scaling, and
// virtualization. To avoid CI flakiness, it is gated by an environment
// variable and will be skipped unless explicitly enabled.
const char* perf_env = std::getenv("SHM_TRANSPORT_PERF_TEST");
if (!perf_env || std::strcmp(perf_env, "1") != 0)
{
GTEST_SKIP() << "Skipping MicrosecondLatencyTest; "
<< "enable with SHM_TRANSPORT_PERF_TEST=1 in a dedicated "
<< "performance environment.";
}
ShmTransportConfig config;
I only see the beginning of the test body. The strict latency assertions are likely further down, e.g.:
EXPECT_LT(p99_latency, 10.0);EXPECT_LT(median_latency, 5.0);
With the gating added above, those assertions can remain unchanged but will only be exercised when SHM_TRANSPORT_PERF_TEST=1 is set in the environment (e.g., in a dedicated performance job). If you prefer a log-only benchmark, you could additionally replace those EXPECT_LT calls with logging (e.g., std::cout or the project's logging macro) in the same test body.
| } | ||
| } | ||
|
|
||
| TEST_F(ShmTransportTest, TryWriteFrameNonBlocking) |
There was a problem hiding this comment.
suggestion (testing): 非阻塞写目前只在 buffer 已满的场景下被测试，写入超时行为没有覆盖到。
当前测试只覆盖了当缓冲区已满时 TryWriteFrame 返回 false 的情况；并没有直接验证 WriteFrameWithTimeout 的超时路径。请增加一个针对性测试：先写满缓冲区，然后以一个非常短的超时时间调用 WriteFrameWithTimeout，断言它返回 false，并且 stats_.write_timeouts / dropped_frames 按预期更新。
建议实现如下：
EXPECT_EQ(static_cast<int>(data[0]), i);
}
}
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
}
TEST_F(ShmTransportTest, WriteFrameWithTimeout_BufferFullTimesOut)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
auto init_result = producer.InitializeProducer(config);
ASSERT_TRUE(init_result.has_value()) << "Producer initialization failed";
ASSERT_TRUE(producer.IsInitialized());
// Fill the buffer completely using non-blocking writes
for (uint32_t i = 0; i < kTestBufferCount; ++i)
{
FrameMetadata metadata{};
metadata.frame_number = i;
std::vector<uint8_t> frame_data(kTestFrameSize, static_cast<uint8_t>(i));
ASSERT_TRUE(
producer.TryWriteFrame(metadata, frame_data.data(), frame_data.size())
) << "Failed to write frame " << i << " while filling buffer";
}
// At this point the ring buffer should be full; a timed write with a very short
// timeout is expected to fail with a timeout and update stats accordingly.
FrameMetadata timeout_metadata{};
timeout_metadata.frame_number = kTestBufferCount;
std::vector<uint8_t> timeout_frame(kTestFrameSize, 0xFF);
const auto timeout = std::chrono::milliseconds(1);
const bool write_result =
producer.WriteFrameWithTimeout(timeout_metadata,
timeout_frame.data(),
timeout_frame.size(),
timeout);
EXPECT_FALSE(write_result) << "WriteFrameWithTimeout should fail when buffer is full";
const auto stats = producer.GetStats();
EXPECT_EQ(stats.write_timeouts, 1u);
EXPECT_EQ(stats.dropped_frames, 1u);
上述测试假设 ShmTransport 及相关类已有如下 API 和类型：
bool ShmTransport::TryWriteFrame(const FrameMetadata&, const uint8_t* data, size_t size);bool ShmTransport::WriteFrameWithTimeout(const FrameMetadata&, const uint8_t* data, size_t size, std::chrono::milliseconds timeout);auto ShmTransport::GetStats() const返回一个带有write_timeouts和dropped_frames字段的结构体。- 一个包含
frame_number字段的FrameMetadata类型。
如果你的实际签名或统计类型不同，请根据现有实现调整参数顺序、类型名以及统计字段访问方式，但要保持测试逻辑不变：填满缓冲区，用极短的超时时间调用WriteFrameWithTimeout，断言它返回false，并验证与超时和丢帧相关的计数器按预期增加（如果其他测试可能已更新统计数据，可以考虑相对初始值进行增量检查，而不是硬编码为1u）。
Original comment in English
suggestion (testing): Non-blocking write is tested only for the buffer-full case; write timeout behaviour is not covered.
Current tests only cover TryWriteFrame returning false when the buffer is full; they don’t directly verify WriteFrameWithTimeout’s timeout path. Please add a focused test that fills the buffer, calls WriteFrameWithTimeout with a very short timeout, and asserts it returns false and updates stats_.write_timeouts / dropped_frames as expected.
Suggested implementation:
EXPECT_EQ(static_cast<int>(data[0]), i);
}
}
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
}
TEST_F(ShmTransportTest, WriteFrameWithTimeout_BufferFullTimesOut)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
auto init_result = producer.InitializeProducer(config);
ASSERT_TRUE(init_result.has_value()) << "Producer initialization failed";
ASSERT_TRUE(producer.IsInitialized());
// Fill the buffer completely using non-blocking writes
for (uint32_t i = 0; i < kTestBufferCount; ++i)
{
FrameMetadata metadata{};
metadata.frame_number = i;
std::vector<uint8_t> frame_data(kTestFrameSize, static_cast<uint8_t>(i));
ASSERT_TRUE(
producer.TryWriteFrame(metadata, frame_data.data(), frame_data.size())
) << "Failed to write frame " << i << " while filling buffer";
}
// At this point the ring buffer should be full; a timed write with a very short
// timeout is expected to fail with a timeout and update stats accordingly.
FrameMetadata timeout_metadata{};
timeout_metadata.frame_number = kTestBufferCount;
std::vector<uint8_t> timeout_frame(kTestFrameSize, 0xFF);
const auto timeout = std::chrono::milliseconds(1);
const bool write_result =
producer.WriteFrameWithTimeout(timeout_metadata,
timeout_frame.data(),
timeout_frame.size(),
timeout);
EXPECT_FALSE(write_result) << "WriteFrameWithTimeout should fail when buffer is full";
const auto stats = producer.GetStats();
EXPECT_EQ(stats.write_timeouts, 1u);
EXPECT_EQ(stats.dropped_frames, 1u);
The above test assumes the following existing APIs and types on ShmTransport and related classes:
bool ShmTransport::TryWriteFrame(const FrameMetadata&, const uint8_t* data, size_t size);bool ShmTransport::WriteFrameWithTimeout(const FrameMetadata&, const uint8_t* data, size_t size, std::chrono::milliseconds timeout);auto ShmTransport::GetStats() constreturning a struct withwrite_timeoutsanddropped_framesfields.- A
FrameMetadatatype with aframe_numberfield.
If your actual signatures or stats type differ, adjust the parameter ordering, type names, and the stats field accesses to match your existing implementation while preserving the test logic: fill the buffer, callWriteFrameWithTimeoutwith a very short timeout, assert it returnsfalse, and verify that the timeout- and drop-related counters increase as expected (possibly by checking for increment relative to a baseline snapshot rather than hard-coded1uif other tests may have already updated the stats).
| EXPECT_FALSE(header.is_valid()); | ||
| } | ||
|
|
||
| TEST_F(ShmTransportTest, FrameHeaderChecksum) |
There was a problem hiding this comment.
suggestion (testing): 校验和目前只在头部类型上做了单元测试，没有在传输层上（启用/禁用两种模式）做端到端验证。
当前只在隔离的情况下验证了 ShmFrameHeader 的 CRC。考虑到存在 config.enable_checksum 以及拷贝与零拷贝 API 在启用/禁用校验和时有不同路径，建议增加以下测试：（1）在 enable_checksum=true 时的集成测试——故意破坏共享内存中的 payload，验证 ReadFrame/AcquireReadBuffer 返回 ShmTransportError::ShmCorrupted；（2）在 enable_checksum=false 时的测试——确认同样的破坏不会触发校验和错误。这样可以覆盖传输层在损坏检测方面的控制流程。
建议实现如下：
// Verify default-constructed header is invalid.
EXPECT_FALSE(header.is_valid());
header.magic = ShmFrameHeader::MAGIC;
header.version = 1;
header.data_size = 100;
EXPECT_TRUE(header.is_valid());
header.magic = 0;
EXPECT_FALSE(header.is_valid());
}
// Integration: checksum enabled should detect corruption in the transport.
TEST_F(ShmTransportTest, ChecksumEnabledDetectsCorruption)
{
// Arrange: create a transport with checksum enabled.
ShmTransportConfig config;
config.enable_checksum = true;
// If there are additional required fields on ShmTransportConfig,
// they should be initialized here consistent with other tests.
auto [writer, reader] = CreateTestTransportPair(config); // uses existing helper in this file
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
// Write one frame.
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the underlying shared-memory payload before reading.
//
// We assume test helpers already expose a way to get at the raw shared-memory
// region for the last written frame, similar to other corruption-related tests.
// For example:
//
// auto* raw_frame = writer.DebugGetLastWrittenFrame();
// ASSERT_NE(raw_frame, nullptr);
// // Flip a byte inside the payload region without touching the header.
// auto* raw_bytes = static_cast<std::uint8_t*>(raw_frame->payload());
// raw_bytes[0] ^= 0xFF;
//
// Replace the following block with the actual helper/API used in this file
// to access and mutate the raw shared-memory frame:
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload()); // payload() should not include header
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF; // corrupt one byte
// Act: attempt to read back the frame via the transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: transport must detect corruption and surface ShmCorrupted.
EXPECT_EQ(read_size, 0u);
EXPECT_EQ(err, ShmTransportError::ShmCorrupted);
}
// Integration: checksum disabled should NOT treat the same corruption as an error.
TEST_F(ShmTransportTest, ChecksumDisabledDoesNotDetectCorruption)
{
// Arrange: create a transport with checksum disabled.
ShmTransportConfig config;
config.enable_checksum = false;
auto [writer, reader] = CreateTestTransportPair(config);
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the same way as in ChecksumEnabledDetectsCorruption.
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload());
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF;
// Act: read via transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: corruption is not flagged via checksum when disabled.
// Depending on implementation, the corrupted payload may be delivered as-is
// or another non-checksum error may occur (e.g., unrelated transport error),
// but it must not be ShmCorrupted due to checksum.
EXPECT_NE(err, ShmTransportError::ShmCorrupted);
// In most implementations, the read should succeed:
EXPECT_EQ(read_size, payload.size());
}
上述示例中的若干 helper 和 API（CreateTestTransportPair、WriteFrame、ReadFrame、DebugGetLastWrittenFrame、payload()、data_size 成员，以及 ShmTransportError::ShmCorrupted）是根据常见模式推断的，需与你的实际代码保持一致：
- 将
CreateTestTransportPair(config)替换为当前测试夹具中用于创建 writer/reader 对的实际 helper（例如CreateTransportPair、CreateClientServer，或通过ShmTransportTest::CreateTransport构造）。 - 根据真实签名调整
writer.WriteFrame(...)和reader.ReadFrame(...)调用。有的代码会使用bool WriteFrame(span<const uint8_t>, ShmTransportError&)，或者直接返回错误枚举而不用输出参数。 - 将
writer.DebugGetLastWrittenFrame()与payload()/data_size的用法替换为当前测试中真正用来访问底层共享内存帧的方式。如果尚无此类 helper，可按现有风格新增一个仅用于测试的访问接口（例如通过friend或#ifdef UNIT_TEST保护的 debug 方法）。 - 确认
ShmTransportError::ShmCorrupted名称无误；如果你的代码使用的是其它枚举名（例如ShmTransportError::kCorrupted或Error::kShmCorrupted），请相应替换。 - 如果存在单独的零拷贝 API（例如
AcquireReadBuffer/ReleaseReadBuffer），可以针对零拷贝路径再增加一组类似的测试，或者在上述测试中使用AcquireReadBuffer替代ReadFrame，视当前测试结构而定。
Original comment in English
suggestion (testing): Checksum is tested on the header type but not integrated through the transport in both enabled/disabled modes.
Currently this only validates ShmFrameHeader’s CRC in isolation. Given config.enable_checksum and the separate paths for checksum-enabled/disabled in copy and zero-copy APIs, please add: (1) an integration test with enable_checksum=true where you corrupt the shared-memory payload and verify ReadFrame/AcquireReadBuffer returns ShmTransportError::ShmCorrupted; and (2) a test with enable_checksum=false confirming the same corruption does not trigger a checksum error. This will cover the transport-level control flow around corruption detection.
Suggested implementation:
// Verify default-constructed header is invalid.
EXPECT_FALSE(header.is_valid());
header.magic = ShmFrameHeader::MAGIC;
header.version = 1;
header.data_size = 100;
EXPECT_TRUE(header.is_valid());
header.magic = 0;
EXPECT_FALSE(header.is_valid());
}
// Integration: checksum enabled should detect corruption in the transport.
TEST_F(ShmTransportTest, ChecksumEnabledDetectsCorruption)
{
// Arrange: create a transport with checksum enabled.
ShmTransportConfig config;
config.enable_checksum = true;
// If there are additional required fields on ShmTransportConfig,
// they should be initialized here consistent with other tests.
auto [writer, reader] = CreateTestTransportPair(config); // uses existing helper in this file
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
// Write one frame.
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the underlying shared-memory payload before reading.
//
// We assume test helpers already expose a way to get at the raw shared-memory
// region for the last written frame, similar to other corruption-related tests.
// For example:
//
// auto* raw_frame = writer.DebugGetLastWrittenFrame();
// ASSERT_NE(raw_frame, nullptr);
// // Flip a byte inside the payload region without touching the header.
// auto* raw_bytes = static_cast<std::uint8_t*>(raw_frame->payload());
// raw_bytes[0] ^= 0xFF;
//
// Replace the following block with the actual helper/API used in this file
// to access and mutate the raw shared-memory frame:
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload()); // payload() should not include header
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF; // corrupt one byte
// Act: attempt to read back the frame via the transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: transport must detect corruption and surface ShmCorrupted.
EXPECT_EQ(read_size, 0u);
EXPECT_EQ(err, ShmTransportError::ShmCorrupted);
}
// Integration: checksum disabled should NOT treat the same corruption as an error.
TEST_F(ShmTransportTest, ChecksumDisabledDoesNotDetectCorruption)
{
// Arrange: create a transport with checksum disabled.
ShmTransportConfig config;
config.enable_checksum = false;
auto [writer, reader] = CreateTestTransportPair(config);
const std::string payload = "0123456789ABCDEFGHIJ"; // 20 bytes
ShmTransportError err = ShmTransportError::Ok;
ASSERT_TRUE(writer.WriteFrame(payload.data(), payload.size(), err));
ASSERT_EQ(err, ShmTransportError::Ok);
// Corrupt the same way as in ChecksumEnabledDetectsCorruption.
auto* raw_frame = writer.DebugGetLastWrittenFrame();
ASSERT_NE(raw_frame, nullptr);
auto* raw_bytes =
static_cast<std::uint8_t*>(raw_frame->payload());
ASSERT_GE(raw_frame->data_size, 1u);
raw_bytes[0] ^= 0xFF;
// Act: read via transport API.
std::string received(payload.size(), '\0');
std::size_t read_size = reader.ReadFrame(received.data(), received.size(), err);
// Assert: corruption is not flagged via checksum when disabled.
// Depending on implementation, the corrupted payload may be delivered as-is
// or another non-checksum error may occur (e.g., unrelated transport error),
// but it must not be ShmCorrupted due to checksum.
EXPECT_NE(err, ShmTransportError::ShmCorrupted);
// In most implementations, the read should succeed:
EXPECT_EQ(read_size, payload.size());
}
The exact helpers and APIs used above (CreateTestTransportPair, WriteFrame, ReadFrame, DebugGetLastWrittenFrame, payload(), data_size member, and ShmTransportError::ShmCorrupted) are inferred from common patterns and the comment you provided:
- Replace
CreateTestTransportPair(config)with whatever the existing test fixture uses to instantiate a writer/reader pair (e.g.CreateTransportPair,CreateClientServer, or construction viaShmTransportTest::CreateTransport). - Adjust
writer.WriteFrame(...)andreader.ReadFrame(...)calls to match the real signatures. Some codebases usebool WriteFrame(span<const uint8_t>, ShmTransportError&)or return an error enum directly instead of using an out-parameter. - Replace
writer.DebugGetLastWrittenFrame()and thepayload()/data_sizeusage with the actual way the tests already access the underlying shared-memory segment/frame. If no such helper exists, add a small test-only accessor consistent with your existing style (e.g., afriendtest or a debug method guarded by#ifdef UNIT_TEST). - Ensure
ShmTransportError::ShmCorruptedis the correct enum name; if your code uses a different identifier (e.g.,ShmTransportError::kCorruptedorError::kShmCorrupted), substitute it accordingly. - If you have separate zero-copy APIs (e.g.,
AcquireReadBuffer/ReleaseReadBuffer), mirror the same corruption scenario in a second pair of tests for the zero-copy path, or extend one of the tests above to useAcquireReadBufferinstead ofReadFrame, depending on how your existing tests are structured.
| EXPECT_FALSE(block.is_valid()); | ||
| } | ||
|
|
||
| TEST_F(ShmTransportTest, ControlBlockBufferStatus) |
There was a problem hiding this comment.
suggestion (testing): ControlBlock 的缓冲区计数逻辑已经做了单元测试，但尚无端到端测试来验证环形缓冲区溢出时的丢帧计数。
为了补充单元测试，建议增加一个集成测试：在没有任何读取的前提下，故意写入超过 ring 容量的帧，随后同时检查 control_block_->dropped_frames 以及 ShmTransportStats 中对应字段，以确保溢出处理和统计在端到端路径上正确打通。
建议实现如下：
// 恢复 magic 但修改 version
block.magic = ShmControlBlock::MAGIC;
block.version = 999; // 错误版本
EXPECT_FALSE(block.is_valid());
}
TEST_F(ShmTransportTest, DroppedFramesOnRingOverflow)
{
// 选择一个很小的 ring 大小，方便在测试中触发溢出
const size_t kRingCapacityFrames = 4;
const size_t kFramesToWrite = 10; // 明显大于 capacity，确保发生溢出
// 初始化一个只写端（producer），不启动 reader，这样不会有消费，ring 会被写满
std::shared_ptr<ShmTransport> writer;
{
ShmTransportOptions opts;
opts.ring_capacity_frames = kRingCapacityFrames;
// NOTE: 这里假设 ShmTransportTest 提供了一个类似 CreateWriterTransport 的辅助方法，
// 或者可以直接构造 ShmTransport。根据现有代码调整下面的初始化方式。
writer = CreateWriterTransport(opts);
}
ASSERT_NE(writer, nullptr);
ASSERT_NE(control_block_, nullptr);
const uint64_t initial_dropped = control_block_->dropped_frames;
// 构造一帧最小有效 payload，用于重复写入
ShmFrame frame{};
frame.timestamp_ns = 1;
frame.data_size = 1;
frame.data[0] = 0xAB;
// 连续写入 kFramesToWrite 帧，不读取
for (size_t i = 0; i < kFramesToWrite; ++i)
{
const bool ok = writer->WriteFrame(frame);
// 写入操作本身不需要全部成功；在 ring 满时，WriteFrame 应触发丢帧计数
(void)ok;
}
// 控制块中的 dropped_frames 应该已经累加（至少大于初始值）
EXPECT_GT(control_block_->dropped_frames, initial_dropped);
// 统计信息中也应该反映同样的 dropped_frames 数量
ShmTransportStats stats{};
writer->GetStats(&stats);
EXPECT_EQ(stats.dropped_frames, control_block_->dropped_frames);
EXPECT_GT(stats.dropped_frames, 0u);
}
- 将
CreateWriterTransport(opts)替换为你在ShmTransportTest中实际使用的写端初始化方式（例如直接构造或使用现有工厂/辅助函数），并确保它使用opts.ring_capacity_frames来配置 ring 大小。 - 确认
control_block_是ShmTransportTest测试夹具中的成员指针/引用，并且其dropped_frames字段存在且会在写端发生溢出时递增；如字段名不同，请相应调整断言。 - 如果
ShmFrame的字段或构造方式不同（例如需要通过工厂或 builder 创建），请相应更新ShmFrame frame{}部分，但要保证每次写入的帧都被视为“有效帧”。 - 若
ShmTransport::WriteFrame或ShmTransport::GetStats的签名不同（例如GetStats()返回值而不是通过指针输出），请按实际签名调整调用方式，但要保留对dropped_frames一致性的断言。 - 如果
ShmTransportStats位于某个命名空间或需要特定头文件，请在文件顶部添加适当的#include和命名空间限定。
Original comment in English
suggestion (testing): ControlBlock buffer accounting is unit-tested, but there is no end-to-end test that asserts dropped frame counting when the ring overflows.
To complement the unit test, please add an integration test that deliberately overruns the ring (producer writes more frames than fit, with no reads) and then asserts both control_block_->dropped_frames and the corresponding field in ShmTransportStats to ensure overflow handling and stats are correctly wired end-to-end.
Suggested implementation:
// 恢复 magic 但修改 version
block.magic = ShmControlBlock::MAGIC;
block.version = 999; // 错误版本
EXPECT_FALSE(block.is_valid());
}
TEST_F(ShmTransportTest, DroppedFramesOnRingOverflow)
{
// 选择一个很小的 ring 大小，方便在测试中触发溢出
const size_t kRingCapacityFrames = 4;
const size_t kFramesToWrite = 10; // 明显大于 capacity，确保发生溢出
// 初始化一个只写端（producer），不启动 reader，这样不会有消费，ring 会被写满
std::shared_ptr<ShmTransport> writer;
{
ShmTransportOptions opts;
opts.ring_capacity_frames = kRingCapacityFrames;
// NOTE: 这里假设 ShmTransportTest 提供了一个类似 CreateWriterTransport 的辅助方法，
// 或者可以直接构造 ShmTransport。根据现有代码调整下面的初始化方式。
writer = CreateWriterTransport(opts);
}
ASSERT_NE(writer, nullptr);
ASSERT_NE(control_block_, nullptr);
const uint64_t initial_dropped = control_block_->dropped_frames;
// 构造一帧最小有效 payload，用于重复写入
ShmFrame frame{};
frame.timestamp_ns = 1;
frame.data_size = 1;
frame.data[0] = 0xAB;
// 连续写入 kFramesToWrite 帧，不读取
for (size_t i = 0; i < kFramesToWrite; ++i)
{
const bool ok = writer->WriteFrame(frame);
// 写入操作本身不需要全部成功；在 ring 满时，WriteFrame 应触发丢帧计数
(void)ok;
}
// 控制块中的 dropped_frames 应该已经累加（至少大于初始值）
EXPECT_GT(control_block_->dropped_frames, initial_dropped);
// 统计信息中也应该反映同样的 dropped_frames 数量
ShmTransportStats stats{};
writer->GetStats(&stats);
EXPECT_EQ(stats.dropped_frames, control_block_->dropped_frames);
EXPECT_GT(stats.dropped_frames, 0u);
}
- 将
CreateWriterTransport(opts)替换为你在ShmTransportTest中实际使用的 writer 端初始化方式（例如构造函数或现有的工厂/辅助函数），并确保它使用opts.ring_capacity_frames来配置 ring 大小。 - 确认
control_block_是ShmTransportTest测试夹具中的成员指针/引用，并且其dropped_frames字段存在且在写端溢出时会被递增；如果名字不同，请相应修改断言中的字段名。 - 如果
ShmFrame的字段或构造方式不同（例如需要通过工厂或 builder 创建），需要将ShmFrame frame{}部分改为你项目中已有的创建方式，但要保持所有写入帧都被视为“有效帧”。 - 如果
ShmTransport::WriteFrame或ShmTransport::GetStats的签名不同（例如GetStats()返回一个值而不是通过指针输出），请据实际签名调整调用方式，但保留对dropped_frames一致性的断言。 - 如果
ShmTransportStats位于命名空间内或需要头文件，请在文件顶部添加相应的#include和命名空间限定符。
| // 并发测试 | ||
| // ========================================================================== | ||
|
|
||
| TEST_F(ShmTransportTest, ConcurrentWriteRead) |
There was a problem hiding this comment.
suggestion (testing): ConcurrentWriteRead 验证了基本的并发行为，但没有覆盖多消费者语义及 reader 相关计数。
当前只测试了单生产者 + 单消费者。由于实现是 SPMC，并维护了 active_readers 计数，建议增加一个测试：让多个 consumer 附着到同一个段，驱动并发 ReadFrame/AcquireReadBuffer，并验证在 InitializeConsumer 和 Shutdown 时 active_readers 能正确更新，从而完整验证 SPMC 行为。
建议实现如下：
// 并发测试
// ==========================================================================
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
=======
// ==========================================================================
// 并发测试
// ==========================================================================
TEST_F(ShmTransportTest, ConcurrentWriteRead)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// TODO: 这里保留原有的并发读写测试逻辑（单生产者 + 单消费者），
// 具体实现依赖于现有的 Write/Read 流程，在本补丁中不做更改。
}
// 多消费者并发测试，验证 SPMC 语义以及 active_readers 统计
TEST_F(ShmTransportTest, MultiConsumerConcurrentReadUpdatesActiveReaders)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
constexpr int kConsumerCount = 4;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// 多个消费者附着到同一个段
std::vector<std::unique_ptr<ShmTransport>> consumers;
consumers.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
auto consumer = std::make_unique<ShmTransport>();
auto init_res = consumer->InitializeConsumer(config);
EXPECT_TRUE(init_res.has_value()) << "Consumer initialization failed for index " << i;
EXPECT_TRUE(consumer->IsInitialized());
consumers.emplace_back(std::move(consumer));
}
// 在此处并发驱动 ReadFrame/AcquireReadBuffer 来验证 SPMC 行为
//
// 为了不假设具体的读 API，这里仅演示多线程并发访问相同的 ShmTransport 实例集合，
// 具体的读调用需要根据现有接口补充（例如 ReadFrame 或 AcquireReadBuffer）。
//
std::atomic<bool> stop{false};
std::vector<std::thread> threads;
threads.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
threads.emplace_back([&, i] {
// 保护性检查，避免空指针
auto* consumer = consumers[i].get();
ASSERT_NE(consumer, nullptr);
// 在 stop 被置为 true 之前，持续执行读取逻辑
while (!stop.load(std::memory_order_acquire)) {
// 在这里调用具体的读取接口，例如：
//
// auto frame_res = consumer->ReadFrame(/*timeout*/);
// if (frame_res.has_value() && frame_res->has_value()) { ... }
//
// 或者：
//
// auto buf_res = consumer->AcquireReadBuffer();
// if (buf_res.has_value() && buf_res->has_value()) { ... }
//
// 为了保持此补丁与现有实现解耦，这里仅做一个短暂 sleep，
// 真正的读取逻辑需要在后续补充。
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
// 让并发读取线程运行一小段时间
std::this_thread::sleep_for(std::chrono::milliseconds(50));
stop.store(true, std::memory_order_release);
for (auto& t : threads) {
if (t.joinable()) {
t.join();
}
}
// 关闭部分消费者，验证 active_readers 统计会递减
// 注意：这里使用占位的 GetActiveReadersForTest()，需要根据实际实现替换。
//
// 期望语义：
// - InitializeConsumer 成功后 active_readers == kConsumerCount
// - 关闭 2 个消费者后 active_readers == kConsumerCount - 2
//
// 如果 ShmTransport 没有对外暴露该计数，需要通过测试辅助接口或友元方式访问共享头。
//
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount);
int shutdown_count = 2;
for (int i = 0; i < shutdown_count; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount - shutdown_count);
// 关闭剩余消费者与生产者，确保不会崩溃且资源全部释放
for (int i = shutdown_count; i < kConsumerCount; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
producer.Shutdown();
EXPECT_FALSE(producer.IsInitialized());
}
- 将关于
active_readers的断言替换为实际可用的访问方式：- 如果已有类似
size_t ShmTransport::GetActiveReadersForTest() const;的接口，请取消注释并使用真实函数名； - 如果没有公开接口，可考虑：
- 在共享内存头结构上新增只读测试辅助函数，或
- 将测试类声明为友元，以便直接访问
active_readers字段。
- 如果已有类似
- 在
MultiConsumerConcurrentReadUpdatesActiveReaders中，用当前实现中的真实读取 API 替换线程循环中的占位逻辑：- 如果存在
ReadFrame(...)或AcquireReadBuffer(...)，在循环中调用并对返回值做轻量验证（例如是否超时/是否有数据），从而真正驱动 SPMC 读路径。
- 如果存在
- 若测试文件尚未包含
<thread>、<atomic>、<vector>等头文件，请在文件顶部添加相应的#include。 - 如果项目中已有自己的线程/同步或测试工具（例如封装好的读写辅助函数），可以用这些工具重写并发部分，以保持与现有测试风格一致。
Original comment in English
suggestion (testing): ConcurrentWriteRead validates basic concurrency but not multi-consumer semantics or reader accounting.
This currently exercises only a single producer and single consumer. Since the implementation is SPMC and tracks active_readers, please add a test that attaches multiple consumers to the same segment, drives concurrent ReadFrame/AcquireReadBuffer, and verifies that active_readers is updated correctly on InitializeConsumer and Shutdown to fully validate the SPMC behavior.
Suggested implementation:
// 并发测试
// ==========================================================================
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
=======
// ==========================================================================
// 并发测试
// ==========================================================================
TEST_F(ShmTransportTest, ConcurrentWriteRead)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// TODO: 这里保留原有的并发读写测试逻辑（单生产者 + 单消费者），
// 具体实现依赖于现有的 Write/Read 流程，在本补丁中不做更改。
}
// 多消费者并发测试，验证 SPMC 语义以及 active_readers 统计
TEST_F(ShmTransportTest, MultiConsumerConcurrentReadUpdatesActiveReaders)
{
ShmTransport producer;
ShmTransportConfig config;
config.shm_name = test_shm_name_;
config.buffer_count = kTestBufferCount;
config.max_frame_size = kTestFrameSize;
constexpr int kConsumerCount = 4;
// 初始化生产者
auto result = producer.InitializeProducer(config);
EXPECT_TRUE(result.has_value()) << "Producer initialization failed";
EXPECT_TRUE(producer.IsInitialized());
// 多个消费者附着到同一个段
std::vector<std::unique_ptr<ShmTransport>> consumers;
consumers.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
auto consumer = std::make_unique<ShmTransport>();
auto init_res = consumer->InitializeConsumer(config);
EXPECT_TRUE(init_res.has_value()) << "Consumer initialization failed for index " << i;
EXPECT_TRUE(consumer->IsInitialized());
consumers.emplace_back(std::move(consumer));
}
// 在此处并发驱动 ReadFrame/AcquireReadBuffer 来验证 SPMC 行为
//
// 为了不假设具体的读 API，这里仅演示多线程并发访问相同的 ShmTransport 实例集合，
// 具体的读调用需要根据现有接口补充（例如 ReadFrame 或 AcquireReadBuffer）。
//
std::atomic<bool> stop{false};
std::vector<std::thread> threads;
threads.reserve(kConsumerCount);
for (int i = 0; i < kConsumerCount; ++i) {
threads.emplace_back([&, i] {
// 保护性检查，避免空指针
auto* consumer = consumers[i].get();
ASSERT_NE(consumer, nullptr);
// 在 stop 被置为 true 之前，持续执行读取逻辑
while (!stop.load(std::memory_order_acquire)) {
// 在这里调用具体的读取接口，例如：
//
// auto frame_res = consumer->ReadFrame(/*timeout*/);
// if (frame_res.has_value() && frame_res->has_value()) { ... }
//
// 或者：
//
// auto buf_res = consumer->AcquireReadBuffer();
// if (buf_res.has_value() && buf_res->has_value()) { ... }
//
// 为了保持此补丁与现有实现解耦，这里仅做一个短暂 sleep，
// 真正的读取逻辑需要在后续补充。
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
});
}
// 让并发读取线程运行一小段时间
std::this_thread::sleep_for(std::chrono::milliseconds(50));
stop.store(true, std::memory_order_release);
for (auto& t : threads) {
if (t.joinable()) {
t.join();
}
}
// 关闭部分消费者，验证 active_readers 统计会递减
// 注意：这里使用占位的 GetActiveReadersForTest()，需要根据实际实现替换。
//
// 期望语义：
// - InitializeConsumer 成功后 active_readers == kConsumerCount
// - 关闭 2 个消费者后 active_readers == kConsumerCount - 2
//
// 如果 ShmTransport 没有对外暴露该计数，需要通过测试辅助接口或友元方式访问共享头。
//
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount);
int shutdown_count = 2;
for (int i = 0; i < shutdown_count; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
// EXPECT_EQ(producer.GetActiveReadersForTest(), kConsumerCount - shutdown_count);
// 关闭剩余消费者与生产者，确保不会崩溃且资源全部释放
for (int i = shutdown_count; i < kConsumerCount; ++i) {
consumers[i]->Shutdown();
EXPECT_FALSE(consumers[i]->IsInitialized());
}
producer.Shutdown();
EXPECT_FALSE(producer.IsInitialized());
}
- 将测试中关于 active_readers 的断言替换为实际可用的访问方式：
- 如果有类似
size_t ShmTransport::GetActiveReadersForTest() const;的接口，请取消注释并使用真实函数名。 - 如果没有公开接口，请考虑：
- 在共享内存头结构上新增只读测试辅助函数，或
- 将测试类声明为友元，以便直接访问
active_readers字段。
- 如果有类似
- 在
MultiConsumerConcurrentReadUpdatesActiveReaders中，用当前实现中的真实读取 API 替换线程循环中的占位部分：- 若有
ReadFrame(...)或AcquireReadBuffer(...)，在循环中调用并对返回值做轻量验证（例如是否超时/有无数据），以真实驱动 SPMC 读路径。
- 若有
- 若测试文件尚未包含
<thread>,<atomic>,<vector>等头文件，请在文件顶部适当位置添加：
#include <thread>,#include <atomic>,#include <vector>. - 如果已有自己的线程/同步工具或测试工具（例如封装好的读写辅助函数），可将并发部分重写为调用这些工具，以保持与现有测试风格一致。
|
✅ Review Complete! The code review has been posted. View Review → |
![arknightsautomachine-ai-reviewer[bot]](https://avatars.githubusercontent.com/in/3271960?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Summary
该 PR 实现了 v0.2.0-alpha.3 计划的共享内存传输层（shm_transport），属于 L0 Sensing 层核心组件。代码质量高，实现了完整的跨平台（Windows/POSIX）零拷贝环形缓冲区传输，包含详细的统计信息、错误处理和全面的测试覆盖。
| File | Changes | Risk Level | Status |
|---|---|---|---|
shm_transport.hpp |
+990/-0 | 🟡 Medium | Added |
shm_transport.cpp |
+1291/-0 | 🟡 Medium | Added |
test_shm_transport.cpp |
+1431/-0 | 🟢 Low | Added |
THIRD_PARTY_LICENSES.md |
+3/-3 | 🟢 Low | Modified |
USER_AGREEMENT.md |
+454/-70 | 🟢 Low | Modified |
ROADMAP.md |
+1/-1 | 🟢 Low | Modified |
🏗️ Architecture Flow
flowchart LR
subgraph L0["L0 Sensing Layer"]
A[Frame Capture] --> B[ShmTransport<br/>Zero-Copy Ring Buffer]
B --> C[Shared Memory<br/>Windows/POSIX]
end
subgraph L1["L1 Perception Layer"]
D[Frame Consumer<br/>Python/C++]
end
C -.->|AcquireReadBuffer| D
🚨 Critical Issues (Must Fix)
无关键问题
⚠️ Warnings (Should Fix)
-
[File: ROADMAP.md, Line 95] 📝 ROADMAP Update Reminder: 此 PR 实现了
v0.2.0-alpha.3的共享内存传输功能。请确保在develop_plan/ROADMAP-S.md中也标记此项为已完成，并验证是否满足 ROADMAP-S 中的验收标准（P99 < 20ms 延迟、零内存泄漏等）。 -
[File: shm_transport.cpp, Line 60] CRC32 查找表生成使用了位运算
-(-(crc & 1))，虽功能正确但可读性较差，建议添加注释说明这是为了消除分支预测失败。 -
[File: shm_transport.cpp, Line 964] 测试代码中使用了
const_cast<std::byte*>(buffer.data.data())[0] ^= std::byte{0xFF};来模拟数据损坏，虽然测试代码中可以接受，但应添加注释说明这是故意的破坏性测试。
💡 Suggestions
-
性能优化:
WriteFrameWithTimeout中使用了std::this_thread::yield()忙等待（line 866），在高竞争场景下可能浪费 CPU。建议考虑使用条件变量或事件通知机制，但鉴于 L0 层实时性要求，当前实现可接受。 -
监控增强:
ShmTransportStats已包含延迟直方图数据，建议添加 P99/P95 百分位数计算（目前只有 min/max/avg），便于符合 ROADMAP-S 的验收标准验证。 -
文档同步: 建议更新
docs/ARCHITECTURE.md添加共享内存传输层的架构说明，特别是零拷贝 API 的使用模式（AcquireWriteBuffer->CommitWriteBuffer）。
📍 Inline Comments
🤖 AI Agent Prompt Generation
🤖 Copy this prompt for AI Agent
Please address the comments from this code review:
## Overall Comments
1. **ROADMAP Sync**: Verify develop_plan/ROADMAP-S.md is updated to mark shm_transport as completed for v0.2.0-alpha.3
2. **Code Comments**: Add explanatory comments for:
- CRC32 bit manipulation in generate_crc32_table() (line ~60)
- Intentional data corruption in test (line ~964)
3. **Optional**: Consider adding P99 latency calculation to ShmTransportStats for easier compliance verification with ROADMAP-S acceptance criteria
## File-specific Changes
- File: core/src/l0_sensing/shm_transport.cpp
- Line 60: Add comment explaining branchless CRC32 bit manipulation
- File: core/tests/test_shm_transport.cpp
- Line 964: Add comment explaining intentional data corruption for test
- File: develop_plan/ROADMAP-S.md
- Mark v0.2.0-alpha.3 shm_transport as completed if not already done
Thank you!Open Questions
- ROADMAP-S Verification: 是否已在 ROADMAP-S.md 中验证并标记 v0.2.0-alpha.3 共享内存传输验收标准（P99 < 20ms、Valgrind 无泄漏）？
- Integration Test: 是否已与
LockFreeFrameBuffer<T>（v0.2.0-alpha.1 接口）进行集成测试验证兼容性？
统计: 1 🚨 | 3
This review was generated by ArknightsAutoMachine AI Reviewer
| ### v0.2.0-alpha.3：传输层 | ||
|
|
||
| - [ ] `src/l0_sensing/shm_transport.cpp`：共享内存传输 | ||
| - [x] `src/l0_sensing/shm_transport.cpp`：共享内存传输 |
There was a problem hiding this comment.
📝 [ROADMAP] 请同步检查 develop_plan/ROADMAP-S.md 中 v0.2.0-alpha 的验收标准，确认共享内存传输满足 P99 < 20ms 和 Valgrind 无泄漏要求，并标记为已完成。
| for (std::uint32_t i = 0; i < 256; ++i) { | ||
| std::uint32_t crc = i; | ||
| for (std::uint32_t j = 0; j < 8; ++j) { | ||
| crc = (crc >> 1) ^ (0xEDB88320 & static_cast<std::uint32_t>(-(static_cast<std::int32_t>(crc) & 1))); |
There was a problem hiding this comment.
-(-(crc & 1)) 意图不够明显，建议添加注释：// Convert bit 0 to 0xFFFFFFFF (all ones) or 0x00000000 (all zeros) for branchless computation
| // 获取数据指针并损坏第一个字节 | ||
| auto& buffer = read_buffer->value(); | ||
| ASSERT_GE(buffer.data.size(), 1u); | ||
| const_cast<std::byte*>(buffer.data.data())[0] ^= std::byte{0xFF}; // 损坏数据 |
There was a problem hiding this comment.
// INTENTIONAL DATA CORRUPTION: Simulating bit flip to test checksum detection
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
添加共享内存传输层实现，支持Windows命名共享内存和POSIX shm，使用环形缓冲区+原子序列号+内存屏障算法，提供高性能零拷贝帧传输功能。同时更新测试文件和开发路线图标记该功能为已完成。
主要功能包括：
Summary by Sourcery
添加一个跨平台的零拷贝共享内存传输层，使用环形缓冲区进行帧传递并提供完善的测试，同时将相关路线图条目标记为已完成。
新功能：
增强：
文档：
测试：
Original summary in English
Summary by Sourcery
Add a cross-platform zero-copy shared memory transport layer with ring-buffer based frame passing and comprehensive tests, and mark the roadmap item as completed.
New Features:
Enhancements:
Documentation:
Tests: