async broadcast channel for inter-task communication

[pn2new]

I am looking for some kind of async broadcast channel for communication between async co-routines.

One of the co-routines drives my board's hardware, and is supposed to send and receive events to/from a web frontend.

So far, I have realized this with the microdot web server and an async websocket for communication with the web frontend. This works as intended.

Of course, I do want to decouple the hardware driver from the websocket so that the former does continue to run when the websocket closes. So I am looking for a broadcast channel (or actually two) to connect my websocket(s) to the controlling routine.

This would be a single producer (potentially) multilple consumer channel to which the websocket tasks would subscribe, and vice versa for communication in the other direction. The channel only needs a tiny buffer that ideally overflows (does not block on sending).

Does something like that already exist in micropython (ESP32 port), or do I have to write my own code?

[andrewleech]

You could possibly use a standard asyncio.Event that all the consumers are blocked waiting for, the publisher has the single buffer that it updates then sets the event, which would wake up the consumers who'd then read the buffer.

The problem then might be how the publisher knows when all consumers have processed the data before clearing the event for them to all go back to waiting again, maybe you could have a double buffer scenario instead; two buffers & two events, only one will ever be set at a time and consumers know that when they're woken for event1 they read buffer1 then immediately await on event2 etc, ping pong between the two.

[rkompass]

Studied the source: Well balanced functionality extension compared to the original.

res = agent(topic, message)
     if isinstance(res, type_coro):
                asyncio.create_task(res)

seems nice application of ASYNCIO lore.

In the docs:

Returns True unless a Queue agent has become full, in which case data for that queue has been lost.

can be interpreted in a way that data on the Queue are lost. (As I did first).
Perhaps a formulation that the Queue is not updated and the message in this case not delivered, woud be better understandable.

[peterhinch]

Good point about the docs. I will amend the text.

I'm open to suggestions about what do do when a Queue fills. I'm trying to figure this out. Throwing seems drastic. Ideally the recipient should be notified. Options I'm considering are:

• Subclassing Queue. The subclass triggers a bound Event when a message is lost, so the recipient can be warned.
• Offer a Queue alternative. This would have the property that the oldest message is discarded (as per ringbuf_queue), also with an Event. In many applications losing the oldest message is preferable.

One benefit of a Queue alternative is that it can support the asynchronous iterator protocol, which is a simple way to retrieve items. (The mission of Queue is modelling the CPython version).

I'm astonished that the message broker functionality is no part of ASYNCIO yet.

I've been using (and writing) cooprative schedulers since, er, a long time ago and I've played with various communication schemes. Mainly implementations of Channel objects in the old activity-channel-pool model of concurrent design - never producing anything of much note.

This was a bit of a lightbulb moment, with a very minimal implementation. I'm still half expecting some nasty snag to emerge :)

[rkompass]

from @andrewleech s message:

The problem then might be how the publisher knows when all consumers have processed the data before clearing the event for them to all go back to waiting again

In this message broker setup every consumer would have its own event.
It certainly would be responsible for clearing that event once it is processed/registered.

The message broker might in turn check that the event is cleared before setting it again and if not signal that an error has occurred somehow (the previos event was not fully processed).

[peterhinch]

Bear in mind that at the instant when a publication occurs there might be from 0 to N recipients, with M <= N of these agent instances being Queue objects. Further, M, the number of subscribed Queue instances, can change dynamically. The publishing task does not know the values of M or N. My current thinking is that having .publish return a value is pointless: there is no practical way for the publishing task to use this information. Publication is essentially shouting into the void, an open loop operation. (This is also the case for MQTT).

If a Queue overflows the cause is in the one receiving task whose agent is that Queue instance. That is the only task that can detect the event or meaningfully respond.

I will examine allowing a RingbufQueue as an agent which has the "discard oldest message on overflow" model. It might be worth adding an Event to RingbufQueue to aid the receiving task in dealing with overflows.

[rkompass]

It might be worth adding an Event to RingbufQueue to aid the receiving task in dealing with overflows.

