Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,13 @@ It is important to note that this mode typically yields higher throughput rates

If `RMW_FASTRTPS_PUBLICATION_MODE` is not set, then both `rmw_fastrtps_cpp` and `rmw_fastrtps_dynamic_cpp` behave as if it were set to `SYNCHRONOUS`.

For the internal `ros_discovery_info` graph subscription, `rmw_fastrtps` requests unique network flow endpoints by default. The `RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS` environment variable can be used to change the unique network flows request mode. The admissible values are:

* `DISABLED`: Unique network flow endpoints not required.
* `STRICT`: Unique network flow endpoins strictly required. This can result in an error if the `ros_discovery_info` graph subscription is configured with a conflicting option, such as static endpoint discovery or custom locators.
* `OPTIONAL`: Unique network flow endpoints optionally required and will not be used if the `ros_discovery_info` graph subscription is configured with a conflicting option, such as static endpoint discovery or custom locators.
* `SYSTEM_DEFAULT`: Unique network flow endpoints requirement decided by system.

### Full QoS configuration

Fast DDS QoS policies can be fully configured through a combination of the [rmw QoS profile] API, and the [Fast DDS XML] file's QoS elements. Configuration depends on the environment variable `RMW_FASTRTPS_USE_QOS_FROM_XML`.
Expand Down
7 changes: 5 additions & 2 deletions rmw_fastrtps_cpp/src/init_rmw_context_impl.cpp
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
// Copyright 2020 Open Source Robotics Foundation, Inc.
// Copyright 2026 Torc Robotics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -29,6 +30,7 @@
#include "rmw_fastrtps_cpp/subscription.hpp"

