`tokio`-style broadcast channels #229

goodboy · 2021-08-08T21:29:00Z

~~Literally the most prototype thing ever~~ actually decently well tested now, but based on discussion in:

Broadcast channels python-trio/trio#987 particularly the comment on what rust has
We need broadcast channels, stat. #204

Nothing more then a runnable module atm..

Some interesting notes:

to consumers i'm pretty sure this looks exactly like a ring buffer
we should be able to swap in any queue type that supports writing to it (like deque.appendleft()) where the oldest value is just discarded, for some examples:
- ringbuf's handling of overflow is much the same
- a new rust based ringbuf impl

Update:

before i lose this thought...

If we wanted to also support slowest consumer style, I'm pretty sure there's an easy hack to just inherit and proxy through reads to trio._channel.MemoryChannelState.data (presuming the queue size to broadcast_receiver() and the underlying mem chan's deque are the same size).

The thinking is, there's a length check to determine if the sender should block in MemorySendChannel.send_nowait():
https://github.com/python-trio/trio/blob/master/trio/_channel.py#L147

If .data here proxied to lambda: max(self._subs.values()) and the order of these branches were reversed, it might just work no?

        elif len(self._state.data) >= self._state.max_buffer_size:
            # causes ``.send()`` to block
            raise trio.WouldBlock
        else:
            self._state.data.append(value)

Follow up:

dropped all the clones stuff, bleh; a big source of errors and confusion imo
probably should just use id(self) for .key instead of uuid4()?
probably still need a couple tests for trying to (re)open a subscription from a now closed caster
need a test for the case where ._recv() is cancelled before the first ask in receives a value and that ensure that no other event queued tasks miss that value and/or re-get their last pulled value
make an issue detailing the slowest-consumer style notes from above. -> Slowest consumer broadcasting? #235

tractor/_live_from_tokio.py

goodboy · 2021-08-08T21:33:19Z

tractor/_live_from_tokio.py

+            # their latest available value.
+            for sub_key, seq in self._subs.items():
+
+                if key == sub_key:


maybe do a ._subs copy and pop instead ?

tractor/_live_from_tokio.py

goodboy · 2021-08-08T23:59:07Z

There. I think this is ready to start adding tests.

I'll leave it for a couple hours to see if any of you lurkers complain 😂

tractor/_broadcast.py

goodboy · 2021-08-09T20:51:37Z

tractor/_broadcast.py

+            # ``AsyncResource`` api.
+            await clone.aclose()
+
+    # TODO:


Last set of design questions i think?

tractor/_broadcast.py

goodboy · 2021-08-09T20:56:13Z

tractor/_broadcast.py

+            # don't decerement the sequence # for this task since we
+            # already retreived the last value
+            subs.pop(key)
+            for sub_key, seq in subs.items():


is there some fancier-faster way to do something like this?

goodboy · 2021-08-09T20:56:58Z

tractor/_broadcast.py

+        else:
+            await self._value_received.wait()
+
+            seq = self._subs[key]


can we drop all of these sequence # checks?

goodboy · 2021-08-10T18:06:02Z

lol, yeah so since moving to the "clones" version the actual N copies consistency was broken obviously because the ._value_received event wasn't common to all clones 😂

Fixed this and now am adding a "broadcast state" type much like default mem chans.

goodboy · 2021-08-10T23:20:37Z

Yah so binance feeds proved that the assumption of slept consumers waking up can expect to read the 0 index is definitely false 😂

We can maybe try this down the road if we invert the indexing?
I gotta think about it.

goodboy · 2021-08-16T15:39:51Z

tractor/_streaming.py

+        if self._broadcaster is None:
+            self._broadcaster = broadcast_receiver(
+                self,
+                self._rx_chan._state.max_buffer_size,


This presumes a trio.MemoryReceiveChannel which is not ideal, so what would be the best default here and how would we reflect it in typing?

goodboy · 2021-08-16T15:44:27Z

tractor/_streaming.py

+            # of the the ``BroadcastReceiver`` before calling
+            # this method for the first time.
+
+            # XXX: why does this work without a recursion issue?!


This was broken actually, but is now fixed by adding support for passing into BroadcastReceiver a recv_afunc kwarg that is by default the underlying's .receive() if not provided.

This makes it easy to wrap receive channel apis for use in task broadcasting without expecting any initial consumer-task of the underlying to also use a broadcast receiver instance when calling .receive(), it however does mutate the machinery to put the broadcaster behind the scenes as it were.

goodboy · 2021-08-19T17:45:14Z

Oh right, to use typing.Protocol we need 3.8+. I wonder is this incentive enough to drop 3.7 for next alpha?

goodboy · 2021-08-19T18:56:28Z

tests/test_local_task_broadcast.py

+            retries = 3
+            size = 100
+            tx, rx = trio.open_memory_channel(size)
+            brx = broadcast_receiver(rx, size)


I'm almost tempted to enforce this as an async context manager, something I kinda wish open_memory_channel() did as well.

goodboy · 2021-08-19T18:56:52Z

tests/test_local_task_broadcast.py

+
+                async with brx.subscribe() as lbrx:
+                    while True:
+                        #     await tractor.breakpoint()


For every set of broadcast receivers which pull from the same producer, we need a singleton state for all of, - subscriptions - the sender ready event - the queue Add a `BroadcastState` dataclass for this and pass it to all subscriptions. This makes the design much more like the built-in memory channels which do something very similar with `MemoryChannelState`. Use a `filter()` on the subs list in the sequence update step, plus some other commented approaches we can try for speed.

Add `ReceiveMsgStream.subscribe()` which allows allocating a broadcast receiver around the stream for use by multiple actor-local consumer tasks. Entering this context manager idempotently mutates the stream's receive machinery which for now can not be undone. Move `.clone()` to the receive stream type. Resolves #204

This allows for wrapping an existing stream by re-assigning its receive method to the allocated broadcaster's `.receive()` so as to avoid expecting any original consumer(s) of the stream to now know about the broadcaster; this instead mutates the stream to delegate to the new receive call behind the scenes any time `.subscribe()` is called. Add a `typing.Protocol` for so called "cloneable channels" until we decide/figure out a better keying system for each subscription and mask all undesired typing failures.

Get rid of all the (requirements for) clones of the underlying receivable. We can just use a uuid generated key for each instance (thinking now this can probably just be `id(self)`). I'm fully convinced now that channel cloning is only a source of confusion and anti-patterns when we already have nurseries to define resource lifetimes. There is no benefit in particular when you allocate subscriptions using a context manager (not sure why `trio.open_memory_channel()` doesn't enforce this). Further refinements: - add a `._closed` state that will error the receiver on reuse - drop module script section; it's been moved to a real test - call the "receiver" duck-type stub a new name

…ibe()` api

Add a couple more tests to check that a parent and sub-task stream can be lagged and recovered (depending on who's slower). Factor some of the test machinery into a new ctx mngr to make it all happen.

The `collections.deque` takes care of array length truncation of values for us implicitly but in the future we'll likely want this value exposed to alternate array implementations. This patch is to provide for that as well as make `mypy` happy since the `dequeu.maxlen` can also be `None`.

goodboy requested review from ryanhiebert, guilledk, chrizzFTD and salotz August 8, 2021 21:29