Optimize QoS to improve responsiveness of reliable endpoints

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

Author: asorbini

README.md

@@ -151,6 +151,7 @@ variables.
 - [RMW_CONNEXT_CYCLONE_COMPATIBILITY_MODE](#RMW_CONNEXT_CYCLONE_COMPATIBILITY_MODE)
 - [RMW_CONNEXT_DISABLE_LARGE_DATA_OPTIMIZATIONS](#RMW_CONNEXT_DISABLE_LARGE_DATA_OPTIMIZATIONS)
 - [RMW_CONNEXT_DISABLE_FAST_ENDPOINT_DISCOVERY](#RMW_CONNEXT_DISABLE_FAST_ENDPOINT_DISCOVERY)
+- [RMW_CONNEXT_DISABLE_RELIABILITY_OPTIMIZATIONS](#RMW_CONNEXT_DISABLE_RELIABILITY_OPTIMIZATIONS)
 - [RMW_CONNEXT_ENDPOINT_QOS_OVERRIDE_POLICY](#RMW_CONNEXT_ENDPOINT_QOS_OVERRIDE_POLICY)
 - [RMW_CONNEXT_INITIAL_PEERS](#RMW_CONNEXT_INITIAL_PEERS)
 - [RMW_CONNEXT_LEGACY_RMW_COMPATIBILITY_MODE](#RMW_CONNEXT_LEGACY_RMW_COMPATIBILITY_MODE)
@@ -207,6 +208,17 @@ Variable `RMW_CONNEXT_DISABLE_FAST_ENDPOINT_DISCOVERY` may be used to disable
 these automatic optimizations, and to leave the DomainParticipant's QoS to
 its defaults.
 
+### RMW_CONNEXT_DISABLE_RELIABILITY_OPTIMIZATIONS
+
+By default, `rmw_connextdds` will modify the QoS of each realiable DataWriter
+and DataReader to improve the responsiveness of the RTPS [reliability protocol](https://community.rti.com/static/documentation/connext-dds/6.0.1/doc/manuals/connext_dds/html_files/RTI_ConnextDDS_CoreLibraries_UsersManual/Content/UsersManual/Using_QosPolicies_to_Tune_the_Reliable_P.htm?tocpath=Part%203%3A%20Advanced%20Concepts%7C11.%20Reliable%20Communications%7C11.3%20Using%20QosPolicies%20to%20Tune%20the%20Reliable%20Protocol%7C_____0#reliable_1394042328_776265).
+
+For example, the ["heartbeat period"](https://community.rti.com/static/documentation/connext-dds/6.0.1/doc/manuals/connext_dds/html_files/RTI_ConnextDDS_CoreLibraries_UsersManual/Content/UsersManual/Controlling_Heartbeats_and_Retries.htm#reliable_1394042328_785637)
+is sped up from 3 seconds to 100 milliseconds.
+
+These optimizations may be disabled using variable
+`RMW_CONNEXT_DISABLE_RELIABILITY_OPTIMIZATIONS`.
+
 ### RMW_CONNEXT_ENDPOINT_QOS_OVERRIDE_POLICY
 
 When this variable is not set or set to `always`, the QoS settings specified in

rmw_connextdds_common/include/rmw_connextdds/context.hpp

@@ -100,6 +100,9 @@ struct rmw_context_impl_s
 #if RMW_CONNEXT_DEFAULT_LARGE_DATA_OPTIMIZATIONS
   bool optimize_large_data{true};
 #endif /* RMW_CONNEXT_DEFAULT_LARGE_DATA_OPTIMIZATIONS */
+#if RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS
+  bool optimize_reliability{true};
+#endif /* RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS */
 
   enum class participant_qos_override_policy_t
   {

rmw_connextdds_common/include/rmw_connextdds/static_config.hpp

@@ -85,6 +85,11 @@
   "RMW_CONNEXT_DISABLE_LARGE_DATA_OPTIMIZATIONS"
 #endif /* RMW_CONNEXT_ENV_DISABLE_LARGE_DATA_OPTIMIZATIONS */
 
+#ifndef RMW_CONNEXT_ENV_DISABLE_RELIABILITY_OPTIMIZATIONS
+#define RMW_CONNEXT_ENV_DISABLE_RELIABILITY_OPTIMIZATIONS \
+  "RMW_CONNEXT_DISABLE_RELIABILITY_OPTIMIZATIONS"
+#endif /* RMW_CONNEXT_ENV_DISABLE_RELIABILITY_OPTIMIZATIONS */
+
 // TODO(security-wg): These are intended to be temporary, and need to be
 // refactored into a proper abstraction.
 #ifndef RMW_CONNEXT_ENV_SECURITY_LOG_FILE
@@ -226,6 +231,58 @@
 #define RMW_CONNEXT_TYPE_OBJECT_MAX_SERIALIZED_SIZE   65000
 #endif /* RMW_CONNEXT_TYPE_OBJECT_MAX_SERIALIZED_SIZE */
 
+/******************************************************************************
+ * Customize the RTPS reliability protocol to speed up its responsiveness.
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS
+#define RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS     1
+#endif /* RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS */
+
+/******************************************************************************
+ * Regular hearbeat period used by any reliable RTPS Writer.
+ * This is an initializer for an instance of type DDS_Duration_t.
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD
+#define RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD       {0, 100000000}   /* 100ms */
+#endif /* RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD */
+
+/******************************************************************************
+ * Fast hearbeat period used by any reliable RTPS Writer to allow
+ * late joiners and out of sync readers to catch up.
+ * This is an initializer for an instance of type DDS_Duration_t.
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD_FAST
+#define RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD_FAST  {0, 20000000}   /* 20ms */
+#endif /* RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD_FAST */
+
+/******************************************************************************
+ * When a DataWriter receives a request for missing DDS samples from a
+ * DataReader and responds by resending the requested DDS samples, it will
+ * ignore additional requests for the same DDS samples during the time period
+ * max_nack_response_delay. We decrease this to be less than the HB period.
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_MAX_NACK_RESPONSE_DELAY
+#define RMW_CONNEXT_DEFAULT_MAX_NACK_RESPONSE_DELAY  {0, 10000000}   /* 10ms */
+#endif /* RMW_CONNEXT_DEFAULT_MAX_NACK_RESPONSE_DELAY */
+
+/******************************************************************************
+ * Maximum number of periodic heartbeats gone unanswered after which a
+ * DataWriter will consider a DataReader as inactive.
+ *
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_MAX_HEARTBEATS
+#define RMW_CONNEXT_DEFAULT_MAX_HEARTBEATS    (10 * 60) /* 1m @ 10hz */
+#endif /* RMW_CONNEXT_DEFAULT_MAX_HEARTBEATS */
+
+/******************************************************************************
+ * When a reliable reader receives a heartbeat from a remote writer and finds
+ * out that it needs to send back an ACK/NACK message, the reader can choose to
+ * delay a while. We set this delay to be compatible with the HB period.
+ ******************************************************************************/
+#ifndef RMW_CONNEXT_DEFAULT_MAX_HEARTBEAT_RESPONSE_DELAY
+#define RMW_CONNEXT_DEFAULT_MAX_HEARTBEAT_RESPONSE_DELAY  {0, 10000000} /* 10ms */
+#endif /* RMW_CONNEXT_DEFAULT_MAX_HEARTBEAT_RESPONSE_DELAY */
+
 /******************************************************************************
  * Automatically tune DataWriterQos to better handle reliable "large data".
  ******************************************************************************/

rmw_connextdds_common/src/common/rmw_context.cpp

@@ -1281,6 +1281,25 @@ rmw_api_connextdds_init(
     RMW_CONNEXT_LOG_DEBUG_A("initial DDS peers: %s", initial_peers)
   }
 
+#if RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS
+  // Check if we should disable the optimizations for the RTPS reliability protocol
+  const char * disable_optimize_reliability_env = nullptr;
+  lookup_rc = rcutils_get_env(
+    RMW_CONNEXT_ENV_DISABLE_RELIABILITY_OPTIMIZATIONS,
+    &disable_optimize_reliability_env);
+
+  if (nullptr != lookup_rc || nullptr == disable_optimize_reliability_env) {
+    RMW_CONNEXT_LOG_ERROR_A_SET(
+      "failed to lookup from environment: "
+      "var=%s, "
+      "rc=%s ",
+      RMW_CONNEXT_ENV_DISABLE_RELIABILITY_OPTIMIZATIONS,
+      lookup_rc)
+    return RMW_RET_ERROR;
+  }
+  ctx->optimize_reliability = '\0' == disable_optimize_reliability_env[0];
+#endif /* RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS */
+
   if (nullptr == RMW_Connext_gv_DomainParticipantFactory) {
     RMW_CONNEXT_ASSERT(1 == RMW_Connext_gv_ContextCount)
     RMW_CONNEXT_LOG_DEBUG("initializing DDS DomainParticipantFactory")

rmw_connextdds_common/src/ndds/dds_api_ndds.cpp

@@ -501,6 +501,27 @@ rmw_connextdds_get_datawriter_qos(
     qos->publish_mode.kind = DDS_ASYNCHRONOUS_PUBLISH_MODE_QOS;
   }
 
+#if RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS
+  // The default settings for the RTPS reliability protocol are not very
+  // responsive, and they cause some unit tests to fail. These optimizations
+  // increase the Heartbeat period from 3s (default) to 100ms. Other vendors
+  // seem to be in a similar range. We also lower the period for "late joiners"
+  // and the "fast" period (used to speed up recovery of readers that reached
+  // the "high watermark" of unacked samples -- 1 by default) to 20ms.
+  if (ctx->optimize_reliability) {
+    qos->protocol.rtps_reliable_writer.heartbeat_period =
+      RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD;
+    qos->protocol.rtps_reliable_writer.late_joiner_heartbeat_period =
+      RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD_FAST;
+    qos->protocol.rtps_reliable_writer.fast_heartbeat_period =
+      RMW_CONNEXT_DEFAULT_HEARTBEAT_PERIOD_FAST;
+    qos->protocol.rtps_reliable_writer.max_heartbeat_retries =
+      RMW_CONNEXT_DEFAULT_MAX_HEARTBEATS;
+    qos->protocol.rtps_reliable_writer.max_nack_response_delay =
+      RMW_CONNEXT_DEFAULT_MAX_NACK_RESPONSE_DELAY;
+  }
+#endif /* RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS */
+
 #if RMW_CONNEXT_DEFAULT_LARGE_DATA_OPTIMIZATIONS
   // Unless disabled, optimize the DataWriter's reliability protocol to
   // better handle large data samples. These are *bounded* types whose
@@ -587,6 +608,17 @@ rmw_connextdds_get_datareader_qos(
     }
   }
 
+#if RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS
+  // The default settings for the RTPS reliability protocol are not very
+  // responsive, and they cause some unit tests to fail. These optimizations
+  // are dual to those applied in rmw_connextdds_get_datawriter_qos().
+  if (ctx->optimize_reliability) {
+    qos->protocol.rtps_reliable_reader.min_heartbeat_response_delay = DDS_DURATION_ZERO;
+    qos->protocol.rtps_reliable_reader.max_heartbeat_response_delay =
+      RMW_CONNEXT_DEFAULT_MAX_HEARTBEAT_RESPONSE_DELAY;
+  }
+#endif /* RMW_CONNEXT_DEFAULT_RELIABILITY_OPTIMIZATIONS */
+
 #if RMW_CONNEXT_DEFAULT_LARGE_DATA_OPTIMIZATIONS
   // Unless disabled, optimize the DataReader's reliability protocol to
   // better handle large data samples. These are *bounded* types whose