Limit SinkAsync coro queue

[phettberg]

When creating a SinkAsync object, using the mode op.AsyncMode.QUEUE , this coro queue could basically grow limitlessly. In case the queue is filled up too fast due to a missed edge case and/or limited system resources (e.g., embedded), this could kill the application.

To handle this, I suggest introducing an optional, additional parameter max_queue_threshold to the SinkAsync class. As the name suggests, this is a "threshold" rather than an absolute limit, which would kind of defeat the purpose of the queue mode. In case the length of the CoroQueue exceeds max_queue_threshold , CoroQueue shall emit a RuntimeError. It is the developers' responsibility to

• define a reasonable threshold for their application, which is not too sharp but also keeps the application running safely.
• handle the runtime error. Either use the default_error_handler and deliberately throw a system error or use error_callback of the SinkAsync class for custom handling.

[semiversus]

(Unlimited length) queues are may be a bad idea here, right. Same applies to MapAsync.

Approach 1: Adding an additional parameter will work.

Approach 2: Another way would be to make ._coro_queue attribute public available. Adding a .queue_length property to CoroQueue would enable to read out the length of the queue (may also helps for LAST and LAST_DISTINCT). The check of a to "filled" queue and the handling what to do would then be in the responsibility of the caller.

What do you think?

[phettberg]

Approach 1 appeals more to me. IMHO, if the unfortunate happens and the queue is growing too fast, the developer already made a mistake in the design. I don't see how a smart mitigation at runtime would work—how do I decide which coroutines are important? They're kind of all are.

The idea is basically to have a last resort where an application crash is avoided but still providing log information for future fixes.

But yes, I had to implement it for SinkAsync and MapAsync and all their decorators, factories, and property decorators.
What do you think @flofeurstein ?

[semiversus]

You are right - it's not easy to check the length of the queue on the caller site, as it is a reactive pipeline. Approach 1 is fine for me.

[flofeurstein]

You are right - it's not easy to check the length of the queue on the caller site, as it is a reactive pipeline. Approach 1 is fine for me.

We had the same idea as you had with Approach 2. But being reactive and having the chance to check the length at the point where we add an element to the queue (instead of polling) is IMO the most important argument for Approach 1.

[phettberg]

I'll proceed with approach 1.

In your opinions - if the threshold is exceeded, should we clear the queue or just block new entries? I want to avoid a runtime error being raised several times ...