#include "rmw_fastrtps_shared_cpp/custom_participant_info.hpp"
#include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"
#include "rmw_fastrtps_shared_cpp/namespace_prefix.hpp"
#include "rmw_fastrtps_shared_cpp/participant.hpp"
#include "rmw_fastrtps_shared_cpp/publisher.hpp"
Expand All @@ -53,9 +55,10 @@ init_context_impl(

// Avoid receiving graph updates from our own publication
subscription_options.ignore_local_publications = true;
// Improve graph discovery by using a unique listening port for its subscription
// Improve graph discovery by using a unique listening port for its subscription,
// unless disabled by the user
subscription_options.require_unique_network_flow_endpoints =
RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info();

std::unique_ptr<rmw_dds_common::Context> common_context(
new(std::nothrow) rmw_dds_common::Context());
Expand Down
7 changes: 5 additions & 2 deletions rmw_fastrtps_dynamic_cpp/src/init_rmw_context_impl.cpp
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
// Copyright 2020 Open Source Robotics Foundation, Inc.
// Copyright 2026 Torc Robotics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -27,6 +28,7 @@
#include "rmw_fastrtps_dynamic_cpp/identifier.hpp"

#include "rmw_fastrtps_shared_cpp/custom_participant_info.hpp"
#include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"
#include "rmw_fastrtps_shared_cpp/listener_thread.hpp"
#include "rmw_fastrtps_shared_cpp/participant.hpp"
#include "rmw_fastrtps_shared_cpp/publisher.hpp"
Expand All @@ -50,9 +52,10 @@ init_context_impl(

// Avoid receiving graph updates from our own publication
subscription_options.ignore_local_publications = true;
// Improve graph discovery by using a unique listening port for its subscription
// Improve graph discovery by using a unique listening port for its subscription,
// unless disabled by the user
subscription_options.require_unique_network_flow_endpoints =
RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info();

std::unique_ptr<rmw_dds_common::Context> common_context(
new(std::nothrow) rmw_dds_common::Context());
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
// Copyright 2020 Open Source Robotics Foundation, Inc.
// Copyright 2026 Torc Robotics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand All @@ -23,6 +24,24 @@
namespace rmw_fastrtps_shared_cpp
{

/// Check if unique network flows should be used for ROS discovery info.
/**
* This function is thread-safe and by default caches the result of the environment
* variable check to avoid repeated lookups during concurrent context initialization.
*
* If `use_cached` is true (the default) the cached value is returned; the cache
* is populated on the first call and subsequent calls ignore changes to the
* environment variable. If `use_cached` is false the environment variable is
* read and mapped to the returned enumeration on each call without modifying the
* cached value.
*
* \param[in] use_cached whether to return the cached value (default: true)
* \return the requirement for unique network flow endpoints for ROS discovery info
*/
RMW_FASTRTPS_SHARED_CPP_PUBLIC
rmw_unique_network_flow_endpoints_requirement_t
get_unique_network_flows_for_ros_discovery_info(bool use_cached = true);

/// Increment `rmw_context_impl_t` reference count, destroying it if the count reaches zero.
/**
* Function that should be called when destroying a node.
Expand Down
73 changes: 73 additions & 0 deletions rmw_fastrtps_shared_cpp/src/init_rmw_context_impl.cpp
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
// Copyright 2020 Open Source Robotics Foundation, Inc.
// Copyright 2026 Torc Robotics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand All @@ -15,6 +16,9 @@
#include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"

#include <cassert>
#include <mutex>

#include "rcutils/env.h"

#include "rmw/error_handling.h"
#include "rmw/init.h"
Expand All @@ -30,6 +34,75 @@

#include "rmw_fastrtps_shared_cpp/listener_thread.hpp"

namespace
{

static const rmw_unique_network_flow_endpoints_requirement_t
DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS =
RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;

rmw_unique_network_flow_endpoints_requirement_t g_unique_network_flows_for_ros_discovery_info =
DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS;
std::once_flag g_init_flag;

static rmw_unique_network_flow_endpoints_requirement_t
read_unique_network_flows_from_env()
{
const char * env_value = nullptr;
const char * error_str = rcutils_get_env(
"RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS", &env_value);

if (error_str != nullptr) {
RCUTILS_LOG_WARN_NAMED(
"rmw_fastrtps_shared_cpp",
"Error getting env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: %s. "
"Using default behavior.",
error_str);
return DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS;
} else if (env_value == nullptr) {
RCUTILS_LOG_WARN_NAMED(
"rmw_fastrtps_shared_cpp",
"Invalid value for env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: null. "
"Using default behavior.");
return DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS;
} else if (env_value[0] == '\0') {
return DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS;
} else if (strcmp(env_value, "SYSTEM_DEFAULT") == 0) {
return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_SYSTEM_DEFAULT;
} else if (strcmp(env_value, "DISABLED") == 0) {
return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_NOT_REQUIRED;
} else if (strcmp(env_value, "OPTIONAL") == 0) {
return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
} else if (strcmp(env_value, "STRICT") == 0) {
return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED;
}

RCUTILS_LOG_WARN_NAMED(
"rmw_fastrtps_shared_cpp",
"Invalid value for env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: %s. "
"Using default behavior.",
env_value);
return DEFAULT_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS;
}

void initialize_unique_network_flows()
{
g_unique_network_flows_for_ros_discovery_info = read_unique_network_flows_from_env();
}

} // anonymous namespace

rmw_unique_network_flow_endpoints_requirement_t
rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info(bool use_cached)
{
if (use_cached) {
std::call_once(g_init_flag, initialize_unique_network_flows);
return g_unique_network_flows_for_ros_discovery_info;
}
// When bypassing the cache, use the shared helper so logic is consistent.
return read_unique_network_flows_from_env();
}

rmw_ret_t
rmw_fastrtps_shared_cpp::decrement_context_impl_ref_count(
rmw_context_t * context)
Expand Down
82 changes: 74 additions & 8 deletions rmw_fastrtps_shared_cpp/src/utils.cpp
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
// Copyright 2026 Torc Robotics, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -140,19 +141,84 @@ create_datareader(
default:
case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_SYSTEM_DEFAULT:
case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_NOT_REQUIRED:
// Unique network flow endpoints not required. We leave the decission to the XML profile.
// Unique network flow endpoints not required. We leave the decision to the XML profile.
break;

case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED:
case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED:
// Ensure we request unique network flow endpoints
using PropertyPolicyHelper = eprosima::fastdds::rtps::PropertyPolicyHelper;
if (nullptr ==
PropertyPolicyHelper::find_property(
updated_qos.properties(),
"fastdds.unique_network_flows"))
{
updated_qos.properties().properties().emplace_back("fastdds.unique_network_flows", "");
bool strict_unique_flows_required =
(RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED ==
subscription_options->require_unique_network_flow_endpoints);

eprosima::fastdds::dds::DomainParticipantQos participant_qos;
subscriber->get_participant()->get_qos(participant_qos);

// Check if locators are explicitly specified at endpoint level.
bool has_explicit_locators =
!datareader_qos.endpoint().unicast_locator_list.empty() ||
!datareader_qos.endpoint().multicast_locator_list.empty() ||
!datareader_qos.endpoint().remote_locator_list.empty();

// Also consider participant defaults as explicit user intent. This covers
// scenarios where only default multicast locators are configured and unique
// flow creation would fail due to missing unicast locators.
if (!has_explicit_locators) {
has_explicit_locators =
!participant_qos.wire_protocol().default_unicast_locator_list.empty() ||
!participant_qos.wire_protocol().default_multicast_locator_list.empty();
}

if (has_explicit_locators) {
if (strict_unique_flows_required) {
RMW_SET_ERROR_MSG(
"RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED cannot be used when "
"endpoint or participant default locators are explicitly configured");
return false;
} else {
RCUTILS_LOG_INFO_NAMED(
"rmw_fastrtps_shared_cpp",
"RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED is ignored when "
"endpoint or participant default locators are explicitly configured");
}
}

// Check if Endpoint Discovery Protocol (EDP) is configured as STATIC
bool is_edp_static =
participant_qos.wire_protocol().builtin.discovery_config.
use_STATIC_EndpointDiscoveryProtocol;

if (is_edp_static) {
if (strict_unique_flows_required) {
RMW_SET_ERROR_MSG(
"RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED cannot be used with "
"static endpoint discovery");
return false;
} else {
RCUTILS_LOG_INFO_NAMED(
"rmw_fastrtps_shared_cpp",
"RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED is ignored with "
"static endpoint discovery");
}
}

using PropertyPolicyHelper = eprosima::fastdds::rtps::PropertyPolicyHelper;
if (nullptr ==
PropertyPolicyHelper::find_property(
updated_qos.properties(),
"fastdds.unique_network_flows"))
{
// For STRICTLY_REQUIRED: always request unique flows (Fast-DDS will enforce it)
// For OPTIONALLY_REQUIRED: only request if no locators are explicitly configured and not
// using static endpoint discovery
bool should_request_unique_flows =
strict_unique_flows_required ||
(!has_explicit_locators && !is_edp_static);

if (should_request_unique_flows) {
updated_qos.properties().properties().emplace_back("fastdds.unique_network_flows", "");
}
}
}
break;
}
Expand Down
9 changes: 9 additions & 0 deletions rmw_fastrtps_shared_cpp/test/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -54,3 +54,12 @@ ament_add_gtest(test_logging test_logging.cpp)
if(TARGET test_logging)
target_link_libraries(test_logging ${PROJECT_NAME} rmw::rmw)
endif()

ament_add_gtest(test_unique_network_flows test_unique_network_flows.cpp)
if(TARGET test_unique_network_flows)
target_link_libraries(test_unique_network_flows
${PROJECT_NAME}
rcutils::rcutils
rmw::rmw
)
endif()
Loading