[19263 ] User configuration for SHM metatraffic

src/cpp/rtps/network/NetworkFactory.cpp

@@ -34,10 +34,36 @@ namespace rtps {
 
 using SendResourceList = fastdds::rtps::SendResourceList;
 
-NetworkFactory::NetworkFactory()
+NetworkFactory::NetworkFactory(
+        const RTPSParticipantAttributes& PParam)
     : maxMessageSizeBetweenTransports_(std::numeric_limits<uint32_t>::max())
     , minSendBufferSize_(std::numeric_limits<uint32_t>::max())
 {
+    const std::string* allow_metatraffic = nullptr;
+    allow_metatraffic = PropertyPolicyHelper::find_property(PParam.properties, "fastdds.shm.allow_metatraffic");
+    if (allow_metatraffic)
+    {
+        if (*allow_metatraffic == "unicast")
+        {
+            allow_shm_unicast_metatraffic_ = true;
+            allow_shm_multicast_metatraffic_ = false;
+        }
+        else if (*allow_metatraffic == "all")
+        {
+            allow_shm_unicast_metatraffic_ = true;
+            allow_shm_multicast_metatraffic_ = true;
+        }
+        else if (*allow_metatraffic == "none")
+        {
+            allow_shm_unicast_metatraffic_ = false;
+            allow_shm_multicast_metatraffic_ = false;
+        }
+        else
+        {
+            EPROSIMA_LOG_WARNING(RTPS_NETWORK, "Unrecognized value '" << *allow_metatraffic << "'" <<
+                    " for 'fastdds.shm.allow_metatraffic'. Using default value: 'none'");
+        }
+    }
 }
 
 bool NetworkFactory::build_send_resources(
@@ -246,9 +272,9 @@ bool NetworkFactory::getDefaultMetatrafficMulticastLocators(
 
     for (auto& transport : mRegisteredTransports)
     {
-        // For better fault-tolerance reasons, SHM multicast metatraffic is avoided if it is already provided
+        // For better fault-tolerance reasons, SHM metatraffic is avoided if it is already provided
         // by another transport
-        if (transport->kind() != LOCATOR_KIND_SHM)
+        if (allow_shm_multicast_metatraffic_ || transport->kind() != LOCATOR_KIND_SHM)
         {
             result |= transport->getDefaultMetatrafficMulticastLocators(locators, metatraffic_multicast_port);
         }
@@ -286,10 +312,28 @@ bool NetworkFactory::getDefaultMetatrafficUnicastLocators(
         uint32_t metatraffic_unicast_port) const
 {
     bool result = false;
+
+    TransportInterface* shm_transport = nullptr;
+
     for (auto& transport : mRegisteredTransports)
     {
-        result |= transport->getDefaultMetatrafficUnicastLocators(locators, metatraffic_unicast_port);
+        // For better fault-tolerance reasons, SHM metatraffic is avoided if it is already provided
+        // by another transport
+        if (allow_shm_unicast_metatraffic_ || transport->kind() != LOCATOR_KIND_SHM)
+        {
+            result |= transport->getDefaultMetatrafficUnicastLocators(locators, metatraffic_unicast_port);
+        }
+        else
+        {
+            shm_transport = transport.get();
+        }
+    }
+
+    if (locators.size() == 0 && shm_transport)
+    {
+        result |= shm_transport->getDefaultMetatrafficUnicastLocators(locators, metatraffic_unicast_port);
     }
+
     return result;
 }
 

src/cpp/rtps/network/NetworkFactory.h

@@ -43,7 +43,8 @@ class NetworkFactory
 {
 public:
 
-    NetworkFactory();
+    NetworkFactory(
+            const RTPSParticipantAttributes& PParam);
 
     /**
      * Allow registration of a transport statically, by specifying the transport type and
@@ -224,6 +225,12 @@ class NetworkFactory
 
     uint32_t minSendBufferSize_;
 
+    // Whether unicast metatraffic on SHM transport is allowed
+    bool allow_shm_unicast_metatraffic_ = false;
+
+    // Whether multicast metatraffic on SHM transport is allowed
+    bool allow_shm_multicast_metatraffic_ = false;
+
     /**
      * Calculate well-known ports.
      */

src/cpp/rtps/participant/RTPSParticipantImpl.cpp

@@ -130,8 +130,8 @@ RTPSParticipantImpl::RTPSParticipantImpl(
     , m_att(PParam)
     , m_guid(guidP, c_EntityId_RTPSParticipant)
     , mp_builtinProtocols(nullptr)
-    , mp_ResourceSemaphore(new Semaphore(0))
     , IdCounter(0)
+    , m_network_Factory(PParam)
     , type_check_fn_(nullptr)
     , client_override_(false)
     , internal_metatraffic_locators_(false)
@@ -512,7 +512,6 @@ RTPSParticipantImpl::~RTPSParticipantImpl()
     }
     m_receiverResourcelist.clear();
 
-    delete mp_ResourceSemaphore;
     delete mp_userParticipant;
     mp_userParticipant = nullptr;
     send_resource_list_.clear();
@@ -2034,22 +2033,6 @@ bool RTPSParticipantImpl::newRemoteEndpointDiscovered(
     return false;
 }
 
-void RTPSParticipantImpl::ResourceSemaphorePost()
-{
-    if (mp_ResourceSemaphore != nullptr)
-    {
-        mp_ResourceSemaphore->post();
-    }
-}
-
-void RTPSParticipantImpl::ResourceSemaphoreWait()
-{
-    if (mp_ResourceSemaphore != nullptr)
-    {
-        mp_ResourceSemaphore->wait();
-    }
-}
-
 void RTPSParticipantImpl::assert_remote_participant_liveliness(
         const GuidPrefix_t& remote_guid)
 {

src/cpp/rtps/participant/RTPSParticipantImpl.h

@@ -240,12 +240,6 @@ class RTPSParticipantImpl
         return (uint32_t)m_att.participantID;
     }
 
-    //!Post to the resource semaphore
-    void ResourceSemaphorePost();
-
-    //!Wait for the resource semaphore
-    void ResourceSemaphoreWait();
-
     //!Get Pointer to the Event Resource.
     ResourceEvent& getEventResource()
     {
@@ -529,8 +523,6 @@ class RTPSParticipantImpl
     ResourceEvent mp_event_thr;
     //! BuiltinProtocols of this RTPSParticipant
     BuiltinProtocols* mp_builtinProtocols;
-    //!Semaphore to wait for the listen thread creation.
-    Semaphore* mp_ResourceSemaphore;
     //!Id counter to correctly assign the ids to writers and readers.
     std::atomic<uint32_t> IdCounter;
     //! Mutex to safely access endpoints collections

test/unittest/rtps/network/CMakeLists.txt

@@ -22,6 +22,10 @@ set(NETWORKFACTORYTESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/Time_t.cpp
 
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
+
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
+
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/ChannelResource.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPChannelResource.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPTransportInterface.cpp

test/unittest/rtps/network/NetworkFactoryTests.cpp

@@ -16,13 +16,15 @@
 
 #include <gtest/gtest.h>
 
+#include <fastdds/rtps/attributes/RTPSParticipantAttributes.h>
 #include <fastrtps/transport/TCPv4TransportDescriptor.h>
 #include <fastrtps/transport/TCPv6TransportDescriptor.h>
 #include <fastrtps/transport/UDPv4TransportDescriptor.h>
 #include <fastrtps/transport/UDPv6TransportDescriptor.h>
 #include <fastrtps/utils/collections/ResourceLimitedVector.hpp>
 #include <fastrtps/utils/IPLocator.h>
 
+
 #include <MockTransport.h>
 #include <rtps/network/NetworkFactory.h>
 
@@ -34,7 +36,8 @@ class NetworkTests : public ::testing::Test
 {
 public:
 
-    NetworkFactory networkFactoryUnderTest;
+    RTPSParticipantAttributes pattr{};
+    NetworkFactory networkFactoryUnderTest{pattr};
     void HELPER_RegisterTransportWithKindAndChannels(
             int kind,
             unsigned int channels);
@@ -648,7 +651,7 @@ TEST_F(NetworkTests, LocatorShrink)
     std::vector<ShrinkLocatorCase_t> test_cases;
     fill_blackbox_locators_test_cases(test_cases);
 
-    NetworkFactory f;
+    NetworkFactory f{pattr};
     UDPv4TransportDescriptor udpv4;
     f.RegisterTransport(&udpv4);
     // TODO: Register more transports

test/unittest/transport/CMakeLists.txt

@@ -53,6 +53,8 @@ set(UDPV4TESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/OStreamConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPv4Transport.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPTransportInterface.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/ChannelResource.cpp
@@ -70,7 +72,9 @@ set(UDPV6TESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/OStreamConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/Time_t.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/network/NetworkFactory.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/ChannelResource.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPChannelResource.cpp
@@ -91,7 +95,9 @@ set(TCPV4TESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/OStreamConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/Time_t.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/messages/RTPSMessageCreator.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/network/NetworkFactory.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/resources/ResourceEvent.cpp
@@ -133,7 +139,9 @@ set(TCPV6TESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/OStreamConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/Time_t.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/messages/RTPSMessageCreator.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/network/NetworkFactory.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/resources/ResourceEvent.cpp
@@ -177,6 +185,8 @@ set(TEST_UDPV4TESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/core/policy/ParameterList.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/utils/IPFinder.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/utils/IPLocator.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/test_UDPv4Transport.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPChannelResource.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/UDPTransportInterface.cpp
@@ -196,7 +206,9 @@ set(SHAREDMEMTESTS_SOURCE
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/OStreamConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutConsumer.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/log/StdoutErrConsumer.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/attributes/PropertyPolicy.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/Time_t.cpp
+    ${PROJECT_SOURCE_DIR}/src/cpp/rtps/flowcontrol/ThroughputControllerDescriptor.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/network/NetworkFactory.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp
     ${PROJECT_SOURCE_DIR}/src/cpp/rtps/transport/shared_mem/SharedMemTransportDescriptor.cpp

test/unittest/transport/TCPv6Tests.cpp

@@ -19,6 +19,7 @@
 #include <gtest/gtest.h>
 
 #include <fastdds/dds/log/Log.hpp>
+#include <fastdds/rtps/attributes/RTPSParticipantAttributes.h>
 #include <fastrtps/transport/TCPv6TransportDescriptor.h>
 #include <fastrtps/utils/IPLocator.h>
 #include <fastrtps/utils/Semaphore.h>
@@ -151,7 +152,8 @@ TEST_F(TCPv6Tests, opening_and_closing_input_channel)
     multicastFilterLocator.port = g_default_port; // arbitrary
     IPLocator::setIPv6(multicastFilterLocator, 0xff31, 0, 0, 0, 0, 0, 0x8000, 0x1234);
 
-    NetworkFactory factory;
+    RTPSParticipantAttributes p_attr{};
+    NetworkFactory factory{p_attr};
     factory.RegisterTransport<TCPv6Transport, TCPv6TransportDescriptor>(descriptor);
     std::vector<std::shared_ptr<ReceiverResource>> receivers;
     factory.BuildReceiverResources(multicastFilterLocator, receivers, 0x8FFF);