Skip to content

Fix loaned message ownership transfer by returning loans via shared_ptr custom deleter #3280

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Aaravanand00 wants to merge 1 commit into
base: rolling
Choose a base branch
from
Open

Fix loaned message ownership transfer by returning loans via shared_ptr custom deleter #3280

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
25 changes: 20 additions & 5 deletions rclcpp/include/rclcpp/subscription.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,17 +390,32 @@ class Subscription : public SubscriptionBase
void * loaned_message,
const rclcpp::MessageInfo & message_info) override
{
auto typed_message = static_cast<ROSMessageType *>(loaned_message);

auto sub_handle = this->get_subscription_handle();
auto loan_mutex = this->loaned_message_mutex_;
auto sptr = std::shared_ptr<ROSMessageType>(
typed_message, [sub_handle, loan_mutex](ROSMessageType * msg) {
Comment thread
Aaravanand00 marked this conversation as resolved.

@clalancette clalancette Oct 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we have a lifetime issue with sub_handle now if a user holds onto the message shared pointer. The deleter captures both sub_handle and loan_mutex shared pointers, so if SubscriptionBase gets destructed, those are both still valid until the user drops it. That is fine.

However, the comments in

rclcpp/rclcpp/include/rclcpp/subscription_base.hpp

Lines 583 to 589 in 249154a

// It is important to declare on_new_message_callback_ before
// subscription_handle_, so on destruction the subscription is
// destroyed first. Otherwise, the rmw subscription callback
// would point briefly to a destroyed function.
std::function<void(size_t)> on_new_message_callback_{nullptr};
// Declare subscription_handle_ after callback
std::shared_ptr<rcl_subscription_t> subscription_handle_;
explicitly say that subscription_handle_ to be destroyed before on_new_message_callback so we don't end up in a situation where the callback can fire on a destroyed subscription_handle_. That is now violated because the subscription_handle_ can outlive the SubscriptionBase. We can probably get away with explicitly clearing the on_new_message_callback in the SubscriptionBase destructor to make sure this doesn't happen.

if (msg) {
std::lock_guard<std::mutex> lock(*loan_mutex);
rcl_ret_t ret = rcl_return_loaned_message_from_subscription(
sub_handle.get(), static_cast<void *>(msg));
if (RCL_RET_OK != ret) {
RCLCPP_ERROR(
rclcpp::get_logger("rclcpp"),
"rcl_return_loaned_message_from_subscription() failed: %s",
rcl_get_error_string().str);
rcl_reset_error();
}
}
});

if (matches_any_intra_process_publishers(&message_info.get_rmw_message_info().publisher_gid)) {
// In this case, the message will be delivered via intra process and
// we should ignore this copy of the message.
return;
}

auto typed_message = static_cast<ROSMessageType *>(loaned_message);
// message is loaned, so we have to make sure that the deleter does not deallocate the message
auto sptr = std::shared_ptr<ROSMessageType>(
typed_message, [](ROSMessageType * msg) {(void) msg;});

std::chrono::time_point<std::chrono::system_clock> now;
if (subscription_topic_statistics_) {
// get current time before executing callback to
Expand Down
11 changes: 11 additions & 0 deletions rclcpp/include/rclcpp/subscription_base.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -650,8 +650,16 @@ class SubscriptionBase : public std::enable_shared_from_this<SubscriptionBase>
take_dynamic_message(
rclcpp::dynamic_typesupport::DynamicMessage & message_out,
rclcpp::MessageInfo & message_info_out);

RCLCPP_PUBLIC
std::shared_ptr<std::mutex>
get_loaned_message_mutex() const
{
return loaned_message_mutex_;
}
Comment on lines +654 to +659

@clalancette clalancette Oct 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not a fan of exposing implementations details like this, particularly around locks. It just makes it much harder to see how the locking is used and expected to be used. I'm going to suggest that we change the implementation to a take_loaned_message API here on SubscriptionBase which holds the lock internally, removing the need for this.

// ===============================================================================================


protected:
template<typename EventCallbackT>
void
Expand Down Expand Up @@ -688,6 +696,9 @@ class SubscriptionBase : public std::enable_shared_from_this<SubscriptionBase>
std::shared_ptr<rcl_node_t> node_handle_;

std::recursive_mutex on_new_message_callback_mutex_;

/// Mutex to protect rcl_take_loaned_message and rcl_return_loaned_message_from_subscription
std::shared_ptr<std::mutex> loaned_message_mutex_;
// It is important to declare on_new_message_callback_ before
// subscription_handle_, so on destruction the subscription is
// destroyed first. Otherwise, the rmw subscription callback
Expand Down
15 changes: 2 additions & 13 deletions rclcpp/src/rclcpp/executor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -575,6 +575,8 @@ Executor::execute_subscription(const rclcpp::SubscriptionBase::SharedPtr & subsc
subscription->get_topic_name(),
[&]()
{
std::shared_ptr<std::mutex> loan_mutex = subscription->get_loaned_message_mutex();
std::lock_guard<std::mutex> lock(*loan_mutex);
rcl_ret_t ret = rcl_take_loaned_message(
subscription->get_subscription_handle().get(),
&loaned_msg,
Expand All @@ -589,19 +591,6 @@ Executor::execute_subscription(const rclcpp::SubscriptionBase::SharedPtr & subsc
return true;
},
[&]() {subscription->handle_loaned_message(loaned_msg, message_info);});
if (nullptr != loaned_msg) {
rcl_ret_t ret = rcl_return_loaned_message_from_subscription(
subscription->get_subscription_handle().get(), loaned_msg);
if (RCL_RET_OK != ret) {
RCLCPP_ERROR(
rclcpp::get_logger("rclcpp"),
"rcl_return_loaned_message_from_subscription() failed for subscription on topic "
"'%s': %s",
subscription->get_topic_name(), rcl_get_error_string().str);
rcl_reset_error();
}
loaned_msg = nullptr;
}
} else {
// This case is taking a copy of the message data from the middleware via
// inter-process communication.
Expand Down
18 changes: 3 additions & 15 deletions rclcpp/src/rclcpp/subscription_base.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ SubscriptionBase::SubscriptionBase(
intra_process_subscription_id_(0),
event_callbacks_(event_callbacks),
type_support_(type_support_handle),
delivered_message_kind_(delivered_message_kind)
delivered_message_kind_(delivered_message_kind),
loaned_message_mutex_(std::make_shared<std::mutex>())
{
auto custom_deletor = [node_handle = this->node_handle_](rcl_subscription_t * rcl_subs)
{
Expand Down Expand Up @@ -337,20 +338,7 @@ SubscriptionBase::setup_intra_process(
bool
SubscriptionBase::can_loan_messages() const
{
bool retval = rcl_subscription_can_loan_messages(subscription_handle_.get());
if (retval) {
// TODO(clalancette): The loaned message interface is currently not safe to use with
// shared_ptr callbacks. If a user takes a copy of the shared_ptr, it can get freed from
// underneath them via rcl_return_loaned_message_from_subscription(). The correct solution is
// to return the loaned message in a custom deleter, but that needs to be carefully handled
// with locking. Warn the user about this until we fix it.
RCLCPP_WARN_ONCE(
this->node_logger_,
"Loaned messages are only safe with const ref subscription callbacks. "
"If you are using any other kind of subscriptions, "
"set the ROS_DISABLE_LOANED_MESSAGES environment variable to 1 (the default).");
}
return retval;
return rcl_subscription_can_loan_messages(subscription_handle_.get());
}

rclcpp::Waitable::SharedPtr
Expand Down
8 changes: 0 additions & 8 deletions rclcpp/test/rclcpp/test_subscription.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,15 +351,7 @@ TEST_F(TestSubscription, rcl_subscription_get_publisher_count_error) {
EXPECT_THROW(sub->get_publisher_count(), rclcpp::exceptions::RCLError);
}

TEST_F(TestSubscription, handle_loaned_message) {
initialize();
auto callback = [](std::shared_ptr<const test_msgs::msg::Empty>) {};
auto sub = node_->create_subscription<test_msgs::msg::Empty>("topic", 10, callback);

test_msgs::msg::Empty msg;
rclcpp::MessageInfo message_info;
EXPECT_NO_THROW(sub->handle_loaned_message(&msg, message_info));
}

/*
Testing on_new_message callbacks.
Expand Down