[24110] Fix wrong update of default locators (backport #6273)

src/cpp/rtps/participant/RTPSParticipantImpl.cpp

@@ -144,7 +144,8 @@ static void set_builtin_transports_from_env_var(
                         "LARGE_DATAv6", BuiltinTransports::LARGE_DATAv6,
                         "P2P", BuiltinTransports::P2P))
                 {
-                    EPROSIMA_LOG_ERROR(RTPS_PARTICIPANT, "Wrong value '" << env_value << "' for environment variable '" <<
+                    EPROSIMA_LOG_ERROR(RTPS_PARTICIPANT,
+                            "Wrong value '" << env_value << "' for environment variable '" <<
                             env_var_name << "'. Leaving as DEFAULT");
                 }
                 // Max_msg_size parser
@@ -242,6 +243,10 @@ Locator_t& RTPSParticipantImpl::applyLocatorAdaptRule(
     {
         metatraffic_unicast_port_ += delta;
     }
+    else if (default_unicast_port_ == loc.port)
+    {
+        default_unicast_port_ += delta;
+    }
     loc.port += delta;
     return loc;
 }
@@ -436,8 +441,8 @@ bool RTPSParticipantImpl::setup_transports()
                     {
                         EPROSIMA_LOG_INFO(RTPS_PARTICIPANT,
                                 "Participant " << m_att.getName() << " with GUID " << m_guid <<
-                                " tries to create a TCP client for discovery server without providing a proper listening port." <<
-                                " No TCP participants will be able to connect to this participant, but it will be able make connections.");
+                                " tries to create a TCP client without providing a proper listening port." <<
+                                " No incomming connections will be accepted, only outgoing connections.");
                     }
                     std::for_each(m_att.builtin.discovery_config.m_DiscoveryServers.begin(),
                             m_att.builtin.discovery_config.m_DiscoveryServers.end(), [&](Locator_t& locator)
@@ -616,13 +621,14 @@ void RTPSParticipantImpl::setup_meta_traffic()
 void RTPSParticipantImpl::setup_user_traffic()
 {
     // Creation of user locator and receiver resources
-    //If no default locators are defined we define some.
+    // If no default locators are defined we define some.
     /* The reasoning here is the following.
        If the parameters of the RTPS Participant don't hold default listening locators for the creation
        of Endpoints, we make some for Unicast only.
        If there is at least one listen locator of any kind, we do not create any default ones.
        If there are no sending locators defined, we create default ones for the transports we implement.
      */
+    default_unicast_port_ = metatraffic_unicast_port_ + m_att.port.offsetd3 - m_att.port.offsetd1;
     if (m_att.defaultUnicastLocatorList.empty() && m_att.defaultMulticastLocatorList.empty())
     {
         //Default Unicast Locators in case they have not been provided
@@ -636,11 +642,10 @@ void RTPSParticipantImpl::setup_user_traffic()
     else
     {
         // Locator with port 0, calculate port.
-        uint32_t unicast_port = metatraffic_unicast_port_ + m_att.port.offsetd3 - m_att.port.offsetd1;
         std::for_each(m_att.defaultUnicastLocatorList.begin(), m_att.defaultUnicastLocatorList.end(),
                 [&](Locator_t& loc)
                 {
-                    m_network_Factory.fill_default_locator_port(loc, unicast_port);
+                    m_network_Factory.fill_default_locator_port(loc, default_unicast_port_);
                 });
         m_network_Factory.NormalizeLocators(m_att.defaultUnicastLocatorList);
 
@@ -1235,8 +1240,8 @@ bool RTPSParticipantImpl::create_writer(
     }
 
     auto callback = [hist, listen, entityId, this]
-                (const GUID_t& guid, WriterAttributes& watt, FlowController* flow_controller,
-                    IPersistenceService* persistence, bool is_reliable) -> BaseWriter*
+            (const GUID_t& guid, WriterAttributes& watt, FlowController* flow_controller,
+            IPersistenceService* persistence, bool is_reliable) -> BaseWriter*
             {
                 BaseWriter* writer = nullptr;
 
@@ -1284,8 +1289,8 @@ bool RTPSParticipantImpl::createReader(
         bool enable)
 {
     auto callback = [hist, listen, this]
-                (const GUID_t& guid, ReaderAttributes& param, IPersistenceService* persistence,
-                    bool is_reliable) -> BaseReader*
+            (const GUID_t& guid, ReaderAttributes& param, IPersistenceService* persistence,
+            bool is_reliable) -> BaseReader*
             {
                 if (is_reliable)
                 {
@@ -1330,8 +1335,8 @@ bool RTPSParticipantImpl::createReader(
     }
 
     auto callback = [hist, listen, &payload_pool, this]
-                (const GUID_t& guid, ReaderAttributes& param, IPersistenceService* persistence,
-                    bool is_reliable) -> BaseReader*
+            (const GUID_t& guid, ReaderAttributes& param, IPersistenceService* persistence,
+            bool is_reliable) -> BaseReader*
             {
                 if (is_reliable)
                 {
@@ -2293,7 +2298,7 @@ void RTPSParticipantImpl::normalize_endpoint_locators(
         EndpointAttributes& endpoint_att)
 {
     // Locators with port 0, calculate port.
-    uint32_t unicast_port = metatraffic_unicast_port_ + m_att.port.offsetd3 - m_att.port.offsetd1;
+    uint32_t unicast_port = default_unicast_port_;
     for (Locator_t& loc : endpoint_att.unicastLocatorList)
     {
         m_network_Factory.fill_default_locator_port(loc, unicast_port);
@@ -2854,7 +2859,8 @@ void RTPSParticipantImpl::environment_file_has_changed()
     }
     else
     {
-        EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK, "Trying to add Discovery Servers to a participant which is not a SERVER, BACKUP " <<
+        EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK,
+                "Trying to add Discovery Servers to a participant which is not a SERVER, BACKUP " <<
                 "or an overriden CLIENT (SIMPLE participant transformed into CLIENT with the environment variable)");
     }
 }
@@ -2880,8 +2886,7 @@ void RTPSParticipantImpl::get_default_metatraffic_locators(
 void RTPSParticipantImpl::get_default_unicast_locators(
         RTPSParticipantAttributes& att)
 {
-    uint32_t unicast_port = metatraffic_unicast_port_ + att.port.offsetd3 - att.port.offsetd1;
-    m_network_Factory.getDefaultUnicastLocators(att.defaultUnicastLocatorList, unicast_port);
+    m_network_Factory.getDefaultUnicastLocators(att.defaultUnicastLocatorList, default_unicast_port_);
     m_network_Factory.NormalizeLocators(att.defaultUnicastLocatorList);
 }
 

src/cpp/rtps/participant/RTPSParticipantImpl.hpp

@@ -555,6 +555,8 @@ class RTPSParticipantImpl
     RTPSParticipantAttributes m_att;
     //! Metatraffic unicast port used by default on this participant
     uint32_t metatraffic_unicast_port_ = 0;
+    //! Default unicast port used by default on this participant
+    uint32_t default_unicast_port_ = 0;
     //!Guid of the RTPSParticipant.
     GUID_t m_guid;
     //! String containing the RTPSParticipant Guid.

test/dds/communication/CMakeLists.txt

@@ -115,6 +115,7 @@ list(APPEND TEST_DEFINITIONS
     mix_zero_copy_communication
     close_TCP_client
     simple_data_sharing_stress
+    custom_metatraffic_but_empty_default
     )
 
 
@@ -129,6 +130,8 @@ list(APPEND XML_CONFIGURATION_FILES
     liveliness_assertion_profile.xml
     liveliness_assertion.360_profile.xml
     shm_communication_subscriber_dies_while_processing_message_profile.xml
+    custom_metatraffic_but_empty_default_pub.xml
+    custom_metatraffic_but_empty_default_sub.xml
     )
 
 list(APPEND PYTHON_FILES

test/dds/communication/SubscriberMain.cpp

@@ -107,7 +107,7 @@ int main(
         {
             if (++arg_count >= argc)
             {
-                std::cout << "--run-for expects a parameter" << std::endl;
+                std::cout << "--timeout expects a parameter" << std::endl;
                 return -1;
             }
 

test/dds/communication/custom_metatraffic_but_empty_default.json

@@ -0,0 +1,19 @@
+{
+    "description" : "Test that an additional call to set_qos() does not break communication if only the defaultUnicastLocatorList is not configured",
+    "participants" : [
+        {
+            "kind" : "publisher",
+            "xmlfile" : "custom_metatraffic_but_empty_default_pub.xml",
+            "samples" : "60",
+            "seed" : "261"
+        },
+        {
+            "kind" : "subscriber",
+            "samples" : "40",
+            "xmlfile" : "custom_metatraffic_but_empty_default_sub.xml",
+            "rescan" : "5",
+            "seed" : "261",
+            "sleep_before_exec" : "2"
+        }
+    ]
+}

test/dds/communication/custom_metatraffic_but_empty_default_pub.xml

@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<profiles xmlns="http://www.eprosima.com">
+    <transport_descriptors>
+        <transport_descriptor>
+            <transport_id>udp_transport</transport_id>
+            <type>UDPv4</type>
+        </transport_descriptor>
+    </transport_descriptors>
+
+    <participant profile_name="test_custom_meta_default_default_pub" is_default_profile="true">
+        <rtps>
+            <useBuiltinTransports>false</useBuiltinTransports>
+            <userTransports>
+                <transport_id>udp_transport</transport_id>
+            </userTransports>
+        </rtps>
+    </participant>
+
+    <data_writer profile_name="simple_publisher_profile" is_default_profile="true">
+        <qos>
+            <reliability>
+                <kind>RELIABLE</kind>
+            </reliability>
+            <data_sharing>
+                <kind>OFF</kind>
+            </data_sharing>
+        </qos>
+    </data_writer>
+</profiles>

test/dds/communication/custom_metatraffic_but_empty_default_sub.xml

@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<profiles xmlns="http://www.eprosima.com">
+    <transport_descriptors>
+        <transport_descriptor>
+            <transport_id>udp_transport</transport_id>
+            <type>UDPv4</type>
+        </transport_descriptor>
+    </transport_descriptors>
+
+    <participant profile_name="test_custom_meta_default_default_sub" is_default_profile="true">
+        <rtps>
+            <useBuiltinTransports>false</useBuiltinTransports>
+            <userTransports>
+                <transport_id>udp_transport</transport_id>
+            </userTransports>
+
+            <builtin>
+                <discovery_config>
+                    <discoveryProtocol>SIMPLE</discoveryProtocol>
+                </discovery_config>
+
+                <metatrafficMulticastLocatorList>
+                    <locator>
+                        <udpv4>
+                            <address>239.255.0.1</address>
+                            <port>15150</port>
+                        </udpv4>
+                    </locator>
+                </metatrafficMulticastLocatorList>
+
+                <metatrafficUnicastLocatorList>
+                    <locator>
+                        <udpv4>
+                            <port>15162</port>
+                        </udpv4>
+                    </locator>
+                </metatrafficUnicastLocatorList>
+            </builtin>
+
+            <!-- Don't set defaultUnicastLocatorList and let Fast automatically configure it -->
+
+        </rtps>
+    </participant>
+
+    <data_reader profile_name="simple_subscriber_profile" is_default_profile="true">
+        <qos>
+            <reliability>
+                <kind>RELIABLE</kind>
+            </reliability>
+            <data_sharing>
+                <kind>OFF</kind>
+            </data_sharing>
+        </qos>
+    </data_reader>
+</profiles>

test/dds/communication/test_build.py

@@ -21,6 +21,8 @@
 import sys
 import time
 
+SLEEP_TAG = "--sleep_before_exec"
+
 script_dir = os.path.dirname(os.path.realpath(__file__))
 seed = str(os.getpid())
 
@@ -60,7 +62,11 @@ def define_args(tests_definition):
                               'publishers',
                               'sleep_before_exec',
                               'interval',
-                              'timeout']
+                              'timeout',
+                              'rescan',
+                              'seed']
+        # Note that arg 'seed' is automatically set, but we can override it by
+        # passing the arg again. This is useful if we need to determine a specific domain
 
         for argument in possible_arguments:
             if argument in test.keys():
@@ -172,23 +178,72 @@ def execute_command(command):
     return subprocess.Popen(command)
 
 
+class ScheduledCmd:
+    def __init__(self, when, kind, cmd):
+        self.when = when
+        self.kind = kind
+        self.cmd = cmd
+
+
+def _extract_sleep(command: list[str]) -> tuple[list[str], int]:
+    """Return (cleaned_command, sleep_seconds)."""
+    if SLEEP_TAG not in command:
+        return command, 0
+
+    cmd = list(command)
+    i = cmd.index(SLEEP_TAG)
+    try:
+        sleep_s = int(cmd[i + 1])
+    except (IndexError, ValueError):
+        raise ValueError(f"{SLEEP_TAG} must be followed by an integer seconds value: {command}")
+
+    # remove tag and value
+    del cmd[i:i+2]
+    return cmd, sleep_s
+
+
 def execute_commands(pub_commands, sub_commands, pubsub_commands, logger):
     """Get test definitions in command lists and execute each process."""
-    pubs_proc = []
-    subs_proc = []
-    pubsubs_proc = []
+    # Manually program execution of subprocess
+    now = time.monotonic()
+    scheduled: list[ScheduledCmd] = []
 
     for subscriber_command in sub_commands:
         logger.info(f'Executing subcriber: {subscriber_command}')
-        subs_proc.append(execute_command(subscriber_command))
+        cmd, s = _extract_sleep(subscriber_command)
+        scheduled.append(ScheduledCmd(when=now + s, kind="sub", cmd=cmd))
 
     for pubsub_command in pubsub_commands:
         logger.info(f'Executing pubsub: {pubsub_command}')
-        pubsubs_proc.append(execute_command(pubsub_command))
+        cmd, s = _extract_sleep(pubsub_command)
+        scheduled.append(ScheduledCmd(when=now + s, kind="pubsub", cmd=cmd))
 
     for publisher_command in pub_commands:
         logger.info(f'Executing publisher: {publisher_command}')
-        pubs_proc.append(execute_command(publisher_command))
+        cmd, s = _extract_sleep(publisher_command)
+        scheduled.append(ScheduledCmd(when=now + s, kind="pub", cmd=cmd))
+
+    # Sort scheduled subprocesses
+    scheduled.sort(key=lambda x: x.when)
+    pubs_proc = []
+    subs_proc = []
+    pubsubs_proc = []
+
+    # Execution of subprocess with scheduling
+    for item in scheduled:
+        delay = item.when - time.monotonic()
+        if delay > 0:
+            time.sleep(delay)
+
+        logger.info(f"Executing {item.kind}: {item.cmd}")
+        p = subprocess.Popen(item.cmd)
+
+        if item.kind == "sub":
+            subs_proc.append(p)
+        elif item.kind == "pubsub":
+            pubsubs_proc.append(p)
+        else:
+            pubs_proc.append(p)
 
     ret_value = 0