Broadcast channels

[njsmith]

[This is a speculative discussion. I don't know if we should do anything here. But I was thinking about it so I figured I'd write it down so it doesn't get lost.]

There are two kinds of multi-consumer channels: the kind where each item is distributed to one receiver (e.g., by round-robin), and the kind where each item is distributed to all receivers ("broadcast"). We already have good support for the former, via memory channels. Implementing broadcast is trickier, though, because you need to pick some policy for handling the case where consumers run at different speeds. I guess the basic options are:

• Apply backpressure based on the slowest consumer: this is appropriate if all the receivers are working together in some sense, and there's no point in letting one consumer get too far ahead of the others. It's straightforward to implement by using a collection of memory channels with bounded buffers, like:

  async def broadcast_send(self, value):
      for channel in self._channels:
          await channel.send(value)

• Apply backpressure based on the faster consumer: this is intuitively tempting in cases where the receivers are unrelated, but it requires unbounded buffer space, so... not necessarily a good idea. It can be approximated by re-using our broadcast_send from the previous bullet point, but with infinite buffers. But, this disables backpressure entirely, rather than applying backpressure based on the fastest receiver. Doing this precisely would require some more work. I'm not sure if it's ever useful.

• Instead of applying backpressure, drop messages: The intuition here is that one common case is where you have a value that updates occasionally, and subscribers want to keep track of the latest version. However, if multiple updates happen in quick succession while a subscriber isn't looking, then they only care about the latest value. One possible API:

  class BroadcastValue(Generic[T]):
      def set(self, value: T) -> None:
          ...
  
      def subscribe(self) -> ReceiveChannel[T]:
          ...

  The semantics of the ReceiveChannel would be: the first time you call receive (or __anext__), it immediately returns the current value, and stores some state in the channel object keeping track of what was the last value returned by this particular channel. After that, each call to receive waits until there's been at least one call to set (compared to the last time this channel returned a value), and then returns the latest value. So in practice, you'd write:

  async for current_value in broadcast_value.subscribe():
      print("Latest update:", current_value)

  and if there's a fast update rate, then the loop might skip over some intermediate values, but if the updates stop then the loop will always catch up with the latest value.

Are there any other interesting policies that I'm missing?

There's some discussion of related ideas in #637

[smurfix]

A channel that drops the oldest enqueued message when a new one arrives (instead of blocking) would be a good building block to have.

The backpressure-on-the-fastest-receiver idea is interesting. But why would you need unbounded buffer space?

• try to send_nowait to all receivers. If at least one accepts the message, you're done.
• otherwise, add yourself to all receivers' parking lots and await being woken up (or cancelled of course).
• Repeat (if not cancelled).

[oremanj]

BroadcastValue is an excellent interface IMO. It could be built pretty easily on top of the "channel that drops the oldest enqueued message when a new one arrives" primitive described by @smurfix.

For the reliable delivery cases, one possible issue with "just send in an arbitrary order" is that it means consumers later in the delivery order have to wait for consumers earlier in it. Theoretically it seems like it might be better to do all the sends in parallel? Either using a nursery, or if that's too high overhead, using some select() ish thing. (Which I will get back to soon... dayjob work has been particularly high-intensity lately and it's not dovetailing as well with Trio work as it did a few months ago.)

[njsmith]

A channel that drops the oldest enqueued message when a new one arrives (instead of blocking) would be a good building block to have.

I think it's worth distinguishing between two use cases:

• There's a pattern for low-level APIs where you use a bounded capacity channel, and if it overflows then you drop the contents and enqueue a special message saying that some messages were lost. (Examples: inotify, posix real-time signals.) This is useful when you actually need the whole stream of messages, there's no way to apply backpressure, you have strictly bounded resources, and there is some (slow) way to recover after an overflow. It would be easy to implement this in Trio. But it's a tremendous pain to use these APIs correctly. You don't want to do it if you have any other choice. So it sounds simple but it's really a low-level experts-only kind of thing.

• The broadcast value use case described above. In some ways this seems very similar to the bounded-queue with a particular drop policy, but it's much simpler to use and reason about, because it's designed for cases where it's totally fine to lose old messages. And if it's totally fine to lose old messages, then you might as well do it aggressively, and hard-code the buffer size to 1, and skip the overflow notification. Technically you could implement it on top of a discard-on-overflow channel, but it's simpler to implement directly.

