Add support for rosidl::Buffer-aware per-endpoint pub/sub by nvcyc · Pull Request #867 · ros2/rmw_fastrtps

nvcyc · 2026-03-16T09:43:08Z

Description

This pull request adds full rosidl::Buffer support to rmw_fastrtps_cpp, enabling per-endpoint DDS DataWriters/DataReaders for zero-copy buffer transport between compatible backends. Backend discovery uses DDS user_data QoS to advertise and detect supported backends.

This pull request modifies the following packages:

`rmw_fastrtps_shared_cpp` -- shared types and QoS

custom_participant_info.hpp: Added BufferDiscoveryCallback type and buffer_discovery_cb_ field for participant-level buffer discovery.
custom_publisher_info.hpp: Added BufferPublisherEndpoint struct and buffer-related fields (is_buffer_aware_, backend_aux_info_, local_endpoint_info_, buffer_mutex_, buffer_endpoints_, pending_buffer_endpoints_, buffer_alive_flag_, participant_, dds_publisher_).
custom_subscriber_info.hpp: Added BufferSubscriptionEndpoint struct (with owns_topic flag for single-process topic reuse) and buffer-related fields. Added RMWSubscriptionEvent::notify_buffer_data_available().
qos.hpp/qos.cpp: Added optional buffer_backends parameter to QoS functions; added encode_buffer_backends_for_user_data() / parse_buffer_backends_from_user_data() (format: BUFBE:name=aux;name2=aux2).
rmw_wait.cpp: Extended WaitSet to also check buffer_data_guard_; wakes subscription when buffer data is available.

`rmw_fastrtps_cpp` -- buffer-aware per-endpoint pub/sub

Backend lifecycle: Calls initialize_buffer_backends() / shutdown_buffer_backends() during rmw_init/rmw_shutdown.
buffer_endpoint_registry: Per-process singleton mapping topic names to discovery callbacks; publishers register subscriber-discovery callbacks and vice versa. Maintains known endpoints so late-registering callbacks see already-discovered peers.
Publisher creation: Detects buffer-aware types via callbacks->has_buffer_fields; encodes backend names in DDS user_data with "cpu" always present. Registers subscriber-discovery callbacks that create per-subscriber DataWriters on unique topics ({topic}/_buf/{pub_gid}_{sub_gid}) with reliable QoS, synchronous publish mode, and data sharing off. Discovery callbacks use buffer_alive_flag_ to prevent use-after-free.
Subscription creation: Parses acceptable_buffer_backends (NULL/empty/"cpu": CPU-only; "any": all; specific names: filtered). Creates per-publisher DataReaders with BufferDataReaderListener that triggers buffer_data_guard_. Topic reuse: In single-process scenarios, uses lookup_topicdescription() to reuse existing topics (owns_topic = false).
Buffer-aware publish (rmw_publish.cpp): Iterates per-subscriber endpoints, calls cdr_serialize_with_endpoint(). Conditionally publishes to the main DataWriter only when non-buffer-aware subscribers exist, avoiding unnecessary CPU conversion.
Buffer-aware take (rmw_take.cpp): Iterates per-publisher DataReaders, takes raw CDR bytes and calls cdr_deserialize_with_endpoint(). Falls back to main DataReader when no buffer endpoint data is available. Guard condition re-arm: Re-sets buffer_data_guard_ when remaining data exists.
Lock-ordering safety: Both publisher and subscriber discovery use a three-phase pattern matching rmw_zenoh_cpp: (1) acquire buffer_mutex_ to check duplicates and mark pending; (2) call DDS create/delete APIs without holding the lock; (3) re-acquire to store endpoints. Destroy functions move endpoints to local lists under lock and clean up DDS resources after releasing.

Is this user-facing behavior change?

This pull request does not change existing rmw_fastrtps behavior for standard messages. For messages with uint8[] fields, the per-endpoint transport is transparent -- publishers and subscribers share backend info via DDS user_data, and CPU fallback ensures correctness when backends are incompatible.

Did you use Generative AI?

Yes. Claude (claude-4.6-opus) via Cursor was used to assist with creating an initial prototype version of the changes contained in this PR.

Additional Information

