27.0.0

Signed-off-by: Chris Lalancette <clalancette@gmail.com>

rclcpp/CHANGELOG.rst

@@ -2,6 +2,142 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+27.0.0 (2024-02-07)
+-------------------
+* Split test_executors up into smaller chunks. (`#2421 <https://github.com/ros2/rclcpp/issues/2421>`_)
+* [events executor] - Fix Behavior with Timer Cancel (`#2375 <https://github.com/ros2/rclcpp/issues/2375>`_)
+* Removed deprecated header (`#2413 <https://github.com/ros2/rclcpp/issues/2413>`_)
+* Make sure to mark RingBuffer methods as 'override'. (`#2410 <https://github.com/ros2/rclcpp/issues/2410>`_)
+* Contributors: Alejandro Hernández Cordero, Chris Lalancette, Matt Condino
+
+26.0.0 (2024-01-24)
+-------------------
+* Increase the cppcheck timeout to 600 seconds. (`#2409 <https://github.com/ros2/rclcpp/issues/2409>`_)
+* Add transient local durability support to publisher and subscriptions when using intra-process communication (`#2303 <https://github.com/ros2/rclcpp/issues/2303>`_)
+* Stop storing the context in the guard condition. (`#2400 <https://github.com/ros2/rclcpp/issues/2400>`_)
+* Contributors: Chris Lalancette, Jeffery Hsu
+
+25.0.0 (2023-12-26)
+-------------------
+* Updated GenericSubscription to AnySubscriptionCallback (`#1928 <https://github.com/ros2/rclcpp/issues/1928>`_)
+* make type support helper supported for service (`#2209 <https://github.com/ros2/rclcpp/issues/2209>`_)
+* Adding QoS to subscription options (`#2323 <https://github.com/ros2/rclcpp/issues/2323>`_)
+* Switch to target_link_libraries. (`#2374 <https://github.com/ros2/rclcpp/issues/2374>`_)
+* aligh with rcl that a rosout publisher of a node might not exist (`#2357 <https://github.com/ros2/rclcpp/issues/2357>`_)
+* Fix data race in EventHandlerBase (`#2349 <https://github.com/ros2/rclcpp/issues/2349>`_)
+* Support users holding onto shared pointers in the message memory pool (`#2336 <https://github.com/ros2/rclcpp/issues/2336>`_)
+* Contributors: Chen Lihui, Chris Lalancette, DensoADAS, Lucas Wendland, mauropasse
+
+24.0.0 (2023-11-06)
+-------------------
+* fix (signal_handler.hpp): spelling (`#2356 <https://github.com/ros2/rclcpp/issues/2356>`_)
+* Updates to not use std::move in some places. (`#2353 <https://github.com/ros2/rclcpp/issues/2353>`_)
+* rclcpp::Time::max() clock type support. (`#2352 <https://github.com/ros2/rclcpp/issues/2352>`_)
+* Serialized Messages with Topic Statistics (`#2274 <https://github.com/ros2/rclcpp/issues/2274>`_)
+* Add a custom deleter when constructing rcl_service_t (`#2351 <https://github.com/ros2/rclcpp/issues/2351>`_)
+* Disable the loaned messages inside the executor. (`#2335 <https://github.com/ros2/rclcpp/issues/2335>`_)
+* Use message_info in SubscriptionTopicStatistics instead of typed message (`#2337 <https://github.com/ros2/rclcpp/issues/2337>`_)
+* Add missing 'enable_rosout' comments (`#2345 <https://github.com/ros2/rclcpp/issues/2345>`_)
+* Adjust rclcpp usage of type description service (`#2344 <https://github.com/ros2/rclcpp/issues/2344>`_)
+* address rate related flaky tests. (`#2329 <https://github.com/ros2/rclcpp/issues/2329>`_)
+* Fixes pointed out by the clang analyzer. (`#2339 <https://github.com/ros2/rclcpp/issues/2339>`_)
+* Remove useless ROSRate class (`#2326 <https://github.com/ros2/rclcpp/issues/2326>`_)
+* Contributors: Alexey Merzlyakov, Chris Lalancette, Jiaqi Li, Lucas Wendland, Michael Carroll, Michael Orlov, Tomoya Fujita, Zard-C
+
+23.2.0 (2023-10-09)
+-------------------
+* add clients & services count (`#2072 <https://github.com/ros2/rclcpp/issues/2072>`_)
+* remove invalid sized allocation test for SerializedMessage. (`#2330 <https://github.com/ros2/rclcpp/issues/2330>`_)
+* Adding API to copy all parameters from one node to another (`#2304 <https://github.com/ros2/rclcpp/issues/2304>`_)
+* Contributors: Minju, Lee, Steve Macenski, Tomoya Fujita
+
+23.1.0 (2023-10-04)
+-------------------
+* Add locking to protect the TimeSource::NodeState::node_base\_ (`#2320 <https://github.com/ros2/rclcpp/issues/2320>`_)
+* Update SignalHandler get_global_signal_handler to avoid complex types in static memory (`#2316 <https://github.com/ros2/rclcpp/issues/2316>`_)
+* Removing Old Connext Tests (`#2313 <https://github.com/ros2/rclcpp/issues/2313>`_)
+* Documentation for list_parameters  (`#2315 <https://github.com/ros2/rclcpp/issues/2315>`_)
+* Decouple rosout publisher init from node init. (`#2174 <https://github.com/ros2/rclcpp/issues/2174>`_)
+* fix the depth to relative in list_parameters (`#2300 <https://github.com/ros2/rclcpp/issues/2300>`_)
+* Contributors: Chris Lalancette, Lucas Wendland, Minju, Lee, Tomoya Fujita, Tully Foote
+
+23.0.0 (2023-09-08)
+-------------------
+* Fix the return type of Rate::period. (`#2301 <https://github.com/ros2/rclcpp/issues/2301>`_)
+* Update API docs links in package READMEs (`#2302 <https://github.com/ros2/rclcpp/issues/2302>`_)
+* Cleanup flaky timers_manager tests. (`#2299 <https://github.com/ros2/rclcpp/issues/2299>`_)
+* Contributors: Chris Lalancette, Christophe Bedard
+
+22.2.0 (2023-09-07)
+-------------------
+* Topic correct typeadapter deduction (`#2294 <https://github.com/ros2/rclcpp/issues/2294>`_)
+* Fix C++20 allocator construct deprecation (`#2292 <https://github.com/ros2/rclcpp/issues/2292>`_)
+* Make Rate to select the clock to work with (`#2123 <https://github.com/ros2/rclcpp/issues/2123>`_)
+* Correct the position of a comment. (`#2290 <https://github.com/ros2/rclcpp/issues/2290>`_)
+* Remove unnecessary lambda captures in the tests. (`#2289 <https://github.com/ros2/rclcpp/issues/2289>`_)
+* Add rcl_logging_interface as an explicit dependency. (`#2284 <https://github.com/ros2/rclcpp/issues/2284>`_)
+* Revamp list_parameters to be more efficient and easier to read. (`#2282 <https://github.com/ros2/rclcpp/issues/2282>`_)
+* Contributors: AiVerisimilitude, Alexey Merzlyakov, Chen Lihui, Chris Lalancette, Jiaqi Li
+
+22.1.0 (2023-08-21)
+-------------------
+* Do not crash Executor when send_response fails due to client failure. (`#2276 <https://github.com/ros2/rclcpp/issues/2276>`_)
+* Adding Custom Unknown Type Error (`#2272 <https://github.com/ros2/rclcpp/issues/2272>`_)
+* Add a pimpl inside rclcpp::Node for future distro backports (`#2228 <https://github.com/ros2/rclcpp/issues/2228>`_)
+* Remove an unused variable from the events executor tests. (`#2270 <https://github.com/ros2/rclcpp/issues/2270>`_)
+* Add spin_all shortcut (`#2246 <https://github.com/ros2/rclcpp/issues/2246>`_)
+* Adding Missing Group Exceptions (`#2256 <https://github.com/ros2/rclcpp/issues/2256>`_)
+* Change associated clocks storage to unordered_set (`#2257 <https://github.com/ros2/rclcpp/issues/2257>`_)
+* associated clocks should be protected by mutex. (`#2255 <https://github.com/ros2/rclcpp/issues/2255>`_)
+* Instrument loaned message publication code path (`#2240 <https://github.com/ros2/rclcpp/issues/2240>`_)
+* Contributors: Chris Lalancette, Christophe Bedard, Emerson Knapp, Luca Della Vedova, Lucas Wendland, Tomoya Fujita, Tony Najjar
+
+22.0.0 (2023-07-11)
+-------------------
+* Implement get_node_type_descriptions_interface for lifecyclenode and add smoke test for it (`#2237 <https://github.com/ros2/rclcpp/issues/2237>`_)
+* Add new node interface TypeDescriptionsInterface to provide GetTypeDescription service (`#2224 <https://github.com/ros2/rclcpp/issues/2224>`_)
+* Move always_false_v to detail namespace (`#2232 <https://github.com/ros2/rclcpp/issues/2232>`_)
+* Revamp the test_subscription.cpp tests. (`#2227 <https://github.com/ros2/rclcpp/issues/2227>`_)
+* warning: comparison of integer expressions of different signedness (`#2219 <https://github.com/ros2/rclcpp/issues/2219>`_)
+* Modifies timers API to select autostart state (`#2005 <https://github.com/ros2/rclcpp/issues/2005>`_)
+* Enable callback group tests for connextdds (`#2182 <https://github.com/ros2/rclcpp/issues/2182>`_)
+* Contributors: Chris Lalancette, Christopher Wecht, Eloy Briceno, Emerson Knapp, Nathan Wiebe Neufeldt, Tomoya Fujita
+
+21.3.0 (2023-06-12)
+-------------------
+* Fix up misspellings of "receive". (`#2208 <https://github.com/ros2/rclcpp/issues/2208>`_)
+* Remove flaky stressAddRemoveNode test (`#2206 <https://github.com/ros2/rclcpp/issues/2206>`_)
+* Use TRACETOOLS\_ prefix for tracepoint-related macros (`#2162 <https://github.com/ros2/rclcpp/issues/2162>`_)
+* Contributors: Chris Lalancette, Christophe Bedard, Michael Carroll
+
+21.2.0 (2023-06-07)
+-------------------
+* remove nolint since ament_cpplint updated for the c++17 header (`#2198 <https://github.com/ros2/rclcpp/issues/2198>`_)
+* Feature/available capacity of ipm (`#2173 <https://github.com/ros2/rclcpp/issues/2173>`_)
+* add mutex to protect events_executor current entity collection (`#2187 <https://github.com/ros2/rclcpp/issues/2187>`_)
+* Declare rclcpp callbacks before the rcl entities (`#2024 <https://github.com/ros2/rclcpp/issues/2024>`_)
+* Contributors: Alberto Soragna, Chen Lihui, DensoADAS, mauropasse
+
+21.1.1 (2023-05-11)
+-------------------
+* Fix race condition in events-executor (`#2177 <https://github.com/ros2/rclcpp/issues/2177>`_)
+* Add missing stdexcept include (`#2186 <https://github.com/ros2/rclcpp/issues/2186>`_)
+* Fix a format-security warning when building with clang (`#2171 <https://github.com/ros2/rclcpp/issues/2171>`_)
+* Fix delivered message kind (`#2175 <https://github.com/ros2/rclcpp/issues/2175>`_)
+* Contributors: Alberto Soragna, Chris Lalancette, methylDragon, Øystein Sture
+
+21.1.0 (2023-04-27)
+-------------------
+
+21.0.0 (2023-04-18)
+-------------------
+* Add support for logging service. (`#2122 <https://github.com/ros2/rclcpp/issues/2122>`_)
+* Picking ABI-incompatible executor changes (`#2170 <https://github.com/ros2/rclcpp/issues/2170>`_)
+* add events-executor and timers-manager in rclcpp (`#2155 <https://github.com/ros2/rclcpp/issues/2155>`_)
+* Create common structures for executors to use (`#2143 <https://github.com/ros2/rclcpp/issues/2143>`_)
+* Implement deliver message kind (`#2168 <https://github.com/ros2/rclcpp/issues/2168>`_)
+* Contributors: Alberto Soragna, Lei Liu, Michael Carroll, methylDragon
+
 20.0.0 (2023-04-13)
 -------------------
 * applied tracepoints for ring_buffer (`#2091 <https://github.com/ros2/rclcpp/issues/2091>`_)

rclcpp/CMakeLists.txt

@@ -10,11 +10,14 @@ 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)
 find_package(rcl_yaml_param_parser REQUIRED)
 find_package(rcpputils REQUIRED)
 find_package(rcutils REQUIRED)
 find_package(rmw REQUIRED)
 find_package(rosgraph_msgs REQUIRED)
+find_package(rosidl_dynamic_typesupport REQUIRED)
+find_package(rosidl_runtime_c REQUIRED)
 find_package(rosidl_runtime_cpp REQUIRED)
 find_package(rosidl_typesupport_c REQUIRED)
 find_package(rosidl_typesupport_cpp REQUIRED)
@@ -43,6 +46,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/context.cpp
   src/rclcpp/contexts/default_context.cpp
   src/rclcpp/detail/add_guard_condition_to_rcl_wait_set.cpp
+  src/rclcpp/detail/resolve_intra_process_buffer_type.cpp
   src/rclcpp/detail/resolve_parameter_overrides.cpp
   src/rclcpp/detail/rmw_implementation_specific_payload.cpp
   src/rclcpp/detail/rmw_implementation_specific_publisher_payload.cpp
@@ -64,6 +68,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/executors/executor_notify_waitable.cpp
   src/rclcpp/executors/multi_threaded_executor.cpp
   src/rclcpp/executors/single_threaded_executor.cpp
+  src/rclcpp/executors/static_executor_entities_collector.cpp
   src/rclcpp/executors/static_single_threaded_executor.cpp
   src/rclcpp/expand_topic_or_service_name.cpp
   src/rclcpp/experimental/executors/events_executor/events_executor.cpp
@@ -91,6 +96,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/node_interfaces/node_time_source.cpp
   src/rclcpp/node_interfaces/node_timers.cpp
   src/rclcpp/node_interfaces/node_topics.cpp
+  src/rclcpp/node_interfaces/node_type_descriptions.cpp
   src/rclcpp/node_interfaces/node_waitables.cpp
   src/rclcpp/node_options.cpp
   src/rclcpp/parameter.cpp
@@ -104,6 +110,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/qos.cpp
   src/rclcpp/event_handler.cpp
   src/rclcpp/qos_overriding_options.cpp
+  src/rclcpp/rate.cpp
   src/rclcpp/serialization.cpp
   src/rclcpp/serialized_message.cpp
   src/rclcpp/service.cpp
@@ -198,22 +205,28 @@ target_include_directories(${PROJECT_NAME} PUBLIC
   "$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>"
   "$<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}/include>"
   "$<INSTALL_INTERFACE:include/${PROJECT_NAME}>")
-target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})
-# specific order: dependents before dependencies
-ament_target_dependencies(${PROJECT_NAME}
-  "ament_index_cpp"
-  "libstatistics_collector"
-  "rcl"
-  "rcl_interfaces"
-  "rcl_yaml_param_parser"
-  "rcpputils"
-  "rcutils"
-  "builtin_interfaces"
-  "rosgraph_msgs"
-  "rosidl_typesupport_cpp"
-  "rosidl_runtime_cpp"
-  "statistics_msgs"
-  "tracetools"
+target_link_libraries(${PROJECT_NAME} PUBLIC
+  ${builtin_interfaces_TARGETS}
+  libstatistics_collector::libstatistics_collector
+  rcl::rcl
+  ${rcl_interfaces_TARGETS}
+  rcl_yaml_param_parser::rcl_yaml_param_parser
+  rcpputils::rcpputils
+  rcutils::rcutils
+  rmw::rmw
+  ${rosgraph_msgs_TARGETS}
+  rosidl_dynamic_typesupport::rosidl_dynamic_typesupport
+  rosidl_runtime_c::rosidl_runtime_c
+  rosidl_runtime_cpp::rosidl_runtime_cpp
+  rosidl_typesupport_cpp::rosidl_typesupport_cpp
+  ${statistics_msgs_TARGETS}
+  tracetools::tracetools
+  ${CMAKE_THREAD_LIBS_INIT}
+)
+
+target_link_libraries(${PROJECT_NAME} PRIVATE
+  ament_index_cpp::ament_index_cpp
+  rcl_logging_interface::rcl_logging_interface
 )
 
 # Causes the visibility macros to use dllexport rather than dllimport,
@@ -235,20 +248,23 @@ ament_export_libraries(${PROJECT_NAME})
 # Export modern CMake targets
 ament_export_targets(${PROJECT_NAME})
 
-# specific order: dependents before dependencies
-ament_export_dependencies(ament_index_cpp)
-ament_export_dependencies(libstatistics_collector)
-ament_export_dependencies(rcl)
-ament_export_dependencies(rcpputils)
-ament_export_dependencies(rcutils)
-ament_export_dependencies(builtin_interfaces)
-ament_export_dependencies(rosgraph_msgs)
-ament_export_dependencies(rosidl_typesupport_cpp)
-ament_export_dependencies(rosidl_typesupport_c)
-ament_export_dependencies(rosidl_runtime_cpp)
-ament_export_dependencies(rcl_yaml_param_parser)
-ament_export_dependencies(statistics_msgs)
-ament_export_dependencies(tracetools)
+ament_export_dependencies(
+  builtin_interfaces
+  libstatistics_collector
+  rcl
+  rcl_interfaces
+  rcl_yaml_param_parser
+  rcpputils
+  rcutils
+  rmw
+  rosgraph_msgs
+  rosidl_dynamic_typesupport
+  rosidl_runtime_c
+  rosidl_runtime_cpp
+  rosidl_typesupport_cpp
+  statistics_msgs
+  tracetools
+)
 
 if(BUILD_TESTING)
   find_package(ament_lint_auto REQUIRED)
@@ -266,7 +282,7 @@ install(
 
 if(TEST cppcheck)
   # must set the property after ament_package()
-  set_tests_properties(cppcheck PROPERTIES TIMEOUT 500)
+  set_tests_properties(cppcheck PROPERTIES TIMEOUT 600)
 endif()
 
 if(TEST cpplint)

rclcpp/README.md

@@ -2,7 +2,7 @@
 
 The ROS client library in C++.
 
-Visit the [rclcpp API documentation](http://docs.ros2.org/latest/api/rclcpp/) for a complete list of its main components and features.
+The link to the latest rclcpp API documentation, which includes a complete list of its main components and features, can be found on the rclcpp package info page, at the [ROS Index](https://index.ros.org/p/rclcpp/).
 
 ## Quality Declaration
 

rclcpp/include/rclcpp/any_service_callback.hpp

@@ -156,7 +156,7 @@ public:
     const std::shared_ptr<rmw_request_id_t> & request_header,
     std::shared_ptr<typename ServiceT::Request> request)
   {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
     if (std::holds_alternative<std::monostate>(callback_)) {
       // TODO(ivanpauno): Remove the set method, and force the users of this class
       // to pass a callback at construnciton.
@@ -182,7 +182,7 @@ public:
       const auto & cb = std::get<SharedPtrWithRequestHeaderCallback>(callback_);
       cb(request_header, std::move(request), response);
     }
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
     return response;
   }
 
@@ -191,9 +191,9 @@ public:
 #ifndef TRACETOOLS_DISABLED
     std::visit(
       [this](auto && arg) {
-        if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+        if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
           char * symbol = tracetools::get_symbol(arg);
-          DO_TRACEPOINT(
+          TRACETOOLS_DO_TRACEPOINT(
             rclcpp_callback_register,
             static_cast<const void *>(this),
             symbol);

rclcpp/include/rclcpp/any_subscription_callback.hpp

@@ -20,7 +20,7 @@
 #include <stdexcept>
 #include <type_traits>
 #include <utility>
-#include <variant>  // NOLINT[build/include_order]
+#include <variant>
 
 #include "rosidl_runtime_cpp/traits.hpp"
 #include "tracetools/tracetools.h"
@@ -30,19 +30,20 @@
 #include "rclcpp/detail/subscription_callback_type_helper.hpp"
 #include "rclcpp/function_traits.hpp"
 #include "rclcpp/message_info.hpp"
+#include "rclcpp/serialization.hpp"
 #include "rclcpp/serialized_message.hpp"
 #include "rclcpp/type_adapter.hpp"
 
 
-template<class>
-inline constexpr bool always_false_v = false;
-
 namespace rclcpp
 {
 
 namespace detail
 {
 
+template<class>
+inline constexpr bool always_false_v = false;
+
 template<typename MessageT, typename AllocatorT>
 struct MessageDeleterHelper
 {
@@ -158,13 +159,14 @@ struct AnySubscriptionCallbackPossibleTypes
 template<
   typename MessageT,
   typename AllocatorT,
-  bool is_adapted_type = rclcpp::TypeAdapter<MessageT>::is_specialized::value
+  bool is_adapted_type = rclcpp::TypeAdapter<MessageT>::is_specialized::value,
+  bool is_serialized_type = serialization_traits::is_serialized_message_class<MessageT>::value
 >
 struct AnySubscriptionCallbackHelper;
 
 /// Specialization for when MessageT is not a TypeAdapter.
 template<typename MessageT, typename AllocatorT>
-struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false, false>
 {
   using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;
 
@@ -194,7 +196,7 @@ struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false>
 
 /// Specialization for when MessageT is a TypeAdapter.
 template<typename MessageT, typename AllocatorT>
-struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true, false>
 {
   using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;
 
@@ -232,6 +234,26 @@ struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true>
   >;
 };
 
+/// Specialization for when MessageT is a SerializedMessage to exclude duplicated declarations.
+template<typename MessageT, typename AllocatorT>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false, true>
+{
+  using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;
+
+  using variant_type = std::variant<
+    typename CallbackTypes::ConstRefSerializedMessageCallback,
+    typename CallbackTypes::ConstRefSerializedMessageWithInfoCallback,
+    typename CallbackTypes::UniquePtrSerializedMessageCallback,
+    typename CallbackTypes::UniquePtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::SharedConstPtrSerializedMessageCallback,
+    typename CallbackTypes::SharedConstPtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::ConstRefSharedConstPtrSerializedMessageCallback,
+    typename CallbackTypes::ConstRefSharedConstPtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::SharedPtrSerializedMessageCallback,
+    typename CallbackTypes::SharedPtrSerializedMessageWithInfoCallback
+  >;
+};
+
 }  // namespace detail
 
 template<
@@ -487,12 +509,14 @@ public:
   }
 
   // Dispatch when input is a ros message and the output could be anything.
-  void
+  template<typename TMsg = ROSMessageType>
+  typename std::enable_if<!serialization_traits::is_serialized_message_class<TMsg>::value,
+    void>::type
   dispatch(
     std::shared_ptr<ROSMessageType> message,
     const rclcpp::MessageInfo & message_info)
   {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
     // Check if the variant is "unset", throw if it is.
     if (callback_variant_.index() == 0) {
       if (std::get<0>(callback_variant_) == nullptr) {
@@ -580,19 +604,19 @@ public:
         }
         // condition to catch unhandled callback types
         else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
         }
       }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
   }
 
   // Dispatch when input is a serialized message and the output could be anything.
   void
   dispatch(
-    std::shared_ptr<rclcpp::SerializedMessage> serialized_message,
+    std::shared_ptr<const rclcpp::SerializedMessage> serialized_message,
     const rclcpp::MessageInfo & message_info)
   {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
     // Check if the variant is "unset", throw if it is.
     if (callback_variant_.index() == 0) {
       if (std::get<0>(callback_variant_) == nullptr) {
@@ -660,10 +684,10 @@ public:
         }
         // condition to catch unhandled callback types
         else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
         }
       }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
   }
 
   void
@@ -671,7 +695,7 @@ public:
     std::shared_ptr<const SubscribedType> message,
     const rclcpp::MessageInfo & message_info)
   {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), true);
     // Check if the variant is "unset", throw if it is.
     if (callback_variant_.index() == 0) {
       if (std::get<0>(callback_variant_) == nullptr) {
@@ -790,10 +814,10 @@ public:
         }
         // condition to catch unhandled callback types
         else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
         }
       }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
   }
 
   void
@@ -801,7 +825,7 @@ public:
     std::unique_ptr<SubscribedType, SubscribedTypeDeleter> message,
     const rclcpp::MessageInfo & message_info)
   {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), true);
     // Check if the variant is "unset", throw if it is.
     if (callback_variant_.index() == 0) {
       if (std::get<0>(callback_variant_) == nullptr) {
@@ -924,10 +948,10 @@ public:
         }
         // condition to catch unhandled callback types
         else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
         }
       }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
   }
 
   constexpr
