forked from goodboy/tractor
1
0
Fork 0

Timeout on remote task cancellation

Turns out you get a bad situation if the target actor who's task you're
trying to cancel has already died (eg. from an external
`KeyboardInterrupt` or other error) and so we need to eventually bail on
the RPC request. Also don't bother closing the channel created in
`open_portal()` manually since the cancel scope should take care of all
that.
contexts
Tyler Goodlet 2019-01-23 19:17:03 -05:00
parent 226312042a
commit 9f41297298
1 changed files with 11 additions and 6 deletions

View File

@ -146,8 +146,8 @@ class Portal:
log.warning( log.warning(
f"Cancelling async gen call {cid} to " f"Cancelling async gen call {cid} to "
f"{self.channel.uid}") f"{self.channel.uid}")
with trio.open_cancel_scope() as cleanup_scope: with trio.move_on_after(0.5) as cs:
cleanup_scope.shield = True cs.shield = True
# TODO: yeah.. it'd be nice if this was just an # TODO: yeah.. it'd be nice if this was just an
# async func on the far end. Gotta figure out a # async func on the far end. Gotta figure out a
# better way then implicitly feeding the ctx # better way then implicitly feeding the ctx
@ -157,6 +157,11 @@ class Portal:
async with aclosing(agen) as agen: async with aclosing(agen) as agen:
async for _ in agen: async for _ in agen:
pass pass
if cs.cancelled_caught:
if not self.channel.connected():
log.warning(
"May have failed to cancel remote task "
f"{cid} for {self.channel.uid}")
raise raise
# TODO: use AsyncExitStack to aclose() all agens # TODO: use AsyncExitStack to aclose() all agens
@ -239,13 +244,13 @@ class Portal:
cancel_scope.shield = True cancel_scope.shield = True
await self.run('self', 'cancel') await self.run('self', 'cancel')
return True return True
if cancel_scope.cancelled_caught:
log.warning(f"May have failed to cancel {self.channel.uid}")
return False
except trio.ClosedResourceError: except trio.ClosedResourceError:
log.warning( log.warning(
f"{self.channel} for {self.channel.uid} was already closed?") f"{self.channel} for {self.channel.uid} was already closed?")
return False return False
else:
log.warning(f"May have failed to cancel {self.channel.uid}")
return False
@dataclass @dataclass
@ -308,8 +313,8 @@ async def open_portal(
if was_connected: if was_connected:
# cancel remote channel-msg loop # cancel remote channel-msg loop
await channel.send(None) await channel.send(None)
await channel.aclose()
# cancel background msg loop task # cancel background msg loop task
msg_loop_cs.cancel() msg_loop_cs.cancel()
nursery.cancel_scope.cancel() nursery.cancel_scope.cancel()