This PR is part of the broader ROS 2 native buffer feature introduced in this post.

Signed-off-by: CY Chen <cyc@nvidia.com>

hidmic

First pass. Another PR that needs many 👀 .

hidmic · 2026-03-17T02:23:22Z

rmw_fastrtps_cpp/src/buffer_endpoint_registry.cpp

+bool gid_equal(const rmw_gid_t & a, const rmw_gid_t & b)
+{
+  return std::memcmp(a.data, b.data, RMW_GID_STORAGE_SIZE) == 0;
+}


@nvcyc there's rmw_compare_gids_equal() already

This still stands.

hidmic · 2026-03-17T02:29:54Z

rmw_fastrtps_cpp/src/rmw_publish.cpp

+
+  for (const auto & endpoint : info->buffer_endpoints_) {
+    uint32_t serialized_size = callbacks->get_serialized_size(ros_message);
+    size_t buffer_size = serialized_size + 4096;


@nvcyc mind to explain those extra 4KB?

Still needs an explanation.

hidmic · 2026-03-17T02:31:40Z

rmw_fastrtps_cpp/src/rmw_publisher.cpp

+        auto gid_to_hex = [](const rmw_gid_t & gid, size_t bytes = 8) -> std::string {
+            static const char hex_chars[] = "0123456789abcdef";
+            std::string result;
+            result.reserve(bytes * 2);
+            for (size_t i = 0; i < bytes && i < RMW_GID_STORAGE_SIZE; ++i) {
+              result += hex_chars[(gid.data[i] >> 4) & 0xF];
+              result += hex_chars[gid.data[i] & 0xF];
+            }
+            return result;
+          };


@nvcyc nit: considering making this a free function.

hidmic · 2026-03-17T02:34:56Z

rmw_fastrtps_cpp/src/rmw_publisher.cpp

+          sub_info.gid.data, RMW_GID_STORAGE_SIZE);
+
+        eprosima::fastdds::dds::TopicQos topic_qos =
+          info->participant_->get_default_topic_qos();


@nvcyc meta: we should notify the user somehow about the actual QoS profile used for buffer aware pub/sub, or fallback to CPU only, or something.

hidmic · 2026-03-17T02:40:38Z

rmw_fastrtps_cpp/src/rmw_take.cpp

+    const_cast<void **>(data_values.buffer())[0] = &data;
+    eprosima::fastdds::dds::SampleInfoSeq info_seq{1};
+
+    auto ret_code = endpoint->data_reader->take(data_values, info_seq, 1);


@nvcyc meta: how do we enforce that we don't starve some endpoints? This implicitly prioritizes endpoints that were discovered earlier than others.

hidmic · 2026-03-17T02:43:12Z

rmw_fastrtps_cpp/src/subscription.cpp

+    if (subscription_options->acceptable_buffer_backends &&
+      strlen(subscription_options->acceptable_buffer_backends) > 0)
+    {
+      std::string acceptable(subscription_options->acceptable_buffer_backends);


@nvcyc there's rcpputils::split() for this.

hidmic · 2026-03-17T02:46:37Z

rmw_fastrtps_shared_cpp/src/qos.cpp

+  if (backends.empty()) {
+    return {};
+  }
+  std::string result = BUFFER_BACKEND_SENTINEL;


@nvcyc fyi: there's std::stringstream.

mjcarroll · 2026-03-18T13:58:00Z

rmw_fastrtps_cpp/src/buffer_endpoint_registry.cpp

+    "rmw_fastrtps_cpp",
+    "BufferEndpointRegistry: firing %zu publisher callback(s) for discovered subscriber on '%s'",
+    callbacks.size(), info.topic_name.c_str());
+  for (const auto & cb : callbacks) {


This looks safe from deadlock risk.

mjcarroll · 2026-03-18T13:58:19Z

rmw_fastrtps_cpp/src/buffer_endpoint_registry.cpp

+    "rmw_fastrtps_cpp",
+    "BufferEndpointRegistry: firing %zu subscriber callback(s) for discovered publisher on '%s'",
+    callbacks.size(), info.topic_name.c_str());
+  for (const auto & cb : callbacks) {


This one looks safe too.

Signed-off-by: CY Chen <cyc@nvidia.com>

hidmic

Looks like things are in flight. @MiguelCompany @richiware we need your 👀 here.

hidmic · 2026-03-20T20:50:25Z

rmw_fastrtps_cpp/src/buffer_endpoint_registry.cpp

+bool gid_equal(const rmw_gid_t & a, const rmw_gid_t & b)
+{
+  return std::memcmp(a.data, b.data, RMW_GID_STORAGE_SIZE) == 0;
+}


This still stands.

hidmic · 2026-03-20T20:51:27Z

rmw_fastrtps_cpp/src/rmw_publish.cpp

+
+  for (const auto & endpoint : info->buffer_endpoints_) {
+    uint32_t serialized_size = callbacks->get_serialized_size(ros_message);
+    size_t buffer_size = serialized_size + 4096;


Still needs an explanation.

Signed-off-by: CY Chen <cyc@nvidia.com>

nvcyc · 2026-03-25T17:36:22Z

rmw_fastrtps_cpp/src/rmw_publisher.cpp

+              return;
+            }
+          }
+


