Bidir streaming #209

goodboy · 2021-05-02T18:23:43Z

First draft bidirectional streaming support as discussed in #53.

Further todos:

more extensive tests for closing down either side of the stream early
port _debug.py remote tty locking to context api.
possibly a better internal design for tracking bidir vs unidir context usage to avoid hacky if not self._ctx._portal checking inside ReceiveMsgStream.aclose()
tests where the consumer tasks use async for stream while sender is running independently in another trio task
docs on the new apis
should we add broadcasting support here for We need broadcast channels, stat. #204 ?
move to a split SendMsgStream / ReceiveMsgStream type set and staple them together using a channel/messaging equivalent of trio.StapledStream?
- currently this would allow making the stream arg to @stream funcs a SendMsgStream instead of a Context
- turns out anyio has something similar: StapledObjectStream (obvs we won't use *Object* since we're msgpack type contrained). Actual code is here.
example in the main readme. i'm thinking that should be a big boon when compared to projects like faust and ray which (i think) are unidirectional only?
from Ems to bidir streaming pikers/piker#190

would be nice if in tractor we can require either a ctx arg, or a named arg with ctx in it and a type annotation of tractor.Context instead of strictly requiring a ctx arg.

Critique and suggestions very welcome from lurkers 😎.

ryanhiebert · 2021-05-02T19:39:54Z

tests/test_2way.py

+
+# possible outcomes:
+# - normal termination: far end returns
+# - premature close: far end relays a stop message to tear down stream


Near end sends a message, not far end, right?

It should eventually be one of either, we'll need tests for both cases.

You should be able to close either side's Context.open_stream() and the other side shuts down gracefully i'm thinking?

That makes sense to me. I figured you were laying out how it might happen from either side

goodboy · 2021-05-12T03:52:57Z

tractor/_streaming.py

+                # NOTE: we're telling the far end actor to cancel a task
+                # corresponding to *this actor*. The far end local channel
+                # instance is passed to `Actor._cancel_task()` implicitly.
+                await self._portal.run_from_ns('self', '_cancel_task', cid=cid)


one outstanding question i still have is whether instead of calling this special cancel method on the "actor runtime" we should have a special cancel message. I mention this same question in #36.

The only other internal change required for the message would basically replace hacky code in the message loop anyway. It would also get us closer to a protocol that isn't tractor specific or something.

goodboy · 2021-05-12T03:55:28Z

tractor/_streaming.py

+            try:
+                yield rchan
+
+            except trio.EndOfChannel:


So if we receive a stop msg then we don't send one since we can assume the far end has already closed.

Should probably add a comment about this.

goodboy · 2021-05-12T03:56:03Z

tractor/_streaming.py

+            except trio.EndOfChannel:
+                raise
+
+            else:


If we're closing just send the stop msg and move on right?

goodboy · 2021-05-12T11:53:37Z

tests/test_advanced_streaming.py

