Set subscriber depth to max from publishers

[davidferlop]

Public API Changes
None

Description
Set the max depth of a subscriber to the maximum max depth of all publishers. Currently, if a publisher has a depth higher than 10 the messages will be lost.

[bjsowa]

Currently, if a publisher has a depth higher than 10 the messages will be lost.

I don't buy this premise. From my understanding, subscriber will lose messages if the queue gets filled faster than the executor is able to process callbacks. This can happen if the publisher outputs messages with too high frequency (in which case setting higher depth will only delay the loss of messages) or the publisher outputs the messages in short bursts (in which case increasing subscriber depth will help). I think both situations can happen even if publisher uses lower depth QoS then subscriber.

I might be wrong though. Curious about your specific use-case.

[davidferlop]

You are correct that it is not enough that the publisher has a bigger queue, it is necessary that it is actually used, that is, that more than 10 messages are sent "simultaneously" (or faster than the subscriber can process them). When sending quick bursts of messages, having a longer queue on the publisher side will allow more messages to be sent "simultaneously", which will fill up the receiving queue faster, regardless of how quick the processing is, leadng to message loss (which can be solved by increasing the subscriber depth).

My specific use case is:
I have a publisher that outputs a burst of messages every second, something like this:

def timer_callback(self):
    for i in range(100):
        self.publisher.publish(String(data=f'Hello World {i}!'))

In that case the 100 messages fill the outgoing queue, get sent "simultaneously" (or close enough), and quickly fill the rosbridge subscriber queue, leading to message loss (only Hello world 90! to Hello world 99! will be received on the websocket client).

[bjsowa]

Have you tried setting queue_length on the client side? This should make rosbridge use internal queue for sending the messages so the ROS callbacks could be processed in a non-blocking fashion.

In that case the 100 messages fill the outgoing queue, get sent "simultaneously" (or close enough), and quickly fill the rosbridge subscriber queue, leading to message loss (only Hello world 90! to Hello world 99! will be received on the websocket client).

I guess we could treat the publisher's queue size as the declaration of how many messages in a bulk can it send but I don't see it as a golden rule. For example, what if there are multiple publishers? @sea-bass Do you have any opinion here?

[davidferlop]

Have you tried setting queue_length on the client side?

Setting queue_length on the client side doesn't really help in this scenario, as the issue is not between rosbridge and the client, but between the publisher and rosbridge. In this case the burst of messages fills the incoming queue on the ROS side of rosbridge before any message is processed, so the other messages are discarded at the ROS level and never seen by rosbridge.

For example, what if there are multiple publishers?

In the case of multiple publishers, the proposed code would simply set the queue to the maximum length between all of them, which will be enough to hold whichever burst is sent from any of the publishers. For the other publishers, which don't use that many messages, the rest of the queue will be unused.

In essence, the queue size of the publisher is the maximum ammount of messages that can be sent in a burst (as any additional messages will be lost by the outgoing queue), though it is true that there is no need to actually fill all the queue. If less messages than that number are sent in a sigle burst, the queue simply isn't filled, which doesn't really cause any issues.

[sea-bass]

Interesting. It is true that the depth=10 should probably not be hardcoded here:

rosbridge_suite/rosbridge_library/src/rosbridge_library/internal/subscribers.py

Line 117 in fde7509

depth=10,

At the same time, setting it to max probably has memory usage implications for users that do not need such high traffic of messages.

So maybe this should be a configurable launch parameter that defaults to 10, but can be increased by users?

[sea-bass]

Actually another idea to control this at the per-topic level... is the issue here also related to the fact that the queue_length is only available in the subscribe part of the protocol, but not the advertise?

Would adding one there help in this case?

[davidferlop]

At the same time, setting it to max probably has memory usage implications for users that do not need such high traffic of messages.

Yes, though I don't think the implications should be that bad, as it will only matter if the queue actually gets filled (wither too many messages too fast, or too big a burst), and in both cases you lose the messages otherwise.

So maybe this should be a configurable launch parameter that defaults to 10, but can be increased by users?

I don't think putting the queue size as a launch parameter is a good idea, as it would set the same queue for all topics, regardless of the need of that, while setting it to "max of publishers" sets that in a per-topic basis.

However, having a use-max-susbscriber-depth boolean parameter that, when set to true, configures the max_depth to the macimum between the publishers; and when set to false keeps the current behavior; might be the best solution here. @bjsowa , @sea-bass, should I implement it that way?

...is the issue here also related to the fact that the queue_length is only available in the subscribe part of the protocol, but not the advertise?

No, the queue_length parameter available on the subscriber creates an internal buffer queue in rosbridge, but it does not affect the ROS queue size. It could be usefult to have the queue_length parameter directly affect the QOS parameters for the rosbridge subscriber, but in that case dealing with multiple subscribers to the same topic will be trickier.

[bjsowa]

@davidferlop Would extending the protocol to be able to set subscriber depth from client side (for example, using the queue_size parameter) work for you?

[davidferlop]

Would extending the protocol to be able to set subscriber depth from client side (for example, using the queue_size parameter) work for you?

Yes, that would work as well.