Add correct timing behaviour for rosbag play #32

Martin-Idel-SI · 2018-09-10T15:21:34Z

This PR adds correct timing behaviour for rosbag play

Uses a shared queue (introduces vendor package for shared queues - header only library!)
Use one thread to get messages from storage
Use one thread to publish according to timestamp in storage
Use a simple heuristic to make sure that the storage will not get starved and doesn't write too many messages to memory

In addition, a few technical aspects have been improved

Only depend on test_msgs (not std_msgs) in rosbag2 tests
Fix pluginlib classloader lifetime for rosbag play

anhosi · 2018-09-17T13:49:29Z

CI:

Linux
Linux-aarch64
macOS
Windows

anhosi · 2018-09-17T14:24:40Z

new CI after fixing test:

Linux
Linux-aarch64
macOS
Windows

Karsten1987 · 2018-09-17T16:45:09Z

rosbag2/src/rosbag2/demo_play.cpp

 #include "rosbag2/rosbag2.hpp"

 int main(int argc, const char ** argv)
 {
  rclcpp::init(argc, argv);

+  auto options = rosbag2::Rosbag2PlayOptions();
+  options.queue_buffer_length_ = 1000;


What exactly does this queue_buffer_length describe? Is it the amount of serialized messages to load?

Yes, it is the queue size up to which the storage is preloaded for smooth replaying.

rosbag2/include/rosbag2/rosbag2_play_options.hpp

Karsten1987 · 2018-09-17T16:51:03Z

rosbag2/src/rosbag2/player.cpp

+
+bool is_pending(const std::future<void> & future)
+{
+  return !(future.valid() && future.wait_for(std::chrono::seconds(0)) == std::future_status::ready);


what does wait_for(std::chrono::seconds(0)) do? Wait forever or return immediately?

It will return immediately.

Karsten1987 · 2018-09-17T17:00:03Z

rosbag2/include/rosbag2/rosbag2_play_options.hpp

+
+namespace rosbag2
+{
+struct Rosbag2PlayOptions


The options should also encompass the qos options in order to guarantee correct replay behavior.

Agreed, however I would prefer to do this later with a followup pr (#34) as we do not yet have the CLI integration.

Karsten1987 · 2018-09-17T17:22:59Z

rosbag2/src/rosbag2/player.cpp

+    message_queue_.enqueue(message);
+  }
+
+  auto queue_lower_boundary = options.queue_buffer_length_ - options.queue_buffer_length_ / 10;


a few comments on this:

Where does that magic number (90%) come from?

The result of this should be casted explicitely to a type. I believe you can lose decimal precision here.

Should this number be part of the Rosbag2PlayOptions struct?

About being part of Rosbag2PlayOptions: I am not sure. It would be an expert user option only and we definitely would need to provide a useful default, e.g. 90%.

There is actually no cast necessary as / is integer division in C++ which is what we want here.

This is also a candidate for Allow to specify more parameters for rosbag play #34.

Karsten1987 · 2018-09-17T20:57:58Z

rosbag2/src/rosbag2/replayable_message.hpp

+namespace rosbag2
+{
+
+struct ReplayableMessage


I think I would prefer that the SerializedBagMessage struct gets enhanced by the time_to_start field (maybe in another name), but I don't see the point of cascading these types, given that the serialized bag message was introduced for this purpose.
Because I believe this time point field can also be used otherwise besides publishing.

I disagree on that one. The SerializedBagMessage represents a message with meta data that can be stored in a bag file.
The ReplayableMessage represents a SerializedBagMessage enriched with play-specific meta data that is needed for playing. These are different things and thus should have different types.

rosbag2/src/rosbag2/rosbag2.cpp

rosbag2/test/rosbag2/rosbag2_rosbag_node_test.cpp

rosbag2/test/rosbag2/rosbag2_test_fixture.hpp

Karsten1987 · 2018-09-17T21:10:37Z

The tests seem to be (at least) flaky on my OSX.
I've run the gmock suite 10 times and rosbag2_record_all_integration_test constantly fails on my machine.

Also, I guess it'd be good to cleanup the tmp_tes_dir.* folder after a build run. This might flood your build folder.

The demo applications are currently not installed. So you can't run them with the ros2 run command.

When running the demo application, I have three files in my workspace

test.bag
test.bag-shm
test.bag-wal
What are these files?

The print of [INFO] [rosbag2_storage]: Stored message is flooding the console pretty heavily

for the record function: I think that an empty topic_names vector should lead to an exception. I believe the get_all_topics...() should be called outside of this function. This makes the API clearer, IMHO. Or the function can have an overload:
record(const std::string & file_name);
record(const std::string & file_name, const std::vector<std::string> topic_names);
whereas the first one fetches all available topics from the network and then calls the second function.
https://github.com/ros2/rosbag2/blob/master/rosbag2/include/rosbag2/rosbag2.hpp#L52
https://github.com/ros2/rosbag2/blob/master/rosbag2/src/rosbag2/rosbag2.cpp#L55-L57

For the generic subscription:
I think it makes sense to subscribe with info at this place to have meta data available at this time:
https://github.com/ros2/rmw/blob/master/rmw/include/rmw/types.h#L226-L231
I could see that this meta data is enhanced by a received time stamp in the future.

Overall:
The demo works and from a first test seems to perform correctly. Nice work! 🥇

rosbag2/include/rosbag2/rosbag2.hpp

Karsten1987 · 2018-09-17T22:06:53Z

rosbag2/src/rosbag2/player.cpp

+    message_queue_.size_approx() < options.queue_buffer_length_ &&
+    is_pending(storage_loading_future))
+  {
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));


I could see that this parameter should be exposed or configurable. Could we pack this into the Rosbag2PlayOptions struct?

Definitely possible. I am not yet sure how useful this parameter will be. I suggest to discuss this here (#34).

anhosi · 2018-09-18T14:30:15Z

We also see this flaky integration test. I created a issue (rosbag2_record_all_integration_test is flaky #35) because we are not sure whether this is a rosbag2 problem.
We noticed the broken test cleanup and are addressing it on the rosbag2_transport branch.
The test.bag-shm and test.bag-wal are internal files of sqlite3.
Not installing the demo programs is intentional and they will be removed with the rosbag2_transport branch as then the CLI will be working.
The "Stored message" logging has been removed.
I decided to use a Rosbag2::record overload for the record all usage.

anhosi · 2018-09-18T15:04:45Z

New CI after working on review comments:

Linux
Linux-aarch64
macOS
Windows

Karsten1987 · 2018-09-18T18:35:11Z

rosbag2/src/rosbag2/player.cpp

 {
  while (
-    message_queue_.size_approx() < options.queue_buffer_length_ &&
-    is_pending(storage_loading_future))
+    message_queue_.size_approx() < options.read_ahead_queue_size &&


does this queue have to be initialized in some way? Or is it guaranteed that size_approx() returns 0 by default initialization?

Karsten1987 · 2018-09-18T18:41:27Z

rosbag2/src/rosbag2/player.cpp

+  while (message_queue_.size_approx() != 0 || is_pending(storage_loading_future)) {
+    ReplayableMessage message;
+    if (message_queue_.try_dequeue(message)) {
+      std::this_thread::sleep_until(start_time + message.time_since_start);


Yeah, my point is that start_time + message.time_since_start could be smaller than now, right?
But from what I read, the sleep_until command doesn't block on a time point in the past. So this should be alright.

Karsten1987 · 2018-09-18T18:43:48Z

rosbag2/src/rosbag2/rosbag2.cpp

+void Rosbag2::record(const std::string & file_name)
+{
+  auto topics_and_types = node_->get_all_topics_with_types();
+  std::vector<std::string> topic_names(topics_and_types.size());


Should this function throw when no topics are available?

Karsten1987 · 2018-09-18T18:44:20Z

rosbag2/src/rosbag2/rosbag2.cpp

-    node->get_all_topics_with_types() :
-    node->get_topics_with_types(topic_names);
-
+  auto topics_and_types = node_->get_topics_with_types(topic_names);
  if (topics_and_types.empty()) {


I think this check should go into the other record function

As long as we do not have topic discovery after startup we should also report if we could not detect the topics that were explicitely specified. So I think is is an error for both record overloads and the location is fine.

Karsten1987

The last commit fixes the record all test for me:

3255d5c

For now the publishing starts only after the reading is completly done. This should change aufter ros2GH-68 is done and a thread-safe queue can be used instead of std::queue.

- The main (play) thread sleeps until the time for publishing the message is reached. - Using std::chrono time_point and duration for type-safe time arithmetic instead of rcutils time types.

- Subscribers need to maintain a longer history if the messages are not consumed fast enough.

The Classloader instance needs to outlive all objects created by it.

Reason: record and play have almost no common code but do the exact opposite with the storage and rclcpp.

- only required in tests - this decreases the amount of packages needed for a clean build without tests

- Download and install headers from moodycamel readerwriterqueue - Download and install headers from moodycamel concurrentqueue - Use readerwriterqueue in code to load and publish concurrently

- Only load if queue contains less than 1000 messages - Wait a millisecond before loading again once the queue is long enough

- gets rid of string_msgs dependency

- Currently, we write sequentially in order of arrival time so reading in id order is fine - This may change at a later time and should not change the reading behaviour, i.e. we need to read in order of timestamps

anhosi · 2018-09-19T08:46:23Z

New CI:

Linux
Linux-aarch64
macOS
Windows

Make the future a class member of player to avoid having to hand it into several functions which is difficult with a move-only type.

Add an explicit overload for record without a topic_names argument to record all topics.

Karsten1987 · 2018-09-20T00:46:52Z

new round of CI:

Karsten1987 · 2018-09-20T01:41:00Z

I'll go ahead and merge this. The flaky test is already ticketed in #37

Signed-off-by: Jacob Bandes-Storch <jacob@foxglove.dev>

Signed-off-by: Jacob Bandes-Storch <jacob@foxglove.dev> Signed-off-by: James Smith <james@foxglove.dev>

tfoote added the in review label Sep 10, 2018

Martin-Idel-SI force-pushed the feature/replay_timing branch 2 times, most recently from 8e8af82 to 6eea477 Compare September 14, 2018 07:06

Karsten1987 reviewed Sep 17, 2018

View reviewed changes

Karsten1987 approved these changes Sep 18, 2018

View reviewed changes

Karsten1987 reviewed Sep 18, 2018

View reviewed changes

anhosi and others added 18 commits September 19, 2018 09:34

ros2GH-69 Read storage content in a separate thread

0ddcbeb

For now the publishing starts only after the reading is completly done. This should change aufter ros2GH-68 is done and a thread-safe queue can be used instead of std::queue.

ros2GH-71 Add integration test for timing behavior

def5392

ros2GH-71 Retain time difference of messages when playing a bag file

cde004c

- The main (play) thread sleeps until the time for publishing the message is reached. - Using std::chrono time_point and duration for type-safe time arithmetic instead of rcutils time types.

ros2GH-71 Improve stability of read test

29dff0e

- Subscribers need to maintain a longer history if the messages are not consumed fast enough.

ros2GH-71 Fix Classloader instance lifetime

ac791f8

The Classloader instance needs to outlive all objects created by it.

ros2GH-71 Extract playing code into a class of its own

fa23d74

Reason: record and play have almost no common code but do the exact opposite with the storage and rclcpp.

ros2GH-70 Do not link explicitly against std_msgs

e92e67d

- only required in tests - this decreases the amount of packages needed for a clean build without tests

ros2GH-70 Fix error message of storage

5ad39d4

ros2GH-70 Fix pluginlib/storage issue for recording

81bfb1b

ros2GH-71 Cleanup: variable naming

c5cda7e

ros2GH-68 Introduce vendor package for shared queue

f6489e0

- Download and install headers from moodycamel readerwriterqueue - Download and install headers from moodycamel concurrentqueue - Use readerwriterqueue in code to load and publish concurrently

ros2GH-70 Load storage continuously instead of as fast as possible

71eac49

- Only load if queue contains less than 1000 messages - Wait a millisecond before loading again once the queue is long enough

ros2GH-70 Add options struct to allow specification of queue size

0f0f94e

ros2GH-72 Wait for messages to fill up

9f67dbb

ros2GH-74 Rename integration tests to play/record tests

d9c87d8

ros2GH-74 Use test_msgs in integration tests

4b5cc9e

- gets rid of string_msgs dependency

ros2GH-70 Rename is_not_ready to is_pending, use bulk reading to queue

75e8e35

ros2GH-70 Harmonize storage_loading_future variable

6ab44a0

Martin-Idel-SI and others added 3 commits September 19, 2018 09:34

ros2GH-88 Read messages in order of their timestamps

3921211

- Currently, we write sequentially in order of arrival time so reading in id order is fine - This may change at a later time and should not change the reading behaviour, i.e. we need to read in order of timestamps

Fix compiler error on Mac

3be2292

ros2GH-8 Fix: use correct ros message type in test

3345e20

anhosi and others added 5 commits September 19, 2018 11:14

ros2GH-8 Cleanup: minor code style fixes

3665a34

ros2GH-8 Refactor future usage in player

463d721

Make the future a class member of player to avoid having to hand it into several functions which is difficult with a move-only type.

ros2GH-8 Cleanup: remove verbose logging for every stored message

3117261

ros2GH-8 Refactor rosbag2 interface

f3da51f

Add an explicit overload for record without a topic_names argument to record all topics.

fix: call vector.reserve instead of default initalization

08cfd51

greimela-si force-pushed the feature/replay_timing branch from 3255d5c to 08cfd51 Compare September 19, 2018 09:22

fix record demo

5f89708

Karsten1987 merged commit bc822f1 into ros2:master Sep 20, 2018

Karsten1987 removed the in review label Sep 20, 2018

Martin-Idel-SI deleted the feature/replay_timing branch September 20, 2018 06:30

This was referenced Sep 28, 2018

rosbag2_play_integration_test times out on Windows #37

Closed

rosbag2_record_all_integration_test is flaky #35

Closed

james-rms pushed a commit to james-rms/rosbag2 that referenced this pull request Nov 17, 2022

Update email address for Foxglove maintainers (ros2#32)

65175a8

Signed-off-by: Jacob Bandes-Storch <jacob@foxglove.dev>

james-rms pushed a commit to james-rms/rosbag2 that referenced this pull request Nov 17, 2022

Update email address for Foxglove maintainers (ros2#32)

d83b081

Signed-off-by: Jacob Bandes-Storch <jacob@foxglove.dev> Signed-off-by: James Smith <james@foxglove.dev>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add correct timing behaviour for rosbag play #32

Add correct timing behaviour for rosbag play #32

Martin-Idel-SI commented Sep 10, 2018

anhosi commented Sep 17, 2018

anhosi commented Sep 17, 2018

Karsten1987 Sep 17, 2018

anhosi Sep 18, 2018

Karsten1987 Sep 17, 2018

anhosi Sep 18, 2018

Karsten1987 Sep 17, 2018

anhosi Sep 18, 2018 •

edited

Loading

Karsten1987 Sep 17, 2018 •

edited

Loading

anhosi Sep 18, 2018 •

edited

Loading

Karsten1987 Sep 17, 2018

anhosi Sep 18, 2018

Karsten1987 commented Sep 17, 2018 •

edited

Loading

Karsten1987 Sep 17, 2018

anhosi Sep 18, 2018

anhosi commented Sep 18, 2018 •

edited

Loading

anhosi commented Sep 18, 2018

Karsten1987 Sep 18, 2018

Karsten1987 Sep 18, 2018

Karsten1987 Sep 18, 2018

Karsten1987 Sep 18, 2018

anhosi Sep 19, 2018

Karsten1987 left a comment •

edited

Loading

anhosi commented Sep 19, 2018

Karsten1987 commented Sep 20, 2018 •

edited

Loading

Karsten1987 commented Sep 20, 2018

Add correct timing behaviour for rosbag play #32

Add correct timing behaviour for rosbag play #32

Conversation

Martin-Idel-SI commented Sep 10, 2018

anhosi commented Sep 17, 2018

anhosi commented Sep 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anhosi Sep 18, 2018 • edited Loading

Choose a reason for hiding this comment

Karsten1987 Sep 17, 2018 • edited Loading

Choose a reason for hiding this comment

anhosi Sep 18, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Karsten1987 commented Sep 17, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anhosi commented Sep 18, 2018 • edited Loading

anhosi commented Sep 18, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Karsten1987 left a comment • edited Loading

Choose a reason for hiding this comment

anhosi commented Sep 19, 2018

Karsten1987 commented Sep 20, 2018 • edited Loading

Karsten1987 commented Sep 20, 2018

anhosi Sep 18, 2018 •

edited

Loading

Karsten1987 Sep 17, 2018 •

edited

Loading

anhosi Sep 18, 2018 •

edited

Loading

Karsten1987 commented Sep 17, 2018 •

edited

Loading

anhosi commented Sep 18, 2018 •

edited

Loading

Karsten1987 left a comment •

edited

Loading

Karsten1987 commented Sep 20, 2018 •

edited

Loading