21.2.0

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

rclcpp/CHANGELOG.rst

@@ -2,6 +2,34 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+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/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"

rclcpp/include/rclcpp/callback_group.hpp

@@ -180,6 +180,13 @@ public:
     return _find_ptrs_if_impl<rclcpp::Waitable, Function>(func, waitable_ptrs_);
   }
 
+  /// Get the total number of entities in this callback group.
+  /**
+   * \return the number of entities in the callback group.
+   */
+  RCLCPP_PUBLIC
+  size_t size() const;
+
   RCLCPP_PUBLIC
   std::atomic_bool &
   can_be_taken_from();

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/executor.hpp

@@ -51,6 +51,7 @@ typedef std::map<rclcpp::CallbackGroup::WeakPtr,
 
 // Forward declaration is used in convenience method signature.
 class Node;
+class ExecutorImplementation;
 
 /// Coordinate the order and timing of available communication tasks.
 /**
@@ -697,6 +698,9 @@ protected:
 
   /// shutdown callback handle registered to Context
   rclcpp::OnShutdownCallbackHandle shutdown_callback_handle_;
+
+  /// Pointer to implementation
+  std::unique_ptr<ExecutorImplementation> impl_;
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/executors/executor_entities_collection.hpp

@@ -198,14 +198,15 @@ build_entities_collection(
  *
  * \param[in] collection Collection of entities corresponding to the current wait set.
  * \param[in] wait_result Result of rclcpp::WaitSet::wait corresponding to the collection.
- * \return A queue of executables that have been marked ready by the waitset.
+ * \param[inout] queue of executables to append new ready executables to
+ * \return number of new ready executables
  */
-std::deque<rclcpp::AnyExecutable>
+size_t
 ready_executables(
   const ExecutorEntitiesCollection & collection,
-  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result
+  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
+  std::deque<rclcpp::AnyExecutable> & executables
 );
-
 }  // namespace executors
 }  // namespace rclcpp
 

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

@@ -33,6 +33,7 @@ public:
 
   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

@@ -44,6 +44,7 @@ public:
 
   virtual bool has_data() const = 0;
   virtual bool use_take_shared_method() const = 0;
+  virtual size_t available_capacity() const = 0;
 };
 
 template<
@@ -143,6 +144,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_;
 

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

@@ -148,6 +148,18 @@ public:
     return is_full_();
   }
 
+  /// 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
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return available_capacity_();
+  }
+
   void clear()
   {
     TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
@@ -189,6 +201,17 @@ 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_;
+  }
+
   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

@@ -306,6 +306,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
   {

rclcpp/include/rclcpp/experimental/subscription_intra_process_base.hpp

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

rclcpp/include/rclcpp/experimental/subscription_intra_process_buffer.hpp

@@ -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/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_options.hpp

@@ -50,6 +50,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 +233,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 +440,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/publisher_base.hpp

@@ -215,6 +215,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/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>

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/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>21.2.0</version>
   <description>The ROS client library in C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp/src/rclcpp/callback_group.cpp

@@ -56,6 +56,16 @@ CallbackGroup::type() const
   return type_;
 }
 
