Refactor: shard profiling collector accumulation#1312
Conversation
- Route PMU, DepGen, TensorDump, and ScopeStats callbacks through collector shard ids. - Accumulate collector output in shard-local state and merge during stop/export to avoid hot-path shared locks. - Add a ScopeStats unit test covering sharded collection before JSON export.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (11)
📝 WalkthroughWalkthroughCollector classes (PmuCollector in a2a3 and a5 platforms, DepGenCollector, ScopeStatsCollector, TensorDumpCollector) are refactored to support per-shard buffer accumulation. Each ChangesSharded Collector Refactor
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Profiler as ProfilerBase
participant Collector as Collector (Pmu/DepGen/ScopeStats/TensorDump)
participant Shard as Per-Shard Storage
participant Final as Merged Output
Profiler->>Collector: on_buffer_collected(info, collector_shard)
Collector->>Collector: normalize_collector_shard(collector_shard)
Collector->>Shard: append record/counter into shard
Note over Collector,Shard: repeated per ready buffer
Collector->>Collector: reconcile_counters() / export
Collector->>Shard: merge_collector_shards() / flush_collector_shards_to_csv()
Shard->>Final: consolidate into records_/collected_/final CSV
Collector->>Final: finalize() clears shard state
Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements sharded collection across several profiling collectors (PMU, DepGen, ScopeStats, and TensorDump) to eliminate mutex contention on the hot path by writing to shard-local files or vectors and merging them during stop-time reconciliation. However, the review highlights critical concurrency hazards across all modified collectors: on the hot path, there are non-thread-safe checks for empty shard containers or merged states that can trigger data races, memory corruption, or crashes when accessed concurrently by multiple threads. Since proper initialization is already performed during setup, these redundant and dangerous lazy-initialization checks should be removed.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if (csv_shard_files_.empty()) { | ||
| reset_collector_shards(); | ||
| } |
There was a problem hiding this comment.
Concurrency Hazard & Redundant Check
Multiple collector threads concurrently invoke on_buffer_collected (and thus append_buffer_to_csv_shard) on the hot path.
If csv_shard_files_ is empty, multiple threads will concurrently call reset_collector_shards(), which modifies shared vectors (csv_shard_paths_, csv_shard_files_, collector_counters_) without any synchronization. This leads to a severe data race, memory corruption, and potential crashes.
Since reset_collector_shards() is already called during init(), csv_shard_files_ is guaranteed to be initialized and sized correctly before any profiling starts. This lazy-initialization check is redundant and introduces a dangerous, non-thread-safe code path.
// csv_shard_files_ is guaranteed to be initialized by init()References
- Avoid redundant scans or loops to re-verify state flags if the flag is already maintained by the primary data-collection path. Such redundant checks can be replaced with a comment noting where the flag is set.
- When a component is accessed by multiple threads, its operations should be protected by a mutex to prevent data races.
| if (csv_shard_files_.empty()) { | ||
| reset_collector_shards(); | ||
| } |
There was a problem hiding this comment.
Concurrency Hazard & Redundant Check
Multiple collector threads concurrently invoke on_buffer_collected (and thus append_buffer_to_csv_shard) on the hot path.
If csv_shard_files_ is empty, multiple threads will concurrently call reset_collector_shards(), which modifies shared vectors (csv_shard_paths_, csv_shard_files_, collector_counters_) without any synchronization. This leads to a severe data race, memory corruption, and potential crashes.
Since reset_collector_shards() is already called during init(), csv_shard_files_ is guaranteed to be initialized and sized correctly before any profiling starts. This lazy-initialization check is redundant and introduces a dangerous, non-thread-safe code path.
// csv_shard_files_ is guaranteed to be initialized by init()References
- Avoid redundant scans or loops to re-verify state flags if the flag is already maintained by the primary data-collection path. Such redundant checks can be replaced with a comment noting where the flag is set.
- When a component is accessed by multiple threads, its operations should be protected by a mutex to prevent data races.
| if (records_by_collector_.empty()) { | ||
| reset_collector_shards(); | ||
| } | ||
| if (collector_shards_merged_) { | ||
| records_.clear(); | ||
| total_collected_ = 0; | ||
| collector_shards_merged_ = false; | ||
| } |
There was a problem hiding this comment.
Concurrency Hazard & Redundant/Incorrect State Reset
Multiple collector threads concurrently invoke on_buffer_collected (and thus append_buffer_records) on the hot path.
- Lazy Initialization Race: If
records_by_collector_is empty, multiple threads will concurrently callreset_collector_shards(), which modifies shared vectors without synchronization, causing a severe data race and memory corruption. - State Reset Race & Bug: If
collector_shards_merged_is true, multiple threads will concurrently clearrecords_and resetcollector_shards_merged_to false. Furthermore, this reset is incomplete because it does not clearrecords_by_collector_, meaning old records would still be retained and duplicated upon the next merge.
Since reset_collector_shards() is already called during init(), these vectors are guaranteed to be initialized and sized correctly before any profiling starts. These checks are redundant, incorrect, and introduce dangerous non-thread-safe code paths.
// records_by_collector_ is guaranteed to be initialized by init()References
- Do not unconditionally reset global synchronization or coordination states in a stale or null-state branch of a multi-threaded coordination protocol. Only release the thread's own election lock to prevent corrupting concurrently or subsequently initiated runs.
- Avoid redundant scans or loops to re-verify state flags if the flag is already maintained by the primary data-collection path. Such redundant checks can be replaced with a comment noting where the flag is set.
- When a component is accessed by multiple threads, its operations should be protected by a mutex to prevent data races.
| if (records_by_collector_.empty()) { | ||
| reset_collector_shards(); | ||
| } | ||
| if (collector_shards_merged_) { | ||
| records_.clear(); | ||
| total_collected_ = 0; | ||
| collector_shards_merged_ = false; | ||
| } |
There was a problem hiding this comment.
Concurrency Hazard & Redundant/Incorrect State Reset
Multiple collector threads concurrently invoke on_buffer_collected (and thus append_buffer_records) on the hot path.
- Lazy Initialization Race: If
records_by_collector_is empty, multiple threads will concurrently callreset_collector_shards(), which modifies shared vectors without synchronization, causing a severe data race and memory corruption. - State Reset Race & Bug: If
collector_shards_merged_is true, multiple threads will concurrently clearrecords_and resetcollector_shards_merged_to false. Furthermore, this reset is incomplete because it does not clearrecords_by_collector_, meaning old records would still be retained and duplicated upon the next merge.
Since reset_collector_shards() is already called during init(), these vectors are guaranteed to be initialized and sized correctly before any profiling starts. These checks are redundant, incorrect, and introduce dangerous non-thread-safe code paths.
// records_by_collector_ is guaranteed to be initialized by init()References
- Do not unconditionally reset global synchronization or coordination states in a stale or null-state branch of a multi-threaded coordination protocol. Only release the thread's own election lock to prevent corrupting concurrently or subsequently initiated runs.
- Avoid redundant scans or loops to re-verify state flags if the flag is already maintained by the primary data-collection path. Such redundant checks can be replaced with a comment noting where the flag is set.
- When a component is accessed by multiple threads, its operations should be protected by a mutex to prevent data races.
| if (collected_by_collector_.empty()) { | ||
| reset_collector_shards(); | ||
| } | ||
| if (collector_shards_merged_) { | ||
| collected_.clear(); | ||
| collector_shards_merged_ = false; | ||
| } |
There was a problem hiding this comment.
Concurrency Hazard & Redundant/Incorrect State Reset
Multiple collector threads concurrently invoke on_buffer_collected (and thus process_dump_buffer) on the hot path.
- Lazy Initialization Race: If
collected_by_collector_is empty, multiple threads will concurrently callreset_collector_shards(), which modifies shared vectors without synchronization, causing a severe data race and memory corruption. - State Reset Race & Bug: If
collector_shards_merged_is true, multiple threads will concurrently clearcollected_and resetcollector_shards_merged_to false. Furthermore, this reset is incomplete because it does not clearcollected_by_collector_, meaning old records would still be retained and duplicated upon the next merge.
Since reset_collector_shards() is already called during initialize(), these vectors are guaranteed to be initialized and sized correctly before any profiling starts. These checks are redundant, incorrect, and introduce dangerous non-thread-safe code paths.
// collected_by_collector_ is guaranteed to be initialized by initialize()References
- Do not unconditionally reset global synchronization or coordination states in a stale or null-state branch of a multi-threaded coordination protocol. Only release the thread's own election lock to prevent corrupting concurrently or subsequently initiated runs.
- Avoid redundant scans or loops to re-verify state flags if the flag is already maintained by the primary data-collection path. Such redundant checks can be replaced with a comment noting where the flag is set.
- When a component is accessed by multiple threads, its operations should be protected by a mutex to prevent data races.
1 participant
Summary
Testing
git diff --check upstream/main..HEADCCACHE_DIR=/tmp/ccache cmake --build tests/ut/cpp/build --target test_scope_stats_collector -j 8./tests/ut/cpp/build/test_scope_stats_collectorRefs #1281
Refs #1237