goodboy
/
tractor
1
0
Fork 1
Code Issues 1 Pull Requests 4 Packages Projects Releases Wiki Activity

Handle broken mem chan on `Actor._push_result()` 

When backpressure is used and a feeder mem chan breaks during msg
delivery (usually because the IPC allocating task already terminated)
instead of raising we simply warn as we do for the non-backpressure
case.

Also, add a proper `Actor.is_arbiter` test inside `._invoke()` to avoid
doing an arbiter-registry lookup if the current actor **is** the
registrar.
ipc_failure_while_streaming
Tyler Goodlet 2022-11-09 19:22:33 -05:00
parent 9fd62cf71f
commit 4f977189c0
1 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tractor/_runtime.py
View File

@ -826,7 +826,12 @@ class Actor:
if ctx._backpressure: if ctx._backpressure:
log.warning(text) log.warning(text)
await send_chan.send(msg) 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: else:
try: try:
raise StreamOverrun(text) from None raise StreamOverrun(text) from None
@ -1371,8 +1376,9 @@ async def async_main(
actor.lifetime_stack.close() actor.lifetime_stack.close()
# Unregister actor from the arbiter # Unregister actor from the arbiter
if registered_with_arbiter and ( if (
actor._arb_addr is not None registered_with_arbiter
and not actor.is_arbiter
): ):
failed = False failed = False
with trio.move_on_after(0.5) as cs: with trio.move_on_after(0.5) as cs: