Make snapshot writing into a new file each time it is triggered

[MichaelOrlov]

This PR will make snapshot writing into a new file each time it is triggered.

• Relates Snapshot should write to a different file every time it is triggered #1099
• This PR is a replacement for the Feature/snapshot write different file each trigger rolling #1839

Note. Snapshot now becomes a blocking call and mutually exclusive with the writer::write(message) method to avoid race conditions, i.e., blocking the same writer_mutex_.
The set_data_ready() method from #1839 is not needed since "snapshot" has now become a blocking call; therefore, there are no race conditions, and we will not dump data from the cyclic buffer twice because data_ready will be settled false after the first dump with message_cache_->notify_data_ready();.

rosbag2/rosbag2_cpp/src/rosbag2_cpp/cache/circular_message_cache.cpp

Lines 95 to 106 in 786c3c4

	void CircularMessageCache::swap_buffers()
	{
	std::lock_guard<std::mutex> producer_lock(producer_buffer_mutex_);
	// Swap buffers only if data is ready. Data not ready when we are calling flushing on exit and
	// we should not dump buffer on exit if snapshot has not been triggered.
	if (data_ready_) {
	std::lock_guard<std::mutex> consumer_lock(consumer_buffer_mutex_);
	consumer_buffer_->clear();
	std::swap(producer_buffer_, consumer_buffer_);
	data_ready_ = false;
	}
	}

However, we still need to call split_bagfile_local(true) to trigger callbacks and open a new storage file.

This PR could be backported. There are no API/ABI breaking changes in it.

[cmuehlbacher]

Thank you for all the work on this feature.

Looks very promising. And great news that it can be backported.

As far as I can tell there should be no data races in there although, there will be several calls of write_messages function with an empty array.

There is only one small remark:
The test snapshot_writes_to_new_file_with_file_compression closes the bag file after taking a snapshot while snapshot_writes_to_new_file does not. It would make sense to have both tests behave similarly.

[christophebedard]

from what I understand, the tests write messages into the cache, then trigger a single snapshot, and then validate everything.

I believe we should test that this works with more than 1 snapshot. Could we check the metadata too, e.g., starting_time, duration, and message_count too?

[christophebedard]

Note. Snapshot now becomes a blocking call and mutually exclusive with the writer::write(message) method to avoid race conditions, i.e., blocking the same writer_mutex_.

I think this makes sense.

[fujitatomoya]

lgtm with green CI, just one minor comment that does not block this PR.

[MichaelOrlov]

@christophebedard I added a check for metadata in the test. To cover the case if snapshot could be called twice - I created a separate test case.

@cmuehlbacher As regards

As far as I can tell there should be no data races in there although, there will be several calls of write_messages function with an empty array.

No it is not. I double-checked in debuuger and we have constraints in tests that write messages will be called only once per snapshot.

As regards:

The test snapshot_writes_to_new_file_with_file_compression closes the bag file after taking a snapshot while > snapshot_writes_to_new_file does not. It would make sense to have both tests behave similarly.

The reason why I use writer->close() in the snapshot_writes_to_new_file_with_file_compression is because otherwise without close it will be a race conditions in tests. The close method will wait for the compressor threads to finish.
I don't want to call writer->close() in another snapshot_writes_to_new_file_with_bag_split test because want to check that file will be finalized when calling snapshot, without calling writer->close() method.

[christophebedard]

Looks good to me with green CI.

[MichaelOrlov]

Pulls: #1842
Gist: https://gist.githubusercontent.com/MichaelOrlov/127b9b542e0ed83d235de01e67be6919/raw/6dec255f6b87621d4194067c15f7fa94ebac1f63/ros2.repos
BUILD args: --packages-above-and-dependencies rosbag2_compression rosbag2_cpp rosbag2_tests
TEST args: --packages-above rosbag2_compression rosbag2_cpp rosbag2_tests
ROS Distro: rolling
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/14765

• Linux
• Linux-aarch64
• Linux-rhel
• Windows

[MichaelOrlov]

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

[mergify]

backport jazzy humble

✅ Backports have been created

• #1849 Make snapshot writing into a new file each time it is triggered (backport #1842) has been created for branch jazzy
• #1850 Make snapshot writing into a new file each time it is triggered (backport #1842) has been created for branch humble

[reinzor]

Great! Thanks for this feature!