@@ -965,9 +989,9 @@ public:
 #ifndef TRACETOOLS_DISABLED
     std::visit(
       [this](auto && callback) {
-        if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+        if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
           char * symbol = tracetools::get_symbol(callback);
-          DO_TRACEPOINT(
+          TRACETOOLS_DO_TRACEPOINT(
             rclcpp_callback_register,
             static_cast<const void *>(this),
             symbol);

rclcpp/include/rclcpp/callback_group.hpp

@@ -129,8 +129,7 @@ public:
    * added to the executor in either case.
    *
    * \param[in] group_type The type of the callback group.
-   * \param[in] get_node_context Lambda to retrieve the node context when
-   *   checking that the creating node is valid and using the guard condition.
+   * \param[in] context A weak pointer to the context associated with this callback group.
    * \param[in] automatically_add_to_executor_with_node A boolean that
    *   determines whether a callback group is automatically added to an executor
    *   with the node with which it is associated.
@@ -138,7 +137,7 @@ public:
   RCLCPP_PUBLIC
   explicit CallbackGroup(
     CallbackGroupType group_type,
-    std::function<rclcpp::Context::SharedPtr(void)> get_node_context,
+    rclcpp::Context::WeakPtr context,
     bool automatically_add_to_executor_with_node = true);
 
   /// Default destructor.
@@ -187,35 +186,14 @@ public:
   RCLCPP_PUBLIC
   size_t size() const;
 
-  /// Return a reference to the 'can be taken' atomic boolean.
-  /**
-   * The resulting bool will be true in the case that no executor is currently
-   * using an executable entity from this group.
-   * The resulting bool will be false in the case that an executor is currently
-   * using an executable entity from this group, and the group policy doesn't
-   * allow a second take (eg mutual exclusion)
-   * \return a reference to the flag
-   */
   RCLCPP_PUBLIC
   std::atomic_bool &
   can_be_taken_from();
 
-  /// Get the group type.
-  /**
-   * \return the group type
-   */
   RCLCPP_PUBLIC
   const CallbackGroupType &
   type() const;
 
-  /// Collect all of the entity pointers contained in this callback group.
-  /**
-   * \param[in] sub_func Function to execute for each subscription
-   * \param[in] service_func Function to execute for each service
-   * \param[in] client_func Function to execute for each client
-   * \param[in] timer_func Function to execute for each timer
-   * \param[in] waitable_fuinc Function to execute for each waitable
-   */
   RCLCPP_PUBLIC
   void collect_all_ptrs(
     std::function<void(const rclcpp::SubscriptionBase::SharedPtr &)> sub_func,
@@ -249,16 +227,6 @@ public:
   bool
   automatically_add_to_executor_with_node() const;
 
-  /// Retrieve the guard condition used to signal changes to this callback group.
-  /**
-   * \param[in] context_ptr context to use when creating the guard condition
-   * \return guard condition if it is valid, otherwise nullptr.
-   */
-  [[deprecated("Use get_notify_guard_condition() without arguments")]]
-  RCLCPP_PUBLIC
-  rclcpp::GuardCondition::SharedPtr
-  get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr);
-
   /// Retrieve the guard condition used to signal changes to this callback group.
   /**
    * \return guard condition if it is valid, otherwise nullptr.
@@ -318,7 +286,7 @@ protected:
   std::shared_ptr<rclcpp::GuardCondition> notify_guard_condition_ = nullptr;
   std::recursive_mutex notify_guard_condition_mutex_;
 
-  std::function<rclcpp::Context::SharedPtr(void)> get_context_;
+  rclcpp::Context::WeakPtr context_;
 
 private:
   template<typename TypeT, typename Function>

rclcpp/include/rclcpp/client.hpp

@@ -20,13 +20,13 @@
 #include <future>
 #include <memory>
 #include <mutex>
-#include <optional>  // NOLINT, cpplint doesn't think this is a cpp std header
+#include <optional>
 #include <sstream>
 #include <string>
 #include <tuple>
 #include <unordered_map>
 #include <utility>
-#include <variant>  // NOLINT
+#include <variant>
 #include <vector>
 
 #include "rcl/client.h"
@@ -363,12 +363,16 @@ protected:
   std::shared_ptr<rclcpp::Context> context_;
   rclcpp::Logger node_logger_;
 
+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_response_callback_ before
+  // client_handle_, so on destruction the client is
+  // destroyed first. Otherwise, the rmw client callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_response_callback_{nullptr};
+  // Declare client_handle_ after callback
   std::shared_ptr<rcl_client_t> client_handle_;
 
   std::atomic<bool> in_use_by_wait_set_{false};
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_response_callback_{nullptr};
 };
 
 template<typename ServiceT>

rclcpp/include/rclcpp/context.hpp

@@ -26,6 +26,7 @@
 #include <unordered_set>
 #include <utility>
 #include <vector>
+#include <stdexcept>
 
 #include "rcl/context.h"
 #include "rcl/guard_condition.h"

rclcpp/include/rclcpp/copy_all_parameter_values.hpp

@@ -0,0 +1,82 @@
+// Copyright 2023 Open Navigation LLC
+//
+// 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__COPY_ALL_PARAMETER_VALUES_HPP_
+#define RCLCPP__COPY_ALL_PARAMETER_VALUES_HPP_
+
+#include <string>
+#include <vector>
+
+#include "rcl_interfaces/srv/list_parameters.hpp"
+#include "rcl_interfaces/msg/parameter_descriptor.hpp"
+#include "rcl_interfaces/msg/set_parameters_result.hpp"
+
+#include "rclcpp/parameter.hpp"
+#include "rclcpp/logger.hpp"
+#include "rclcpp/logging.hpp"
+
+namespace rclcpp
+{
+
+/**
+ * Copy all parameters from one source node to another destination node.
+ * May throw exceptions if parameters from source are uninitialized or undeclared.
+ * \param source Node to copy parameters from
+ * \param destination Node to copy parameters to
+ * \param override_existing_params Default false. Whether to override existing destination params
+ * if both the source and destination contain the same parameter.
+ */
+template<typename NodeT1, typename NodeT2>
+void
+copy_all_parameter_values(
+  const NodeT1 & source, const NodeT2 & destination, const bool override_existing_params = false)
+{
+  using Parameters = std::vector<rclcpp::Parameter>;
+  using Descriptions = std::vector<rcl_interfaces::msg::ParameterDescriptor>;
+  auto source_params = source->get_node_parameters_interface();
+  auto dest_params = destination->get_node_parameters_interface();
+  rclcpp::Logger logger = destination->get_node_logging_interface()->get_logger();
+
+  std::vector<std::string> param_names = source_params->list_parameters({}, 0).names;
+  Parameters params = source_params->get_parameters(param_names);
+  Descriptions descriptions = source_params->describe_parameters(param_names);
+
+  for (unsigned int idx = 0; idx != params.size(); idx++) {
+    if (!dest_params->has_parameter(params[idx].get_name())) {
+      dest_params->declare_parameter(
+        params[idx].get_name(), params[idx].get_parameter_value(), descriptions[idx]);
+    } else if (override_existing_params) {
+      try {
+        rcl_interfaces::msg::SetParametersResult result =
+          dest_params->set_parameters_atomically({params[idx]});
+        if (!result.successful) {
+          // Parameter update rejected or read-only
+          RCLCPP_WARN(
+            logger,
+            "Unable to set parameter (%s): %s!",
+            params[idx].get_name().c_str(), result.reason.c_str());
+        }
+      } catch (const rclcpp::exceptions::InvalidParameterTypeException & e) {
+        RCLCPP_WARN(
+          logger,
+          "Unable to set parameter (%s): incompatable parameter type (%s)!",
+          params[idx].get_name().c_str(), e.what());
+      }
+    }
+  }
+}
+
+}  // namespace rclcpp
+
+#endif  // RCLCPP__COPY_ALL_PARAMETER_VALUES_HPP_

rclcpp/include/rclcpp/create_generic_subscription.hpp

@@ -45,13 +45,15 @@ namespace rclcpp
  * Not all publisher options are currently respected, the only relevant options for this
  * publisher are `event_callbacks`, `use_default_callbacks`, and `%callback_group`.
  */
-template<typename AllocatorT = std::allocator<void>>
+template<
+  typename CallbackT,
+  typename AllocatorT = std::allocator<void>>
 std::shared_ptr<GenericSubscription> create_generic_subscription(
   rclcpp::node_interfaces::NodeTopicsInterface::SharedPtr topics_interface,
   const std::string & topic_name,
   const std::string & topic_type,
   const rclcpp::QoS & qos,
-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+  CallbackT && callback,
   const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options = (
     rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>()
   )
@@ -60,13 +62,20 @@ std::shared_ptr<GenericSubscription> create_generic_subscription(
   auto ts_lib = rclcpp::get_typesupport_library(
     topic_type, "rosidl_typesupport_cpp");
 
+  auto allocator = options.get_allocator();
+
+  using rclcpp::AnySubscriptionCallback;
+  AnySubscriptionCallback<rclcpp::SerializedMessage, AllocatorT>
+  any_subscription_callback(*allocator);
+  any_subscription_callback.set(std::forward<CallbackT>(callback));
+
   auto subscription = std::make_shared<GenericSubscription>(
     topics_interface->get_node_base_interface(),
     std::move(ts_lib),
     topic_name,
     topic_type,
     qos,
-    callback,
+    any_subscription_callback,
     options);
 
   topics_interface->add_subscription(subscription, options.callback_group);

rclcpp/include/rclcpp/create_subscription.hpp

@@ -50,8 +50,8 @@ template<
   typename SubscriptionT,
   typename MessageMemoryStrategyT,
   typename NodeParametersT,
-  typename NodeTopicsT,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType>
+  typename NodeTopicsT
+>
 typename std::shared_ptr<SubscriptionT>
 create_subscription(
   NodeParametersT & node_parameters,
@@ -70,7 +70,7 @@ create_subscription(
   using rclcpp::node_interfaces::get_node_topics_interface;
   auto node_topics_interface = get_node_topics_interface(node_topics);
 
-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
   subscription_topic_stats = nullptr;
 
   if (rclcpp::detail::resolve_enable_topic_statistics(
@@ -80,8 +80,7 @@ create_subscription(
     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::to_string(options.topic_stats_options.publish_period.count()) + " ms");
     }
 
     std::shared_ptr<Publisher<statistics_msgs::msg::MetricsMessage>>
@@ -89,14 +88,14 @@ create_subscription(
       node_parameters,
       node_topics_interface,
       options.topic_stats_options.publish_topic,
-      qos);
+      options.topic_stats_options.qos);
 
-    subscription_topic_stats = std::make_shared<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
-      >(node_topics_interface->get_node_base_interface()->get_name(), publisher);
+    subscription_topic_stats =
+      std::make_shared<rclcpp::topic_statistics::SubscriptionTopicStatistics>(
+      node_topics_interface->get_node_base_interface()->get_name(), publisher);
 
     std::weak_ptr<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
+      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();

rclcpp/include/rclcpp/create_timer.hpp

@@ -90,7 +90,8 @@ create_timer(
   rclcpp::Clock::SharedPtr clock,
   rclcpp::Duration period,
   CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::CallbackGroup::SharedPtr group = nullptr,
+  bool autostart = true)
 {
   return create_timer(
     clock,
@@ -98,7 +99,8 @@ create_timer(
     std::forward<CallbackT>(callback),
     group,
     node_base.get(),
-    node_timers.get());
+    node_timers.get(),
+    autostart);
 }
 
 /// Create a timer with a given clock
@@ -109,7 +111,8 @@ create_timer(
   rclcpp::Clock::SharedPtr clock,
   rclcpp::Duration period,
   CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::CallbackGroup::SharedPtr group = nullptr,
+  bool autostart = true)
 {
   return create_timer(
     clock,
@@ -117,7 +120,8 @@ create_timer(
     std::forward<CallbackT>(callback),
     group,
     rclcpp::node_interfaces::get_node_base_interface(node).get(),
-    rclcpp::node_interfaces::get_node_timers_interface(node).get());
+    rclcpp::node_interfaces::get_node_timers_interface(node).get(),
+    autostart);
 }
 
 /// Convenience method to create a general timer with node resources.
@@ -132,6 +136,7 @@ create_timer(
  * \param group callback group
  * \param node_base node base interface
  * \param node_timers node timer interface
+ * \param autostart defines if the timer should start it's countdown on initialization or not.
  * \return shared pointer to a generic timer
  * \throws std::invalid_argument if either clock, node_base or node_timers
  * are nullptr, or period is negative or too large
@@ -144,7 +149,8 @@ create_timer(
   CallbackT callback,
   rclcpp::CallbackGroup::SharedPtr group,
   node_interfaces::NodeBaseInterface * node_base,
-  node_interfaces::NodeTimersInterface * node_timers)
+  node_interfaces::NodeTimersInterface * node_timers,
+  bool autostart = true)
 {
   if (clock == nullptr) {
     throw std::invalid_argument{"clock cannot be null"};
@@ -160,7 +166,7 @@ create_timer(
 
   // Add a new generic timer.
   auto timer = rclcpp::GenericTimer<CallbackT>::make_shared(
-    std::move(clock), period_ns, std::move(callback), node_base->get_context());
+    std::move(clock), period_ns, std::move(callback), node_base->get_context(), autostart);
   node_timers->add_timer(timer, group);
   return timer;
 }
@@ -187,7 +193,8 @@ create_wall_timer(
   CallbackT callback,
   rclcpp::CallbackGroup::SharedPtr group,
   node_interfaces::NodeBaseInterface * node_base,
-  node_interfaces::NodeTimersInterface * node_timers)
+  node_interfaces::NodeTimersInterface * node_timers,
+  bool autostart = true)
 {
   if (node_base == nullptr) {
     throw std::invalid_argument{"input node_base cannot be null"};
@@ -201,7 +208,7 @@ create_wall_timer(
 
   // Add a new wall timer.
   auto timer = rclcpp::WallTimer<CallbackT>::make_shared(
-    period_ns, std::move(callback), node_base->get_context());
+    period_ns, std::move(callback), node_base->get_context(), autostart);
   node_timers->add_timer(timer, group);
   return timer;
 }

rclcpp/include/rclcpp/detail/resolve_intra_process_buffer_type.hpp

@@ -47,6 +47,11 @@ resolve_intra_process_buffer_type(
   return resolved_buffer_type;
 }
 
+RCLCPP_PUBLIC
+rclcpp::IntraProcessBufferType
+resolve_intra_process_buffer_type(
+  const rclcpp::IntraProcessBufferType buffer_type);
+
 }  // namespace detail
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/event_handler.hpp

@@ -260,6 +260,16 @@ public:
     }
   }
 
+  ~EventHandler()
+  {
+    // Since the rmw event listener holds a reference to the
+    // "on ready" callback, we need to clear it on destruction of this class.
+    // This clearing is not needed for other rclcpp entities like pub/subs, since
+    // they do own the underlying rmw entities, which are destroyed
+    // on their rclcpp destructors, thus no risk of dangling pointers.
+    clear_on_ready_callback();
+  }
+
   /// Take data so that the callback cannot be scheduled again
   std::shared_ptr<void>
   take_data() override

rclcpp/include/rclcpp/exceptions/exceptions.hpp

@@ -206,6 +206,14 @@ public:
   const std::vector<std::string> unknown_ros_args;
 };
 
+/// Thrown when an unknown type is passed
+class UnknownTypeError : public std::runtime_error
+{
+public:
+  explicit UnknownTypeError(const std::string & type)
+  : std::runtime_error("Unknown type: " + type) {}
+};
+
 /// Thrown when an invalid rclcpp::Event object or SharedPtr is encountered.
 class InvalidEventError : public std::runtime_error
 {
@@ -222,6 +230,14 @@ public:
   : std::runtime_error("event already registered") {}
 };
 
+/// Thrown when a callback group is missing from the node, when it wants to utilize the group.
+class MissingGroupNodeException : public std::runtime_error
+{
+public:
+  explicit MissingGroupNodeException(const std::string & obj_type)
+  : std::runtime_error("cannot create: " + obj_type + " , callback group not in node") {}
+};
+
 /// Thrown if passed parameters are inconsistent or invalid
 class InvalidParametersException : public std::runtime_error
 {

rclcpp/include/rclcpp/executor.hpp

@@ -19,7 +19,6 @@
 #include <cassert>
 #include <chrono>
 #include <cstdlib>
-#include <deque>
 #include <iostream>
 #include <list>
 #include <map>
@@ -30,24 +29,26 @@
 
 #include "rcl/guard_condition.h"
 #include "rcl/wait.h"
-#include "rclcpp/executors/executor_notify_waitable.hpp"
 #include "rcpputils/scope_exit.hpp"
 
 #include "rclcpp/context.hpp"
 #include "rclcpp/contexts/default_context.hpp"
 #include "rclcpp/guard_condition.hpp"
 #include "rclcpp/executor_options.hpp"
-#include "rclcpp/executors/executor_entities_collection.hpp"
-#include "rclcpp/executors/executor_entities_collector.hpp"
 #include "rclcpp/future_return_code.hpp"
+#include "rclcpp/memory_strategies.hpp"
+#include "rclcpp/memory_strategy.hpp"
 #include "rclcpp/node_interfaces/node_base_interface.hpp"
 #include "rclcpp/utilities.hpp"
 #include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_set.hpp"
 
 namespace rclcpp
 {
 
+typedef std::map<rclcpp::CallbackGroup::WeakPtr,
+    rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
+    std::owner_less<rclcpp::CallbackGroup::WeakPtr>> WeakCallbackGroupsToNodesMap;
+
 // Forward declaration is used in convenience method signature.
 class Node;
 class ExecutorImplementation;
@@ -296,6 +297,21 @@ public:
   virtual void
   spin_some(std::chrono::nanoseconds max_duration = std::chrono::nanoseconds(0));
 
+  /// Add a node, complete all immediately available work exhaustively, and remove the node.
+  /**
+   * \param[in] node Shared pointer to the node to add.
+   */
+  RCLCPP_PUBLIC
+  void
+  spin_node_all(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node,
+    std::chrono::nanoseconds max_duration);
+
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  RCLCPP_PUBLIC
+  void
+  spin_node_all(std::shared_ptr<rclcpp::Node> node, std::chrono::nanoseconds max_duration);
+
   /// Collect and execute work repeatedly within a duration or until no more work is available.
   /**
    * This function can be overridden. The default implementation is suitable for a
@@ -402,6 +418,17 @@ public:
   void
   cancel();
 
+  /// Support dynamic switching of the memory strategy.
+  /**
+   * Switching the memory strategy while the executor is spinning in another threading could have
+   * unintended consequences.
+   * \param[in] memory_strategy Shared pointer to the memory strategy to set.
+   * \throws std::runtime_error if memory_strategy is null
+   */
+  RCLCPP_PUBLIC
+  void
+  set_memory_strategy(memory_strategy::MemoryStrategy::SharedPtr memory_strategy);
+
   /// Returns true if the executor is currently spinning.
   /**
    * This function can be called asynchronously from any thread.
@@ -486,11 +513,6 @@ protected:
   static void
   execute_client(rclcpp::ClientBase::SharedPtr client);
 
-  /// Gather all of the waitable entities from associated nodes and callback groups.
-  RCLCPP_PUBLIC
-  void
-  collect_entities();
-
   /// Block until more work becomes avilable or timeout is reached.
   /**
    * Builds a set of waitable entities, which are passed to the middleware.
@@ -502,6 +524,62 @@ protected:
   void
   wait_for_work(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
 
+  /// Find node associated with a callback group
+  /**
+   * \param[in] weak_groups_to_nodes map of callback groups to nodes
+   * \param[in] group callback group to find assocatiated node
+   * \return Pointer to associated node if found, else nullptr
+   */
+  RCLCPP_PUBLIC
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
+  get_node_by_group(
+    const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes,
+    rclcpp::CallbackGroup::SharedPtr group);
+
+  /// Return true if the node has been added to this executor.
+  /**
+   * \param[in] node_ptr a shared pointer that points to a node base interface
+   * \param[in] weak_groups_to_nodes map to nodes to lookup
+   * \return true if the node is associated with the executor, otherwise false
+   */
+  RCLCPP_PUBLIC
+  bool
+  has_node(
+    const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes) const;
+
+  /// Find the callback group associated with a timer
+  /**
+   * \param[in] timer Timer to find associated callback group
+   * \return Pointer to callback group node if found, else nullptr
+   */
+  RCLCPP_PUBLIC
+  rclcpp::CallbackGroup::SharedPtr
+  get_group_by_timer(rclcpp::TimerBase::SharedPtr timer);
+
+  /// Add a callback group to an executor
+  /**
+   * \see rclcpp::Executor::add_callback_group
+   */
+  RCLCPP_PUBLIC
+  virtual void
+  add_callback_group_to_map(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    WeakCallbackGroupsToNodesMap & weak_groups_to_nodes,
+    bool notify = true) RCPPUTILS_TSA_REQUIRES(mutex_);
+
+  /// Remove a callback group from the executor.
+  /**
+   * \see rclcpp::Executor::remove_callback_group
+   */
+  RCLCPP_PUBLIC
+  virtual void
+  remove_callback_group_from_map(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    WeakCallbackGroupsToNodesMap & weak_groups_to_nodes,
+    bool notify = true) RCPPUTILS_TSA_REQUIRES(mutex_);
+
   /// Check for executable in ready state and populate union structure.
   /**
    * \param[out] any_executable populated union structure of ready executable
@@ -512,6 +590,33 @@ protected:
   bool
   get_next_ready_executable(AnyExecutable & any_executable);
 
+  /// Check for executable in ready state and populate union structure.
+  /**
+   * This is the implementation of get_next_ready_executable that takes into
+   * account the current state of callback groups' association with nodes and
+   * executors.
+   *
+   * This checks in a particular order for available work:
+   *   * Timers
+   *   * Subscriptions
+   *   * Services
+   *   * Clients
+   *   * Waitable
+   *
+   * If the next executable is not associated with this executor/node pair,
+   * then this method will return false.
+   *
+   * \param[out] any_executable populated union structure of ready executable
+   * \param[in] weak_groups_to_nodes mapping of callback groups to nodes
+   * \return true if an executable was ready and any_executable was populated,
+   *   otherwise false
+   */
+  RCLCPP_PUBLIC
+  bool
+  get_next_ready_executable_from_map(
+    AnyExecutable & any_executable,
+    const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
+
   /// Wait for executable in ready state and populate union structure.
   /**
    * If an executable is ready, it will return immediately, otherwise
@@ -529,6 +634,21 @@ protected:
     AnyExecutable & any_executable,
     std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
 
+  /// Add all callback groups that can be automatically added from associated nodes.
+  /**
+   * The executor, before collecting entities, verifies if any callback group from
+   * nodes associated with the executor, which is not already associated to an executor,
+   * can be automatically added to this executor.
+   * This takes care of any callback group that has been added to a node but not explicitly added
+   * to the executor.
+   * It is important to note that in order for the callback groups to be automatically added to an
+   * executor through this function, the node of the callback groups needs to have been added
+   * through the `add_node` method.
+   */
+  RCLCPP_PUBLIC
+  virtual void
+  add_callback_groups_from_nodes_associated_to_executor() RCPPUTILS_TSA_REQUIRES(mutex_);
+
   /// Spinning state, used to prevent multi threaded calls to spin and to cancel blocking spins.
   std::atomic_bool spinning;
 
@@ -538,8 +658,16 @@ protected:
   /// Guard condition for signaling the rmw layer to wake up for system shutdown.
   std::shared_ptr<rclcpp::GuardCondition> shutdown_guard_condition_;
 
+  /// Wait set for managing entities that the rmw layer waits on.
+  rcl_wait_set_t wait_set_ = rcl_get_zero_initialized_wait_set();
+
+  // Mutex to protect the subsequent memory_strategy_.
   mutable std::mutex mutex_;
 
+  /// The memory strategy: an interface for handling user-defined memory allocation strategies.
+  memory_strategy::MemoryStrategy::SharedPtr
+  memory_strategy_ RCPPUTILS_TSA_PT_GUARDED_BY(mutex_);
+
   /// The context associated with this executor.
   std::shared_ptr<rclcpp::Context> context_;
 
@@ -549,33 +677,39 @@ protected:
   virtual void
   spin_once_impl(std::chrono::nanoseconds timeout);
 
-  /// Waitable containing guard conditions controlling the executor flow.
-  /**
-   * This waitable contains the interrupt and shutdown guard condition, as well
-   * as the guard condition associated with each node and callback group.
-   * By default, if any change is detected in the monitored entities, the notify
-   * waitable will awake the executor and rebuild  the collections.
-   */
-  std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> notify_waitable_;
+  typedef std::map<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
+      const rclcpp::GuardCondition *,
+      std::owner_less<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>>
+    WeakNodesToGuardConditionsMap;
 
-  std::atomic_bool entities_need_rebuild_;
+  typedef std::map<rclcpp::CallbackGroup::WeakPtr,
+      const rclcpp::GuardCondition *,
+      std::owner_less<rclcpp::CallbackGroup::WeakPtr>>
+    WeakCallbackGroupsToGuardConditionsMap;
 
-  /// Collector used to associate executable entities from nodes and guard conditions
-  rclcpp::executors::ExecutorEntitiesCollector collector_;
+  /// maps nodes to guard conditions
+  WeakNodesToGuardConditionsMap
+  weak_nodes_to_guard_conditions_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
 
-  /// Waitset to be waited on.
-  rclcpp::WaitSet wait_set_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
+  /// maps callback groups to guard conditions
+  WeakCallbackGroupsToGuardConditionsMap
+  weak_groups_to_guard_conditions_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
 
-  /// Hold the current state of the collection being waited on by the waitset
-  rclcpp::executors::ExecutorEntitiesCollection current_collection_ RCPPUTILS_TSA_GUARDED_BY(
-    mutex_);
+  /// maps callback groups associated to nodes
+  WeakCallbackGroupsToNodesMap
+  weak_groups_associated_with_executor_to_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
 
-  /// Hold the current state of the notify waitable being waited on by the waitset
-  std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> current_notify_waitable_
-  RCPPUTILS_TSA_GUARDED_BY(mutex_);
+  /// maps callback groups to nodes associated with executor
+  WeakCallbackGroupsToNodesMap
+  weak_groups_to_nodes_associated_with_executor_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
 
-  /// Hold the list of executables currently available to be executed.
-  std::deque<rclcpp::AnyExecutable> ready_executables_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
+  /// maps all callback groups to nodes
+  WeakCallbackGroupsToNodesMap
+  weak_groups_to_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
+
+  /// nodes that are associated with the executor
+  std::list<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>
+  weak_nodes_ RCPPUTILS_TSA_GUARDED_BY(mutex_);
 
   /// shutdown callback handle registered to Context
   rclcpp::OnShutdownCallbackHandle shutdown_callback_handle_;

rclcpp/include/rclcpp/executors.hpp

@@ -29,6 +29,18 @@
 namespace rclcpp
 {
 
+/// Create a default single-threaded executor and execute all available work exhaustively.
+/** \param[in] node_ptr Shared pointer to the node to spin. */
+RCLCPP_PUBLIC
+void
+spin_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  std::chrono::nanoseconds max_duration);
+
+RCLCPP_PUBLIC
+void
+spin_all(rclcpp::Node::SharedPtr node_ptr, std::chrono::nanoseconds max_duration);
+
 /// Create a default single-threaded executor and execute any immediately available work.
 /** \param[in] node_ptr Shared pointer to the node to spin. */
 RCLCPP_PUBLIC

rclcpp/include/rclcpp/executors/static_executor_entities_collector.hpp

@@ -0,0 +1,357 @@
+// Copyright 2020 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__EXECUTORS__STATIC_EXECUTOR_ENTITIES_COLLECTOR_HPP_
+#define RCLCPP__EXECUTORS__STATIC_EXECUTOR_ENTITIES_COLLECTOR_HPP_
+
+#include <chrono>
+#include <list>
+#include <map>
+#include <memory>
+#include <vector>
+
+#include "rcl/guard_condition.h"
+#include "rcl/wait.h"
+
+#include "rclcpp/experimental/executable_list.hpp"
+#include "rclcpp/macros.hpp"
+#include "rclcpp/memory_strategy.hpp"
+#include "rclcpp/visibility_control.hpp"
+#include "rclcpp/waitable.hpp"
+
+namespace rclcpp
+{
+namespace executors
+{
+typedef std::map<rclcpp::CallbackGroup::WeakPtr,
+    rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
+    std::owner_less<rclcpp::CallbackGroup::WeakPtr>> WeakCallbackGroupsToNodesMap;
+
+class StaticExecutorEntitiesCollector final
+  : public rclcpp::Waitable,
+  public std::enable_shared_from_this<StaticExecutorEntitiesCollector>
+{
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(StaticExecutorEntitiesCollector)
+
+  // Constructor
+  RCLCPP_PUBLIC
+  StaticExecutorEntitiesCollector() = default;
+
+  // Destructor
+  RCLCPP_PUBLIC
+  ~StaticExecutorEntitiesCollector();
+
+  /// Initialize StaticExecutorEntitiesCollector
+  /**
+   * \param p_wait_set A reference to the wait set to be used in the executor
+   * \param memory_strategy Shared pointer to the memory strategy to set.
+   * \throws std::runtime_error if memory strategy is null
+   */
+  RCLCPP_PUBLIC
+  void
+  init(
+    rcl_wait_set_t * p_wait_set,
+    rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy);
+
+  /// Finalize StaticExecutorEntitiesCollector to clear resources
+  RCLCPP_PUBLIC
+  bool
+  is_init() {return initialized_;}
+
+  RCLCPP_PUBLIC
+  void
+  fini();
+
+  /// Execute the waitable.
+  RCLCPP_PUBLIC
+  void
+  execute(std::shared_ptr<void> & data) override;
+
+  /// Take the data so that it can be consumed with `execute`.
+  /**
+   * For `StaticExecutorEntitiesCollector`, this always return `nullptr`.
+   * \sa rclcpp::Waitable::take_data()
+   */
+  RCLCPP_PUBLIC
+  std::shared_ptr<void>
+  take_data() override;
+
+  /// Function to add_handles_to_wait_set and wait for work and
+  /**
+   * block until the wait set is ready or until the timeout has been exceeded.
+   * \throws std::runtime_error if wait set couldn't be cleared or filled.
+   * \throws any rcl errors from rcl_wait, \see rclcpp::exceptions::throw_from_rcl_error()
+   */
+  RCLCPP_PUBLIC
+  void
+  refresh_wait_set(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
+
+  /**
+   * \throws std::runtime_error if it couldn't add guard condition to wait set
+   */
+  RCLCPP_PUBLIC
+  void
+  add_to_wait_set(rcl_wait_set_t * wait_set) override;
+
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_ready_guard_conditions() override;
+
+  /// Add a callback group to an executor.
+  /**
+   * \see rclcpp::Executor::add_callback_group
+   */
+  RCLCPP_PUBLIC
+  bool
+  add_callback_group(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
+
+  /// Add a callback group to an executor.
+  /**
+   * \see rclcpp::Executor::add_callback_group
+   * \return boolean whether the node from the callback group is new
+   */
+  RCLCPP_PUBLIC
+  bool
+  add_callback_group(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
+
+  /// Remove a callback group from the executor.
+  /**
+   * \see rclcpp::Executor::remove_callback_group
+   */
+  RCLCPP_PUBLIC
+  bool
+  remove_callback_group(
+    rclcpp::CallbackGroup::SharedPtr group_ptr);
+
+  /// Remove a callback group from the executor.
+  /**
+   * \see rclcpp::Executor::remove_callback_group_from_map
+   */
+  RCLCPP_PUBLIC
+  bool
+  remove_callback_group_from_map(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
+
+  /**
+   * \see rclcpp::Executor::add_node()
+   * \throw std::runtime_error if node was already added
+   */
+  RCLCPP_PUBLIC
+  bool
+  add_node(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
+
+  /**
+   * \see rclcpp::Executor::remove_node()
+   * \throw std::runtime_error if no guard condition is associated with node.
+   */
+  RCLCPP_PUBLIC
+  bool
+  remove_node(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr);
+
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_all_callback_groups();
+
+  /// Get callback groups that belong to executor.
+  /**
+   * \see rclcpp::Executor::get_manually_added_callback_groups()
+   */
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_manually_added_callback_groups();
+
+  /// Get callback groups that belong to executor.
+  /**
+   * \see rclcpp::Executor::get_automatically_added_callback_groups_from_nodes()
+   */
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_automatically_added_callback_groups_from_nodes();
+
+  /// Complete all available queued work without blocking.
+  /**
+   * This function checks if after the guard condition was triggered
+   * (or a spurious wakeup happened) we are really ready to execute
+   * i.e. re-collect entities
+   */
+  RCLCPP_PUBLIC
+  bool
+  is_ready(rcl_wait_set_t * wait_set) override;
+
+  /// Return number of timers
+  /**
+   * \return number of timers
+   */
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_timers() {return exec_list_.number_of_timers;}
+
+  /// Return number of subscriptions
+  /**
+   * \return number of subscriptions
+   */
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_subscriptions() {return exec_list_.number_of_subscriptions;}
+
+  /// Return number of services
+  /**
+   * \return number of services
+   */
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_services() {return exec_list_.number_of_services;}
+
+  /// Return number of clients
+  /**
+   * \return number of clients
+   */
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_clients() {return exec_list_.number_of_clients;}
+
+  /// Return number of waitables
+  /**
+   * \return number of waitables
+   */
+  RCLCPP_PUBLIC
+  size_t
+  get_number_of_waitables() {return exec_list_.number_of_waitables;}
+
+  /** Return a SubscritionBase Sharedptr by index.
+   * \param[in] i The index of the SubscritionBase
+   * \return a SubscritionBase shared pointer
+   * \throws std::out_of_range if the argument is higher than the size of the structrue.
+   */
+  RCLCPP_PUBLIC
+  rclcpp::SubscriptionBase::SharedPtr
+  get_subscription(size_t i) {return exec_list_.subscription[i];}
+
+  /** Return a TimerBase Sharedptr by index.
+   * \param[in] i The index of the TimerBase
+   * \return a TimerBase shared pointer
+   * \throws std::out_of_range if the argument is higher than the size.
+   */
+  RCLCPP_PUBLIC
+  rclcpp::TimerBase::SharedPtr
+  get_timer(size_t i) {return exec_list_.timer[i];}
+
+  /** Return a ServiceBase Sharedptr by index.
+   * \param[in] i The index of the ServiceBase
+   * \return a ServiceBase shared pointer
+   * \throws std::out_of_range if the argument is higher than the size.
+   */
+  RCLCPP_PUBLIC
+  rclcpp::ServiceBase::SharedPtr
+  get_service(size_t i) {return exec_list_.service[i];}
+
+  /** Return a ClientBase Sharedptr by index
+   * \param[in] i The index of the ClientBase
+   * \return a ClientBase shared pointer
+   * \throws std::out_of_range if the argument is higher than the size.
+   */
+  RCLCPP_PUBLIC
+  rclcpp::ClientBase::SharedPtr
+  get_client(size_t i) {return exec_list_.client[i];}
+
+  /** Return a Waitable Sharedptr by index
+   * \param[in] i The index of the Waitable
+   * \return a Waitable shared pointer
+   * \throws std::out_of_range if the argument is higher than the size.
+   */
+  RCLCPP_PUBLIC
+  rclcpp::Waitable::SharedPtr
+  get_waitable(size_t i) {return exec_list_.waitable[i];}
+
+private:
+  /// Function to reallocate space for entities in the wait set.
+  /**
+   * \throws std::runtime_error if wait set couldn't be cleared or resized.
+   */
+  void
+  prepare_wait_set();
+
+  void
+  fill_executable_list();
+
+  void
+  fill_memory_strategy();
+
+  /// Return true if the node belongs to the collector
+  /**
+   * \param[in] node_ptr a node base interface shared pointer
+   * \param[in] weak_groups_to_nodes map to nodes to lookup
+   * \return boolean whether a node belongs the collector
+   */
+  bool
+  has_node(
+    const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes) const;
+
+  /// Add all callback groups that can be automatically added by any executor
+  /// and is not already associated with an executor from nodes
+  /// that are associated with executor
+  /**
+   * \see rclcpp::Executor::add_callback_groups_from_nodes_associated_to_executor()
+   */
+  void
+  add_callback_groups_from_nodes_associated_to_executor();
+
+  void
+  fill_executable_list_from_map(const WeakCallbackGroupsToNodesMap & weak_groups_to_nodes);
+
+  /// Memory strategy: an interface for handling user-defined memory allocation strategies.
+  rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy_;
+
+  // maps callback groups to nodes.
+  WeakCallbackGroupsToNodesMap weak_groups_associated_with_executor_to_nodes_;
+  // maps callback groups to nodes.
+  WeakCallbackGroupsToNodesMap weak_groups_to_nodes_associated_with_executor_;
+
+  typedef std::map<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
+      const rclcpp::GuardCondition *,
+      std::owner_less<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr>>
+    WeakNodesToGuardConditionsMap;
+  WeakNodesToGuardConditionsMap weak_nodes_to_guard_conditions_;
+
+  /// List of weak nodes registered in the static executor
+  std::list<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> weak_nodes_;
+
+  // Mutex to protect vector of new nodes.
+  std::mutex new_nodes_mutex_;
+  std::vector<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> new_nodes_;
+
+  /// Wait set for managing entities that the rmw layer waits on.
+  rcl_wait_set_t * p_wait_set_ = nullptr;
+
+  /// Executable list: timers, subscribers, clients, services and waitables
+  rclcpp::experimental::ExecutableList exec_list_;
+
+  /// Bool to check if the entities collector has been initialized
+  bool initialized_ = false;
+};
+
+}  // namespace executors
+}  // namespace rclcpp
+
+#endif  // RCLCPP__EXECUTORS__STATIC_EXECUTOR_ENTITIES_COLLECTOR_HPP_

rclcpp/include/rclcpp/executors/static_single_threaded_executor.hpp

@@ -15,13 +15,24 @@
 #ifndef RCLCPP__EXECUTORS__STATIC_SINGLE_THREADED_EXECUTOR_HPP_
 #define RCLCPP__EXECUTORS__STATIC_SINGLE_THREADED_EXECUTOR_HPP_
 
-#include <atomic>
 #include <chrono>
+#include <cassert>
+#include <cstdlib>
 #include <memory>
+#include <vector>
+#include <string>
+
+#include "rmw/rmw.h"
 
 #include "rclcpp/executor.hpp"
-#include "rclcpp/executors/executor_entities_collection.hpp"
-#include "rclcpp/executors/single_threaded_executor.hpp"
+#include "rclcpp/executors/static_executor_entities_collector.hpp"
+#include "rclcpp/experimental/executable_list.hpp"
+#include "rclcpp/macros.hpp"
+#include "rclcpp/memory_strategies.hpp"
+#include "rclcpp/node.hpp"
+#include "rclcpp/rate.hpp"
+#include "rclcpp/utilities.hpp"
+#include "rclcpp/visibility_control.hpp"
 
 namespace rclcpp
 {
@@ -54,7 +65,7 @@ public:
   explicit StaticSingleThreadedExecutor(
     const rclcpp::ExecutorOptions & options = rclcpp::ExecutorOptions());
 
-  /// Default destructor.
+  /// Default destrcutor.
   RCLCPP_PUBLIC
   virtual ~StaticSingleThreadedExecutor();
 
@@ -105,20 +116,92 @@ public:
   void
   spin_all(std::chrono::nanoseconds max_duration) override;
 
+  /// Add a callback group to an executor.
+  /**
+   * \sa rclcpp::Executor::add_callback_group
+   */
+  RCLCPP_PUBLIC
+  void
+  add_callback_group(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    bool notify = true) override;
+
+  /// Remove callback group from the executor
+  /**
+   * \sa rclcpp::Executor::remove_callback_group
+   */
+  RCLCPP_PUBLIC
+  void
+  remove_callback_group(
+    rclcpp::CallbackGroup::SharedPtr group_ptr,
+    bool notify = true) override;
+
+  /// Add a node to the executor.
+  /**
+   * \sa rclcpp::Executor::add_node
+   */
+  RCLCPP_PUBLIC
+  void
+  add_node(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    bool notify = true) override;
+
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  /**
+   * \sa rclcpp::StaticSingleThreadedExecutor::add_node
+   */
+  RCLCPP_PUBLIC
+  void
+  add_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify = true) override;
+
+  /// Remove a node from the executor.
+  /**
+   * \sa rclcpp::Executor::remove_node
+   */
+  RCLCPP_PUBLIC
+  void
+  remove_node(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+    bool notify = true) override;
+
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  /**
+   * \sa rclcpp::Executor::remove_node
+   */
+  RCLCPP_PUBLIC
+  void
+  remove_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify = true) override;
+
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_all_callback_groups() override;
+
+  /// Get callback groups that belong to executor.
+  /**
+   * \sa rclcpp::Executor::get_manually_added_callback_groups()
+   */
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_manually_added_callback_groups() override;
+
+  /// Get callback groups that belong to executor.
+  /**
+   * \sa rclcpp::Executor::get_automatically_added_callback_groups_from_nodes()
+   */
+  RCLCPP_PUBLIC
+  std::vector<rclcpp::CallbackGroup::WeakPtr>
+  get_automatically_added_callback_groups_from_nodes() override;
+
 protected:
   /**
    * @brief Executes ready executables from wait set.
-   * @param collection entities to evaluate for ready executables.
-   * @param wait_result result to check for ready executables.
    * @param spin_once if true executes only the first ready executable.
    * @return true if any executable was ready.
    */
   RCLCPP_PUBLIC
   bool
-  execute_ready_executables(
-    const rclcpp::executors::ExecutorEntitiesCollection & collection,
-    rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
-    bool spin_once);
+  execute_ready_executables(bool spin_once = false);
 
   RCLCPP_PUBLIC
   void
@@ -130,6 +213,8 @@ protected:
 
 private:
   RCLCPP_DISABLE_COPY(StaticSingleThreadedExecutor)
+
+  StaticExecutorEntitiesCollector::SharedPtr entities_collector_;
 };
 
 }  // namespace executors

rclcpp/include/rclcpp/experimental/buffers/buffer_implementation_base.hpp

@@ -15,6 +15,8 @@
 #ifndef RCLCPP__EXPERIMENTAL__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
 #define RCLCPP__EXPERIMENTAL__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
 
+#include <vector>
+
 namespace rclcpp
 {
 namespace experimental
@@ -31,8 +33,11 @@ public:
   virtual BufferT dequeue() = 0;
   virtual void enqueue(BufferT request) = 0;
 
+  virtual std::vector<BufferT> get_all_data() = 0;
+
   virtual void clear() = 0;
   virtual bool has_data() const = 0;
+  virtual size_t available_capacity() const = 0;
 };
 
 }  // namespace buffers

rclcpp/include/rclcpp/experimental/buffers/intra_process_buffer.hpp

@@ -19,6 +19,7 @@
 #include <stdexcept>
 #include <type_traits>
 #include <utility>
+#include <vector>
 
 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/allocator/allocator_deleter.hpp"
@@ -44,6 +45,7 @@ public:
 
   virtual bool has_data() const = 0;
   virtual bool use_take_shared_method() const = 0;
+  virtual size_t available_capacity() const = 0;
 };
 
 template<
@@ -65,6 +67,9 @@ public:
 
   virtual MessageSharedPtr consume_shared() = 0;
   virtual MessageUniquePtr consume_unique() = 0;
+
+  virtual std::vector<MessageSharedPtr> get_all_data_shared() = 0;
+  virtual std::vector<MessageUniquePtr> get_all_data_unique() = 0;
 };
 
 template<
@@ -95,7 +100,7 @@ public:
 
     buffer_ = std::move(buffer_impl);
 
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_buffer_to_ipb,
       static_cast<const void *>(buffer_.get()),
       static_cast<const void *>(this));
@@ -128,6 +133,16 @@ public:
     return consume_unique_impl<BufferT>();
   }
 
+  std::vector<MessageSharedPtr> get_all_data_shared() override
+  {
+    return get_all_data_shared_impl();
+  }
+
+  std::vector<MessageUniquePtr> get_all_data_unique() override
+  {
+    return get_all_data_unique_impl();
+  }
+
   bool has_data() const override
   {
     return buffer_->has_data();
@@ -143,6 +158,11 @@ public:
     return std::is_same<BufferT, MessageSharedPtr>::value;
   }
 
+  size_t available_capacity() const override
+  {
+    return buffer_->available_capacity();
+  }
+
 private:
   std::unique_ptr<BufferImplementationBase<BufferT>> buffer_;
 
@@ -237,6 +257,71 @@ private:
   {
     return buffer_->dequeue();
   }
+
+  // MessageSharedPtr to MessageSharedPtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageSharedPtr>::value,
+    std::vector<MessageSharedPtr>
+  >::type
+  get_all_data_shared_impl()
+  {
+    return buffer_->get_all_data();
+  }
+
+  // MessageUniquePtr to MessageSharedPtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageUniquePtr>::value,
+    std::vector<MessageSharedPtr>
+  >::type
+  get_all_data_shared_impl()
+  {
+    std::vector<MessageSharedPtr> result;
+    auto uni_ptr_vec = buffer_->get_all_data();
+    result.reserve(uni_ptr_vec.size());
+    for (MessageUniquePtr & uni_ptr : uni_ptr_vec) {
+      result.emplace_back(std::move(uni_ptr));
+    }
+    return result;
+  }
+
+  // MessageSharedPtr to MessageUniquePtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageSharedPtr>::value,
+    std::vector<MessageUniquePtr>
+  >::type
+  get_all_data_unique_impl()
+  {
+    std::vector<MessageUniquePtr> result;
+    auto shared_ptr_vec = buffer_->get_all_data();
+    result.reserve(shared_ptr_vec.size());
+    for (MessageSharedPtr shared_msg : shared_ptr_vec) {
+      MessageUniquePtr unique_msg;
+      MessageDeleter * deleter = std::get_deleter<MessageDeleter, const MessageT>(shared_msg);
+      auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
+      MessageAllocTraits::construct(*message_allocator_.get(), ptr, *shared_msg);
+      if (deleter) {
+        unique_msg = MessageUniquePtr(ptr, *deleter);
+      } else {
+        unique_msg = MessageUniquePtr(ptr);
+      }
+      result.push_back(std::move(unique_msg));
+    }
+    return result;
+  }
+
+  // MessageUniquePtr to MessageUniquePtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageUniquePtr>::value,
+    std::vector<MessageUniquePtr>
+  >::type
+  get_all_data_unique_impl()
+  {
+    return buffer_->get_all_data();
+  }
 };
 
 }  // namespace buffers

rclcpp/include/rclcpp/experimental/buffers/ring_buffer_implementation.hpp

@@ -15,6 +15,7 @@
 #ifndef RCLCPP__EXPERIMENTAL__BUFFERS__RING_BUFFER_IMPLEMENTATION_HPP_
 #define RCLCPP__EXPERIMENTAL__BUFFERS__RING_BUFFER_IMPLEMENTATION_HPP_
 
+#include <memory>
 #include <mutex>
 #include <stdexcept>
 #include <utility>
@@ -52,7 +53,10 @@ public:
     if (capacity == 0) {
       throw std::invalid_argument("capacity must be a positive, non-zero value");
     }
-    TRACEPOINT(rclcpp_construct_ring_buffer, static_cast<const void *>(this), capacity_);
+    TRACETOOLS_TRACEPOINT(
+      rclcpp_construct_ring_buffer,
+      static_cast<const void *>(this),
+      capacity_);
   }
 
   virtual ~RingBufferImplementation() {}
