Populate message info for intra-process messages by thomasmoore-torc · Pull Request #3271 · ros2/rclcpp

thomasmoore-torc · 2026-09-14T14:48:27Z

rclcpp: Populate message info for intra-process messages

Branch: intra-process-message-info (this repo: ros2/rclcpp)
Related PR: #3272 — "Iterate over IntraProcessBuffer instead of creating a vector" (intra-process-reduce-copies). See Landing order below — no functional dependency, but the two touch overlapping code.

Summary

Subscriptions that receive a message via intra-process delivery currently get a default-constructed, empty rmw_message_info_t — no sequence number, no publisher GID, no from_intra_process flag. The same subscription receiving the same message via inter-process (rmw) delivery gets a fully populated one. Anything relying on message_info (deduplicating by publisher GID, checking from_intra_process, etc.) behaves inconsistently depending on whether a given publish happens to stay in-process or not.

This populates message_info for intra-process messages too, so subscribers see consistent, meaningful info regardless of delivery path.

Changes

Populate message info for intra-process messages — threads an rmw_message_info_t alongside the message through the intra-process publish path (Publisher::do_intra_process_publish/do_intra_process_ros_message_publish_and_return_shared → IntraProcessManager → IntraProcessBuffer → subscription).
- IntraProcessBuffer storage changes from a bare shared_ptr/unique_ptr to a bundled IntraProcessBufferData (message + rmw_message_info_t), added with a copy constructor that deep-copies the unique-ptr alternative (needed so get_all_data()-based paths, e.g. transient-local replay, keep working with move-only storage).
- Fixes do_transient_local_publish to correctly convert between shared/unique storage and shared/unique subscription requests via std::visit instead of assuming the two always match.
Move IntraProcessBufferData to its own header — pulls the new struct out into intra_process_buffer_data.hpp to keep intra_process_buffer.hpp focused and avoid a circular-ish coupling between buffer and data definitions.
Update intra-process unit tests — updates test_intra_process_buffer.cpp and test_intra_process_manager.cpp for the new Data/IntraProcessBufferData API and adds coverage for message-info propagation across both shared- and unique-ownership subscriptions.

Compatibility

Internal to rclcpp's intra-process implementation (rclcpp::experimental::*); no public API signature changes. Behavior change: intra-process subscribers now observe non-default message_info where they previously saw an empty one.

Testing

Built and tested against rolling in a ros:rolling-ros-base container via colcon build/colcon test for rclcpp (3091 tests, 0 failures), plus uncrustify/cpplint lint targets. All green.

Landing order

intra-process-reduce-copies (opened separately) touches the same files (intra_process_buffer.hpp, intra_process_manager.hpp, and both intra-process test files) as part of an unrelated copy-reduction optimization. The two branches are independently buildable and testable, but not free to merge in either order without a rebase — real textual conflicts exist between them (confirmed via a trial merge). No functional dependency either way; whichever of the two merges second will need to reconcile against the other.

🤖 Generated with Claude Code

(cherry picked from commit 8f255f3da12d5d0cc8aa5a49f69bb74339226820) (cherry picked from commit 94a3d5856c046e3a7b6bda942394cd8662926c28) Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

(cherry picked from commit f2302935bd1b41a75f56c6a9f4ef5b7fbe4d7e9b) (cherry picked from commit 8a875ee1df60afcdb7ed68ceb27f097acdcdacf0) Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

fujitatomoya · 2026-09-20T23:30:56Z

+
+    rmw_message_info_t message_info{};
+    message_info.from_intra_process = true;
+    message_info.publication_sequence_number = publisher_it->second.publication_sequence_number++;


what if 2 publishers are publishing to the same topic at the same time?
looks like this sequence counters mutated under a shared (reader) lock is not thread safe?
IMO std::atomic could be used here.

fujitatomoya · 2026-09-20T23:31:29Z

        continue;
      }

+      message_info.reception_sequence_number = subscription_it->second.reception_sequence_number++;


same racy condition here?

fujitatomoya · 2026-09-20T23:40:22Z

-      unique_msg = MessageUniquePtr(ptr);
+      // Promote to a shared pointer
+      auto unique_msg = std::move(std::get<MessageUniquePtr>(data.message));
+      data.message = MessageSharedPtr(unique_msg.release());


the message could be allocated via the allocator but will be destroyed with plain delete, this looks like it leads to mismatch allocate/deallocate consistency? eventually Undefined Behaivor.

Suggested change

data.message = MessageSharedPtr(unique_msg.release());

data.message = MessageSharedPtr(std::move(unique_msg));

fujitatomoya · 2026-09-20T23:43:33Z

-        return nullptr;
-      }
+    auto data = this->buffer_->consume();
+    if (data.message.index() == std::variant_npos) {


the check never fires for empty buffer? index() returns variant_npos only for a variant that is valueless by exception.

fujitatomoya · 2026-09-20T23:45:35Z

+        }, message);
      trigger_guard_condition();
    }
+    buffer_->add(std::move(data));


the gurad condition will be triggered before the data is in the queue? i think this needs to be moved to before triggering the guard condition? at least, previous code keeps it in that way.

fujitatomoya · 2026-09-20T23:47:11Z

@thomasmoore-torc i agree with you.

