Gracefully handle SIGINT and SIGTERM in rosbag2 recorder (#1301)

* Gracefully handle SIGINT and SIGTERM signal for rosbag2 recorder

- Call recorder->stop() and exec_->cancel() instead of rclcpp::shutdown

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

* Use singleton for static executor in the rosbag2_py::Recorder

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

* Rollback to the non-static executor and don't call executor->cancel()

- In case when signal will arrive we will trigger our internal exit_
variable and wait while current exec_->spin_all(10msec) will exit.

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

* Spin recorder node in a separate thread for better handling exit

- Run exec->spin() in a separate thread, because we need to call
exec->cancel() after recorder->stop() to be able to send notifications
about bag split and close.
- Wait on conditional variable for exit_ flag

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

* Address race condition in rosbag2_py.test_record_cancel

- Add `record_thread.join()` before trying to parse metadata.

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

---------

Signed-off-by: Michael Orlov <[email protected]>
(cherry picked from commit 46a23e9)

# Conflicts:
#	rosbag2_py/src/rosbag2_py/_transport.cpp
#	rosbag2_py/test/test_transport.py

rosbag2_py/src/rosbag2_py/_transport.cpp

@@ -141,17 +141,20 @@ class Player
 
 class Recorder
 {
-private:
- std::unique_ptr<rclcpp::executors::SingleThreadedExecutor> exec_;
-
 public:
  Recorder()
  {
- rclcpp::init(0, nullptr);
- exec_ = std::make_unique<rclcpp::executors::SingleThreadedExecutor>();
+ auto init_options = rclcpp::InitOptions();
+ init_options.shutdown_on_signal = false;
+ rclcpp::init(0, nullptr, init_options, rclcpp::SignalHandlerOptions::None);
+
  std::signal(
  SIGTERM, [](int /* signal */) {
- rclcpp::shutdown();
+ rosbag2_py::Recorder::cancel();
+ });
+ std::signal(
+ SIGINT, [](int /* signal */) {
+ rosbag2_py::Recorder::cancel();
  });
  }
 
@@ -164,6 +167,8 @@ class Recorder
  const rosbag2_storage::StorageOptions & storage_options,
  RecordOptions & record_options)
  {
+ exit_ = false;
+ auto exec = std::make_unique<rclcpp::executors::SingleThreadedExecutor>();
  if (record_options.rmw_serialization_format.empty()) {
  record_options.rmw_serialization_format = std::string(rmw_get_serialization_format());
  }
@@ -173,20 +178,55 @@ class Recorder
  std::move(writer), storage_options, record_options);
  recorder->record();
 
- exec_->add_node(recorder);
- // Release the GIL for long-running record, so that calling Python code can use other threads
+ exec->add_node(recorder);
+ // Run exec->spin() in a separate thread, because we need to call exec->cancel() after
+ // recorder->stop() to be able to send notifications about bag split and close.
+ auto spin_thread = std::thread(
+ [&exec]() {
+ exec->spin();
+ });
  {
+ // Release the GIL for long-running record, so that calling Python code can use other threads
  py::gil_scoped_release release;
- exec_->spin();
+ std::unique_lock<std::mutex> lock(wait_for_exit_mutex_);
+ wait_for_exit_cv_.wait(lock, [] {return rosbag2_py::Recorder::exit_.load();});
+ recorder->stop();
+ }
+ exec->cancel();
+ if (spin_thread.joinable()) {
+ spin_thread.join();
  }
+ exec->remove_node(recorder);
  }
 
- void cancel()
+ static void cancel()
  {
- exec_->cancel();
+ exit_ = true;
+ wait_for_exit_cv_.notify_all();
  }
+
+protected:
+ static std::atomic_bool exit_;
+ static std::condition_variable wait_for_exit_cv_;
+ std::mutex wait_for_exit_mutex_;
 };
 
+<<<<<<< HEAD
+=======
+std::atomic_bool Recorder::exit_{false};
+std::condition_variable Recorder::wait_for_exit_cv_{};
+
+// Return a RecordOptions struct with defaults set for rewriting bags.
+rosbag2_transport::RecordOptions bag_rewrite_default_record_options()
+{
+ rosbag2_transport::RecordOptions options{};
+ // We never want to drop messages when converting bags, so set the compression queue size to 0
+ // (unbounded).
+ options.compression_queue_size = 0;
+ return options;
+}
+
+>>>>>>> 46a23e9 (Gracefully handle SIGINT and SIGTERM in rosbag2 recorder (#1301))
 // Simple wrapper to read the output config YAML into structs
 void bag_rewrite(
  const std::vector<rosbag2_storage::StorageOptions> & input_options,
@@ -283,8 +323,15 @@ PYBIND11_MODULE(_transport, m) {
 
  py::class_<rosbag2_py::Recorder>(m, "Recorder")
  .def(py::init())
+<<<<<<< HEAD
  .def("record", &rosbag2_py::Recorder::record)
  .def("cancel", &rosbag2_py::Recorder::cancel)
+=======
+ .def(
+ "record", &rosbag2_py::Recorder::record, py::arg("storage_options"), py::arg("record_options"),
+ py::arg("node_name") = "rosbag2_recorder")
+ .def_static("cancel", &rosbag2_py::Recorder::cancel)
+>>>>>>> 46a23e9 (Gracefully handle SIGINT and SIGTERM in rosbag2 recorder (#1301))
  ;
 
  m.def(

rosbag2_py/test/test_transport.py

@@ -81,7 +81,19 @@ def test_record_cancel(tmp_path):
 
  recorder.cancel()
 
+<<<<<<< HEAD
  metadata_path = Path(bag_path) / 'metadata.yaml'
  db3_path = Path(bag_path) / 'test_record_cancel_0.db3'
  assert wait_for(lambda: metadata_path.is_file() and db3_path.is_file(),
+=======
+ metadata_io = rosbag2_py.MetadataIo()
+ assert wait_for(lambda: metadata_io.metadata_file_exists(bag_path),
+ timeout=rclpy.duration.Duration(seconds=3))
+ record_thread.join()
+
+ metadata = metadata_io.read_metadata(bag_path)
+ assert(len(metadata.relative_file_paths))
+ storage_path = Path(metadata.relative_file_paths[0])
+ assert wait_for(lambda: storage_path.is_file(),
+>>>>>>> 46a23e9 (Gracefully handle SIGINT and SIGTERM in rosbag2 recorder (#1301))
  timeout=rclpy.duration.Duration(seconds=3))