@@ -63,13 +67,13 @@ public:
    *
    * \param request the element to be stored in the ring buffer
    */
-  void enqueue(BufferT request)
+  void enqueue(BufferT request) override
   {
     std::lock_guard<std::mutex> lock(mutex_);
 
     write_index_ = next_(write_index_);
     ring_buffer_[write_index_] = std::move(request);
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_ring_buffer_enqueue,
       static_cast<const void *>(this),
       write_index_,
@@ -89,7 +93,7 @@ public:
    *
    * \return the element that is being removed from the ring buffer
    */
-  BufferT dequeue()
+  BufferT dequeue() override
   {
     std::lock_guard<std::mutex> lock(mutex_);
 
@@ -98,7 +102,7 @@ public:
     }
 
     auto request = std::move(ring_buffer_[read_index_]);
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_ring_buffer_dequeue,
       static_cast<const void *>(this),
       read_index_,
@@ -110,6 +114,17 @@ public:
     return request;
   }
 
+  /// Get all the elements from the ring buffer
+  /**
+   * This member function is thread-safe.
+   *
+   * \return a vector containing all the elements from the ring buffer
+   */
+  std::vector<BufferT> get_all_data() override
+  {
+    return get_all_data_impl();
+  }
+
   /// Get the next index value for the ring buffer
   /**
    * This member function is thread-safe.
@@ -129,7 +144,7 @@ public:
    *
    * \return `true` if there is data and `false` otherwise
    */
-  inline bool has_data() const
+  inline bool has_data() const override
   {
     std::lock_guard<std::mutex> lock(mutex_);
     return has_data_();
@@ -148,9 +163,21 @@ public:
     return is_full_();
   }
 
-  void clear()
+  /// Get the remaining capacity to store messages
+  /**
+   * This member function is thread-safe.
+   *
+   * \return the number of free capacity for new messages
+   */
+  size_t available_capacity() const override
   {
-    TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
+    std::lock_guard<std::mutex> lock(mutex_);
+    return available_capacity_();
+  }
+
+  void clear() override
+  {
+    TRACETOOLS_TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
   }
 
 private:
@@ -189,6 +216,82 @@ private:
     return size_ == capacity_;
   }
 
+  /// Get the remaining capacity to store messages
+  /**
+   * This member function is not thread-safe.
+   *
+   * \return the number of free capacity for new messages
+   */
+  inline size_t available_capacity_() const
+  {
+    return capacity_ - size_;
+  }
+
+  /// Traits for checking if a type is std::unique_ptr
+  template<typename ...>
+  struct is_std_unique_ptr final : std::false_type {};
+  template<class T, typename ... Args>
+  struct is_std_unique_ptr<std::unique_ptr<T, Args...>> final : std::true_type
+  {
+    typedef T Ptr_type;
+  };
+
+  /// Get all the elements from the ring buffer
+  /**
+   * This member function is thread-safe.
+   * Two versions for the implementation of the function.
+   * One for buffer containing unique_ptr and the other for other types
+   *
+   * \return a vector containing all the elements from the ring buffer
+   */
+  template<typename T = BufferT, std::enable_if_t<is_std_unique_ptr<T>::value &&
+    std::is_copy_constructible<
+      typename is_std_unique_ptr<T>::Ptr_type
+    >::value,
+    void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    std::vector<BufferT> result_vtr;
+    result_vtr.reserve(size_);
+    for (size_t id = 0; id < size_; ++id) {
+      result_vtr.emplace_back(
+        new typename is_std_unique_ptr<T>::Ptr_type(
+          *(ring_buffer_[(read_index_ + id) % capacity_])));
+    }
+    return result_vtr;
+  }
+
+  template<typename T = BufferT, std::enable_if_t<
+      std::is_copy_constructible<T>::value, void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    std::vector<BufferT> result_vtr;
+    result_vtr.reserve(size_);
+    for (size_t id = 0; id < size_; ++id) {
+      result_vtr.emplace_back(ring_buffer_[(read_index_ + id) % capacity_]);
+    }
+    return result_vtr;
+  }
+
+  template<typename T = BufferT, std::enable_if_t<!is_std_unique_ptr<T>::value &&
+    !std::is_copy_constructible<T>::value, void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    throw std::logic_error("Underlined type results in invalid get_all_data_impl()");
+    return {};
+  }
+
+  template<typename T = BufferT, std::enable_if_t<is_std_unique_ptr<T>::value &&
+    !std::is_copy_constructible<typename is_std_unique_ptr<T>::Ptr_type>::value,
+    void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    throw std::logic_error("Underlined type in unique_ptr results in invalid get_all_data_impl()");
+    return {};
+  }
+
   size_t capacity_;
 
   std::vector<BufferT> ring_buffer_;

rclcpp/include/rclcpp/experimental/executors/events_executor/events_executor.hpp

@@ -243,6 +243,11 @@ private:
   std::function<void(size_t, int)>
   create_waitable_callback(const rclcpp::Waitable * waitable_id);
 
+  /// Utility to add the notify waitable to an entities collection
+  void
+  add_notify_waitable_to_collection(
+    rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection);
+
   /// Searches for the provided entity_id in the collection and returns the entity if valid
   template<typename CollectionType>
   typename CollectionType::EntitySharedPtr
@@ -269,9 +274,12 @@ private:
   rclcpp::experimental::executors::EventsQueue::UniquePtr events_queue_;
 
   std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollector> entities_collector_;
-  std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollection> current_entities_collection_;
   std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> notify_waitable_;
 
+  /// Mutex to protect the current_entities_collection_
+  std::recursive_mutex collection_mutex_;
+  std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollection> current_entities_collection_;
+
   /// Flag used to reduce the number of unnecessary waitable events
   std::atomic<bool> notify_waitable_event_pushed_ {false};
 

rclcpp/include/rclcpp/experimental/intra_process_manager.hpp

@@ -28,6 +28,7 @@
 #include <typeinfo>
 
 #include "rclcpp/allocator/allocator_deleter.hpp"
+#include "rclcpp/experimental/buffers/intra_process_buffer.hpp"
 #include "rclcpp/experimental/ros_message_intra_process_buffer.hpp"
 #include "rclcpp/experimental/subscription_intra_process.hpp"
 #include "rclcpp/experimental/subscription_intra_process_base.hpp"
@@ -112,9 +113,40 @@ public:
    * \param subscription the SubscriptionIntraProcess to register.
    * \return an unsigned 64-bit integer which is the subscription's unique id.
    */
-  RCLCPP_PUBLIC
+  template<
+    typename ROSMessageType,
+    typename Alloc = std::allocator<ROSMessageType>
+  >
   uint64_t
-  add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription);
+  add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription)
+  {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+
+    uint64_t sub_id = IntraProcessManager::get_next_unique_id();
+
+    subscriptions_[sub_id] = subscription;
+
+    // adds the subscription id to all the matchable publishers
+    for (auto & pair : publishers_) {
+      auto publisher = pair.second.lock();
+      if (!publisher) {
+        continue;
+      }
+      if (can_communicate(publisher, subscription)) {
+        uint64_t pub_id = pair.first;
+        insert_sub_id_for_pub(sub_id, pub_id, subscription->use_take_shared_method());
+        if (publisher->is_durability_transient_local() &&
+          subscription->is_durability_transient_local())
+        {
+          do_transient_local_publish<ROSMessageType, Alloc>(
+            pub_id, sub_id,
+            subscription->use_take_shared_method());
+        }
+      }
+    }
+
+    return sub_id;
+  }
 
   /// Unregister a subscription using the subscription's unique id.
   /**
@@ -131,14 +163,21 @@ public:
    * This method stores the publisher intra process object, together with
    * the information of its wrapped publisher (i.e. topic name and QoS).
    *
+   * If the publisher's durability is transient local, its buffer pointer should
+   * be passed and the method will store it as well.
+   *
    * In addition this generates a unique intra process id for the publisher.
    *
    * \param publisher publisher to be registered with the manager.
+   * \param buffer publisher's buffer to be stored if its duability is transient local.
    * \return an unsigned 64-bit integer which is the publisher's unique id.
    */
   RCLCPP_PUBLIC
   uint64_t
-  add_publisher(rclcpp::PublisherBase::SharedPtr publisher);
+  add_publisher(
+    rclcpp::PublisherBase::SharedPtr publisher,
+    rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr buffer =
+    rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr());
 
   /// Unregister a publisher using the publisher's unique id.
   /**
@@ -292,6 +331,34 @@ public:
     }
   }
 
+  template<
+    typename MessageT,
+    typename Alloc,
+    typename Deleter,
+    typename ROSMessageType>
+  void
+  add_shared_msg_to_buffer(
+    std::shared_ptr<const MessageT> message,
+    uint64_t subscription_id)
+  {
+    add_shared_msg_to_buffers<MessageT, Alloc, Deleter, ROSMessageType>(message, {subscription_id});
+  }
+
+  template<
+    typename MessageT,
+    typename Alloc,
+    typename Deleter,
+    typename ROSMessageType>
+  void
+  add_owned_msg_to_buffer(
+    std::unique_ptr<MessageT, Deleter> message,
+    uint64_t subscription_id,
+    typename allocator::AllocRebind<MessageT, Alloc>::allocator_type & allocator)
+  {
+    add_owned_msg_to_buffers<MessageT, Alloc, Deleter, ROSMessageType>(
+      std::move(message), {subscription_id}, allocator);
+  }
+
   /// Return true if the given rmw_gid_t matches any stored Publishers.
   RCLCPP_PUBLIC
   bool
@@ -306,6 +373,11 @@ public:
   rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr
   get_subscription_intra_process(uint64_t intra_process_subscription_id);
 
+  /// Return the lowest available capacity for all subscription buffers for a publisher id.
+  RCLCPP_PUBLIC
+  size_t
+  lowest_available_capacity(const uint64_t intra_process_publisher_id) const;
+
 private:
   struct SplittedSubscriptions
   {
@@ -319,6 +391,9 @@ private:
   using PublisherMap =
     std::unordered_map<uint64_t, rclcpp::PublisherBase::WeakPtr>;
 
+  using PublisherBufferMap =
+    std::unordered_map<uint64_t, rclcpp::experimental::buffers::IntraProcessBufferBase::WeakPtr>;
+
   using PublisherToSubscriptionIdsMap =
     std::unordered_map<uint64_t, SplittedSubscriptions>;
 
@@ -337,6 +412,54 @@ private:
     rclcpp::PublisherBase::SharedPtr pub,
     rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr sub) const;
 
+  template<
+    typename ROSMessageType,
+    typename Alloc = std::allocator<ROSMessageType>
+  >
+  void do_transient_local_publish(
+    const uint64_t pub_id, const uint64_t sub_id,
+    const bool use_take_shared_method)
+  {
+    using ROSMessageTypeAllocatorTraits = allocator::AllocRebind<ROSMessageType, Alloc>;
+    using ROSMessageTypeAllocator = typename ROSMessageTypeAllocatorTraits::allocator_type;
+    using ROSMessageTypeDeleter = allocator::Deleter<ROSMessageTypeAllocator, ROSMessageType>;
+
+    auto publisher_buffer = publisher_buffers_[pub_id].lock();
+    if (!publisher_buffer) {
+      throw std::runtime_error("publisher buffer has unexpectedly gone out of scope");
+    }
+    auto buffer = std::dynamic_pointer_cast<
+      rclcpp::experimental::buffers::IntraProcessBuffer<
+        ROSMessageType,
+        ROSMessageTypeAllocator,
+        ROSMessageTypeDeleter
+      >
+      >(publisher_buffer);
+    if (!buffer) {
+      throw std::runtime_error(
+              "failed to dynamic cast publisher's IntraProcessBufferBase to "
+              "IntraProcessBuffer<ROSMessageType,ROSMessageTypeAllocator,"
+              "ROSMessageTypeDeleter> which can happen when the publisher and "
+              "subscription use different allocator types, which is not supported");
+    }
+    if (use_take_shared_method) {
+      auto data_vec = buffer->get_all_data_shared();
+      for (auto shared_data : data_vec) {
+        this->template add_shared_msg_to_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter, ROSMessageType>(
+          shared_data, sub_id);
+      }
+    } else {
+      auto data_vec = buffer->get_all_data_unique();
+      for (auto & owned_data : data_vec) {
+        auto allocator = ROSMessageTypeAllocator();
+        this->template add_owned_msg_to_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter, ROSMessageType>(
+          std::move(owned_data), sub_id, allocator);
+      }
+    }
+  }
+
   template<
     typename MessageT,
     typename Alloc,
@@ -462,7 +585,7 @@ private:
           auto ptr = MessageAllocTraits::allocate(allocator, 1);
           MessageAllocTraits::construct(allocator, ptr, *message);
 
-          subscription->provide_intra_process_data(std::move(MessageUniquePtr(ptr, deleter)));
+          subscription->provide_intra_process_data(MessageUniquePtr(ptr, deleter));
         }
 
         continue;
@@ -481,13 +604,13 @@ private:
                 "subscription use different allocator types, which is not supported");
       }
 
-      if constexpr (rclcpp::TypeAdapter<MessageT>::is_specialized::value) {
+      if constexpr (rclcpp::TypeAdapter<MessageT, ROSMessageType>::is_specialized::value) {
         ROSMessageTypeAllocator ros_message_alloc(allocator);
-        auto ptr = ros_message_alloc.allocate(1);
-        ros_message_alloc.construct(ptr);
+        auto ptr = ROSMessageTypeAllocatorTraits::allocate(ros_message_alloc, 1);
+        ROSMessageTypeAllocatorTraits::construct(ros_message_alloc, ptr);
         ROSMessageTypeDeleter deleter;
         allocator::set_allocator_for_deleter(&deleter, &allocator);
-        rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*message, *ptr);
+        rclcpp::TypeAdapter<MessageT, ROSMessageType>::convert_to_ros_message(*message, *ptr);
         auto ros_msg = std::unique_ptr<ROSMessageType, ROSMessageTypeDeleter>(ptr, deleter);
         ros_message_subscription->provide_intra_process_message(std::move(ros_msg));
       } else {
@@ -505,7 +628,7 @@ private:
             MessageAllocTraits::construct(allocator, ptr, *message);
 
             ros_message_subscription->provide_intra_process_message(
-              std::move(MessageUniquePtr(ptr, deleter)));
+              MessageUniquePtr(ptr, deleter));
           }
         }
       }
@@ -515,6 +638,7 @@ private:
   PublisherToSubscriptionIdsMap pub_to_subs_;
   SubscriptionMap subscriptions_;
   PublisherMap publishers_;
+  PublisherBufferMap publisher_buffers_;
 
   mutable std::shared_timed_mutex mutex_;
 };

rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp

@@ -87,7 +87,7 @@ public:
       buffer_type),
     any_callback_(callback)
   {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_subscription_callback_added,
       static_cast<const void *>(this),
       static_cast<const void *>(&any_callback_));

rclcpp/include/rclcpp/experimental/subscription_intra_process_base.hpp

@@ -62,6 +62,15 @@ public:
   void
   add_to_wait_set(rcl_wait_set_t * wait_set) override;
 
+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  available_capacity() const = 0;
+
+  RCLCPP_PUBLIC
+  bool
+  is_durability_transient_local() const;
+
   bool
   is_ready(rcl_wait_set_t * wait_set) override = 0;
 

rclcpp/include/rclcpp/experimental/subscription_intra_process_buffer.hpp

@@ -93,7 +93,7 @@ public:
       buffer_type,
       qos_profile,
       std::make_shared<Alloc>(subscribed_type_allocator_));
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_ipb_to_subscription,
       static_cast<const void *>(buffer_.get()),
       static_cast<const void *>(this));
@@ -169,6 +169,11 @@ public:
     return buffer_->use_take_shared_method();
   }
 
+  size_t available_capacity() const override
+  {
+    return buffer_->available_capacity();
+  }
+
 protected:
   void
   trigger_guard_condition() override

rclcpp/include/rclcpp/experimental/timers_manager.hpp

@@ -22,10 +22,10 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <thread>
 #include <utility>
 #include <vector>
-
 #include "rclcpp/context.hpp"
 #include "rclcpp/timer.hpp"
 
@@ -172,13 +172,14 @@ public:
    * @brief Get the amount of time before the next timer triggers.
    * This function is thread safe.
    *
-   * @return std::chrono::nanoseconds to wait,
+   * @return std::optional<std::chrono::nanoseconds> to wait,
    * the returned value could be negative if the timer is already expired
    * or std::chrono::nanoseconds::max() if there are no timers stored in the object.
+   * If the head timer was cancelled, then this will return a nullopt.
    * @throws std::runtime_error if the timers thread was already running.
    */
   RCLCPP_PUBLIC
-  std::chrono::nanoseconds get_head_timeout();
+  std::optional<std::chrono::nanoseconds> get_head_timeout();
 
 private:
   RCLCPP_DISABLE_COPY(TimersManager)
@@ -512,12 +513,13 @@ private:
    * @brief Get the amount of time before the next timer triggers.
    * This function is not thread safe, acquire a mutex before calling it.
    *
-   * @return std::chrono::nanoseconds to wait,
+   * @return std::optional<std::chrono::nanoseconds> to wait,
    * the returned value could be negative if the timer is already expired
    * or std::chrono::nanoseconds::max() if the heap is empty.
+   * If the head timer was cancelled, then this will return a nullopt.
    * This function is not thread safe, acquire the timers_mutex_ before calling it.
    */
-  std::chrono::nanoseconds get_head_timeout_unsafe();
+  std::optional<std::chrono::nanoseconds> get_head_timeout_unsafe();
 
   /**
    * @brief Executes all the timers currently ready when the function is invoked
@@ -527,7 +529,7 @@ private:
   void execute_ready_timers_unsafe();
 
   // Callback to be called when timer is ready
-  std::function<void(const rclcpp::TimerBase *)> on_ready_callback_ = nullptr;
+  std::function<void(const rclcpp::TimerBase *)> on_ready_callback_;
 
   // Thread used to run the timers execution task
   std::thread timers_thread_;

rclcpp/include/rclcpp/generic_publisher.hpp

@@ -77,7 +77,7 @@ public:
   : rclcpp::PublisherBase(
       node_base,
       topic_name,
-      *rclcpp::get_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
+      *rclcpp::get_message_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
       options.template to_rcl_publisher_options<rclcpp::SerializedMessage>(qos),
       // NOTE(methylDragon): Passing these args separately is necessary for event binding
       options.event_callbacks,

rclcpp/include/rclcpp/generic_subscription.hpp

@@ -74,18 +74,21 @@ public:
     const std::string & topic_name,
     const std::string & topic_type,
     const rclcpp::QoS & qos,
-    // TODO(nnmm): Add variant for callback with message info. See issue #1604.
-    std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+    AnySubscriptionCallback<rclcpp::SerializedMessage, AllocatorT> callback,
     const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options)
   : SubscriptionBase(
       node_base,
-      *rclcpp::get_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
+      *rclcpp::get_message_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
       topic_name,
       options.to_rcl_subscription_options(qos),
       options.event_callbacks,
       options.use_default_callbacks,
       DeliveredMessageKind::SERIALIZED_MESSAGE),
-    callback_(callback),
+    callback_([callback](
+        std::shared_ptr<const rclcpp::SerializedMessage> serialized_message,
+        const rclcpp::MessageInfo & message_info) mutable {
+        callback.dispatch(serialized_message, message_info);
+      }),
     ts_lib_(ts_lib)
   {}
 
@@ -151,7 +154,9 @@ public:
 private:
   RCLCPP_DISABLE_COPY(GenericSubscription)
 
-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback_;
+  std::function<void(
+      std::shared_ptr<const rclcpp::SerializedMessage>,
+      const rclcpp::MessageInfo)> callback_;
   // The type support library should stay loaded, so it is stored in the GenericSubscription
   std::shared_ptr<rcpputils::SharedLibrary> ts_lib_;
 };

rclcpp/include/rclcpp/guard_condition.hpp

@@ -48,7 +48,7 @@ public:
    */
   RCLCPP_PUBLIC
   explicit GuardCondition(
-    rclcpp::Context::SharedPtr context =
+    const rclcpp::Context::SharedPtr & context =
     rclcpp::contexts::get_global_default_context(),
     rcl_guard_condition_options_t guard_condition_options =
     rcl_guard_condition_get_default_options());
@@ -57,11 +57,6 @@ public:
   virtual
   ~GuardCondition();
 
-  /// Return the context used when creating this guard condition.
-  RCLCPP_PUBLIC
-  rclcpp::Context::SharedPtr
-  get_context() const;
-
   /// Return the underlying rcl guard condition structure.
   RCLCPP_PUBLIC
   rcl_guard_condition_t &
@@ -128,7 +123,6 @@ public:
   set_on_trigger_callback(std::function<void(size_t)> callback);
 
 protected:
-  rclcpp::Context::SharedPtr context_;
   rcl_guard_condition_t rcl_guard_condition_;
   std::atomic<bool> in_use_by_wait_set_{false};
   std::recursive_mutex reentrant_mutex_;

rclcpp/include/rclcpp/logger.hpp

@@ -126,9 +126,6 @@ private:
   std::shared_ptr<std::pair<std::string, std::string>> logger_sublogger_pairname_ = nullptr;
 
 public:
-  RCLCPP_PUBLIC
-  Logger(const Logger &) = default;
-
   /// Get the name of this logger.
   /**
    * \return the full name of the logger including any prefixes, or

rclcpp/include/rclcpp/node.hpp

@@ -56,6 +56,7 @@
 #include "rclcpp/node_interfaces/node_time_source_interface.hpp"
 #include "rclcpp/node_interfaces/node_timers_interface.hpp"
 #include "rclcpp/node_interfaces/node_topics_interface.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
 #include "rclcpp/node_interfaces/node_waitables_interface.hpp"
 #include "rclcpp/node_options.hpp"
 #include "rclcpp/parameter.hpp"
@@ -232,13 +233,15 @@ public:
    * \param[in] period Time interval between triggers of the callback.
    * \param[in] callback User-defined callback function.
    * \param[in] group Callback group to execute this timer's callback in.
+   * \param[in] autostart The state of the clock on initialization.
    */
   template<typename DurationRepT = int64_t, typename DurationT = std::milli, typename CallbackT>
   typename rclcpp::WallTimer<CallbackT>::SharedPtr
   create_wall_timer(
     std::chrono::duration<DurationRepT, DurationT> period,
     CallbackT callback,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::CallbackGroup::SharedPtr group = nullptr,
+    bool autostart = true);
 
   /// Create a timer that uses the node clock to drive the callback.
   /**
@@ -355,12 +358,14 @@ public:
  * `%callback_group`.
    * \return Shared pointer to the created generic subscription.
    */
-  template<typename AllocatorT = std::allocator<void>>
+  template<
+    typename CallbackT,
+    typename AllocatorT = std::allocator<void>>
   std::shared_ptr<rclcpp::GenericSubscription> create_generic_subscription(
     const std::string & topic_name,
     const std::string & topic_type,
     const rclcpp::QoS & qos,
-    std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+    CallbackT && callback,
     const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options = (
       rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>()
     )
@@ -969,7 +974,16 @@ public:
 
   /// Return a list of parameters with any of the given prefixes, up to the given depth.
   /**
-   * \todo: properly document and test this method.
+   * Parameters are separated into a hierarchy using the "." (dot) character.
+   * The "prefixes" argument is a way to select only particular parts of the hierarchy.
+   *
+   * \param[in] prefixes The list of prefixes that should be searched for within the
+   * current parameters. If this vector of prefixes is empty, then list_parameters
+   * will return all parameters.
+   * \param[in] depth An unsigned integer that represents the recursive depth to search.
+   * If this depth = 0, then all parameters that fit the prefixes will be returned.
+   * \returns A ListParametersResult message which contains both an array of unique prefixes
+   * and an array of names that were matched to the prefixes given.
    */
   RCLCPP_PUBLIC
   rcl_interfaces::msg::ListParametersResult
@@ -1302,6 +1316,26 @@ public:
   size_t
   count_subscribers(const std::string & topic_name) const;
 
+  /// Return the number of clients created for a given service.
+  /**
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   * \return number of clients that have been created for the given service.
+   * \throws std::runtime_error if clients could not be counted
+   */
+  RCLCPP_PUBLIC
+  size_t
+  count_clients(const std::string & service_name) const;
+
+  /// Return the number of services created for a given service.
+  /**
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   * \return number of services that have been created for the given service.
+   * \throws std::runtime_error if services could not be counted
+   */
+  RCLCPP_PUBLIC
+  size_t
+  count_services(const std::string & service_name) const;
+
   /// Return the topic endpoint information about publishers on a given topic.
   /**
    * The returned parameter is a list of topic endpoint information, where each item will contain
@@ -1454,6 +1488,11 @@ public:
   rclcpp::node_interfaces::NodeTimeSourceInterface::SharedPtr
   get_node_time_source_interface();
 
+  /// Return the Node's internal NodeTypeDescriptionsInterface implementation.
+  RCLCPP_PUBLIC
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
+  get_node_type_descriptions_interface();
+
   /// Return the sub-namespace, if this is a sub-node, otherwise an empty string.
   /**
    * The returned sub-namespace is either the accumulated sub-namespaces which
@@ -1586,11 +1625,18 @@ private:
   rclcpp::node_interfaces::NodeClockInterface::SharedPtr node_clock_;
   rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters_;
   rclcpp::node_interfaces::NodeTimeSourceInterface::SharedPtr node_time_source_;
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr node_type_descriptions_;
   rclcpp::node_interfaces::NodeWaitablesInterface::SharedPtr node_waitables_;
 
   const rclcpp::NodeOptions node_options_;
   const std::string sub_namespace_;
   const std::string effective_namespace_;
+
+  class NodeImpl;
+  // This member is meant to be a place to backport features into stable distributions,
+  // and new features targeting Rolling should not use this.
+  // See the comment in node.cpp for more information.
+  std::shared_ptr<NodeImpl> hidden_impl_{nullptr};
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/node_impl.hpp

@@ -110,14 +110,16 @@ typename rclcpp::WallTimer<CallbackT>::SharedPtr
 Node::create_wall_timer(
   std::chrono::duration<DurationRepT, DurationT> period,
   CallbackT callback,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::CallbackGroup::SharedPtr group,
+  bool autostart)
 {
   return rclcpp::create_wall_timer(
     period,
     std::move(callback),
     group,
     this->node_base_.get(),
-    this->node_timers_.get());
+    this->node_timers_.get(),
+    autostart);
 }
 
 template<typename DurationRepT, typename DurationT, typename CallbackT>
@@ -219,13 +221,13 @@ Node::create_generic_publisher(
   );
 }
 
-template<typename AllocatorT>
+template<typename CallbackT, typename AllocatorT>
 std::shared_ptr<rclcpp::GenericSubscription>
 Node::create_generic_subscription(
   const std::string & topic_name,
   const std::string & topic_type,
   const rclcpp::QoS & qos,
-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+  CallbackT && callback,
   const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options)
 {
   return rclcpp::create_generic_subscription(
@@ -233,7 +235,7 @@ Node::create_generic_subscription(
     extend_name_with_sub_namespace(topic_name, this->get_sub_namespace()),
     topic_type,
     qos,
-    std::move(callback),
+    std::forward<CallbackT>(callback),
     options
   );
 }

rclcpp/include/rclcpp/node_interfaces/node_graph.hpp

@@ -113,6 +113,14 @@ public:
   size_t
   count_subscribers(const std::string & topic_name) const override;
 
+  RCLCPP_PUBLIC
+  size_t
+  count_clients(const std::string & service_name) const override;
+
+  RCLCPP_PUBLIC
+  size_t
+  count_services(const std::string & service_name) const override;
+
   RCLCPP_PUBLIC
   const rcl_guard_condition_t *
   get_graph_guard_condition() const override;

rclcpp/include/rclcpp/node_interfaces/node_graph_interface.hpp

@@ -305,6 +305,24 @@ public:
   size_t
   count_subscribers(const std::string & topic_name) const = 0;
 
+  /// Return the number of clients created for a given service.
+  /*
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   */
+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  count_clients(const std::string & service_name) const = 0;
+
+  /// Return the number of services created for a given service.
+  /*
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   */
+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  count_services(const std::string & service_name) const = 0;
+
   /// Return the rcl guard condition which is triggered when the ROS graph changes.
   RCLCPP_PUBLIC
   virtual

rclcpp/include/rclcpp/node_interfaces/node_interfaces.hpp

@@ -30,6 +30,7 @@
   rclcpp::node_interfaces::NodeTimeSourceInterface, \
   rclcpp::node_interfaces::NodeTimersInterface, \
   rclcpp::node_interfaces::NodeTopicsInterface, \
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface, \
   rclcpp::node_interfaces::NodeWaitablesInterface
 
 
@@ -118,6 +119,7 @@ public:
    *   - rclcpp::node_interfaces::NodeTimeSourceInterface
    *   - rclcpp::node_interfaces::NodeTimersInterface
    *   - rclcpp::node_interfaces::NodeTopicsInterface
+   *   - rclcpp::node_interfaces::NodeTypeDescriptionsInterface
    *   - rclcpp::node_interfaces::NodeWaitablesInterface
    *
    * Or you use custom interfaces as long as you make a template specialization

rclcpp/include/rclcpp/node_interfaces/node_logging.hpp

@@ -23,6 +23,9 @@
 #include "rclcpp/node_interfaces/node_logging_interface.hpp"
 #include "rclcpp/visibility_control.hpp"
 
+#include "rcl_interfaces/srv/get_logger_levels.hpp"
+#include "rcl_interfaces/srv/set_logger_levels.hpp"
+
 namespace rclcpp
 {
 namespace node_interfaces
@@ -35,7 +38,7 @@ public:
   RCLCPP_SMART_PTR_ALIASES_ONLY(NodeLoggingInterface)
 
   RCLCPP_PUBLIC
-  explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base);
+  explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base);
 
   RCLCPP_PUBLIC
   virtual
@@ -49,13 +52,21 @@ public:
   const char *
   get_logger_name() const override;
 
