Builtin flow controller

Signed-off-by: Eugenio Collado <[email protected]>

Author: EugenioCollado

include/fastdds/rtps/attributes/RTPSParticipantAttributes.hpp

@@ -391,6 +391,9 @@ class BuiltinAttributes
     //! Set to true to avoid multicast traffic on builtin endpoints
     bool avoid_builtin_multicast = true;
 
+    //! Flow controller name to use for the builtin writers
+    std::string flow_controller_name = "";
+
     BuiltinAttributes() = default;
 
     virtual ~BuiltinAttributes() = default;
@@ -410,6 +413,7 @@ class BuiltinAttributes
                (this->writerHistoryMemoryPolicy == b.writerHistoryMemoryPolicy) &&
                (this->writerPayloadSize == b.writerPayloadSize) &&
                (this->mutation_tries == b.mutation_tries) &&
+               (this->flow_controller_name == b.flow_controller_name) &&
                (this->avoid_builtin_multicast == b.avoid_builtin_multicast);
     }
 

resources/xsd/fastdds_profiles.xsd

@@ -720,7 +720,8 @@
         ├ writerHistoryMemoryPolicy             [0~1],
         ├ readerPayloadSize                     [uint32],
         ├ writerPayloadSize                     [uint32],
-        └ mutation_tries                        [uint32]-->
+        ├ mutation_tries                        [uint32],
+        └ flow_controller_name                  [string] -->
     <xs:complexType name="builtinAttributesType">
         <xs:all>
             <xs:element name="discovery_config" type="discoverySettingsType" minOccurs="0" maxOccurs="1"/>
@@ -735,6 +736,7 @@
             <xs:element name="readerPayloadSize" type="uint32" minOccurs="0" maxOccurs="1"/>
             <xs:element name="writerPayloadSize" type="uint32" minOccurs="0" maxOccurs="1"/>
             <xs:element name="mutation_tries" type="uint32" minOccurs="0" maxOccurs="1"/>
+            <xs:element name="flow_controller_name" type="string" minOccurs="0" maxOccurs="1"/>
         </xs:all>
     </xs:complexType>
 

src/cpp/fastdds/domain/DomainParticipantImpl.cpp

@@ -2090,6 +2090,7 @@ bool DomainParticipantImpl::can_qos_be_updated(
                 from.wire_protocol().builtin.writerHistoryMemoryPolicy) ||
                 !(to.wire_protocol().builtin.writerPayloadSize == from.wire_protocol().builtin.writerPayloadSize) ||
                 !(to.wire_protocol().builtin.mutation_tries == from.wire_protocol().builtin.mutation_tries) ||
+                !(to.wire_protocol().builtin.flow_controller_name == from.wire_protocol().builtin.flow_controller_name) ||
                 !(to.wire_protocol().builtin.avoid_builtin_multicast ==
                 from.wire_protocol().builtin.avoid_builtin_multicast) ||
                 !(to.wire_protocol().builtin.discovery_config.discoveryProtocol ==

src/cpp/rtps/builtin/discovery/participant/PDP.cpp

@@ -50,6 +50,7 @@
 #include <rtps/builtin/discovery/participant/PDPEndpoints.hpp>
 #include <rtps/builtin/discovery/participant/PDPListener.h>
 #include <rtps/builtin/liveliness/WLP.hpp>
+#include <rtps/flowcontrol/FlowControllerFactory.hpp>
 #include <rtps/history/TopicPayloadPoolRegistry.hpp>
 #include <rtps/network/utils/external_locators.hpp>
 #include <rtps/participant/RTPSParticipantImpl.hpp>
@@ -1700,6 +1701,7 @@ WriterAttributes PDP::static_create_builtin_writer_attributes(
     if (!pattr.flow_controllers.empty())
     {
         attributes.mode = ASYNCHRONOUS_WRITER;
+        attributes.flow_controller_name = (pattr.builtin.flow_controller_name != "") ? pattr.builtin.flow_controller_name : fastdds::rtps::async_flow_controller_name;
     }
 
     attributes.times.heartbeat_period = pdp_heartbeat_period;

src/cpp/rtps/flowcontrol/FlowControllerImpl.hpp

@@ -1127,7 +1127,6 @@ class FlowControllerImpl : public FlowController
             const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time)
     {
         bool ret_value = false;
-        assert(!change->writer_info.is_linked.load());
         // Sync delivery failed. Try to store for asynchronous delivery.
 #if HAVE_STRICT_REALTIME
         std::unique_lock<fastdds::TimedMutex> lock(async_mode.changes_interested_mutex, std::defer_lock);

src/cpp/xmlparser/XMLElementParser.cpp

@@ -593,6 +593,7 @@ XMLP_ret XMLParser::getXMLBuiltinAttributes(
             <xs:element name="readerHistoryMemoryPolicy" type="historyMemoryPolicyType" minOccurs="0"/>
             <xs:element name="writerHistoryMemoryPolicy" type="historyMemoryPolicyType" minOccurs="0"/>
             <xs:element name="mutation_tries" type="uint32Type" minOccurs="0"/>
+            <xs:element name="flow_controller_name" type="stringType" minOccurs="0"/>
         </xs:all>
        </xs:complexType>
      */
@@ -709,6 +710,14 @@ XMLP_ret XMLParser::getXMLBuiltinAttributes(
                 return XMLP_ret::XML_ERROR;
             }
         }
+        else if (strcmp(name, FLOW_CONTROLLER_NAME) == 0)
+        {
+            // flow_controller_name - stringType
+            if (XMLP_ret::XML_OK != getXMLString(p_aux0, &builtin.flow_controller_name, ident))
+            {
+                return XMLP_ret::XML_ERROR;
+            }
+        }
         else
         {
             EPROSIMA_LOG_ERROR(XMLPARSER, "Invalid element found into 'builtinAttributesType'. Name: " << name);

test/mock/rtps/RTPSParticipantAttributes/fastdds/rtps/attributes/RTPSParticipantAttributes.hpp

@@ -377,6 +377,9 @@ class BuiltinAttributes
     //! Mutation tries if the port is being used.
     uint32_t mutation_tries = 100u;
 
+    //! Flow controller name to use for the builtin writers
+    std::string flow_controller_name = "";
+
     //! Set to true to avoid multicast traffic on builtin endpoints
     bool avoid_builtin_multicast = true;
 
@@ -399,6 +402,7 @@ class BuiltinAttributes
                (this->writerHistoryMemoryPolicy == b.writerHistoryMemoryPolicy) &&
                (this->writerPayloadSize == b.writerPayloadSize) &&
                (this->mutation_tries == b.mutation_tries) &&
+               (this->flow_controller_name == b.flow_controller_name) &&
                (this->avoid_builtin_multicast == b.avoid_builtin_multicast);
     }