Add Record, Stop, StartDiscovery, StopDiscovery and IsDiscoveryRunning services for Recorder #2248

MichaelOrlov · 2025-11-26T05:05:07Z

Description

This PR is a replacement for the stale PR #1840 and adds the following services to the rosbag2_transport::Recorder

Record
Stop
StartDiscovery
StopDiscovery
IsDiscoveryRunning
Fixes Add record, stop, start_discovery and stop_discovery services for rosbag2_transport::Recorder #1634
Relates #[FEA] Support distributed recording to get ultra high performance. #1548

Is this user-facing behavior change?

Yes. The user will have the ability to manage the Rosbag2 recorder instance remotely via the service calls.

Did you use Generative AI?

Yes. I am using GitHub Copilot, GPT-5.0, mostly to help me with tests and Doxygen comments.

Additional Information

No APi/ABI breaking changes -> can be backported

MichaelOrlov · 2025-12-06T19:09:25Z

@ros-pull-request-builder retest this please

MichaelOrlov · 2025-12-12T17:06:16Z

@christophebedard @fujitatomoya Friendly ping here for review. Or maybe @ahcorde or @Barry-Xu-2018 some of you can review this PR?

Barry-Xu-2018

LGTM.

I have a small question.

In RecorderImpl::stop(), in_recording_ is set to false only after the stop processing is completed.

rosbag2/rosbag2_transport/src/rosbag2_transport/recorder.cpp

Lines 321 to 334 in a8b9c82

    
           void RecorderImpl::stop() 
        
           { 
        
             std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_); 
        
             if (!in_recording_) { 
        
               RCLCPP_DEBUG(node->get_logger(), "Recording has already been stopped or not running."); 
        
               return; 
        
             } 
        
             stop_discovery(); 
        
             pause(); 
        
             subscriptions_.clear(); 
        
             writer_->close();  // Call writer->close() to finalize current bag file and write metadata 
        
             in_recording_ = false;

In RecorderImpl::record(), in_recording_ is set to true before record processing is complete.

rosbag2/rosbag2_transport/src/rosbag2_transport/recorder.cpp