I don't think we should mix these up. In particular, I'm skeptical that it ever makes sense to use a drop-oldest policy + buffer size >1... If you can afford to drop old messages, you might as well do it all the time, and exercise that case properly.

The backpressure-on-the-fastest-receiver idea is interesting. But why would you need unbounded buffer space?

If the slowest receiver is N messages behind the fastest receiver, then you need to buffer N messages. If you don't have any way for the slowest receiver to transmit backpressure, then N can grow without bound.

[njsmith]

For the reliable delivery cases, one possible issue with "just send in an arbitrary order" is that it means consumers later in the delivery order have to wait for consumers earlier in it. Theoretically it seems like it might be better to do all the sends in parallel?

Yeah, I'm not sure it makes a huge difference – it's more-or-less equivalent to changing the buffer size by 1 – but doing the sends in parallel does feel more elegant.

[smurfix]

The backpressure-on-the-fastest-receiver idea is interesting. But why would you need unbounded buffer space?

If the slowest receiver is N messages behind the fastest receiver, then you need to buffer N messages. If you don't have any way for the slowest receiver to transmit backpressure, then N can grow without bound.

Well, in that case, sure. I was thinking of the "we have more than one receiver and at least one needs to process the message while others can take it or leave it" case. I actually may need that for a distributed network test system: at least one receiver must run each test, while others may also do so if they have spare capacity.

[belm0]

The semantics of BroadcastValue fits several instances in my existing codebase (notably most of the uses of Event.clear())-- but how about making the value optional? Often we just want to pass the value out of band.

With the value optional, the question becomes what to name it. BroadcastSignal()? No one wants to overload "signal"-- but that's what it is.

[njsmith]

@belm0 I guess in general we want to encourage people to pass things in-band, because it makes the state tracking easier to reason about ("share memory by communicating" does have some merit). But if you wanted to reuse the machinery without any real state, I guess you could just increment a counter or something?

[belm0]

Regarding BroadcastValue again-- I believe that direction has the most utility. However I'm thinking of merging this idea with the ValueEvent we have in our project. ValueEvent supports predicates for value transitions so the listener can choose what to be awoken with. The predicate feature is used a lot. Perhaps ValueEvent could be built on top of BroadcastValue, though I'm not sure because ValueEvent does other things (e.g. level triggering).

Roughly:

class BroadcastValue(Generic[T]):
    def set(self, value: T) -> None:
        ...

    def subscribe(self, predicate=None) -> ReceiveChannel[T]:
        """Yield value changes when predicate(value, old_value) is True.  Defaults to all changes."""
        ...

(In 90% of cases our predicates don't use old_value, and comparing new and old is somewhat dubious given that values may be dropped.)

[smurfix]

I use a version of ValueEvent that's specifically not about broadcasting, but about passing exactly one value from one task to exactly one other task. Typical uses are passing a task's value back to the caller, or managing a server's async responses.

The crux of these ideas is that they are not cancellation safe, i.e. when that task is cancelled or otherwise exits before setting the value, how does the caller get notified? Conversely, when the caller dies, somebody please notify the callee that its work is no longer required.

[njsmith]

Something that just occurred to me, though maybe it was obvious to everyone else: if broadcasting is about handling arbitrary numbers of consumers, then an important special case is when there are zero consumers. For BroadcastValue this case is trivial, but for the other approaches it's a bit less clear.

And the other thing any broadcast strategy has to think about is: can the set of consumers change over time, and if so then how do you handle consumers arriving and consumers departing? For BroadcastValue, again, this is obvious. For the backpressure-based strategies, it's not as clear.

There's a pattern for low-level APIs where you use a bounded capacity channel, and if it overflows then you drop the contents and enqueue a special message saying that some messages were lost. [...] But it's a tremendous pain to use these APIs correctly. You don't want to do it if you have any other choice.

I guess there might be one high-level case where this style could be useful, and that's logging. Maybe if your log consumer can't keep up, you want to just discard messages, but add a note saying that that's what you're doing.

But I feel like this still has a lot of subtleties that I don't fully understand.

If you're logging over something like UDP, then this isn't an issue – the transport layer automatically handles message boundaries and discarding messages on overflow. So I guess this is mostly useful for when you're logging over a bytestream, like TCP or stdout. If you're logging over a bytestream, then you need a task to handle the bytestream and mediate access, because otherwise you'll get garbled messages if a task gets cancelled while it's in the middle of logging. So you want something like an in-memory channel, with multiple producers who probably only want to call synchronous methods, and a single consumer. And the channel needs a buffer, plus a flag to indicate overflow, and whenever we consume from the channel we need to do some delicate handling to make sure the "messages dropped" indication gets recorded at the right point in the message stream.

That last part makes things slightly awkward, but I guess we can do it with a object that encapsulates a MemoryChannel plus a flag, where logging is:

• try to send_nowait
• if that gets WouldBlock, set the flag instead

And the consumer does:

• block waiting for a message
• once one message arrives, immediately use receive_nowait to pull out all the messages
• and then pull out the flag, inserting it into the message stream immediately after all the messages we just pulled out

... But this has a bug. Consider this sequence:

• the channel fills up
• the overflow flag gets set
• the consumer calls blocking receive, which of course wakes up immediately, but still executes a checkpoint
• while the consumer is waiting to be rescheduled, another message gets logged into the open slot that was created by receive
• the consumer is rescheduled and follows the algorithm above

In this case the final logs look like:

[message] [message] [new message] [overflow indication]

But they should look like:

[message] [message] [overflow indication] [new message]

So this is interesting. I'm not sure how important this is and there are certainly lots of ways to solve it (e.g. it wouldn't be too hard to create a custom LogChannel from scratch instead of trying to compose it out of existing pieces). But I'm always intrigued when we find a case that our abstractions don't make easy, because maybe it's a clue pointing to better abstractions.

