Add support for content-filtered topics

[iuhilnehc-ynos]

related to ros2/design#282

Signed-off-by: Chen Lihui [email protected]

[fujitatomoya]

overall it seems to be okay. test would be ideal.

[asorbini]

Things look pretty good.

My main concern is with functions RMW_Connext_Subscriber::set_cft_expression_parameters/get_cft_expression_parameters which, besides the comments I added inline, have a few other issues that will need to be resolved:

• Function RMW_Connext_Subscriber::set_cft_expression_parameters seems to duplicate some logic from RMW_Connext_Subscriber::finalize and RMW_Connext_Subscriber::create, in order to delete and recreate the underlying DDS_DataReader. Instead of duplicating this logic, I would refactor the other two functions and rely on them here.
  • RMW_Connext_Subscriber::finalize could be made to take an argument const bool reset_cft, which, if true, would force the function to perform only a subset of the operations it would usually perform (ie. only: invalidate the status condition wrapper, return loaned samples, delete content filter, delete data reader, and -- not currently done by finalize() -- delete the status condition wrapper).
  • Some of the logic in RMW_Connext_Subscriber::create could be refactored into a separate RMW_Connext_Subscriber::initialize_datareader function to encapsulate the logic required to determine DataReaderQos and create the DataReader. The function might have a parameter (e.g. const bool on_cft_reset) to make it behave differently if needed.
  • These are just suggestions, and I'm open to other ways to refactor the code, but I'd like to try to extract the logic to "delete and re-initialize the DataReader" into a separate function, hopefully trying to avoid duplication with the create/finalize functions as much as possible.
• Another problem in these functions is that, as it is, they will not compile with RTI Connext DDS Micro, because this version doesn't support content-filtered topic, thus symbols like DDS_ContentFilteredTopic and DDS_ContentFilteredTopic_get_filter_expression will not be available.
  • I suggest to introduce at least two new functions, defined in dds_api.hpp, and implemented in dds_api_ndds.cpp and dds_api_rtime.cpp to abstract the logic to get and set the filter expression on a content filtered topic object using only symbols available in both implementations, e.g.:
  • rmw_ret_t rmw_connextdds_get_cft_filter_expression(DDS_TopicDescription * const topic_desc, char ** const expr_out)
  • rmw_ret_t rmw_connextdds_set_cft_filter_expression(DDS_TopicDescription * const topic_desc, struct DDS_StringSeq * const cft_params)

Once these issues are resolved, I think we will be good to merge (or very very close to it :) )

[asorbini]

I just pushed a commit (598979b) which introduced some conflicts with this PR because it refactors the code related to WaitSets and Conditions.

There are now two different implementations of WaitSets and Conditions (which can be selected at compile-time), but as far as this PR is concerned, the only relevant difference I think is a small change in the signature of RMW_Connext_StatusCondition::RMW_Connext_StatusCondition(...).

I apologize for the additional work, but hopefully the conflicts should be easy to fix. Let me know if you have any questions.

[iuhilnehc-ynos]

I just pushed a commit (598979b) which introduced some conflicts with this PR because it refactors the code related to WaitSets and Conditions.

There are now two different implementations of WaitSets and Conditions (which can be selected at compile-time), but as far as this PR is concerned, the only relevant difference I think is a small change in the signature of RMW_Connext_StatusCondition::RMW_Connext_StatusCondition(...).

I apologize for the additional work, but hopefully the conflicts should be easy to fix. Let me know if you have any questions.

Thanks for the notification. Even if I fixed all the issues, I think we should not merge this PR, because it leads to a compilation problem since something update about rmw is not merged.

[asorbini]

Thanks for the notification. Even if I fixed all the issues, I think we should not merge this PR, because it leads to a compilation problem since something update about rmw is not merged.

I had actually missed it, but yes, this PR can only be merged after changes to ros2/rmw have been merged to introduce field filter_expression in the subscription options.

Which actually lead me to request another change to make sure that the code continues to build with older ROS2 releases.

[iuhilnehc-ynos]

hopefully the conflicts should be easy to fix. Let me know if you have any questions.

I have rebased from the latest branch of master.

[asorbini]

Things are looking good, thank you @iuhilnehc-ynos for addressing all my many comments! (only one left)

[asorbini]

I created #15 to get rid of all "feature flags" in master. If you want, I think you could rebase your changes on asorbini/rolling-no-ff (or rebase them from master, once that PR is merged).

[iuhilnehc-ynos]

rebase them from master, once that PR is merged

I'll wait to rebase from master until that PR is merged because there might exist other updates in the future.

[asorbini]

Latest changes look good, holding off on approval until we can rebase on top of master after merging #15.

[asorbini]

#15 was merged, so you can rebase from master now and remove the feature flags

[asorbini]

All good 🚀

I've been thinking is_cft_supported might be better renamed to is_cft_enabled, but I won't block the PR for that (since that is also related to the API changes in rmw).

[asorbini]

