feat: introduce the advanced publisher and subscriber #368
Conversation
YuanYuYuan
force-pushed
the
feat/advanced-pub-sub
branch
from
e0c490f to
1f9357f
Compare
|
Update: The deadlock fix has been merged. Mark this PR as ready for review. |
Test Failure
No new failure is found. |
|
Update: bump up zenoh-cpp version to include the memory leak fix eclipse-zenoh/zenoh-cpp#363. |
|
Will rebase once #405 is merged. |
…claration deadlock
YuanYuYuan
force-pushed
the
feat/advanced-pub-sub
branch
from
08b0906 to
b76ee4c
Compare
There was a problem hiding this comment.
Please see feedback inline.
I have the following additional questions
- It looks like this PR does not fix [Test Failure] rclcpp_action / test_client #326. But from my experience, it changes the behavior from deterministically failing to failing sometimes. ie, the test is now flaky. Do you observe the same?
- With
AdvancedSubscribersdo we no longer need to callGet()when we detect anAdvancedPublisherwith publication cache? This was needed to fix Subscribers sometimes not getting messages from topics using durability transient_local #263.
| @@ -22,7 +22,7 @@ set(ZENOHC_CARGO_FLAGS "--no-default-features$<SEMICOLON>--features=shared-memor | |||
| # - https://github.com/eclipse-zenoh/zenoh/pull/1696 (Fix SHM Garbage Collection (GC) policy) | |||
| ament_vendor(zenoh_c_vendor | |||
| VCS_URL https://github.com/eclipse-zenoh/zenoh-c.git | |||
| VCS_VERSION 61d8fcc136ce4ed36d921a32244da4f3d81a6097 | |||
| VCS_VERSION 85ca060fa4037239ca4102a3a61f96626cc6b434 | |||
There was a problem hiding this comment.
Please replace the comment above and below to include changes in this commit hash.
| @@ -189,15 +189,6 @@ class GraphCache final | |||
| /// Returns true if the entity is a publisher or client. False otherwise. | |||
| static bool is_entity_pub(const liveliness::Entity & entity); | |||
|
|
|||
| void set_querying_subscriber_callback( | |||
There was a problem hiding this comment.
You'll also need to delete querying_subs_cbs_ and update graph_cache.cpp to not iterate over querying_subs_cbs_
| zenoh::KeyExpr pub_ke(entity->topic_info()->topic_keyexpr_); | ||
| auto adv_pub_opts = zenoh::ext::SessionExt::AdvancedPublisherOptions::create_default(); | ||
| adv_pub_opts.publisher_detection = true; | ||
| adv_pub_opts.sample_miss_detection = true; |
There was a problem hiding this comment.
Looking at the documentation for this option, it sounds like this should be set true only if adapted_qos_profile.reliability == RMW_QOS_POLICY_RELIABILITY_RELIABLE? Could you clarify how this differs from setting pub_opts.reliability = Z_RELIABILITY_RELIABLE?
| pub_cache_opts.history = adapted_qos_profile.depth; | ||
| pub_cache_opts.queryable_complete = true; | ||
|
|
||
| std::string queryable_prefix = entity->zid(); | ||
| pub_cache_opts.queryable_prefix = zenoh::KeyExpr(queryable_prefix); | ||
|
|
||
| pub_cache = session->ext().declare_publication_cache( | ||
| pub_ke, std::move(pub_cache_opts), &result); |
There was a problem hiding this comment.
We previously needed to do this to fix #263. Can you confirm that we no longer need to explicitly call QueryingSubscriber::get() when we discover an AdvancedPublisher in the graph?
| adv_sub_opts.history->detect_late_publishers = true; | ||
| adv_sub_opts.history->max_samples = entity_->topic_info()->qos_.depth; | ||
| adv_sub_opts.recovery = zenoh::ext::SessionExt::AdvancedSubscriberOptions::RecoveryOptions{}; | ||
| adv_sub_opts.recovery->periodic_queries_period_ms = 1000; |
There was a problem hiding this comment.
Again looking at the documentation, it sounds like recovery options should only be enabled if reliability is RELIABLE. Could you clarify on the behavior?
With this PR, we replace the
QueryingSubscriberandPublicationCache(+ a normal publisher)with the
AdvancedSubscriberandAdvancedPublisher.NOTE: A deadlock is found when undeclaring advanced subscribers. It will be fixed once eclipse-zenoh/zenoh#1685 is merged and synced with zenoh-c.