Builtin flow controller

Signed-off-by: Eugenio Collado <eugeniocollado@eprosima.com>

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/blackbox/api/dds-pim/PubSubParticipant.hpp

@@ -701,6 +701,14 @@ class PubSubParticipant
         return false;
     }
 
+    PubSubParticipant& flow_controller(
+            const std::shared_ptr<eprosima::fastdds::rtps::FlowControllerDescriptor>& flow_controller)
+    {
+        participant_qos_.flow_controllers().clear();
+        participant_qos_.flow_controllers().push_back(flow_controller);
+        return *this;
+    }
+
     PubSubParticipant& initial_peers(
             const eprosima::fastdds::rtps::LocatorList& initial_peers)
     {

test/blackbox/common/BlackboxTestsDiscovery.cpp

@@ -1298,7 +1298,7 @@ TEST_P(Discovery, EndpointCreationMultithreaded)
     endpoint_thr.join();
 }
 
-// Regression test for redmine issue 16253
+//! Regression test for redmine issue 16253
 TEST_P(Discovery, AsymmeticIgnoreParticipantFlags)
 {
     if (INTRAPROCESS != GetParam())
@@ -1416,7 +1416,7 @@ TEST_P(Discovery, single_unicast_pdp_response)
     main_wire_protocol.builtin.discovery_config.initial_announcements.count = 1;
     main_wire_protocol.builtin.discovery_config.initial_announcements.period = { 0, 100000000 };
 
-    // The main participant will use the test transport and a specific announcments configuration
+    // The main participant will use the test transport and a specific announcements configuration
     main_participant->disable_builtin_transport().add_user_transport_to_pparams(test_transport)
             .wire_protocol(main_wire_protocol);
 
@@ -1474,6 +1474,137 @@ TEST_P(Discovery, single_unicast_pdp_response)
     participants.clear();
 }
 