+size_t
+CallbackGroup::size() const
+{
+  return
+    subscription_ptrs_.size() +
+    service_ptrs_.size() +
+    client_ptrs_.size() +
+    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,

rclcpp/src/rclcpp/executor.cpp

@@ -41,11 +41,14 @@ using namespace std::chrono_literals;
 using rclcpp::exceptions::throw_from_rcl_error;
 using rclcpp::Executor;
 
+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)),
-  memory_strategy_(options.memory_strategy)
+  memory_strategy_(options.memory_strategy),
+  impl_(std::make_unique<rclcpp::ExecutorImplementation>())
 {
   // Store the context for later use.
   context_ = options.context;
@@ -600,7 +603,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:
       {

rclcpp/src/rclcpp/executors/executor_entities_collection.cpp

@@ -20,12 +20,13 @@ namespace executors
 {
 bool ExecutorEntitiesCollection::empty() const
 {
-  return subscriptions.empty() &&
-         timers.empty() &&
-         guard_conditions.empty() &&
-         clients.empty() &&
-         services.empty() &&
-         waitables.empty();
+  return
+    subscriptions.empty() &&
+    timers.empty() &&
+    guard_conditions.empty() &&
+    clients.empty() &&
+    services.empty() &&
+    waitables.empty();
 }
 
 void ExecutorEntitiesCollection::clear()
@@ -38,7 +39,6 @@ void ExecutorEntitiesCollection::clear()
   waitables.clear();
 }
 
-
 void
 build_entities_collection(
   const std::vector<rclcpp::CallbackGroup::WeakPtr> & callback_groups,
@@ -94,109 +94,136 @@ build_entities_collection(
   }
 }
 
-template<typename EntityCollectionType>
-void check_ready(
-  EntityCollectionType & collection,
-  std::deque<rclcpp::AnyExecutable> & executables,
-  size_t size_of_waited_entities,
-  typename EntityCollectionType::Key * waited_entities,
-  std::function<bool(rclcpp::AnyExecutable &,
-  typename EntityCollectionType::EntitySharedPtr &)> fill_executable)
+size_t
+ready_executables(
+  const ExecutorEntitiesCollection & collection,
+  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
+  std::deque<rclcpp::AnyExecutable> & executables
+)
 {
-  for (size_t ii = 0; ii < size_of_waited_entities; ++ii) {
-    if (!waited_entities[ii]) {continue;}
-    auto entity_iter = collection.find(waited_entities[ii]);
-    if (entity_iter != collection.end()) {
+  size_t added = 0;
+  if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
+    return added;
+  }
+  auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
+
+  // Cache shared pointers to groups to avoid extra work re-locking them
+  std::map<rclcpp::CallbackGroup::WeakPtr,
+    rclcpp::CallbackGroup::SharedPtr,
+    std::owner_less<rclcpp::CallbackGroup::WeakPtr>> group_map;
+
+  auto group_cache = [&group_map](const rclcpp::CallbackGroup::WeakPtr & weak_cbg_ptr)
+    {
+      if (group_map.count(weak_cbg_ptr) == 0) {
+        group_map.insert({weak_cbg_ptr, weak_cbg_ptr.lock()});
+      }
+      return group_map.find(weak_cbg_ptr)->second;
+    };
+
+  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;
       }
-
-      auto callback_group = entity_iter->second.callback_group.lock();
-      if (callback_group && !callback_group->can_be_taken_from().load()) {
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      if (!entity->call()) {
         continue;
       }
       rclcpp::AnyExecutable exec;
-
-      exec.callback_group = callback_group;
-      if (fill_executable(exec, entity)) {
-        executables.push_back(exec);
-      }
+      exec.timer = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
     }
   }
-}
 
-std::deque<rclcpp::AnyExecutable>
-ready_executables(
-  const ExecutorEntitiesCollection & collection,
-  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result
-)
-{
-  std::deque<rclcpp::AnyExecutable> ret;
-
-  if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
-    return ret;
-  }
-  auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
-  check_ready(
-    collection.timers,
-    ret,
-    rcl_wait_set.size_of_timers,
-    rcl_wait_set.timers,
-    [](rclcpp::AnyExecutable & exec, auto timer) {
-      if (!timer->call()) {
-        return false;
+  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;
       }
-      exec.timer = timer;
-      return true;
-    });
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.subscription = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }
 
-  check_ready(
-    collection.subscriptions,
-    ret,
-    rcl_wait_set.size_of_subscriptions,
-    rcl_wait_set.subscriptions,
-    [](rclcpp::AnyExecutable & exec, auto subscription) {
-      exec.subscription = subscription;
-      return 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;
+      }
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.service = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }
 
-
-  check_ready(
-    collection.services,
-    ret,
-    rcl_wait_set.size_of_services,
-    rcl_wait_set.services,
-    [](rclcpp::AnyExecutable & exec, auto service) {
-      exec.service = service;
-      return true;
-    });
-
-  check_ready(
-    collection.clients,
-    ret,
-    rcl_wait_set.size_of_clients,
-    rcl_wait_set.clients,
-    [](rclcpp::AnyExecutable & exec, auto client) {
-      exec.client = client;
-      return 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;
+      }
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.client = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }
 
   for (auto & [handle, entry] : collection.waitables) {
     auto waitable = entry.entity.lock();
-    if (waitable && waitable->is_ready(&rcl_wait_set)) {
-      auto group = entry.callback_group.lock();
-      if (group && !group->can_be_taken_from().load()) {
-        continue;
-      }
-
-      rclcpp::AnyExecutable exec;
-      exec.waitable = waitable;
-      exec.callback_group = group;
-      ret.push_back(exec);
+    if (!waitable) {
+      continue;
     }
+    if (!waitable->is_ready(&rcl_wait_set)) {
+      continue;
+    }
+    auto group_info = group_cache(entry.callback_group);
+    if (group_info && !group_info->can_be_taken_from().load()) {
+      continue;
+    }
+    rclcpp::AnyExecutable exec;
+    exec.waitable = waitable;
+    exec.callback_group = group_info;
+    exec.data = waitable->take_data();
+    executables.push_back(exec);
+    added++;
   }
-  return ret;
+
+  return added;
 }
 
 }  // namespace executors

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()
@@ -269,10 +273,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 +290,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 +306,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 +329,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 +400,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 +414,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 +427,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 +501,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/intra_process_manager.cpp

@@ -225,5 +225,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/logging_mutex.cpp

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

rclcpp/src/rclcpp/node.cpp

