Add option to set compression threads priority

[jmachowinski]

Added option to the the nice value of the compression thread.

[arneboe]

@emersonknapp Could you review this (or find someone to to it)? This is an important feature that enables us to use rosbag2 in production environments without grinding the whole system to a halt when writing large bag files.

[MichaelOrlov]

@jmachowinski Thank you for your contribution.
I think we can move forward with this PR and the proposed approach but need to address some minor issues and add missing parts in the dependent rosbag2 packages.

In particular, will need to:

1. update rosbag2_py package with the new data structure field.
2. update rosbag2_transport::RecordOptions package with the new data structure field.
3. update ros2bag/verb/record.py with a new command line option for compression threads priority value.
4. Consider if it will be possible to add some unit tests to check if threads priority sets up as expected.

I will doublecheck yet Windows implementation and let you know if changes are needed.

[jmachowinski]

In particular, will need to:

1. update `rosbag2_py` package with the new data structure field.

2. update `rosbag2_transport::RecordOptions` package with the new data structure field.

3. update `ros2bag/verb/record.py` with a new command line option for compression threads priority value.

4. Consider if it will be possible to add some unit tests to check if threads priority sets up as expected.

The python stuff is nothing, were a commit from ME would meet any quality standards... @kohrt can you have a look at this ?

[fujitatomoya]

besides all comments, and

Manual page getpriority(2)
...
BUGS
       According  to POSIX, the nice value is a per-process setting.  However, under the current Linux/NPTL implementation of
       POSIX threads, the nice value is a per-thread attribute: different threads in the same process can have different nice
       values.   Portable  applications should avoid relying on the Linux behavior, which may be made standards conformant in
       the future.

IMO, i am not inclined to take this fix.

related REP(ROS Enhancement Proposal): ros-infrastructure/rep#385

[jmachowinski]

2. update `rosbag2_transport::RecordOptions` package with the new data structure field.

Done

4. Consider if it will be possible to add some unit tests to check if threads priority sets up as expected.

Added a test case

[MichaelOrlov]

@jmachowinski Could you please rename compression_threads_nice_value parameter to something more generic like
compression_threads_priority and update descriptions in the header files?
Since on Windows we don't have such entities as nice value we do really have a thread priority.
I remember I've added a comment about it but it seems my comment disappeared in Github after you made a git push with --force parameter.

[MichaelOrlov]

@jmachowinski The selected approach for tests is not going to work. The test shall be rewritten.
See my comments in code.
I would propose creating a regular non-fake sequential_compression_writer and capture std::err output to verify logged error messages.
Something like

auto writer = ReaderWriterFactory::make_writer(record_options);
testing::internal::CaptureStderr();
writer->open(zero_cache_storage_options);
std::string test_output = testing::internal::GetCapturedStderr();
EXPECT_TRUE(
  test_output.find(
    "Could not set priority for compression thread to'") == std::string::npos);

and make two test cases with valid thread priority and invalid thread priority values.

Also please rename compression_threads_nice_value and thread_nice_value parameters to the compression_threads_priority and thread_priority respectively.

[jmachowinski]

@jmachowinski Could you please rename compression_threads_nice_value parameter to something more generic like compression_threads_priority and update descriptions in the header files? Since on Windows we don't have such entities as nice value we do really have a thread priority. I remember I've added a comment about it but it seems my comment disappeared in Github after you made a git push with --force parameter.

A priority is something special in the linux world, therefore I would not name it that.

[MichaelOrlov]

@jmachowinski As regards

A priority is something special in the linux world, therefore I would not name it that.

The thread priority is a generic term and will be applicable to the cross-platform implementation.
The fact that in the current implementation, the thread priority on Unix/Linux platforms will be settling up via nice value is an implementation details and would be considered as temporary workaround until [REP-2017] Thread attributes configuration support in rcl #385 will not be fully implemented.
Eventually, we will replace part of the code that is setting/getting nice value to the API from the [REP-2017]. But what is important is that the variables and parameter's names shall be agnostic to this temporary workaround.