That was my thinking from the beginning after studying the code: A Queue without an event is not complete because the receiver does not notice immediatly when the Queue is updated. OTOH if it does not clear the event that may be an indication that the Queue is not updated/emptied.
But this assumes that each published message is to be processed immediately. You probably intend to have a way of storing a number of events (interrupts??) allowing for a receiver to react more lazily.

From the beginning we notice that the Queue is overall problematic. If a recipient wants to have a Queue he may implement it internally and update it in the receiving routine.

[beetlegigg]

@peterhinch
Forgive my blunder into this topic as a hobby programmer who only took up programming in my retirement, but I have made, to my mind at least, great use of your examples of async, mqtt, and the async events.

When an async function that will be called on demand within in a program, one thing I'm not so clear on is whether to create the function as a task on program startup, using the async events set/clear flags initiate the function to run, and then to pause execution until its required again, or to just create the task anew each time the function is required to be executed. I suspect that theres no definitive 'rule' on which to base this and its probably of no consequence either way (I state this in blissful ignorance :) )

My somewhat limited understanding of this new message broker idea is that tasks or indeed other functions can be registered to a broker class which will then assign a topic id and set an async event, or create a new task, etc, when some other bit of the program publishes to that topic. The publisher can also send data when notifying the broker to set the required task (or whatever) in motion. I'm not too clear as yet on how that data gets translated into the aysnc functions parameters, but I think this will be clear if a few working examples are provided.

Again, I not too clear on when one's program should choose to use the broker mechanism or just use the existing program techniques, but it does seem to be a nice way of handling the situation when there are a number of data collectors that need to fire off various tasks when the data is received.

I look forward to having a go with the broker idea once you've had a chance to develop it, so I send this message more in support and interest about this development, though sadly not as one who can provide any sensible comments at the level required. But thanks for providing a very interesting development in the use of async programming.

[peterhinch]

Both techniques are valid. In general it's simplest to create a task whenever it's needed and let it run to completion. The exception is where the task needs to retain state: then it's fine to have it wait on an Event and resume execution when the event occurs.

Re Broker if you're a beginner it may be best to wait a little while. I'm still developing this idea and the documentation is limited. Dive in when the "under development" note disappears. My starting point would be to use standard techniques. The Broker technique comes into its own when there is a problem with inter-task communication: it's a better solution that having a host of shared global objects. I think it may have wider application (see the MQTT example), but I'm new to this too so I can't yet advise :)

[peterhinch]

The Broker primitive has been updated. See the docs and the code. It has the following features:

• Improved validation and exception handling.
• A Broker.Verbose flag enables printing of non-fatal states such as queue overrun.
• RingbufQueue is now supported and recommended.
• Agent instances may be passed arguments.
• An Agent instance may subscribe to multiple topics (was always the case) but...
• An Agent instance may subscribe to a single topic multiple times.
• User-defined Agent classes are supported.
• Docs are improved with more examples.

Code has inevitably grown but at 59LOC it is still quite lightweight. Comments/code review welcome.

@rkompass Re our discussion of queues - I think queue support is essential to the Broker concept.

The Queue class is primarily designed for CPython compatibility hence its API is set in stone. RingbufQueue was designed as a MicroPython optimised alternative. It supports retrieval via an asynchronous iterator which is similar in concept to setting an Event when a data item arrives (see the example in the Broker docs).

We also discussed setting an Event when an overrun occurs. I think this option is best achieved by a user Agent based on RingbufQueue (via composition or inheritance). Most microcontroller applications will be designed so that overruns can never occur - a user Agent might be used as a temporary debugging tool. However there are applications where overruns with discard of oldest data are expected and no notification is required. This is default behaviour.

[EDIT] Example of a queue which sets an Event on overflow:

class MyQueue(Agent, RingbufQueue):
    def __init__(self, size):
        RingbufQueue.__init__(self, size)
        self.oflow = asyncio.Event()

    def put(self, topic, message):
        try:
            self.put_nowait((topic, message))
        except:
            self.oflow.set()