@@ -167,7 +167,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())),
@@ -225,6 +225,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(

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_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/publisher_base.cpp

@@ -384,3 +384,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/subscription_base.cpp

@@ -52,7 +52,7 @@ SubscriptionBase::SubscriptionBase(
   intra_process_subscription_id_(0),
   event_callbacks_(event_callbacks),
   type_support_(type_support_handle),
-  delivered_message_type_(delivered_message_kind)
+  delivered_message_kind_(delivered_message_kind)
 {
   auto custom_deletor = [node_handle = this->node_handle_](rcl_subscription_t * rcl_subs)
     {
@@ -261,13 +261,13 @@ SubscriptionBase::get_message_type_support_handle() const
 bool
 SubscriptionBase::is_serialized() const
 {
-  return delivered_message_type_ == rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE;
+  return delivered_message_kind_ == rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE;
 }
 
 rclcpp::DeliveredMessageKind
-SubscriptionBase::get_subscription_type() const
+SubscriptionBase::get_delivered_message_kind() const
 {
-  return delivered_message_type_;
+  return delivered_message_kind_;
 }
 
 size_t

rclcpp/test/rclcpp/CMakeLists.txt

@@ -649,6 +649,13 @@ if(TARGET test_wait_for_message)
   target_link_libraries(test_wait_for_message ${PROJECT_NAME})
 endif()
 
+ament_add_gtest(test_logger_service test_logger_service.cpp)
+if(TARGET test_logger_service)
+  ament_target_dependencies(test_logger_service
+    "rcl_interfaces")
+  target_link_libraries(test_logger_service ${PROJECT_NAME})
+endif()
+
 ament_add_gtest(test_interface_traits test_interface_traits.cpp
   APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_interface_traits)

rclcpp/test/rclcpp/executors/test_executors.cpp

@@ -20,12 +20,14 @@
 #include <gtest/gtest.h>
 
 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <limits>
 #include <memory>
 #include <string>
 #include <thread>
 #include <utility>
+#include <vector>
 
 #include "rcl/error_handling.h"
 #include "rcl/time.h"
@@ -43,18 +45,10 @@ template<typename T>
 class TestExecutors : public ::testing::Test
 {
 public:
-  static void SetUpTestCase()
-  {
-    rclcpp::init(0, nullptr);
-  }
-
-  static void TearDownTestCase()
-  {
-    rclcpp::shutdown();
-  }
-
   void SetUp()
   {
+    rclcpp::init(0, nullptr);
+
     const auto test_info = ::testing::UnitTest::GetInstance()->current_test_info();
     std::stringstream test_name;
     test_name << test_info->test_case_name() << "_" << test_info->name();
@@ -75,6 +69,8 @@ public:
     publisher.reset();
     subscription.reset();
     node.reset();
+
+    rclcpp::shutdown();
   }
 
   rclcpp::Node::SharedPtr node;
@@ -729,6 +725,131 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteInterrupted)
   spinner.join();
 }
 
+// This test verifies that the add_node operation is robust wrt race conditions.
+// It's mostly meant to prevent regressions in the events-executor, but the operation should be
+// thread-safe in all executor implementations.
+// The initial implementation of the events-executor contained a bug where the executor
+// would end up in an inconsistent state and stop processing interrupt/shutdown notifications.
+// Manually adding a node to the executor results in a) producing a notify waitable event
+// and b) refreshing the executor collections.
+// The inconsistent state would happen if the event was processed before the collections were
+// finished to be refreshed: the executor would pick up the event but be unable to process it.
+// This would leave the `notify_waitable_event_pushed_` flag to true, preventing additional
+// notify waitable events to be pushed.
+// The behavior is observable only under heavy load, so this test spawns several worker
+// threads. Due to the nature of the bug, this test may still succeed even if the
+// bug is present. However repeated runs will show its flakiness nature and indicate
+// an eventual regression.
+TYPED_TEST(TestExecutors, testRaceConditionAddNode)
+{
+  using ExecutorType = TypeParam;
+  // rmw_connextdds doesn't support events-executor
+  if (
+    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
+    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
+  {
+    GTEST_SKIP();
+  }
+
+  // Spawn some threads to do some heavy work
+  std::atomic<bool> should_cancel = false;
+  std::vector<std::thread> stress_threads;
+  for (size_t i = 0; i < 5 * std::thread::hardware_concurrency(); i++) {
+    stress_threads.emplace_back(
+      [&should_cancel, i]() {
+        // This is just some arbitrary heavy work
+        volatile size_t total = 0;
+        for (size_t k = 0; k < 549528914167; k++) {
+          if (should_cancel) {
+            break;
+          }
+          total += k * (i + 42);
+        }
+      });
+  }
+
+  // Create an executor
+  auto executor = std::make_shared<ExecutorType>();
+  // Start spinning
+  auto executor_thread = std::thread(
+    [executor]() {
+      executor->spin();
+    });
+  // Add a node to the executor
+  executor->add_node(this->node);
+
+  // Cancel the executor (make sure that it's already spinning first)
+  while (!executor->is_spinning() && rclcpp::ok()) {
+    continue;
+  }
+  executor->cancel();
+
+  // Try to join the thread after cancelling the executor
+  // This is the "test". We want to make sure that we can still cancel the executor
+  // regardless of the presence of race conditions
+  executor_thread.join();
+
+  // The test is now completed: we can join the stress threads
+  should_cancel = true;
+  for (auto & t : stress_threads) {
+    t.join();
+  }
+}
+
+// This test verifies the thread-safety of adding and removing a node
+// while the executor is spinning and events are ready.
+// This test does not contain expectations, but rather it verifies that
+// we can run a "stressful routine" without crashing.
+TYPED_TEST(TestExecutors, stressAddRemoveNode)
+{
+  using ExecutorType = TypeParam;
+  // rmw_connextdds doesn't support events-executor
+  if (
+    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
+    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
+  {
+    GTEST_SKIP();
+  }
+
+  ExecutorType executor;
+
+  // A timer that is "always" ready (the timer callback doesn't do anything)
+  auto timer = this->node->create_wall_timer(std::chrono::nanoseconds(1), []() {});
+
+  // This thread spins the executor until it's cancelled
+  std::thread spinner_thread([&]() {
+      executor.spin();
+    });
+
+  // This thread publishes data in a busy loop (the node has a subscription)
+  std::thread publisher_thread1([&]() {
+      for (size_t i = 0; i < 100000; i++) {
+        this->publisher->publish(test_msgs::msg::Empty());
+      }
+    });
+  std::thread publisher_thread2([&]() {
+      for (size_t i = 0; i < 100000; i++) {
+        this->publisher->publish(test_msgs::msg::Empty());
+      }
+    });
+
+  // This thread adds/remove the node that contains the entities in a busy loop
+  std::thread add_remove_thread([&]() {
+      for (size_t i = 0; i < 100000; i++) {
+        executor.add_node(this->node);
+        executor.remove_node(this->node);
+      }
+    });
+
+  // Wait for the threads that do real work to finish
+  publisher_thread1.join();
+  publisher_thread2.join();
+  add_remove_thread.join();
+
+  executor.cancel();
+  spinner_thread.join();
+}
+
 // Check spin_until_future_complete with node base pointer (instantiates its own executor)
 TEST(TestExecutors, testSpinUntilFutureCompleteNodeBasePtr)
 {

rclcpp/test/rclcpp/test_intra_process_buffer.cpp

@@ -238,3 +238,75 @@ TEST(TestIntraProcessBuffer, unique_buffer_consume) {
   EXPECT_EQ(original_value, *popped_unique_msg);
   EXPECT_EQ(original_message_pointer, popped_message_pointer);
 }
+
+/*
+  Check the available buffer capacity while storing and consuming data from an intra-process
+  buffer.
+  The initial available buffer capacity should equal the buffer size.
+  Inserting a message should decrease the available buffer capacity by 1.
+  Consuming a message should increase the available buffer capacity by 1.
+ */
+TEST(TestIntraProcessBuffer, available_capacity) {
+  using MessageT = char;
+  using Alloc = std::allocator<void>;
+  using Deleter = std::default_delete<MessageT>;
+  using SharedMessageT = std::shared_ptr<const MessageT>;
+  using UniqueMessageT = std::unique_ptr<MessageT, Deleter>;
+  using UniqueIntraProcessBufferT = rclcpp::experimental::buffers::TypedIntraProcessBuffer<
+    MessageT, Alloc, Deleter, UniqueMessageT>;
+
+  constexpr auto history_depth = 5u;
+
+  auto buffer_impl =
+    std::make_unique<rclcpp::experimental::buffers::RingBufferImplementation<UniqueMessageT>>(
+    history_depth);
+
+  UniqueIntraProcessBufferT intra_process_buffer(std::move(buffer_impl));
+
+  EXPECT_EQ(history_depth, intra_process_buffer.available_capacity());
+
+  auto original_unique_msg = std::make_unique<char>('a');
+  auto original_message_pointer = reinterpret_cast<std::uintptr_t>(original_unique_msg.get());
+  auto original_value = *original_unique_msg;
+
+  intra_process_buffer.add_unique(std::move(original_unique_msg));
+
+  EXPECT_EQ(history_depth - 1u, intra_process_buffer.available_capacity());
+
+  SharedMessageT popped_shared_msg;
+  popped_shared_msg = intra_process_buffer.consume_shared();
+  auto popped_message_pointer = reinterpret_cast<std::uintptr_t>(popped_shared_msg.get());
+
+  EXPECT_EQ(history_depth, intra_process_buffer.available_capacity());
+  EXPECT_EQ(original_value, *popped_shared_msg);
+  EXPECT_EQ(original_message_pointer, popped_message_pointer);
+
+  original_unique_msg = std::make_unique<char>('b');
+  original_message_pointer = reinterpret_cast<std::uintptr_t>(original_unique_msg.get());
+  original_value = *original_unique_msg;
+
+  intra_process_buffer.add_unique(std::move(original_unique_msg));
+
+  auto second_unique_msg = std::make_unique<char>('c');
+  auto second_message_pointer = reinterpret_cast<std::uintptr_t>(second_unique_msg.get());
+  auto second_value = *second_unique_msg;
+
+  intra_process_buffer.add_unique(std::move(second_unique_msg));
+
+  EXPECT_EQ(history_depth - 2u, intra_process_buffer.available_capacity());
+
+  UniqueMessageT popped_unique_msg;
+  popped_unique_msg = intra_process_buffer.consume_unique();
+  popped_message_pointer = reinterpret_cast<std::uintptr_t>(popped_unique_msg.get());
+
+  EXPECT_EQ(history_depth - 1u, intra_process_buffer.available_capacity());
+  EXPECT_EQ(original_value, *popped_unique_msg);
+  EXPECT_EQ(original_message_pointer, popped_message_pointer);
+
+  popped_unique_msg = intra_process_buffer.consume_unique();
+  popped_message_pointer = reinterpret_cast<std::uintptr_t>(popped_unique_msg.get());
+
+  EXPECT_EQ(history_depth, intra_process_buffer.available_capacity());
+  EXPECT_EQ(second_value, *popped_unique_msg);
+  EXPECT_EQ(second_message_pointer, popped_message_pointer);
+}

rclcpp/test/rclcpp/test_intra_process_manager.cpp

@@ -156,18 +156,26 @@ public:
   {
     message_ptr = reinterpret_cast<std::uintptr_t>(msg.get());
     shared_msg = msg;
+    ++num_msgs;
   }
 
   void add(MessageUniquePtr msg)
   {
     message_ptr = reinterpret_cast<std::uintptr_t>(msg.get());
     unique_msg = std::move(msg);
+    ++num_msgs;
   }
 
   void pop(std::uintptr_t & msg_ptr)
   {
     msg_ptr = message_ptr;
     message_ptr = 0;
+    --num_msgs;
+  }
+
+  size_t size() const
+  {
+    return num_msgs;
   }
 
   // need to store the messages somewhere otherwise the memory address will be reused
@@ -175,6 +183,8 @@ public:
   MessageUniquePtr unique_msg;
 
   std::uintptr_t message_ptr;
+  // count add and pop
+  size_t num_msgs = 0u;
 };
 
 }  // namespace mock
@@ -221,6 +231,10 @@ public:
     return topic_name.c_str();
   }
 
+  virtual
+  size_t
+  available_capacity() const = 0;
+
   rclcpp::QoS qos_profile;
   std::string topic_name;
 };
@@ -280,6 +294,12 @@ public:
     return take_shared_method;
   }
 
+  size_t
+  available_capacity() const override
+  {
+    return qos_profile.depth() - buffer->size();
+  }
+
   bool take_shared_method;
 
   typename rclcpp::experimental::buffers::mock::IntraProcessBuffer<MessageT>::UniquePtr buffer;
@@ -712,3 +732,91 @@ TEST(TestIntraProcessManager, multiple_subscriptions_different_type) {
   EXPECT_EQ(original_message_pointer, received_message_pointer_10);
   EXPECT_NE(original_message_pointer, received_message_pointer_11);
 }
+
+/*
+   This tests the method "lowest_available_capacity":
+   - Creates 1 publisher.
+   - The available buffer capacity should be at least history size.
+   - Add 2 subscribers.
+   - Add everything to the intra-process manager.
+   - All the entities are expected to have different ids.
+   - Check the subscriptions count for the publisher.
+   - The available buffer capacity should be the history size.
+   - Publish one message (without receiving it).
+   - The available buffer capacity should decrease by 1.
+   - Publish another message (without receiving it).
+   - The available buffer capacity should decrease by 1.
+   - One subscriber receives one message.
+   - The available buffer capacity should stay the same,
+     as the other subscriber still has not freed its buffer.
+   - The other subscriber receives one message.
+   - The available buffer capacity should increase by 1.
+   - One subscription goes out of scope.
+   - The available buffer capacity should not change.
+ */
+TEST(TestIntraProcessManager, lowest_available_capacity) {
+  using IntraProcessManagerT = rclcpp::experimental::IntraProcessManager;
+  using MessageT = rcl_interfaces::msg::Log;
+  using PublisherT = rclcpp::mock::Publisher<MessageT>;
+  using SubscriptionIntraProcessT = rclcpp::experimental::mock::SubscriptionIntraProcess<MessageT>;
+
+  constexpr auto history_depth = 10u;
+
+  auto ipm = std::make_shared<IntraProcessManagerT>();
+
+  auto p1 = std::make_shared<PublisherT>(rclcpp::QoS(history_depth).best_effort());
+
+  auto s1 = std::make_shared<SubscriptionIntraProcessT>(rclcpp::QoS(history_depth).best_effort());
+  auto s2 = std::make_shared<SubscriptionIntraProcessT>(rclcpp::QoS(history_depth).best_effort());
+
+  auto p1_id = ipm->add_publisher(p1);
+  p1->set_intra_process_manager(p1_id, ipm);
+
+  auto c1 = ipm->lowest_available_capacity(p1_id);
+
+  ASSERT_LE(0u, c1);
+
+  auto s1_id = ipm->add_subscription(s1);
+  auto s2_id = ipm->add_subscription(s2);
+
+  bool unique_ids = s1_id != s2_id && p1_id != s1_id;
+  ASSERT_TRUE(unique_ids);
+
+  size_t p1_subs = ipm->get_subscription_count(p1_id);
+  size_t non_existing_pub_subs = ipm->get_subscription_count(42);
+  ASSERT_EQ(2u, p1_subs);
+  ASSERT_EQ(0u, non_existing_pub_subs);
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  auto non_existing_pub_c = ipm->lowest_available_capacity(42);
+
+  ASSERT_EQ(history_depth, c1);
+  ASSERT_EQ(0u, non_existing_pub_c);
+
+  auto unique_msg = std::make_unique<MessageT>();
+  p1->publish(std::move(unique_msg));
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  ASSERT_EQ(history_depth - 1u, c1);
+
+  unique_msg = std::make_unique<MessageT>();
+  p1->publish(std::move(unique_msg));
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  ASSERT_EQ(history_depth - 2u, c1);
+
+  s1->pop();
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  ASSERT_EQ(history_depth - 2u, c1);
+
+  s2->pop();
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  ASSERT_EQ(history_depth - 1u, c1);
+
+  ipm->get_subscription_intra_process(s1_id).reset();
+
+  c1 = ipm->lowest_available_capacity(p1_id);
+  ASSERT_EQ(history_depth - 1u, c1);
+}

rclcpp/test/rclcpp/test_logger_service.cpp

@@ -0,0 +1,214 @@
+// Copyright 2023 Sony Group Corporation.
+//
+// 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 <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "rclcpp/rclcpp.hpp"
+#include "rclcpp/node.hpp"
+#include "rcl_interfaces/srv/get_logger_levels.hpp"
+#include "rcl_interfaces/srv/set_logger_levels.hpp"
+
+using namespace std::chrono_literals;
+
+class TestLoggerService : public ::testing::Test
+{
+protected:
+  void SetUp()
+  {
+    rclcpp::init(0, nullptr);
+    rclcpp::NodeOptions options = rclcpp::NodeOptions();
+    options.enable_logger_service(true);
+    node_ = std::make_shared<rclcpp::Node>("test_logger_service", "/test", options);
+  }
+
+  void TearDown()
+  {
+    rclcpp::shutdown();
+  }
+
+  rclcpp::Node::SharedPtr node_;
+  std::thread thread_;
+};
+
+TEST_F(TestLoggerService, check_connect_get_logger_service) {
+  auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
+    "/test/test_logger_service/get_logger_levels");
+  ASSERT_TRUE(client->wait_for_service(2s));
+}
+
+TEST_F(TestLoggerService, check_connect_set_logger_service) {
+  auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
+    "/test/test_logger_service/set_logger_levels");
+  ASSERT_TRUE(client->wait_for_service(2s));
+}
+
+TEST_F(TestLoggerService, test_set_and_get_one_logging_level) {
+  std::string test_logger_name = "rcl";
+  uint8_t test_logger_level = 20;
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
+      "/test/test_logger_service/set_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
+    auto level = rcl_interfaces::msg::LoggerLevel();
+    level.name = test_logger_name;
+    level.level = test_logger_level;
+    request->levels.push_back(level);
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->results.size(), 1u);
+    ASSERT_TRUE(result_get->results[0].successful);
+    ASSERT_STREQ(result_get->results[0].reason.c_str(), "");
+  }
+
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
+      "/test/test_logger_service/get_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::GetLoggerLevels::Request>();
+    request->names.emplace_back(test_logger_name);
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->levels.size(), 1u);
+    ASSERT_STREQ(result_get->levels[0].name.c_str(), test_logger_name.c_str());
+    ASSERT_EQ(result_get->levels[0].level, test_logger_level);
+  }
+}
+
+TEST_F(TestLoggerService, test_set_and_get_multi_logging_level) {
+  std::vector<std::pair<std::string, uint8_t>> test_data {
+    {"rcl", 30},
+    {"rclcpp", 40},
+    {"/test/test_logger_service", 50}
+  };
+
+  // Set multi log levels
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
+      "/test/test_logger_service/set_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
+    for (auto & set_level : test_data) {
+      auto level = rcl_interfaces::msg::LoggerLevel();
+      level.name = std::get<0>(set_level);
+      level.level = std::get<1>(set_level);
+      request->levels.push_back(level);
+    }
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->results.size(), test_data.size());
+    for (uint32_t i = 0; i < test_data.size(); i++) {
+      ASSERT_TRUE(result_get->results[0].successful);
+    }
+  }
+
+  // Get multi log levels
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::GetLoggerLevels>(
+      "/test/test_logger_service/get_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::GetLoggerLevels::Request>();
+    for (auto & set_level : test_data) {
+      request->names.emplace_back(std::get<0>(set_level));
+    }
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->levels.size(), test_data.size());
+    for (uint32_t i = 0; i < test_data.size(); i++) {
+      ASSERT_STREQ(result_get->levels[i].name.c_str(), std::get<0>(test_data[i]).c_str());
+      ASSERT_EQ(result_get->levels[i].level, std::get<1>(test_data[i]));
+    }
+  }
+}
+
+TEST_F(TestLoggerService, test_set_logging_level_with_invalid_param) {
+  std::vector<std::pair<std::string, uint8_t>> test_data {
+    {"rcl", 12},
+    {"/test/test_logger_service", 22}
+  };
+
+  // Set multi log levels
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
+      "/test/test_logger_service/set_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
+    for (auto & set_level : test_data) {
+      auto level = rcl_interfaces::msg::LoggerLevel();
+      level.name = std::get<0>(set_level);
+      level.level = std::get<1>(set_level);
+      request->levels.push_back(level);
+    }
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->results.size(), test_data.size());
+    for (uint32_t i = 0; i < test_data.size(); i++) {
+      ASSERT_FALSE(result_get->results[i].successful);
+      // Check string starts with prefix
+      ASSERT_EQ(
+        result_get->results[i].reason.rfind("Unable to determine severity_string for severity", 0),
+        0);
+    }
+  }
+}
+
+TEST_F(TestLoggerService, test_set_logging_level_with_partial_invalid_param) {
+  std::vector<std::pair<std::string, uint8_t>> test_data {
+    {"rcl", 20},
+    {"rclcpp", 22},
+    {"/test/test_logger_service", 30}
+  };
+
+  // Set multi log levels
+  {
+    auto client = node_->create_client<rcl_interfaces::srv::SetLoggerLevels>(
+      "/test/test_logger_service/set_logger_levels");
+    ASSERT_TRUE(client->wait_for_service(1s));
+    auto request = std::make_shared<rcl_interfaces::srv::SetLoggerLevels::Request>();
+    for (auto & set_level : test_data) {
+      auto level = rcl_interfaces::msg::LoggerLevel();
+      level.name = std::get<0>(set_level);
+      level.level = std::get<1>(set_level);
+      request->levels.push_back(level);
+    }
+    auto result = client->async_send_request(request);
+    ASSERT_EQ(
+      rclcpp::spin_until_future_complete(node_, result),
+      rclcpp::FutureReturnCode::SUCCESS);
+    auto result_get = result.get();
+    ASSERT_EQ(result_get->results.size(), test_data.size());
+    ASSERT_TRUE(result_get->results[0].successful);
+    ASSERT_FALSE(result_get->results[1].successful);
+    ASSERT_TRUE(result_get->results[2].successful);
+  }
+}

