Mitigate discovery race condition between clents and services #132

asorbini · 2023-09-25T19:45:30Z

This PR introduces changes meant to reduce the impact of a race condition between clients and services.

The race condition is described in detail by this comment, but to summarize:

It is possible for the service's request DataReader to receive a request from the client, handle it, and try to write the response before the service's reply DataWriter has actually matched the client's reply DataReader. If this is the case, the response will be lost.
The data race and loss of reply message may also occur in the case of an "asymmetric discovery" between the client's reply DataReader and the service's reply DataWriter, where the latter has discovered and matched the former, but not viceversa.

The only way to fully resolve this race condition is described by the DDS-RPC 1.0 specification, section 7.6.2.2:

[..] To ensure robustness, service discovery must ensure that no replies are lost due to race conditions in the discovery process of the underlying DDS entities [..]:

The client-side DataWriter (request DataWriter) shall not send out any request to a service DataReader unless the client-side DataReader (reply DataReader) has discovered and matched the corresponding service DataWriter (reply DataWriter).

The service-side DataReader (request DataReader) shall reject requests from a client DataWriter (request
DataWriter) as long as the service-side DataWriter (reply DataWriter) has not discovered and matched the
corresponding client-side DataReader (reply DataReader).

The proposed solution follows the approach implemented by rmw_fastrtps_cpp:

The client (ab)uses the `related_sample_identity" metadata field of the request message to communicate its DataReader's GUID to the service.
The service keeps track of discovered pairs of [DataReader, DataWriter] belonging to remote clients.
Before sending a reply back to the client, the service waits for the client's DataReader to be matched locally, or throws a timeout error if that doesn't happen within the configured period (read from DataWriterQos::reliability::max_blocking_time).

These changes, paired with the implementation of rmw_service_server_is_available(), provide a partial implementation of the solution prescribed by DDS-RPC, with the following limitations:

RMW_Connext_Client::::is_service_available() only verifies that for every remote subscription matched by the local request DataWriter, there is a matched remote publication with the same "GUID prefix" (i.e. belonging to the same remote DomainParticipant) on the local reply DataReader. This test may lead to false positives if a remote DomainParticipant has created two or more "services" (i.e. pairs of request DR/reply DW) with the same (topic) name, e.g. in the case of a remote process with two ROS 2 Nodes that create a service with the same name. This is a very likely case for "1-to-many" services (where a client makes a request on many services at once).
The client must use rmw_service_server_is_available() (e.g. by calling of rclcpp::wait_for_service()) before sending a request to make sure it'll (most likely) receive at least one reply from a service, but because of the false positives in RMW_Connext_Client::is_service_available(), it is still possible for the client to end up with "inconsistent" matches and no reply.
A more robust solution requires each service to communicate their endpoints' GUIDs through discovery so that clients may correctly associate each pair of discovered endpoints and then make sure they are both matched before sending a request.
This information is currently encoded by the ROS Graph, by means of an endpoint's node and topic name, but it is not easily queried and correlated with the client's matches. It also relies on the assumption that a node can't create two services with the same name (not sure if that is the case...).

asorbini

Added some comments to help with reviewing.

I will now double-check the changes to RMW_Connext_Client::take_response(), then run CI.

asorbini · 2023-09-25T19:46:46Z

rmw_connextdds_common/include/rmw_connextdds/dds_api.hpp

@@ -46,6 +46,10 @@ enum RMW_Connext_MessageType
  RMW_CONNEXT_MESSAGE_REPLY
 };

+#ifndef DDS_GUID_INITIALIZER


Definition moved to this common header from rmw_impl.cpp

asorbini · 2023-09-25T19:48:38Z

rmw_connextdds_common/include/rmw_connextdds/dds_api.hpp

@@ -281,4 +285,10 @@ rmw_connextdds_get_cft_filter_expression(
  rcutils_allocator_t * const allocator,
  rmw_subscription_content_filter_options_t * const options);

+rmw_ret_t
+rmw_connextdds_is_subscription_matched(


New internal API function to check if a remote reader is currently matched by a local writer.

Only implemented for Connext Pro, and only used with the "extended" RPC mapping. The function is never called by the Micro implementation (because it doesn't support the "exteded" RPC mapping).

asorbini · 2023-09-25T19:49:23Z

rmw_connextdds_common/include/rmw_connextdds/resource_limits.hpp

@@ -55,4 +55,13 @@
 #define RMW_CONNEXT_LIMIT_KEEP_ALL_SAMPLES              1000
 #endif /* RMW_CONNEXT_LIMIT_SAMPLES_MAX */

+#ifndef RMW_CONNEXT_LIMIT_DEFAULT_BLOCKING_TIME_DEFAULT
+#define RMW_CONNEXT_LIMIT_DEFAULT_BLOCKING_TIME_DEFAULT     100000 /* 100ms */


Static limit used by RMW_Connext_Publisher::max_blocking_time when the value of DataWriterQos::reliability::max_blocking_time is "auto".

asorbini · 2023-09-25T19:49:32Z

rmw_connextdds_common/include/rmw_connextdds/resource_limits.hpp

+#endif /* RMW_CONNEXT_LIMIT_DEFAULT_BLOCKING_TIME_DEFAULT */
+
+#ifndef RMW_CONNEXT_LIMIT_DEFAULT_BLOCKING_TIME_INFINITE
+#define RMW_CONNEXT_LIMIT_DEFAULT_BLOCKING_TIME_INFINITE    (3600000000LU * 24LU * 365LU) /* 1 year */


Static limit used by RMW_Connext_Publisher::max_blocking_time when the value of DataWriterQos::reliability::max_blocking_time is "infinite".

asorbini · 2023-09-25T19:51:58Z

rmw_connextdds_common/include/rmw_connextdds/type_support.hpp

@@ -45,6 +45,7 @@ struct RMW_Connext_RequestReplyMessage
 {
  bool request;
  rmw_gid_t gid;
+  rmw_gid_t writer_gid;


Enhance this data structure with an extra field so that we can store both reader and writer GUIDs on the service.

asorbini · 2023-09-25T19:54:40Z