Lines 353 to 356 in a8b9c82

    
           void RecorderImpl::record(const std::string & uri) 
        
           { 
        
             std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_); 
        
             if (in_recording_.exchange(true)) {

In RecorderImpl::stop(), is it possible to set in_recording_ to false before stop processing is started ? (I'm not sure if there were any specific considerations for current implementation.)

Considering that there is some processing time between when RecorderImpl::stop() is called and when in_recording_ is set to false, if a service callback is invoked during this period, the first check of in_recording_ would be meaningless.

  srv_stop_ = node->create_service<rosbag2_interfaces::srv::Stop>(
    "~/stop",
    [this](
      const std::shared_ptr<rmw_request_id_t>/* request_header */,
      const std::shared_ptr<rosbag2_interfaces::srv::Stop::Request>/* request */,
      const std::shared_ptr<rosbag2_interfaces::srv::Stop::Response>/* response */)
    {
      if (!in_recording_) {
        RCLCPP_WARN(node->get_logger(),
          "Received Stop request while not in recording. Ignoring request.");
      } else {
        try {
          this->stop();
        } catch (const std::exception & e) {
          RCLCPP_ERROR(node->get_logger(), "Error during Stop request: %s", e.what());
        }
      }
    }
  );

MichaelOrlov · 2025-12-23T20:13:56Z

@Barry-Xu-2018, good question and catch-up.
We use if (in_recording_.exchange(true)) in the record(uri) method as a leftover from the times when we were using it without a mutex lock.

rosbag2/rosbag2_transport/src/rosbag2_transport/recorder.cpp

Lines 353 to 356 in a8b9c82

    
           void RecorderImpl::record(const std::string & uri) 
        
           { 
        
             std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_); 
        
             if (in_recording_.exchange(true)) {

You are right, we shall set in_recording_ = true; at the end of the record(uri) method when the recorder and all its underlying data types are fully initialized and ready for recording to avoid race conditions and undefined behavior.

I made a fix in my last 0b2e478 commit

MichaelOrlov · 2025-12-23T20:16:03Z

@christophebedard @fujitatomoya @ahcorde This PR was reviewed, and findings were addressed. Can some of you formally approve this PR or review it additionally?

rosbag2_transport/src/rosbag2_transport/recorder.cpp

MichaelOrlov · 2026-01-06T03:34:45Z

Pulls: #2248
Gist: https://gist.githubusercontent.com/MichaelOrlov/a037975e5fc0e47f16e9b4531842f99f/raw/4f6447b9d365c81b919c094079a426d215d6c412/ros2.repos
BUILD args: --packages-above-and-dependencies rosbag2_transport rosbag2_tests
TEST args: --packages-above rosbag2_transport rosbag2_tests
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/17856

Linux
Linux-aarch64
Linux-rhel
Windows

Signed-off-by: Michael Orlov <[email protected]>

- Rationale: We use in_recording_ check in service calls, therefore the recorder and all its underlying data types shall be fully initialized and ready before we set this flag to true. Signed-off-by: Michael Orlov <[email protected]>

Signed-off-by: Michael Orlov <[email protected]>

MichaelOrlov · 2026-01-07T18:09:07Z

Made a rebase and resolved conflicts.

MichaelOrlov · 2026-01-07T18:09:59Z

@ahcorde @fujitatomoya A friendly ping for a formal approval or additional review here.

MichaelOrlov · 2026-01-07T21:14:25Z

Re-run CI after rebase and fixing conflicts
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/17878

Linux
Linux-aarch64
Linux-rhel
Windows

fujitatomoya

just a few minor comments. overall lgtm with all comments are resolved.

fujitatomoya · 2026-01-08T06:54:50Z

rosbag2_interfaces/srv/IsDiscoveryRunning.srv

+---
+# True if discovery is running, false otherwise
+bool running
+


(this comment does not block the PR)

although i understand there is already ~/is_paused as a service (this is consistent for the current implementation), i would use a read only parameter instead here. this is purely a state query with no side effects, and i believe exactly what parameters are designed for. in this way, we can reduce service endpoint to take advantage of internal parameter service.

fujitatomoya · 2026-01-08T06:57:31Z

rosbag2_transport/src/rosbag2_transport/recorder.cpp

+      const std::shared_ptr<rosbag2_interfaces::srv::Record::Request> request,
+      const std::shared_ptr<rosbag2_interfaces::srv::Record::Response> response)
+    {
+      if (in_recording_) {


are we missing mutex here?

Suggested change

if (in_recording_) {

std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_);

if (in_recording_) {

fujitatomoya · 2026-01-08T06:59:57Z

rosbag2_transport/src/rosbag2_transport/recorder.cpp

+      const std::shared_ptr<rosbag2_interfaces::srv::Stop::Request>/* request */,
+      const std::shared_ptr<rosbag2_interfaces::srv::Stop::Response>/* response */)
+    {
+      if (!in_recording_) {


same here?

Suggested change

if (!in_recording_) {

std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_);

if (!in_recording_) {

fujitatomoya · 2026-01-08T07:08:25Z

rosbag2_transport/src/rosbag2_transport/recorder.cpp

+        try {
+          this->stop();
+        } catch (const std::exception & e) {
+          RCLCPP_ERROR(node->get_logger(), "Error during Stop request: %s", e.what());


probably we can consider adding return_code/error_string to the Stop response for consistency with other services? errors are only logged, not returned to the caller.

fujitatomoya · 2026-01-08T07:11:46Z

rosbag2_transport/src/rosbag2_transport/recorder.cpp

 #include <algorithm>
+#include <chrono>
+#include <cstdint>
+#include <condition_variable>


is this required to add? doesn't appear to be used.

MichaelOrlov mentioned this pull request Nov 26, 2025

Add record, stop, start_discovery, stop_discovery and is_discovery_stopped services for rosbag2_transport::Recorder #1840

Closed

MichaelOrlov marked this pull request as ready for review November 26, 2025 05:07

MichaelOrlov requested review from christophebedard, emersonknapp and fujitatomoya December 6, 2025 19:08

MichaelOrlov requested review from Barry-Xu-2018 and ahcorde December 12, 2025 17:04

Barry-Xu-2018 approved these changes Dec 23, 2025

View reviewed changes

Tiryoh reviewed Dec 25, 2025

View reviewed changes

rosbag2_transport/src/rosbag2_transport/recorder.cpp Show resolved Hide resolved

Tiryoh approved these changes Jan 6, 2026

View reviewed changes

MichaelOrlov added 5 commits January 7, 2026 09:58

Add record/stop and discovery_start[stop], is_discovery_running services

86a1103

Signed-off-by: Michael Orlov <[email protected]>

Move services creation into create_control_services() and call from ctor

3fef96f

Signed-off-by: Michael Orlov <[email protected]>

Update README.md file with newly added services

f5207c7

Signed-off-by: Michael Orlov <[email protected]>

Make record_options_.start_paused to be used only in constructor

06e26d2

Signed-off-by: Michael Orlov <[email protected]>

MichaelOrlov force-pushed the morlov/add-record-stop-discovery-services branch from 2978286 to 06e26d2 Compare January 7, 2026 18:08

fujitatomoya approved these changes Jan 8, 2026

View reviewed changes

	void RecorderImpl::stop()
	{
	std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_);
	if (!in_recording_) {
	RCLCPP_DEBUG(node->get_logger(), "Recording has already been stopped or not running.");
	return;
	}

	stop_discovery();
	pause();
	subscriptions_.clear();
	writer_->close(); // Call writer->close() to finalize current bag file and write metadata

	in_recording_ = false;

	void RecorderImpl::record(const std::string & uri)
	{
	std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_);
	if (in_recording_.exchange(true)) {

	if (in_recording_) {
	std::lock_guard<std::mutex> state_lock(start_stop_transition_mutex_);
	if (in_recording_) {

Add Record, Stop, StartDiscovery, StopDiscovery and IsDiscoveryRunning services for Recorder #2248

Are you sure you want to change the base?

Add Record, Stop, StartDiscovery, StopDiscovery and IsDiscoveryRunning services for Recorder #2248

Conversation

MichaelOrlov commented Nov 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Is this user-facing behavior change?

Did you use Generative AI?

Additional Information

Uh oh!

MichaelOrlov commented Dec 6, 2025

Uh oh!

MichaelOrlov commented Dec 12, 2025

Uh oh!

Barry-Xu-2018 left a comment

Choose a reason for hiding this comment

Uh oh!

MichaelOrlov commented Dec 23, 2025

Uh oh!

MichaelOrlov commented Dec 23, 2025

Uh oh!

Uh oh!

MichaelOrlov commented Jan 6, 2026

Uh oh!

MichaelOrlov commented Jan 7, 2026

Uh oh!

MichaelOrlov commented Jan 7, 2026

Uh oh!

MichaelOrlov commented Jan 7, 2026

Uh oh!

fujitatomoya left a comment

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

fujitatomoya Jan 8, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

MichaelOrlov commented Nov 26, 2025 •

edited

Loading