From 8f9cd4c595be38a673d8ad7d026b47ba428bb038 Mon Sep 17 00:00:00 2001 From: TinyGeek Date: Sun, 11 Oct 2026 21:02:53 +0900 Subject: [PATCH 1/2] Fix intra-process message copy allocation leak Guard raw allocations while constructing copies for owning intra-process buffers. After successful construction, transfer ownership to the existing message deleter. Cover custom and ROS owning buffers with throwing-copy regression cases. Assisted-by: Codex:gpt-6-sol Signed-off-by: TinyGeek --- .../experimental/intra_process_manager.hpp | 17 +- rclcpp/test/rclcpp/CMakeLists.txt | 6 + .../test_intra_process_copy_exception.cpp | 178 ++++++++++++++++++ 3 files changed, 198 insertions(+), 3 deletions(-) create mode 100644 rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp diff --git a/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp b/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp index a8eba4baf7..66b5b228ba 100644 --- a/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp +++ b/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp @@ -628,9 +628,15 @@ class IntraProcessManager // Copy the message since we have additional subscriptions to serve Deleter deleter = message.get_deleter(); auto ptr = MessageAllocTraits::allocate(allocator, 1); + auto deallocate = [&allocator](MessageT * allocation) { + MessageAllocTraits::deallocate(allocator, allocation, 1); + }; + std::unique_ptr allocation_guard(ptr, deallocate); MessageAllocTraits::construct(allocator, ptr, *message); + MessageUniquePtr copied_message(ptr, deleter); + allocation_guard.release(); - subscription->provide_intra_process_data(MessageUniquePtr(ptr, deleter)); + subscription->provide_intra_process_data(std::move(copied_message)); } continue; @@ -670,10 +676,15 @@ class IntraProcessManager Deleter deleter = message.get_deleter(); allocator::set_allocator_for_deleter(&deleter, &allocator); auto ptr = MessageAllocTraits::allocate(allocator, 1); + auto deallocate = [&allocator](MessageT * allocation) { + MessageAllocTraits::deallocate(allocator, allocation, 1); + }; + std::unique_ptr allocation_guard(ptr, deallocate); MessageAllocTraits::construct(allocator, ptr, *message); + MessageUniquePtr copied_message(ptr, deleter); + allocation_guard.release(); - ros_message_subscription->provide_intra_process_message( - MessageUniquePtr(ptr, deleter)); + ros_message_subscription->provide_intra_process_message(std::move(copied_message)); } } } diff --git a/rclcpp/test/rclcpp/CMakeLists.txt b/rclcpp/test/rclcpp/CMakeLists.txt index 29a5700d0f..0d33107a19 100644 --- a/rclcpp/test/rclcpp/CMakeLists.txt +++ b/rclcpp/test/rclcpp/CMakeLists.txt @@ -288,6 +288,12 @@ ament_add_ros_isolated_gtest(test_parameter_map test_parameter_map.cpp) if(TARGET test_parameter_map) target_link_libraries(test_parameter_map ${PROJECT_NAME} rcl::rcl rcl_yaml_param_parser::rcl_yaml_param_parser rcutils::rcutils) endif() +ament_add_ros_isolated_gtest( + test_intra_process_copy_exception test_intra_process_copy_exception.cpp) +if(TARGET test_intra_process_copy_exception) + target_link_libraries(test_intra_process_copy_exception ${PROJECT_NAME} std_msgs::std_msgs) +endif() + ament_add_ros_isolated_gtest(test_publisher test_publisher.cpp TIMEOUT 120) ament_add_test_label(test_publisher mimick) if(TARGET test_publisher) diff --git a/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp b/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp new file mode 100644 index 0000000000..6c9766851e --- /dev/null +++ b/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp @@ -0,0 +1,178 @@ +// Copyright 2026 TinyGeek +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include +#include +#include +#include +#include +#include "rclcpp/rclcpp.hpp" +#include "std_msgs/msg/string.hpp" + +namespace +{ +struct CopyMessage +{ + int value = 42; + bool fail_copy = false; + CopyMessage() = default; + CopyMessage(const CopyMessage & other) + : value(other.value), fail_copy(other.fail_copy) + { + if (fail_copy) {throw std::runtime_error("controlled copy failure");} + } +}; +struct AllocationCounts +{ + std::size_t allocated = 0; + std::size_t deallocated = 0; +}; +template +struct CountingAllocator +{ + using value_type = T; + std::shared_ptr counts; + CountingAllocator() + : counts(std::make_shared()) {} + explicit CountingAllocator(std::shared_ptr value) + : counts(std::move(value)) {} + template + CountingAllocator(const CountingAllocator & other) + : counts(other.counts) {} + T * allocate(std::size_t n) + { + auto pointer = std::allocator{}.allocate(n); + if constexpr (std::is_same::value) {counts->allocated += n;} + return pointer; + } + void deallocate(T * pointer, std::size_t n) + { + if constexpr (std::is_same::value) {counts->deallocated += n;} + std::allocator{}.deallocate(pointer, n); + } + template + struct rebind {using other = CountingAllocator;}; +}; +template +bool operator==(const CountingAllocator & a, const CountingAllocator & b) +{return a.counts == b.counts;} +template +bool operator!=(const CountingAllocator & a, const CountingAllocator & b) +{return !(a == b);} + +using Alloc = CountingAllocator; +using Deleter = rclcpp::allocator::Deleter; +using UniqueMessage = std::unique_ptr; +using Base = rclcpp::experimental::SubscriptionIntraProcessBase; + +class DataBuffer : public rclcpp::experimental::SubscriptionIntraProcessBuffer< + CopyMessage, Alloc, Deleter, CopyMessage> +{ +public: + explicit DataBuffer(const std::shared_ptr & allocator) + : SubscriptionIntraProcessBuffer( + allocator, rclcpp::contexts::get_global_default_context(), "/copy", rclcpp::QoS(10), + rclcpp::IntraProcessBufferType::UniquePtr) {} + std::shared_ptr take_data() override {return nullptr;} + void execute(const std::shared_ptr &) override {} + UniqueMessage consume() {return buffer_->consume_unique();} +}; + +class ROSBuffer : public rclcpp::experimental::SubscriptionROSMsgIntraProcessBuffer< + CopyMessage, Alloc, Deleter> +{ +public: + ROSBuffer() + : SubscriptionROSMsgIntraProcessBuffer( + rclcpp::contexts::get_global_default_context(), "/copy", rclcpp::QoS(10)) {} + bool use_take_shared_method() const override {return false;} + std::size_t available_capacity() const override {return 10;} + bool is_ready(const rcl_wait_set_t &) override {return false;} + std::shared_ptr take_data() override {return nullptr;} + void execute(const std::shared_ptr &) override {} + void provide_intra_process_message(ConstMessageSharedPtr) override + {throw std::logic_error("unexpected shared delivery");} + void provide_intra_process_message(MessageUniquePtr value) override {message = std::move(value);} + UniqueMessage message; + +protected: + void trigger_guard_condition() override {gc_.trigger();} +}; + +struct Case {bool ros_only; bool fail_copy;}; +class TestIntraProcessCopyException : public ::testing::TestWithParam +{ +public: + static void SetUpTestSuite() {rclcpp::init(0, nullptr);} + static void TearDownTestSuite() {rclcpp::shutdown();} +}; + +TEST_P(TestIntraProcessCopyException, balances_message_allocations) +{ + const auto param = GetParam(); + auto counts = std::make_shared(); + auto allocator = std::make_shared(counts); + { + auto node = std::make_shared("copy_allocation_test"); + auto publisher = node->create_publisher("copy", 10); + rclcpp::experimental::IntraProcessManager manager; + const auto id = manager.add_publisher(publisher); + std::shared_ptr data_a; + std::shared_ptr data_b; + std::shared_ptr ros_a; + std::shared_ptr ros_b; + if (param.ros_only) { + ros_a = std::make_shared(); + ros_b = std::make_shared(); + manager.add_subscription>(ros_a); + manager.add_subscription>(ros_b); + } else { + data_a = std::make_shared(allocator); + data_b = std::make_shared(allocator); + manager.add_subscription>(data_a); + manager.add_subscription>(data_b); + } + auto pointer = std::allocator_traits::allocate(*allocator, 1); + std::allocator_traits::construct(*allocator, pointer); + pointer->fail_copy = param.fail_copy; + Deleter deleter(allocator.get()); + UniqueMessage message(pointer, deleter); + const auto original = message.get(); + auto publish = [&]() { + manager.do_intra_process_publish>( + id, std::move(message), *allocator); + }; + if (param.fail_copy) { + EXPECT_THROW(publish(), std::runtime_error); + } else { + ASSERT_NO_THROW(publish()); + auto first = param.ros_only ? std::move(ros_a->message) : data_a->consume(); + auto second = param.ros_only ? std::move(ros_b->message) : data_b->consume(); + ASSERT_NE(nullptr, first); + ASSERT_NE(nullptr, second); + EXPECT_EQ(42, first->value); + EXPECT_EQ(42, second->value); + EXPECT_NE(first.get(), second.get()); + EXPECT_EQ(original, second.get()); + first->value = 9; + EXPECT_EQ(42, second->value); + } + } + EXPECT_EQ(2u, counts->allocated); + EXPECT_EQ(counts->allocated, counts->deallocated); +} +INSTANTIATE_TEST_SUITE_P( + BufferKinds, TestIntraProcessCopyException, + ::testing::Values(Case{false, false}, Case{false, true}, Case{true, false}, Case{true, true})); +} // namespace From 85b0e94eebb13835f37cb02706df592e5b72a940 Mon Sep 17 00:00:00 2001 From: TinyGeek Date: Thu, 15 Oct 2026 14:17:16 +0900 Subject: [PATCH 2/2] Guard intra-process message construction and adapted ROS conversion Use an exception-safe allocator helper for owning message copies and take ownership of adapted ROS messages before invoking conversion code. Cover construction, conversion, and delivery exceptions in regression tests. Refs #3320 Assisted-by: Codex:gpt-6-sol Signed-off-by: TinyGeek --- .../rclcpp/allocator/allocator_common.hpp | 15 ++ .../experimental/intra_process_manager.hpp | 24 +-- .../test_intra_process_copy_exception.cpp | 174 +++++++++++++++++- 3 files changed, 191 insertions(+), 22 deletions(-) diff --git a/rclcpp/include/rclcpp/allocator/allocator_common.hpp b/rclcpp/include/rclcpp/allocator/allocator_common.hpp index 18cab9e9e1..e828946eec 100644 --- a/rclcpp/include/rclcpp/allocator/allocator_common.hpp +++ b/rclcpp/include/rclcpp/allocator/allocator_common.hpp @@ -17,6 +17,7 @@ #include #include +#include #include "rcl/allocator.h" @@ -57,6 +58,20 @@ inline constexpr bool has_get_rcl_allocator_v = template using AllocRebind = typename std::allocator_traits::template rebind_traits; +template +typename Traits::pointer +allocate_and_construct(typename Traits::allocator_type & allocator, Args &&... args) +{ + auto ptr = Traits::allocate(allocator, 1); + try { + Traits::construct(allocator, ptr, std::forward(args)...); + } catch (...) { + Traits::deallocate(allocator, ptr, 1); + throw; + } + return ptr; +} + template [[deprecated("Conversion of C++ allocators to C style is not valid, as the size on deallocate" "can not be determined. This will be remove in future versions of ros.")]] diff --git a/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp b/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp index 66b5b228ba..effd6fafc3 100644 --- a/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp +++ b/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp @@ -28,6 +28,7 @@ #include #include +#include "rclcpp/allocator/allocator_common.hpp" #include "rclcpp/allocator/allocator_deleter.hpp" #include "rclcpp/experimental/buffers/intra_process_buffer.hpp" #include "rclcpp/experimental/ros_message_intra_process_buffer.hpp" @@ -627,14 +628,8 @@ class IntraProcessManager } else { // Copy the message since we have additional subscriptions to serve Deleter deleter = message.get_deleter(); - auto ptr = MessageAllocTraits::allocate(allocator, 1); - auto deallocate = [&allocator](MessageT * allocation) { - MessageAllocTraits::deallocate(allocator, allocation, 1); - }; - std::unique_ptr allocation_guard(ptr, deallocate); - MessageAllocTraits::construct(allocator, ptr, *message); + auto ptr = allocator::allocate_and_construct(allocator, *message); MessageUniquePtr copied_message(ptr, deleter); - allocation_guard.release(); subscription->provide_intra_process_data(std::move(copied_message)); } @@ -657,12 +652,13 @@ class IntraProcessManager if constexpr (rclcpp::TypeAdapter::is_specialized::value) { ROSMessageTypeAllocator ros_message_alloc(allocator); - auto ptr = ROSMessageTypeAllocatorTraits::allocate(ros_message_alloc, 1); - ROSMessageTypeAllocatorTraits::construct(ros_message_alloc, ptr); ROSMessageTypeDeleter deleter; allocator::set_allocator_for_deleter(&deleter, &allocator); - rclcpp::TypeAdapter::convert_to_ros_message(*message, *ptr); + auto ptr = allocator::allocate_and_construct( + ros_message_alloc); auto ros_msg = std::unique_ptr(ptr, deleter); + rclcpp::TypeAdapter::convert_to_ros_message( + *message, *ros_msg); ros_message_subscription->provide_intra_process_message(std::move(ros_msg)); } else { if constexpr (std::is_same::value) { @@ -675,14 +671,8 @@ class IntraProcessManager // Copy the message since we have additional subscriptions to serve Deleter deleter = message.get_deleter(); allocator::set_allocator_for_deleter(&deleter, &allocator); - auto ptr = MessageAllocTraits::allocate(allocator, 1); - auto deallocate = [&allocator](MessageT * allocation) { - MessageAllocTraits::deallocate(allocator, allocation, 1); - }; - std::unique_ptr allocation_guard(ptr, deallocate); - MessageAllocTraits::construct(allocator, ptr, *message); + auto ptr = allocator::allocate_and_construct(allocator, *message); MessageUniquePtr copied_message(ptr, deleter); - allocation_guard.release(); ros_message_subscription->provide_intra_process_message(std::move(copied_message)); } diff --git a/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp b/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp index 6c9766851e..652da3e28f 100644 --- a/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp +++ b/rclcpp/test/rclcpp/test_intra_process_copy_exception.cpp @@ -13,20 +13,75 @@ // limitations under the License. #include +#include #include #include #include #include -#include "rclcpp/rclcpp.hpp" +#include "rclcpp/allocator/allocator_common.hpp" +#include "rclcpp/allocator/allocator_deleter.hpp" +#include "rclcpp/contexts/default_context.hpp" +#include "rclcpp/experimental/intra_process_manager.hpp" +#include "rclcpp/experimental/ros_message_intra_process_buffer.hpp" +#include "rclcpp/experimental/subscription_intra_process_buffer.hpp" +#include "rclcpp/node.hpp" +#include "rclcpp/publisher.hpp" +#include "rclcpp/qos.hpp" +#include "rclcpp/type_adapter.hpp" +#include "rclcpp/utilities.hpp" #include "std_msgs/msg/string.hpp" +struct AdaptedMessage +{ + int value = 42; +}; + +struct TrackedROSMessage +{ + inline static bool throw_default = false; + inline static int alive = 0; + int value = 0; + + TrackedROSMessage() + { + if (throw_default) {throw std::runtime_error("controlled ROS construction failure");} + ++alive; + } + + ~TrackedROSMessage() {--alive;} +}; + +template<> +struct rclcpp::TypeAdapter +{ + using is_specialized = std::true_type; + using custom_type = AdaptedMessage; + using ros_message_type = TrackedROSMessage; + inline static bool throw_conversion = false; + + static void convert_to_ros_message(const custom_type & source, ros_message_type & destination) + { + destination.value = source.value; + if (throw_conversion) {throw std::runtime_error("controlled TypeAdapter failure");} + } + + static void convert_to_custom(const ros_message_type & source, custom_type & destination) + { + destination.value = source.value; + } +}; + namespace { struct CopyMessage { int value = 42; bool fail_copy = false; - CopyMessage() = default; + inline static bool throw_default = false; + CopyMessage() + { + if (throw_default) {throw std::runtime_error("controlled default construction failure");} + } CopyMessage(const CopyMessage & other) : value(other.value), fail_copy(other.fail_copy) { @@ -93,9 +148,10 @@ class ROSBuffer : public rclcpp::experimental::SubscriptionROSMsgIntraProcessBuf CopyMessage, Alloc, Deleter> { public: - ROSBuffer() + explicit ROSBuffer(bool throw_on_provide = false) : SubscriptionROSMsgIntraProcessBuffer( - rclcpp::contexts::get_global_default_context(), "/copy", rclcpp::QoS(10)) {} + rclcpp::contexts::get_global_default_context(), "/copy", rclcpp::QoS(10)), + throw_on_provide_(throw_on_provide) {} bool use_take_shared_method() const override {return false;} std::size_t available_capacity() const override {return 10;} bool is_ready(const rcl_wait_set_t &) override {return false;} @@ -103,9 +159,39 @@ class ROSBuffer : public rclcpp::experimental::SubscriptionROSMsgIntraProcessBuf void execute(const std::shared_ptr &) override {} void provide_intra_process_message(ConstMessageSharedPtr) override {throw std::logic_error("unexpected shared delivery");} - void provide_intra_process_message(MessageUniquePtr value) override {message = std::move(value);} + void provide_intra_process_message(MessageUniquePtr value) override + { + if (throw_on_provide_) {throw std::runtime_error("controlled provide failure");} + message = std::move(value); + } UniqueMessage message; +protected: + void trigger_guard_condition() override {gc_.trigger();} + +private: + bool throw_on_provide_; +}; + +class AdaptedROSBuffer : public rclcpp::experimental::SubscriptionROSMsgIntraProcessBuffer< + TrackedROSMessage, std::allocator, + std::default_delete> +{ +public: + AdaptedROSBuffer() + : SubscriptionROSMsgIntraProcessBuffer( + rclcpp::contexts::get_global_default_context(), "/copy", rclcpp::QoS(10)) {} + bool use_take_shared_method() const override {return false;} + std::size_t available_capacity() const override {return 10;} + bool is_ready(const rcl_wait_set_t &) override {return false;} + std::shared_ptr take_data() override {return nullptr;} + void execute(const std::shared_ptr &) override {} + void provide_intra_process_message(ConstMessageSharedPtr) override + {throw std::logic_error("unexpected shared delivery");} + void provide_intra_process_message(MessageUniquePtr value) override + {message = std::move(value);} + MessageUniquePtr message; + protected: void trigger_guard_condition() override {gc_.trigger();} }; @@ -175,4 +261,82 @@ TEST_P(TestIntraProcessCopyException, balances_message_allocations) INSTANTIATE_TEST_SUITE_P( BufferKinds, TestIntraProcessCopyException, ::testing::Values(Case{false, false}, Case{false, true}, Case{true, false}, Case{true, true})); + +TEST_F(TestIntraProcessCopyException, default_constructor_exception_deallocates_storage) +{ + auto counts = std::make_shared(); + Alloc allocator(counts); + using Traits = rclcpp::allocator::AllocRebind>; + + CopyMessage::throw_default = true; + EXPECT_THROW( + rclcpp::allocator::allocate_and_construct(allocator), std::runtime_error); + CopyMessage::throw_default = false; + + EXPECT_EQ(1u, counts->allocated); + EXPECT_EQ(counts->allocated, counts->deallocated); +} + +TEST_F(TestIntraProcessCopyException, adapted_ros_conversion_exception_destroys_message) +{ + using Adapter = rclcpp::TypeAdapter; + TrackedROSMessage::alive = 0; + auto node = std::make_shared("adapted_copy_allocation_test"); + auto publisher = node->create_publisher("copy", 10); + rclcpp::experimental::IntraProcessManager manager; + const auto id = manager.add_publisher(publisher); + auto subscription = std::make_shared(); + manager.add_subscription>(subscription); + std::allocator allocator; + + TrackedROSMessage::throw_default = true; + EXPECT_THROW( + (manager.do_intra_process_publish>( + id, std::make_unique(), allocator)), + std::runtime_error); + TrackedROSMessage::throw_default = false; + EXPECT_EQ(0, TrackedROSMessage::alive); + + Adapter::throw_conversion = true; + EXPECT_THROW( + (manager.do_intra_process_publish>( + id, std::make_unique(), allocator)), + std::runtime_error); + Adapter::throw_conversion = false; + EXPECT_EQ(0, TrackedROSMessage::alive); + + manager.do_intra_process_publish>( + id, std::make_unique(), allocator); + ASSERT_NE(nullptr, subscription->message); + EXPECT_EQ(42, subscription->message->value); + subscription->message.reset(); + EXPECT_EQ(0, TrackedROSMessage::alive); +} + +TEST_F(TestIntraProcessCopyException, provide_exception_releases_copied_message) +{ + auto counts = std::make_shared(); + auto allocator = std::make_shared(counts); + { + auto node = std::make_shared("provide_exception_test"); + auto publisher = node->create_publisher("copy", 10); + rclcpp::experimental::IntraProcessManager manager; + const auto id = manager.add_publisher(publisher); + auto first = std::make_shared(true); + auto last = std::make_shared(); + manager.add_subscription>(first); + manager.add_subscription>(last); + + auto pointer = std::allocator_traits::allocate(*allocator, 1); + std::allocator_traits::construct(*allocator, pointer); + UniqueMessage message(pointer, Deleter(allocator.get())); + EXPECT_THROW( + (manager.do_intra_process_publish>( + id, std::move(message), *allocator)), + std::runtime_error); + EXPECT_EQ(nullptr, last->message); + } + EXPECT_EQ(2u, counts->allocated); + EXPECT_EQ(counts->allocated, counts->deallocated); +} } // namespace