Subscribers with QoS History policy set to 'KeepLast' don't behave as expected

[lukerob01]

Firstly, I would like to thank you for creating such an innovative crate. It is nice to see someone trying to expose ROS in a rust-first way as opposed to just following the C++ implementation.

Issue

The issue I am describing happens when a subscriber is created with a QoS profile where 'History' is set to 'KeepLast', and when that subscriber is read much less frequently than the rate at which Node::spin_once is called. This can come up easily in robotics code. For example, say you wanted to infrequently read the latest value of a sensor topic (e.g. reading heading every time a waypoint is reached), you might think of creating a subscriber with a QoS profile set to QosProfile::default().keep_last(1) so that you got the most recently received message each time the subscriber was read. The problem with the example I have just given is that, although reading the descriptions of the QoS policies would lead you to this solution, it would not currently work in r2r.

Example

I have attached a minimal example with a talker and a slow listener. Both are set with the QoS profile listed above with a depth of 1. The talker sends messages at 10 Hz, and the slow listener reads messages (with let msg = subscriber.next().await) at 2 Hz. Both run the Node::spin_once loop in a separate blocking thread as per r2r examples.

Expected behaviour

The subscriber prints the most recent message to the console every 500 ms, leading to terminal output which looks like the following:

received_msg: Counter value: 1
received_msg: Counter value: 6
received_msg: Counter value: 11
received_msg: Counter value: 16
received_msg: Counter value: 21
received_msg: Counter value: 26
...

Behaviour seen

The subscriber still prints messages to the console every 500 ms, but they are all old. It starts by printing messages in order and then settles to messages with a constant offset to the most recent message. An example is shown below:

Counter value: 1
Counter value: 2
Counter value: 3
...
Counter value: 12
Counter value: 13
Counter value: 14
Counter value: 17
Counter value: 22
Counter value: 27
Counter value: 32
Counter value: 37
Counter value: 42
...

Reason

This happens due to the underlying architecture of r2r. Although Node::spin_once will obey the QoS policy (I assume) and pick the most recent message to add to the subscriber stream, in almost all programs, this method is called in a very fast loop. This has the result that it keeps adding messages to the subscriber stream as they arrive, regardless of if the stream is being emptied. (This explains the first 14 messages of the observed output.) After a while, the spin_once method will fill the subscriber stream and start dropping messages; however, as soon as a message is read by an executor, the stream has space again and the spin_once method will fill it back up again. (This explains the remaining messages in the observed output.)

Workaround

There are a few work arounds for this. For example, when you want to infrequently read from this topic, instead of just writing

let msg = subscriber.next().await?;

you could instead empty the queue with

let mut msg = subscriber.next().await?;
while let Some(Some(stale_msg)) = subscriber.next().now_or_never() {
    msg = stale_msg;
}

(although if the stream has been full for a while, this would still not be the most recent message). Additionally, you could also spawn a task that continually reads the next message and updates a shared variable or something like a Tokio watch channel with the most recent value (essentially implementing the QoS policy manually). However, both of these methods are quite involved and a new user would not expect to have to do this themselves when they give this specific QoS policy to the Node::subscribe method (it caught me out today).

Solutions

I can't think of any solution that would work with the current return type from Node::subscribe. The only way I could see to fix this would be to change the method of inter-task communication between the node and the subscriber handle depending on what the QoS History profile is set to. Either way, this could perhaps be better documented at the Node and QoS structs and maybe an example given.

Sorry for such a long post and I apologise if I've misunderstood anything about the crate (I'm quite new to it).

qos_test.zip