Context oriented error relay and MsgStream overruns

examples/full_fledged_streaming_service.py

@@ -7,7 +7,7 @@
 async def stream_data(seed):
     for i in range(seed):
         yield i
-        await trio.sleep(0)  # trigger scheduler
+        await trio.sleep(0.0001)  # trigger scheduler
 
 
 # this is the third actor; the aggregator

newsfragments/261.misc.rst

@@ -0,0 +1,37 @@
+Add cross-actor-task ``Context`` oriented error relay, a new
+stream overrun error-signal ``StreamOverrun``, and support
+disabling ``MsgStream`` backpressure as the default before a stream
+is opened or by choice of the user.
+
+We added stricter semantics around ``tractor.Context.open_stream():``
+particularly to do with streams which are only opened at one end.
+Previously, if only one end opened a stream there was no way for that
+sender to know if msgs are being received until first, the feeder mem
+chan on the receiver side hit a backpressure state and then that
+condition delayed its msg loop processing task to eventually create
+backpressure on the associated IPC transport. This is non-ideal in the
+case where the receiver side never opened a stream by mistake since it
+results in silent block of the sender and no adherence to the underlying
+mem chan buffer size settings (which is still unsolved btw).
+
+To solve this we add non-backpressure style message pushing inside
+``Actor._push_result()`` by default and only use the backpressure
+``trio.MemorySendChannel.send()`` call **iff** the local end of the
+context has entered ``Context.open_stream():``. This way if the stream
+was never opened but the mem chan is overrun, we relay back to the
+sender a (new exception) ``SteamOverrun`` error which is raised in the
+sender's scope with a special error message about the stream never
+having been opened. Further, this behaviour (non-backpressure style
+where senders can expect an error on overruns) can now be enabled with
+``.open_stream(backpressure=False)`` and the underlying mem chan size
+can be specified with a kwarg ``msg_buffer_size: int``.
+
+Further bug fixes and enhancements in this changeset include:
+- fix a race we were ignoring where if the callee task opened a context
+  it could enter ``Context.open_stream()`` before calling
+  ``.started()``.
+- Disallow calling ``Context.started()`` more then once.
+- Enable ``Context`` linked tasks error relaying via the new
+  ``Context._maybe_raise_from_remote_msg()`` which (for now) uses
+  a simple ``trio.Nursery.start_soon()`` to raise the error via closure
+  in the local scope.