Should call rosidl_buffer_backend_registry::BufferBackendRegistry::get_instance().call notify_endpoint_discovered() around here.

nvcyc · 2026-03-25T17:36:33Z

rmw_fastrtps_cpp/src/rmw_subscription.cpp

+              return;
+            }
+          }
+


Should call rosidl_buffer_backend_registry::BufferBackendRegistry::get_instance().call notify_endpoint_discovered() around here.

nvcyc · 2026-03-25T18:18:35Z

rmw_fastrtps_cpp/src/rmw_publish.cpp

+  size_t total_matched = info->publisher_event_->subscription_count();
+  size_t buffer_count = state.endpoints.size() + state.pending.size();
+  bool has_non_buffer_subscribers = (total_matched > buffer_count);


Return directly if has_non_buffer_subscribers is non-zero as a fallback when legacy RMWs appear in any peer nodes. That is, every subscriber will get CPU-based messages from their main data reader if there is one or more subscriber that is based on legacy (older version) RMWs that do not support buffer backend protocol (and only work with main data reader/writer).

Add support for rosidl::Buffer-aware per-endpoint pub/sub

9ac694a

Signed-off-by: CY Chen <cyc@nvidia.com>

nvcyc requested review from MiguelCompany, ahcorde, hidmic, karsten-nvidia, mjcarroll, skyegalaxy and tfoote March 16, 2026 09:44

hidmic reviewed Mar 17, 2026

View reviewed changes

nvcyc marked this pull request as draft March 17, 2026 03:15

mjcarroll reviewed Mar 18, 2026

View reviewed changes

Add buffer backend init/shutdown functions

b8e49a9

Signed-off-by: CY Chen <cyc@nvidia.com>

nvcyc mentioned this pull request Mar 19, 2026

Add rosidl_buffer_backend_registry ros2/rosidl#944

Open

Fix lint error

95e47c8

Signed-off-by: CY Chen <cyc@nvidia.com>

nvcyc force-pushed the native_buffer branch from 7577fc1 to 95e47c8 Compare March 19, 2026 17:22

hidmic reviewed Mar 20, 2026

View reviewed changes

nvcyc and others added 2 commits March 22, 2026 17:59

Rename backend aux info to backend metadata

fbe3e51

Signed-off-by: CY Chen <cyc@nvidia.com>

defer buffer DataWriter/DataReader creation to publish/take path

5eaa0e2

nvcyc commented Mar 25, 2026

View reviewed changes

Conversation

nvcyc commented Mar 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

rmw_fastrtps_shared_cpp -- shared types and QoS

rmw_fastrtps_cpp -- buffer-aware per-endpoint pub/sub

Is this user-facing behavior change?

Did you use Generative AI?

Additional Information

Uh oh!

hidmic left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hidmic left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nvcyc Mar 25, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

nvcyc commented Mar 16, 2026 •

edited

Loading

`rmw_fastrtps_shared_cpp` -- shared types and QoS

`rmw_fastrtps_cpp` -- buffer-aware per-endpoint pub/sub

nvcyc Mar 25, 2026 •

edited

Loading