rclcpp/test/rclcpp/test_node_options.cpp

@@ -266,6 +266,11 @@ TEST(TestNodeOptions, bool_setters_and_getters) {
   EXPECT_FALSE(options.automatically_declare_parameters_from_overrides());
   options.automatically_declare_parameters_from_overrides(true);
   EXPECT_TRUE(options.automatically_declare_parameters_from_overrides());
+
+  options.enable_logger_service(false);
+  EXPECT_FALSE(options.enable_logger_service());
+  options.enable_logger_service(true);
+  EXPECT_TRUE(options.enable_logger_service());
 }
 
 TEST(TestNodeOptions, parameter_event_qos) {

rclcpp/test/rclcpp/test_publisher.cpp

@@ -629,6 +629,41 @@ TEST_P(TestPublisherWaitForAllAcked, check_wait_for_all_acked_with_QosPolicy) {
   EXPECT_TRUE(pub->wait_for_all_acked(std::chrono::milliseconds(6000)));
 }
 
+TEST_F(TestPublisher, lowest_available_ipm_capacity) {
+  constexpr auto history_depth = 10u;
+
+  initialize(rclcpp::NodeOptions().use_intra_process_comms(true));
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> options_ipm_disabled;
+  options_ipm_disabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Disable;
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> options_ipm_enabled;
+  options_ipm_enabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
+
+  auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Strings>) {};
+  auto pub_ipm_disabled = node->create_publisher<test_msgs::msg::Strings>(
+    "topic", history_depth,
+    options_ipm_disabled);
+  auto pub_ipm_enabled = node->create_publisher<test_msgs::msg::Strings>(
+    "topic", history_depth,
+    options_ipm_enabled);
+  auto sub = node->create_subscription<test_msgs::msg::Strings>(
+    "topic",
+    history_depth,
+    do_nothing);
+
+  ASSERT_EQ(1, pub_ipm_enabled->get_intra_process_subscription_count());
+  ASSERT_EQ(0, pub_ipm_disabled->lowest_available_ipm_capacity());
+  ASSERT_EQ(history_depth, pub_ipm_enabled->lowest_available_ipm_capacity());
+
+  auto msg = std::make_shared<test_msgs::msg::Strings>();
+  ASSERT_NO_THROW(pub_ipm_disabled->publish(*msg));
+  ASSERT_NO_THROW(pub_ipm_enabled->publish(*msg));
+
+  ASSERT_EQ(0, pub_ipm_disabled->lowest_available_ipm_capacity());
+  ASSERT_EQ(history_depth - 1u, pub_ipm_enabled->lowest_available_ipm_capacity());
+}
+
 INSTANTIATE_TEST_SUITE_P(
   TestWaitForAllAckedWithParm,
   TestPublisherWaitForAllAcked,

rclcpp/test/rclcpp/test_rosout_subscription.cpp

@@ -70,7 +70,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_getchild) {
     // before calling get_child of Logger
     {
       RCLCPP_INFO(
-        rclcpp::get_logger(logger_name), this->rosout_msg_data.c_str());
+        rclcpp::get_logger(logger_name), "%s", this->rosout_msg_data.c_str());
       auto future = received_msg_promise.get_future();
       auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
       ASSERT_EQ(rclcpp::FutureReturnCode::TIMEOUT, return_code);
@@ -83,7 +83,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_getchild) {
     // after calling get_child of Logger
     // 1. use child_logger directly
     {
-      RCLCPP_INFO(child_logger, this->rosout_msg_data.c_str());
+      RCLCPP_INFO(child_logger, "%s", this->rosout_msg_data.c_str());
       auto future = received_msg_promise.get_future();
       auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
       ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);
@@ -93,7 +93,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_getchild) {
 
     // 2. use rclcpp::get_logger
     {
-      RCLCPP_INFO(rclcpp::get_logger(logger_name), this->rosout_msg_data.c_str());
+      RCLCPP_INFO(rclcpp::get_logger(logger_name), "%s", this->rosout_msg_data.c_str());
       auto future = received_msg_promise.get_future();
       auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
       ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);
@@ -104,7 +104,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_getchild) {
 
   // `child_logger` is end of life, there is no sublogger
   {
-    RCLCPP_INFO(rclcpp::get_logger(logger_name), this->rosout_msg_data.c_str());
+    RCLCPP_INFO(rclcpp::get_logger(logger_name), "%s", this->rosout_msg_data.c_str());
     auto future = received_msg_promise.get_future();
     auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
     ASSERT_EQ(rclcpp::FutureReturnCode::TIMEOUT, return_code);
@@ -119,7 +119,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_parent_log) {
 
   rclcpp::Logger logger = this->node->get_logger();
   ASSERT_EQ(logger.get_name(), logger_name);
-  RCLCPP_INFO(logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(logger, "%s", this->rosout_msg_data.c_str());
   auto future = received_msg_promise.get_future();
   auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);
@@ -133,14 +133,14 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_child_log) {
   this->rosout_msg_name = logger_name;
 
   rclcpp::Logger logger = this->node->get_logger();
-  RCLCPP_INFO(logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(logger, "%s", this->rosout_msg_data.c_str());
   auto future = received_msg_promise.get_future();
   auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::TIMEOUT, return_code);
   received_msg_promise = {};
 
   logger = this->node->get_logger().get_child("child1");
-  RCLCPP_INFO(logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(logger, "%s", this->rosout_msg_data.c_str());
   future = received_msg_promise.get_future();
   return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);
@@ -148,14 +148,14 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_child_log) {
   received_msg_promise = {};
 
   logger = this->node->get_logger().get_child("child2");
-  RCLCPP_INFO(logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(logger, "%s", this->rosout_msg_data.c_str());
   future = received_msg_promise.get_future();
   return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::TIMEOUT, return_code);
   received_msg_promise = {};
 
   this->rosout_msg_name = "ns.test_rosout_subscription.child2";
-  RCLCPP_INFO(logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(logger, "%s", this->rosout_msg_data.c_str());
   future = received_msg_promise.get_future();
   return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);
@@ -171,7 +171,7 @@ TEST_F(TestRosoutSubscription, test_rosoutsubscription_getchild_hierarchy) {
   rclcpp::Logger grandchild_logger =
     this->node->get_logger().get_child("child").get_child("grandchild");
   ASSERT_EQ(grandchild_logger.get_name(), logger_name);
-  RCLCPP_INFO(grandchild_logger, this->rosout_msg_data.c_str());
+  RCLCPP_INFO(grandchild_logger, "%s", this->rosout_msg_data.c_str());
   auto future = received_msg_promise.get_future();
   auto return_code = rclcpp::spin_until_future_complete(this->node, future, 3s);
   ASSERT_EQ(rclcpp::FutureReturnCode::SUCCESS, return_code);

rclcpp_action/CHANGELOG.rst

@@ -3,6 +3,18 @@ Changelog for package rclcpp_action
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 
+21.2.0 (2023-06-07)
+-------------------
+
+21.1.1 (2023-05-11)
+-------------------
+
+21.1.0 (2023-04-27)
+-------------------
+
+21.0.0 (2023-04-18)
+-------------------
+
 20.0.0 (2023-04-13)
 -------------------
 * extract the result response before the callback is issued. (`#2132 <https://github.com/ros2/rclcpp/issues/2132>`_)

rclcpp_action/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_action</name>
-  <version>20.0.0</version>
+  <version>21.2.0</version>
   <description>Adds action APIs for C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_components/CHANGELOG.rst

@@ -2,6 +2,18 @@
 Changelog for package rclcpp_components
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+21.2.0 (2023-06-07)
+-------------------
+
+21.1.1 (2023-05-11)
+-------------------
+
+21.1.0 (2023-04-27)
+-------------------
+
+21.0.0 (2023-04-18)
+-------------------
+
 20.0.0 (2023-04-13)
 -------------------
 * Update all rclcpp packages to C++17. (`#2121 <https://github.com/ros2/rclcpp/issues/2121>`_)

rclcpp_components/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_components</name>
-  <version>20.0.0</version>
+  <version>21.2.0</version>
   <description>Package containing tools for dynamically loadable components</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_lifecycle/CHANGELOG.rst

@@ -3,6 +3,21 @@ Changelog for package rclcpp_lifecycle
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 
+21.2.0 (2023-06-07)
+-------------------
+
+21.1.1 (2023-05-11)
+-------------------
+
+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>`_)
+* Support publishing loaned messages in LifecyclePublisher (`#2159 <https://github.com/ros2/rclcpp/issues/2159>`_)
+* Contributors: Lei Liu, Michael Babenko
+
 20.0.0 (2023-04-13)
 -------------------
 * Fixes to silence some clang warnings. (`#2127 <https://github.com/ros2/rclcpp/issues/2127>`_)

rclcpp_lifecycle/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_lifecycle</name>
-  <version>20.0.0</version>
+  <version>21.2.0</version>
   <description>Package containing a prototype for lifecycle implementation</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_lifecycle/src/lifecycle_node.cpp

@@ -76,7 +76,7 @@ LifecycleNode::LifecycleNode(
       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())),