Add support for listener callbacks

[asorbini]

This PR implements the new listener-related functions that were introduced by #44.

Signed-off-by: Andrea Sorbini [email protected]

[asorbini]

Linux CI with only Connext enabled for initial validation:

[asorbini]

Linux CI with only Connext enabled for initial validation:

I investigated the failures, I believe the issues lie either in some test problem or they are unrelated to this PR (and they manifested because I ran CI with just this RMW enabled, which usually doesn't happen).

• rclcpp.TestSubscription.on_new_message_callback: in the last case, the test only waits while c3 == 0 while instead the loop should continue for as long as c3 < 3.
• rclcpp.TestWaitForAllAckedWithParm/TestPublisherWaitForAllAcked.check_wait_for_all_acked_with_QosPolicy/0: unrelated issue. Problem is with the test which only waits for 500ms to receive an ACK. Connext's default settings use a 3s periodic heartbeat, which may cause the ACK not to be received within the timeout. Increasing the wait to 6s makes the test pass consistently.
• rclpy.rclpy.test.test_client.TestClient.test_service_timestamps: unrelated problem. The issue lies with the test which doesn't take into account the possible return value (None, None) which indicates a successful take_request() with no request actually taken (i.e. an "empty polling" of the requests)..
• launch_testing_examples.launch_testing_examples.check_node_launch_test and launch_testing_examples.launch_testing_examples.set_param_launch: unrelated issue. It seems like the failures can be avoided by increasing the wait timeout or by configuring Connext to more quickly repair samples on topic "ros_discovery_info".
• Errors in rviz_rendering seem to be unrelated to the RMW layer because they occur also with other RMW implementations.

[ivanpauno]

rclcpp.TestSubscription.on_new_message_callback: in the last case, the test only waits while c3 == 0 while instead the loop should continue for as long as c3 < 3.

Could we fix that test?
It's directly related to the listener callbacks API, so it would be good to get it fixed to make sure that's the only issue.

[ivanpauno]

Code LGTM!

[asorbini]

Could we fix that test? It's directly related to the listener callbacks API, so it would be good to get it fixed to make sure that's the only issue.

I opened ros2/rclcpp#1900 to fix the unit tests.

[asorbini]

Could we fix that test? It's directly related to the listener callbacks API, so it would be good to get it fixed to make sure that's the only issue.

I opened ros2/rclcpp#1900 to fix the unit tests.

Also opened ros2/rclcpp#1901 to fix the issue with the wait_for_acknowledgments() test.

[alsora]

I just want to double check here, as I don't really know the underlying connext APIs.
When is a sample marked as read?

If this notify_new_data function is invoked whenever the DDS receives a message, I would expect unread_samples to always be 1.

[asorbini]

I just want to double check here, as I don't really know the underlying connext APIs.
When is a sample marked as read?

In general in DDS, samples have a "sample state" associated with them, which is "NOT_READ" when they are added to a DataReader's cache, and it transitions to "READ" once they have been returned at least once by a call to DataReader::read() or DataReader::take() (in the case of take() the "READ" state cannot be really observed because the sample is removed from the cache, but theoretically the sample transitions...).

Both read() and take() take a "sample state" argument to possibly filter the samples in the cache (in addition to two more states, "Instance" and "View". So the count_unread_samples() function takes advantage of this to scan the reader's cache and only ever receive samples that haven't yet been observed by the "application layer" (i.e. the RMW/ROS2).

The RMW will then take()'s all the samples out of the cache (when the upper layers call rmw_take_message()) and it does to without constraints on the "sample state".

If this notify_new_data function is invoked whenever the DDS receives a message, I would expect unread_samples to always be 1.

There is no guarantee in DDS that the callback will be invoked for every samples. In fact the protocol explicitly allows an implementation to coalesce multiple events in a single notification. This is true for every of the observable events, not just DATA_AVAILABLE.

Short of explicit guarantees of an implementation, it is not correct to assume that a DDS listener will be notified for every received sample. The listener must actually inspect the reader cache to determine how many samples are available.

[asorbini]

Btw, I ended up with this implementation that counts the messages in the reader queue upon DATA_AVAILABLE, because the implementation from rmw_fastrtps_cpp which assumes only 1 sample didn't work for Connext (sometimes the listener would receive only one notification while having multiple samples in the queue)

[alsora]

Thank you for the detailed description, this makes sense!

[asorbini]

One thing I'd like to point out about this implementation is that there is always a possibility that some samples that were notified to the callback will not actually be read by the application layer because of QoS settings (mainly KEEP_LAST history). The samples could be counted but then be pushed out of the cache by newer samples before they can be take()n out of it.

[cwecht]

I this true for both connext dds variantions, Pro and Micro? I'm just asking because the call to count_unread_samples adds quite some overhead here. I don't remember the exact numbers, but I think it was at least 1µs. It's not that much, but it is a gain, which can add up quite quickly for latency sensitive applications with "sufficiently large" chains of subscriptions.

[alsora]

Looks good to me, I just want to confirm how the notify_new_data function works exactly.

[asorbini]

@clalancette @ivanpauno can this PR be merged now even with the freeze? It doesn't change API/ABI.

[audrow]

@clalancette @ivanpauno can this PR be merged now even with the freeze? It doesn't change API/ABI.

Hi @asorbini, I don't think we should merge this because it is a feature addition and the freeze is in place. If it doesn't break API/ABI, it should be able to be added in a patch release.

[cwecht]

Did anyone ever test this PR with irobot's events executors? https://github.com/irobot-ros/events-executor
If I run their tests with rmw_connextdds enabled and these changes here applied to the rmw_connextdds humble branch, I get these kinds of errors/warnings:

This PR is needed though, if one is trying to use events executors with connextdds. It would be nice if someone could have a look at this issue.

[cwecht]

#109 fixes the issue with the events executors.

[alsora]

@asorbini FYI we recently opened the PR for the events-executor in rclcpp and we currently have disabled it from running with connext.
To test this PR you can easily remove the "if connext -> skip" parts of the tests
ros2/rclcpp#2140

for example here (and in every test in this file https://github.com/ros2/rclcpp/pull/2140/files#diff-e1ac0eadd712d463d84ffc15b59e669fbf53f75c6f81e4c06f03f4d9c91b0ceaR47)

[asorbini]

@alsora thank you for the heads up, that's very useful to validate these changes.

Do you have a .repos file that I can use to test that PR or is it just those rclcpp changes + current rollng?

[alsora]

rolling + that PR is sufficient to test, no other changes to other repos are needed.
In the last CI run, all the unit-tests passed ros2/rclcpp#2140 (comment).
Note that there is 1 flaky test that we are currently investigating (TEST_F(TestEventsExecutor, destroy_entities)).

Feel free to share the results if you need help debugging failures.

[cwecht]

What's the status of this PR? Given that ros2/rclcpp#2155 is merged now, it would be really greate to have this PR "officially" approved and merged in order to use rclcpp with ConnextDDS to it's full potential out of the box.

[clalancette]

What's the status of this PR? Given that ros2/rclcpp#2155 is merged now, it would be really greate to have this PR "officially" approved and merged in order to use rclcpp with ConnextDDS to it's full potential out of the box.

At this point, it needs another approval (there have been changes since the previous ones), and CI run on it. I can do the latter, but I'd prefer @asorbini or @alsora did another review/approval first.

[alsora]

I'm not really familiar with connext, but overall the PR looks sane.

From my perspective I would suggest to create a rclcpp branch where the checks that disable events-executor with connext are removed, and then, if it passes all unit-tests, it should be good to go.

[cwecht]

@alsora here is the PR to enable the callback groupd tests in rclcpp again: ros2/rclcpp#2182

[cwecht]

The tests reenabled in ros2/rclcpp#2182 are now succeeding. Although they seem to be somewhat flakly, I'd suppose, that this PR should be mergable now?

[clalancette]

Although they seem to be somewhat flakly, I'd suppose, that this PR should be mergable now?

We'd really prefer to not merge in "somewhat flaky" tests. We are just down to handful of those now, and I'd like not to increase that number.