Graph executables comparison

rclcpp/CMakeLists.txt

@@ -72,7 +72,11 @@ 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/events_executor/events_executor_rt.cpp
+  src/rclcpp/experimental/executors/graph_executor.cpp
+  src/rclcpp/experimental/graph_executable.cpp
   src/rclcpp/experimental/timers_manager.cpp
+  src/rclcpp/experimental/timers_manager_rt.cpp
   src/rclcpp/future_return_code.cpp
   src/rclcpp/generic_publisher.cpp
   src/rclcpp/generic_subscription.cpp

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;
@@ -133,6 +134,17 @@ 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);
+  }
+
+  // Add input topic name and link to rcl_subscription_t pointer
+  sub->add_input_topic(sub->get_topic_name());
+  sub->add_key(sub->get_subscription_handle().get());
+
   node_topics_interface->add_subscription(sub, options.callback_group);
 
   return std::dynamic_pointer_cast<SubscriptionT>(sub);
@@ -183,12 +195,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 +227,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/executors.hpp

@@ -21,7 +21,7 @@
 #include "rclcpp/executors/multi_threaded_executor.hpp"
 #include "rclcpp/executors/single_threaded_executor.hpp"
 #include "rclcpp/executors/static_single_threaded_executor.hpp"
-#include "rclcpp/experimental/executors/events_executor/events_executor.hpp"
+#include "rclcpp/experimental/executors/events_executor/events_executor_rt.hpp"
 #include "rclcpp/node.hpp"
 #include "rclcpp/utilities.hpp"
 #include "rclcpp/visibility_control.hpp"