[jmachowinski]

@MichaelOrlov
You latest changes trigger this warning :
Warning: class_loader.impl: SEVERE WARNING!!! A namespace collision has occurred with plugin factory for class FakeCompressor. New factory will OVERWRITE existing one. This situation occurs when libraries containing plugins are directly linked against an executable (the one running right now generating this message). Please separate plugins out into their own library or just don't link against the library and use either class_loader::ClassLoader/MultiLibraryClassLoader to open.

I suggest we move the check back into a class inside of the test function.

[jmachowinski]

@jmachowinski Could you please rename compression_threads_nice_value parameter to something more generic like compression_threads_priority

Done

[MichaelOrlov]

I suggest we move the check back into a class inside of the test function.

Hold on let me double-check. We didn't change the way how the library was linked.
This warning might have already existed before.

[jmachowinski]

I suggest we move the check back into a class inside of the test function.

Hold on let me double-check. We didn't change the way how the library was linked. This warning might have already existed before.

You added test/rosbag2_compression/fake_compressor.cpp to test_sequential_compression_writer this, triggers the warning

[MichaelOrlov]

@jmachowinski You can disregard this warning.
I made some investigation and found out that the case about what this warning signaling is safe to be used in this particular scenario.
I would prefer to have a such runtime warning in one test rather than duplicated class for fake_compressor.

If this warning is really bothering you and you want to get rid of it you can fix it by specifying "zstd" compression format instead of the DefaultTestCompressor in the open_succeeds_on_supported_compression_format test.

rosbag2/rosbag2_compression/test/rosbag2_compression/test_sequential_compression_writer.cpp

Lines 195 to 199 in 96c11b0

	TEST_F(SequentialCompressionWriterTest, open_succeeds_on_supported_compression_format)
	{
	rosbag2_compression::CompressionOptions compression_options{
	DefaultTestCompressor, rosbag2_compression::CompressionMode::FILE,
	kDefaultCompressionQueueSize, kDefaultCompressionQueueThreads};

But in this case, will need to add rosbag2_compression_zstd as a runtime dependenises for the tests in the package.xml file

[MichaelOrlov]

@jmachowinski Thanks for the renames and fixes.
We do have currently uint8_t for thread_priority and it is totally aligned with the range of the nice value on Linux/POSIX since it varies from -19 to 20.
Although it doesn't align with the possible thread's priority values on Windows.
According to the https://learn.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-setthreadpriority one of the biggest values could be settled up is

THREAD_MODE_BACKGROUND_END 0x00020000

In decimal, this is 131 072. On Windows, the type is int but this is not good to use int type in data structures since not strictly defined length of the integer type. i.e. it could vary from different platforms and compilators.
I think we need to change the type for the thread's priority at least to the int32_t or even better int64_t to accommodate all possible values.
We can use a static cast to int when passing it to set priority functions.

[MichaelOrlov]

@jmachowinski Could you please fix DCO and cpplint warnings?

[jmachowinski]

The linter is angry a bit:

I calmed it down ;-)

[MichaelOrlov]

Re-run Windows CI after an attempt to fix linker errors

• Windows

[MichaelOrlov]

@jmachowinski I am sorry, but we just merged another big PR and this branch has to be rebased and conflicts shall be resolved.
Could you please rebase your branch on top of the latest rolling ?

[jmachowinski]

rebased

[MichaelOrlov]

@jmachowinski since we added recently a feature for composable recorder. This is when recorder_options values getting from a node params. We need to add compression_thread_priority to the relevant parser.

Could you please add

record_options.compression_threads_priority = param_utils::declare_integer_node_params<int32_t>(
    node, "compression_threads_priority", std::numeric_limits<int32_t>::min(),
    std::numeric_limits<int32_t>::max(), record_options.compression_threads_priority);

