tractor/_root.py

@@ -243,6 +243,7 @@ async def open_root_actor(
                 logger.cancel("Shutting down root actor")
                 await actor.cancel()
     finally:
+        _state._current_actor = None
         logger.runtime("Root actor terminated")
 
 

tractor/_runtime.py

@@ -826,12 +826,7 @@ class Actor:
 
             if ctx._backpressure:
                 log.warning(text)
-                try:
                 await send_chan.send(msg)
-                except trio.BrokenResourceError:
-                    # XXX: local consumer has closed their side
-                    # so cancel the far end streaming task
-                    log.warning(f"{chan} is already closed")
             else:
                 try:
                     raise StreamOverrun(text) from None
@@ -1376,9 +1371,8 @@ async def async_main(
         actor.lifetime_stack.close()
 
         # Unregister actor from the arbiter
-        if (
-            registered_with_arbiter
-            and not actor.is_arbiter
+        if registered_with_arbiter and (
+                actor._arb_addr is not None
         ):
             failed = False
             with trio.move_on_after(0.5) as cs:

tractor/_streaming.py

@@ -69,7 +69,7 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
     '''
     def __init__(
         self,
-        ctx: Context,  # typing: ignore # noqa
+        ctx: 'Context',  # typing: ignore # noqa
         rx_chan: trio.MemoryReceiveChannel,
         _broadcaster: Optional[BroadcastReceiver] = None,
 
@@ -82,9 +82,6 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
         self._eoc: bool = False
         self._closed: bool = False
 
-    def ctx(self) -> Context:
-        return self._ctx
-
     # delegate directly to underlying mem channel
     def receive_nowait(self):
         msg = self._rx_chan.receive_nowait()
@@ -274,8 +271,7 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
         self,
 
     ) -> AsyncIterator[BroadcastReceiver]:
-        '''
-        Allocate and return a ``BroadcastReceiver`` which delegates
+        '''Allocate and return a ``BroadcastReceiver`` which delegates
         to this message stream.
 
         This allows multiple local tasks to receive each their own copy
@@ -384,7 +380,7 @@ class Context:
     # only set on the callee side
     _scope_nursery: Optional[trio.Nursery] = None
 
-    _backpressure: bool = True
+    _backpressure: bool = False
 
     async def send_yield(self, data: Any) -> None:
 
@@ -613,14 +609,7 @@ class Context:
                 # XXX: Make the stream "one-shot use".  On exit, signal
                 # ``trio.EndOfChannel``/``StopAsyncIteration`` to the
                 # far end.
-                try:
                 await self.send_stop()
-                except trio.BrokenResourceError:
-                    log.warning(
-                        f"Couldn't close: stream already broken?\n"
-                        f'actor: {self.chan.uid}\n'
-                        f'ctx id: {self.cid}'
-                    )
 
             finally:
                 if self._portal:

tractor/trionics/_mngrs.py

@@ -133,12 +133,12 @@ async def gather_contexts(
         # deliver control once all managers have started up
         await all_entered.wait()
 
-        try:
+        # NOTE: order *should* be preserved in the output values
+        # since ``dict``s are now implicitly ordered.
         yield tuple(unwrapped.values())
-        finally:
-            # NOTE: this is ABSOLUTELY REQUIRED to avoid
-            # the following wacky bug:
-            # <tractorbugurlhere>
+
+        # we don't need a try/finally since cancellation will be triggered
+        # by the surrounding nursery on error.
         parent_exit.set()