diff --git a/rclcpp/CMakeLists.txt b/rclcpp/CMakeLists.txt index d56e8d3919..30d1164d25 100644 --- a/rclcpp/CMakeLists.txt +++ b/rclcpp/CMakeLists.txt @@ -8,7 +8,6 @@ find_package(ament_cmake_ros REQUIRED) find_package(ament_cmake_ros_core REQUIRED) find_package(ament_index_cpp REQUIRED) find_package(builtin_interfaces REQUIRED) -find_package(libstatistics_collector REQUIRED) find_package(rcl REQUIRED) find_package(rcl_interfaces REQUIRED) find_package(rcl_logging_interface REQUIRED) @@ -22,7 +21,6 @@ find_package(rosidl_runtime_c REQUIRED) find_package(rosidl_runtime_cpp REQUIRED) find_package(rosidl_typesupport_c REQUIRED) find_package(rosidl_typesupport_cpp REQUIRED) -find_package(statistics_msgs REQUIRED) find_package(tracetools REQUIRED) if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang") @@ -200,7 +198,6 @@ target_include_directories(${PROJECT_NAME} PUBLIC "$") target_link_libraries(${PROJECT_NAME} PUBLIC builtin_interfaces::builtin_interfaces - libstatistics_collector::libstatistics_collector rcl::rcl rcl_interfaces::rcl_interfaces rcl_yaml_param_parser::rcl_yaml_param_parser @@ -212,7 +209,6 @@ target_link_libraries(${PROJECT_NAME} PUBLIC rosidl_runtime_c::rosidl_runtime_c rosidl_runtime_cpp::rosidl_runtime_cpp rosidl_typesupport_cpp::rosidl_typesupport_cpp - statistics_msgs::statistics_msgs tracetools::tracetools ${CMAKE_THREAD_LIBS_INIT} # Note we link public on purpose, as we want to export @@ -246,7 +242,6 @@ ament_export_targets(${PROJECT_NAME}) ament_export_dependencies( builtin_interfaces - libstatistics_collector rcl rcl_interfaces rcl_yaml_param_parser @@ -258,7 +253,6 @@ ament_export_dependencies( rosidl_runtime_c rosidl_runtime_cpp rosidl_typesupport_cpp - statistics_msgs tracetools ) diff --git a/rclcpp/QUALITY_DECLARATION.md b/rclcpp/QUALITY_DECLARATION.md index 7aba6a2c4b..5c5fec9889 100644 --- a/rclcpp/QUALITY_DECLARATION.md +++ b/rclcpp/QUALITY_DECLARATION.md @@ -195,12 +195,6 @@ It is **Quality Level 1**, see its [Quality Declaration document](https://github It is **Quality Level 1**, see its [Quality Declaration document](https://github.com/ros2/rmw/blob/rolling/rmw/QUALITY_DECLARATION.md). -#### `statistics_msgs` - -The `statistics_msgs` package contains ROS 2 message definitions for reporting statistics for topics and system resources. - -It is **Quality Level 1**, see its [Quality Declaration document](https://github.com/ros2/rcl_interfaces/blob/rolling/statistics_msgs/QUALITY_DECLARATION.md). - #### `tracetools` The `tracetools` package provides utilities for instrumenting the code in `rclcpp` so that it may be traced for debugging and performance analysis. diff --git a/rclcpp/include/rclcpp/create_subscription.hpp b/rclcpp/include/rclcpp/create_subscription.hpp index 0bf9a7c19a..94e6fd908b 100644 --- a/rclcpp/include/rclcpp/create_subscription.hpp +++ b/rclcpp/include/rclcpp/create_subscription.hpp @@ -22,19 +22,14 @@ #include #include -#include "rclcpp/detail/resolve_enable_topic_statistics.hpp" - -#include "rclcpp/node_interfaces/get_node_timers_interface.hpp" +#include "rclcpp/detail/qos_parameters.hpp" #include "rclcpp/node_interfaces/get_node_topics_interface.hpp" -#include "rclcpp/node_interfaces/node_timers_interface.hpp" +#include "rclcpp/node_interfaces/node_parameters_interface.hpp" #include "rclcpp/node_interfaces/node_topics_interface.hpp" -#include "rclcpp/create_publisher.hpp" #include "rclcpp/qos.hpp" #include "rclcpp/subscription_factory.hpp" #include "rclcpp/subscription_options.hpp" -#include "rclcpp/timer.hpp" -#include "rclcpp/topic_statistics/subscription_topic_statistics.hpp" #include "rmw/qos_profiles.h" namespace rclcpp @@ -69,59 +64,11 @@ create_subscription( using rclcpp::node_interfaces::get_node_topics_interface; auto node_topics_interface = get_node_topics_interface(node_topics); - std::shared_ptr - subscription_topic_stats = nullptr; - - if (rclcpp::detail::resolve_enable_topic_statistics( - options, - *node_topics_interface->get_node_base_interface())) - { - if (options.topic_stats_options.publish_period <= std::chrono::milliseconds(0)) { - throw std::invalid_argument( - "topic_stats_options.publish_period must be greater than 0, specified value of " + - std::to_string(options.topic_stats_options.publish_period.count()) + " ms"); - } - - std::shared_ptr> - publisher = rclcpp::detail::create_publisher( - node_parameters, - node_topics_interface, - options.topic_stats_options.publish_topic, - options.topic_stats_options.qos); - - subscription_topic_stats = - std::make_shared( - node_topics_interface->get_node_base_interface()->get_name(), publisher); - - std::weak_ptr< - rclcpp::topic_statistics::SubscriptionTopicStatistics - > weak_subscription_topic_stats(subscription_topic_stats); - auto sub_call_back = [weak_subscription_topic_stats]() { - auto subscription_topic_stats = weak_subscription_topic_stats.lock(); - if (subscription_topic_stats) { - subscription_topic_stats->publish_message_and_reset_measurements(); - } - }; - - auto node_timer_interface = node_topics_interface->get_node_timers_interface(); - - auto timer = create_wall_timer( - std::chrono::duration_cast( - options.topic_stats_options.publish_period), - sub_call_back, - options.callback_group, - node_topics_interface->get_node_base_interface(), - node_timer_interface - ); - - subscription_topic_stats->set_publisher_timer(timer); - } - auto factory = rclcpp::create_subscription_factory( std::forward(callback), options, msg_mem_strat, - subscription_topic_stats + options.subscription_statistics_monitor ); const rclcpp::QoS & actual_qos = options.qos_overriding_options.get_policy_kinds().size() ? @@ -161,8 +108,7 @@ create_subscription( * \param options * \param msg_mem_strat * \return the created subscription - * \throws std::invalid_argument if topic statistics is enabled and the publish period is - * less than or equal to zero. + * \throws std::invalid_argument if the QoS is incompatible with intra-process. */ template< typename MessageT, diff --git a/rclcpp/include/rclcpp/detail/resolve_enable_topic_statistics.hpp b/rclcpp/include/rclcpp/detail/resolve_enable_topic_statistics.hpp deleted file mode 100644 index 5265f8af30..0000000000 --- a/rclcpp/include/rclcpp/detail/resolve_enable_topic_statistics.hpp +++ /dev/null @@ -1,53 +0,0 @@ -// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved. -// -// 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. - -#ifndef RCLCPP__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_ -#define RCLCPP__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_ - -#include - -#include "rclcpp/topic_statistics_state.hpp" - -namespace rclcpp -{ -namespace detail -{ - -/// Return whether or not topic statistics is enabled, resolving "NodeDefault" if needed. -template -bool -resolve_enable_topic_statistics(const OptionsT & options, const NodeBaseT & node_base) -{ - bool topic_stats_enabled = false; - switch (options.topic_stats_options.state) { - case TopicStatisticsState::Enable: - topic_stats_enabled = true; - break; - case TopicStatisticsState::Disable: - topic_stats_enabled = false; - break; - case TopicStatisticsState::NodeDefault: - topic_stats_enabled = node_base.get_enable_topic_statistics_default(); - break; - default: - throw std::runtime_error("Unrecognized EnableTopicStatistics value"); - } - - return topic_stats_enabled; -} - -} // namespace detail -} // namespace rclcpp - -#endif // RCLCPP__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_ diff --git a/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp b/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp index 6fbbf5503f..9d271db951 100644 --- a/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp +++ b/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp @@ -18,7 +18,6 @@ #include #include -#include #include #include #include @@ -33,7 +32,7 @@ #include "rclcpp/experimental/subscription_intra_process_buffer.hpp" #include "rclcpp/logging.hpp" #include "rclcpp/qos.hpp" -#include "rclcpp/time.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" #include "rclcpp/type_support_decl.hpp" #include "tracetools/tracetools.h" @@ -74,7 +73,8 @@ class SubscriptionIntraProcess using ConstMessageSharedPtr = typename SubscriptionIntraProcessBufferT::ConstDataSharedPtr; using MessageUniquePtr = typename SubscriptionIntraProcessBufferT::SubscribedTypeUniquePtr; using BufferUniquePtr = typename SubscriptionIntraProcessBufferT::BufferUniquePtr; - using StatsHandlerFn = std::function; + using SubscriptionStatisticsMonitorSharedPtr = + rclcpp::SubscriptionStatisticsMonitor::SharedPtr; SubscriptionIntraProcess( AnySubscriptionCallback callback, @@ -83,7 +83,7 @@ class SubscriptionIntraProcess const std::string & topic_name, const rclcpp::QoS & qos_profile, rclcpp::IntraProcessBufferType buffer_type, - StatsHandlerFn stats_handler = nullptr) + SubscriptionStatisticsMonitorSharedPtr monitor = nullptr) : SubscriptionIntraProcessBuffer( std::make_shared(*allocator), @@ -92,7 +92,7 @@ class SubscriptionIntraProcess qos_profile, buffer_type), any_callback_(callback), - stats_handler_(std::move(stats_handler)) + monitor_(std::move(monitor)) { TRACETOOLS_TRACEPOINT( rclcpp_subscription_callback_added, @@ -206,7 +206,7 @@ class SubscriptionIntraProcess const auto nanos = std::chrono::time_point_cast( std::chrono::system_clock::now()); - if (stats_handler_) { + if (monitor_) { RCLCPP_WARN_ONCE( rclcpp::get_logger("rclcpp"), "Intra-process communication does not support accurate message age statistics"); @@ -214,6 +214,7 @@ class SubscriptionIntraProcess // an invalid value taken from an un-initialised timestamp. IPC delivery // has little/no transport latency by definition, so near-zero age is expected. msg_info.source_timestamp = nanos.time_since_epoch().count(); + monitor_->before_message_dispatch(msg_info); } auto shared_ptr = std::static_pointer_cast>( @@ -228,13 +229,13 @@ class SubscriptionIntraProcess } shared_ptr.reset(); - if (stats_handler_) { - stats_handler_(msg_info, rclcpp::Time(nanos.time_since_epoch().count())); + if (monitor_) { + monitor_->after_message_dispatch(msg_info); } } AnySubscriptionCallback any_callback_; - StatsHandlerFn stats_handler_; + SubscriptionStatisticsMonitorSharedPtr monitor_; }; } // namespace experimental diff --git a/rclcpp/include/rclcpp/node_impl.hpp b/rclcpp/include/rclcpp/node_impl.hpp index 079250d58a..9328cdfe22 100644 --- a/rclcpp/include/rclcpp/node_impl.hpp +++ b/rclcpp/include/rclcpp/node_impl.hpp @@ -43,7 +43,6 @@ #include "rclcpp/create_generic_service.hpp" #include "rclcpp/create_subscription.hpp" #include "rclcpp/create_timer.hpp" -#include "rclcpp/detail/resolve_enable_topic_statistics.hpp" #include "rclcpp/parameter.hpp" #include "rclcpp/qos.hpp" #include "rclcpp/timer.hpp" diff --git a/rclcpp/include/rclcpp/node_interfaces/node_base.hpp b/rclcpp/include/rclcpp/node_interfaces/node_base.hpp index 2a71ae2f3e..e011ebfdb3 100644 --- a/rclcpp/include/rclcpp/node_interfaces/node_base.hpp +++ b/rclcpp/include/rclcpp/node_interfaces/node_base.hpp @@ -53,7 +53,6 @@ class NodeBase : public NodeBaseInterface, public std::enable_shared_from_this node_handle_; diff --git a/rclcpp/include/rclcpp/node_interfaces/node_base_interface.hpp b/rclcpp/include/rclcpp/node_interfaces/node_base_interface.hpp index 4986fd7729..35bf378c88 100644 --- a/rclcpp/include/rclcpp/node_interfaces/node_base_interface.hpp +++ b/rclcpp/include/rclcpp/node_interfaces/node_base_interface.hpp @@ -170,12 +170,6 @@ class NodeBaseInterface bool get_use_intra_process_default() const = 0; - /// Return the default preference for enabling topic statistics collection. - RCLCPP_PUBLIC - virtual - bool - get_enable_topic_statistics_default() const = 0; - /// Expand and remap a given topic or service name. RCLCPP_PUBLIC virtual diff --git a/rclcpp/include/rclcpp/node_options.hpp b/rclcpp/include/rclcpp/node_options.hpp index 212f94fe21..453d94cbbb 100644 --- a/rclcpp/include/rclcpp/node_options.hpp +++ b/rclcpp/include/rclcpp/node_options.hpp @@ -46,7 +46,6 @@ class NodeOptions * - use_global_arguments = true * - enable_rosout = true * - use_intra_process_comms = false - * - enable_topic_statistics = false * - start_parameter_services = true * - start_parameter_event_publisher = true * - clock_type = RCL_ROS_TIME @@ -208,23 +207,6 @@ class NodeOptions NodeOptions & use_intra_process_comms(bool use_intra_process_comms); - /// Return the enable_topic_statistics flag. - RCLCPP_PUBLIC - bool - enable_topic_statistics() const; - - /// Set the enable_topic_statistics flag, return this for parameter idiom. - /** - * If true, topic statistics collection and publication will be enabled - * for all subscriptions. - * This can be used to override the global topic statistics setting. - * - * Defaults to false. - */ - RCLCPP_PUBLIC - NodeOptions & - enable_topic_statistics(bool enable_topic_statistics); - /// Return the start_parameter_services flag. RCLCPP_PUBLIC bool @@ -455,8 +437,6 @@ class NodeOptions bool use_intra_process_comms_ {false}; - bool enable_topic_statistics_ {false}; - bool start_parameter_services_ {true}; bool start_parameter_event_publisher_ {true}; diff --git a/rclcpp/include/rclcpp/subscription.hpp b/rclcpp/include/rclcpp/subscription.hpp index 5f5e090294..4ff629a7a4 100644 --- a/rclcpp/include/rclcpp/subscription.hpp +++ b/rclcpp/include/rclcpp/subscription.hpp @@ -18,7 +18,6 @@ #include #include -#include #include #include #include @@ -46,7 +45,7 @@ #include "rclcpp/type_support_decl.hpp" #include "rclcpp/visibility_control.hpp" #include "rclcpp/waitable.hpp" -#include "rclcpp/topic_statistics/subscription_topic_statistics.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" #include "tracetools/tracetools.h" namespace rclcpp @@ -90,8 +89,8 @@ class Subscription : public SubscriptionBase using ROSMessageTypeDeleter = allocator::Deleter; private: - using SubscriptionTopicStatisticsSharedPtr = - std::shared_ptr; + using SubscriptionStatisticsMonitorSharedPtr = + std::shared_ptr; public: RCLCPP_SMART_PTR_DEFINITIONS(Subscription) @@ -109,7 +108,7 @@ class Subscription : public SubscriptionBase * \param[in] callback User defined callback to call when a message is received. * \param[in] options Options for the subscription. * \param[in] message_memory_strategy The memory strategy to be used for managing message memory. - * \param[in] subscription_topic_statistics Optional pointer to a topic statistics subcription. + * \param[in] subscription_statistics_monitor Optional monitor for subscription statistics. * \throws std::invalid_argument if the QoS is uncompatible with intra-process (if one * of the following conditions are true: qos_profile.history == RMW_QOS_POLICY_HISTORY_KEEP_ALL, * qos_profile.depth == 0 or qos_profile.durability != RMW_QOS_POLICY_DURABILITY_VOLATILE). @@ -123,7 +122,7 @@ class Subscription : public SubscriptionBase AnySubscriptionCallback callback, const rclcpp::SubscriptionOptionsWithAllocator & options, typename MessageMemoryStrategyT::SharedPtr message_memory_strategy, - SubscriptionTopicStatisticsSharedPtr subscription_topic_statistics = nullptr) + SubscriptionStatisticsMonitorSharedPtr subscription_statistics_monitor = nullptr) : SubscriptionBase( node_base, type_support_handle, @@ -163,18 +162,6 @@ class Subscription : public SubscriptionBase ROSMessageT, AllocatorT>; - // Build a type-erased stats handler to avoid a circular include chain - // via publisher.hpp and callback_group.hpp - typename SubscriptionIntraProcessT::StatsHandlerFn stats_handler = nullptr; - if (subscription_topic_statistics) { - stats_handler = - [subscription_topic_statistics]( - const rmw_message_info_t & info, const rclcpp::Time & time) - { - subscription_topic_statistics->handle_message(info, time); - }; - } - // First create a SubscriptionIntraProcess which will be given to the intra-process manager. auto context = node_base->get_context(); subscription_intra_process_ = std::make_shared( @@ -184,7 +171,7 @@ class Subscription : public SubscriptionBase this->get_topic_name(), // important to get like this, as it has the fully-qualified name qos_profile, resolve_intra_process_buffer_type(options_.intra_process_buffer_type, callback), - std::move(stats_handler)); + subscription_statistics_monitor); TRACETOOLS_TRACEPOINT( rclcpp_subscription_init, static_cast(get_subscription_handle().get()), @@ -198,9 +185,7 @@ class Subscription : public SubscriptionBase this->setup_intra_process(intra_process_subscription_id, ipm); } - if (subscription_topic_statistics != nullptr) { - this->subscription_topic_statistics_ = std::move(subscription_topic_statistics); - } + this->subscription_statistics_monitor_ = std::move(subscription_statistics_monitor); TRACETOOLS_TRACEPOINT( rclcpp_subscription_init, @@ -347,19 +332,16 @@ class Subscription : public SubscriptionBase } auto typed_message = std::static_pointer_cast(message); - std::chrono::time_point now; - if (subscription_topic_statistics_) { - // get current time before executing callback to - // exclude callback duration from topic statistics result. - now = std::chrono::system_clock::now(); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->before_message_dispatch( + message_info.get_rmw_message_info()); } any_callback_.dispatch(typed_message, message_info); - if (subscription_topic_statistics_) { - const auto nanos = std::chrono::time_point_cast(now); - const auto time = rclcpp::Time(nanos.time_since_epoch().count()); - subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->after_message_dispatch( + message_info.get_rmw_message_info()); } } @@ -368,19 +350,16 @@ class Subscription : public SubscriptionBase const std::shared_ptr & serialized_message, const rclcpp::MessageInfo & message_info) override { - std::chrono::time_point now; - if (subscription_topic_statistics_) { - // get current time before executing callback to - // exclude callback duration from topic statistics result. - now = std::chrono::system_clock::now(); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->before_message_dispatch( + message_info.get_rmw_message_info()); } any_callback_.dispatch(serialized_message, message_info); - if (subscription_topic_statistics_) { - const auto nanos = std::chrono::time_point_cast(now); - const auto time = rclcpp::Time(nanos.time_since_epoch().count()); - subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->after_message_dispatch( + message_info.get_rmw_message_info()); } } @@ -400,19 +379,16 @@ class Subscription : public SubscriptionBase auto sptr = std::shared_ptr( typed_message, [](ROSMessageType * msg) {(void) msg;}); - std::chrono::time_point now; - if (subscription_topic_statistics_) { - // get current time before executing callback to - // exclude callback duration from topic statistics result. - now = std::chrono::system_clock::now(); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->before_message_dispatch( + message_info.get_rmw_message_info()); } any_callback_.dispatch(sptr, message_info); - if (subscription_topic_statistics_) { - const auto nanos = std::chrono::time_point_cast(now); - const auto time = rclcpp::Time(nanos.time_since_epoch().count()); - subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time); + if (subscription_statistics_monitor_) { + subscription_statistics_monitor_->after_message_dispatch( + message_info.get_rmw_message_info()); } } @@ -504,8 +480,8 @@ class Subscription : public SubscriptionBase typename message_memory_strategy::MessageMemoryStrategy::SharedPtr message_memory_strategy_; - /// Component which computes and publishes topic statistics for this subscriber - SubscriptionTopicStatisticsSharedPtr subscription_topic_statistics_{nullptr}; + /// Optional external monitor for subscription statistics. + SubscriptionStatisticsMonitorSharedPtr subscription_statistics_monitor_{nullptr}; }; } // namespace rclcpp diff --git a/rclcpp/include/rclcpp/subscription_factory.hpp b/rclcpp/include/rclcpp/subscription_factory.hpp index 0e9d9fefe5..5446e83466 100644 --- a/rclcpp/include/rclcpp/subscription_factory.hpp +++ b/rclcpp/include/rclcpp/subscription_factory.hpp @@ -30,8 +30,8 @@ #include "rclcpp/qos.hpp" #include "rclcpp/subscription.hpp" #include "rclcpp/subscription_options.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" #include "rclcpp/subscription_traits.hpp" -#include "rclcpp/topic_statistics/subscription_topic_statistics.hpp" #include "rclcpp/visibility_control.hpp" namespace rclcpp @@ -68,7 +68,7 @@ struct SubscriptionFactory * \param[in] callback The user-defined callback function to receive a message * \param[in] options Additional options for the creation of the Subscription. * \param[in] msg_mem_strat The message memory strategy to use for allocating messages. - * \param[in] subscription_topic_stats Optional stats callback for topic_statistics + * \param[in] subscription_statistics_monitor Optional monitor for subscription statistics. */ template< typename MessageT, @@ -82,8 +82,8 @@ create_subscription_factory( CallbackT && callback, const rclcpp::SubscriptionOptionsWithAllocator & options, typename MessageMemoryStrategyT::SharedPtr msg_mem_strat, - std::shared_ptr - subscription_topic_stats = nullptr + std::shared_ptr + subscription_statistics_monitor = nullptr ) { auto allocator = options.get_allocator(); @@ -94,7 +94,7 @@ create_subscription_factory( SubscriptionFactory factory { // factory function that creates a MessageT specific SubscriptionT - [options, msg_mem_strat, any_subscription_callback, subscription_topic_stats]( + [options, msg_mem_strat, any_subscription_callback, subscription_statistics_monitor]( rclcpp::node_interfaces::NodeBaseInterface * node_base, const std::string & topic_name, const rclcpp::QoS & qos @@ -111,7 +111,7 @@ create_subscription_factory( any_subscription_callback, options, msg_mem_strat, - subscription_topic_stats); + subscription_statistics_monitor); // This is used for setting up things like intra process comms which // require this->shared_from_this() which cannot be called from // the constructor. diff --git a/rclcpp/include/rclcpp/subscription_options.hpp b/rclcpp/include/rclcpp/subscription_options.hpp index 822259289a..47c00e6616 100644 --- a/rclcpp/include/rclcpp/subscription_options.hpp +++ b/rclcpp/include/rclcpp/subscription_options.hpp @@ -15,7 +15,6 @@ #ifndef RCLCPP__SUBSCRIPTION_OPTIONS_HPP_ #define RCLCPP__SUBSCRIPTION_OPTIONS_HPP_ -#include #include #include #include @@ -29,8 +28,7 @@ #include "rclcpp/event_handler.hpp" #include "rclcpp/qos_overriding_options.hpp" #include "rclcpp/subscription_content_filter_options.hpp" -#include "rclcpp/topic_statistics_state.hpp" -#include "rclcpp/visibility_control.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" namespace rclcpp { @@ -65,26 +63,14 @@ struct SubscriptionOptionsBase std::shared_ptr rmw_implementation_payload = nullptr; - // Options to configure topic statistics collector in the subscription. - struct TopicStatisticsOptions - { - // Enable and disable topic statistics calculation and publication. Defaults to disabled. - TopicStatisticsState state = TopicStatisticsState::NodeDefault; - - // Topic to which topic statistics get published when enabled. Defaults to /statistics. - std::string publish_topic = "/statistics"; - - // Topic statistics publication period in ms. Defaults to one second. - // Only values greater than zero are allowed. - std::chrono::milliseconds publish_period{std::chrono::seconds(1)}; - - // An optional QoS which can provide topic_statistics with a stable QoS separate from - // the subscription's current QoS settings which could be unstable. - // Explicitly set the enough depth to avoid missing the statistics messages. - rclcpp::QoS qos = SystemDefaultsQoS().keep_last(10); - }; - - TopicStatisticsOptions topic_stats_options; + /// Optional external monitor for subscription statistics. + /** + * If set, the subscription will call before_message_dispatch() and + * after_message_dispatch() on this monitor for each received message. The + * monitor is responsible for any aggregation, timers, and publication of + * statistics. + */ + std::shared_ptr subscription_statistics_monitor; QosOverridingOptions qos_overriding_options; diff --git a/rclcpp/include/rclcpp/subscription_statistics_monitor.hpp b/rclcpp/include/rclcpp/subscription_statistics_monitor.hpp new file mode 100644 index 0000000000..53a361f074 --- /dev/null +++ b/rclcpp/include/rclcpp/subscription_statistics_monitor.hpp @@ -0,0 +1,62 @@ +// Copyright 2026 Open Source Robotics Foundation, Inc. +// +// 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. + +#ifndef RCLCPP__SUBSCRIPTION_STATISTICS_MONITOR_HPP_ +#define RCLCPP__SUBSCRIPTION_STATISTICS_MONITOR_HPP_ + +#include + +#include "rmw/types.h" + +#include "rclcpp/visibility_control.hpp" + +namespace rclcpp +{ + +/** + * Pure virtual interface that allows external code to monitor message reception + * for a single subscription. + * + * Implementations are responsible for any collection, aggregation, and + * publication of statistics. rclcpp calls before_message_dispatch() before + * the user callback and after_message_dispatch() after the user callback for + * each received message. + */ +class SubscriptionStatisticsMonitor +{ +public: + using SharedPtr = std::shared_ptr; + using WeakPtr = std::weak_ptr; + + RCLCPP_PUBLIC + virtual ~SubscriptionStatisticsMonitor() = default; + + /// Called once before the user callback is dispatched for a received message. + /** + * \param[in] message_info RMW message information for the received message. + */ + RCLCPP_PUBLIC + virtual void before_message_dispatch(const rmw_message_info_t & message_info) = 0; + + /// Called once after the user callback has been dispatched for a received message. + /** + * \param[in] message_info RMW message information for the received message. + */ + RCLCPP_PUBLIC + virtual void after_message_dispatch(const rmw_message_info_t & message_info) = 0; +}; + +} // namespace rclcpp + +#endif // RCLCPP__SUBSCRIPTION_STATISTICS_MONITOR_HPP_ diff --git a/rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp b/rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp deleted file mode 100644 index 7a6db13502..0000000000 --- a/rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp +++ /dev/null @@ -1,235 +0,0 @@ -// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved. -// -// 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. - -#ifndef RCLCPP__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_ -#define RCLCPP__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_ - -#include -#include -#include -#include - -#include "libstatistics_collector/collector/generate_statistics_message.hpp" -#include "libstatistics_collector/moving_average_statistics/types.hpp" -#include "libstatistics_collector/topic_statistics_collector/constants.hpp" -#include "libstatistics_collector/topic_statistics_collector/received_message_age.hpp" -#include "libstatistics_collector/topic_statistics_collector/received_message_period.hpp" - -#include "rcl/time.h" -#include "rclcpp/time.hpp" -#include "rclcpp/publisher.hpp" -#include "rclcpp/timer.hpp" - -#include "statistics_msgs/msg/metrics_message.hpp" - -namespace rclcpp -{ -namespace topic_statistics -{ - -constexpr const char kDefaultPublishTopicName[]{"/statistics"}; -constexpr const std::chrono::milliseconds kDefaultPublishingPeriod{std::chrono::seconds(1)}; - -using libstatistics_collector::collector::GenerateStatisticMessage; -using statistics_msgs::msg::MetricsMessage; -using libstatistics_collector::moving_average_statistics::StatisticData; - -/** - * Class used to collect, measure, and publish topic statistics data. Current statistics - * supported for subscribers are received message age and received message period. - */ -class SubscriptionTopicStatistics -{ - using TopicStatsCollector = libstatistics_collector::TopicStatisticsCollector; - using ReceivedMessageAge = libstatistics_collector::ReceivedMessageAgeCollector; - using ReceivedMessagePeriod = libstatistics_collector::ReceivedMessagePeriodCollector; - -public: - /// Construct a SubscriptionTopicStatistics object. - /** - * This object wraps utilities, defined in libstatistics_collector, to collect, - * measure, and publish topic statistics data. This throws an invalid_argument - * if the input publisher is null. - * - * \param node_name the name of the node, which created this instance, in order to denote - * topic source - * \param publisher instance constructed by the node in order to publish statistics data. - * This class owns the publisher. - * \throws std::invalid_argument if publisher pointer is nullptr - */ - SubscriptionTopicStatistics( - const std::string & node_name, - rclcpp::Publisher::SharedPtr publisher) - : node_name_(node_name), - publisher_(std::move(publisher)) - { - // TODO(dbbonnie): ros-tooling/aws-roadmap/issues/226, received message age - - if (nullptr == publisher_) { - throw std::invalid_argument("publisher pointer is nullptr"); - } - - bring_up(); - } - - virtual ~SubscriptionTopicStatistics() - { - tear_down(); - } - - /// Handle a message received by the subscription to collect statistics. - /** - * This method acquires a lock to prevent race conditions to collectors list. - * - * \param message_info the message info corresponding to the received message - * \param now_nanoseconds current time in nanoseconds - */ - virtual void handle_message( - const rmw_message_info_t & message_info, - const rclcpp::Time now_nanoseconds) const - { - std::lock_guard lock(mutex_); - for (const auto & collector : subscriber_statistics_collectors_) { - collector->OnMessageReceived(message_info, now_nanoseconds.nanoseconds()); - } - } - - /// Set the timer used to publish statistics messages. - /** - * \param publisher_timer the timer to fire the publisher, created by the node - */ - void set_publisher_timer(rclcpp::TimerBase::SharedPtr publisher_timer) - { - publisher_timer_ = publisher_timer; - } - - /// Publish a populated MetricsStatisticsMessage. - /** - * This method acquires a lock to prevent race conditions to collectors list. - */ - virtual void publish_message_and_reset_measurements() - { - std::vector msgs; - rclcpp::Time window_end{get_current_nanoseconds_since_epoch()}; - - { - std::lock_guard lock(mutex_); - for (auto & collector : subscriber_statistics_collectors_) { - const auto collected_stats = collector->GetStatisticsResults(); - collector->ClearCurrentMeasurements(); - - auto message = libstatistics_collector::collector::GenerateStatisticMessage( - node_name_, - collector->GetMetricName(), - collector->GetMetricUnit(), - window_start_, - window_end, - collected_stats); - msgs.push_back(message); - } - } - - for (auto & msg : msgs) { - publisher_->publish(msg); - } - window_start_ = window_end; - } - -protected: - /// Return a vector of all the currently collected data. - /** - * This method acquires a lock to prevent race conditions to collectors list. - * - * \return a vector of all the collected data - */ - std::vector get_current_collector_data() const - { - std::vector data; - std::lock_guard lock(mutex_); - for (const auto & collector : subscriber_statistics_collectors_) { - data.push_back(collector->GetStatisticsResults()); - } - return data; - } - -private: - /// Construct and start all collectors and set window_start_. - /** - * This method acquires a lock to prevent race conditions to collectors list. - */ - void bring_up() - { - auto received_message_age = std::make_unique(); - received_message_age->Start(); - auto received_message_period = std::make_unique(); - received_message_period->Start(); - { - std::lock_guard lock(mutex_); - subscriber_statistics_collectors_.emplace_back(std::move(received_message_age)); - subscriber_statistics_collectors_.emplace_back(std::move(received_message_period)); - } - - window_start_ = rclcpp::Time(get_current_nanoseconds_since_epoch()); - } - - /// Stop all collectors, clear measurements, stop publishing timer, and reset publisher. - /** - * This method acquires a lock to prevent race conditions to collectors list. - */ - void tear_down() - { - { - std::lock_guard lock(mutex_); - for (auto & collector : subscriber_statistics_collectors_) { - collector->Stop(); - } - - subscriber_statistics_collectors_.clear(); - } - - if (publisher_timer_) { - publisher_timer_->cancel(); - publisher_timer_.reset(); - } - - publisher_.reset(); - } - - /// Return the current nanoseconds (count) since epoch. - /** - * \return the current nanoseconds (count) since epoch - */ - int64_t get_current_nanoseconds_since_epoch() const - { - const auto now = std::chrono::system_clock::now(); - return std::chrono::duration_cast(now.time_since_epoch()).count(); - } - - /// Mutex to protect the subsequence vectors - mutable std::mutex mutex_; - /// Collection of statistics collectors - std::vector> subscriber_statistics_collectors_{}; - /// Node name used to generate topic statistics messages to be published - const std::string node_name_; - /// Publisher, created by the node, used to publish topic statistics messages - rclcpp::Publisher::SharedPtr publisher_; - /// Timer which fires the publisher - rclcpp::TimerBase::SharedPtr publisher_timer_; - /// The start of the collection window, used in the published topic statistics message - rclcpp::Time window_start_; -}; -} // namespace topic_statistics -} // namespace rclcpp - -#endif // RCLCPP__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_ diff --git a/rclcpp/include/rclcpp/topic_statistics_state.hpp b/rclcpp/include/rclcpp/topic_statistics_state.hpp deleted file mode 100644 index b407622a4e..0000000000 --- a/rclcpp/include/rclcpp/topic_statistics_state.hpp +++ /dev/null @@ -1,35 +0,0 @@ -// Copyright 2019 Open Source Robotics Foundation, Inc. -// -// 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. - -#ifndef RCLCPP__TOPIC_STATISTICS_STATE_HPP_ -#define RCLCPP__TOPIC_STATISTICS_STATE_HPP_ - -namespace rclcpp -{ - -/// Represent the state of topic statistics collector. -/// Used as argument in create_subscriber. -enum class TopicStatisticsState -{ - /// Explicitly enable topic statistics at subscription level. - Enable, - /// Explicitly disable topic statistics at subscription level. - Disable, - /// Take topic statistics state from the node. - NodeDefault -}; - -} // namespace rclcpp - -#endif // RCLCPP__TOPIC_STATISTICS_STATE_HPP_ diff --git a/rclcpp/package.xml b/rclcpp/package.xml index eb06999fff..c49bb06d97 100644 --- a/rclcpp/package.xml +++ b/rclcpp/package.xml @@ -37,7 +37,6 @@ rosidl_typesupport_c rosidl_typesupport_cpp - libstatistics_collector rcl rcl_logging_interface rcl_yaml_param_parser @@ -45,7 +44,6 @@ rcutils rmw rosidl_dynamic_typesupport - statistics_msgs tracetools ament_cmake_google_benchmark @@ -58,6 +56,7 @@ rmw rmw_implementation_cmake rosidl_default_generators + std_msgs test_msgs diff --git a/rclcpp/src/rclcpp/node.cpp b/rclcpp/src/rclcpp/node.cpp index a0ba436a0a..b90b52c0e0 100644 --- a/rclcpp/src/rclcpp/node.cpp +++ b/rclcpp/src/rclcpp/node.cpp @@ -181,8 +181,7 @@ Node::Node( namespace_, options.context(), *(options.get_rcl_node_options()), - options.use_intra_process_comms(), - options.enable_topic_statistics())), + options.use_intra_process_comms())), node_graph_(new rclcpp::node_interfaces::NodeGraph(node_base_.get())), node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_)), node_timers_(new rclcpp::node_interfaces::NodeTimers(node_base_.get())), diff --git a/rclcpp/src/rclcpp/node_interfaces/node_base.cpp b/rclcpp/src/rclcpp/node_interfaces/node_base.cpp index 9c27776264..ba364652a3 100644 --- a/rclcpp/src/rclcpp/node_interfaces/node_base.cpp +++ b/rclcpp/src/rclcpp/node_interfaces/node_base.cpp @@ -40,11 +40,9 @@ NodeBase::NodeBase( rclcpp::Context::SharedPtr context, const rcl_node_options_t & rcl_node_options, bool use_intra_process_default, - bool enable_topic_statistics_default, rclcpp::CallbackGroup::SharedPtr default_callback_group) : context_(std::move(context)), use_intra_process_default_(use_intra_process_default), - enable_topic_statistics_default_(enable_topic_statistics_default), node_handle_(nullptr), default_callback_group_(std::move(default_callback_group)), associated_with_executor_(false), @@ -301,12 +299,6 @@ NodeBase::get_use_intra_process_default() const return use_intra_process_default_; } -bool -NodeBase::get_enable_topic_statistics_default() const -{ - return enable_topic_statistics_default_; -} - std::string NodeBase::resolve_topic_or_service_name( const std::string & name, bool is_service, bool only_expand) const diff --git a/rclcpp/src/rclcpp/node_options.cpp b/rclcpp/src/rclcpp/node_options.cpp index 04ebf47079..1458179231 100644 --- a/rclcpp/src/rclcpp/node_options.cpp +++ b/rclcpp/src/rclcpp/node_options.cpp @@ -74,7 +74,6 @@ NodeOptions::operator=(const NodeOptions & other) this->use_global_arguments_ = other.use_global_arguments_; this->enable_rosout_ = other.enable_rosout_; this->use_intra_process_comms_ = other.use_intra_process_comms_; - this->enable_topic_statistics_ = other.enable_topic_statistics_; this->start_parameter_services_ = other.start_parameter_services_; this->start_parameter_event_publisher_ = other.start_parameter_event_publisher_; this->clock_type_ = other.clock_type_; @@ -224,19 +223,6 @@ NodeOptions::use_intra_process_comms(bool use_intra_process_comms) return *this; } -bool -NodeOptions::enable_topic_statistics() const -{ - return this->enable_topic_statistics_; -} - -NodeOptions & -NodeOptions::enable_topic_statistics(bool enable_topic_statistics) -{ - this->enable_topic_statistics_ = enable_topic_statistics; - return *this; -} - bool NodeOptions::start_parameter_services() const { diff --git a/rclcpp/test/rclcpp/CMakeLists.txt b/rclcpp/test/rclcpp/CMakeLists.txt index 868329f5cd..737bb731d3 100644 --- a/rclcpp/test/rclcpp/CMakeLists.txt +++ b/rclcpp/test/rclcpp/CMakeLists.txt @@ -1,6 +1,7 @@ find_package(ament_cmake_gtest REQUIRED) find_package(rmw_implementation_cmake REQUIRED) +find_package(std_msgs REQUIRED) add_definitions(-DTEST_RESOURCES_DIRECTORY="${TEST_RESOURCES_DIRECTORY}") @@ -323,7 +324,7 @@ if(TARGET test_subscription_publisher_with_same_type_adapter) target_link_libraries(test_subscription_publisher_with_same_type_adapter ${PROJECT_NAME} ${cpp_typesupport_target} - statistics_msgs::statistics_msgs + std_msgs::std_msgs ) endif() @@ -572,17 +573,6 @@ if(TARGET test_wait_set) target_link_libraries(test_wait_set ${PROJECT_NAME} rcl_interfaces::rcl_interfaces test_msgs::test_msgs) endif() -ament_add_ros_isolated_gtest(test_subscription_topic_statistics topic_statistics/test_subscription_topic_statistics.cpp - APPEND_LIBRARY_DIRS "${append_library_dirs}" -) -if(TARGET test_subscription_topic_statistics) - target_link_libraries(test_subscription_topic_statistics - ${PROJECT_NAME} - libstatistics_collector::libstatistics_collector - statistics_msgs::statistics_msgs - test_msgs::test_msgs - ) -endif() ament_add_ros_isolated_gtest(test_subscription_options test_subscription_options.cpp) if(TARGET test_subscription_options) diff --git a/rclcpp/test/rclcpp/test_create_subscription.cpp b/rclcpp/test/rclcpp/test_create_subscription.cpp index b5a5b68e35..4899cb7905 100644 --- a/rclcpp/test/rclcpp/test_create_subscription.cpp +++ b/rclcpp/test/rclcpp/test_create_subscription.cpp @@ -14,14 +14,14 @@ #include -#include #include #include "rclcpp/create_subscription.hpp" #include "rclcpp/node.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" #include "test_msgs/msg/empty.hpp" -using namespace std::chrono_literals; +#include "rmw/types.h" class TestCreateSubscription : public ::testing::Test { @@ -77,13 +77,28 @@ TEST_F(TestCreateSubscription, create_separated_node_topics_and_parameters) { EXPECT_STREQ("/ns/topic_name", subscription->get_topic_name()); } -TEST_F(TestCreateSubscription, create_with_statistics) { +class MockSubscriptionStatisticsMonitor : public rclcpp::SubscriptionStatisticsMonitor +{ +public: + void before_message_dispatch(const rmw_message_info_t &) override + { + before_message_dispatch_count++; + } + void after_message_dispatch(const rmw_message_info_t &) override + { + after_message_dispatch_count++; + } + + size_t before_message_dispatch_count{0}; + size_t after_message_dispatch_count{0}; +}; + +TEST_F(TestCreateSubscription, create_with_monitor) { auto node = std::make_shared("my_node", "/ns"); const rclcpp::QoS qos(10); auto options = rclcpp::SubscriptionOptions(); - options.topic_stats_options.state = rclcpp::TopicStatisticsState::Enable; - options.topic_stats_options.publish_topic = "topic_statistics"; - options.topic_stats_options.publish_period = 5min; + auto monitor = std::make_shared(); + options.subscription_statistics_monitor = monitor; auto callback = [](test_msgs::msg::Empty::ConstSharedPtr) {}; auto subscription = diff --git a/rclcpp/test/rclcpp/test_interface_traits.cpp b/rclcpp/test/rclcpp/test_interface_traits.cpp index 071ac22ce7..bf079adb75 100644 --- a/rclcpp/test/rclcpp/test_interface_traits.cpp +++ b/rclcpp/test/rclcpp/test_interface_traits.cpp @@ -33,8 +33,7 @@ class MyNode "my_node_namespace", rclcpp::contexts::get_global_default_context(), *options.get_rcl_node_options(), - options.use_intra_process_comms(), - options.enable_topic_statistics()); + options.use_intra_process_comms()); } }; diff --git a/rclcpp/test/rclcpp/test_node_options.cpp b/rclcpp/test/rclcpp/test_node_options.cpp index 07a522f497..925b46cfda 100644 --- a/rclcpp/test/rclcpp/test_node_options.cpp +++ b/rclcpp/test/rclcpp/test_node_options.cpp @@ -216,7 +216,6 @@ TEST(TestNodeOptions, copy) { .use_global_arguments(false) .enable_rosout(false) .use_intra_process_comms(true) - .enable_topic_statistics(true) .start_parameter_services(false) .enable_logger_service(true) .start_parameter_event_publisher(false) @@ -234,8 +233,6 @@ TEST(TestNodeOptions, copy) { EXPECT_EQ(non_default_options.enable_rosout(), copied_options.enable_rosout()); EXPECT_EQ(non_default_options.use_intra_process_comms(), copied_options.use_intra_process_comms()); - EXPECT_EQ(non_default_options.enable_topic_statistics(), - copied_options.enable_topic_statistics()); EXPECT_EQ(non_default_options.start_parameter_services(), copied_options.start_parameter_services()); EXPECT_EQ(non_default_options.enable_logger_service(), copied_options.enable_logger_service()); @@ -295,11 +292,6 @@ TEST(TestNodeOptions, bool_setters_and_getters) { options.use_intra_process_comms(true); EXPECT_TRUE(options.use_intra_process_comms()); - options.enable_topic_statistics(false); - EXPECT_FALSE(options.enable_topic_statistics()); - options.enable_topic_statistics(true); - EXPECT_TRUE(options.enable_topic_statistics()); - options.start_parameter_services(false); EXPECT_FALSE(options.start_parameter_services()); options.start_parameter_services(true); diff --git a/rclcpp/test/rclcpp/test_subscription_options.cpp b/rclcpp/test/rclcpp/test_subscription_options.cpp index 1bcde41130..eb57f7ffa8 100644 --- a/rclcpp/test/rclcpp/test_subscription_options.cpp +++ b/rclcpp/test/rclcpp/test_subscription_options.cpp @@ -14,23 +14,16 @@ #include -#include #include #include #include #include "rclcpp/node.hpp" -#include "rclcpp/node_options.hpp" #include "rclcpp/subscription_options.hpp" +#include "rclcpp/subscription_statistics_monitor.hpp" #include "../utils/rclcpp_gtest_macros.hpp" - -using namespace std::chrono_literals; - -namespace -{ -constexpr const char defaultPublishTopic[] = "/statistics"; -} +#include "rmw/types.h" class TestSubscriptionOptions : public ::testing::Test { @@ -59,53 +52,27 @@ class TestSubscriptionOptions : public ::testing::Test rclcpp::Node::SharedPtr node; }; -TEST_F(TestSubscriptionOptions, topic_statistics_options_default_and_set) { - auto options = rclcpp::SubscriptionOptions(); - - EXPECT_EQ(options.topic_stats_options.state, rclcpp::TopicStatisticsState::NodeDefault); - EXPECT_EQ(options.topic_stats_options.publish_topic, defaultPublishTopic); - EXPECT_EQ(options.topic_stats_options.publish_period, 1s); - EXPECT_EQ(options.topic_stats_options.qos, rclcpp::SystemDefaultsQoS().keep_last(10)); - - options.topic_stats_options.state = rclcpp::TopicStatisticsState::Enable; - options.topic_stats_options.publish_topic = "topic_statistics"; - options.topic_stats_options.publish_period = 5min; - options.topic_stats_options.qos = rclcpp::BestAvailableQoS(); - - EXPECT_EQ(options.topic_stats_options.state, rclcpp::TopicStatisticsState::Enable); - EXPECT_EQ(options.topic_stats_options.publish_topic, "topic_statistics"); - EXPECT_EQ(options.topic_stats_options.publish_period, 5min); - EXPECT_EQ(options.topic_stats_options.qos, rclcpp::BestAvailableQoS()); -} - -TEST_F(TestSubscriptionOptions, topic_statistics_options_node_default_mode) { - initialize(); - auto subscription_options = rclcpp::SubscriptionOptions(); - - EXPECT_EQ( - subscription_options.topic_stats_options.state, - rclcpp::TopicStatisticsState::NodeDefault); - EXPECT_FALSE( - rclcpp::detail::resolve_enable_topic_statistics( - subscription_options, - *(node->get_node_base_interface()))); +class MockSubscriptionStatisticsMonitor : public rclcpp::SubscriptionStatisticsMonitor +{ +public: + void before_message_dispatch(const rmw_message_info_t &) override + { + before_message_dispatch_count++; + } + void after_message_dispatch(const rmw_message_info_t &) override + { + after_message_dispatch_count++; + } - initialize(rclcpp::NodeOptions().enable_topic_statistics(true)); - EXPECT_TRUE( - rclcpp::detail::resolve_enable_topic_statistics( - subscription_options, - *(node->get_node_base_interface()))); + size_t before_message_dispatch_count{0}; + size_t after_message_dispatch_count{0}; +}; - subscription_options.topic_stats_options.state = rclcpp::TopicStatisticsState::Disable; - EXPECT_FALSE( - rclcpp::detail::resolve_enable_topic_statistics( - subscription_options, - *(node->get_node_base_interface()))); +TEST_F(TestSubscriptionOptions, subscription_statistics_monitor_default_and_set) { + auto options = rclcpp::SubscriptionOptions(); + EXPECT_EQ(nullptr, options.subscription_statistics_monitor); - subscription_options.topic_stats_options.state = static_cast(5); - RCLCPP_EXPECT_THROW_EQ( - rclcpp::detail::resolve_enable_topic_statistics( - subscription_options, - *(node->get_node_base_interface())), - std::runtime_error("Unrecognized EnableTopicStatistics value")); + auto monitor = std::make_shared(); + options.subscription_statistics_monitor = monitor; + EXPECT_EQ(monitor, options.subscription_statistics_monitor); } diff --git a/rclcpp/test/rclcpp/test_subscription_publisher_with_same_type_adapter.cpp b/rclcpp/test/rclcpp/test_subscription_publisher_with_same_type_adapter.cpp index d733d9758b..c313a4cd2a 100644 --- a/rclcpp/test/rclcpp/test_subscription_publisher_with_same_type_adapter.cpp +++ b/rclcpp/test/rclcpp/test_subscription_publisher_with_same_type_adapter.cpp @@ -26,7 +26,7 @@ #include "rclcpp/rclcpp.hpp" #include "rclcpp/msg/string.hpp" -#include "statistics_msgs/msg/statistic_data_point.hpp" +#include "std_msgs/msg/float64.hpp" static const int g_max_loops = 200; static const std::chrono::milliseconds g_sleep_per_loop(10); @@ -73,18 +73,17 @@ struct TypeAdapter }; template<> -struct TypeAdapter +struct TypeAdapter { using is_specialized = std::true_type; using custom_type = double; - using ros_message_type = statistics_msgs::msg::StatisticDataPoint; + using ros_message_type = std_msgs::msg::Float64; static void convert_to_ros_message( const custom_type & source, ros_message_type & destination) { - destination.data_type = 0; destination.data = source; } @@ -541,7 +540,7 @@ TEST_F( test_intra_process_within_one_node, type_adapted_messages_ros_message_ref_pub_received_by_intra_process_subscription) { - using DoubleTypeAdapter = rclcpp::TypeAdapter; + using DoubleTypeAdapter = rclcpp::TypeAdapter; const double message_data = 0.894; const std::string topic_name = "topic_name"; @@ -562,7 +561,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -582,7 +581,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -602,7 +601,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -623,7 +622,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -643,7 +642,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -664,7 +663,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -684,7 +683,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -706,7 +705,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -726,7 +725,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -748,7 +747,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - statistics_msgs::msg::StatisticDataPoint msg; + std_msgs::msg::Float64 msg; msg.data = message_data; pub->publish(msg); @@ -762,7 +761,7 @@ TEST_F( test_intra_process_within_one_node, type_adapted_messages_unique_ptr_ros_message_pub_received_by_intra_process_subscription) { - using DoubleTypeAdapter = rclcpp::TypeAdapter; + using DoubleTypeAdapter = rclcpp::TypeAdapter; const double message_data = 0.7508; const std::string topic_name = "topic_name"; @@ -773,7 +772,7 @@ TEST_F( auto pub = node->create_publisher(topic_name, 1); { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, - // callback const std::string &, publish with unique statistics_msgs::msg::StatisticDataPoint + // callback const std::string &, publish with unique std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -783,7 +782,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -793,7 +792,7 @@ TEST_F( } { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback const std::string & with message info, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -804,7 +803,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -815,7 +814,7 @@ TEST_F( { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::unique_ptr, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -825,7 +824,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -835,7 +834,7 @@ TEST_F( } { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::unique_ptr with message info, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -847,7 +846,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -858,7 +857,7 @@ TEST_F( { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::shared_ptr, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -868,7 +867,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -878,7 +877,7 @@ TEST_F( } { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::shared_ptr with message info, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -890,7 +889,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -901,7 +900,7 @@ TEST_F( { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::shared_ptr, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -911,7 +910,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -921,7 +920,7 @@ TEST_F( } { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback std::shared_ptr with message info, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -933,7 +932,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -944,7 +943,7 @@ TEST_F( { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback const std::shared_ptr &, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -954,7 +953,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); @@ -964,7 +963,7 @@ TEST_F( } { // create_publisher with TypeAdapt struct, create_subscription with TypeAdapt struct, // callback const std::shared_ptr & with message info, publish with unique - // statistics_msgs::msg::StatisticDataPoint + // std_msgs::msg::Float64 bool is_received = false; auto callback = [message_data, &is_received]( @@ -976,7 +975,7 @@ TEST_F( }; auto sub = node->create_subscription(topic_name, 1, callback); - auto pu_msg = std::make_unique(); + auto pu_msg = std::make_unique(); pu_msg->data = message_data; pub->publish(std::move(pu_msg)); diff --git a/rclcpp/test/rclcpp/topic_statistics/test_subscription_topic_statistics.cpp b/rclcpp/test/rclcpp/topic_statistics/test_subscription_topic_statistics.cpp deleted file mode 100644 index 439fc685e3..0000000000 --- a/rclcpp/test/rclcpp/topic_statistics/test_subscription_topic_statistics.cpp +++ /dev/null @@ -1,480 +0,0 @@ -// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved. -// -// 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 -#include -#include -#include -#include - -#include "libstatistics_collector/moving_average_statistics/types.hpp" - -#include "rclcpp/create_publisher.hpp" -#include "rclcpp/node.hpp" -#include "rclcpp/qos.hpp" -#include "rclcpp/rclcpp.hpp" -#include "rclcpp/subscription_options.hpp" - -#include "rclcpp/topic_statistics/subscription_topic_statistics.hpp" - -#include "statistics_msgs/msg/metrics_message.hpp" -#include "statistics_msgs/msg/statistic_data_type.hpp" - -#include "test_msgs/msg/empty.hpp" -#include "test_msgs/msg/strings.hpp" - -#include "test_topic_stats_utils.hpp" - -namespace -{ -constexpr const std::chrono::seconds defaultStatisticsPublishPeriod{1}; -constexpr const char kTestPubNodeName[]{"test_pub_stats_node"}; -constexpr const char kTestSubNodeName[]{"test_sub_stats_node"}; -constexpr const char kTestSubStatsTopic[]{"/test_sub_stats_topic"}; -constexpr const char kTestSubStatsEmptyTopic[]{"/test_sub_stats_empty_topic"}; -constexpr const char kTestTopicStatisticsTopic[]{"/test_topic_statistics_topic"}; -constexpr const char kMessageAgeSourceLabel[]{"message_age"}; -constexpr const char kMessagePeriodSourceLabel[]{"message_period"}; -constexpr const uint64_t kNoSamples{0}; -constexpr const std::chrono::seconds kTestTimeout{10}; -constexpr const uint64_t kNumExpectedWindows{4}; -constexpr const uint64_t kNumExpectedMessages{kNumExpectedWindows * 2}; -constexpr const uint64_t kNumExpectedMessageAgeMessages{kNumExpectedWindows}; -constexpr const uint64_t kNumExpectedMessagePeriodMessages{kNumExpectedWindows}; -constexpr const std::chrono::seconds kUnstableMessageAgeWindowDuration{ - defaultStatisticsPublishPeriod * (kNumExpectedWindows / 2)}; -// kUnstableMessageAgeWindowDuration can take following value. -// Min: defaultStatisticsPublishPeriod * 2 -// Max: defaultStatisticsPublishPeriod * (kNumExpectedWindows - 2) -constexpr const std::chrono::seconds kUnstableMessageAgeOffset{std::chrono::seconds{1}}; -} // namespace - -using test_msgs::msg::Empty; -using rclcpp::topic_statistics::SubscriptionTopicStatistics; -using statistics_msgs::msg::MetricsMessage; -using statistics_msgs::msg::StatisticDataPoint; -using statistics_msgs::msg::StatisticDataType; -using libstatistics_collector::moving_average_statistics::StatisticData; - -/** - * Wrapper class to test and expose parts of the SubscriptionTopicStatistics class. - */ -class TestSubscriptionTopicStatistics : public SubscriptionTopicStatistics -{ -public: - TestSubscriptionTopicStatistics( - const std::string & node_name, - rclcpp::Publisher::SharedPtr publisher) - : SubscriptionTopicStatistics(node_name, std::move(publisher)) - { - } - - ~TestSubscriptionTopicStatistics() override = default; - - /// Exposed for testing - using SubscriptionTopicStatistics::get_current_collector_data; -}; - -/** - * PublisherNode wrapper: used to create publisher node - */ -template -class PublisherNode : public rclcpp::Node -{ -public: - PublisherNode( - const std::string & name, const std::string & topic, - const std::chrono::milliseconds & publish_period = std::chrono::milliseconds{100}, - bool use_intra_process_comms = false) - : Node(name, rclcpp::NodeOptions().use_intra_process_comms(use_intra_process_comms)) - { - publisher_ = create_publisher(topic, 10); - publish_timer_ = this->create_wall_timer( - publish_period, [this]() { - this->publish_message(); - }); - } - - ~PublisherNode() override = default; - -private: - void publish_message() - { - auto msg = MessageT{}; - publisher_->publish(msg); - } - - typename rclcpp::Publisher::SharedPtr publisher_; - rclcpp::TimerBase::SharedPtr publish_timer_; -}; - -/** - * TransitionMessageStamp publisher emulator node : used to emulate publishing messages by - * directly calling rclcpp::Subscription::handle_message(msg_shared_ptr, message_info). - * The message age results change during the test. - */ -template -class TransitionMessageStampPublisherEmulator : public rclcpp::Node -{ -public: - TransitionMessageStampPublisherEmulator( - const std::string & name, - const std::chrono::seconds transition_duration, const std::chrono::seconds message_age_offset, - typename rclcpp::Subscription::SharedPtr subscription, - const std::chrono::milliseconds & publish_period = std::chrono::milliseconds{100}) - : Node(name), transition_duration_(transition_duration), message_age_offset_(message_age_offset), - subscription_(std::move(subscription)) - { - publish_timer_ = this->create_wall_timer(publish_period, [this]() {this->publish_message();}); - start_time_ = this->now(); - } - -private: - void publish_message() - { - std::shared_ptr msg_shared_ptr = std::make_shared(); - rmw_message_info_t rmw_message_info = rmw_get_zero_initialized_message_info(); - - auto now = this->now(); - auto elapsed_time = now - start_time_; - if (elapsed_time < transition_duration_) { - // Apply only to the topic statistics in the first half - // Subtract offset so message_age is always >= offset. - rmw_message_info.source_timestamp = (now - message_age_offset_).nanoseconds(); - } else { - rmw_message_info.source_timestamp = now.nanoseconds(); - } - rclcpp::MessageInfo message_info{rmw_message_info}; - subscription_->handle_message(msg_shared_ptr, message_info); - } - - std::chrono::seconds transition_duration_; - std::chrono::seconds message_age_offset_; - typename rclcpp::Subscription::SharedPtr subscription_; - rclcpp::Time start_time_; - rclcpp::TimerBase::SharedPtr publish_timer_; -}; - -/** - * Message subscriber node: used to create subscriber with enabled topic statistics collectors - * - */ -template -class SubscriberWithTopicStatistics : public rclcpp::Node -{ -public: - SubscriberWithTopicStatistics( - const std::string & name, const std::string & topic, - std::chrono::milliseconds publish_period = defaultStatisticsPublishPeriod, - bool use_intra_process_comms = false) - : Node(name, rclcpp::NodeOptions().use_intra_process_comms(use_intra_process_comms)) - { - // Manually enable topic statistics via options - auto options = rclcpp::SubscriptionOptions(); - options.topic_stats_options.state = rclcpp::TopicStatisticsState::Enable; - options.topic_stats_options.publish_period = publish_period; - - auto callback = []([[maybe_unused]] typename MessageT::UniquePtr msg) { - // This function is intentionally left empty. - }; - subscription_ = create_subscription>( - topic, - use_intra_process_comms ? rclcpp::QoS(10) : rclcpp::QoS(rclcpp::KeepAll()), - callback, - options); - } - ~SubscriberWithTopicStatistics() override = default; - - typename rclcpp::Subscription::SharedPtr get_subscription() - { - return subscription_; - } - -private: - typename rclcpp::Subscription::SharedPtr subscription_; -}; - -/** - * Test fixture to bring up and teardown rclcpp - */ -class TestSubscriptionTopicStatisticsFixture : public ::testing::Test -{ -protected: - void SetUp() override - { - rclcpp::init(0 /* argc */, nullptr /* argv */); - } - - void TearDown() override - { - rclcpp::shutdown(); - } -}; - -/** - * Check if a received statistics message observed data and contains some calculation - * \param message_to_check - */ -void check_if_statistic_message_is_populated(const MetricsMessage & message_to_check) -{ - for (const auto & stats_point : message_to_check.statistics) { - const auto type = stats_point.data_type; - switch (type) { - case StatisticDataType::STATISTICS_DATA_TYPE_SAMPLE_COUNT: - EXPECT_LT(0, stats_point.data) << "unexpected sample count " << stats_point.data; - break; - case StatisticDataType::STATISTICS_DATA_TYPE_AVERAGE: - EXPECT_LT(0, stats_point.data) << "unexpected avg " << stats_point.data; - break; - case StatisticDataType::STATISTICS_DATA_TYPE_MINIMUM: - EXPECT_LT(0, stats_point.data) << "unexpected min " << stats_point.data; - break; - case StatisticDataType::STATISTICS_DATA_TYPE_MAXIMUM: - EXPECT_LT(0, stats_point.data) << "unexpected max " << stats_point.data; - break; - case StatisticDataType::STATISTICS_DATA_TYPE_STDDEV: - EXPECT_LT(0, stats_point.data) << "unexpected stddev " << stats_point.data; - break; - default: - FAIL() << "received unknown statistics type: " << std::dec << - static_cast(type); - } - } -} - -/** - * Test an invalid argument is thrown for a bad input publish period. - */ -TEST_F(TestSubscriptionTopicStatisticsFixture, test_invalid_publish_period) -{ - ASSERT_THROW( - SubscriberWithTopicStatistics( - "test_period_node", "should_throw_invalid_arg", std::chrono::milliseconds(0) - ), - std::invalid_argument); -} - -/** - * Test that we can manually construct the subscription topic statistics utility class - * without any errors and defaults to empty measurements. - */ -TEST_F(TestSubscriptionTopicStatisticsFixture, test_manual_construction) -{ - auto empty_subscriber = std::make_shared>( - kTestSubNodeName, - kTestSubStatsEmptyTopic); - - // Manually create publisher tied to the node - auto topic_stats_publisher = - empty_subscriber->create_publisher( - kTestTopicStatisticsTopic, - 10); - - // Construct a separate instance - auto sub_topic_stats = std::make_unique( - empty_subscriber->get_name(), - topic_stats_publisher); - - // Expect no data has been collected / no samples received - for (const auto & data : sub_topic_stats->get_current_collector_data()) { - EXPECT_TRUE(std::isnan(data.average)); - EXPECT_TRUE(std::isnan(data.min)); - EXPECT_TRUE(std::isnan(data.max)); - EXPECT_TRUE(std::isnan(data.standard_deviation)); - EXPECT_EQ(kNoSamples, data.sample_count); - } -} - -/** - * Publish messages that do not have a header timestamp, test that all statistics messages - * were received, and verify the statistics message contents. - */ -TEST_F(TestSubscriptionTopicStatisticsFixture, test_receive_stats_for_message_no_header) -{ - // Create an empty publisher - auto empty_publisher = std::make_shared>( - kTestPubNodeName, - kTestSubStatsEmptyTopic); - // empty_subscriber has a topic statistics instance as part of its subscription - // this will listen to and generate statistics for the empty message - - // Create a listener for topic statistics messages - auto statistics_listener = std::make_shared( - "test_receive_single_empty_stats_message_listener", - "/statistics", - kNumExpectedMessages); - - auto empty_subscriber = std::make_shared>( - kTestSubNodeName, - kTestSubStatsEmptyTopic); - - rclcpp::executors::SingleThreadedExecutor ex; - ex.add_node(empty_publisher); - ex.add_node(statistics_listener); - ex.add_node(empty_subscriber); - - // Spin and get future - ex.spin_until_future_complete(statistics_listener->GetFuture(), kTestTimeout); - - // Compare message counts, sample count should be the same as published and received count - EXPECT_EQ(kNumExpectedMessages, statistics_listener->GetNumberOfMessagesReceived()); - - // Check the received message total count - const auto received_messages = statistics_listener->GetReceivedMessages(); - EXPECT_EQ(kNumExpectedMessages, received_messages.size()); - - // check the type of statistics that were received and their counts - uint64_t message_age_count{0}; - uint64_t message_period_count{0}; - - for (const auto & msg : received_messages) { - if (msg.metrics_source == kMessageAgeSourceLabel) { - message_age_count++; - } - if (msg.metrics_source == kMessagePeriodSourceLabel) { - message_period_count++; - } - } - EXPECT_EQ(kNumExpectedMessageAgeMessages, message_age_count); - EXPECT_EQ(kNumExpectedMessagePeriodMessages, message_period_count); - - // Check the collected statistics for message period. - for (const auto & msg : received_messages) { - check_if_statistic_message_is_populated(msg); - } -} - -TEST_F(TestSubscriptionTopicStatisticsFixture, test_receive_stats_include_window_reset) -{ - // msg_subscriber_with_topic_statistics has a topic statistics instance as part of its - // subscription this will listen to and generate statistics - auto msg_subscriber_with_topic_statistics = - std::make_shared>( - kTestSubNodeName, - kTestSubStatsTopic); - - // Create a message publisher - auto msg_publisher = - std::make_shared>( - kTestPubNodeName, kUnstableMessageAgeWindowDuration, - kUnstableMessageAgeOffset, msg_subscriber_with_topic_statistics->get_subscription()); - - // Create a listener for topic statistics messages - auto statistics_listener = std::make_shared( - "test_receive_stats_include_window_reset", "/statistics", kNumExpectedMessages); - - rclcpp::executors::SingleThreadedExecutor ex; - ex.add_node(msg_publisher); - ex.add_node(statistics_listener); - ex.add_node(msg_subscriber_with_topic_statistics); - - // Spin and get future - ex.spin_until_future_complete(statistics_listener->GetFuture(), kTestTimeout); - - const auto received_messages = statistics_listener->GetReceivedMessages(); - EXPECT_EQ(kNumExpectedMessages, received_messages.size()); - - auto message_age_offset = - std::chrono::duration(kUnstableMessageAgeOffset).count(); - - // Check that the first statistic contains the offset inside of its window - auto head_message = received_messages[0]; - for (const auto & stats_point : head_message.statistics) { - const auto type = stats_point.data_type; - switch (type) { - case StatisticDataType::STATISTICS_DATA_TYPE_MINIMUM: - case StatisticDataType::STATISTICS_DATA_TYPE_MAXIMUM: - EXPECT_GE(stats_point.data, message_age_offset); - break; - default: - break; - } - } - - // Check that the last statistic does not contain the offset outside of its window - auto tail_message = received_messages[received_messages.size() - 1]; - for (const auto & stats_point : tail_message.statistics) { - const auto type = stats_point.data_type; - switch (type) { - case StatisticDataType::STATISTICS_DATA_TYPE_MINIMUM: - case StatisticDataType::STATISTICS_DATA_TYPE_MAXIMUM: - EXPECT_LT(stats_point.data, message_age_offset); - break; - default: - break; - } - } -} - -/** - * Test topic statistics are collected when use_intra_process_comms is enabled. - * This validates a fix for ros2/rclcpp#2911 where IPC subscriptions never called the - * stat handler causing all statistics to report NaN values. - * Also verifies message_age is non-NaN, validating that source_timestamp is set correctly. - */ -TEST_F(TestSubscriptionTopicStatisticsFixture, test_stats_with_intra_process_comms) -{ - auto empty_publisher = std::make_shared>( - kTestPubNodeName, - kTestSubStatsEmptyTopic, - std::chrono::milliseconds{100}, - true); - - auto statistics_listener = std::make_shared( - "test_ipc_stats_listener", - "/statistics", - kNumExpectedMessages); - - auto empty_subscriber = std::make_shared>( - kTestSubNodeName, - kTestSubStatsEmptyTopic, - defaultStatisticsPublishPeriod, - true); - - rclcpp::executors::SingleThreadedExecutor ex; - ex.add_node(empty_publisher); - ex.add_node(statistics_listener); - ex.add_node(empty_subscriber); - - ex.spin_until_future_complete(statistics_listener->GetFuture(), kTestTimeout); - - const auto received_messages = statistics_listener->GetReceivedMessages(); - EXPECT_EQ(kNumExpectedMessages, received_messages.size()); - - uint64_t message_age_count{0}; - uint64_t message_period_count{0}; - - for (const auto & msg : received_messages) { - if (msg.metrics_source == kMessageAgeSourceLabel) { - message_age_count++; - // Verify message_age stats are non-NaN to validates source_timestamp fix - for (const auto & stats_point : msg.statistics) { - EXPECT_FALSE(std::isnan(stats_point.data)); - } - } - if (msg.metrics_source == kMessagePeriodSourceLabel) { - message_period_count++; - } - } - - EXPECT_EQ(kNumExpectedMessageAgeMessages, message_age_count); - EXPECT_EQ(kNumExpectedMessagePeriodMessages, message_period_count); -} diff --git a/rclcpp/test/rclcpp/topic_statistics/test_topic_stats_utils.hpp b/rclcpp/test/rclcpp/topic_statistics/test_topic_stats_utils.hpp deleted file mode 100644 index 8d5be3da0f..0000000000 --- a/rclcpp/test/rclcpp/topic_statistics/test_topic_stats_utils.hpp +++ /dev/null @@ -1,153 +0,0 @@ -// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved. -// -// 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 -#include - -#include "statistics_msgs/msg/metrics_message.hpp" -#include "rclcpp/node.hpp" - -#ifndef RCLCPP__TOPIC_STATISTICS__TEST_TOPIC_STATS_UTILS_HPP_ -#define RCLCPP__TOPIC_STATISTICS__TEST_TOPIC_STATS_UTILS_HPP_ - -namespace rclcpp -{ -namespace topic_statistics -{ - -using statistics_msgs::msg::MetricsMessage; - -/** -* Provide an interface to wait for a promise to be satisfied via its future. -*/ -class PromiseSetter -{ -public: - /** - * Reassign the promise member and return it's future. Acquires a mutex in order - * to mutate member variables. - * - * \return the promise member's future, called upon PeriodicMeasurement - */ - std::shared_future GetFuture() - { - std::unique_lock ulock{mutex_}; - use_future_ = true; - promise_ = std::promise(); - return promise_.get_future(); - } - -protected: - /** - * Set the promise to true, which signals the corresponding future. Acquires a mutex and sets - * the promise to true iff GetFuture was invoked before this. - */ - void SetPromise() - { - std::unique_lock ulock{mutex_}; - if (use_future_) { - // only set if GetFuture was called - promise_.set_value(true); - use_future_ = false; // the promise needs to be reassigned to set again - } - } - -private: - mutable std::mutex mutex_; - std::promise promise_; - bool use_future_{false}; -}; - -/** - * Node which listens for published MetricsMessages. This uses the PromiseSetter API - * in order to signal, via a future, that rclcpp should stop spinning upon - * message handling. - */ -class MetricsMessageSubscriber : public rclcpp::Node, public PromiseSetter -{ -public: - /** - * Constructs a MetricsMessageSubscriber. - * \param name the node name - * \param name the topic name - * \param number of messages to receive to trigger the PromiseSetter future - */ - MetricsMessageSubscriber( - const std::string & name, - const std::string & topic_name, - const uint64_t number_of_messages_to_receive = 2) - : rclcpp::Node(name), - number_of_messages_to_receive_(number_of_messages_to_receive) - { - auto callback = [this](MetricsMessage::UniquePtr msg) { - this->MetricsMessageCallback(*msg); - }; - subscription_ = create_subscription>( - topic_name, - 10 /*history_depth*/, - callback); - } - - /** - * Acquires a mutex in order to get the last message received member. - * \return the last message received - */ - std::vector GetReceivedMessages() const - { - std::unique_lock ulock{mutex_}; - return received_messages_; - } - - /** - * Return the number of messages received by this subscriber. - * \return the number of messages received by the subscriber callback - */ - uint64_t GetNumberOfMessagesReceived() const - { - return num_messages_received_; - } - -private: - /** - * Subscriber callback. Acquires a mutex to set the last message received and - * sets the promise to true. - * \param msg - */ - void MetricsMessageCallback(const MetricsMessage & msg) - { - std::unique_lock ulock{mutex_}; - ++num_messages_received_; - received_messages_.push_back(msg); - if (num_messages_received_ >= number_of_messages_to_receive_) { - PromiseSetter::SetPromise(); - } - } - - std::vector received_messages_; - rclcpp::Subscription::SharedPtr subscription_; - mutable std::mutex mutex_; - std::atomic num_messages_received_{0}; - const uint64_t number_of_messages_to_receive_; -}; - -} // namespace topic_statistics -} // namespace rclcpp - -#endif // RCLCPP__TOPIC_STATISTICS__TEST_TOPIC_STATS_UTILS_HPP_