From f3b1e719663aacfb13a5ffa2cb09b17246968417 Mon Sep 17 00:00:00 2001 From: Thomas Moore Date: Sat, 13 Jun 2026 21:49:52 -0400 Subject: [PATCH 1/2] Unlock mutex when getting status in take_event to prevent deadlock Signed-off-by: Thomas Moore --- .../src/custom_publisher_info.cpp | 42 +++++++++++++-- .../src/custom_subscriber_info.cpp | 51 ++++++++++++++++--- 2 files changed, 82 insertions(+), 11 deletions(-) diff --git a/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp b/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp index 8d8308f953..23b6301a98 100644 --- a/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp +++ b/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp @@ -108,6 +108,14 @@ bool RMWPublisherEvent::take_event( { assert(rmw_fastrtps_shared_cpp::internal::is_event_supported(event_type)); + // NOTE: the DataWriter status queries below (get_*_status) acquire the writer's + // internal mutex. Fast-DDS may invoke this publisher's listener callbacks -- which + // take on_new_event_m_ -- while holding that writer mutex. To avoid a lock-order + // inversion (ABBA deadlock) with the writer's threads, on_new_event_m_ must NOT be + // held across a writer get_*_status() call. Each case that queries the writer releases + // the lock around the query, copies into a local, then re-acquires and adopts the + // value only if a listener update has not arrived in the meantime. (Symmetric to the + // subscription-side fix in custom_subscriber_info.cpp.) rcpputils::unique_lock lock_mutex(on_new_event_m_); switch (event_type) { @@ -117,7 +125,14 @@ bool RMWPublisherEvent::take_event( if (liveliness_changed_) { liveliness_changed_ = false; } else { - publisher_info_->data_writer_->get_liveliness_lost_status(liveliness_lost_status_); + eprosima::fastdds::dds::LivelinessLostStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_liveliness_lost_status(current); + lock_mutex.lock(); + if (!liveliness_changed_) { + liveliness_lost_status_ = current; + } + liveliness_changed_ = false; } rmw_data->total_count = liveliness_lost_status_.total_count; rmw_data->total_count_change = liveliness_lost_status_.total_count_change; @@ -130,8 +145,14 @@ bool RMWPublisherEvent::take_event( if (deadline_changed_) { deadline_changed_ = false; } else { - publisher_info_->data_writer_->get_offered_deadline_missed_status( - offered_deadline_missed_status_); + eprosima::fastdds::dds::OfferedDeadlineMissedStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_offered_deadline_missed_status(current); + lock_mutex.lock(); + if (!deadline_changed_) { + offered_deadline_missed_status_ = current; + } + deadline_changed_ = false; } rmw_data->total_count = offered_deadline_missed_status_.total_count; rmw_data->total_count_change = offered_deadline_missed_status_.total_count_change; @@ -144,8 +165,14 @@ bool RMWPublisherEvent::take_event( if (incompatible_qos_changed_) { incompatible_qos_changed_ = false; } else { - publisher_info_->data_writer_->get_offered_incompatible_qos_status( - incompatible_qos_status_); + eprosima::fastdds::dds::OfferedIncompatibleQosStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_offered_incompatible_qos_status(current); + lock_mutex.lock(); + if (!incompatible_qos_changed_) { + incompatible_qos_status_ = current; + } + incompatible_qos_changed_ = false; } rmw_data->total_count = incompatible_qos_status_.total_count; rmw_data->total_count_change = incompatible_qos_status_.total_count_change; @@ -157,6 +184,9 @@ bool RMWPublisherEvent::take_event( break; case RMW_EVENT_PUBLISHER_INCOMPATIBLE_TYPE: { + // get_inconsistent_topic_status() operates on the Topic, not the DataWriter, so + // it does not take the writer mutex and is not part of the writer-mutex inversion; + // left querying under the lock unchanged. auto rmw_data = static_cast(event_info); if (inconsistent_topic_changed_) { inconsistent_topic_changed_ = false; @@ -174,7 +204,9 @@ bool RMWPublisherEvent::take_event( auto rmw_data = static_cast(event_info); eprosima::fastdds::dds::PublicationMatchedStatus matched_status; + lock_mutex.unlock(); publisher_info_->data_writer_->get_publication_matched_status(matched_status); + lock_mutex.lock(); rmw_data->total_count = static_cast(matched_status.total_count); rmw_data->current_count = static_cast(matched_status.current_count); diff --git a/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp b/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp index 36e1b5cdc1..40638291ce 100644 --- a/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp +++ b/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp @@ -128,6 +128,14 @@ bool RMWSubscriptionEvent::take_event( { assert(rmw_fastrtps_shared_cpp::internal::is_event_supported(event_type)); + // NOTE: the DataReader status queries below (get_*_status) acquire the reader's + // internal mutex. Fast-DDS may invoke this subscription's listener callbacks -- + // which take on_new_event_m_ -- while holding that reader mutex (e.g. + // StatefulReader::processHeartbeatMsg -> on_sample_lost). To avoid a lock-order + // inversion (ABBA deadlock) with the reader's receive threads, on_new_event_m_ must + // NOT be held across a reader get_*_status() call. Each case that queries the reader + // releases the lock around the query, copies into a local, then re-acquires and + // adopts the value only if a listener update has not arrived in the meantime. rcpputils::unique_lock lock_mutex(on_new_event_m_); switch (event_type) { @@ -137,7 +145,14 @@ bool RMWSubscriptionEvent::take_event( if (liveliness_changed_) { liveliness_changed_ = false; } else { - subscriber_info_->data_reader_->get_liveliness_changed_status(liveliness_changed_status_); + eprosima::fastdds::dds::LivelinessChangedStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_liveliness_changed_status(current); + lock_mutex.lock(); + if (!liveliness_changed_) { + liveliness_changed_status_ = current; + } + liveliness_changed_ = false; } rmw_data->alive_count = liveliness_changed_status_.alive_count; rmw_data->not_alive_count = liveliness_changed_status_.not_alive_count; @@ -153,8 +168,14 @@ bool RMWSubscriptionEvent::take_event( if (deadline_changed_) { deadline_changed_ = false; } else { - subscriber_info_->data_reader_->get_requested_deadline_missed_status( - requested_deadline_missed_status_); + eprosima::fastdds::dds::RequestedDeadlineMissedStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_requested_deadline_missed_status(current); + lock_mutex.lock(); + if (!deadline_changed_) { + requested_deadline_missed_status_ = current; + } + deadline_changed_ = false; } rmw_data->total_count = requested_deadline_missed_status_.total_count; rmw_data->total_count_change = requested_deadline_missed_status_.total_count_change; @@ -167,7 +188,14 @@ bool RMWSubscriptionEvent::take_event( if (sample_lost_changed_) { sample_lost_changed_ = false; } else { - subscriber_info_->data_reader_->get_sample_lost_status(sample_lost_status_); + eprosima::fastdds::dds::SampleLostStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_sample_lost_status(current); + lock_mutex.lock(); + if (!sample_lost_changed_) { + sample_lost_status_ = current; + } + sample_lost_changed_ = false; } rmw_data->total_count = sample_lost_status_.total_count; rmw_data->total_count_change = sample_lost_status_.total_count_change; @@ -180,8 +208,14 @@ bool RMWSubscriptionEvent::take_event( if (incompatible_qos_changed_) { incompatible_qos_changed_ = false; } else { - subscriber_info_->data_reader_->get_requested_incompatible_qos_status( - incompatible_qos_status_); + eprosima::fastdds::dds::RequestedIncompatibleQosStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_requested_incompatible_qos_status(current); + lock_mutex.lock(); + if (!incompatible_qos_changed_) { + incompatible_qos_status_ = current; + } + incompatible_qos_changed_ = false; } rmw_data->total_count = incompatible_qos_status_.total_count; rmw_data->total_count_change = incompatible_qos_status_.total_count_change; @@ -193,6 +227,9 @@ bool RMWSubscriptionEvent::take_event( break; case RMW_EVENT_SUBSCRIPTION_INCOMPATIBLE_TYPE: { + // get_inconsistent_topic_status() operates on the Topic, not the DataReader, so + // it does not take the reader mutex and is not part of the reader-mutex inversion; + // left querying under the lock unchanged. auto rmw_data = static_cast(event_info); if (inconsistent_topic_changed_) { inconsistent_topic_changed_ = false; @@ -209,7 +246,9 @@ bool RMWSubscriptionEvent::take_event( auto rmw_data = static_cast(event_info); eprosima::fastdds::dds::SubscriptionMatchedStatus matched_status; + lock_mutex.unlock(); subscriber_info_->data_reader_->get_subscription_matched_status(matched_status); + lock_mutex.lock(); rmw_data->total_count = static_cast(matched_status.total_count); rmw_data->total_count_change = static_cast(matched_status.total_count_change); From 5a77dccd4d917ccfcddb05f1a4b39e53af144d3b Mon Sep 17 00:00:00 2001 From: Thomas Moore Date: Sun, 14 Jun 2026 02:09:47 +0000 Subject: [PATCH 2/2] Unlock mutex when getting status in set_on_new_event_callback to prevent deadlock Signed-off-by: Thomas Moore --- .../src/custom_publisher_info.cpp | 85 ++++++++++++++----- .../src/custom_subscriber_info.cpp | 58 +++++++++++-- 2 files changed, 112 insertions(+), 31 deletions(-) diff --git a/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp b/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp index 23b6301a98..99acdff5f5 100644 --- a/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp +++ b/rmw_fastrtps_shared_cpp/src/custom_publisher_info.cpp @@ -236,52 +236,91 @@ void RMWPublisherEvent::set_on_new_event_callback( const void * user_data, rmw_event_callback_t callback) { - rcpputils::unique_lock lock_mutex(on_new_event_m_); - + // NOTE: like take_event(), this method must not hold on_new_event_m_ across any + // DataWriter call (get_status_mask / get_*_status / set_listener): those acquire the + // writer's internal mutex, and Fast-DDS may invoke this publisher's listener callbacks + // -- which take on_new_event_m_ -- while holding that writer mutex. Holding + // on_new_event_m_ across a writer call is therefore a lock-order inversion (ABBA + // deadlock). The lock is released around every writer query below; each get_*_status + // copies into a local and re-acquires before adopting, keeping any value a listener + // delivered during the unlocked window. get_inconsistent_topic_status() operates on + // the Topic (not the writer) and is left under the lock, as in take_event(). eprosima::fastdds::dds::StatusMask status_mask = publisher_info_->data_writer_->get_status_mask(); + rcpputils::unique_lock lock_mutex(on_new_event_m_); + if (callback) { switch (event_type) { case RMW_EVENT_LIVELINESS_LOST: - publisher_info_->data_writer_->get_liveliness_lost_status(liveliness_lost_status_); + { + eprosima::fastdds::dds::LivelinessLostStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_liveliness_lost_status(current); + lock_mutex.lock(); + if (!liveliness_changed_) { + liveliness_lost_status_ = current; + } - if (liveliness_lost_status_.total_count_change > 0) { - callback(user_data, liveliness_lost_status_.total_count_change); - liveliness_lost_status_.total_count_change = 0; + if (liveliness_lost_status_.total_count_change > 0) { + callback(user_data, liveliness_lost_status_.total_count_change); + liveliness_lost_status_.total_count_change = 0; + } } break; case RMW_EVENT_OFFERED_DEADLINE_MISSED: - publisher_info_->data_writer_->get_offered_deadline_missed_status( - offered_deadline_missed_status_); + { + eprosima::fastdds::dds::OfferedDeadlineMissedStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_offered_deadline_missed_status(current); + lock_mutex.lock(); + if (!deadline_changed_) { + offered_deadline_missed_status_ = current; + } - if (offered_deadline_missed_status_.total_count_change > 0) { - callback(user_data, offered_deadline_missed_status_.total_count_change); - offered_deadline_missed_status_.total_count_change = 0; + if (offered_deadline_missed_status_.total_count_change > 0) { + callback(user_data, offered_deadline_missed_status_.total_count_change); + offered_deadline_missed_status_.total_count_change = 0; + } } break; case RMW_EVENT_OFFERED_QOS_INCOMPATIBLE: - publisher_info_->data_writer_->get_offered_incompatible_qos_status( - incompatible_qos_status_); + { + eprosima::fastdds::dds::OfferedIncompatibleQosStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_offered_incompatible_qos_status(current); + lock_mutex.lock(); + if (!incompatible_qos_changed_) { + incompatible_qos_status_ = current; + } - if (incompatible_qos_status_.total_count_change > 0) { - callback(user_data, incompatible_qos_status_.total_count_change); - incompatible_qos_status_.total_count_change = 0; + if (incompatible_qos_status_.total_count_change > 0) { + callback(user_data, incompatible_qos_status_.total_count_change); + incompatible_qos_status_.total_count_change = 0; + } } break; case RMW_EVENT_PUBLISHER_INCOMPATIBLE_TYPE: - publisher_info_->data_writer_->get_topic()->get_inconsistent_topic_status( - inconsistent_topic_status_); - if (inconsistent_topic_status_.total_count_change > 0) { - callback(user_data, inconsistent_topic_status_.total_count_change); - inconsistent_topic_status_.total_count_change = 0; + { + // get_inconsistent_topic_status() operates on the Topic, not the DataWriter, + // so it does not take the writer mutex; left querying under the lock. + publisher_info_->data_writer_->get_topic()->get_inconsistent_topic_status( + inconsistent_topic_status_); + if (inconsistent_topic_status_.total_count_change > 0) { + callback(user_data, inconsistent_topic_status_.total_count_change); + inconsistent_topic_status_.total_count_change = 0; + } } break; case RMW_EVENT_PUBLICATION_MATCHED: { if (matched_status_.total_count_change > 0) { callback(user_data, matched_status_.total_count_change); - publisher_info_->data_writer_->get_publication_matched_status(matched_status_); + eprosima::fastdds::dds::PublicationMatchedStatus current; + lock_mutex.unlock(); + publisher_info_->data_writer_->get_publication_matched_status(current); + lock_mutex.lock(); + matched_status_ = current; matched_status_.total_count_change = 0; matched_status_.current_count_change = 0; } @@ -306,6 +345,8 @@ void RMWPublisherEvent::set_on_new_event_callback( } } + // set_listener() acquires the writer's internal mutex; release on_new_event_m_ first. + lock_mutex.unlock(); publisher_info_->data_writer_->set_listener(publisher_info_->data_writer_listener_, status_mask); } diff --git a/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp b/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp index 40638291ce..faaaa41b85 100644 --- a/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp +++ b/rmw_fastrtps_shared_cpp/src/custom_subscriber_info.cpp @@ -278,16 +278,31 @@ void RMWSubscriptionEvent::set_on_new_event_callback( const void * user_data, rmw_event_callback_t callback) { - rcpputils::unique_lock lock_mutex(on_new_event_m_); - + // NOTE: like take_event(), this method must not hold on_new_event_m_ across any + // DataReader call (get_status_mask / get_*_status / set_listener): those acquire the + // reader's internal mutex, and Fast-DDS may invoke this subscription's listener + // callbacks -- which take on_new_event_m_ -- while holding that reader mutex. Holding + // on_new_event_m_ across a reader call is therefore a lock-order inversion (ABBA + // deadlock). The lock is released around every reader query below; each get_*_status + // copies into a local and re-acquires before adopting, keeping any value a listener + // delivered during the unlocked window. get_inconsistent_topic_status() operates on + // the Topic (not the reader) and is left under the lock, as in take_event(). eprosima::fastdds::dds::StatusMask status_mask = subscriber_info_->data_reader_->get_status_mask(); + rcpputils::unique_lock lock_mutex(on_new_event_m_); + if (callback) { switch (event_type) { case RMW_EVENT_LIVELINESS_CHANGED: { - subscriber_info_->data_reader_->get_liveliness_changed_status(liveliness_changed_status_); + eprosima::fastdds::dds::LivelinessChangedStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_liveliness_changed_status(current); + lock_mutex.lock(); + if (!liveliness_changed_) { + liveliness_changed_status_ = current; + } if ((liveliness_changed_status_.alive_count_change > 0) || (liveliness_changed_status_.not_alive_count_change > 0)) @@ -303,8 +318,13 @@ void RMWSubscriptionEvent::set_on_new_event_callback( break; case RMW_EVENT_REQUESTED_DEADLINE_MISSED: { - subscriber_info_->data_reader_->get_requested_deadline_missed_status( - requested_deadline_missed_status_); + eprosima::fastdds::dds::RequestedDeadlineMissedStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_requested_deadline_missed_status(current); + lock_mutex.lock(); + if (!deadline_changed_) { + requested_deadline_missed_status_ = current; + } if (requested_deadline_missed_status_.total_count_change > 0) { callback(user_data, requested_deadline_missed_status_.total_count_change); @@ -314,7 +334,13 @@ void RMWSubscriptionEvent::set_on_new_event_callback( break; case RMW_EVENT_MESSAGE_LOST: { - subscriber_info_->data_reader_->get_sample_lost_status(sample_lost_status_); + eprosima::fastdds::dds::SampleLostStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_sample_lost_status(current); + lock_mutex.lock(); + if (!sample_lost_changed_) { + sample_lost_status_ = current; + } if (sample_lost_status_.total_count_change > 0) { callback(user_data, sample_lost_status_.total_count_change); @@ -324,8 +350,13 @@ void RMWSubscriptionEvent::set_on_new_event_callback( break; case RMW_EVENT_REQUESTED_QOS_INCOMPATIBLE: { - subscriber_info_->data_reader_->get_requested_incompatible_qos_status( - incompatible_qos_status_); + eprosima::fastdds::dds::RequestedIncompatibleQosStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_requested_incompatible_qos_status(current); + lock_mutex.lock(); + if (!incompatible_qos_changed_) { + incompatible_qos_status_ = current; + } if (incompatible_qos_status_.total_count_change > 0) { callback(user_data, incompatible_qos_status_.total_count_change); @@ -335,6 +366,8 @@ void RMWSubscriptionEvent::set_on_new_event_callback( break; case RMW_EVENT_SUBSCRIPTION_INCOMPATIBLE_TYPE: { + // get_inconsistent_topic_status() operates on the Topic, not the DataReader, + // so it does not take the reader mutex; left querying under the lock. subscriber_info_->topic_->get_inconsistent_topic_status(inconsistent_topic_status_); if (inconsistent_topic_status_.total_count_change > 0) { callback(user_data, inconsistent_topic_status_.total_count_change); @@ -346,11 +379,16 @@ void RMWSubscriptionEvent::set_on_new_event_callback( { if (matched_status_.total_count_change > 0) { callback(user_data, matched_status_.total_count_change); - subscriber_info_->data_reader_->get_subscription_matched_status(matched_status_); + eprosima::fastdds::dds::SubscriptionMatchedStatus current; + lock_mutex.unlock(); + subscriber_info_->data_reader_->get_subscription_matched_status(current); + lock_mutex.lock(); + matched_status_ = current; matched_status_.total_count_change = 0; matched_status_.current_count_change = 0; } } + break; default: break; } @@ -370,6 +408,8 @@ void RMWSubscriptionEvent::set_on_new_event_callback( } } + // set_listener() acquires the reader's internal mutex; release on_new_event_m_ first. + lock_mutex.unlock(); subscriber_info_->data_reader_->set_listener( subscriber_info_->data_reader_listener_, status_mask); }