Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR updates OnDemand’s documentation and strengthens pub/sub robustness and throughput by adding participant GUID-based stale-drop protection + reconnect recovery, improving VarStore batch write behavior, and adding regression tests for rapid restart scenarios.
Changes:
- Add GUID tracking on both pub/sub sides to ignore stale DROPPED events and support reconnect recovery (restore cached TableDefine + resend subscriptions).
- Optimize data-path performance by switching sub-side data ingestion to VarStore
write_batch()and improving lite-variable lookup to O(1). - Refresh/expand architecture/design docs and add new Pub/Sub rapid-restart test cases (TC32–TC34).
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Rewrites top-level architecture/design documentation with expanded sections and diagrams. |
| docs/variable_store_design.md | Major expansion of VarStore design doc (seqlock + guards + dirty tracking + resizing). |
| docs/technical_challenges.md | New doc capturing key scaling/concurrency issues and corresponding solutions. |
| docs/onDemand_overview.md | Removed (content effectively consolidated into new README/docs). |
| docs/ondemand_architecture.md | Removed (replaced by updated docs). |
| docs/design_ondemand_lite_broadcast.md | Removed (superseded by updated documentation). |
| docs/design_ondemand_definitions.md | Removed (superseded by updated documentation). |
| core/ondemand/variable_store.h | Refactors write_batch() to pre-scan + batch expand + single OpGuard write pass. |
| core/ondemand/ondemand_pubsub_test.cc | Adds TC32–TC34 to validate stale-drop protection in rapid restart scenarios. |
| core/ondemand/on_demand_sub.h | Adds subscription replay storage, GUID tracking, and PubTableDefine cache declarations. |
| core/ondemand/on_demand_sub.cc | Implements batch ingestion, PubTableDefine caching, GUID stale-drop ignore, and resubscription on reconnect. |
| core/ondemand/on_demand_pub.h | Lite bucket index changes (O(1) lookup) and adds GUID tracking members. |
| core/ondemand/on_demand_pub.cc | Updates lite index maintenance, lazy-create path, and pub-side stale-drop ignore. |
| core/main.cc | Changes demo defaults and increases callback logging detail. |
| common/txdds_wrapper/txdds_listeners.h | Extends ParticipantInfo to include GUID string extraction. |
| common/fastdds_wrapper/fastdds_listeners.h | Extends ParticipantInfo to include GUID string extraction (guidPrefix). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+344
to
351
| /*批量写入：一次 OpGuard 覆盖所有变量*/ | ||
| if (!writeIds.empty()) { | ||
| WriteResult result = varStore_.write_batch( | ||
| writeIds.data(), writeDatas.data(), writeSizes.data(), writeIds.size()); | ||
| if (result == WriteResult::SUCCESS || result == WriteResult::NOT_READY) { | ||
| written = writeIds.size(); | ||
| } | ||
| } |
Comment on lines
+1401
to
+1405
| { | ||
| std::lock_guard<std::mutex> lock(subscriptionItemsMutex_); | ||
| subscriptionItems_.clear(); | ||
| } | ||
|
|
Comment on lines
+1138
to
+1143
| auto res = varStore_.write_batch(ids, datas, sizes, count); | ||
| if (res != WriteResult::SUCCESS) { | ||
| ONDEMANDLOG(error) | ||
| << "setVarDataBatch: write_batch failed with result=" << static_cast<int>(res); | ||
| } | ||
| return res; |
| uint32_t maxNodeNum_; // 最大订阅者数，超过后拒绝新订阅者注册 | ||
| std::unordered_map<uint64_t, uint32_t> nodeVarCount_; // nodeHash -> 已订阅变量数 | ||
|
|
||
| /* participant GUID 跟护：participantName -> guidPrefix*/ |
Comment on lines
+208
to
+210
| LOG(info) << "Callback: var=" << dst.var_name << " type=" << dst.type_name | ||
| << " version=" << dst.type_version << " size=" << dst.data_size | ||
| << " ts_ms=" << dst.timestamp_ms; |
Comment on lines
+275
to
279
| /* Step 1: relaxed 扫描，收集需要扩容的变量（不加锁） */ | ||
| std::vector<uint32_t> expandIds; | ||
| std::vector<uint32_t> expandSizes; | ||
| for (size_t i = 0; i < count; ++i) { | ||
| uint32_t id = ids[i]; |
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
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.
No description provided.