We need broadcast channels, stat.

[goodboy]

This ties in with #40 and #53 both of which heavily rely on the discussion in python-trio/trio#987.

The main gotcha is that trio's mem chans don't have broadcasting built-in (frankly making them mostly an SC anti-pattern if you have followed the .clone() semantics discussion). This mis-assumption was what originally fueled #203.

The obvious main use case is for broadcasting stream data on the receive side of a stream such that multiple rx-side tasks get each their own copy of a new message. NB: We already have a prototype for producer side ala tractor.msg.pub, though I'm not sure it's the best solution either.

Depending on how we move forward with bidirectional-streaming in #53 it might be worth offering a .clone() or similar method which provides a broadcast reference token to consumer tasks.

Definitely needs a deeper dive.

For reference here's the current short list of broadcast chan impls:

• the eventual_values predicate matching iterator in trio-util

• the "lossless", slowest consumer limiting, event queue aioevent gist from @basak

• slurry's Pipeline.tap() effectively does the,

  Apply backpressure based on the slowest consumer:
  from Broadcast channels python-trio/trio#987 (comment)

  but, with non-zero buffer size reccomendations:

  max_buffer_size: Although not recommended in general, it is possible to
  set a buffer on the output channel. (default 0) See

Any more implementations are greatly welcome from the lurker pack 😉

[goodboy]

Interesting stuff from the rust method njs linked: https://tokio-rs.github.io/tokio/doc/tokio/sync/broadcast/index.html#lagging

This broadcast channel implementation handles this case by setting a hard upper bound on the number of values the channel may retain at any given time. This upper bound is passed to the channel function as an argument.

If a value is sent when the channel is at capacity, the oldest value currently held by the channel is released. This frees up space for the new value. Any receiver that has not yet seen the released value will return RecvError::Lagged the next time recv is called.

I kind of like this since it's very reminiscent of real-time DSP style overruns and would allow for detecting too-slow consumers via errors (that may allow restart logic to entail?).