+import tractor
+
+
+_registry: Dict[str, Set[tractor.ReceiveMsgStream]] = {


Linking comment f9dd2ad#r50700925.

@gc-ss mentioned,

… but if you don't like that, what if we make _registry an argument to publisher?

Yup can do that, I just didn't bc turns out the consumers can update via sending updates on their individual streams.

It would be perfect if publisher accepted an environment variable (or a Context?)

This can also be done though I'm not sure what you mean by environment. A Context isn't really required explicitly here but could be used to control cancellation if wanted.

For an alt to _registry we could do something like,

class PublisherState: subscriptions: dict[str, tractor.MsgStream] = {}

kinda thing?

I'm not sure what you mean by environment. A Context isn't really required explicitly here but could be used to control cancellation if wanted

By environment or Context, I meant the context/environment/configuration due to which the function might behave in a non-deterministic manner but the values of which don't necessarily change between function invocations - like a remote host address:port or username/password.

In this case - think about two actors communicating with each other. The exact same actors (types) might behave in very different ways if their context/environment/configuration were different with everything else (eg: arguments) being the same.

Unlike arguments though, their context/environment/configuration dont change between function invocations (like arguments do).

For example, the registry does not need to change between function invocations.

I mean, yeah a service actor can easily create wtv objects they need and mutate them as actor-local state.
They can go further and offer a seperate api to mutate that object/state from other actors if needed.

I'm just not sure we need to show that in this example since the point is just a super basic dynamic pubsub system.
We do have a pretty terrible example in the docs that could be improved.

Bonus points for sweet PRs 😉

goodboy · 2021-05-12T12:08:43Z

tests/test_advanced_streaming.py

+                )
+
+            # make one dynamic subscriber
+            await n.run_in_actor(


Hmm, I'm thinking we could just move the dynamic case into the immediate task here instead of sleeping?

tractor/_streaming.py

tests/test_advanced_streaming.py

gc-ss · 2021-05-12T22:47:53Z

tractor/_actor.py


        try:
            log.debug(f"Delivering {msg} from {actorid} to caller {cid}")
            # maintain backpressure
            await send_chan.send(msg)

        except trio.BrokenResourceError:
+            # TODO: what is the right way to handle the case where the


This is a very difficult decision to make. It would have been less complex with old school unidirectional streams.

We might look into using both NACKs and ACKs. This coupled with atleast once delivery can ensure durable message processing

This coupled with atleast once delivery can ensure durable message processing

Not sure what you mean.
TCP already has reliability built in, for other protocols this might be some for a user to consider yes.

Also, the messaging processing isn't the problem here, it's whether a stop/cancel message made it to the far end and what to do in specifically the case where no ack for that msg is received.
That's just two general's.

I'm pretty sure (at least right now) trio takes the same naive approach of just "hoping for the best".

For interested lurkers:

For example, the first general could send 100 messengers, anticipating that the probability of all being captured is low. With this approach the first general will attack no matter what, and the second general will attack if any message is received. Alternatively the first general could send a stream of messages and the second general could send acknowledgments to each, with each general feeling more comfortable with every message received. As seen in the proof, however, neither can be certain that the attack will be coordinated.

TCP obvs already has built-in mechanisms for it's reliability requirements so currently we don't have to think too hard since failures should bubble up from the transport layer.

The main question was moreso about cancellation race conditions that can arise where the local channel is killed after it's sent the stop and whether or not we should wait / shield the mem chan until the msg is processed later (also presumably this is all before the sub-actor is killed).

Also to clarify, tractor's streams can in theory be restarted but i'm pretty sure internal feeder mem chans should be re-created?

goodboy · 2021-05-13T12:52:32Z

tractor/_actor.py

-            log.debug(f"{send_chan} was terminated at remote end")
-            # indicate to consumer that far end has stopped
-            return await send_chan.aclose()
+        # if 'stop' in msg:


yeah i think we can drop this now.
processing is now inside the ReceiveMsgStream.receive() body.

goodboy · 2021-05-13T12:53:53Z

tractor/_debug.py

@@ -38,7 +37,9 @@
 _in_debug = False

 # lock in root actor preventing multi-access to local tty
-_debug_lock = trio.StrictFIFOLock()
+_debug_lock: trio.StrictFIFOLock = trio.StrictFIFOLock()


Might pull this debugger stuff into a new PR to keep things more separate?

The main thing was being able to use the context api for tty locking which is a lot cleaner now 🏄🏼

Nah, keeping it here for now; it's all internals anyway.

goodboy · 2021-06-05T18:31:48Z

tests/test_advanced_streaming.py

+
+    global _registry
+
+    # syn caller


s/syn/sync

goodboy · 2021-06-14T00:29:04Z

There's still quite a few things on the todo list but I think this is good enough to get peeps using on master and we can always follow up.

Add clear teardown semantics for `Context` such that the remote side cancellation propagation happens only on error or if client code explicitly requests it (either by exit flag to `Portal.open_context()` or by manually calling `Context.cancel()`). Add `Context.result()` to wait on and capture the final result from a remote context function; any lingering msg sequence will be consumed/discarded. Changes in order to make this possible: - pass the runtime msg loop's feeder receive channel in to the context on the calling (portal opening) side such that a final 'return' msg can be waited upon using `Context.result()` which delivers the final return value from the callee side `@tractor.context` async function. - always await a final result from the target context function in `Portal.open_context()`'s `__aexit__()` if the context has not been (requested to be) cancelled by client code on block exit. - add an internal `Context._cancel_called` for context "cancel requested" tracking (much like `trio`'s cancel scope). - allow flagging a stream as terminated using an internal `._eoc` flag which will mark the stream as stopped for iteration. - drop `StopAsyncIteration` catching in `.receive()`; it does nothing.

Revert this change since it really is poking at internals and doesn't make a lot of sense. If the context is going to be cancelled then the msg loop will tear down the feed memory channel when ready, we don't need to be clobbering it and confusing the runtime machinery lol.

Another face palm that was causing serious issues for code that is using the `.shielded` feature.. Add a bunch more detailed comments for all this subtlety and hopefully get it right once and for all. Also aggregated the `trio` errors that should trigger closure inside `.aclose()`, hopefully that's right too.

…n logging

goodboy · 2021-07-06T11:43:37Z

Ok so I've put up #219 as a replacement for this which pulls out all debugger work that relies on this change set.
This will hopefully make the patch simpler to grok and also more easy to resolve wtv strange CI stuff is going on.

I will be moving the todo list to that PR as well.

goodboy · 2021-07-06T12:29:43Z

Replaced by #219.

goodboy requested review from ryanhiebert, guilledk, salotz and chrizzFTD May 2, 2021 18:23

ryanhiebert reviewed May 2, 2021

View reviewed changes

goodboy force-pushed the drop_run branch from 67e0830 to 85be3a4 Compare May 6, 2021 16:06

goodboy force-pushed the bi_streaming branch from 5177c81 to 697195b Compare May 6, 2021 16:07

goodboy force-pushed the drop_run branch from 85be3a4 to 73e123b Compare May 7, 2021 15:22

goodboy force-pushed the bi_streaming branch from 697195b to 050717b Compare May 7, 2021 15:22

Base automatically changed from drop_run to master May 7, 2021 16:02

goodboy commented May 12, 2021

View reviewed changes

tractor/_streaming.py

except trio.EndOfChannel:

raise

else:

Copy link

Owner Author

goodboy May 12, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we're closing just send the stop msg and move on right?

goodboy commented May 12, 2021

View reviewed changes

tractor/_streaming.py Outdated Show resolved Hide resolved

gc-ss reviewed May 12, 2021

View reviewed changes

tests/test_advanced_streaming.py Show resolved Hide resolved

gc-ss reviewed May 12, 2021

View reviewed changes

This was referenced May 12, 2021

Spec out the async-ipc protocol #36

Open

Task sychronization func decorator #54

Open

goodboy commented May 13, 2021

View reviewed changes

goodboy mentioned this pull request May 17, 2021

Actor state via messages #190

Draft

goodboy force-pushed the bi_streaming branch from fc3b7a2 to 4a16f03 Compare May 31, 2021 13:26

goodboy mentioned this pull request Jun 1, 2021

Ems to bidir streaming pikers/piker#190

Merged

5 tasks

goodboy commented Jun 5, 2021

View reviewed changes

tests/test_advanced_streaming.py

global _registry

# syn caller

Copy link

Owner Author

goodboy Jun 5, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

s/syn/sync

goodboy mentioned this pull request Jun 9, 2021

Binance orders pikers/piker#194

Open

goodboy marked this pull request as ready for review June 14, 2021 00:28

goodboy requested a review from ryanhiebert June 14, 2021 00:29

goodboy added 20 commits July 5, 2021 08:44

Specially raise a ContextCancelled for a task-context rpc

f8e2d40

Fix exception typing

0083145

Wait for debugger lock task context termination

83c4b93

Adjustments for non-frozen context dataclass change

201392a

Drop trailing comma

87f1af0

Add detailed @tractor.context cancellation/termination tests

f2b1ef3

Speedup the dynamic pubsub test

43ce533

Modernize streaming tests

197d291

Consider relaying context error via raised-in-scope-nursery task

17dc6aa

Add some brief todo notes on idea of shielded breakpoint

ced5d42

Add temp warning msg for context cancel call

627f107

First try: pack cancelled tracebacks and ship to caller

17fca76

Always shield cancel the caller on cancel-causing-errors, add teardow…

6f22ee8

…n logging

De-densify some code

6e75913

Add pre-stream open error conditions

377b8c1

Expect context cancelled when we cancel

8371621

Avoid mutate during interate error

e1533d3

goodboy force-pushed the bi_streaming branch from dca0378 to e1533d3 Compare July 5, 2021 13:23

This was referenced Jul 5, 2021

Debugger hardening #217

Closed

Always hard kill sub-procs on teardown #218

Closed

goodboy changed the base branch from master to transport_cleaning July 5, 2021 16:33

goodboy mentioned this pull request Jul 5, 2021

Bi streaming no debugger stuff #219

Merged

5 tasks

Base automatically changed from transport_cleaning to master July 6, 2021 12:20

goodboy closed this Jul 6, 2021

goodboy mentioned this pull request Jul 31, 2021

Streaming API refinements #223

Open

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Bidir streaming #209

Bidir streaming #209

goodboy commented May 2, 2021 •

edited

ryanhiebert May 2, 2021

goodboy May 2, 2021

ryanhiebert May 2, 2021

goodboy May 12, 2021 •

edited

goodboy May 12, 2021 •

edited

goodboy May 12, 2021

goodboy May 12, 2021

goodboy May 12, 2021

gc-ss May 12, 2021

goodboy May 12, 2021

goodboy May 12, 2021

gc-ss May 12, 2021

goodboy May 13, 2021

goodboy May 13, 2021

goodboy May 13, 2021

goodboy May 13, 2021

goodboy May 13, 2021

goodboy Jun 2, 2021

goodboy Jun 5, 2021

goodboy commented Jun 14, 2021

goodboy commented Jul 6, 2021

goodboy commented Jul 6, 2021

		import tractor


		_registry: Dict[str, Set[tractor.ReceiveMsgStream]] = {

Bidir streaming #209

Bidir streaming #209

Conversation

goodboy commented May 2, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

goodboy May 12, 2021 • edited

Choose a reason for hiding this comment

goodboy May 12, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

goodboy commented Jun 14, 2021

goodboy commented Jul 6, 2021

goodboy commented Jul 6, 2021

goodboy commented May 2, 2021 •

edited

goodboy May 12, 2021 •

edited

goodboy May 12, 2021 •

edited