diff --git a/tractor/runtime/_supervise.py b/tractor/runtime/_supervise.py index 9cb30941..020e3f34 100644 --- a/tractor/runtime/_supervise.py +++ b/tractor/runtime/_supervise.py @@ -438,8 +438,6 @@ class ActorNursery: ), bind_addrs=bind_addrs, loglevel=loglevel, - # use the run_in_actor nursery - nursery=self._ria_nursery, infect_asyncio=infect_asyncio, inherit_parent_main=inherit_parent_main, proc_kwargs=proc_kwargs @@ -580,6 +578,51 @@ class ActorNursery: self._join_procs.set() +async def _reap_ria_portals( + an: ActorNursery, + errors: dict[tuple[str, str], BaseException], + ria_children: list[tuple[Portal, Actor]]|None = None, +) -> None: + ''' + Wait on and stash the final result/error from every + `.run_in_actor()`-spawned child then cancel its actor + runtime, one `_spawn.cancel_on_completion()` task per + child. + + Replaces the per-child reaper task formerly spawned by + the spawn backends (keyed off + `._cancel_after_result_on_exit` membership) which + required routing such children into the (now removable) + `._ria_nursery`. Only call AFTER `._join_procs` is set + so user code inside the nursery block retains exclusive + result-await access; see the "manually await results" + note in `spawn._mp.mp_proc()`. + + ''' + if ria_children is None: + ria_children: list[tuple[Portal, Actor]] = [ + (portal, subactor) + for subactor, _, portal in an._children.values() + if portal in an._cancel_after_result_on_exit + ] + if not ria_children: + return + + async with ( + collapse_eg(), + trio.open_nursery() as tn, + ): + portal: Portal + subactor: Actor + for portal, subactor in ria_children: + tn.start_soon( + _spawn.cancel_on_completion, + portal, + subactor, + errors, + ) + + @acm async def _open_and_supervise_one_cancels_all_nursery( actor: Actor, @@ -641,6 +684,11 @@ async def _open_and_supervise_one_cancels_all_nursery( ) an._join_procs.set() + # collect results (and errors) from all + # `.run_in_actor()` children then cancel + # each, one reaper task per child. + await _reap_ria_portals(an, errors) + except BaseException as _inner_err: inner_err = _inner_err errors[actor.aid.uid] = inner_err @@ -704,9 +752,39 @@ async def _open_and_supervise_one_cancels_all_nursery( # '------ - ------' ) + # snapshot `.run_in_actor()` children + # BEFORE cancelling: each backend + # spawn-task pops its `._children` + # entry as the proc gets reaped. + ria_children: list = [ + (portal, subactor) + for subactor, _, portal + in an._children.values() + if portal in + an._cancel_after_result_on_exit + ] # cancel all subactors await an.cancel() + # then collect any already-relayed + # results/errors from ria children. + # Tightly bounded: anything + # collectable is already queued in + # the local ctx (relayed BEFORE the + # cancel above); a child hard-killed + # without relaying just parks its + # reaper which then self-cleans (a + # `trio.Cancelled` result is never + # stashed), mirroring the old + # backend-side reaper-vs-`soft_kill` + # cancel race. + with trio.move_on_after(0.5): + await _reap_ria_portals( + an, + errors, + ria_children=ria_children, + ) + # ria_nursery scope end # TODO: this is the handler around the ``.run_in_actor()`` diff --git a/tractor/spawn/_mp.py b/tractor/spawn/_mp.py index d0c8af32..4a680cc6 100644 --- a/tractor/spawn/_mp.py +++ b/tractor/spawn/_mp.py @@ -48,7 +48,6 @@ from ._entry import _mp_main # by `try_set_start_method()` after module load time. from . import _spawn from ._spawn import ( - cancel_on_completion, proc_waiter, soft_kill, ) @@ -187,30 +186,16 @@ async def mp_proc( with trio.CancelScope(shield=True): await actor_nursery._join_procs.wait() - async with trio.open_nursery() as nursery: - if portal in actor_nursery._cancel_after_result_on_exit: - nursery.start_soon( - cancel_on_completion, - portal, - subactor, - errors - ) - - # This is a "soft" (cancellable) join/reap which - # will remote cancel the actor on a ``trio.Cancelled`` - # condition. - await soft_kill( - proc, - proc_waiter, - portal - ) - - # cancel result waiter that may have been spawned in - # tandem if not done already - log.warning( - "Cancelling existing result waiter task for " - f"{subactor.aid.uid}") - nursery.cancel_scope.cancel() + # This is a "soft" (cancellable) join/reap which + # will remote cancel the actor on a ``trio.Cancelled`` + # condition. Any `.run_in_actor()` result-reaping + # happens up in the `ActorNursery` machinery (see + # `_supervise._reap_ria_portals()`), NOT here. + await soft_kill( + proc, + proc_waiter, + portal + ) finally: # hard reap sequence diff --git a/tractor/spawn/_trio.py b/tractor/spawn/_trio.py index 7d47175a..53955f4f 100644 --- a/tractor/spawn/_trio.py +++ b/tractor/spawn/_trio.py @@ -50,7 +50,6 @@ from tractor.msg import ( pretty_struct, ) from ._spawn import ( - cancel_on_completion, hard_kill, soft_kill, ) @@ -195,31 +194,16 @@ async def trio_proc( with trio.CancelScope(shield=True): await actor_nursery._join_procs.wait() - async with trio.open_nursery() as nursery: - if portal in actor_nursery._cancel_after_result_on_exit: - nursery.start_soon( - cancel_on_completion, - portal, - subactor, - errors - ) - - # This is a "soft" (cancellable) join/reap which - # will remote cancel the actor on a ``trio.Cancelled`` - # condition. - await soft_kill( - proc, - trio.Process.wait, # XXX, uses `pidfd_open()` below. - portal - ) - - # cancel result waiter that may have been spawned in - # tandem if not done already - log.cancel( - 'Cancelling portal result reaper task\n' - f'c)> {subactor.aid.reprol()!r}\n' - ) - nursery.cancel_scope.cancel() + # This is a "soft" (cancellable) join/reap which + # will remote cancel the actor on a ``trio.Cancelled`` + # condition. Any `.run_in_actor()` result-reaping + # happens up in the `ActorNursery` machinery (see + # `_supervise._reap_ria_portals()`), NOT here. + await soft_kill( + proc, + trio.Process.wait, # XXX, uses `pidfd_open()` below. + portal + ) finally: # XXX NOTE XXX: The "hard" reap since no actor zombies are