diff --git a/rosbag2_transport/include/rosbag2_transport/recorder.hpp b/rosbag2_transport/include/rosbag2_transport/recorder.hpp index efc9eb0c36..0ce39ca1b2 100644 --- a/rosbag2_transport/include/rosbag2_transport/recorder.hpp +++ b/rosbag2_transport/include/rosbag2_transport/recorder.hpp @@ -164,6 +164,7 @@ class Recorder : public rclcpp::Node std::unordered_set topic_unknown_types_; rclcpp::Service::SharedPtr srv_snapshot_; std::atomic paused_ = false; + // Keyboard handler std::shared_ptr keyboard_handler_; // Toogle paused key callback handle @@ -172,8 +173,8 @@ class Recorder : public rclcpp::Node // Variables for event publishing rclcpp::Publisher::SharedPtr split_event_pub_; - bool event_publisher_thread_should_exit_ = false; - bool write_split_has_occurred_ = false; + std::atomic event_publisher_thread_should_exit_ = false; + std::atomic write_split_has_occurred_ = false; rosbag2_cpp::bag_events::BagSplitInfo bag_split_info_; std::mutex event_publisher_thread_mutex_; std::condition_variable event_publisher_thread_wake_cv_; diff --git a/rosbag2_transport/src/rosbag2_transport/recorder.cpp b/rosbag2_transport/src/rosbag2_transport/recorder.cpp index a76777e02e..4d0c2649a1 100644 --- a/rosbag2_transport/src/rosbag2_transport/recorder.cpp +++ b/rosbag2_transport/src/rosbag2_transport/recorder.cpp @@ -121,7 +121,13 @@ void Recorder::stop() { stop_discovery_ = true; if (discovery_future_.valid()) { - discovery_future_.wait(); + auto status = discovery_future_.wait_for(2 * record_options_.topic_polling_interval); + if (status != std::future_status::ready) { + RCLCPP_ERROR_STREAM( + get_logger(), + "discovery_future_.wait_for(" << record_options_.topic_polling_interval.count() << + ") return status: " << (status == std::future_status::timeout ? "timeout" : "deferred")); + } } paused_ = true; subscriptions_.clear(); @@ -135,6 +141,7 @@ void Recorder::stop() if (event_publisher_thread_.joinable()) { event_publisher_thread_.join(); } + RCLCPP_INFO(get_logger(), "Recording stopped"); } void Recorder::record() @@ -190,15 +197,13 @@ void Recorder::record() discovery_future_ = std::async(std::launch::async, std::bind(&Recorder::topics_discovery, this)); } + RCLCPP_INFO(get_logger(), "Recording..."); } void Recorder::event_publisher_thread_main() { RCLCPP_INFO(get_logger(), "Event publisher thread: Starting"); - - bool should_exit = false; - - while (!should_exit) { + while (!event_publisher_thread_should_exit_.load()) { std::unique_lock lock(event_publisher_thread_mutex_); event_publisher_thread_wake_cv_.wait( lock, @@ -222,10 +227,7 @@ void Recorder::event_publisher_thread_main() "Failed to publish message on '/events/write_split' topic."); } } - - should_exit = event_publisher_thread_should_exit_; } - RCLCPP_INFO(get_logger(), "Event publisher thread: Exiting"); }