Anything relying on message_info (deduplicating by publisher GID, checking from_intra_process, etc.) behaves inconsistently depending on whether a given publish happens to stay in-process or not.

especially this behavior is not transparent and consistent for the applicaiton. got several comments above to be addressed. please have them checked out!

@fujitatomoya

Addresses five issues raised by @fujitatomoya on the PR: - publication_sequence_number and reception_sequence_number were plain uint64_t members mutated (via postfix ++) from do_intra_process_publish(), which only holds a shared (reader) lock on mutex_. Concurrent publishes could race on these counters. Made both std::atomic<uint64_t>; since std::atomic has no copy/move assignment, PublisherData/SubscriptionData gained explicit constructors and their insertion sites switched to try_emplace() instead of default-construct-then-assign. - IntraProcessBuffer::add_impl() (SharedPtr case) promoted a MessageUniquePtr to a MessageSharedPtr via `MessageSharedPtr(unique_msg.release())`, which drops the unique_ptr's (possibly allocator-aware) deleter in favor of plain delete -- an allocate/deallocate mismatch. Changed to `MessageSharedPtr(std::move(unique_msg))`, which preserves the original deleter via shared_ptr's converting constructor. - SubscriptionIntraProcess::take_data() checked `data.message.index() == std::variant_npos` to detect an empty buffer, but variant_npos only occurs for a valueless-by-exception variant. A default-constructed Data (what an empty buffer's consume() returns) holds a null pointer in its default alternative instead, so the check never fired. The original pre-refactor code checked pointer nullness directly (`if (!shared_msg) return nullptr;` / `if (!unique_msg) return nullptr;`); restored that via `std::visit` across both alternatives. - SubscriptionIntraProcessBuffer::provide_intra_process_message() called trigger_guard_condition() before buffer_->add(), unlike the sibling provide_intra_process_data() which does it in the correct order. This left a window where a consumer woken by the guard condition could find the buffer still empty. Reordered to add() then trigger. Signed-off-by: Thomas Moore <thomas.moore@torc.ai> Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Each test was verified to fail against the pre-fix code and pass against the fix in the previous commit: - test_intra_process_buffer.cpp: shared_buffer_add_preserves_custom_deleter adds a unique_ptr with a tracking deleter to a SharedPtr-typed buffer and confirms the *original* deleter runs when the promoted shared_ptr's refcount hits zero, not a plain `delete`. - test_subscription_intra_process.cpp (new file, plus scaffolding to construct a real SubscriptionIntraProcess directly, without a full Node/executor): take_data_returns_nullptr_when_buffer_is_empty covers the variant_npos check; provide_intra_process_message_adds_data_before_triggering patches rcl_trigger_guard_condition (via mocking_utils) to observe whether the buffer already has data at the moment the guard condition actually fires. - test_publisher.cpp: concurrent_publish_produces_unique_sequence_numbers publishes from 8 threads at once and checks every delivered publication_sequence_number is unique. Against the pre-fix plain uint64_t counter this fails reliably (8/8 runs); this is a stronger regression guard than a typical data-race test since the race window here is wide enough to hit consistently under load, not just occasionally. Signed-off-by: Thomas Moore <thomas.moore@torc.ai> Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

thomasmoore-torc added 3 commits September 14, 2026 14:19

Populate message info for intra-process messages

6f44d2c

(cherry picked from commit 8f255f3da12d5d0cc8aa5a49f69bb74339226820) (cherry picked from commit 94a3d5856c046e3a7b6bda942394cd8662926c28) Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

Move IntraProcessBufferData to its own header

05f5e58

(cherry picked from commit f2302935bd1b41a75f56c6a9f4ef5b7fbe4d7e9b) (cherry picked from commit 8a875ee1df60afcdb7ed68ceb27f097acdcdacf0) Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

Update intra-process unit tests

bef63b7

Signed-off-by: Thomas Moore <thomas.moore@torc.ai>

thomasmoore-torc mentioned this pull request Sep 14, 2026

Iterate over IntraProcessBuffer instead of creating a vector #3272

Closed

fujitatomoya requested changes Sep 20, 2026

View reviewed changes

thomasmoore-torc and others added 2 commits September 21, 2026 15:05

Navigation Menu

Populate message info for intra-process messages - #3271

Populate message info for intra-process messages#3271
thomasmoore-torc wants to merge 5 commits into
ros2:rollingfrom
thomasmoore-torc:intra-process-message-info

thomasmoore-torc commented Sep 14, 2026 •

edited

Loading

Uh oh!

fujitatomoya Sep 20, 2026

Uh oh!

fujitatomoya Sep 20, 2026

Uh oh!

fujitatomoya Sep 20, 2026

Uh oh!

fujitatomoya Sep 20, 2026

Uh oh!

fujitatomoya Sep 20, 2026

Uh oh!

fujitatomoya commented Sep 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	data.message = MessageSharedPtr(unique_msg.release());
	data.message = MessageSharedPtr(std::move(unique_msg));

Conversation

thomasmoore-torc commented Sep 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

rclcpp: Populate message info for intra-process messages

Summary

Changes

Compatibility

Testing

Landing order

Uh oh!

fujitatomoya Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Sep 20, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya commented Sep 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

thomasmoore-torc commented Sep 14, 2026 •

edited

Loading