@iuhilnehc-ynos I have created branch asorbini/content-filtered-topics by rebasing your commits on top of master, and adding 92bad59.

This commit updates the synchronization of the set/get CFT parameters functions to make sure that they don't overlap with any concurrent WaitSet::wait() (e.g. performed by an executor to which the subscription is attached).

If you want, you can rebase your topic-content_filtered_topic-rebased branch on top of this new branch and we can then run some CI to validate this feature and finally merge it.

Btw, you will also need to rebase ros2/rmw#302 in order to build the new branch. I have done so in branch asorbini/content-filtered-topic of asorbini/rmw

[iuhilnehc-ynos]

@asorbini

To make a function exec_when_detached on 92bad59 seems reasonable.

I tested with my new branch iuhilnehc-ynos/topic-content_filtered_topic-updated based on the asorbini/content-filtered-topics, but rmw_wait failed to trigger events.

rmw_connextdds/rmw_connextdds_common/include/rmw_connextdds/rmw_waitset_std.hpp

Lines 94 to 107 in 92bad59

	if (this->in_ea) {
	this->detached_condition.wait(
	lock, [this]() {
	return !this->in_ea;
	});
	}
	already_active = check_trigger();
	if (!already_active) {
	this->waitset_mutex = waitset_mutex;
	this->waitset_condition = waitset_condition;
	}
	this->in_ea = true;

The attach of a RMW_Connext_Condition coule be called twice on RMW_Connext_WaitSet::attach

1. rmw_connextdds/rmw_connextdds_common/src/common/rmw_impl_waitset_std.cpp

   Lines 253 to 258 in 92bad59

   	RMW_Connext_SubscriberStatusCondition * const cond = sub->condition();
   	cond->attach(
   	&this->mutex_internal, &this->condition, wait_active,
   	[cond]() {
   	return cond->triggered_data;
   	});

2. rmw_connextdds/rmw_connextdds_common/src/common/rmw_impl_waitset_std.cpp

   Lines 303 to 308 in 92bad59

   	RMW_Connext_SubscriberStatusCondition * const cond = sub->condition();
   	cond->attach(
   	&this->mutex_internal, &this->condition, wait_active,
   	[cond, event]() {
   	return cond->has_status(event->event_type);
   	});

It leads the second attach blocked on

rmw_connextdds/rmw_connextdds_common/include/rmw_connextdds/rmw_waitset_std.hpp

Lines 95 to 98 in 92bad59

	this->detached_condition.wait(
	lock, [this]() {
	return !this->in_ea;
	});

Do you have any suggestions to fix it?

[asorbini]

@iuhilnehc-ynos I'm looking into reproducing this issue and digging into solution, but I'm not sure if I got all required packages (and branches). It's been a while since I worked on this PR so I'm a bit rusty on the code and I can't answer your questions without some testing :)

Would it be possible for you to provide a .repos manifest with all repositories/branches that you are using for testing?
Also, instructions on how you are doing the test would be useful (I'm guessing that you are using a modified listener from demo_nodes_cpp, but confirmation would be useful).

Hopefully I already got all I need and I'll be able to come to an answer soon (building various packages, at the moment).

[asorbini]

Yeah, I'm gonna need some help in getting the test environment back up. As soon as you provide that repos file, I'll give it another try.

[iuhilnehc-ynos]

@asorbini

Thanks for your response.
I'll provide ros2 repos files to run a demo for CFT, please wait a moment.

[iuhilnehc-ynos]

@asorbini

I have updated https://github.com/iuhilnehc-ynos/ros2/tree/topic-debug-connextdds-cft.

After building, you can try running the following commands (the original demo_nodes_cpp of ros2) at first

// console A
$ RMW_IMPLEMENTATION=rmw_connextdds ros2 run demo_nodes_cpp talker    (blocking, so there is no INFO message)

// console B
$ RMW_IMPLEMENTATION=rmw_connextdds ros2 run demo_nodes_cpp listener (it can't receive the message obviously)

[asorbini]

@iuhilnehc-ynos thank you! I was able to get everything built and I'm about to dig into it, I should have something later today. FYI I had to comment out rcl/test_subscription_content_filtered_topic_options.cpp from rcl/CMakeLists.txt since the file is not present (it is present in the rmw package coincidentally).

[asorbini]

@iuhilnehc-ynos Before I could start debugging the concurrency issue, I had a talk with @GerardoPardo and we identified an interesting alternative approach which would eliminate the problem altogether. So I went ahead, implemented it, and created #68 with the resulting changes.

I tested them with the cft-demo application (from https://github.com/iuhilnehc-ynos/ros2/blob/topic-debug-connextdds-cft/cft-demo.repos) and using a ROS 2 tree built from the ros2.repos you provided (only swapped rmw_connextdds for branch asorbini/cft on the main repository) and everything seemed to work as expected.

If you're ok with it, let's see if this approach can simplify our lives and continue the conversation on that PR.

[iuhilnehc-ynos]

Use #68 instead of this PR, close it.