+  RCLCPP_PUBLIC
+  void
+  create_logger_services(
+    node_interfaces::NodeServicesInterface::SharedPtr node_services) override;
+
 private:
   RCLCPP_DISABLE_COPY(NodeLogging)
 
   /// Handle to the NodeBaseInterface given in the constructor.
-  rclcpp::node_interfaces::NodeBaseInterface * node_base_;
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
 
   rclcpp::Logger logger_;
+
+  rclcpp::Service<rcl_interfaces::srv::GetLoggerLevels>::SharedPtr get_loggers_service_;
+  rclcpp::Service<rcl_interfaces::srv::SetLoggerLevels>::SharedPtr set_loggers_service_;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_interfaces/node_logging_interface.hpp

@@ -19,6 +19,7 @@
 
 #include "rclcpp/logger.hpp"
 #include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"
 #include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
 #include "rclcpp/visibility_control.hpp"
 
@@ -54,6 +55,13 @@ public:
   virtual
   const char *
   get_logger_name() const = 0;
+
+  /// create logger services
+  RCLCPP_PUBLIC
+  virtual
+  void
+  create_logger_services(
+    node_interfaces::NodeServicesInterface::SharedPtr node_services) = 0;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_interfaces/node_type_descriptions.hpp

@@ -0,0 +1,63 @@
+// Copyright 2023 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__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
+#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
+
+#include <memory>
+
+#include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/node_base_interface.hpp"
+#include "rclcpp/node_interfaces/node_logging_interface.hpp"
+#include "rclcpp/node_interfaces/node_parameters_interface.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"
+#include "rclcpp/node_interfaces/node_topics_interface.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
+#include "rclcpp/visibility_control.hpp"
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+/// Implementation of the NodeTypeDescriptions part of the Node API.
+class NodeTypeDescriptions : public NodeTypeDescriptionsInterface
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptions)
+
+  RCLCPP_PUBLIC
+  explicit NodeTypeDescriptions(
+    NodeBaseInterface::SharedPtr node_base,
+    NodeLoggingInterface::SharedPtr node_logging,
+    NodeParametersInterface::SharedPtr node_parameters,
+    NodeServicesInterface::SharedPtr node_services);
+
+  RCLCPP_PUBLIC
+  virtual
+  ~NodeTypeDescriptions();
+
+private:
+  RCLCPP_DISABLE_COPY(NodeTypeDescriptions)
+
+  // Pimpl hides helper types and functions used for wrapping a C service, which would be
+  // awkward to expose in this header.
+  class NodeTypeDescriptionsImpl;
+  std::unique_ptr<NodeTypeDescriptionsImpl> impl_;
+};
+
+}  // namespace node_interfaces
+}  // namespace rclcpp
+
+#endif  // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_

rclcpp/include/rclcpp/node_interfaces/node_type_descriptions_interface.hpp

@@ -0,0 +1,44 @@
+// Copyright 2023 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__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
+#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
+
+#include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
+#include "rclcpp/visibility_control.hpp"
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+/// Pure virtual interface class for the NodeTypeDescriptions part of the Node API.
+class NodeTypeDescriptionsInterface
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptionsInterface)
+
+  RCLCPP_PUBLIC
+  virtual
+  ~NodeTypeDescriptionsInterface() = default;
+};
+
+}  // namespace node_interfaces
+}  // namespace rclcpp
+
+RCLCPP_NODE_INTERFACE_HELPERS_SUPPORT(
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface, type_descriptions)
+
+#endif  // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_

rclcpp/include/rclcpp/node_options.hpp

@@ -43,6 +43,7 @@ public:
    *   - arguments = {}
    *   - parameter_overrides = {}
    *   - use_global_arguments = true
+   *   - enable_rosout = true
    *   - use_intra_process_comms = false
    *   - enable_topic_statistics = false
    *   - start_parameter_services = true
@@ -50,6 +51,7 @@ public:
    *   - clock_type = RCL_ROS_TIME
    *   - clock_qos = rclcpp::ClockQoS()
    *   - use_clock_thread = true
+   *   - enable_logger_service = false
    *   - rosout_qos = rclcpp::RosoutQoS()
    *   - parameter_event_qos = rclcpp::ParameterEventQoS
    *     - with history setting and depth from rmw_qos_profile_parameter_events
@@ -232,6 +234,24 @@ public:
   NodeOptions &
   start_parameter_services(bool start_parameter_services);
 
+  /// Return the enable_logger_service flag.
+  RCLCPP_PUBLIC
+  bool
+  enable_logger_service() const;
+
+  /// Set the enable_logger_service flag, return this for logger idiom.
+  /**
+   * If true, ROS services are created to allow external nodes to get
+   * and set logger levels of this node.
+   *
+   * If false, loggers will still be configured and set logger levels locally,
+   * but logger levels cannot be changed remotely .
+   *
+   */
+  RCLCPP_PUBLIC
+  NodeOptions &
+  enable_logger_service(bool enable_log_service);
+
   /// Return the start_parameter_event_publisher flag.
   RCLCPP_PUBLIC
   bool
@@ -421,6 +441,8 @@ private:
 
   bool use_clock_thread_ {true};
 
+  bool enable_logger_service_ {false};
+
   rclcpp::QoS parameter_event_qos_ = rclcpp::ParameterEventsQoS(
     rclcpp::QoSInitialization::from_rmw(rmw_qos_profile_parameter_events)
   );

rclcpp/include/rclcpp/parameter_events_filter.hpp

@@ -48,7 +48,7 @@ public:
    *
    * Example Usage:
    *