+//! Regression test for redmine issue 22506
+//! Test using a user's flowcontroller limiting the bandwidth and two remote participants waiting for the PDP sample.
+TEST_P(Discovery, single_unicast_pdp_response_flowcontroller)
+{
+    // Leverage intraprocess so transport is only used for participant discovery
+    if (INTRAPROCESS != GetParam())
+    {
+        GTEST_SKIP() << "Only makes sense on INTRAPROCESS";
+        return;
+    }
+
+    using namespace eprosima::fastdds::dds;
+
+    // All participants would restrict communication to UDP localhost.
+    // The main participant should send a single initial announcement, and have a big announcement period.
+    // This is to ensure that we only check the datagrams sent in response to the participant discovery,
+    // and not the ones sent in the periodic announcements.
+    // The main participant will use the test transport to count the number of unicast messages sent.
+
+    // This will hold the multicast port. Since the test is not always run in the same domain, we'll need to set
+    // its value when the first multicast datagram is sent.
+    std::atomic<uint32_t> multicast_port{ 0 };
+    // Declare a test transport that will count the number of unicast messages sent
+    std::atomic<size_t> num_unicast_sends{ 0 };
+    auto test_transport = std::make_shared<test_UDPv4TransportDescriptor>();
+    test_transport->interfaceWhiteList.push_back("127.0.0.1");
+    test_transport->locator_filter_ = [&num_unicast_sends, &multicast_port](
+        const eprosima::fastdds::rtps::Locator& destination)
+            {
+                if (IPLocator::isMulticast(destination))
+                {
+                    uint32_t port = 0;
+                    multicast_port.compare_exchange_strong(port, destination.port);
+                }
+                else
+                {
+                    num_unicast_sends.fetch_add(1u, std::memory_order_seq_cst);
+                }
+
+                // Do not discard any message
+                return false;
+            };
+
+    // Create the main participant
+    auto main_participant = std::make_shared<PubSubParticipant<HelloWorldPubSubType>>(0, 0, 0, 0);
+    WireProtocolConfigQos main_wire_protocol;
+    main_wire_protocol.builtin.avoid_builtin_multicast = true;
+    main_wire_protocol.builtin.discovery_config.leaseDuration = c_TimeInfinite;
+    main_wire_protocol.builtin.discovery_config.leaseDuration_announcementperiod = { 3600, 0 };
+    main_wire_protocol.builtin.discovery_config.initial_announcements.count = 1;
+    main_wire_protocol.builtin.discovery_config.initial_announcements.period = { 0, 100000000u };
+    main_wire_protocol.builtin.flow_controller_name = "TestFlowController";
+    // main_wire_protocol.builtin.flow_controller_name = "FastDDSFlowControllerDefault";
+
+    // Flowcontroller to limit the bandwidth
+    auto test_flow_controller = std::make_shared<eprosima::fastdds::rtps::FlowControllerDescriptor>();
+    test_flow_controller->name = "TestFlowController";
+    test_flow_controller->max_bytes_per_period = 3120; //3120 OK
+    test_flow_controller->period_ms = static_cast<uint64_t>(100);
+
+    // The main participant will use the test transport, specific announcements configuration and a flowcontroller
+    main_participant->disable_builtin_transport().add_user_transport_to_pparams(test_transport)
+            .wire_protocol(main_wire_protocol)
+            .flow_controller(test_flow_controller);
+
+    // Start the main participant
+    ASSERT_TRUE(main_participant->init_participant());
+
+    // Wait for the initial announcements to be sent
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    // This would have set the multicast port
+    EXPECT_NE(multicast_port, 0u);
+
+    // The rest of the participants only send announcements to the main participant
+    // Calculate the metatraffic unicast port of the main participant
+    uint32_t port = multicast_port + main_wire_protocol.port.offsetd1 - main_wire_protocol.port.offsetd0;
+
+    // The rest of the participants only send announcements to the main participant
+    auto udp_localhost_transport = std::make_shared<test_UDPv4TransportDescriptor>();
+    udp_localhost_transport->interfaceWhiteList.push_back("127.0.0.1");
+    Locator peer_locator;
+    IPLocator::createLocator(LOCATOR_KIND_UDPv4, "127.0.0.1", port, peer_locator);
+    WireProtocolConfigQos wire_protocol;
+    wire_protocol.builtin.avoid_builtin_multicast = true;
+    wire_protocol.builtin.initialPeersList.push_back(peer_locator);
+    wire_protocol.builtin.discovery_config.leaseDuration = c_TimeInfinite;
+    wire_protocol.builtin.discovery_config.leaseDuration_announcementperiod = { 3600, 0 };
+    wire_protocol.builtin.discovery_config.initial_announcements.count = 1;
+    wire_protocol.builtin.discovery_config.initial_announcements.period = { 0, 100000000u };
+
+    std::vector<std::shared_ptr<PubSubParticipant<HelloWorldPubSubType>>> participants;
+    for (size_t i = 0; i < 20; ++i)
+    {
+        auto participant = std::make_shared<PubSubParticipant<HelloWorldPubSubType>>(0, 0, 0, 0);
+        // All participants use the same transport
+        participant->disable_builtin_transport().add_user_transport_to_pparams(udp_localhost_transport)
+                .wire_protocol(wire_protocol);
+        participants.push_back(participant);
+    }
+
+    // Start the rest of the participants
+    for (auto& participant : participants)
+    {
+        ASSERT_TRUE(participant->init_participant());
+        participant->wait_discovery(std::chrono::seconds(1), 1, true);
+        // participant->wait_discovery();
+        // std::this_thread::sleep_for(std::chrono::milliseconds(30));
+        std::cout << "\n\nnum_unicast_sends: " << num_unicast_sends.load(std::memory_order::memory_order_seq_cst) << " VS " << participants.size() << std::endl;
+    }
+
+    // EXPECT_EQ(num_unicast_sends.load(std::memory_order::memory_order_seq_cst),
+    //         participants.size());
+
+    // Destroy main participant
+    main_participant.reset();
+    for (auto& participant : participants)
+    {
+        participant->wait_discovery(std::chrono::seconds(1), 0, true);
+        std::cout << "\n\nnum_unicast_sends: " << num_unicast_sends.load(std::memory_order::memory_order_seq_cst) << " VS " << participants.size() << std::endl;
+    }
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    // Check that only two unicast messages per participant were sent
+    std::cout << "\n\nnum_unicast_sends: " << num_unicast_sends.load(std::memory_order::memory_order_seq_cst) << " VS " << participants.size() << std::endl;
+    EXPECT_EQ(num_unicast_sends.load(std::memory_order::memory_order_seq_cst),
+            participants.size() + participants.size());
+
+    // Clean up
+    participants.clear();
+}
+
 #ifdef INSTANTIATE_TEST_SUITE_P
 #define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
 #else

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);
     }