to the RecordOptions get_record_options_from_node_params(rclcpp::Node & node) and modify the relevant test to make sure that the value parsed correctly?

[MichaelOrlov]

@jmachowinski I've made the requested change in the node params parser and relevant test for the compression_thread_priority by myself.
Please review and merge or cherry-pick changes to your branch if you don't have any concerns.

[jmachowinski]

Looks good to me, another try to merge it ?

[MichaelOrlov]

Re-run CI after an attempt to fix Windows build errors and rebase.
Job: ci_launcher
ci_launcher ran: https://ci.ros2.org/job/ci_launcher/13044

• Linux
• Linux-aarch64
• Windows

[MichaelOrlov]

@jmachowinski Unfortunately Windows build still fails with the following error messages

:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp(26,1): warning C4273: 'FakeCompressor::FakeCompressor': inconsistent dll linkage [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.hpp(29,3): message : see previous definition of '{ctor}' (compiling source file C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp(38,1): warning C4273: 'FakeCompressor::compress_uri': inconsistent dll linkage [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.hpp(31,15): message : see previous definition of 'compress_uri' (compiling source file C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp(44,44): warning C4273: 'FakeCompressor::compress_serialized_bag_message': inconsistent dll linkage [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.hpp(33,8): message : see previous definition of 'compress_serialized_bag_message' (compiling source file C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp(47,1): warning C4273: 'FakeCompressor::get_compression_identifier': inconsistent dll linkage [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.hpp(37,15): message : see previous definition of 'get_compression_identifier' (compiling source file C:\ci\ws\src\cellumation\rosbag2\rosbag2_compression\test\rosbag2_compression\fake_compressor.cpp) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

     Creating library C:/ci/ws/build/rosbag2_compression/Release/fake_plugin.lib and object C:/ci/ws/build/rosbag2_compression/Release/fake_plugin.exp

fake_compressor.obj : error LNK2019: unresolved external symbol "__declspec(dllimport) public: __cdecl FakeCompressor::FakeCompressor(void)" (__imp_??0FakeCompressor@@QEAA@XZ) referenced in function "public: virtual class rosbag2_compression::BaseCompressorInterface * __cdecl class_loader::impl::MetaObject<class FakeCompressor,class rosbag2_compression::BaseCompressorInterface>::create(void)const " (?create@?$MetaObject@VFakeCompressor@@VBaseCompressorInterface@rosbag2_compression@@@impl@class_loader@@UEBAPEAVBaseCompressorInterface@rosbag2_compression@@XZ) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

fake_compressor.obj : error LNK2019: unresolved external symbol "__declspec(dllimport) public: virtual __cdecl FakeCompressor::~FakeCompressor(void)" (__imp_??1FakeCompressor@@UEAA@XZ) referenced in function "public: virtual void * __cdecl FakeCompressor::`scalar deleting destructor'(unsigned int)" (??_GFakeCompressor@@UEAAPEAXI@Z) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

fake_compressor.obj : error LNK2019: unresolved external symbol "__declspec(dllimport) const FakeCompressor::`vftable'" (__imp_??_7FakeCompressor@@6B@) referenced in function "public: __cdecl FakeCompressor::FakeCompressor(int &)" (??0FakeCompressor@@QEAA@AEAH@Z) [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

C:\ci\ws\build\rosbag2_compression\Release\fake_plugin.dll : fatal error LNK1120: 3 unresolved externals [C:\ci\ws\build\rosbag2_compression\fake_plugin.vcxproj]

[jmachowinski]

I just set up a windows machine....

[jmachowinski]

Compiles and tests now fine under windows.

[MichaelOrlov]

Re-run Windows CI job after the attempt to fix build errors.

• Windows

[ros-discourse]

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

https://discourse.ros.org/t/ros-2-tsc-meeting-minutes-for-2023-12-14/35153/1

[ros-discourse]

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

https://discourse.ros.org/t/ros-2-tsc-meeting-minutes-for-2024-01-18/35779/1