jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

Specially raise a `ContextCancelled` for a task-context rpc

bistream_backup
Tyler Goodlet 2021-06-13 18:03:50 -04:00
parent a1488a1773
commit 4601c88574
1 changed files with 10 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
tractor/_actor.py
View File

@ -28,6 +28,7 @@ from ._exceptions import (
unpack_error, unpack_error,
ModuleNotExposed, ModuleNotExposed,
is_multi_cancelled, is_multi_cancelled,
ContextCancelled,
TransportClosed, TransportClosed,
) )
from . import _debug from . import _debug
@ -152,9 +153,12 @@ async def _invoke(
task_status.started(cs) task_status.started(cs)
await chan.send({'return': await coro, 'cid': cid}) await chan.send({'return': await coro, 'cid': cid})
# if cs.cancelled_caught: if cs.cancelled_caught:
# # task was cancelled so relay to the cancel to caller # task-contex was cancelled so relay to the cancel to caller
# await chan.send({'return': await coro, 'cid': cid}) raise ContextCancelled(
f'{func.__name__} cancelled itself',
suberror_type=trio.Cancelled,
)
else: else:
# regular async function # regular async function
@ -168,7 +172,8 @@ async def _invoke(
# TODO: maybe we'll want differnet "levels" of debugging # TODO: maybe we'll want differnet "levels" of debugging
# eventualy such as ('app', 'supervisory', 'runtime') ? # eventualy such as ('app', 'supervisory', 'runtime') ?
if not isinstance(err, trio.ClosedResourceError) and ( if not isinstance(err, trio.ClosedResourceError) and (
not is_multi_cancelled(err) not is_multi_cancelled(err)) and (
not isinstance(err, ContextCancelled)
): ):
# XXX: is there any case where we'll want to debug IPC # XXX: is there any case where we'll want to debug IPC
# disconnects? I can't think of a reason that inspecting # disconnects? I can't think of a reason that inspecting
@ -303,7 +308,7 @@ class Actor:
self._parent_chan: Optional[Channel] = None self._parent_chan: Optional[Channel] = None
self._forkserver_info: Optional[ self._forkserver_info: Optional[
Tuple[Any, Any, Any, Any, Any]] = None Tuple[Any, Any, Any, Any, Any]] = None
self._actoruid2nursery: Dict[str, 'ActorNursery'] = {} # type: ignore self._actoruid2nursery: Dict[str, 'ActorNursery'] = {} # type: ignore # noqa
async def wait_for_peer( async def wait_for_peer(
self, uid: Tuple[str, str] self, uid: Tuple[str, str]