Graph executables comparison

rclcpp/CHANGELOG.rst

@@ -2,6 +2,24 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+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>`_)

rclcpp/CMakeLists.txt

@@ -46,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
@@ -71,6 +72,8 @@ set(${PROJECT_NAME}_SRCS
   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
+  src/rclcpp/experimental/executors/graph_executor.cpp
+  src/rclcpp/experimental/graph_executable.cpp
   src/rclcpp/experimental/timers_manager.cpp
   src/rclcpp/future_return_code.cpp
   src/rclcpp/generic_publisher.cpp
@@ -281,7 +284,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/include/rclcpp/any_subscription_callback.hpp

@@ -30,6 +30,7 @@
 #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"
 
@@ -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,7 +509,9 @@ class AnySubscriptionCallback
   }
 
   // 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)
@@ -589,7 +613,7 @@ class AnySubscriptionCallback
   // 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)
   {
     TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);

rclcpp/include/rclcpp/callback_group.hpp

@@ -129,16 +129,15 @@ class CallbackGroup
    * 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.
    */
   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.
@@ -228,16 +227,6 @@ class CallbackGroup
   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.
@@ -297,7 +286,7 @@ class CallbackGroup
   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/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

@@ -64,7 +64,8 @@ create_subscription(
   ),
   typename MessageMemoryStrategyT::SharedPtr msg_mem_strat = (
     MessageMemoryStrategyT::create_default()
-  )
+  ),
+  std::initializer_list<rclcpp::PublisherBase::SharedPtr> publishers = {}
 )
 {
   using rclcpp::node_interfaces::get_node_topics_interface;
@@ -88,7 +89,7 @@ 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>(
@@ -133,6 +134,13 @@ create_subscription(
     qos;
 
   auto sub = node_topics_interface->create_subscription(topic_name, factory, actual_qos);
+
+  // Get any subscriptions downstream from the publisher and add their graph_node_t to this one
+  for (auto & publisher : publishers) {
+    auto topic_name = publisher->get_topic_name();
+    sub->add_output_topic(topic_name);
+  }
+
   node_topics_interface->add_subscription(sub, options.callback_group);
 
   return std::dynamic_pointer_cast<SubscriptionT>(sub);
@@ -183,12 +191,14 @@ create_subscription(
   ),
   typename MessageMemoryStrategyT::SharedPtr msg_mem_strat = (
     MessageMemoryStrategyT::create_default()
-  )
+  ),
+  std::initializer_list<rclcpp::PublisherBase::SharedPtr> publishers = {}
 )
 {
   return rclcpp::detail::create_subscription<
     MessageT, CallbackT, AllocatorT, SubscriptionT, MessageMemoryStrategyT>(
-    node, node, topic_name, qos, std::forward<CallbackT>(callback), options, msg_mem_strat);
+    node, node, topic_name, qos, std::forward<CallbackT>(callback),
+    options, msg_mem_strat, publishers);
 }
 
 /// Create and return a subscription of the given MessageT type.
@@ -213,13 +223,14 @@ create_subscription(
   ),
   typename MessageMemoryStrategyT::SharedPtr msg_mem_strat = (
     MessageMemoryStrategyT::create_default()
-  )
+  ),
+  std::initializer_list<rclcpp::PublisherBase::SharedPtr> publishers = {}
 )
 {
   return rclcpp::detail::create_subscription<
     MessageT, CallbackT, AllocatorT, SubscriptionT, MessageMemoryStrategyT>(
     node_parameters, node_topics, topic_name, qos,
-    std::forward<CallbackT>(callback), options, msg_mem_strat);
+    std::forward<CallbackT>(callback), options, msg_mem_strat, publishers);
 }
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/create_timer.hpp

@@ -121,6 +121,7 @@ create_timer(
     group,
     rclcpp::node_interfaces::get_node_base_interface(node).get(),
     rclcpp::node_interfaces::get_node_timers_interface(node).get(),
+    {},
     autostart);
 }
 
@@ -150,6 +151,7 @@ create_timer(
   rclcpp::CallbackGroup::SharedPtr group,
   node_interfaces::NodeBaseInterface * node_base,
   node_interfaces::NodeTimersInterface * node_timers,
+  std::initializer_list<rclcpp::PublisherBase::SharedPtr> publishers = {},
   bool autostart = true)
 {
   if (clock == nullptr) {
@@ -168,6 +170,13 @@ create_timer(
   auto timer = rclcpp::GenericTimer<CallbackT>::make_shared(
     std::move(clock), period_ns, std::move(callback), node_base->get_context(), autostart);
   node_timers->add_timer(timer, group);
+
+  // Get any subscriptions downstream from the publisher and add their graph_node_t to this one
+  for (auto & publisher : publishers) {
+    auto topic_name = publisher->get_topic_name();
+    timer->add_output_topic(topic_name);
+  }
+
   return timer;
 }
 
@@ -194,6 +203,7 @@ create_wall_timer(
   rclcpp::CallbackGroup::SharedPtr group,
   node_interfaces::NodeBaseInterface * node_base,
   node_interfaces::NodeTimersInterface * node_timers,
+  std::initializer_list<rclcpp::PublisherBase::SharedPtr> publishers = {},
   bool autostart = true)
 {
   if (node_base == nullptr) {
@@ -206,9 +216,16 @@ create_wall_timer(
 
   const std::chrono::nanoseconds period_ns = detail::safe_cast_to_period_in_ns(period);
 
-  // Add a new wall timer.
+  // Create a new wall timer.
   auto timer = rclcpp::WallTimer<CallbackT>::make_shared(
     period_ns, std::move(callback), node_base->get_context(), autostart);
+
+  // Get any subscriptions downstream from the publisher and add their graph_node_t to this one
+  for (auto & publisher : publishers) {
+    auto topic_name = publisher->get_topic_name();
+    timer->add_output_topic(topic_name);
+  }
+
   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/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,6 +33,8 @@ class BufferImplementationBase
   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;