Use rw_lock to protect mcap metadata lists.

[fujitatomoya]

address #1542

[MichaelOrlov]

@fujitatomoya Thanks for trying to fix this issue with race condition.
At first glance, the fix is trivial. However, we can do way better.

• We don't need to use shared_mutex because there are no situations in rosbag2 when write() method could be called from multiple threads and unlikely would be because
  1. it is protected with mutex on upper rosbag2_cpp::writer::write(..) method for the cases when we are writing without cache.
  2. We are using SingleThreadedExecutor which is processing all subscription callbacks one by one sequentially in the one thread.
  3. When double-buffer cache is enabled we are using another method

     rosbag2/rosbag2_storage_mcap/src/mcap_storage.cpp

     Lines 784 to 790 in b8a9b06

     	void MCAPStorage::write(
     	const std::vector<std::shared_ptr<const rosbag2_storage::SerializedBagMessage>> & msgs)
     	{
     	for (const auto & msg : msgs) {
     	write(msg);
     	}
     	}

• And here we can do optimization. It will be more efficient to lock mutex once before we iterate through the vector of messages and call the write method for each message. Other API calls as create_topic(..) or remove_topic(..) have lower priority and could wait until we will finish dumping the vector of messages from our double-buffer cache to the storage. It will be more elegant to create a new private method for writing one message without lock. We can name it write_lock_free(msg) or write_locked(msg) and lock mutex in public write(..) method right before calling newly created write_lock_free(msg) and in the
  

  rosbag2/rosbag2_storage_mcap/src/mcap_storage.cpp

  Lines 784 to 790 in b8a9b06

  	void MCAPStorage::write(
  	const std::vector<std::shared_ptr<const rosbag2_storage::SerializedBagMessage>> & msgs)
  	{
  	for (const auto & msg : msgs) {
  	write(msg);
  	}
  	}

  
  right before for loop.

[fujitatomoya]

@MichaelOrlov thanks!

because there are no situations in rosbag2 when write() method could be called from multiple threads and unlikely would be because

I see, rosbag2_cpp::writer::write protects the methods with mutex lock.

whole point is rosbag2_cpp::cache::CacheConsumer::exec_consuming(), this one is staying outside of the lock mechanism.

We are using SingleThreadedExecutor which is processing all subscription callbacks one by one sequentially in the one thread.

currently true. minor concern was if we support MultiThreadedExecutor in the future, we will need to come back to this discussion. probably that is not concern, i guess rosbag2 takes the different path to take advantage of cache with dedicated single thread for the performance?

It will be more elegant to create a new private method for writing one message without lock. We can name it write_lock_free(msg) or write_locked(msg) and lock mutex in public write(..) method right before calling newly created write_lock_free(msg)

okay, can do that.

• will use std::mutex, instead of shared_mutex.
• conceal lock free implementation in private scope, then control lock in public scope.

[fujitatomoya]

@MichaelOrlov requesting another look! thanks!

[MichaelOrlov]

@fujitatomoya Thanks for the followup. As reagrds.

currently true. minor concern was if we support MultiThreadedExecutor in the future, we will need to come back to this discussion. probably that is not concern, i guess rosbag2 takes the different path to take advantage of cache with dedicated single thread for the performance?

Yes. Currently using double-buffer cache gives the same performance gain as a multithreaded executor but in two threads and decouples writing to the storage and transport layer. Which is very important due to the possible delays with filesystem or DB operations.
Even if we would start using MutithreadedExecutor for whatever reasons in the future we still bounded to the locking on one std::mutex on the rosbag2_cpp writer(message) layer here

rosbag2/rosbag2_cpp/src/rosbag2_cpp/writer.cpp

Lines 104 to 108 in edda376

	void Writer::write(std::shared_ptr<const rosbag2_storage::SerializedBagMessage> message)
	{
	std::lock_guard<std::mutex> writer_lock(writer_mutex_);
	writer_impl_->write(message);
	}

And would need to reconsider and make changes on that level as well.
Switching to shared_mutex on the rosbag2_storage_mcap layer if would needed in the future will not cause difficulties even with backporting since the whole changes in the .cpp file only and will be API/ABI compatible.

[MichaelOrlov]

@fujitatomoya Thanks. LGTM with green CI.

[MichaelOrlov]

Gist: https://gist.githubusercontent.com/MichaelOrlov/a06b14d20610f8f8cc8007db7fc61efe/raw/06daf70cbe74c9b1eb6715c512d655cd101f648f/ros2.repos
BUILD args: --packages-above-and-dependencies rosbag2_storage_mcap rosbag2_tests
TEST args: --packages-above rosbag2_storage_mcap rosbag2_tests
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/13233

• Linux
• Linux-aarch64
• Windows

[MichaelOrlov]

https://github.com/Mergifyio backport iron humble

[mergify]

backport iron humble

✅ Backports have been created

• #1566 Use rw_lock to protect mcap metadata lists. (backport #1561) has been created for branch iron but encountered conflicts
• #1567 Use rw_lock to protect mcap metadata lists. (backport #1561) has been created for branch humble but encountered conflicts

[ros-discourse]

This pull request has been mentioned on ROS Discourse. There might be relevant details there:

https://discourse.ros.org/t/ros-2-meeting-minutes-2024-02-15/36221/1