goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

Add back broken send chan ignore block

stricter_context_starting
Tyler Goodlet 2021-12-06 17:04:17 -05:00
parent efba5229fc
commit 4856285dee
1 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
tractor/_actor.py
View File

@ -663,6 +663,18 @@ class Actor:
# consumer task # consumer task
await ctx._maybe_raise_from_remote_msg(msg) await ctx._maybe_raise_from_remote_msg(msg)
except trio.BrokenResourceError:
# TODO: what is the right way to handle the case where the
# local task has already sent a 'stop' / StopAsyncInteration
# to the other side but and possibly has closed the local
# feeder mem chan? Do we wait for some kind of ack or just
# let this fail silently and bubble up (currently)?
# XXX: local consumer has closed their side
# so cancel the far end streaming task
log.warning(f"{send_chan} consumer is already closed")
return
except trio.WouldBlock: except trio.WouldBlock:
# XXX: always push an error even if the local # XXX: always push an error even if the local
# receiver is in overrun state. # receiver is in overrun state.
@ -691,16 +703,11 @@ class Actor:
err_msg = pack_error(err) err_msg = pack_error(err)
err_msg['cid'] = cid err_msg['cid'] = cid
try: try:
# TODO: what is the right way to handle the case where the
# local task has already sent a 'stop' / StopAsyncInteration
# to the other side but and possibly has closed the local
# feeder mem chan? Do we wait for some kind of ack or just
# let this fail silently and bubble up (currently)?
await chan.send(err_msg) await chan.send(err_msg)
except trio.BrokenResourceError: except trio.BrokenResourceError:
# XXX: local consumer has closed their side # XXX: local consumer has closed their side
# so cancel the far end streaming task # so cancel the far end streaming task
log.warning(f"{send_chan} consumer is already closed") log.warning(f"{chan} is already closed")
def get_context( def get_context(
self, self,