ROS 2 Architecture Patterns | Robotics Development Hub

Building production-grade ROS 2 systems requires more than just knowing the API—it demands understanding architectural patterns that scale. This guide covers the essential patterns for building robust, maintainable, and performant ROS 2 applications. 1. Node Composition Patterns 1.1 Component-Based Nodes (Recommended) ROS 2's component architecture allows multiple nodes to run in a single process, reducing IPC overhead and enabling intra-process communication. // C++ Component Header #include #include namespace my_robot { class SensorFusionNode : public rclcpp::Node { public: explicit SensorFusionNode(const rclcpp::NodeOptions& options) : Node("sensor_fusion", options) { // Initialize publishers, subscribers, services imu_sub_ = create_subscription ( "imu/data", rclcpp::SensorDataQoS(), std::bind(&SensorFusionNode::imuCallback, this, _1)); fused_pub_ = create_publisher ( "imu/fused", rclcpp::SensorDataQoS()); } private: void imuCallback(const sensor_msgs::msg::Imu::SharedPtr msg) { // Fusion logic here fused_pub_->publish(*msg); } rclcpp::Subscription ::SharedPtr imu_sub_; rclcpp::Publisher ::SharedPtr fused_pub_; }; } RCLCPP_COMPONENTS_REGISTER_NODE(my_robot::SensorFusionNode) # Python Component import rclpy from rclpy.node import Node from sensor_msgs.msg import Imu class SensorFusionNode(Node): def __init__(self): super().__init__('sensor_fusion') self.imu_sub = self.create_subscription( Imu, 'imu/data', self.imu_callback, 10) self.fused_pub = self.create_publisher(Imu, 'imu/fused', 10) def imu_callback(self, msg): # Fusion logic self.fused_pub.publish(msg) def main(args=None): rclpy.init(args=args) node = SensorFusionNode() rclpy.spin(node) rclpy.shutdown() 1.2 Composition Launch Files 2. Lifecycle Management 2.1 Lifecycle Node States ROS 2 lifecycle nodes provide deterministic state transitions: unconfigured → inactive → active → finalized . # Lifecycle Node Implementation (Python) import rclpy from rclpy.lifecycle import LifecycleNode, State, TransitionCallbackReturn from std_msgs.msg import String class ManagedSensorNode(LifecycleNode): def __init__(self): super().__init__('managed_sensor') self.publisher_ = None self.timer_ = None def on_configure(self, state: State) -> TransitionCallbackReturn: self.get_logger().info(f'Configuring from {state.label}') self.publisher_ = self.create_lifecycle_publisher(String, 'sensor_data', 10) return TransitionCallbackReturn.SUCCESS def on_activate(self, state: State) -> TransitionCallbackReturn: self.get_logger().info(f'Activating from {state.label}') self.timer_ = self.create_timer(1.0, self.publish_data) return TransitionCallbackReturn.SUCCESS def on_deactivate(self, state: State) -> TransitionCallbackReturn: self.get_logger().info(f'Deactivating from {state.label}') self.timer_.cancel() return TransitionCallbackReturn.SUCCESS def on_cleanup(self, state: State) -> TransitionCallbackReturn: self.get_logger().info(f'Cleaning up from {state.label}') self.destroy_lifecycle_publisher(self.publisher_) return TransitionCallbackReturn.SUCCESS def on_shutdown(self, state: State) -> TransitionCallbackReturn: self.get_logger().info(f'Shutting down from {state.label}') return TransitionCallbackReturn.SUCCESS def publish_data(self): msg = String() msg.data = f"Sensor reading at {self.get_clock().now()}" self.publisher_.publish(msg) 2.2 Lifecycle Management via CLI # Configure all lifecycle nodes ros2 lifecycle set /managed_sensor configure # Activate ros2 lifecycle set /managed_sensor activate # Deactivate ros2 lifecycle set /managed_sensor deactivate # Shutdown ros2 lifecycle set /managed_sensor shutdown 3. DDS Tuning & QoS Profiles 3.1 Understanding QoS Policies Policy Options Use Case Reliability RELIABLE, BEST_EFFORT RELIABLE for commands, BEST_EFFORT for sensor streams Durability TRANSIENT_LOCAL, VOLATILE TRANSIENT_LOCAL for latched topics (map, params) History KEEP_LAST(N), KEEP_ALL KEEP_LAST(1) for latest state, KEEP_ALL for logs Deadline Duration Detect missed publications Lifespan Duration Auto-expire stale data 3.2 Built-in QoS Profiles # Sensor data - high frequency, loss tolerant rclcpp::SensorDataQoS() // BEST_EFFORT, KEEP_LAST(1) # Parameters / config - reliable, latched rclcpp::ParameterEventsQoS() // RELIABLE, TRANSIENT_LOCAL, KEEP_LAST(1) # Services - reliable request/response rclcpp::ServicesQoS() // RELIABLE, KEEP_LAST(10) # Custom profile for command topics rclcpp::QoS(10).reliable().transient_local().deadline(rclcpp::Duration(1, 0)) 3.3 DDS Implementation Selection # Fast DDS (default) - good general purpose export RMW_IMPLEMENTATION=rmw_fastrtps_cpp # Cyclone DDS - better for high throughput, real-time export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp # Zenoh - for WAN/edge scenarios export RMW_IMPLEMENTATION=rmw_zenoh_cpp # Per-node override (Fast DDS XML profile) # Create fastdds.xml: 0 20 7400 export FASTRTPS_DEFAULT_PROFILES_FILE=/path/to/fastdds.xml 4. Real-Time Executor Configuration 4.1 Executor Types # Single-threaded (default) rclpy.spin(node) # Multi-threaded for parallel callbacks executor = rclpy.executors.MultiThreadedExecutor(num_threads=4) executor.add_node(node) executor.spin() # StaticSingleThreadedExecutor - for real-time determinism # (requires real-time kernel and proper priority configuration) executor = rclpy.executors.StaticSingleThreadedExecutor() executor.add_node(node) executor.spin() 4.2 Callback Groups for Concurrency Control # Mutually exclusive callback group (sequential execution) mutually_exclusive_group = node.create_callback_group( rclpy.callback_groups.MutuallyExclusiveCallbackGroup()) # Reentrant callback group (parallel execution) reentrant_group = node.create_callback_group( rclpy.callback_groups.ReentrantCallbackGroup()) # Assign to subscriptions/timers/services sub = node.create_subscription( Imu, 'imu/data', callback, 10, callback_group=reentrant_group) timer = node.create_timer( 0.01, control_loop, callback_group=mutually_exclusive_group) 5. Parameter Management Patterns 5.1 Declarative Parameters class ConfigurableNode(Node): def __init__(self): super().__init__('configurable_node') # Declare with descriptors for validation from rcl_interfaces.msg import ParameterDescriptor, FloatingPointRange range_desc = FloatingPointRange(from_value=0.0, to_value=100.0, step=0.1) param_desc = ParameterDescriptor( description='Control loop frequency in Hz', floating_point_range=[range_desc], read_only=False ) self.declare_parameter('control_frequency', 50.0, param_desc) self.declare_parameter('robot_name', 'default_robot') self.declare_parameter('gains', [1.0, 0.1, 0.01]) # Array parameter # Parameter change callback self.add_on_set_parameters_callback(self.param_callback) def param_callback(self, params): for param in params: if param.name == 'control_frequency': self.get_logger().info(f'Frequency changed to {param.value}') self.update_timer_period(1.0 / param.value) return SetParametersResult(successful=True) 5.2 YAML Parameter Files # params.yaml /configurable_node: ros__parameters: control_frequency: 100.0 robot_name: "ur5e_arm" gains: [2.0, 0.5, 0.1] enable_logging: true topics: joint_states: "joint_states" cmd_vel: "cmd_vel" # Launch with parameters ros2 run my_pkg configurable_node --ros-args --params-file params.yaml 6. Inter-Process vs Intra-Process Communication 6.1 Enabling Intra-Process # In component container launch # Or programmatically rclcpp::NodeOptions options; options.use_intra_process_comms(true); auto node = std::make_shared (options); 6.2 Zero-Copy Data Transfer # Publisher with loaned messages (zero-copy) auto publisher = node->create_publisher ( "pointcloud", rclcpp::SensorDataQoS().keep_last(1)); // Loan message - avoids copy auto msg = publisher->borrow_loaned_message(); msg.get().header.stamp = node->now(); msg.get().height = 1; msg.get().width = 100000; // ... fill point cloud data publisher->publish(std::move(msg)); 7. Namespace & Topic Remapping 7.1 Hierarchical Namespaces # Results in: # /robot1/driver, /robot1/controller # /robot2/driver, /robot2/controller 7.2 Topic Remapping # CLI remapping ros2 run my_robot localization --ros-args --remap scan:=/robot1/scan 8. Error Handling & Diagnostics 8.1 Structured Logging # Use appropriate log levels RCLCPP_DEBUG(node->get_logger(), "Debug: processing frame %d", frame_id); RCLCPP_INFO(node->get_logger(), "Initialized sensor driver"); RCLCPP_WARN(node->get_logger(), "Dropped frame, latency: %f ms", latency); RCLCPP_ERROR(node->get_logger(), "Failed to connect to hardware: %s", err.what()); RCLCPP_FATAL(node->get_logger(), "Unrecoverable error, shutting down"); 8.2 Diagnostic Updater # diagnostic_updater for health monitoring #include class MonitoredNode : public rclcpp::Node { public: MonitoredNode() : Node("monitored_node") { updater_.setHardwareID("robot_001"); updater_.add("CPU Temperature", this, &MonitoredNode::checkCpuTemp); updater_.add("Memory Usage", this, &MonitoredNode::checkMemory); updater_.add("Sensor Status", this, &MonitoredNode::checkSensors); // Force update at 1Hz timer_ = create_wall_timer(1s, [this]() { updater_.force_update(); }); } private: void checkCpuTemp(diagnostic_updater::DiagnosticStatusWrapper& stat) { float temp = readCpuTemperature(); if (temp > 80.0) { stat.summary(diagnostic_msgs::msg::DiagnosticStatus::ERROR, "Overheating"); } else if (temp > 70.0) { stat.summary(diagnostic_msgs::msg::DiagnosticStatus::WARN, "High temperature"); } else { stat.summary(diagnostic_msgs::msg::DiagnosticStatus::OK, "Normal"); } stat.add("Temperature (C)", temp); } diagnostic_updater::Updater updater_; rclcpp::TimerBase::SharedPtr timer_; }; 9. Security (ROS 2 Security) 9.1 SROS2 Setup # Generate keystore ros2 security create_keystore /path/to/keystore # Create permissions ros2 security create_permission /path/to/keystore \ --identity robot1 \ --allow-publish /robot1/cmd_vel \ --allow-subscribe /robot1/scan \ --allow-service /robot1/* # Launch with security export ROS_SECURITY_KEYSTORE=/path/to/keystore export ROS_SECURITY_ENABLE=true export ROS_SECURITY_STRATEGY=Enforce 10. Testing Patterns 10.1 Unit Testing with gtest #include #include #include "my_robot/kinematics.hpp" TEST(KinematicsTest, ForwardKinematics) { std::vector joints = {0.0, -M_PI/2, 0.0, -M_PI/2, 0.0, 0.0}; auto pose = my_robot::forward_kinematics(joints); EXPECT_NEAR(pose.position.x, 0.5, 1e-3); EXPECT_NEAR(pose.position.y, 0.0, 1e-3); EXPECT_NEAR(pose.position.z, 0.3, 1e-3); } TEST(KinematicsTest, InverseKinematicsMultipleSolutions) { geometry_msgs::msg::Pose target; target.position.x = 0.5; target.position.y = 0.0; target.position.z = 0.3; auto solutions = my_robot::inverse_kinematics(target); EXPECT_GT(solutions.size(), 0); } 10.2 Integration Testing with launch_testing # test_launch.py import launch import launch_testing import pytest @pytest.mark.launch_test def generate_test_description(): return launch.LaunchDescription([ launch.actions.Node( package='my_robot', executable='driver', name='driver', output='screen' ), launch_testing.actions.ReadyToTest() ]) class TestDriverIntegration: def test_driver_starts(self, proc_output): proc_output.assertWaitFor('Driver initialized', timeout=10) def test_driver_publishes_data(self, proc_output): proc_output.assertWaitFor('Publishing scan data', timeout=5) 11. Performance Profiling 11.1 Tracing with ros2_tracing # Enable tracing export ROS_TRACE_DIR=/tmp/ros2_trace ros2 run my_robot my_node # Analyze with tracecompass or babeltrace babeltrace /tmp/ros2_trace | grep -E "(callback|publish|subscription)" 11.2 CPU/Memory Profiling # perf for CPU profiling perf record -g -- ros2 run my_robot my_node perf report # heaptrack for memory heaptrack ros2 run my_robot my_node heaptrack_gui heaptrack.my_node.*.gz Summary Checklist ✅ Use component-based nodes for intra-process communication ✅ Implement lifecycle nodes for managed components ✅ Choose QoS profiles per topic semantics ✅ Select DDS implementation for your deployment ✅ Configure executors and callback groups for real-time needs ✅ Use declarative parameters with validation ✅ Enable zero-copy for high-bandwidth topics ✅ Structure namespaces for multi-robot systems ✅ Implement diagnostics and structured logging ✅ Apply ROS 2 Security for production deployments ✅ Write unit and integration tests ✅ Profile regularly in CI/CD These patterns form the foundation for scalable ROS 2 systems. Start with component composition and lifecycle management, then layer on QoS tuning and real-time configuration as your requirements evolve. Related Articles Motion Planning Algorithms Robot Simulation with Gazebo Autonomous Navigation Stack Embedded Robotics