-   * If you have recieved a parameter event and are only interested in parameters foo and
+   * If you have received a parameter event and are only interested in parameters foo and
    * bar being added or changed but don't care about deletion.
    *
    * ```cpp

rclcpp/include/rclcpp/parameter_value.hpp

@@ -24,6 +24,7 @@
 
 #include "rcl_interfaces/msg/parameter_type.hpp"
 #include "rcl_interfaces/msg/parameter_value.hpp"
+#include "rclcpp/exceptions/exceptions.hpp"
 #include "rclcpp/visibility_control.hpp"
 
 namespace rclcpp

rclcpp/include/rclcpp/publisher.hpp

@@ -32,6 +32,9 @@
 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/allocator/allocator_deleter.hpp"
 #include "rclcpp/detail/resolve_use_intra_process.hpp"
+#include "rclcpp/detail/resolve_intra_process_buffer_type.hpp"
+#include "rclcpp/experimental/buffers/intra_process_buffer.hpp"
+#include "rclcpp/experimental/create_intra_process_buffer.hpp"
 #include "rclcpp/experimental/intra_process_manager.hpp"
 #include "rclcpp/get_message_type_support_handle.hpp"
 #include "rclcpp/is_ros_compatible_type.hpp"
@@ -109,6 +112,12 @@ public:
   [[deprecated("use std::shared_ptr<const PublishedType>")]] =
     std::shared_ptr<const PublishedType>;
 
+  using BufferSharedPtr = typename rclcpp::experimental::buffers::IntraProcessBuffer<
+    ROSMessageType,
+    ROSMessageTypeAllocator,
+    ROSMessageTypeDeleter
+    >::SharedPtr;
+
   RCLCPP_SMART_PTR_DEFINITIONS(Publisher<MessageT, AllocatorT>)
 
   /// Default constructor.
@@ -171,11 +180,14 @@ public:
         throw std::invalid_argument(
                 "intraprocess communication is not allowed with a zero qos history depth value");
       }
-      if (qos.durability() != rclcpp::DurabilityPolicy::Volatile) {
-        throw std::invalid_argument(
-                "intraprocess communication allowed only with volatile durability");
+      if (qos.durability() == rclcpp::DurabilityPolicy::TransientLocal) {
+        buffer_ = rclcpp::experimental::create_intra_process_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter>(
+          rclcpp::detail::resolve_intra_process_buffer_type(options_.intra_process_buffer_type),
+          qos,
+          std::make_shared<ROSMessageTypeAllocator>(ros_message_type_allocator_));
       }
-      uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this());
+      uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this(), buffer_);
       this->setup_intra_process(
         intra_process_publisher_id,
         ipm);
@@ -242,9 +254,18 @@ public:
     if (inter_process_publish_needed) {
       auto shared_msg =
         this->do_intra_process_ros_message_publish_and_return_shared(std::move(msg));
+      if (buffer_) {
+        buffer_->add_shared(shared_msg);
+      }
       this->do_inter_process_publish(*shared_msg);
     } else {
-      this->do_intra_process_ros_message_publish(std::move(msg));
+      if (buffer_) {
+        auto shared_msg =
+          this->do_intra_process_ros_message_publish_and_return_shared(std::move(msg));
+        buffer_->add_shared(shared_msg);
+      } else {
+        this->do_intra_process_ros_message_publish(std::move(msg));
+      }
     }
   }
 
@@ -309,14 +330,22 @@ public:
       get_subscription_count() > get_intra_process_subscription_count();
 
     if (inter_process_publish_needed) {
-      ROSMessageType ros_msg;
+      auto ros_msg_ptr = std::make_shared<ROSMessageType>();
       // TODO(clalancette): This is unnecessarily doing an additional conversion
       // that may have already been done in do_intra_process_publish_and_return_shared().
       // We should just reuse that effort.
-      rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, ros_msg);
+      rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, *ros_msg_ptr);
       this->do_intra_process_publish(std::move(msg));
-      this->do_inter_process_publish(ros_msg);
+      this->do_inter_process_publish(*ros_msg_ptr);
+      if (buffer_) {
+        buffer_->add_shared(ros_msg_ptr);
+      }
     } else {
+      if (buffer_) {
+        auto ros_msg_ptr = std::make_shared<ROSMessageType>();
+        rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, *ros_msg_ptr);
+        buffer_->add_shared(ros_msg_ptr);
+      }
       this->do_intra_process_publish(std::move(msg));
     }
   }
@@ -390,7 +419,7 @@ public:
     if (this->can_loan_messages()) {
       // we release the ownership from the rclpp::LoanedMessage instance
       // and let the middleware clean up the memory.
-      this->do_loaned_message_publish(std::move(loaned_msg.release()));
+      this->do_loaned_message_publish(loaned_msg.release());
     } else {
       // we don't release the ownership, let the middleware copy the ros message
       // and thus the destructor of rclcpp::LoanedMessage cleans up the memory.
@@ -421,7 +450,7 @@ protected:
   void
   do_inter_process_publish(const ROSMessageType & msg)
   {
-    TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(&msg));
+    TRACETOOLS_TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(&msg));
     auto status = rcl_publish(publisher_handle_.get(), &msg, nullptr);
 
     if (RCL_RET_PUBLISHER_INVALID == status) {
@@ -456,6 +485,7 @@ protected:
   do_loaned_message_publish(
     std::unique_ptr<ROSMessageType, std::function<void(ROSMessageType *)>> msg)
   {
+    TRACETOOLS_TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(msg.get()));
     auto status = rcl_publish_loaned_message(publisher_handle_.get(), msg.get(), nullptr);
 
     if (RCL_RET_PUBLISHER_INVALID == status) {
@@ -484,7 +514,7 @@ protected:
     if (!msg) {
       throw std::runtime_error("cannot publish msg which is a null pointer");
     }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_intra_publish,
       static_cast<const void *>(publisher_handle_.get()),
       msg.get());
@@ -506,7 +536,7 @@ protected:
     if (!msg) {
       throw std::runtime_error("cannot publish msg which is a null pointer");
     }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_intra_publish,
       static_cast<const void *>(publisher_handle_.get()),
       msg.get());
@@ -529,7 +559,7 @@ protected:
     if (!msg) {
       throw std::runtime_error("cannot publish msg which is a null pointer");
     }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_intra_publish,
       static_cast<const void *>(publisher_handle_.get()),
       msg.get());
@@ -580,6 +610,8 @@ protected:
   PublishedTypeDeleter published_type_deleter_;
   ROSMessageTypeAllocator ros_message_type_allocator_;
   ROSMessageTypeDeleter ros_message_type_deleter_;
+
+  BufferSharedPtr buffer_{nullptr};
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/publisher_base.hpp

@@ -139,6 +139,12 @@ public:
   size_t
   get_intra_process_subscription_count() const;
 
+  /// Get if durability is transient local
+  /** \return If durability is transient local*/
+  RCLCPP_PUBLIC
+  bool
+  is_durability_transient_local() const;
+
   /// Manually assert that this Publisher is alive (for RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC).
   /**
    * If the rmw Liveliness policy is set to RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC, the creator
@@ -215,6 +221,17 @@ public:
   std::vector<rclcpp::NetworkFlowEndpoint>
   get_network_flow_endpoints() const;
 
+  /// Return the lowest available capacity for all subscription buffers.
+  /**
+   * For intraprocess communication return the lowest buffer capacity for all subscriptions.
+   * If intraprocess is disabled or no intraprocess subscriptions present, return maximum of size_t.
+   * On failure return 0.
+   * \return lowest buffer capacity for all subscriptions
+   */
+  RCLCPP_PUBLIC
+  size_t
+  lowest_available_ipm_capacity() const;
+
   /// Wait until all published messages are acknowledged or until the specified timeout elapses.
   /**
    * This method waits until all published messages are acknowledged by all matching

rclcpp/include/rclcpp/publisher_options.hpp

@@ -24,6 +24,7 @@
 
 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/detail/rmw_implementation_specific_publisher_payload.hpp"
+#include "rclcpp/intra_process_buffer_type.hpp"
 #include "rclcpp/intra_process_setting.hpp"
 #include "rclcpp/qos.hpp"
 #include "rclcpp/event_handler.hpp"
@@ -40,6 +41,9 @@ struct PublisherOptionsBase
   /// Setting to explicitly set intraprocess communications.
   IntraProcessSetting use_intra_process_comm = IntraProcessSetting::NodeDefault;
 
+  /// Setting the data-type stored in the intraprocess buffer
+  IntraProcessBufferType intra_process_buffer_type = IntraProcessBufferType::SharedPtr;
+
   /// Callbacks for various events related to publishers.
   PublisherEventCallbacks event_callbacks;
 

rclcpp/include/rclcpp/qos_event.hpp

@@ -1,22 +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__QOS_EVENT_HPP_
-#define RCLCPP__QOS_EVENT_HPP_
-
-#warning This header is obsolete, please include rclcpp/event_handler.hpp instead
-
-#include "rclcpp/event_handler.hpp"
-
-#endif  // RCLCPP__QOS_EVENT_HPP_

rclcpp/include/rclcpp/rate.hpp

@@ -19,6 +19,8 @@
 #include <memory>
 #include <thread>
 
+#include "rclcpp/clock.hpp"
+#include "rclcpp/duration.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/utilities.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -31,9 +33,20 @@ class RateBase
 public:
   RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(RateBase)
 
+  RCLCPP_PUBLIC
   virtual ~RateBase() {}
+
+  RCLCPP_PUBLIC
   virtual bool sleep() = 0;
+
+  [[deprecated("use get_type() instead")]]
+  RCLCPP_PUBLIC
   virtual bool is_steady() const = 0;
+
+  RCLCPP_PUBLIC
+  virtual rcl_clock_type_t get_type() const = 0;
+
+  RCLCPP_PUBLIC
   virtual void reset() = 0;
 };
 
@@ -42,14 +55,13 @@ using std::chrono::duration_cast;
 using std::chrono::nanoseconds;
 
 template<class Clock = std::chrono::high_resolution_clock>
-class GenericRate : public RateBase
+class [[deprecated("use rclcpp::Rate class instead of GenericRate")]] GenericRate : public RateBase
 {
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(GenericRate)
 
   explicit GenericRate(double rate)
-  : GenericRate<Clock>(
-      duration_cast<nanoseconds>(duration<double>(1.0 / rate)))
+  : period_(duration_cast<nanoseconds>(duration<double>(1.0 / rate))), last_interval_(Clock::now())
   {}
   explicit GenericRate(std::chrono::nanoseconds period)
   : period_(period), last_interval_(Clock::now())
@@ -87,12 +99,18 @@ public:
     return true;
   }
 
+  [[deprecated("use get_type() instead")]]
   virtual bool
   is_steady() const
   {
     return Clock::is_steady;
   }
 
+  virtual rcl_clock_type_t get_type() const
+  {
+    return Clock::is_steady ? RCL_STEADY_TIME : RCL_SYSTEM_TIME;
+  }
+
   virtual void
   reset()
   {
@@ -112,8 +130,59 @@ private:
   std::chrono::time_point<Clock, ClockDurationNano> last_interval_;
 };
 
-using Rate = GenericRate<std::chrono::system_clock>;
-using WallRate = GenericRate<std::chrono::steady_clock>;
+class Rate : public RateBase
+{
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(Rate)
+
+  RCLCPP_PUBLIC
+  explicit Rate(
+    const double rate,
+    Clock::SharedPtr clock = std::make_shared<Clock>(RCL_SYSTEM_TIME));
+
+  RCLCPP_PUBLIC
+  explicit Rate(
+    const Duration & period,
+    Clock::SharedPtr clock = std::make_shared<Clock>(RCL_SYSTEM_TIME));
+
+  RCLCPP_PUBLIC
+  virtual bool
+  sleep();
+
+  [[deprecated("use get_type() instead")]]
+  RCLCPP_PUBLIC
+  virtual bool
+  is_steady() const;
+
+  RCLCPP_PUBLIC
+  virtual rcl_clock_type_t
+  get_type() const;
+
+  RCLCPP_PUBLIC
+  virtual void
+  reset();
+
+  RCLCPP_PUBLIC
+  std::chrono::nanoseconds
+  period() const;
+
+private:
+  RCLCPP_DISABLE_COPY(Rate)
+
+  Clock::SharedPtr clock_;
+  Duration period_;
+  Time last_interval_;
+};
+
+class WallRate : public Rate
+{
+public:
+  RCLCPP_PUBLIC
+  explicit WallRate(const double rate);
+
+  RCLCPP_PUBLIC
+  explicit WallRate(const Duration & period);
+};
 
 }  // namespace rclcpp
 

rclcpp/include/rclcpp/rclcpp.hpp

@@ -54,6 +54,7 @@
  *   - rclcpp::ParameterValue
  *   - rclcpp::AsyncParametersClient
  *   - rclcpp::SyncParametersClient
+ *   - rclcpp::copy_all_parameter_values()
  *   - rclcpp/parameter.hpp
  *   - rclcpp/parameter_value.hpp
  *   - rclcpp/parameter_client.hpp
@@ -95,6 +96,9 @@
  * - Get the number of publishers or subscribers on a topic:
  *   - rclcpp::Node::count_publishers()
  *   - rclcpp::Node::count_subscribers()
+ * - Get the number of clients or servers on a service:
+ *   - rclcpp::Node::count_clients()
+ *   - rclcpp::Node::count_services()
  *
  * And components related to logging:
  *
@@ -164,6 +168,7 @@
 #include <csignal>
 #include <memory>
 
+#include "rclcpp/copy_all_parameter_values.hpp"
 #include "rclcpp/executors.hpp"
 #include "rclcpp/guard_condition.hpp"
 #include "rclcpp/logging.hpp"

rclcpp/include/rclcpp/service.hpp

@@ -265,15 +265,19 @@ protected:
 
   std::shared_ptr<rcl_node_t> node_handle_;
 
+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_request_callback_ before
+  // service_handle_, so on destruction the service is
+  // destroyed first. Otherwise, the rmw service callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_request_callback_{nullptr};
+  // Declare service_handle_ after callback
   std::shared_ptr<rcl_service_t> service_handle_;
   bool owns_rcl_handle_ = true;
 
   rclcpp::Logger node_logger_;
 
   std::atomic<bool> in_use_by_wait_set_{false};
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_request_callback_{nullptr};
 };
 
 template<typename ServiceT>
@@ -348,7 +352,7 @@ public:
 
       rclcpp::exceptions::throw_from_rcl_error(ret, "could not create service");
     }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_service_callback_added,
       static_cast<const void *>(get_service_handle().get()),
       static_cast<const void *>(&any_callback_));
@@ -383,7 +387,7 @@ public:
     }
 
     service_handle_ = service_handle;
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_service_callback_added,
       static_cast<const void *>(get_service_handle().get()),
       static_cast<const void *>(&any_callback_));
@@ -420,7 +424,7 @@ public:
     // In this case, rcl owns the service handle memory
     service_handle_ = std::shared_ptr<rcl_service_t>(new rcl_service_t);
     service_handle_->impl = service_handle->impl;
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_service_callback_added,
       static_cast<const void *>(get_service_handle().get()),
       static_cast<const void *>(&any_callback_));
@@ -482,6 +486,14 @@ public:
   {
     rcl_ret_t ret = rcl_send_response(get_service_handle().get(), &req_id, &response);
 
+    if (ret == RCL_RET_TIMEOUT) {
+      RCLCPP_WARN(
+        node_logger_.get_child("rclcpp"),
+        "failed to send response to %s (timeout): %s",
+        this->get_service_name(), rcl_get_error_string().str);
+      rcl_reset_error();
+      return;
+    }
     if (ret != RCL_RET_OK) {
       rclcpp::exceptions::throw_from_rcl_error(ret, "failed to send response");
     }

rclcpp/include/rclcpp/strategies/message_pool_memory_strategy.hpp

@@ -15,10 +15,17 @@
 #ifndef RCLCPP__STRATEGIES__MESSAGE_POOL_MEMORY_STRATEGY_HPP_
 #define RCLCPP__STRATEGIES__MESSAGE_POOL_MEMORY_STRATEGY_HPP_
 
+#include <array>
+#include <cstring>
 #include <memory>
+#include <mutex>
+#include <stdexcept>
+#include <type_traits>
 
 #include "rosidl_runtime_cpp/traits.hpp"
 
+#include "rclcpp/logger.hpp"
+#include "rclcpp/logging.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/message_memory_strategy.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -50,13 +57,24 @@ class MessagePoolMemoryStrategy
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(MessagePoolMemoryStrategy)
 
-  /// Default constructor
   MessagePoolMemoryStrategy()
-  : next_array_index_(0)
   {
+    pool_mutex_ = std::make_shared<std::mutex>();
+
+    pool_ = std::shared_ptr<std::array<MessageT *, Size>>(
+      new std::array<MessageT *, Size>,
+      [](std::array<MessageT *, Size> * arr) {
+        for (size_t i = 0; i < Size; ++i) {
+          free((*arr)[i]);
+        }
+        delete arr;
+      });
+
+    free_list_ = std::make_shared<CircularArray<Size>>();
+
     for (size_t i = 0; i < Size; ++i) {
-      pool_[i].msg_ptr_ = std::make_shared<MessageT>();
-      pool_[i].used = false;
+      (*pool_)[i] = static_cast<MessageT *>(malloc(sizeof(MessageT)));
+      free_list_->push_back(i);
     }
   }
 
@@ -68,43 +86,85 @@ public:
    */
   std::shared_ptr<MessageT> borrow_message()
   {
-    size_t current_index = next_array_index_;
-    next_array_index_ = (next_array_index_ + 1) % Size;
-    if (pool_[current_index].used) {
-      throw std::runtime_error("Tried to access message that was still in use! Abort.");
+    std::lock_guard<std::mutex> lock(*pool_mutex_);
+    if (free_list_->size() == 0) {
+      throw std::runtime_error("No more free slots in the pool");
     }
-    pool_[current_index].msg_ptr_->~MessageT();
-    new (pool_[current_index].msg_ptr_.get())MessageT;
 
-    pool_[current_index].used = true;
-    return pool_[current_index].msg_ptr_;
+    size_t current_index = free_list_->pop_front();
+
+    return std::shared_ptr<MessageT>(
+      new((*pool_)[current_index]) MessageT(),
+      [pool = this->pool_, pool_mutex = this->pool_mutex_,
+      free_list = this->free_list_](MessageT * p) {
+        std::lock_guard<std::mutex> lock(*pool_mutex);
+        for (size_t i = 0; i < Size; ++i) {
+          if ((*pool)[i] == p) {
+            p->~MessageT();
+            free_list->push_back(i);
+            break;
+          }
+        }
+      });
   }
 
   /// Return a message to the message pool.
   /**
-   * Manage metadata in the message pool ring buffer to release the message.
+   * This does nothing since the message isn't returned to the pool until the user has dropped
+   * all references.
    * \param[in] msg Shared pointer to the message to return.
    */
   void return_message(std::shared_ptr<MessageT> & msg)
   {
-    for (size_t i = 0; i < Size; ++i) {
-      if (pool_[i].msg_ptr_ == msg) {
-        pool_[i].used = false;
-        return;
-      }
-    }
-    throw std::runtime_error("Unrecognized message ptr in return_message.");
+    (void)msg;
   }
 
 protected:
-  struct PoolMember
+  template<size_t N>
+  class CircularArray
   {
-    std::shared_ptr<MessageT> msg_ptr_;
-    bool used;
+public:
+    void push_back(const size_t v)
+    {
+      if (size_ + 1 > N) {
+        throw std::runtime_error("Tried to push too many items into the array");
+      }
+      array_[(front_ + size_) % N] = v;
+      ++size_;
+    }
+
+    size_t pop_front()
+    {
+      if (size_ < 1) {
+        throw std::runtime_error("Tried to pop item from empty array");
+      }
+
+      size_t val = array_[front_];
+
+      front_ = (front_ + 1) % N;
+      --size_;
+
+      return val;
+    }
+
+    size_t size() const
+    {
+      return size_;
+    }
+
+private:
+    size_t front_ = 0;
+    size_t size_ = 0;
+    std::array<size_t, N> array_;
   };
 
-  std::array<PoolMember, Size> pool_;
-  size_t next_array_index_;
+  // It's very important that these are shared_ptrs, since users of this class might hold a
+  // reference to a pool item longer than the lifetime of the class.  In that scenario, the
+  // shared_ptr ensures that the lifetime of these variables outlives this class, and hence ensures
+  // the custom destructor for each pool item can successfully run.
+  std::shared_ptr<std::mutex> pool_mutex_;
+  std::shared_ptr<std::array<MessageT *, Size>> pool_;
+  std::shared_ptr<CircularArray<Size>> free_list_;
 };
 
 }  // namespace message_pool_memory_strategy

rclcpp/include/rclcpp/subscription.hpp

@@ -104,7 +104,7 @@ public:
 
 private:
   using SubscriptionTopicStatisticsSharedPtr =
-    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>;
+    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>;
 
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(Subscription)
@@ -163,10 +163,6 @@ public:
         throw std::invalid_argument(
                 "intraprocess communication is not allowed with 0 depth qos policy");
       }
-      if (qos_profile.durability() != rclcpp::DurabilityPolicy::Volatile) {
-        throw std::invalid_argument(
-                "intraprocess communication allowed only with volatile durability");
-      }
 
       using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<
         MessageT,
@@ -185,7 +181,7 @@ public:
         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));
-      TRACEPOINT(
+      TRACETOOLS_TRACEPOINT(
         rclcpp_subscription_init,
         static_cast<const void *>(get_subscription_handle().get()),
         static_cast<const void *>(subscription_intra_process_.get()));
@@ -193,7 +189,8 @@ public:
       // Add it to the intra process manager.
       using rclcpp::experimental::IntraProcessManager;
       auto ipm = context->get_sub_context<IntraProcessManager>();
-      uint64_t intra_process_subscription_id = ipm->add_subscription(subscription_intra_process_);
+      uint64_t intra_process_subscription_id = ipm->template add_subscription<
+        ROSMessageType, ROSMessageTypeAllocator>(subscription_intra_process_);
       this->setup_intra_process(intra_process_subscription_id, ipm);
     }
 
@@ -201,11 +198,11 @@ public:
       this->subscription_topic_statistics_ = std::move(subscription_topic_statistics);
     }
 
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_subscription_init,
       static_cast<const void *>(get_subscription_handle().get()),
       static_cast<const void *>(this));
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_subscription_callback_added,
       static_cast<const void *>(this),
       static_cast<const void *>(&any_callback_));
@@ -316,7 +313,7 @@ public:
     if (subscription_topic_statistics_) {
       const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
       const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
     }
   }
 
@@ -325,8 +322,20 @@ public:
     const std::shared_ptr<rclcpp::SerializedMessage> & serialized_message,
     const rclcpp::MessageInfo & message_info) override
   {
-    // TODO(wjwwood): enable topic statistics for serialized messages
+    std::chrono::time_point<std::chrono::system_clock> 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();
+    }
+
     any_callback_.dispatch(serialized_message, message_info);
+
+    if (subscription_topic_statistics_) {
+      const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
+      const auto time = rclcpp::Time(nanos.time_since_epoch().count());
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
+    }
   }
 
   void
@@ -357,7 +366,7 @@ public:
     if (subscription_topic_statistics_) {
       const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
       const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
     }
   }
 

rclcpp/include/rclcpp/subscription_base.hpp

@@ -260,13 +260,13 @@ public:
   bool
   is_serialized() const;
 
-  /// Return the type of the subscription.
+  /// Return the delivered message kind.
   /**
    * \return `DeliveredMessageKind`, which adjusts how messages are received and delivered.
    */
   RCLCPP_PUBLIC
   DeliveredMessageKind
-  get_subscription_type() const;
+  get_delivered_message_kind() const;
 
   /// Get matching publisher count.
   /** \return The number of publishers on this topic. */
@@ -645,6 +645,14 @@ protected:
   rclcpp::node_interfaces::NodeBaseInterface * const node_base_;
 
   std::shared_ptr<rcl_node_t> node_handle_;
+
+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_message_callback_ before
+  // subscription_handle_, so on destruction the subscription is
+  // destroyed first. Otherwise, the rmw subscription callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_message_callback_{nullptr};
+  // Declare subscription_handle_ after callback
   std::shared_ptr<rcl_subscription_t> subscription_handle_;
   std::shared_ptr<rcl_subscription_t> intra_process_subscription_handle_;
   rclcpp::Logger node_logger_;
@@ -663,15 +671,12 @@ private:
   RCLCPP_DISABLE_COPY(SubscriptionBase)
 
   rosidl_message_type_support_t type_support_;
-  DeliveredMessageKind delivered_message_type_;
+  DeliveredMessageKind delivered_message_kind_;
 
   std::atomic<bool> subscription_in_use_by_wait_set_{false};
   std::atomic<bool> intra_process_subscription_waitable_in_use_by_wait_set_{false};
   std::unordered_map<rclcpp::EventHandlerBase *,
     std::atomic<bool>> qos_events_in_use_by_wait_set_;
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_message_callback_{nullptr};
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/subscription_factory.hpp

@@ -75,15 +75,14 @@ template<
   typename CallbackT,
   typename AllocatorT,
   typename SubscriptionT = rclcpp::Subscription<MessageT, AllocatorT>,
-  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType
+  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType
 >
 SubscriptionFactory
 create_subscription_factory(
   CallbackT && callback,
   const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options,
   typename MessageMemoryStrategyT::SharedPtr msg_mem_strat,
-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
   subscription_topic_stats = nullptr
 )
 {

rclcpp/include/rclcpp/subscription_options.hpp

@@ -77,6 +77,10 @@ struct SubscriptionOptionsBase
     // 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.
+    rclcpp::QoS qos = SystemDefaultsQoS();
   };
 
   TopicStatisticsOptions topic_stats_options;

rclcpp/include/rclcpp/time.hpp

@@ -189,7 +189,7 @@ public:
    */
   RCLCPP_PUBLIC
   static Time
-  max();
+  max(rcl_clock_type_t clock_type = RCL_SYSTEM_TIME);  // NOLINT
 
   /// Get the seconds since epoch
   /**

rclcpp/include/rclcpp/timer.hpp

@@ -53,12 +53,17 @@ public:
    * \param clock A clock to use for time and sleeping
    * \param period The interval at which the timer fires
    * \param context node context
+   * \param autostart timer state on initialization
+   *
+   * In order to activate a timer that is not started on initialization,
+   * user should call the reset() method.
    */
   RCLCPP_PUBLIC
   explicit TimerBase(
     Clock::SharedPtr clock,
     std::chrono::nanoseconds period,
-    rclcpp::Context::SharedPtr context);
+    rclcpp::Context::SharedPtr context,
+    bool autostart = true);
 
   /// TimerBase destructor
   RCLCPP_PUBLIC
@@ -216,21 +221,22 @@ public:
    * \param[in] period The interval at which the timer fires.
    * \param[in] callback User-specified callback function.
    * \param[in] context custom context to be used.
+   * \param autostart timer state on initialization
    */
   explicit GenericTimer(
     Clock::SharedPtr clock, std::chrono::nanoseconds period, FunctorT && callback,
-    rclcpp::Context::SharedPtr context
+    rclcpp::Context::SharedPtr context, bool autostart = true
   )
-  : TimerBase(clock, period, context), callback_(std::forward<FunctorT>(callback))
+  : TimerBase(clock, period, context, autostart), callback_(std::forward<FunctorT>(callback))
   {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_timer_callback_added,
       static_cast<const void *>(get_timer_handle().get()),
       reinterpret_cast<const void *>(&callback_));
 #ifndef TRACETOOLS_DISABLED
-    if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+    if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
       char * symbol = tracetools::get_symbol(callback_);
-      DO_TRACEPOINT(
+      TRACETOOLS_DO_TRACEPOINT(
         rclcpp_callback_register,
         reinterpret_cast<const void *>(&callback_),
         symbol);
@@ -269,9 +275,9 @@ public:
   void
   execute_callback() override
   {
-    TRACEPOINT(callback_start, reinterpret_cast<const void *>(&callback_), false);
+    TRACETOOLS_TRACEPOINT(callback_start, reinterpret_cast<const void *>(&callback_), false);
     execute_callback_delegate<>();
-    TRACEPOINT(callback_end, reinterpret_cast<const void *>(&callback_));
+    TRACETOOLS_TRACEPOINT(callback_end, reinterpret_cast<const void *>(&callback_));
   }
 
   // void specialization
@@ -330,13 +336,15 @@ public:
    * \param period The interval at which the timer fires
    * \param callback The callback function to execute every interval
    * \param context node context
+   * \param autostart timer state on initialization
    */
   WallTimer(
     std::chrono::nanoseconds period,
     FunctorT && callback,
-    rclcpp::Context::SharedPtr context)
+    rclcpp::Context::SharedPtr context,
+    bool autostart = true)
   : GenericTimer<FunctorT>(
-      std::make_shared<Clock>(RCL_STEADY_TIME), period, std::move(callback), context)
+      std::make_shared<Clock>(RCL_STEADY_TIME), period, std::move(callback), context, autostart)
   {}
 
 protected:

rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp

@@ -48,21 +48,12 @@ 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.
- *
- * \tparam CallbackMessageT the subscribed message type
- */
-template<typename CallbackMessageT>
+  */
 class SubscriptionTopicStatistics
 {
-  using TopicStatsCollector =
-    libstatistics_collector::topic_statistics_collector::TopicStatisticsCollector<
-    CallbackMessageT>;
-  using ReceivedMessageAge =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessageAgeCollector<
-    CallbackMessageT>;
-  using ReceivedMessagePeriod =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessagePeriodCollector<
-    CallbackMessageT>;
+  using TopicStatsCollector = libstatistics_collector::TopicStatisticsCollector;
+  using ReceivedMessageAge = libstatistics_collector::ReceivedMessageAgeCollector;
+  using ReceivedMessagePeriod = libstatistics_collector::ReceivedMessagePeriodCollector;
 
 public:
   /// Construct a SubscriptionTopicStatistics object.
@@ -101,16 +92,16 @@ public:
   /**
    * This method acquires a lock to prevent race conditions to collectors list.
    *
-   * \param received_message the message received by the subscription
+   * \param message_info the message info corresponding to the received message
    * \param now_nanoseconds current time in nanoseconds
    */
   virtual void handle_message(
-    const CallbackMessageT & received_message,
+    const rmw_message_info_t & message_info,
     const rclcpp::Time now_nanoseconds) const
   {
     std::lock_guard<std::mutex> lock(mutex_);
     for (const auto & collector : subscriber_statistics_collectors_) {
-      collector->OnMessageReceived(received_message, now_nanoseconds.nanoseconds());
+      collector->OnMessageReceived(message_info, now_nanoseconds.nanoseconds());
     }
   }
 

rclcpp/include/rclcpp/typesupport_helpers.hpp

@@ -22,6 +22,7 @@
 
 #include "rcpputils/shared_library.hpp"
 #include "rosidl_runtime_cpp/message_type_support_decl.hpp"
+#include "rosidl_runtime_cpp/service_type_support_decl.hpp"
 
 #include "rclcpp/visibility_control.hpp"
 
@@ -40,11 +41,15 @@ get_typesupport_library(const std::string & type, const std::string & typesuppor
 /// Extract the type support handle from the library.
 /**
  * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \deprecated Use get_message_typesupport_handle() instead
+ *
  * \param[in] type The topic type, e.g. "std_msgs/msg/String"
  * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
  * \param[in] library The shared type support library
  * \return A type support handle
  */
+[[deprecated("Use `get_message_typesupport_handle` instead")]]
 RCLCPP_PUBLIC
 const rosidl_message_type_support_t *
 get_typesupport_handle(
@@ -52,6 +57,40 @@ get_typesupport_handle(
   const std::string & typesupport_identifier,
   rcpputils::SharedLibrary & library);
 
+/// Extract the message type support handle from the library.
+/**
+ * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \param[in] type The topic type, e.g. "std_msgs/msg/String"
+ * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
+ * \param[in] library The shared type support library
+ * \throws std::runtime_error if the symbol of type not found in the library.
+ * \return A message type support handle
+ */
+RCLCPP_PUBLIC
+const rosidl_message_type_support_t *
+get_message_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library);
+
+/// Extract the service type support handle from the library.
+/**
+ * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \param[in] type The service type, e.g. "std_srvs/srv/Empty"
+ * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
+ * \param[in] library The shared type support library
+ * \throws std::runtime_error if the symbol of type not found in the library.
+ * \return A service type support handle
+ */
+RCLCPP_PUBLIC
+const rosidl_service_type_support_t *
+get_service_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library);
+
 }  // namespace rclcpp
 
 #endif  // RCLCPP__TYPESUPPORT_HELPERS_HPP_

rclcpp/include/rclcpp/wait_set_policies/detail/storage_policy_common.hpp

@@ -104,7 +104,7 @@ protected:
       // TODO(wjwwood): support custom allocator, maybe restrict to polymorphic allocator
       rcl_get_default_allocator());
     if (RCL_RET_OK != ret) {
-      rclcpp::exceptions::throw_from_rcl_error(ret, "Failed to create wait set");
+      rclcpp::exceptions::throw_from_rcl_error(ret);
     }
 
     // (Re)build the wait set for the first time.
@@ -192,7 +192,8 @@ protected:
       size_t services_from_waitables = 0;
       size_t events_from_waitables = 0;
       for (const auto & waitable_entry : waitables) {
-        if (!waitable_entry.waitable) {
+        auto waitable_ptr_pair = get_raw_pointer_from_smart_pointer(waitable_entry.waitable);
+        if (nullptr == waitable_ptr_pair.second) {
           // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
           if (HasStrongOwnership) {
             // This will not happen in fixed sized storage, as it holds
@@ -203,13 +204,13 @@ protected:
           needs_pruning_ = true;
           continue;
         }
-        const auto & waitable = waitable_entry.waitable;
-        subscriptions_from_waitables += waitable->get_number_of_ready_subscriptions();
-        guard_conditions_from_waitables += waitable->get_number_of_ready_guard_conditions();
-        timers_from_waitables += waitable->get_number_of_ready_timers();
-        clients_from_waitables += waitable->get_number_of_ready_clients();
-        services_from_waitables += waitable->get_number_of_ready_services();
-        events_from_waitables += waitable->get_number_of_ready_events();
+        rclcpp::Waitable & waitable = *waitable_ptr_pair.second;
+        subscriptions_from_waitables += waitable.get_number_of_ready_subscriptions();
+        guard_conditions_from_waitables += waitable.get_number_of_ready_guard_conditions();
+        timers_from_waitables += waitable.get_number_of_ready_timers();
+        clients_from_waitables += waitable.get_number_of_ready_clients();
+        services_from_waitables += waitable.get_number_of_ready_services();
+        events_from_waitables += waitable.get_number_of_ready_events();
       }
       rcl_ret_t ret = rcl_wait_set_resize(
         &rcl_wait_set_,
@@ -221,7 +222,7 @@ protected:
         events_from_waitables
       );
       if (RCL_RET_OK != ret) {
-        rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't resize the wait set");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
       was_resized = true;
       // Assumption: the calling code ensures this function is not called
@@ -237,13 +238,15 @@ protected:
     if (!was_resized) {
       rcl_ret_t ret = rcl_wait_set_clear(&rcl_wait_set_);
       if (RCL_RET_OK != ret) {
-        rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't clear the wait set");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     }
 
     // Add subscriptions.
     for (const auto & subscription_entry : subscriptions) {
-      if (!subscription_entry.subscription) {
+      auto subscription_ptr_pair =
+        get_raw_pointer_from_smart_pointer(subscription_entry.subscription);
+      if (nullptr == subscription_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
           // This will not happen in fixed sized storage, as it holds
@@ -254,13 +257,12 @@ protected:
         needs_pruning_ = true;
         continue;
       }
-
       rcl_ret_t ret = rcl_wait_set_add_subscription(
         &rcl_wait_set_,
-        subscription_entry.subscription->get_subscription_handle().get(),
+        subscription_ptr_pair.second->get_subscription_handle().get(),
         nullptr);
       if (RCL_RET_OK != ret) {
-        rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't fill wait set");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     }
 
@@ -269,7 +271,8 @@ protected:
       [this](const auto & inner_guard_conditions)
       {
         for (const auto & guard_condition : inner_guard_conditions) {
-          if (!guard_condition) {
+          auto guard_condition_ptr_pair = get_raw_pointer_from_smart_pointer(guard_condition);
+          if (nullptr == guard_condition_ptr_pair.second) {
             // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
             if (HasStrongOwnership) {
               // This will not happen in fixed sized storage, as it holds
@@ -282,10 +285,10 @@ protected:
           }
           rcl_ret_t ret = rcl_wait_set_add_guard_condition(
             &rcl_wait_set_,
-            &guard_condition->get_rcl_guard_condition(),
+            &guard_condition_ptr_pair.second->get_rcl_guard_condition(),
             nullptr);
           if (RCL_RET_OK != ret) {
-            rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't fill wait set");
+            rclcpp::exceptions::throw_from_rcl_error(ret);
           }
         }
       };
@@ -298,7 +301,8 @@ protected:
 
     // Add timers.
     for (const auto & timer : timers) {
-      if (!timer) {
+      auto timer_ptr_pair = get_raw_pointer_from_smart_pointer(timer);
+      if (nullptr == timer_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
           // This will not happen in fixed sized storage, as it holds
@@ -311,16 +315,17 @@ protected:
       }
       rcl_ret_t ret = rcl_wait_set_add_timer(
         &rcl_wait_set_,
-        timer->get_timer_handle().get(),
+        timer_ptr_pair.second->get_timer_handle().get(),
         nullptr);
       if (RCL_RET_OK != ret) {
-        rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't fill wait set");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     }
 
     // Add clients.
     for (const auto & client : clients) {
-      if (!client) {
+      auto client_ptr_pair = get_raw_pointer_from_smart_pointer(client);
+      if (nullptr == client_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
           // This will not happen in fixed sized storage, as it holds
@@ -333,17 +338,17 @@ protected:
       }
       rcl_ret_t ret = rcl_wait_set_add_client(
         &rcl_wait_set_,
-        client->get_client_handle().get(),
+        client_ptr_pair.second->get_client_handle().get(),
         nullptr);
       if (RCL_RET_OK != ret) {
         rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     }
 
-
     // Add services.
     for (const auto & service : services) {
-      if (!service) {
+      auto service_ptr_pair = get_raw_pointer_from_smart_pointer(service);
+      if (nullptr == service_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
           // This will not happen in fixed sized storage, as it holds
@@ -356,16 +361,17 @@ protected:
       }
       rcl_ret_t ret = rcl_wait_set_add_service(
         &rcl_wait_set_,
-        service->get_service_handle().get(),
+        service_ptr_pair.second->get_service_handle().get(),
         nullptr);
       if (RCL_RET_OK != ret) {
-        rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't fill wait set");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     }
 
     // Add waitables.
     for (auto & waitable_entry : waitables) {
-      if (!waitable_entry.waitable) {
+      auto waitable_ptr_pair = get_raw_pointer_from_smart_pointer(waitable_entry.waitable);
+      if (nullptr == waitable_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
           // This will not happen in fixed sized storage, as it holds
@@ -376,7 +382,8 @@ protected:
         needs_pruning_ = true;
         continue;
       }
-      waitable_entry.waitable->add_to_wait_set(&rcl_wait_set_);
+      rclcpp::Waitable & waitable = *waitable_ptr_pair.second;
+      waitable.add_to_wait_set(&rcl_wait_set_);
     }
   }
 

rclcpp/include/rclcpp/wait_set_policies/dynamic_storage.hpp

@@ -204,19 +204,15 @@ public:
   void
   storage_rebuild_rcl_wait_set(const ArrayOfExtraGuardConditions & extra_guard_conditions)
   {
-    this->storage_acquire_ownerships();
-
     this->storage_rebuild_rcl_wait_set_with_sets(
-      shared_subscriptions_,
-      shared_guard_conditions_,
+      subscriptions_,
+      guard_conditions_,
       extra_guard_conditions,
-      shared_timers_,
-      shared_clients_,
-      shared_services_,
-      shared_waitables_
+      timers_,
+      clients_,
+      services_,
+      waitables_
     );
-
-    this->storage_release_ownerships();
   }
 
   template<class EntityT, class SequenceOfEntitiesT>
@@ -411,7 +407,6 @@ public:
         }
       };
     // Lock all the weak pointers and hold them until released.
-    lock_all(subscriptions_, shared_subscriptions_);
     lock_all(guard_conditions_, shared_guard_conditions_);
     lock_all(timers_, shared_timers_);
     lock_all(clients_, shared_clients_);
@@ -443,7 +438,6 @@ public:
           shared_ptr.reset();
         }
       };
-    reset_all(shared_subscriptions_);
     reset_all(shared_guard_conditions_);
     reset_all(shared_timers_);
     reset_all(shared_clients_);

rclcpp/include/rclcpp/wait_set_policies/sequential_synchronization.hpp

@@ -290,7 +290,7 @@ protected:
         return create_wait_result(WaitResultKind::Empty);
       } else {
         // Some other error case, throw.
-        rclcpp::exceptions::throw_from_rcl_error(ret, "rcl_wait() failed");
+        rclcpp::exceptions::throw_from_rcl_error(ret);
       }
     } while (should_loop());
 

rclcpp/package.xml

@@ -2,7 +2,7 @@
 <?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
 <package format="2">
   <name>rclcpp</name>
-  <version>20.0.0</version>
+  <version>27.0.0</version>
   <description>The ROS client library in C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>
@@ -22,6 +22,7 @@
   <build_depend>builtin_interfaces</build_depend>
   <build_depend>rcl_interfaces</build_depend>
   <build_depend>rosgraph_msgs</build_depend>
+  <build_depend>rosidl_runtime_c</build_depend>
   <build_depend>rosidl_runtime_cpp</build_depend>
   <build_depend>rosidl_typesupport_c</build_depend>
   <build_depend>rosidl_typesupport_cpp</build_depend>
@@ -29,12 +30,14 @@
   <build_export_depend>builtin_interfaces</build_export_depend>
   <build_export_depend>rcl_interfaces</build_export_depend>
   <build_export_depend>rosgraph_msgs</build_export_depend>
+  <build_export_depend>rosidl_runtime_c</build_export_depend>
   <build_export_depend>rosidl_runtime_cpp</build_export_depend>
   <build_export_depend>rosidl_typesupport_c</build_export_depend>
   <build_export_depend>rosidl_typesupport_cpp</build_export_depend>
 
   <depend>libstatistics_collector</depend>
   <depend>rcl</depend>
+  <depend>rcl_logging_interface</depend>
   <depend>rcl_yaml_param_parser</depend>
   <depend>rcpputils</depend>
   <depend>rcutils</depend>

rclcpp/src/rclcpp/any_executable.cpp

@@ -16,16 +16,13 @@
 
 using rclcpp::AnyExecutable;
 
-RCLCPP_PUBLIC
 AnyExecutable::AnyExecutable()
 : subscription(nullptr),
   timer(nullptr),
   service(nullptr),
   client(nullptr),
-  waitable(nullptr),
   callback_group(nullptr),
-  node_base(nullptr),
-  data(nullptr)
+  node_base(nullptr)
 {}
 
 AnyExecutable::~AnyExecutable()

rclcpp/src/rclcpp/callback_group.cpp

@@ -31,12 +31,12 @@ using rclcpp::CallbackGroupType;
 
 CallbackGroup::CallbackGroup(
   CallbackGroupType group_type,
-  std::function<rclcpp::Context::SharedPtr(void)> get_context,
+  rclcpp::Context::WeakPtr context,
   bool automatically_add_to_executor_with_node)
 : type_(group_type), associated_with_executor_(false),
   can_be_taken_from_(true),
   automatically_add_to_executor_with_node_(automatically_add_to_executor_with_node),
-  get_context_(get_context)
+  context_(context)
 {}
 
 CallbackGroup::~CallbackGroup()
@@ -66,7 +66,6 @@ CallbackGroup::size() const
     timer_ptrs_.size() +
     waitable_ptrs_.size();
 }
-
 void CallbackGroup::collect_all_ptrs(
   std::function<void(const rclcpp::SubscriptionBase::SharedPtr &)> sub_func,
   std::function<void(const rclcpp::ServiceBase::SharedPtr &)> service_func,
@@ -124,40 +123,12 @@ CallbackGroup::automatically_add_to_executor_with_node() const
   return automatically_add_to_executor_with_node_;
 }
 
-// \TODO(mjcarroll) Deprecated, remove on tock
-rclcpp::GuardCondition::SharedPtr
-CallbackGroup::get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr)
-{
-  std::lock_guard<std::recursive_mutex> lock(notify_guard_condition_mutex_);
-  if (notify_guard_condition_ && context_ptr != notify_guard_condition_->get_context()) {
-    if (associated_with_executor_) {
-      trigger_notify_guard_condition();
-    }
-    notify_guard_condition_ = nullptr;
-  }
-
-  if (!notify_guard_condition_) {
-    notify_guard_condition_ = std::make_shared<rclcpp::GuardCondition>(context_ptr);
-  }
-
-  return notify_guard_condition_;
-}
-
 rclcpp::GuardCondition::SharedPtr
 CallbackGroup::get_notify_guard_condition()
 {
   std::lock_guard<std::recursive_mutex> lock(notify_guard_condition_mutex_);
-  if (!this->get_context_) {
-    throw std::runtime_error("Callback group was created without context and not passed context");
-  }
-  auto context_ptr = this->get_context_();
+  rclcpp::Context::SharedPtr context_ptr = context_.lock();
   if (context_ptr && context_ptr->is_valid()) {
-    if (notify_guard_condition_ && context_ptr != notify_guard_condition_->get_context()) {
-      if (associated_with_executor_) {
-        trigger_notify_guard_condition();
-      }
-      notify_guard_condition_ = nullptr;
-    }
     if (!notify_guard_condition_) {
       notify_guard_condition_ = std::make_shared<rclcpp::GuardCondition>(context_ptr);
     }

rclcpp/src/rclcpp/client.cpp

@@ -125,7 +125,6 @@ bool
 ClientBase::wait_for_service_nanoseconds(std::chrono::nanoseconds timeout)
 {
   auto start = std::chrono::steady_clock::now();
-  // make an event to reuse, rather than create a new one each time
   auto node_ptr = node_graph_.lock();
   if (!node_ptr) {
     throw InvalidNodeError();
@@ -138,6 +137,7 @@ ClientBase::wait_for_service_nanoseconds(std::chrono::nanoseconds timeout)
     // check was non-blocking, return immediately
     return false;
   }
+  // make an event to reuse, rather than create a new one each time
   auto event = node_ptr->get_graph_event();
   // update the time even on the first loop to account for time spent in the first call
   // to this->server_is_ready()

rclcpp/src/rclcpp/context.cpp

@@ -462,7 +462,7 @@ template<Context::ShutdownType shutdown_type>
 std::vector<rclcpp::Context::ShutdownCallback>
 Context::get_shutdown_callback() const
 {
-  const auto get_callback_vector = [this](auto & mutex, auto & callback_set) {
+  const auto get_callback_vector = [](auto & mutex, auto & callback_set) {
       const std::lock_guard<std::mutex> lock(mutex);
       std::vector<rclcpp::Context::ShutdownCallback> callbacks;
       for (auto & callback : callback_set) {
@@ -496,7 +496,7 @@ Context::sleep_for(const std::chrono::nanoseconds & nanoseconds)
       std::unique_lock<std::mutex> lock(interrupt_mutex_);
       auto start = std::chrono::steady_clock::now();
       // this will release the lock while waiting
-      interrupt_condition_variable_.wait_for(lock, nanoseconds);
+      interrupt_condition_variable_.wait_for(lock, time_left);
       time_left -= std::chrono::steady_clock::now() - start;
     }
   } while (time_left > std::chrono::nanoseconds::zero() && this->is_valid());

rclcpp/src/rclcpp/detail/resolve_intra_process_buffer_type.cpp

@@ -0,0 +1,37 @@
+// Copyright 2023 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.
+
+#include <rclcpp/detail/resolve_intra_process_buffer_type.hpp>
+
+namespace rclcpp
+{
+
+namespace detail
+{
+rclcpp::IntraProcessBufferType
+resolve_intra_process_buffer_type(
+  const rclcpp::IntraProcessBufferType buffer_type)
+{
+  if (buffer_type == IntraProcessBufferType::CallbackDefault) {
+    throw std::invalid_argument(
+            "IntraProcessBufferType::CallbackDefault is not allowed "
+            "when there is no callback function");
+  }
+
+  return buffer_type;
+}
+
+}  // namespace detail
+
+}  // namespace rclcpp

rclcpp/src/rclcpp/event_handler.cpp

@@ -39,13 +39,6 @@ UnsupportedEventTypeException::UnsupportedEventTypeException(
 
 EventHandlerBase::~EventHandlerBase()
 {
-  // Since the rmw event listener holds a reference to
-  // this callback, we need to clear it on destruction of this class.
-  // This clearing is not needed for other rclcpp entities like pub/subs, since
-  // they do own the underlying rmw entities, which are destroyed
-  // on their rclcpp destructors, thus no risk of dangling pointers.
-  clear_on_ready_callback();
-
   if (rcl_event_fini(&event_handle_) != RCL_RET_OK) {
     RCUTILS_LOG_ERROR_NAMED(
       "rclcpp",

rclcpp/src/rclcpp/executor.cpp

@@ -13,8 +13,6 @@
 // limitations under the License.
 
 #include <algorithm>
-#include <chrono>
-#include <iterator>
 #include <memory>
 #include <map>
 #include <string>
@@ -24,14 +22,13 @@
 
 #include "rcl/allocator.h"
 #include "rcl/error_handling.h"
-#include "rclcpp/executors/executor_notify_waitable.hpp"
-#include "rclcpp/subscription_wait_set_mask.hpp"
 #include "rcpputils/scope_exit.hpp"
 
 #include "rclcpp/dynamic_typesupport/dynamic_message.hpp"
 #include "rclcpp/exceptions.hpp"
 #include "rclcpp/executor.hpp"
 #include "rclcpp/guard_condition.hpp"
+#include "rclcpp/memory_strategy.hpp"
 #include "rclcpp/node.hpp"
 #include "rclcpp/utilities.hpp"
 
@@ -41,24 +38,16 @@
 
 using namespace std::chrono_literals;
 
+using rclcpp::exceptions::throw_from_rcl_error;
 using rclcpp::Executor;
-class rclcpp::ExecutorImplementation {};
 
-/// Mask to indicate to the waitset to only add the subscription.
-/// The events and intraprocess waitable are already added via the callback group.
-static constexpr rclcpp::SubscriptionWaitSetMask kDefaultSubscriptionMask = {true, false, false};
+class rclcpp::ExecutorImplementation {};
 
 Executor::Executor(const rclcpp::ExecutorOptions & options)
 : spinning(false),
   interrupt_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
   shutdown_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
-  notify_waitable_(std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(
-      [this]() {
-        this->entities_need_rebuild_.store(true);
-      })),
-  collector_(notify_waitable_),
-  wait_set_({}, {}, {}, {}, {}, {}, options.context),
-  current_notify_waitable_(notify_waitable_),
+  memory_strategy_(options.memory_strategy),
   impl_(std::make_unique<rclcpp::ExecutorImplementation>())
 {
   // Store the context for later use.
@@ -72,56 +61,74 @@ Executor::Executor(const rclcpp::ExecutorOptions & options)
       }
     });
 
-  notify_waitable_->add_guard_condition(interrupt_guard_condition_);
-  notify_waitable_->add_guard_condition(shutdown_guard_condition_);
+  // The number of guard conditions is always at least 2: 1 for the ctrl-c guard cond,
+  // and one for the executor's guard cond (interrupt_guard_condition_)
+  memory_strategy_->add_guard_condition(*shutdown_guard_condition_.get());
+
+  // Put the executor's guard condition in
+  memory_strategy_->add_guard_condition(*interrupt_guard_condition_.get());
+  rcl_allocator_t allocator = memory_strategy_->get_allocator();
+
+  rcl_ret_t ret = rcl_wait_set_init(
+    &wait_set_,
+    0, 2, 0, 0, 0, 0,
+    context_->get_rcl_context().get(),
+    allocator);
+  if (RCL_RET_OK != ret) {
+    RCUTILS_LOG_ERROR_NAMED(
+      "rclcpp",
+      "failed to create wait set: %s", rcl_get_error_string().str);
+    rcl_reset_error();
+    throw_from_rcl_error(ret, "Failed to create wait set in Executor constructor");
+  }
 }
 
 Executor::~Executor()
 {
-  std::lock_guard<std::mutex> guard(mutex_);
-
-  notify_waitable_->remove_guard_condition(interrupt_guard_condition_);
-  notify_waitable_->remove_guard_condition(shutdown_guard_condition_);
-
-  std::cout << "Executor::~Executor" << std::endl;
-
-  current_collection_.timers.update(
-    {}, {},
-    [this](auto timer) {
-      std::cout << "remove_timer(" << timer.get() << ")" << std::endl;
-      wait_set_.remove_timer(timer);
+  // Disassociate all callback groups.
+  for (auto & pair : weak_groups_to_nodes_) {
+    auto group = pair.first.lock();
+    if (group) {
+      std::atomic_bool & has_executor = group->get_associated_with_executor_atomic();
+      has_executor.store(false);
+    }
+  }
+  // Disassociate all nodes.
+  std::for_each(
+    weak_nodes_.begin(), weak_nodes_.end(), []
+      (rclcpp::node_interfaces::NodeBaseInterface::WeakPtr weak_node_ptr) {
+      auto shared_node_ptr = weak_node_ptr.lock();
+      if (shared_node_ptr) {
+        std::atomic_bool & has_executor = shared_node_ptr->get_associated_with_executor_atomic();
+        has_executor.store(false);
+      }
     });
+  weak_nodes_.clear();
+  weak_groups_associated_with_executor_to_nodes_.clear();
+  weak_groups_to_nodes_associated_with_executor_.clear();
+  weak_groups_to_nodes_.clear();
+  for (const auto & pair : weak_groups_to_guard_conditions_) {
+    auto guard_condition = pair.second;
+    memory_strategy_->remove_guard_condition(guard_condition);
+  }
+  weak_groups_to_guard_conditions_.clear();
 
-  current_collection_.subscriptions.update(
-    {}, {},
-    [this](auto subscription) {
-      std::cout << "remove_subscription(" << subscription.get() << ")" << std::endl;
-      wait_set_.remove_subscription(subscription, kDefaultSubscriptionMask);
-    });
+  for (const auto & pair : weak_nodes_to_guard_conditions_) {
+    auto guard_condition = pair.second;
+    memory_strategy_->remove_guard_condition(guard_condition);
+  }
+  weak_nodes_to_guard_conditions_.clear();
 
-  current_collection_.clients.update(
-    {}, {},
-    [this](auto client) {
-      std::cout << "remove_client(" << client.get() << ")" << std::endl;
-      wait_set_.remove_client(client);});
-
-  current_collection_.services.update(
-    {}, {},
-    [this](auto service) {
-      std::cout << "remove_service(" << service.get() << ")" << std::endl;
-      wait_set_.remove_service(service);});
-
-  current_collection_.guard_conditions.update(
-    {}, {},
-    [this](auto guard_condition) {
-      std::cout << "remove_guard_condition(" << guard_condition.get() << ")" << std::endl;
-        wait_set_.remove_guard_condition(guard_condition);});
-
-  current_collection_.waitables.update(
-    {}, {},
-    [this](auto waitable) {
-      std::cout << "remove_waitable(" << waitable.get() << ")" << std::endl;
-      wait_set_.remove_waitable(waitable);});
+  // Finalize the wait set.
+  if (rcl_wait_set_fini(&wait_set_) != RCL_RET_OK) {
+    RCUTILS_LOG_ERROR_NAMED(
+      "rclcpp",
+      "failed to destroy wait set: %s", rcl_get_error_string().str);
+    rcl_reset_error();
+  }
+  // Remove and release the sigint guard condition
+  memory_strategy_->remove_guard_condition(shutdown_guard_condition_.get());
+  memory_strategy_->remove_guard_condition(interrupt_guard_condition_.get());
 
   // Remove shutdown callback handle registered to Context
   if (!context_->remove_on_shutdown_callback(shutdown_callback_handle_)) {
@@ -135,39 +142,95 @@ Executor::~Executor()
 std::vector<rclcpp::CallbackGroup::WeakPtr>
 Executor::get_all_callback_groups()
 {
-  this->collector_.update_collections();
-  return this->collector_.get_all_callback_groups();
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  std::lock_guard<std::mutex> guard{mutex_};
+  for (const auto & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  for (auto const & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
 }
 
 std::vector<rclcpp::CallbackGroup::WeakPtr>
 Executor::get_manually_added_callback_groups()
 {
-  this->collector_.update_collections();
-  return this->collector_.get_manually_added_callback_groups();
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  std::lock_guard<std::mutex> guard{mutex_};
+  for (auto const & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
 }
 
 std::vector<rclcpp::CallbackGroup::WeakPtr>
 Executor::get_automatically_added_callback_groups_from_nodes()
 {
-  this->collector_.update_collections();
-  return this->collector_.get_automatically_added_callback_groups();
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  std::lock_guard<std::mutex> guard{mutex_};
+  for (auto const & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
 }
 
 void
-Executor::add_callback_group(
+Executor::add_callback_groups_from_nodes_associated_to_executor()
+{
+  for (auto & weak_node : weak_nodes_) {
+    auto node = weak_node.lock();
+    if (node) {
+      node->for_each_callback_group(
+        [this, node](rclcpp::CallbackGroup::SharedPtr shared_group_ptr)
+        {
+          if (
+            shared_group_ptr->automatically_add_to_executor_with_node() &&
+            !shared_group_ptr->get_associated_with_executor_atomic().load())
+          {
+            this->add_callback_group_to_map(
+              shared_group_ptr,
+              node,
+              weak_groups_to_nodes_associated_with_executor_,
+              true);
+          }
+        });
+    }
+  }
+}
+
+void
+Executor::add_callback_group_to_map(
   rclcpp::CallbackGroup::SharedPtr group_ptr,
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes,
   bool notify)
 {
-  (void) node_ptr;
-  this->collector_.add_callback_group(group_ptr);
+  // If the callback_group already has an executor
+  std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
+  if (has_executor.exchange(true)) {
+    throw std::runtime_error("Callback group has already been added to an executor.");
+  }
 
-  if (!spinning.load()) {
-    std::lock_guard<std::mutex> guard(mutex_);
-    this->collect_entities();
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
+  auto insert_info =
+    weak_groups_to_nodes.insert(std::make_pair(weak_group_ptr, node_ptr));
+  bool was_inserted = insert_info.second;
+  if (!was_inserted) {
+    throw std::runtime_error("Callback group was already added to executor.");
+  }
+  // Also add to the map that contains all callback groups
+  weak_groups_to_nodes_.insert(std::make_pair(weak_group_ptr, node_ptr));
+
+  if (node_ptr->get_context()->is_valid()) {
+    auto callback_group_guard_condition = group_ptr->get_notify_guard_condition();
+    weak_groups_to_guard_conditions_[weak_group_ptr] = callback_group_guard_condition.get();
+    // Add the callback_group's notify condition to the guard condition handles
+    memory_strategy_->add_guard_condition(*callback_group_guard_condition);
   }
 
   if (notify) {
+    // Interrupt waiting to handle new node
     try {
       interrupt_guard_condition_->trigger();
     } catch (const rclcpp::exceptions::RCLError & ex) {
@@ -178,23 +241,91 @@ Executor::add_callback_group(
   }
 }
 
+void
+Executor::add_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  bool notify)
+{
+  std::lock_guard<std::mutex> guard{mutex_};
+  this->add_callback_group_to_map(
+    group_ptr,
+    node_ptr,
+    weak_groups_associated_with_executor_to_nodes_,
+    notify);
+}
+
 void
 Executor::add_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify)
 {
-  this->collector_.add_node(node_ptr);
-
-  if (!spinning.load()) {
-    std::lock_guard<std::mutex> guard(mutex_);
-    this->collect_entities();
+  // If the node already has an executor
+  std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
+  if (has_executor.exchange(true)) {
+    throw std::runtime_error(
+            std::string("Node '") + node_ptr->get_fully_qualified_name() +
+            "' has already been added to an executor.");
   }
+  std::lock_guard<std::mutex> guard{mutex_};
+  node_ptr->for_each_callback_group(
+    [this, node_ptr, notify](rclcpp::CallbackGroup::SharedPtr group_ptr)
+    {
+      if (!group_ptr->get_associated_with_executor_atomic().load() &&
+      group_ptr->automatically_add_to_executor_with_node())
+      {
+        this->add_callback_group_to_map(
+          group_ptr,
+          node_ptr,
+          weak_groups_to_nodes_associated_with_executor_,
+          notify);
+      }
+    });
 
-  if (notify) {
-    try {
-      interrupt_guard_condition_->trigger();
-    } catch (const rclcpp::exceptions::RCLError & ex) {
-      throw std::runtime_error(
-              std::string(
-                "Failed to trigger guard condition on node add: ") + ex.what());
+  const auto gc = node_ptr->get_shared_notify_guard_condition();
+  weak_nodes_to_guard_conditions_[node_ptr] = gc.get();
+  // Add the node's notify condition to the guard condition handles
+  memory_strategy_->add_guard_condition(*gc);
+  weak_nodes_.push_back(node_ptr);
+}
+
+void
+Executor::remove_callback_group_from_map(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes,
+  bool notify)
+{
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr;
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
+  auto iter = weak_groups_to_nodes.find(weak_group_ptr);
+  if (iter != weak_groups_to_nodes.end()) {
+    node_ptr = iter->second.lock();
+    if (node_ptr == nullptr) {
+      throw std::runtime_error("Node must not be deleted before its callback group(s).");
+    }
+    weak_groups_to_nodes.erase(iter);
+    weak_groups_to_nodes_.erase(group_ptr);
+    std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
+    has_executor.store(false);
+  } else {
+    throw std::runtime_error("Callback group needs to be associated with executor.");
+  }
+  // If the node was matched and removed, interrupt waiting.
+  if (!has_node(node_ptr, weak_groups_to_nodes_associated_with_executor_) &&
+    !has_node(node_ptr, weak_groups_associated_with_executor_to_nodes_))
+  {
+    auto iter = weak_groups_to_guard_conditions_.find(weak_group_ptr);
+    if (iter != weak_groups_to_guard_conditions_.end()) {
+      memory_strategy_->remove_guard_condition(iter->second);
+    }
+    weak_groups_to_guard_conditions_.erase(weak_group_ptr);
+
+    if (notify) {
+      try {
+        interrupt_guard_condition_->trigger();
+      } catch (const rclcpp::exceptions::RCLError & ex) {
+        throw std::runtime_error(
+                std::string(
+                  "Failed to trigger guard condition on callback group remove: ") + ex.what());
+      }
     }
   }
 }
@@ -204,21 +335,11 @@ Executor::remove_callback_group(
   rclcpp::CallbackGroup::SharedPtr group_ptr,
   bool notify)
 {
-  this->collector_.remove_callback_group(group_ptr);
-
-  if (!spinning.load()) {
-    std::lock_guard<std::mutex> guard(mutex_);
-    this->collect_entities();
-  }
-  if (notify) {
-    try {
-      interrupt_guard_condition_->trigger();
-    } catch (const rclcpp::exceptions::RCLError & ex) {
-      throw std::runtime_error(
-              std::string(
-                "Failed to trigger guard condition on callback group remove: ") + ex.what());
-    }
-  }
+  std::lock_guard<std::mutex> guard{mutex_};
+  this->remove_callback_group_from_map(
+    group_ptr,
+    weak_groups_associated_with_executor_to_nodes_,
+    notify);
 }
 
 void
@@ -230,22 +351,48 @@ Executor::add_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify)
 void
 Executor::remove_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify)
 {
-  this->collector_.remove_node(node_ptr);
-
-  if (!spinning.load()) {
-    std::lock_guard<std::mutex> guard(mutex_);
-    this->collect_entities();
+  if (!node_ptr->get_associated_with_executor_atomic().load()) {
+    throw std::runtime_error("Node needs to be associated with an executor.");
   }
 
-  if (notify) {
-    try {
-      interrupt_guard_condition_->trigger();
-    } catch (const rclcpp::exceptions::RCLError & ex) {
-      throw std::runtime_error(
-              std::string(
-                "Failed to trigger guard condition on node remove: ") + ex.what());
+  std::lock_guard<std::mutex> guard{mutex_};
+  bool found_node = false;
+  auto node_it = weak_nodes_.begin();
+  while (node_it != weak_nodes_.end()) {
+    bool matched = (node_it->lock() == node_ptr);
+    if (matched) {
+      found_node = true;
+      node_it = weak_nodes_.erase(node_it);
+    } else {
+      ++node_it;
     }
   }
+  if (!found_node) {
+    throw std::runtime_error("Node needs to be associated with this executor.");
+  }
+
+  for (auto it = weak_groups_to_nodes_associated_with_executor_.begin();
+    it != weak_groups_to_nodes_associated_with_executor_.end(); )
+  {
+    auto weak_node_ptr = it->second;
+    auto shared_node_ptr = weak_node_ptr.lock();
+    auto group_ptr = it->first.lock();
+
+    // Increment iterator before removing in case it's invalidated
+    it++;
+    if (shared_node_ptr == node_ptr) {
+      remove_callback_group_from_map(
+        group_ptr,
+        weak_groups_to_nodes_associated_with_executor_,
+        notify);
+    }
+  }
+
+  memory_strategy_->remove_guard_condition(node_ptr->get_shared_notify_guard_condition().get());
+  weak_nodes_to_guard_conditions_.erase(node_ptr);
+
+  std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
+  has_executor.store(false);
 }
 
 void
@@ -284,6 +431,22 @@ void Executor::spin_some(std::chrono::nanoseconds max_duration)
   return this->spin_some_impl(max_duration, false);
 }
 
+void
+Executor::spin_node_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node,
+  std::chrono::nanoseconds max_duration)
+{
+  this->add_node(node, false);
+  spin_all(max_duration);
+  this->remove_node(node, false);
+}
+
+void
+Executor::spin_node_all(std::shared_ptr<rclcpp::Node> node, std::chrono::nanoseconds max_duration)
+{
+  this->spin_node_all(node->get_node_base_interface(), max_duration);
+}
+
 void Executor::spin_all(std::chrono::nanoseconds max_duration)
 {
   if (max_duration < 0ns) {
@@ -312,44 +475,21 @@ Executor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
     throw std::runtime_error("spin_some() called while already spinning");
   }
   RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
-
-  size_t work_in_queue = 0;
-  bool has_waited = false;
-
-  {
-    std::lock_guard<std::mutex> lock(mutex_);
-    work_in_queue = ready_executables_.size();
-  }
-  // The logic below is to guarantee that we:
-  // a) run all of the work in the queue before we spin the first time
-  // b) spin at least once
-  // c) run all of the work in the queue after we spin
-
+  bool work_available = false;
   while (rclcpp::ok(context_) && spinning.load() && max_duration_not_elapsed()) {
     AnyExecutable any_exec;
-    if (work_in_queue > 0) {
-      // If there is work in the queue, then execute it
-      // This covers the case that there are things left in the queue from a
-      // previous spin.
-      if (get_next_ready_executable(any_exec)) {
-        execute_any_executable(any_exec);
-      }
-    } else if (!has_waited && !work_in_queue) {
-      // Once the ready queue is empty, then we need to wait at least once.
-      wait_for_work(std::chrono::milliseconds(0));
-      has_waited = true;
-    } else if (has_waited && !work_in_queue) {
-      // Once we have emptied the ready queue, but have already waited:
-      if (!exhaustive) {
-        // In the case of spin some, then we can exit
-        break;
-      } else {
-        // In the case of spin all, then we will allow ourselves to wait again.
-        has_waited = false;
-      }
+    if (!work_available) {
+      wait_for_work(std::chrono::milliseconds::zero());
+    }
+    if (get_next_ready_executable(any_exec)) {
+      execute_any_executable(any_exec);
+      work_available = true;
+    } else {
+      if (!work_available || !exhaustive) {
+        break;
+      }
+      work_available = false;
     }
-    std::lock_guard<std::mutex> lock(mutex_);
-    work_in_queue = ready_executables_.size();
   }
 }
 
@@ -384,21 +524,30 @@ Executor::cancel()
   }
 }
 
+void
+Executor::set_memory_strategy(rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy)
+{
+  if (memory_strategy == nullptr) {
+    throw std::runtime_error("Received NULL memory strategy in executor.");
+  }
+  std::lock_guard<std::mutex> guard{mutex_};
+  memory_strategy_ = memory_strategy;
+}
+
 void
 Executor::execute_any_executable(AnyExecutable & any_exec)
 {
   if (!spinning.load()) {
     return;
   }
-
   if (any_exec.timer) {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_executor_execute,
       static_cast<const void *>(any_exec.timer->get_timer_handle().get()));
     execute_timer(any_exec.timer);
   }
   if (any_exec.subscription) {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
       rclcpp_executor_execute,
       static_cast<const void *>(any_exec.subscription->get_subscription_handle().get()));
     execute_subscription(any_exec.subscription);
@@ -412,10 +561,16 @@ Executor::execute_any_executable(AnyExecutable & any_exec)
   if (any_exec.waitable) {
     any_exec.waitable->execute(any_exec.data);
   }
-
   // Reset the callback_group, regardless of type
-  if (any_exec.callback_group) {
-    any_exec.callback_group->can_be_taken_from().store(true);
+  any_exec.callback_group->can_be_taken_from().store(true);
+  // Wake the wait, because it may need to be recalculated or work that
+  // was previously blocked is now available.
+  try {
+    interrupt_guard_condition_->trigger();
+  } catch (const rclcpp::exceptions::RCLError & ex) {
+    throw std::runtime_error(
+            std::string(
+              "Failed to trigger guard condition from execute_any_executable: ") + ex.what());
   }
 }
 
@@ -464,7 +619,7 @@ Executor::execute_subscription(rclcpp::SubscriptionBase::SharedPtr subscription)
   rclcpp::MessageInfo message_info;
   message_info.get_rmw_message_info().from_intra_process = false;
 
-  switch (subscription->get_subscription_type()) {
+  switch (subscription->get_delivered_message_kind()) {
     // Deliver ROS message
     case rclcpp::DeliveredMessageKind::ROS_MESSAGE:
       {
@@ -514,6 +669,11 @@ Executor::execute_subscription(rclcpp::SubscriptionBase::SharedPtr subscription)
             subscription->get_topic_name(),
             [&]() {return subscription->take_type_erased(message.get(), message_info);},
             [&]() {subscription->handle_message(message, message_info);});
+          // TODO(clalancette): In the case that the user is using the MessageMemoryPool,
+          // and they take a shared_ptr reference to the message in the callback, this can
+          // inadvertently return the message to the pool when the user is still using it.
+          // This is a bug that needs to be fixed in the pool, and we should probably have
+          // a custom deleter for the message that actually does the return_message().
           subscription->return_message(message);
         }
         break;
@@ -584,117 +744,228 @@ Executor::execute_client(
     [&]() {client->handle_response(request_header, response);});
 }
 
-void
-Executor::collect_entities()
-{
-  // Get the current list of available waitables from the collector.
-  rclcpp::executors::ExecutorEntitiesCollection collection;
-  this->collector_.update_collections();
-  auto callback_groups = this->collector_.get_all_callback_groups();
-  rclcpp::executors::build_entities_collection(callback_groups, collection);
-
-  // Make a copy of notify waitable so we can continue to mutate the original
-  // one outside of the execute loop.
-  // This prevents the collection of guard conditions in the waitable from changing
-  // while we are waiting on it.
-  if (notify_waitable_) {
-    current_notify_waitable_ = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(
-      *notify_waitable_);
-    auto notify_waitable = std::static_pointer_cast<rclcpp::Waitable>(current_notify_waitable_);
-    collection.waitables.insert({notify_waitable.get(), {notify_waitable, {}}});
-  }
-
-  std::cout << "current_collection.timers: " << current_collection_.timers.size() << std::endl;
-  std::cout << "next_collection.timers: " << collection.timers.size() << std::endl;
-  // Update each of the groups of entities in the current collection, adding or removing
-  // from the wait set as necessary.
-  current_collection_.timers.update(
-    collection.timers,
-    [this](auto timer) {
-        std::cout << "add_timer(" << timer << ")" << std::endl;
-        wait_set_.add_timer(timer);},
-    [this](auto timer) {
-        std::cout << "remove_timer(" << timer << ")" << std::endl;
-        wait_set_.remove_timer(timer);});
-
-  std::cout << "current_collection.subscriptions: " << current_collection_.subscriptions.size() << std::endl;
-  std::cout << "next_collection.subscriptions: " << collection.subscriptions.size() << std::endl;
-  current_collection_.subscriptions.update(
-    collection.subscriptions,
-    [this](auto subscription) {
-      std::cout << "add_subscription(" << subscription << ")" << std::endl;
-      wait_set_.add_subscription(subscription, kDefaultSubscriptionMask);
-    },
-    [this](auto subscription) {
-      std::cout << "remove_subscription(" << subscription << ")" << std::endl;
-      wait_set_.remove_subscription(subscription, kDefaultSubscriptionMask);
-    });
-
-  current_collection_.clients.update(
-    collection.clients,
-    [this](auto client) {wait_set_.add_client(client);},
-    [this](auto client) {wait_set_.remove_client(client);});
-
-  current_collection_.services.update(
-    collection.services,
-    [this](auto service) {wait_set_.add_service(service);},
-    [this](auto service) {wait_set_.remove_service(service);});
-
-  current_collection_.guard_conditions.update(
-    collection.guard_conditions,
-    [this](auto guard_condition) {wait_set_.add_guard_condition(guard_condition);},
-    [this](auto guard_condition) {wait_set_.remove_guard_condition(guard_condition);});
-
-  current_collection_.waitables.update(
-    collection.waitables,
-    [this](auto waitable) {
-      wait_set_.add_waitable(waitable);
-    },
-    [this](auto waitable) {wait_set_.remove_waitable(waitable);});
-
-  this->entities_need_rebuild_.store(false);
-}
-
 void
 Executor::wait_for_work(std::chrono::nanoseconds timeout)
 {
-  TRACEPOINT(rclcpp_executor_wait_for_work, timeout.count());
+  TRACETOOLS_TRACEPOINT(rclcpp_executor_wait_for_work, timeout.count());
+  {
+    std::lock_guard<std::mutex> guard(mutex_);
 
-  std::lock_guard<std::mutex> guard(mutex_);
-  if (current_collection_.empty() || this->entities_need_rebuild_.load()) {
-    this->collect_entities();
+    // Check weak_nodes_ to find any callback group that is not owned
+    // by an executor and add it to the list of callbackgroups for
+    // collect entities. Also exchange to false so it is not
+    // allowed to add to another executor
+    add_callback_groups_from_nodes_associated_to_executor();
+
+    // Collect the subscriptions and timers to be waited on
+    memory_strategy_->clear_handles();
+    bool has_invalid_weak_groups_or_nodes =
+      memory_strategy_->collect_entities(weak_groups_to_nodes_);
+
+    if (has_invalid_weak_groups_or_nodes) {
+      std::vector<rclcpp::CallbackGroup::WeakPtr> invalid_group_ptrs;
+      for (auto pair : weak_groups_to_nodes_) {
+        auto weak_group_ptr = pair.first;
+        auto weak_node_ptr = pair.second;
+        if (weak_group_ptr.expired() || weak_node_ptr.expired()) {
+          invalid_group_ptrs.push_back(weak_group_ptr);
+          auto node_guard_pair = weak_nodes_to_guard_conditions_.find(weak_node_ptr);
+          if (node_guard_pair != weak_nodes_to_guard_conditions_.end()) {
+            auto guard_condition = node_guard_pair->second;
+            weak_nodes_to_guard_conditions_.erase(weak_node_ptr);
+            memory_strategy_->remove_guard_condition(guard_condition);
+          }
+        }
+      }
+      std::for_each(
+        invalid_group_ptrs.begin(), invalid_group_ptrs.end(),
+        [this](rclcpp::CallbackGroup::WeakPtr group_ptr) {
+          if (weak_groups_to_nodes_associated_with_executor_.find(group_ptr) !=
+          weak_groups_to_nodes_associated_with_executor_.end())
+          {
+            weak_groups_to_nodes_associated_with_executor_.erase(group_ptr);
+          }
+          if (weak_groups_associated_with_executor_to_nodes_.find(group_ptr) !=
+          weak_groups_associated_with_executor_to_nodes_.end())
+          {
+            weak_groups_associated_with_executor_to_nodes_.erase(group_ptr);
+          }
+          auto callback_guard_pair = weak_groups_to_guard_conditions_.find(group_ptr);
+          if (callback_guard_pair != weak_groups_to_guard_conditions_.end()) {
+            auto guard_condition = callback_guard_pair->second;
+            weak_groups_to_guard_conditions_.erase(group_ptr);
+            memory_strategy_->remove_guard_condition(guard_condition);
+          }
+          weak_groups_to_nodes_.erase(group_ptr);
+        });
+    }
+
+    // clear wait set
+    rcl_ret_t ret = rcl_wait_set_clear(&wait_set_);
+    if (ret != RCL_RET_OK) {
+      throw_from_rcl_error(ret, "Couldn't clear wait set");
+    }
+
+    // The size of waitables are accounted for in size of the other entities
+    ret = rcl_wait_set_resize(
+      &wait_set_, memory_strategy_->number_of_ready_subscriptions(),
+      memory_strategy_->number_of_guard_conditions(), memory_strategy_->number_of_ready_timers(),
+      memory_strategy_->number_of_ready_clients(), memory_strategy_->number_of_ready_services(),
+      memory_strategy_->number_of_ready_events());
+    if (RCL_RET_OK != ret) {
+      throw_from_rcl_error(ret, "Couldn't resize the wait set");
+    }
+
+    if (!memory_strategy_->add_handles_to_wait_set(&wait_set_)) {
+      throw std::runtime_error("Couldn't fill wait set");
+    }
   }
 
-  auto wait_result = wait_set_.wait(timeout);
-  if (wait_result.kind() == WaitResultKind::Empty) {
+  rcl_ret_t status =
+    rcl_wait(&wait_set_, std::chrono::duration_cast<std::chrono::nanoseconds>(timeout).count());
+  if (status == RCL_RET_WAIT_SET_EMPTY) {
     RCUTILS_LOG_WARN_NAMED(
       "rclcpp",
-      "empty wait set received in wait(). This should never happen.");
+      "empty wait set received in rcl_wait(). This should never happen.");
+  } else if (status != RCL_RET_OK && status != RCL_RET_TIMEOUT) {
+    using rclcpp::exceptions::throw_from_rcl_error;
+    throw_from_rcl_error(status, "rcl_wait() failed");
   }
-  rclcpp::executors::ready_executables(current_collection_, wait_result, ready_executables_);
+
+  // check the null handles in the wait set and remove them from the handles in memory strategy
+  // for callback-based entities
+  std::lock_guard<std::mutex> guard(mutex_);
+  memory_strategy_->remove_null_handles(&wait_set_);
+}
+
+rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
+Executor::get_node_by_group(
+  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
+  weak_groups_to_nodes,
+  rclcpp::CallbackGroup::SharedPtr group)
+{
+  if (!group) {
+    return nullptr;
+  }
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr(group);
+  const auto finder = weak_groups_to_nodes.find(weak_group_ptr);
+  if (finder != weak_groups_to_nodes.end()) {
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr = finder->second.lock();
+    return node_ptr;
+  }
+  return nullptr;
+}
+
+rclcpp::CallbackGroup::SharedPtr
+Executor::get_group_by_timer(rclcpp::TimerBase::SharedPtr timer)
+{
+  std::lock_guard<std::mutex> guard{mutex_};
+  for (const auto & pair : weak_groups_associated_with_executor_to_nodes_) {
+    auto group = pair.first.lock();
+    if (!group) {
+      continue;
+    }
+    auto timer_ref = group->find_timer_ptrs_if(
+      [timer](const rclcpp::TimerBase::SharedPtr & timer_ptr) -> bool {
+        return timer_ptr == timer;
+      });
+    if (timer_ref) {
+      return group;
+    }
+  }
+
+  for (const auto & pair : weak_groups_to_nodes_associated_with_executor_) {
+    auto group = pair.first.lock();
+    if (!group) {
+      continue;
+    }
+    auto timer_ref = group->find_timer_ptrs_if(
+      [timer](const rclcpp::TimerBase::SharedPtr & timer_ptr) -> bool {
+        return timer_ptr == timer;
+      });
+    if (timer_ref) {
+      return group;
+    }
+  }
+  return nullptr;
 }
 
 bool
 Executor::get_next_ready_executable(AnyExecutable & any_executable)
 {
-  TRACEPOINT(rclcpp_executor_get_next_ready);
+  bool success = get_next_ready_executable_from_map(any_executable, weak_groups_to_nodes_);
+  return success;
+}
 
-  std::lock_guard<std::mutex> guard(mutex_);
-  if (ready_executables_.size() == 0) {
-    return false;
+bool
+Executor::get_next_ready_executable_from_map(
+  AnyExecutable & any_executable,
+  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
+  weak_groups_to_nodes)
+{
+  TRACETOOLS_TRACEPOINT(rclcpp_executor_get_next_ready);
+  bool success = false;
+  std::lock_guard<std::mutex> guard{mutex_};
+  // Check the timers to see if there are any that are ready
+  memory_strategy_->get_next_timer(any_executable, weak_groups_to_nodes);
+  if (any_executable.timer) {
+    success = true;
+  }
+  if (!success) {
+    // Check the subscriptions to see if there are any that are ready
+    memory_strategy_->get_next_subscription(any_executable, weak_groups_to_nodes);
+    if (any_executable.subscription) {
+      success = true;
+    }
+  }
+  if (!success) {
+    // Check the services to see if there are any that are ready
+    memory_strategy_->get_next_service(any_executable, weak_groups_to_nodes);
+    if (any_executable.service) {
+      success = true;
+    }
+  }
+  if (!success) {
+    // Check the clients to see if there are any that are ready
+    memory_strategy_->get_next_client(any_executable, weak_groups_to_nodes);
+    if (any_executable.client) {
+      success = true;
+    }
+  }
+  if (!success) {
+    // Check the waitables to see if there are any that are ready
+    memory_strategy_->get_next_waitable(any_executable, weak_groups_to_nodes);
+    if (any_executable.waitable) {
+      any_executable.data = any_executable.waitable->take_data();
+      success = true;
+    }
+  }
+  // At this point any_executable should be valid with either a valid subscription
+  // or a valid timer, or it should be a null shared_ptr
+  if (success) {
+    rclcpp::CallbackGroup::WeakPtr weak_group_ptr = any_executable.callback_group;
+    auto iter = weak_groups_to_nodes.find(weak_group_ptr);
+    if (iter == weak_groups_to_nodes.end()) {
+      success = false;
+    }
   }
 
-  any_executable = ready_executables_.front();
-  ready_executables_.pop_front();
-
-  if (any_executable.callback_group &&
-    any_executable.callback_group->type() == CallbackGroupType::MutuallyExclusive)
-  {
-    assert(any_executable.callback_group->can_be_taken_from().load());
-    any_executable.callback_group->can_be_taken_from().store(false);
+  if (success) {
+    // If it is valid, check to see if the group is mutually exclusive or
+    // not, then mark it accordingly ..Check if the callback_group belongs to this executor
+    if (any_executable.callback_group && any_executable.callback_group->type() == \
+      CallbackGroupType::MutuallyExclusive)
+    {
+      // It should not have been taken otherwise
+      assert(any_executable.callback_group->can_be_taken_from().load());
+      // Set to false to indicate something is being run from this group
+      // This is reset to true either when the any_executable is executed or when the
+      // any_executable is destructued
+      any_executable.callback_group->can_be_taken_from().store(false);
+    }
   }
-
-  return true;
+  // If there is no ready executable, return false
+  return success;
 }
 
 bool
@@ -717,6 +988,22 @@ Executor::get_next_executable(AnyExecutable & any_executable, std::chrono::nanos
   return success;
 }
 
+// Returns true iff the weak_groups_to_nodes map has node_ptr as the value in any of its entry.
+bool
+Executor::has_node(
+  const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
+  weak_groups_to_nodes) const
+{
+  return std::find_if(
+    weak_groups_to_nodes.begin(),
+    weak_groups_to_nodes.end(),
+    [&](const WeakCallbackGroupsToNodesMap::value_type & other) -> bool {
+      auto other_ptr = other.second.lock();
+      return other_ptr == node_ptr;
+    }) != weak_groups_to_nodes.end();
+}
+
 bool
 Executor::is_spinning()
 {

rclcpp/src/rclcpp/executors.cpp

@@ -14,6 +14,21 @@
 
 #include "rclcpp/executors.hpp"
 
+void
+rclcpp::spin_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  std::chrono::nanoseconds max_duration)
+{
+  rclcpp::executors::SingleThreadedExecutor exec;
+  exec.spin_node_all(node_ptr, max_duration);
+}
+
+void
+rclcpp::spin_all(rclcpp::Node::SharedPtr node_ptr, std::chrono::nanoseconds max_duration)
+{
+  rclcpp::spin_all(node_ptr->get_node_base_interface(), max_duration);
+}
+
 void
 rclcpp::spin_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
 {

rclcpp/src/rclcpp/executors/executor_entities_collection.cpp

@@ -222,7 +222,9 @@ ready_executables(
     executables.push_back(exec);
     added++;
   }
+
   return added;
 }
+
 }  // namespace executors
 }  // namespace rclcpp

rclcpp/src/rclcpp/executors/executor_entities_collector.cpp

@@ -105,8 +105,7 @@ void
 ExecutorEntitiesCollector::remove_node(
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
 {
-  std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
-  if (!has_executor.exchange(false)) {
+  if (!node_ptr->get_associated_with_executor_atomic().load()) {
     throw std::runtime_error(
             std::string("Node '") + node_ptr->get_fully_qualified_name() +
             "' needs to be associated with an executor.");
@@ -162,6 +161,7 @@ ExecutorEntitiesCollector::remove_callback_group(rclcpp::CallbackGroup::SharedPt
     throw std::runtime_error("Node must not be deleted before its callback group(s).");
   }
   */
+
   auto weak_group_ptr = rclcpp::CallbackGroup::WeakPtr(group_ptr);
   std::lock_guard<std::mutex> lock(mutex_);
   bool associated = manually_added_groups_.count(group_ptr) != 0;
@@ -314,11 +314,7 @@ ExecutorEntitiesCollector::process_queues()
     if (node_it != weak_nodes_.end()) {
       remove_weak_node(node_it);
     } else {
-      // The node may have been destroyed and removed from the colletion before
-      // we processed the queues.  Don't throw if the pointer is already expired.
-      if (!weak_node_ptr.expired()) {
-        throw std::runtime_error("Node needs to be associated with this executor.");
-      }
+      throw std::runtime_error("Node needs to be associated with this executor.");
     }
 
     auto node_ptr = weak_node_ptr.lock();

rclcpp/src/rclcpp/executors/executor_notify_waitable.cpp

@@ -46,18 +46,20 @@ void
 ExecutorNotifyWaitable::add_to_wait_set(rcl_wait_set_t * wait_set)
 {
   std::lock_guard<std::mutex> lock(guard_condition_mutex_);
+
   for (auto weak_guard_condition : this->notify_guard_conditions_) {
     auto guard_condition = weak_guard_condition.lock();
-    if (!guard_condition) {continue;}
+    if (guard_condition) {
+      auto rcl_guard_condition = &guard_condition->get_rcl_guard_condition();
 
-    rcl_guard_condition_t * cond = &guard_condition->get_rcl_guard_condition();
+      rcl_ret_t ret = rcl_wait_set_add_guard_condition(
+        wait_set,
+        rcl_guard_condition, NULL);
 
-    rcl_ret_t ret = rcl_wait_set_add_guard_condition(
-      wait_set, cond, NULL);
-
-    if (RCL_RET_OK != ret) {
-      rclcpp::exceptions::throw_from_rcl_error(
-        ret, "failed to add guard condition to wait set");
+      if (RCL_RET_OK != ret) {
+        rclcpp::exceptions::throw_from_rcl_error(
+          ret, "failed to add guard condition to wait set");
+      }
     }
   }
 }

rclcpp/src/rclcpp/executors/multi_threaded_executor.cpp

@@ -99,19 +99,6 @@ MultiThreadedExecutor::run(size_t this_thread_number)
 
     execute_any_executable(any_exec);
 
-    if (any_exec.callback_group &&
-      any_exec.callback_group->type() == CallbackGroupType::MutuallyExclusive &&
-      any_exec.callback_group->size() > 1)
-    {
-      try {
-        interrupt_guard_condition_->trigger();
-      } catch (const rclcpp::exceptions::RCLError & ex) {
-        throw std::runtime_error(
-                std::string(
-                  "Failed to trigger guard condition on callback group change: ") + ex.what());
-      }
-    }
-
     // Clear the callback_group to prevent the AnyExecutable destructor from
     // resetting the callback group `can_be_taken_from`
     any_exec.callback_group.reset();

rclcpp/src/rclcpp/executors/static_executor_entities_collector.cpp

@@ -0,0 +1,524 @@
+// Copyright 2020 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.
+
+#include "rclcpp/executors/static_executor_entities_collector.hpp"
+
+#include <algorithm>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "rclcpp/memory_strategy.hpp"
+#include "rclcpp/executors/static_single_threaded_executor.hpp"
+#include "rclcpp/detail/add_guard_condition_to_rcl_wait_set.hpp"
+
+using rclcpp::executors::StaticExecutorEntitiesCollector;
+
+StaticExecutorEntitiesCollector::~StaticExecutorEntitiesCollector()
+{
+  // Disassociate all callback groups and thus nodes.
+  for (const auto & pair : weak_groups_associated_with_executor_to_nodes_) {
+    auto group = pair.first.lock();
+    if (group) {
+      std::atomic_bool & has_executor = group->get_associated_with_executor_atomic();
+      has_executor.store(false);
+    }
+  }
+  for (const auto & pair : weak_groups_to_nodes_associated_with_executor_) {
+    auto group = pair.first.lock();
+    if (group) {
+      std::atomic_bool & has_executor = group->get_associated_with_executor_atomic();
+      has_executor.store(false);
+    }
+  }
+  // Disassociate all nodes
+  for (const auto & weak_node : weak_nodes_) {
+    auto node = weak_node.lock();
+    if (node) {
+      std::atomic_bool & has_executor = node->get_associated_with_executor_atomic();
+      has_executor.store(false);
+    }
+  }
+  weak_groups_associated_with_executor_to_nodes_.clear();
+  weak_groups_to_nodes_associated_with_executor_.clear();
+  exec_list_.clear();
+  weak_nodes_.clear();
+  weak_nodes_to_guard_conditions_.clear();
+}
+
+void
+StaticExecutorEntitiesCollector::init(
+  rcl_wait_set_t * p_wait_set,
+  rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy)
+{
+  // Empty initialize executable list
+  exec_list_ = rclcpp::experimental::ExecutableList();
+  // Get executor's wait_set_ pointer
+  p_wait_set_ = p_wait_set;
+  // Get executor's memory strategy ptr
+  if (memory_strategy == nullptr) {
+    throw std::runtime_error("Received NULL memory strategy in executor waitable.");
+  }
+  memory_strategy_ = memory_strategy;
+
+  // Get memory strategy and executable list. Prepare wait_set_
+  std::shared_ptr<void> shared_ptr;
+  execute(shared_ptr);
+
+  // The entities collector is now initialized
+  initialized_ = true;
+}
+
+void
+StaticExecutorEntitiesCollector::fini()
+{
+  memory_strategy_->clear_handles();
+  exec_list_.clear();
+}
+
+std::shared_ptr<void>
+StaticExecutorEntitiesCollector::take_data()
+{
+  return nullptr;
+}
+
+void
+StaticExecutorEntitiesCollector::execute(std::shared_ptr<void> & data)
+{
+  (void) data;
+  // Fill memory strategy with entities coming from weak_nodes_
+  fill_memory_strategy();
+  // Fill exec_list_ with entities coming from weak_nodes_ (same as memory strategy)
+  fill_executable_list();
+  // Resize the wait_set_ based on memory_strategy handles (rcl_wait_set_resize)
+  prepare_wait_set();
+  // Add new nodes guard conditions to map
+  std::lock_guard<std::mutex> guard{new_nodes_mutex_};
+  for (const auto & weak_node : new_nodes_) {
+    if (auto node_ptr = weak_node.lock()) {
+      weak_nodes_to_guard_conditions_[node_ptr] =
+        node_ptr->get_shared_notify_guard_condition().get();
+    }
+  }
+  new_nodes_.clear();
+}
+
+void
+StaticExecutorEntitiesCollector::fill_memory_strategy()
+{
+  memory_strategy_->clear_handles();
+  bool has_invalid_weak_groups_or_nodes =
+    memory_strategy_->collect_entities(weak_groups_to_nodes_associated_with_executor_);
+  // Clean up any invalid nodes, if they were detected
+  if (has_invalid_weak_groups_or_nodes) {
+    std::vector<rclcpp::CallbackGroup::WeakPtr> invalid_group_ptrs;
+    for (const auto & pair : weak_groups_to_nodes_associated_with_executor_) {
+      auto & weak_group_ptr = pair.first;
+      auto & weak_node_ptr = pair.second;
+      if (weak_group_ptr.expired() || weak_node_ptr.expired()) {
+        invalid_group_ptrs.push_back(weak_group_ptr);
+      }
+    }
+    std::for_each(
+      invalid_group_ptrs.begin(), invalid_group_ptrs.end(),
+      [this](rclcpp::CallbackGroup::WeakPtr group_ptr) {
+        weak_groups_to_nodes_associated_with_executor_.erase(group_ptr);
+      });
+  }
+  has_invalid_weak_groups_or_nodes =
+    memory_strategy_->collect_entities(weak_groups_associated_with_executor_to_nodes_);
+  // Clean up any invalid nodes, if they were detected
+  if (has_invalid_weak_groups_or_nodes) {
+    std::vector<rclcpp::CallbackGroup::WeakPtr> invalid_group_ptrs;
+    for (const auto & pair : weak_groups_associated_with_executor_to_nodes_) {
+      auto & weak_group_ptr = pair.first;
+      const auto & weak_node_ptr = pair.second;
+      if (weak_group_ptr.expired() || weak_node_ptr.expired()) {
+        invalid_group_ptrs.push_back(weak_group_ptr);
+      }
+    }
+    std::for_each(
+      invalid_group_ptrs.begin(), invalid_group_ptrs.end(),
+      [this](rclcpp::CallbackGroup::WeakPtr group_ptr) {
+        weak_groups_associated_with_executor_to_nodes_.erase(group_ptr);
+      });
+  }
+
+  // Add the static executor waitable to the memory strategy
+  memory_strategy_->add_waitable_handle(this->shared_from_this());
+}
+
+void
+StaticExecutorEntitiesCollector::fill_executable_list()
+{
+  exec_list_.clear();
+  add_callback_groups_from_nodes_associated_to_executor();
+  fill_executable_list_from_map(weak_groups_associated_with_executor_to_nodes_);
+  fill_executable_list_from_map(weak_groups_to_nodes_associated_with_executor_);
+  // Add the executor's waitable to the executable list
+  exec_list_.add_waitable(shared_from_this());
+}
+void
+StaticExecutorEntitiesCollector::fill_executable_list_from_map(
+  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
+  weak_groups_to_nodes)
+{
+  for (const auto & pair : weak_groups_to_nodes) {
+    auto group = pair.first.lock();
+    auto node = pair.second.lock();
+    if (!node || !group || !group->can_be_taken_from().load()) {
+      continue;
+    }
+    group->find_timer_ptrs_if(
+      [this](const rclcpp::TimerBase::SharedPtr & timer) {
+        if (timer) {
+          exec_list_.add_timer(timer);
+        }
+        return false;
+      });
+    group->find_subscription_ptrs_if(
+      [this](const rclcpp::SubscriptionBase::SharedPtr & subscription) {
+        if (subscription) {
+          exec_list_.add_subscription(subscription);
+        }
+        return false;
+      });
+    group->find_service_ptrs_if(
+      [this](const rclcpp::ServiceBase::SharedPtr & service) {
+        if (service) {
+          exec_list_.add_service(service);
+        }
+        return false;
+      });
+    group->find_client_ptrs_if(
+      [this](const rclcpp::ClientBase::SharedPtr & client) {
+        if (client) {
+          exec_list_.add_client(client);
+        }
+        return false;
+      });
+    group->find_waitable_ptrs_if(
+      [this](const rclcpp::Waitable::SharedPtr & waitable) {
+        if (waitable) {
+          exec_list_.add_waitable(waitable);
+        }
+        return false;
+      });
+  }
+}
+
+void
+StaticExecutorEntitiesCollector::prepare_wait_set()
+{
+  // clear wait set
+  if (rcl_wait_set_clear(p_wait_set_) != RCL_RET_OK) {
+    throw std::runtime_error("Couldn't clear wait set");
+  }
+
+  // The size of waitables are accounted for in size of the other entities
+  rcl_ret_t ret = rcl_wait_set_resize(
+    p_wait_set_, memory_strategy_->number_of_ready_subscriptions(),
+    memory_strategy_->number_of_guard_conditions(), memory_strategy_->number_of_ready_timers(),
+    memory_strategy_->number_of_ready_clients(), memory_strategy_->number_of_ready_services(),
+    memory_strategy_->number_of_ready_events());
+
+  if (RCL_RET_OK != ret) {
+    throw std::runtime_error(
+            std::string("Couldn't resize the wait set: ") + rcl_get_error_string().str);
+  }
+}
+
+void
+StaticExecutorEntitiesCollector::refresh_wait_set(std::chrono::nanoseconds timeout)
+{
+  // clear wait set (memset to '0' all wait_set_ entities
+  // but keeps the wait_set_ number of entities)
+  if (rcl_wait_set_clear(p_wait_set_) != RCL_RET_OK) {
+    throw std::runtime_error("Couldn't clear wait set");
+  }
+
+  if (!memory_strategy_->add_handles_to_wait_set(p_wait_set_)) {
+    throw std::runtime_error("Couldn't fill wait set");
+  }
+
+  rcl_ret_t status =
+    rcl_wait(p_wait_set_, std::chrono::duration_cast<std::chrono::nanoseconds>(timeout).count());
+
+  if (status == RCL_RET_WAIT_SET_EMPTY) {
+    RCUTILS_LOG_WARN_NAMED(
+      "rclcpp",
+      "empty wait set received in rcl_wait(). This should never happen.");
+  } else if (status != RCL_RET_OK && status != RCL_RET_TIMEOUT) {
+    using rclcpp::exceptions::throw_from_rcl_error;
+    throw_from_rcl_error(status, "rcl_wait() failed");
+  }
+}
+
+void
+StaticExecutorEntitiesCollector::add_to_wait_set(rcl_wait_set_t * wait_set)
+{
+  // Add waitable guard conditions (one for each registered node) into the wait set.
+  for (const auto & pair : weak_nodes_to_guard_conditions_) {
+    auto & gc = pair.second;
+    detail::add_guard_condition_to_rcl_wait_set(*wait_set, *gc);
+  }
+}
+
+size_t StaticExecutorEntitiesCollector::get_number_of_ready_guard_conditions()
+{
+  std::lock_guard<std::mutex> guard{new_nodes_mutex_};
+  return weak_nodes_to_guard_conditions_.size() + new_nodes_.size();
+}
+
+bool
+StaticExecutorEntitiesCollector::add_node(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
+{
+  bool is_new_node = false;
+  // If the node already has an executor
+  std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
+  if (has_executor.exchange(true)) {
+    throw std::runtime_error("Node has already been added to an executor.");
+  }
+  node_ptr->for_each_callback_group(
+    [this, node_ptr, &is_new_node](rclcpp::CallbackGroup::SharedPtr group_ptr)
+    {
+      if (
+        !group_ptr->get_associated_with_executor_atomic().load() &&
+        group_ptr->automatically_add_to_executor_with_node())
+      {
+        is_new_node = (add_callback_group(
+          group_ptr,
+          node_ptr,
+          weak_groups_to_nodes_associated_with_executor_) ||
+        is_new_node);
+      }
+    });
+  weak_nodes_.push_back(node_ptr);
+  return is_new_node;
+}
+
+bool
+StaticExecutorEntitiesCollector::add_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes)
+{
+  // If the callback_group already has an executor
+  std::atomic_bool & has_executor = group_ptr->get_associated_with_executor_atomic();
+  if (has_executor.exchange(true)) {
+    throw std::runtime_error("Callback group has already been added to an executor.");
+  }
+  bool is_new_node = !has_node(node_ptr, weak_groups_associated_with_executor_to_nodes_) &&
+    !has_node(node_ptr, weak_groups_to_nodes_associated_with_executor_);
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
+  auto insert_info = weak_groups_to_nodes.insert(
+    std::make_pair(weak_group_ptr, node_ptr));
+  bool was_inserted = insert_info.second;
+  if (!was_inserted) {
+    throw std::runtime_error("Callback group was already added to executor.");
+  }
+  if (is_new_node) {
+    std::lock_guard<std::mutex> guard{new_nodes_mutex_};
+    new_nodes_.push_back(node_ptr);
+    return true;
+  }
+  return false;
+}
+
+bool
+StaticExecutorEntitiesCollector::add_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
+{
+  return add_callback_group(group_ptr, node_ptr, weak_groups_associated_with_executor_to_nodes_);
+}
+
+bool
+StaticExecutorEntitiesCollector::remove_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr)
+{
+  return this->remove_callback_group_from_map(
+    group_ptr,
+    weak_groups_associated_with_executor_to_nodes_);
+}
+
+bool
+StaticExecutorEntitiesCollector::remove_callback_group_from_map(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap & weak_groups_to_nodes)
+{
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr;
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr = group_ptr;
+  auto iter = weak_groups_to_nodes.find(weak_group_ptr);
+  if (iter != weak_groups_to_nodes.end()) {
+    node_ptr = iter->second.lock();
+    if (node_ptr == nullptr) {
+      throw std::runtime_error("Node must not be deleted before its callback group(s).");
+    }
+    weak_groups_to_nodes.erase(iter);
+  } else {
+    throw std::runtime_error("Callback group needs to be associated with executor.");
+  }
+  // If the node was matched and removed, interrupt waiting.
+  if (!has_node(node_ptr, weak_groups_associated_with_executor_to_nodes_) &&
+    !has_node(node_ptr, weak_groups_to_nodes_associated_with_executor_))
+  {
+    rclcpp::node_interfaces::NodeBaseInterface::WeakPtr node_weak_ptr(node_ptr);
+    weak_nodes_to_guard_conditions_.erase(node_weak_ptr);
+    return true;
+  }
+  return false;
+}
+
+bool
+StaticExecutorEntitiesCollector::remove_node(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
+{
+  if (!node_ptr->get_associated_with_executor_atomic().load()) {
+    return false;
+  }
+  bool node_found = false;
+  auto node_it = weak_nodes_.begin();
+  while (node_it != weak_nodes_.end()) {
+    bool matched = (node_it->lock() == node_ptr);
+    if (matched) {
+      weak_nodes_.erase(node_it);
+      node_found = true;
+      break;
+    }
+    ++node_it;
+  }
+  if (!node_found) {
+    return false;
+  }
+  std::vector<rclcpp::CallbackGroup::SharedPtr> found_group_ptrs;
+  std::for_each(
+    weak_groups_to_nodes_associated_with_executor_.begin(),
+    weak_groups_to_nodes_associated_with_executor_.end(),
+    [&found_group_ptrs, node_ptr](std::pair<rclcpp::CallbackGroup::WeakPtr,
+    rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> key_value_pair) {
+      auto & weak_node_ptr = key_value_pair.second;
+      auto shared_node_ptr = weak_node_ptr.lock();
+      auto group_ptr = key_value_pair.first.lock();
+      if (shared_node_ptr == node_ptr) {
+        found_group_ptrs.push_back(group_ptr);
+      }
+    });
+  std::for_each(
+    found_group_ptrs.begin(), found_group_ptrs.end(), [this]
+      (rclcpp::CallbackGroup::SharedPtr group_ptr) {
+      this->remove_callback_group_from_map(
+        group_ptr,
+        weak_groups_to_nodes_associated_with_executor_);
+    });
+  std::atomic_bool & has_executor = node_ptr->get_associated_with_executor_atomic();
+  has_executor.store(false);
+  return true;
+}
+
+bool
+StaticExecutorEntitiesCollector::is_ready(rcl_wait_set_t * p_wait_set)
+{
+  // Check wait_set guard_conditions for added/removed entities to/from a node
+  for (size_t i = 0; i < p_wait_set->size_of_guard_conditions; ++i) {
+    if (p_wait_set->guard_conditions[i] != NULL) {
+      auto found_guard_condition = std::find_if(
+        weak_nodes_to_guard_conditions_.begin(), weak_nodes_to_guard_conditions_.end(),
+        [&](std::pair<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr,
+        const GuardCondition *> pair) -> bool {
+          const rcl_guard_condition_t & rcl_gc = pair.second->get_rcl_guard_condition();
+          return &rcl_gc == p_wait_set->guard_conditions[i];
+        });
+      if (found_guard_condition != weak_nodes_to_guard_conditions_.end()) {
+        return true;
+      }
+    }
+  }
+  // None of the guard conditions triggered belong to a registered node
+  return false;
+}
+
+// Returns true iff the weak_groups_to_nodes map has node_ptr as the value in any of its entry.
+bool
+StaticExecutorEntitiesCollector::has_node(
+  const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
+  weak_groups_to_nodes) const
+{
+  return std::find_if(
+    weak_groups_to_nodes.begin(),
+    weak_groups_to_nodes.end(),
+    [&](const WeakCallbackGroupsToNodesMap::value_type & other) -> bool {
+      auto other_ptr = other.second.lock();
+      return other_ptr == node_ptr;
+    }) != weak_groups_to_nodes.end();
+}
+
+void
+StaticExecutorEntitiesCollector::add_callback_groups_from_nodes_associated_to_executor()
+{
+  for (const auto & weak_node : weak_nodes_) {
+    auto node = weak_node.lock();
+    if (node) {
+      node->for_each_callback_group(
+        [this, node](rclcpp::CallbackGroup::SharedPtr shared_group_ptr)
+        {
+          if (shared_group_ptr->automatically_add_to_executor_with_node() &&
+          !shared_group_ptr->get_associated_with_executor_atomic().load())
+          {
+            add_callback_group(
+              shared_group_ptr,
+              node,
+              weak_groups_to_nodes_associated_with_executor_);
+          }
+        });
+    }
+  }
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticExecutorEntitiesCollector::get_all_callback_groups()
+{
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  for (const auto & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  for (const auto & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticExecutorEntitiesCollector::get_manually_added_callback_groups()
+{
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  for (const auto & group_node_ptr : weak_groups_associated_with_executor_to_nodes_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticExecutorEntitiesCollector::get_automatically_added_callback_groups_from_nodes()
+{
+  std::vector<rclcpp::CallbackGroup::WeakPtr> groups;
+  for (const auto & group_node_ptr : weak_groups_to_nodes_associated_with_executor_) {
+    groups.push_back(group_node_ptr.first);
+  }
+  return groups;
+}

rclcpp/src/rclcpp/executors/static_single_threaded_executor.cpp

@@ -12,21 +12,31 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "rclcpp/executors/executor_entities_collection.hpp"
-#include "rclcpp/executors/executor_notify_waitable.hpp"
+#include "rclcpp/executors/static_single_threaded_executor.hpp"
+
+#include <chrono>
+#include <memory>
+#include <utility>
+#include <vector>
+
 #include "rcpputils/scope_exit.hpp"
 
-#include "rclcpp/executors/static_single_threaded_executor.hpp"
-#include "rclcpp/any_executable.hpp"
-
 using rclcpp::executors::StaticSingleThreadedExecutor;
+using rclcpp::experimental::ExecutableList;
 
-StaticSingleThreadedExecutor::StaticSingleThreadedExecutor(const rclcpp::ExecutorOptions & options)
+StaticSingleThreadedExecutor::StaticSingleThreadedExecutor(
+  const rclcpp::ExecutorOptions & options)
 : rclcpp::Executor(options)
 {
+  entities_collector_ = std::make_shared<StaticExecutorEntitiesCollector>();
 }
 
-StaticSingleThreadedExecutor::~StaticSingleThreadedExecutor() {}
+StaticSingleThreadedExecutor::~StaticSingleThreadedExecutor()
+{
+  if (entities_collector_->is_init()) {
+    entities_collector_->fini();
+  }
+}
 
 void
 StaticSingleThreadedExecutor::spin()
@@ -36,25 +46,14 @@ StaticSingleThreadedExecutor::spin()
   }
   RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
 
-  // This is essentially the contents of the rclcpp::Executor::wait_for_work method,
-  // except we need to keep the wait result to reproduce the StaticSingleThreadedExecutor
-  // behavior.
+  // Set memory_strategy_ and exec_list_ based on weak_nodes_
+  // Prepare wait_set_ based on memory_strategy_
+  entities_collector_->init(&wait_set_, memory_strategy_);
+
   while (rclcpp::ok(this->context_) && spinning.load()) {
-    std::deque<rclcpp::AnyExecutable> to_exec;
-
-    std::lock_guard<std::mutex> guard(mutex_);
-    if (current_collection_.empty() || this->entities_need_rebuild_.load()) {
-      this->collect_entities();
-    }
-
-    auto wait_result = wait_set_.wait(std::chrono::nanoseconds(-1));
-    if (wait_result.kind() == WaitResultKind::Empty) {
-      RCUTILS_LOG_WARN_NAMED(
-        "rclcpp",
-        "empty wait set received in wait(). This should never happen.");
-      continue;
-    }
-    execute_ready_executables(current_collection_, wait_result, false);
+    // Refresh wait set and wait for work
+    entities_collector_->refresh_wait_set();
+    execute_ready_executables();
   }
 }
 
@@ -81,6 +80,11 @@ StaticSingleThreadedExecutor::spin_all(std::chrono::nanoseconds max_duration)
 void
 StaticSingleThreadedExecutor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
 {
+  // Make sure the entities collector has been initialized
+  if (!entities_collector_->is_init()) {
+    entities_collector_->init(&wait_set_, memory_strategy_);
+  }
+
   auto start = std::chrono::steady_clock::now();
   auto max_duration_not_elapsed = [max_duration, start]() {
       if (std::chrono::nanoseconds(0) == max_duration) {
@@ -101,21 +105,9 @@ StaticSingleThreadedExecutor::spin_some_impl(std::chrono::nanoseconds max_durati
 
   while (rclcpp::ok(context_) && spinning.load() && max_duration_not_elapsed()) {
     // Get executables that are ready now
-    std::lock_guard<std::mutex> guard(mutex_);
-    if (current_collection_.empty() || this->entities_need_rebuild_.load()) {
-      this->collect_entities();
-    }
-
-    auto wait_result = wait_set_.wait(std::chrono::nanoseconds(0));
-    if (wait_result.kind() == WaitResultKind::Empty) {
-      RCUTILS_LOG_WARN_NAMED(
-        "rclcpp",
-        "empty wait set received in wait(). This should never happen.");
-      continue;
-    }
-
+    entities_collector_->refresh_wait_set(std::chrono::milliseconds::zero());
     // Execute ready executables
-    bool work_available = execute_ready_executables(current_collection_, wait_result, false);
+    bool work_available = execute_ready_executables();
     if (!work_available || !exhaustive) {
       break;
     }
@@ -125,122 +117,164 @@ StaticSingleThreadedExecutor::spin_some_impl(std::chrono::nanoseconds max_durati
 void
 StaticSingleThreadedExecutor::spin_once_impl(std::chrono::nanoseconds timeout)
 {
+  // Make sure the entities collector has been initialized
+  if (!entities_collector_->is_init()) {
+    entities_collector_->init(&wait_set_, memory_strategy_);
+  }
+
   if (rclcpp::ok(context_) && spinning.load()) {
-    std::lock_guard<std::mutex> guard(mutex_);
-    if (current_collection_.empty() || this->entities_need_rebuild_.load()) {
-      this->collect_entities();
-    }
-
-    auto wait_result = wait_set_.wait(std::chrono::nanoseconds(timeout));
-    if (wait_result.kind() == WaitResultKind::Empty) {
-      RCUTILS_LOG_WARN_NAMED(
-        "rclcpp",
-        "empty wait set received in wait(). This should never happen.");
-      return;
-    }
-
+    // Wait until we have a ready entity or timeout expired
+    entities_collector_->refresh_wait_set(timeout);
     // Execute ready executables
-    execute_ready_executables(current_collection_, wait_result, true);
+    execute_ready_executables(true);
   }
 }
 
-// This preserves the "scheduling semantics" of the StaticSingleThreadedExecutor
-// from the original implementation.
-bool StaticSingleThreadedExecutor::execute_ready_executables(
-  const rclcpp::executors::ExecutorEntitiesCollection & collection,
-  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
-  bool spin_once)
+void
+StaticSingleThreadedExecutor::add_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr,
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  bool notify)
+{
+  bool is_new_node = entities_collector_->add_callback_group(group_ptr, node_ptr);
+  if (is_new_node && notify) {
+    // Interrupt waiting to handle new node
+    interrupt_guard_condition_->trigger();
+  }
+}
+
+void
+StaticSingleThreadedExecutor::add_node(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify)
+{
+  bool is_new_node = entities_collector_->add_node(node_ptr);
+  if (is_new_node && notify) {
+    // Interrupt waiting to handle new node
+    interrupt_guard_condition_->trigger();
+  }
+}
+
+void
+StaticSingleThreadedExecutor::add_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify)
+{
+  this->add_node(node_ptr->get_node_base_interface(), notify);
+}
+
+void
+StaticSingleThreadedExecutor::remove_callback_group(
+  rclcpp::CallbackGroup::SharedPtr group_ptr, bool notify)
+{
+  bool node_removed = entities_collector_->remove_callback_group(group_ptr);
+  // If the node was matched and removed, interrupt waiting
+  if (node_removed && notify) {
+    interrupt_guard_condition_->trigger();
+  }
+}
+
+void
+StaticSingleThreadedExecutor::remove_node(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify)
+{
+  bool node_removed = entities_collector_->remove_node(node_ptr);
+  if (!node_removed) {
+    throw std::runtime_error("Node needs to be associated with this executor.");
+  }
+  // If the node was matched and removed, interrupt waiting
+  if (notify) {
+    interrupt_guard_condition_->trigger();
+  }
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticSingleThreadedExecutor::get_all_callback_groups()
+{
+  return entities_collector_->get_all_callback_groups();
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticSingleThreadedExecutor::get_manually_added_callback_groups()
+{
+  return entities_collector_->get_manually_added_callback_groups();
+}
+
+std::vector<rclcpp::CallbackGroup::WeakPtr>
+StaticSingleThreadedExecutor::get_automatically_added_callback_groups_from_nodes()
+{
+  return entities_collector_->get_automatically_added_callback_groups_from_nodes();
+}
+
+void
+StaticSingleThreadedExecutor::remove_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify)
+{
+  this->remove_node(node_ptr->get_node_base_interface(), notify);
+}
+
+bool
+StaticSingleThreadedExecutor::execute_ready_executables(bool spin_once)
 {
   bool any_ready_executable = false;
 
-  if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
-    return any_ready_executable;
+  // Execute all the ready subscriptions
+  for (size_t i = 0; i < wait_set_.size_of_subscriptions; ++i) {
+    if (i < entities_collector_->get_number_of_subscriptions()) {
+      if (wait_set_.subscriptions[i]) {
+        execute_subscription(entities_collector_->get_subscription(i));
+        if (spin_once) {
+          return true;
+        }
+        any_ready_executable = true;
+      }
+    }
   }
-
-  auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
-
-  for (size_t ii = 0; ii < rcl_wait_set.size_of_timers; ++ii) {
-    if (nullptr == rcl_wait_set.timers[ii]) {continue;}
-    auto entity_iter = collection.timers.find(rcl_wait_set.timers[ii]);
-    if (entity_iter != collection.timers.end()) {
-      auto entity = entity_iter->second.entity.lock();
-      if (!entity) {
-        continue;
+  // Execute all the ready timers
+  for (size_t i = 0; i < wait_set_.size_of_timers; ++i) {
+    if (i < entities_collector_->get_number_of_timers()) {
+      if (wait_set_.timers[i] && entities_collector_->get_timer(i)->is_ready()) {
+        auto timer = entities_collector_->get_timer(i);
+        timer->call();
+        execute_timer(std::move(timer));
+        if (spin_once) {
+          return true;
+        }
+        any_ready_executable = true;
       }
-      if (!entity->call()) {
-        continue;
+    }
+  }
+  // Execute all the ready services
+  for (size_t i = 0; i < wait_set_.size_of_services; ++i) {
+    if (i < entities_collector_->get_number_of_services()) {
+      if (wait_set_.services[i]) {
+        execute_service(entities_collector_->get_service(i));
+        if (spin_once) {
+          return true;
+        }
+        any_ready_executable = true;
       }
-      execute_timer(entity);
+    }
+  }
+  // Execute all the ready clients
+  for (size_t i = 0; i < wait_set_.size_of_clients; ++i) {
+    if (i < entities_collector_->get_number_of_clients()) {
+      if (wait_set_.clients[i]) {
+        execute_client(entities_collector_->get_client(i));
+        if (spin_once) {
+          return true;
+        }
+        any_ready_executable = true;
+      }
+    }
+  }
+  // Execute all the ready waitables
+  for (size_t i = 0; i < entities_collector_->get_number_of_waitables(); ++i) {
+    auto waitable = entities_collector_->get_waitable(i);
+    if (waitable->is_ready(&wait_set_)) {
+      auto data = waitable->take_data();
+      waitable->execute(data);
       if (spin_once) {
         return true;
       }
       any_ready_executable = true;
     }
   }
-
-  for (size_t ii = 0; ii < rcl_wait_set.size_of_subscriptions; ++ii) {
-    if (nullptr == rcl_wait_set.subscriptions[ii]) {continue;}
-    auto entity_iter = collection.subscriptions.find(rcl_wait_set.subscriptions[ii]);
-    if (entity_iter != collection.subscriptions.end()) {
-      auto entity = entity_iter->second.entity.lock();
-      if (!entity) {
-        continue;
-      }
-      execute_subscription(entity);
-      if (spin_once) {
-        return true;
-      }
-      any_ready_executable = true;
-    }
-  }
-
-  for (size_t ii = 0; ii < rcl_wait_set.size_of_services; ++ii) {
-    if (nullptr == rcl_wait_set.services[ii]) {continue;}
-    auto entity_iter = collection.services.find(rcl_wait_set.services[ii]);
-    if (entity_iter != collection.services.end()) {
-      auto entity = entity_iter->second.entity.lock();
-      if (!entity) {
-        continue;
-      }
-      execute_service(entity);
-      if (spin_once) {
-        return true;
-      }
-      any_ready_executable = true;
-    }
-  }
-
-  for (size_t ii = 0; ii < rcl_wait_set.size_of_clients; ++ii) {
-    if (nullptr == rcl_wait_set.clients[ii]) {continue;}
-    auto entity_iter = collection.clients.find(rcl_wait_set.clients[ii]);
-    if (entity_iter != collection.clients.end()) {
-      auto entity = entity_iter->second.entity.lock();
-      if (!entity) {
-        continue;
-      }
-      execute_client(entity);
-      if (spin_once) {
-        return true;
-      }
-      any_ready_executable = true;
-    }
-  }
-
-  for (auto & [handle, entry] : collection.waitables) {
-    auto waitable = entry.entity.lock();
-    if (!waitable) {
-      continue;
-    }
-    if (!waitable->is_ready(&rcl_wait_set)) {
-      continue;
-    }
-
-    auto data = waitable->take_data();
-    waitable->execute(data);
-    if (spin_once) {
-      return true;
-    }
-    any_ready_executable = true;
-  }
   return any_ready_executable;
 }

rclcpp/src/rclcpp/experimental/executors/events_executor/events_executor.cpp

@@ -50,6 +50,9 @@ EventsExecutor::EventsExecutor(
   timers_manager_ =
     std::make_shared<rclcpp::experimental::TimersManager>(context_, timer_on_ready_cb);
 
+  this->current_entities_collection_ =
+    std::make_shared<rclcpp::executors::ExecutorEntitiesCollection>();
+
   notify_waitable_ = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(
     [this]() {
       // This callback is invoked when:
@@ -61,6 +64,10 @@ EventsExecutor::EventsExecutor(
       this->refresh_current_collection_from_callback_groups();
     });
 
+  // Make sure that the notify waitable is immediately added to the collection
+  // to avoid missing events
+  this->add_notify_waitable_to_collection(current_entities_collection_->waitables);
+
   notify_waitable_->add_guard_condition(interrupt_guard_condition_);
   notify_waitable_->add_guard_condition(shutdown_guard_condition_);
 
@@ -87,9 +94,6 @@ EventsExecutor::EventsExecutor(
 
   this->entities_collector_ =
     std::make_shared<rclcpp::executors::ExecutorEntitiesCollector>(notify_waitable_);
-
-  this->current_entities_collection_ =
-    std::make_shared<rclcpp::executors::ExecutorEntitiesCollection>();
 }
 
 EventsExecutor::~EventsExecutor()
@@ -202,11 +206,12 @@ EventsExecutor::spin_once_impl(std::chrono::nanoseconds timeout)
     timeout = std::chrono::nanoseconds::max();
   }
 
-  // Select the smallest between input timeout and timer timeout
+  // Select the smallest between input timeout and timer timeout.
+  // Cancelled timers are not considered.
   bool is_timer_timeout = false;
   auto next_timer_timeout = timers_manager_->get_head_timeout();
-  if (next_timer_timeout < timeout) {
-    timeout = next_timer_timeout;
+  if (next_timer_timeout.has_value() && next_timer_timeout.value() < timeout) {
+    timeout = next_timer_timeout.value();
     is_timer_timeout = true;
   }
 
@@ -269,10 +274,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
   switch (event.type) {
     case ExecutorEventType::CLIENT_EVENT:
       {
-        auto client = this->retrieve_entity(
-          static_cast<const rcl_client_t *>(event.entity_key),
-          current_entities_collection_->clients);
-
+        rclcpp::ClientBase::SharedPtr client;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          client = this->retrieve_entity(
+            static_cast<const rcl_client_t *>(event.entity_key),
+            current_entities_collection_->clients);
+        }
         if (client) {
           for (size_t i = 0; i < event.num_events; i++) {
             execute_client(client);
@@ -283,9 +291,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
       }
     case ExecutorEventType::SUBSCRIPTION_EVENT:
       {
-        auto subscription = this->retrieve_entity(
-          static_cast<const rcl_subscription_t *>(event.entity_key),
-          current_entities_collection_->subscriptions);
+        rclcpp::SubscriptionBase::SharedPtr subscription;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          subscription = this->retrieve_entity(
+            static_cast<const rcl_subscription_t *>(event.entity_key),
+            current_entities_collection_->subscriptions);
+        }
         if (subscription) {
           for (size_t i = 0; i < event.num_events; i++) {
             execute_subscription(subscription);
@@ -295,10 +307,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
       }
     case ExecutorEventType::SERVICE_EVENT:
       {
-        auto service = this->retrieve_entity(
-          static_cast<const rcl_service_t *>(event.entity_key),
-          current_entities_collection_->services);
-
+        rclcpp::ServiceBase::SharedPtr service;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          service = this->retrieve_entity(
+            static_cast<const rcl_service_t *>(event.entity_key),
+            current_entities_collection_->services);
+        }
         if (service) {
           for (size_t i = 0; i < event.num_events; i++) {
             execute_service(service);
@@ -315,9 +330,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
       }
     case ExecutorEventType::WAITABLE_EVENT:
       {
-        auto waitable = this->retrieve_entity(
-          static_cast<const rclcpp::Waitable *>(event.entity_key),
-          current_entities_collection_->waitables);
+        rclcpp::Waitable::SharedPtr waitable;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          waitable = this->retrieve_entity(
+            static_cast<const rclcpp::Waitable *>(event.entity_key),
+            current_entities_collection_->waitables);
+        }
         if (waitable) {
           for (size_t i = 0; i < event.num_events; i++) {
             auto data = waitable->take_data_by_entity_id(event.waitable_data);
@@ -382,6 +401,7 @@ EventsExecutor::get_automatically_added_callback_groups_from_nodes()
 void
 EventsExecutor::refresh_current_collection_from_callback_groups()
 {
+  // Build the new collection
   this->entities_collector_->update_collections();
   auto callback_groups = this->entities_collector_->get_all_callback_groups();
   rclcpp::executors::ExecutorEntitiesCollection new_collection;
@@ -395,18 +415,11 @@ EventsExecutor::refresh_current_collection_from_callback_groups()
   // retrieved in the "standard" way.
   // To do it, we need to add the notify waitable as an entry in both the new and
   // current collections such that it's neither added or removed.
-  rclcpp::CallbackGroup::WeakPtr weak_group_ptr;
-  new_collection.waitables.insert(
-  {
-    this->notify_waitable_.get(),
-    {this->notify_waitable_, weak_group_ptr}
-  });
+  this->add_notify_waitable_to_collection(new_collection.waitables);
 
-  this->current_entities_collection_->waitables.insert(
-  {
-    this->notify_waitable_.get(),
-    {this->notify_waitable_, weak_group_ptr}
-  });
+  // Acquire lock before modifying the current collection
+  std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+  this->add_notify_waitable_to_collection(current_entities_collection_->waitables);
 
   this->refresh_current_collection(new_collection);
 }
@@ -415,6 +428,9 @@ void
 EventsExecutor::refresh_current_collection(
   const rclcpp::executors::ExecutorEntitiesCollection & new_collection)
 {
+  // Acquire lock before modifying the current collection
+  std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+
   current_entities_collection_->timers.update(
     new_collection.timers,
     [this](rclcpp::TimerBase::SharedPtr timer) {timers_manager_->add_timer(timer);},
@@ -486,3 +502,16 @@ EventsExecutor::create_waitable_callback(const rclcpp::Waitable * entity_key)
     };
   return callback;
 }
+
+void
+EventsExecutor::add_notify_waitable_to_collection(
+  rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection)
+{
+  // The notify waitable is not associated to any group, so use an invalid one
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr;
+  collection.insert(
+  {
+    this->notify_waitable_.get(),
+    {this->notify_waitable_, weak_group_ptr}
+  });
+}

rclcpp/src/rclcpp/experimental/timers_manager.cpp

@@ -28,9 +28,9 @@ using rclcpp::experimental::TimersManager;
 TimersManager::TimersManager(
   std::shared_ptr<rclcpp::Context> context,
   std::function<void(const rclcpp::TimerBase *)> on_ready_callback)
+: on_ready_callback_(on_ready_callback),
+  context_(context)
 {
-  context_ = context;
-  on_ready_callback_ = on_ready_callback;
 }
 
 TimersManager::~TimersManager()
@@ -100,7 +100,7 @@ void TimersManager::stop()
   }
 }
 
-std::chrono::nanoseconds TimersManager::get_head_timeout()
+std::optional<std::chrono::nanoseconds> TimersManager::get_head_timeout()
 {
   // Do not allow to interfere with the thread running
   if (running_) {
@@ -169,7 +169,7 @@ void TimersManager::execute_ready_timer(const rclcpp::TimerBase * timer_id)
   }
 }
 
-std::chrono::nanoseconds TimersManager::get_head_timeout_unsafe()
+std::optional<std::chrono::nanoseconds> TimersManager::get_head_timeout_unsafe()
 {
   // If we don't have any weak pointer, then we just return maximum timeout
   if (weak_timers_heap_.empty()) {
@@ -191,7 +191,9 @@ std::chrono::nanoseconds TimersManager::get_head_timeout_unsafe()
     }
     head_timer = locked_heap.front();
   }
-
+  if (head_timer->is_canceled()) {
+    return std::nullopt;
+  }
   return head_timer->time_until_trigger();
 }
 
@@ -242,17 +244,34 @@ void TimersManager::run_timers()
     // Lock mutex
     std::unique_lock<std::mutex> lock(timers_mutex_);
 
-    std::chrono::nanoseconds time_to_sleep = get_head_timeout_unsafe();
+    std::optional<std::chrono::nanoseconds> time_to_sleep = get_head_timeout_unsafe();
 
-    // No need to wait if a timer is already available
-    if (time_to_sleep > std::chrono::nanoseconds::zero()) {
-      if (time_to_sleep != std::chrono::nanoseconds::max()) {
-        // Wait until timeout or notification that timers have been updated
-        timers_cv_.wait_for(lock, time_to_sleep, [this]() {return timers_updated_;});
-      } else {
-        // Wait until notification that timers have been updated
-        timers_cv_.wait(lock, [this]() {return timers_updated_;});
-      }
+    // If head timer was cancelled, try to reheap and get a new head.
+    // This avoids an edge condition where head timer is cancelled, but other
+    // valid timers remain in the heap.
+    if (!time_to_sleep.has_value()) {
+      // Re-heap to (possibly) move cancelled timer from head of heap. If
+      // entire heap is cancelled, this will still result in a nullopt.
+      TimersHeap locked_heap = weak_timers_heap_.validate_and_lock();
+      locked_heap.heapify();
+      weak_timers_heap_.store(locked_heap);
+      time_to_sleep = get_head_timeout_unsafe();
+    }
+
+    // If no timers, or all timers cancelled, wait for an update.
+    if (!time_to_sleep.has_value() || (time_to_sleep.value() == std::chrono::nanoseconds::max()) ) {
+      // Wait until notification that timers have been updated
+      timers_cv_.wait(lock, [this]() {return timers_updated_;});
+
+      // Re-heap in case ordering changed due to a cancelled timer
+      // re-activating.
+      TimersHeap locked_heap = weak_timers_heap_.validate_and_lock();
+      locked_heap.heapify();
+      weak_timers_heap_.store(locked_heap);
+    } else if (time_to_sleep.value() != std::chrono::nanoseconds::zero()) {
+      // If time_to_sleep is zero, we immediately execute. Otherwise, wait
+      // until timeout or notification that timers have been updated
+      timers_cv_.wait_for(lock, time_to_sleep.value(), [this]() {return timers_updated_;});
     }
 
     // Reset timers updated flag

rclcpp/src/rclcpp/generic_subscription.cpp

@@ -49,9 +49,9 @@ GenericSubscription::handle_message(
 void
 GenericSubscription::handle_serialized_message(
   const std::shared_ptr<rclcpp::SerializedMessage> & message,
-  const rclcpp::MessageInfo &)
+  const rclcpp::MessageInfo & message_info)
 {
-  callback_(message);
+  callback_(message, message_info);
 }
 
 void

rclcpp/src/rclcpp/guard_condition.cpp

@@ -23,16 +23,17 @@ namespace rclcpp
 {
 
 GuardCondition::GuardCondition(
-  rclcpp::Context::SharedPtr context,
+  const rclcpp::Context::SharedPtr & context,
   rcl_guard_condition_options_t guard_condition_options)
-: context_(context), rcl_guard_condition_{rcl_get_zero_initialized_guard_condition()}
+: rcl_guard_condition_{rcl_get_zero_initialized_guard_condition()}
 {
-  if (!context_) {
+  if (!context) {
     throw std::invalid_argument("context argument unexpectedly nullptr");
   }
+
   rcl_ret_t ret = rcl_guard_condition_init(
     &this->rcl_guard_condition_,
-    context_->get_rcl_context().get(),
+    context->get_rcl_context().get(),
     guard_condition_options);
   if (RCL_RET_OK != ret) {
     rclcpp::exceptions::throw_from_rcl_error(ret, "failed to create guard condition");
@@ -53,12 +54,6 @@ GuardCondition::~GuardCondition()
   }
 }
 
-rclcpp::Context::SharedPtr
-GuardCondition::get_context() const
-{
-  return context_;
-}
-
 rcl_guard_condition_t &
 GuardCondition::get_rcl_guard_condition()
 {

rclcpp/src/rclcpp/intra_process_manager.cpp

@@ -32,13 +32,24 @@ IntraProcessManager::~IntraProcessManager()
 {}
 
 uint64_t
-IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
+IntraProcessManager::add_publisher(
+  rclcpp::PublisherBase::SharedPtr publisher,
+  rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr buffer)
 {
   std::unique_lock<std::shared_timed_mutex> lock(mutex_);
 
   uint64_t pub_id = IntraProcessManager::get_next_unique_id();
 
   publishers_[pub_id] = publisher;
+  if (publisher->is_durability_transient_local()) {
+    if (buffer) {
+      publisher_buffers_[pub_id] = buffer;
+    } else {
+      throw std::runtime_error(
+              "transient_local publisher needs to pass"
+              "a valid publisher buffer ptr when calling add_publisher()");
+    }
+  }
 
   // Initialize the subscriptions storage for this publisher.
   pub_to_subs_[pub_id] = SplittedSubscriptions();
@@ -58,30 +69,6 @@ IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
   return pub_id;
 }
 
-uint64_t
-IntraProcessManager::add_subscription(SubscriptionIntraProcessBase::SharedPtr subscription)
-{
-  std::unique_lock<std::shared_timed_mutex> lock(mutex_);
-
-  uint64_t sub_id = IntraProcessManager::get_next_unique_id();
-
-  subscriptions_[sub_id] = subscription;
-
-  // adds the subscription id to all the matchable publishers
-  for (auto & pair : publishers_) {
-    auto publisher = pair.second.lock();
-    if (!publisher) {
-      continue;
-    }
-    if (can_communicate(publisher, subscription)) {
-      uint64_t pub_id = pair.first;
-      insert_sub_id_for_pub(sub_id, pub_id, subscription->use_take_shared_method());
-    }
-  }
-
-  return sub_id;
-}
-
 void
 IntraProcessManager::remove_subscription(uint64_t intra_process_subscription_id)
 {
@@ -112,6 +99,7 @@ IntraProcessManager::remove_publisher(uint64_t intra_process_publisher_id)
   std::unique_lock<std::shared_timed_mutex> lock(mutex_);
 
   publishers_.erase(intra_process_publisher_id);
+  publisher_buffers_.erase(intra_process_publisher_id);
   pub_to_subs_.erase(intra_process_publisher_id);
 }
 
@@ -225,5 +213,52 @@ IntraProcessManager::can_communicate(
   return true;
 }
 
+size_t
+IntraProcessManager::lowest_available_capacity(const uint64_t intra_process_publisher_id) const
+{
+  size_t capacity = std::numeric_limits<size_t>::max();
+
+  auto publisher_it = pub_to_subs_.find(intra_process_publisher_id);
+  if (publisher_it == pub_to_subs_.end()) {
+    // Publisher is either invalid or no longer exists.
+    RCLCPP_WARN(
+      rclcpp::get_logger("rclcpp"),
+      "Calling lowest_available_capacity for invalid or no longer existing publisher id");
+    return 0u;
+  }
+
+  if (publisher_it->second.take_shared_subscriptions.empty() &&
+    publisher_it->second.take_ownership_subscriptions.empty())
+  {
+    // no subscriptions available
+    return 0u;
+  }
+
+  auto available_capacity = [this, &capacity](const uint64_t intra_process_subscription_id)
+    {
+      auto subscription_it = subscriptions_.find(intra_process_subscription_id);
+      if (subscription_it != subscriptions_.end()) {
+        auto subscription = subscription_it->second.lock();
+        if (subscription) {
+          capacity = std::min(capacity, subscription->available_capacity());
+        }
+      } else {
+        // Subscription is either invalid or no longer exists.
+        RCLCPP_WARN(
+          rclcpp::get_logger("rclcpp"),
+          "Calling available_capacity for invalid or no longer existing subscription id");
+      }
+    };
+
+  for (const auto sub_id : publisher_it->second.take_shared_subscriptions) {
+    available_capacity(sub_id);
+  }
+
+  for (const auto sub_id : publisher_it->second.take_ownership_subscriptions) {
+    available_capacity(sub_id);
+  }
+
+  return capacity;
+}
 }  // namespace experimental
 }  // namespace rclcpp

rclcpp/src/rclcpp/logger.cpp

@@ -17,6 +17,7 @@
 #include <utility>
 
 #include "rcl_logging_interface/rcl_logging_interface.h"
+#include "rcl/error_handling.h"
 #include "rcl/logging_rosout.h"
 
 #include "rclcpp/exceptions.hpp"
@@ -80,10 +81,12 @@ Logger::get_child(const std::string & suffix)
   {
     std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
     rcl_ret = rcl_logging_rosout_add_sublogger((*name_).c_str(), suffix.c_str());
-    if (RCL_RET_OK != rcl_ret) {
+    if (RCL_RET_NOT_FOUND == rcl_ret) {
+      rcl_reset_error();
+    } else if (RCL_RET_OK != rcl_ret) {
       exceptions::throw_from_rcl_error(
         rcl_ret, "failed to call rcl_logging_rosout_add_sublogger",
-        rcutils_get_error_state(), rcutils_reset_error);
+        rcl_get_error_state(), rcl_reset_error);
     }
   }
 
@@ -98,9 +101,7 @@ Logger::get_child(const std::string & suffix)
           logger_sublogger_pairname_ptr->second.c_str());
         delete logger_sublogger_pairname_ptr;
         if (RCL_RET_OK != rcl_ret) {
-          exceptions::throw_from_rcl_error(
-            rcl_ret, "failed to call rcl_logging_rosout_remove_sublogger",
-            rcutils_get_error_state(), rcutils_reset_error);
+          rcl_reset_error();
         }
       });
   }

rclcpp/src/rclcpp/logging_mutex.cpp

@@ -14,6 +14,7 @@
 
 #include <memory>
 #include <mutex>
+#include <stdexcept>
 
 #include "rcutils/macros.h"
 

rclcpp/src/rclcpp/node.cpp

@@ -36,6 +36,7 @@
 #include "rclcpp/node_interfaces/node_time_source.hpp"
 #include "rclcpp/node_interfaces/node_timers.hpp"
 #include "rclcpp/node_interfaces/node_topics.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
 #include "rclcpp/node_interfaces/node_waitables.hpp"
 #include "rclcpp/qos_overriding_options.hpp"
 
@@ -109,6 +110,22 @@ create_effective_namespace(const std::string & node_namespace, const std::string
 
 }  // namespace
 
+/// Internal implementation to provide hidden and API/ABI stable changes to the Node.
+/**
+ * This class is intended to be an "escape hatch" within a stable distribution, so that certain
+ * smaller features and bugfixes can be backported, having a place to put new members, while
+ * maintaining the ABI.
+ *
+ * This is not intended to be a parking place for new features, it should be used for backports
+ * only, left empty and unallocated in Rolling.
+ */
+class Node::NodeImpl
+{
+public:
+  NodeImpl() = default;
+  ~NodeImpl() = default;
+};
+
 Node::Node(
   const std::string & node_name,
   const NodeOptions & options)
@@ -167,7 +184,7 @@ Node::Node(
       options.use_intra_process_comms(),
       options.enable_topic_statistics())),
   node_graph_(new rclcpp::node_interfaces::NodeGraph(node_base_.get())),
-  node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_.get())),
+  node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_)),
   node_timers_(new rclcpp::node_interfaces::NodeTimers(node_base_.get())),
   node_topics_(new rclcpp::node_interfaces::NodeTopics(node_base_.get(), node_timers_.get())),
   node_services_(new rclcpp::node_interfaces::NodeServices(node_base_.get())),
@@ -206,6 +223,12 @@ Node::Node(
       options.clock_qos(),
       options.use_clock_thread()
     )),
+  node_type_descriptions_(new rclcpp::node_interfaces::NodeTypeDescriptions(
+      node_base_,
+      node_logging_,
+      node_parameters_,
+      node_services_
+    )),
   node_waitables_(new rclcpp::node_interfaces::NodeWaitables(node_base_.get())),
   node_options_(options),
   sub_namespace_(""),
@@ -225,6 +248,10 @@ Node::Node(
     node_topics_->resolve_topic_name("/parameter_events"),
     options.parameter_event_qos(),
     rclcpp::detail::PublisherQosParametersTraits{});
+
+  if (options.enable_logger_service()) {
+    node_logging_->create_logger_services(node_services_);
+  }
 }
 
 Node::Node(
@@ -242,7 +269,8 @@ Node::Node(
   node_waitables_(other.node_waitables_),
   node_options_(other.node_options_),
   sub_namespace_(extend_sub_namespace(other.get_sub_namespace(), sub_namespace)),
-  effective_namespace_(create_effective_namespace(other.get_namespace(), sub_namespace_))
+  effective_namespace_(create_effective_namespace(other.get_namespace(), sub_namespace_)),
+  hidden_impl_(other.hidden_impl_)
 {
   // Validate new effective namespace.
   int validation_result;
@@ -494,6 +522,18 @@ Node::count_subscribers(const std::string & topic_name) const
   return node_graph_->count_subscribers(topic_name);
 }
 
+size_t
+Node::count_clients(const std::string & service_name) const
+{
+  return node_graph_->count_clients(service_name);
+}
+
+size_t
+Node::count_services(const std::string & service_name) const
+{
+  return node_graph_->count_services(service_name);
+}
+
 std::vector<rclcpp::TopicEndpointInfo>
 Node::get_publishers_info_by_topic(const std::string & topic_name, bool no_mangle) const
 {
@@ -587,6 +627,12 @@ Node::get_node_topics_interface()
   return node_topics_;
 }
 
+rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
+Node::get_node_type_descriptions_interface()
+{
+  return node_type_descriptions_;
+}
+
 rclcpp::node_interfaces::NodeServicesInterface::SharedPtr
 Node::get_node_services_interface()
 {

rclcpp/src/rclcpp/node_interfaces/node_base.cpp

@@ -20,6 +20,9 @@
 #include "rclcpp/node_interfaces/node_base.hpp"
 
 #include "rcl/arguments.h"
+#include "rcl/node_type_cache.h"
+#include "rcl/logging.h"
+#include "rcl/logging_rosout.h"
 #include "rclcpp/exceptions.hpp"
 #include "rcutils/logging_macros.h"
 #include "rmw/validate_namespace.h"
@@ -54,17 +57,12 @@ NodeBase::NodeBase(
   std::shared_ptr<std::recursive_mutex> logging_mutex = get_global_logging_mutex();
 
   rcl_ret_t ret;
-  {
-    std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
-    // TODO(ivanpauno): /rosout Qos should be reconfigurable.
-    // TODO(ivanpauno): Instead of mutually excluding rcl_node_init with the global logger mutex,
-    // rcl_logging_rosout_init_publisher_for_node could be decoupled from there and be called
-    // here directly.
-    ret = rcl_node_init(
-      rcl_node.get(),
-      node_name.c_str(), namespace_.c_str(),
-      context_->get_rcl_context().get(), &rcl_node_options);
-  }
+
+  // TODO(ivanpauno): /rosout Qos should be reconfigurable.
+  ret = rcl_node_init(
+    rcl_node.get(),
+    node_name.c_str(), namespace_.c_str(),
+    context_->get_rcl_context().get(), &rcl_node_options);
   if (ret != RCL_RET_OK) {
     if (ret == RCL_RET_NODE_INVALID_NAME) {
       rcl_reset_error();  // discard rcl_node_init error
@@ -114,13 +112,29 @@ NodeBase::NodeBase(
     throw_from_rcl_error(ret, "failed to initialize rcl node");
   }
 
+  // The initialization for the rosout publisher
+  if (rcl_logging_rosout_enabled() && rcl_node_options.enable_rosout) {
+    std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
+    ret = rcl_logging_rosout_init_publisher_for_node(rcl_node.get());
+    if (ret != RCL_RET_OK) {
+      throw_from_rcl_error(ret, "failed to initialize rosout publisher");
+    }
+  }
+
   node_handle_.reset(
     rcl_node.release(),
-    [logging_mutex](rcl_node_t * node) -> void {
-      std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
-      // TODO(ivanpauno): Instead of mutually excluding rcl_node_fini with the global logger mutex,
-      // rcl_logging_rosout_fini_publisher_for_node could be decoupled from there and be called
-      // here directly.
+    [logging_mutex, rcl_node_options](rcl_node_t * node) -> void {
+      {
+        std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
+        if (rcl_logging_rosout_enabled() && rcl_node_options.enable_rosout) {
+          rcl_ret_t ret = rcl_logging_rosout_fini_publisher_for_node(node);
+          if (ret != RCL_RET_OK) {
+            RCUTILS_LOG_ERROR_NAMED(
+              "rclcpp",
+              "Error in destruction of rosout publisher: %s", rcl_get_error_string().str);
+          }
+        }
+      }
       if (rcl_node_fini(node) != RCL_RET_OK) {
         RCUTILS_LOG_ERROR_NAMED(
           "rclcpp",
@@ -204,14 +218,9 @@ NodeBase::create_callback_group(
   rclcpp::CallbackGroupType group_type,
   bool automatically_add_to_executor_with_node)
 {
-  auto weak_context = this->get_context()->weak_from_this();
-  auto get_node_context = [weak_context]() -> rclcpp::Context::SharedPtr {
-      return weak_context.lock();
-    };
-
   auto group = std::make_shared<rclcpp::CallbackGroup>(
     group_type,
-    get_node_context,
+    context_->weak_from_this(),
     automatically_add_to_executor_with_node);
   std::lock_guard<std::mutex> lock(callback_groups_mutex_);
   callback_groups_.push_back(group);

rclcpp/src/rclcpp/node_interfaces/node_graph.cpp

@@ -498,6 +498,50 @@ NodeGraph::count_subscribers(const std::string & topic_name) const
   return count;
 }
 
+size_t
+NodeGraph::count_clients(const std::string & service_name) const
+{
+  auto rcl_node_handle = node_base_->get_rcl_node_handle();
+
+  auto fqdn = rclcpp::expand_topic_or_service_name(
+    service_name,
+    rcl_node_get_name(rcl_node_handle),
+    rcl_node_get_namespace(rcl_node_handle),
+    true);
+
+  size_t count;
+  auto ret = rcl_count_clients(rcl_node_handle, fqdn.c_str(), &count);
+  if (ret != RMW_RET_OK) {
+    // *INDENT-OFF*
+    throw std::runtime_error(
+      std::string("could not count clients: ") + rmw_get_error_string().str);
+    // *INDENT-ON*
+  }
+  return count;
+}
+
+size_t
+NodeGraph::count_services(const std::string & service_name) const
+{
+  auto rcl_node_handle = node_base_->get_rcl_node_handle();
+
+  auto fqdn = rclcpp::expand_topic_or_service_name(
+    service_name,
+    rcl_node_get_name(rcl_node_handle),
+    rcl_node_get_namespace(rcl_node_handle),
+    true);
+
+  size_t count;
+  auto ret = rcl_count_services(rcl_node_handle, fqdn.c_str(), &count);
+  if (ret != RMW_RET_OK) {
+    // *INDENT-OFF*
+    throw std::runtime_error(
+      std::string("could not count services: ") + rmw_get_error_string().str);
+    // *INDENT-ON*
+  }
+  return count;
+}
+
 const rcl_guard_condition_t *
 NodeGraph::get_graph_guard_condition() const
 {

rclcpp/src/rclcpp/node_interfaces/node_logging.cpp

@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "rclcpp/node_impl.hpp"
 #include "rclcpp/node_interfaces/node_logging.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"
 
 using rclcpp::node_interfaces::NodeLogging;
 
-NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base)
+NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base)
 : node_base_(node_base)
 {
   logger_ = rclcpp::get_logger(NodeLogging::get_logger_name());
@@ -37,3 +39,55 @@ NodeLogging::get_logger_name() const
 {
   return rcl_node_get_logger_name(node_base_->get_rcl_node_handle());
 }
+
+void NodeLogging::create_logger_services(
+  node_interfaces::NodeServicesInterface::SharedPtr node_services)
+{
+  rclcpp::ServicesQoS qos_profile;
+  const std::string node_name = node_base_->get_name();
+  auto callback_group = node_base_->get_default_callback_group();
+
+  get_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::GetLoggerLevels>(
+    node_base_, node_services,
+    node_name + "/get_logger_levels",
+    [](
+      const std::shared_ptr<rmw_request_id_t>,
+      const std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Request> request,
+      std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Response> response)
+    {
+      for (auto & name : request->names) {
+        rcl_interfaces::msg::LoggerLevel logger_level;
+        logger_level.name = name;
+        auto ret = rcutils_logging_get_logger_level(name.c_str());
+        if (ret < 0) {
+          logger_level.level = 0;
+        } else {
+          logger_level.level = static_cast<uint8_t>(ret);
+        }
+        response->levels.push_back(std::move(logger_level));
+      }
+    },
+    qos_profile, callback_group);
+
+  set_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::SetLoggerLevels>(
+    node_base_, node_services,
+    node_name + "/set_logger_levels",
+    [](
+      const std::shared_ptr<rmw_request_id_t>,
+      const std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Request> request,
+      std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Response> response)
+    {
+      rcl_interfaces::msg::SetLoggerLevelsResult result;
+      for (auto & level : request->levels) {
+        auto ret = rcutils_logging_set_logger_level(level.name.c_str(), level.level);
+        if (ret != RCUTILS_RET_OK) {
+          result.successful = false;
+          result.reason = rcutils_get_error_string().str;
+        } else {
+          result.successful = true;
+        }
+        response->results.push_back(std::move(result));
+      }
+    },
+    qos_profile, callback_group);
+}

rclcpp/src/rclcpp/node_interfaces/node_parameters.cpp

@@ -1038,37 +1038,50 @@ NodeParameters::list_parameters(const std::vector<std::string> & prefixes, uint6
   // TODO(mikaelarguedas) define parameter separator different from "/" to avoid ambiguity
   // using "." for now
   const char * separator = ".";
-  for (auto & kv : parameters_) {
-    bool get_all = (prefixes.size() == 0) &&
-      ((depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) ||
-      (static_cast<uint64_t>(std::count(kv.first.begin(), kv.first.end(), *separator)) < depth));
-    bool prefix_matches = std::any_of(
-      prefixes.cbegin(), prefixes.cend(),
-      [&kv, &depth, &separator](const std::string & prefix) {
-        if (kv.first == prefix) {
-          return true;
-        } else if (kv.first.find(prefix + separator) == 0) {
-          size_t length = prefix.length();
-          std::string substr = kv.first.substr(length);
-          // Cast as unsigned integer to avoid warning
-          return (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) ||
-          (static_cast<uint64_t>(std::count(substr.begin(), substr.end(), *separator)) < depth);
-        }
-        return false;
-      });
-    if (get_all || prefix_matches) {
-      result.names.push_back(kv.first);
-      size_t last_separator = kv.first.find_last_of(separator);
-      if (std::string::npos != last_separator) {
-        std::string prefix = kv.first.substr(0, last_separator);
-        if (
-          std::find(result.prefixes.cbegin(), result.prefixes.cend(), prefix) ==
-          result.prefixes.cend())
-        {
-          result.prefixes.push_back(prefix);
+
+  auto separators_less_than_depth = [&depth, &separator](const std::string & str) -> bool {
+      return static_cast<uint64_t>(std::count(str.begin(), str.end(), *separator)) < depth;
+    };
+
+  bool recursive = (prefixes.size() == 0) &&
+    (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE);
+
+  for (const std::pair<const std::string, ParameterInfo> & kv : parameters_) {
+    if (!recursive) {
+      bool get_all = (prefixes.size() == 0) && separators_less_than_depth(kv.first);
+      if (!get_all) {
+        bool prefix_matches = std::any_of(
+          prefixes.cbegin(), prefixes.cend(),
+          [&kv, &depth, &separator, &separators_less_than_depth](const std::string & prefix) {
+            if (kv.first == prefix) {
+              return true;
+            } else if (kv.first.find(prefix + separator) == 0) {
+              if (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) {
+                return true;
+              }
+              std::string substr = kv.first.substr(prefix.length() + 1);
+              return separators_less_than_depth(substr);
+            }
+            return false;
+          });
+
+        if (!prefix_matches) {
+          continue;
         }
       }
     }
+
+    result.names.push_back(kv.first);
+    size_t last_separator = kv.first.find_last_of(separator);
+    if (std::string::npos != last_separator) {
+      std::string prefix = kv.first.substr(0, last_separator);
+      if (
+        std::find(result.prefixes.cbegin(), result.prefixes.cend(), prefix) ==
+        result.prefixes.cend())
+      {
+        result.prefixes.push_back(prefix);
+      }
+    }
   }
   return result;
 }

rclcpp/src/rclcpp/node_interfaces/node_services.cpp

@@ -32,8 +32,7 @@ NodeServices::add_service(
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create service, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("service");
     }
   } else {
     group = node_base_->get_default_callback_group();
@@ -58,8 +57,7 @@ NodeServices::add_client(
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create client, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("client");
     }
   } else {
     group = node_base_->get_default_callback_group();

rclcpp/src/rclcpp/node_interfaces/node_timers.cpp

@@ -34,8 +34,7 @@ NodeTimers::add_timer(
 {
   if (callback_group) {
     if (!node_base_->callback_group_in_node(callback_group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create timer, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("timer");
     }
   } else {
     callback_group = node_base_->get_default_callback_group();
@@ -50,7 +49,7 @@ NodeTimers::add_timer(
             std::string("failed to notify wait set on timer creation: ") + ex.what());
   }
 
-  TRACEPOINT(
+  TRACETOOLS_TRACEPOINT(
     rclcpp_timer_link_node,
     static_cast<const void *>(timer->get_timer_handle().get()),
     static_cast<const void *>(node_base_->get_rcl_node_handle()));

rclcpp/src/rclcpp/node_interfaces/node_topics.cpp

@@ -58,7 +58,7 @@ NodeTopics::add_publisher(
   // Assign to a group.
   if (callback_group) {
     if (!node_base_->callback_group_in_node(callback_group)) {
-      throw std::runtime_error("Cannot create publisher, callback group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("publisher");
     }
   } else {
     callback_group = node_base_->get_default_callback_group();
@@ -97,8 +97,7 @@ NodeTopics::add_subscription(
   // Assign to a group.
   if (callback_group) {
     if (!node_base_->callback_group_in_node(callback_group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create subscription, callback group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("subscription");
     }
   } else {
     callback_group = node_base_->get_default_callback_group();

rclcpp/src/rclcpp/node_interfaces/node_type_descriptions.cpp

@@ -0,0 +1,153 @@
+// Copyright 2023 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.
+
+#include <memory>
+#include <string>
+#include <thread>
+
+#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
+#include "rclcpp/parameter_client.hpp"
+
+#include "type_description_interfaces/srv/get_type_description.h"
+
+namespace
+{
+// Helper wrapper for rclcpp::Service to access ::Request and ::Response types for allocation.
+struct GetTypeDescription__C
+{
+  using Request = type_description_interfaces__srv__GetTypeDescription_Request;
+  using Response = type_description_interfaces__srv__GetTypeDescription_Response;
+  using Event = type_description_interfaces__srv__GetTypeDescription_Event;
+};
+}  // namespace
+
+// Helper function for C typesupport.
+namespace rosidl_typesupport_cpp
+{
+template<>
+rosidl_service_type_support_t const *
+get_service_type_support_handle<GetTypeDescription__C>()
+{
+  return ROSIDL_GET_SRV_TYPE_SUPPORT(type_description_interfaces, srv, GetTypeDescription);
+}
+}  // namespace rosidl_typesupport_cpp
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+class NodeTypeDescriptions::NodeTypeDescriptionsImpl
+{
+public:
+  using ServiceT = GetTypeDescription__C;
+
+  rclcpp::Logger logger_;
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
+  rclcpp::Service<ServiceT>::SharedPtr type_description_srv_;
+
+  NodeTypeDescriptionsImpl(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
+    rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
+    rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
+    rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
+  : logger_(node_logging->get_logger()),
+    node_base_(node_base)
+  {
+    const std::string enable_param_name = "start_type_description_service";
+
+    bool enabled = false;
+    try {
+      auto enable_param = node_parameters->declare_parameter(
+        enable_param_name,
+        rclcpp::ParameterValue(true),
+        rcl_interfaces::msg::ParameterDescriptor()
+        .set__name(enable_param_name)
+        .set__type(rclcpp::PARAMETER_BOOL)
+        .set__description("Start the ~/get_type_description service for this node.")
+        .set__read_only(true));
+      enabled = enable_param.get<bool>();
+    } catch (const rclcpp::exceptions::InvalidParameterTypeException & exc) {
+      RCLCPP_ERROR(logger_, "%s", exc.what());
+      throw;
+    }
+
+    if (enabled) {
+      auto * rcl_node = node_base->get_rcl_node_handle();
+      std::shared_ptr<rcl_service_t> rcl_srv(
+        new rcl_service_t,
+        [rcl_node, logger = this->logger_](rcl_service_t * service)
+        {
+          if (rcl_service_fini(service, rcl_node) != RCL_RET_OK) {
+            RCLCPP_ERROR(
+              logger,
+              "Error in destruction of rcl service handle [~/get_type_description]: %s",
+              rcl_get_error_string().str);
+            rcl_reset_error();
+          }
+          delete service;
+        });
+      *rcl_srv = rcl_get_zero_initialized_service();
+      rcl_ret_t rcl_ret = rcl_node_type_description_service_init(rcl_srv.get(), rcl_node);
+
+      if (rcl_ret != RCL_RET_OK) {
+        RCLCPP_ERROR(
+          logger_, "Failed to initialize ~/get_type_description service: %s",
+          rcl_get_error_string().str);
+        throw std::runtime_error(
+                "Failed to initialize ~/get_type_description service.");
+      }
+
+      rclcpp::AnyServiceCallback<ServiceT> cb;
+      cb.set(
+        [this](
+          std::shared_ptr<rmw_request_id_t> header,
+          std::shared_ptr<ServiceT::Request> request,
+          std::shared_ptr<ServiceT::Response> response
+        ) {
+          rcl_node_type_description_service_handle_request(
+            node_base_->get_rcl_node_handle(),
+            header.get(),
+            request.get(),
+            response.get());
+        });
+
+      type_description_srv_ = std::make_shared<Service<ServiceT>>(
+        node_base_->get_shared_rcl_node_handle(),
+        rcl_srv,
+        cb);
+      node_services->add_service(
+        std::dynamic_pointer_cast<ServiceBase>(type_description_srv_),
+        nullptr);
+    }
+  }
+};
+
+NodeTypeDescriptions::NodeTypeDescriptions(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
+  rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
+  rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
+  rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
+: impl_(new NodeTypeDescriptionsImpl(
+      node_base,
+      node_logging,
+      node_parameters,
+      node_services))
+{}
+
+NodeTypeDescriptions::~NodeTypeDescriptions()
+{}
+
+}  // namespace node_interfaces
+}  // namespace rclcpp

rclcpp/src/rclcpp/node_interfaces/node_waitables.cpp

@@ -32,8 +32,7 @@ NodeWaitables::add_waitable(
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacobperron): use custom exception
-      throw std::runtime_error("Cannot create waitable, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("waitable");
     }
   } else {
     group = node_base_->get_default_callback_group();

rclcpp/src/rclcpp/node_options.cpp

@@ -248,6 +248,19 @@ NodeOptions::start_parameter_services(bool start_parameter_services)
   return *this;
 }
 
+bool
+NodeOptions::enable_logger_service() const
+{
+  return this->enable_logger_service_;
+}
+
+NodeOptions &
+NodeOptions::enable_logger_service(bool enable_logger_service)
+{
+  this->enable_logger_service_ = enable_logger_service;
+  return *this;
+}
+
 bool
 NodeOptions::start_parameter_event_publisher() const
 {

rclcpp/src/rclcpp/parameter_value.cpp

@@ -129,8 +129,7 @@ ParameterValue::ParameterValue(const rcl_interfaces::msg::ParameterValue & value
     case PARAMETER_NOT_SET:
       break;
     default:
-      // TODO(wjwwood): use custom exception
-      throw std::runtime_error("Unknown type: " + std::to_string(value.type));
+      throw rclcpp::exceptions::UnknownTypeError(std::to_string(value.type));
   }
 }
 

rclcpp/src/rclcpp/publisher_base.cpp

@@ -270,6 +270,13 @@ PublisherBase::get_intra_process_subscription_count() const
   return ipm->get_subscription_count(intra_process_publisher_id_);
 }
 
+bool
+PublisherBase::is_durability_transient_local() const
+{
+  return rcl_publisher_get_actual_qos(publisher_handle_.get())->durability ==
+         RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL;
+}
+
 rclcpp::QoS
 PublisherBase::get_actual_qos() const
 {
@@ -384,3 +391,22 @@ std::vector<rclcpp::NetworkFlowEndpoint> PublisherBase::get_network_flow_endpoin
 
   return network_flow_endpoint_vector;
 }
+
+size_t PublisherBase::lowest_available_ipm_capacity() const
+{
+  if (!intra_process_is_enabled_) {
+    return 0u;
+  }
+
+  auto ipm = weak_ipm_.lock();
+
+  if (!ipm) {
+    // TODO(ivanpauno): should this raise an error?
+    RCLCPP_WARN(
+      rclcpp::get_logger("rclcpp"),
+      "Intra process manager died for a publisher.");
+    return 0u;
+  }
+
+  return ipm->lowest_available_capacity(intra_process_publisher_id_);
+}

rclcpp/src/rclcpp/rate.cpp

@@ -0,0 +1,106 @@
+// Copyright 2023 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.
+
+#include "rclcpp/rate.hpp"
+
+#include <chrono>
+#include <stdexcept>
+
+namespace rclcpp
+{
+
+Rate::Rate(
+  const double rate, Clock::SharedPtr clock)
+: clock_(clock), period_(0, 0), last_interval_(clock_->now())
+{
+  if (rate <= 0.0) {
+    throw std::invalid_argument{"rate must be greater than 0"};
+  }
+  period_ = Duration::from_seconds(1.0 / rate);
+}
+
+Rate::Rate(
+  const Duration & period, Clock::SharedPtr clock)
+: clock_(clock), period_(period), last_interval_(clock_->now())
+{
+  if (period <= Duration(0, 0)) {
+    throw std::invalid_argument{"period must be greater than 0"};
+  }
+}
+
+bool
+Rate::sleep()
+{
+  // Time coming into sleep
+  auto now = clock_->now();
+  // Time of next interval
+  auto next_interval = last_interval_ + period_;
+  // Detect backwards time flow
+  if (now < last_interval_) {
+    // Best thing to do is to set the next_interval to now + period
+    next_interval = now + period_;
+  }
+  // Update the interval
+  last_interval_ += period_;
+  // If the time_to_sleep is negative or zero, don't sleep
+  if (next_interval <= now) {
+    // If an entire cycle was missed then reset next interval.
+    // This might happen if the loop took more than a cycle.
+    // Or if time jumps forward.
+    if (now > next_interval + period_) {
+      last_interval_ = now + period_;
+    }
+    // Either way do not sleep and return false
+    return false;
+  }
+  // Calculate the time to sleep
+  auto time_to_sleep = next_interval - now;
+  // Sleep (will get interrupted by ctrl-c, may not sleep full time)
+  clock_->sleep_for(time_to_sleep);
+  return true;
+}
+
+bool
+Rate::is_steady() const
+{
+  return clock_->get_clock_type() == RCL_STEADY_TIME;
+}
+
+rcl_clock_type_t
+Rate::get_type() const
+{
+  return clock_->get_clock_type();
+}
+
+void
+Rate::reset()
+{
+  last_interval_ = clock_->now();
+}
+
+std::chrono::nanoseconds
+Rate::period() const
+{
+  return std::chrono::nanoseconds(period_.nanoseconds());
+}
+
+WallRate::WallRate(const double rate)
+: Rate(rate, std::make_shared<Clock>(RCL_STEADY_TIME))
+{}
+
+WallRate::WallRate(const Duration & period)
+: Rate(period, std::make_shared<Clock>(RCL_STEADY_TIME))
+{}
+
+}  // namespace rclcpp

rclcpp/src/rclcpp/signal_handler.cpp

@@ -113,7 +113,7 @@ SignalHandler::get_logger()
 SignalHandler &
 SignalHandler::get_global_signal_handler()
 {
-  static SignalHandler signal_handler;
+  static SignalHandler & signal_handler = *new SignalHandler();
   return signal_handler;
 }