Intra-process pipeline between different nodes in same process

[jackieokay]

In ros2/rmw_connext#76 I complained about nondeterministic startup behavior for 2 nodes communicating in the same process (but this was before the intra-process pipeline was implemented)). It was argued that this is acceptable behavior, since the two nodes are being initialized through DDS and DDS discovery can't make timing guarantees.

I would expect nodes within the same process to be able to avoid this nondeterministic startup behavior if the intra-process pipeline were utilized. This would be an acceptable solution.

However, the intra-process pipeline currently does not work for communication between two different nodes in the same process. I assume because the intra-process manager is a child of the context, not the node, that this feature should be implemented.

In branch multiple_nodes of system_tests, I added a case for publishing messages from one node to another node in the same process:

https://github.com/ros2/system_tests/blob/multiple_nodes/test_rclcpp/test/test_intra_process.cpp#L154

Currently the callback fails to trigger.

This feature is probably not urgent right now, but important for node composability.

[wjwwood]

The intraprocess communications still use DDS to pass information. So, I wouldn't expect it to behave any differently.

I'll look at the test, but these demos do work: https://github.com/ros2/ros2/wiki/Intra-Process-Communication

[wjwwood]

So, my theory as to why your test doesn't work, but my demos do, is that your test finishes before the two DDS participants have had time to discover each other. My demos just don't start working until the discovery process is complete, see:

https://github.com/ros2/demos/blob/master/intra_process_demo/src/cyclic_pipeline/cyclic_pipeline.cpp#L63

I added a sleep in your test and it passes:

diff --git a/test_rclcpp/test/test_intra_process.cpp b/test_rclcpp/test/test_intra_process.cpp
index 7dc3c64..17126d7 100644
--- a/test_rclcpp/test/test_intra_process.cpp
+++ b/test_rclcpp/test/test_intra_process.cpp
@@ -201,6 +201,8 @@ TEST(test_intra_process_two_nodes, nominal_usage) {
     publisher->publish(msg);
     ASSERT_EQ(0, counter);

+    rclcpp::sleep_for(1_s);
+
     // wait for the first callback
     {
       size_t i = 0;
@@ -231,10 +233,12 @@ TEST(test_intra_process_two_nodes, nominal_usage) {
     publisher->publish(msg);
     ASSERT_EQ(1, counter);

+    rclcpp::sleep_for(1_s);
+
     int i = 0;
     while (counter < 5 && i < 5) {
       executor.spin_some();
-      printf("spin_some() -  %zu/5\n", ++i);
+      printf("spin_some() -  %i/5\n", ++i);
     }
     ASSERT_EQ(5, counter);

@@ -248,7 +252,8 @@ TEST(test_intra_process_two_nodes, nominal_usage) {

   // check that no further callbacks have been invoked
   printf("spin_node_some() - no callbacks expected\n");
-  executor.spin_node_some(node1);
-  executor.spin_node_some(node2);
+  rclcpp::executors::SingleThreadedExecutor executor2;
+  executor2.spin_node_some(node1);
+  executor2.spin_node_some(node2);
   ASSERT_EQ(5, counter);
 }

(some other small fixes in there)

[jackieokay]

duplicate of ros2/rmw_connext#112

[clalancette]

I'm going to close this out since it is a duplicate.