Some ideas for fixing this situation:

• I think it might be enough to modify the consumer so that before it calls receive, it first tries receive_nowait? This is super subtle though, if it works at all.

• we could add some way to "force" a write to a memory channel even when its buffer is full, and then use this to write a "overflow" message instead of setting a flag. But we'd also need some way to peek at the end of the existing buffer, so if there's already an overflow message in there we skip adding a new one.

... None is this is seeming super appealing. Maybe a custom LogChannel is the best option.

[smurfix]

Simple solution to the message flag problem: teach the sender to attach a sequence number to each message.

I'd rather not over-engineer this. A broadcast on UDP is a lossy transaction that doesn't have any loss indication and might even suffer from message reordering. If we add a broadcast channel that doesn't have these two faults it's a strict improvement.

Beyond that we have some incompatible design requirements (e.g. does a lossy receiver still receive the messages before the overflow occurred, given that it needs to re-sync anyway?) which remind me of the multiple choices we have for doing a Future/EventWithValue object, which is why we still don't have the latter either. :-/

[njsmith]

Simple solution to the message flag problem: teach the sender to attach a sequence number to each message.

OK yeah that's probably better than all that stuff I wrote up above :-)

---

Returning to BroadcastValue: I've seen a few cases recently where I wanted something like this, but without actually transmitting any value. (E.g. in design sketches for snekomatic, where I want to notify background tasks "hey I just changed the database, you should check it again").

I'm thinking, the basic structure of BroadcastValue, where it internally keeps a counter, and each consumer keeps its own snapshot of the counter, to prevent lost updates – that's a basic primitive, that covers a lot of the use cases for Condition but in a less error-prone way. I think it also covers a lot of the "blinker" use cases where people often reach for Event.clear and end up with bugs due to lost updates, that we discussed in #637. Adding an associated value that gets sent along with each update is a nice add-on, not definitely not necessary. And speaking of counters, maybe we should expose the counter, in case people want to detect lost updates?

So I'm not quite sure how we should organize all these ideas into, like, Python classes.

One option would be to pack it all into one class, that you can use like:

# Simple value broadcasting:
broadcast_value.set(...)

async for generation, value in broadcast_value:
    if generation != last_generation + 1:
        print("missed some updates")
    print(value)

# Representing an outside value, like a database:
broadcast_value.notify()  # bumps generation counter without changing value

async for _, _ in broadcast_value:
    print("database was updated")

This might be trying to jam too many ideas into a single class though, and end up making it more confusing than if we had multiple classes? And I'm not really sure the generation counter is useful. Really the main reason I put it in the sketch is that notify only makes sense if you understand that there's a generation counter involved, and that's easier to explain if the counter is an explicit value that users can see and experiment with, compared to if it's a piece of hidden implicit state.

I also wonder about whether we should expose a way to inspect the current value and/or generation counter directly, without subscribing. Is that helpful, or an invitation to subtle race conditions due to the values being out of sync with subscribers?