Asyncio one-to-many "queue"

[iot49]

In an asyncio project I have data producer (e.g. sensor) and a varying number of consumers of its values. I would like to ensure that each consumer gets every value. The processor is sufficiently fast to keep up with the data.

My present solution relies on asyncio.Event that the consumers wait on. Whenever a new value is available, the producer calls set (immediately followed by clear) on the event, putting all consumers on the ready queue. Immediately after calling clear the producer issues await asyncio.sleep_ms(0) to allow the consumer tasks to run.

The problem is that when values arrive in fast bursts consumers sometimes "skip" values. How can I avoid this - i.e. implement some kind of "back pressure". Changing the delay inawait asyncio.sleep_ms(0) to 100ms is a "solution", but of course the value 100ms is a bit arbitrary and presumably depends on the number of consumers.

I do not understand why the problem occurs in the first place - my assumption is that await asyncio.sleep_ms(0) pauses the running thread (the producer) at which point the event loop processes all ready tasks (i.e. all consumers) before resuming the producer task. Apparently this is not happening - why?

[peterhinch]

If a consumer is not waiting on the event at the time when the producer sets it, the consumer will miss the event. To synchronise this, the producer should set the event but not clear it. This ensures that consumers that are running late will work: when they await the event they will immediately run, because the event is already set.

This only solves one part of the problem. You need to figure out how to clear the event, and how to ensure that the producer doesn't set the event again before it has been cleared (otherwise data will be lost).

One way is to use gather() to clear the event when all consumers have terminated. This could set an "acknowledege" event that the producer waits on.

There is no single best way to do this - it depends on the application design. See the tutorial for the various synchronisation primitives.

It's important to understand what

asyncio.sleep(0)

actually does. It simply allows other tasks to get a slice of execution. It cannot be expected to synchronise tasks.

[iot49]

What I feared - there is no trivial solution.

Rust (embassy project) has a queue that keeps track of all subscribers (clients), ensuring that each of them reads all values. I may need to implement something similar. Perhaps explicit "subscription" to the event with a counter that keeps track of the number of reads and gates set. Need to think about it.

Gather won't work in my case as the tasks run for a long time, essentially executing a loop waiting for values.

Thanks!