From 2a59cefbe80ae770b4c0bec074eeb51d07cfb0be Mon Sep 17 00:00:00 2001 From: goodboy Date: Mon, 6 Jul 2026 12:52:29 -0400 Subject: [PATCH] Remove `run_in_actor()` + the ria reap cluster MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The final excision of #477: with zero in-repo callers left (all tests/examples/docs migrated to `to_actor.run()` et al) the entire legacy one-shot machinery drops out, - `runtime/_supervise.py`: `ActorNursery.run_in_actor()`, the `._cancel_after_result_on_exit` portal-set and the `_reap_ria_portals()` teardown-reaper (both its happy-path block-exit call AND the error-path snapshot + 0.5s-bounded collection) are deleted — one-shot result-waiting now lives entirely in the caller's task via `to_actor.run()`, whose enclosing cancel-scope bounds the wait by construction (the correct-scoping fix for the unbounded-reap hang class; the `d1fb4a1a` guard test now passes structurally). - `runtime/_portal.py`: `Portal._submit_for_result()`, `._expect_result_ctx`, `._final_result_msg/_pld`, `.wait_for_result()` + the deprecated `.result()` alias are gone — a `Portal` no longer has any "main result" notion. NB `Context.wait_for_result()` is a different (very alive) API and is untouched. - `spawn/_spawn.py`: `exhaust_portal()` + `cancel_on_completion()` (the reaper tasks) deleted; backend comment sweeps in `_trio.py`/`_mp.py`. - `_exceptions.py`: the `NoResult` sentinel dies with its lone reader. - `tests/test_ringbuf.py`: drop a daemon-portal `.result()` call that was already a warn + `NoResult` no-op (the ctx-acm exit does the real result-wait); unshadow the 2nd `sctx` as `rctx`. - comment/docstring x-ref sweeps: `msg/types.py`, `_context.py`, `to_actor/`, `tests/test_to_actor.py`. Gate: `test_to_actor test_spawning test_cancellation test_infected_asyncio test_local test_rpc` = 81 passed, 3 xfailed on `trio`; +`test_ringbuf` = 70 passed, 3 skipped, 3 xfailed on `mp_spawn`. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- tests/test_ringbuf.py | 7 +- tests/test_to_actor.py | 6 +- tractor/_context.py | 2 +- tractor/_exceptions.py | 4 - tractor/msg/types.py | 3 +- tractor/runtime/_portal.py | 106 ------------------- tractor/runtime/_supervise.py | 186 ++-------------------------------- tractor/spawn/_mp.py | 4 +- tractor/spawn/_spawn.py | 96 +----------------- tractor/spawn/_trio.py | 4 +- tractor/to_actor/__init__.py | 4 +- tractor/to_actor/_api.py | 8 +- 12 files changed, 28 insertions(+), 402 deletions(-) diff --git a/tests/test_ringbuf.py b/tests/test_ringbuf.py index e55a87b9..9db53f58 100644 --- a/tests/test_ringbuf.py +++ b/tests/test_ringbuf.py @@ -131,9 +131,12 @@ def test_ringbuf( child_read_shm, **common_kwargs, total_bytes=total_bytes, - ) as (sctx, _sent), + ) as (rctx, _sent), ): - await recv_p.result() + # ctx-acm exits await each child task's + # `Return` (the prior `recv_p.result()` here + # was a daemon-portal no-op). + pass await send_p.cancel_actor() await recv_p.cancel_actor() diff --git a/tests/test_to_actor.py b/tests/test_to_actor.py index 40ade1c4..f8aba6da 100644 --- a/tests/test_to_actor.py +++ b/tests/test_to_actor.py @@ -1,8 +1,8 @@ ''' `tractor.to_actor`: one-shot single-remote-task API suite. -Verifies the "spiritual successor" to (and eventual -replacement of) `ActorNursery.run_in_actor()`; see +Verifies the "spiritual successor" to (and replacement of) +the removed legacy `ActorNursery.run_in_actor()`; see https://github.com/goodboy/tractor/issues/477 ''' @@ -82,7 +82,7 @@ async def test_remote_error_relayed_to_caller_task( A remote task error is raised directly in the caller's task as a boxed `RemoteActorError` instead of surfacing at actor-nursery teardown as with the - legacy `.run_in_actor()` API. + removed legacy `.run_in_actor()` API. ''' with pytest.raises(RemoteActorError) as excinfo: diff --git a/tractor/_context.py b/tractor/_context.py index d538e361..1aa372cf 100644 --- a/tractor/_context.py +++ b/tractor/_context.py @@ -780,7 +780,7 @@ class Context: # `Portal.open_context()` has been opened since it's # assumed that other portal APIs like, # - `Portal.run()`, - # - `ActorNursery.run_in_actor()` + # - `to_actor.run()` # do their own error checking at their own call points and # result processing. diff --git a/tractor/_exceptions.py b/tractor/_exceptions.py index d838544b..5686533a 100644 --- a/tractor/_exceptions.py +++ b/tractor/_exceptions.py @@ -1161,10 +1161,6 @@ class TransportClosed(Exception): ) -class NoResult(RuntimeError): - "No final result is expected for this actor" - - class ModuleNotExposed(ModuleNotFoundError): "The requested module is not exposed for RPC" diff --git a/tractor/msg/types.py b/tractor/msg/types.py index 4f3e33cc..6fe775a4 100644 --- a/tractor/msg/types.py +++ b/tractor/msg/types.py @@ -304,12 +304,11 @@ class Start( It is called by all the following public APIs: - - `ActorNursery.run_in_actor()` + - `to_actor.run()` - `Portal.run()` `|_.run_from_ns()` `|_.open_stream_from()` - `|_._submit_for_result()` - `Context.open_context()` diff --git a/tractor/runtime/_portal.py b/tractor/runtime/_portal.py index 738599ac..6ee20ed7 100644 --- a/tractor/runtime/_portal.py +++ b/tractor/runtime/_portal.py @@ -50,13 +50,11 @@ from ..ipc import Channel from ..log import get_logger from ..msg import ( # Error, - PayloadMsg, NamespacePath, Return, ) from .._exceptions import ( ActorTooSlowError, - NoResult, TransportClosed, ) from .._context import ( @@ -102,14 +100,6 @@ class Portal: ) -> None: self._chan: Channel = channel - # during the portal's lifetime - self._final_result_pld: Any|None = None - self._final_result_msg: PayloadMsg|None = None - - # When set to a ``Context`` (when _submit_for_result is called) - # it is expected that ``result()`` will be awaited at some - # point. - self._expect_result_ctx: Context|None = None self._streams: set[MsgStream] = set() # TODO, this should be PRIVATE (and never used publicly)! since it's just @@ -137,102 +127,6 @@ class Portal: ) return self.chan - # TODO: factor this out into a `.highlevel` API-wrapper that uses - # a single `.open_context()` call underneath. - async def _submit_for_result( - self, - ns: str, - func: str, - **kwargs - ) -> None: - - if self._expect_result_ctx is not None: - raise RuntimeError( - 'A pending main result has already been submitted' - ) - - self._expect_result_ctx: Context = await self.actor.start_remote_task( - self.channel, - nsf=NamespacePath(f'{ns}:{func}'), - kwargs=kwargs, - portal=self, - ) - - # TODO: we should deprecate this API right? since if we remove - # `.run_in_actor()` (and instead move it to a `.highlevel` - # wrapper api (around a single `.open_context()` call) we don't - # really have any notion of a "main" remote task any more? - # - # @api_frame - async def wait_for_result( - self, - hide_tb: bool = True, - ) -> Any: - ''' - Return the final result delivered by a `Return`-msg from the - remote peer actor's "main" task's `return` statement. - - ''' - __tracebackhide__: bool = hide_tb - # Check for non-rpc errors slapped on the - # channel for which we always raise - exc = self.channel._exc - if exc: - raise exc - - # not expecting a "main" result - if self._expect_result_ctx is None: - peer_id: str = f'{self.channel.aid.reprol()!r}' - log.warning( - f'Portal to peer {peer_id} will not deliver a final result?\n' - f'\n' - f'Context.result() can only be called by the parent of ' - f'a sub-actor when it was spawned with ' - f'`ActorNursery.run_in_actor()`' - f'\n' - f'Further this `ActorNursery`-method-API will deprecated in the' - f'near fututre!\n' - ) - return NoResult - - # expecting a "main" result - assert self._expect_result_ctx - - if self._final_result_msg is None: - try: - ( - self._final_result_msg, - self._final_result_pld, - ) = await self._expect_result_ctx._pld_rx.recv_msg( - ipc=self._expect_result_ctx, - expect_msg=Return, - ) - except BaseException as err: - # TODO: wrap this into `@api_frame` optionally with - # some kinda filtering mechanism like log levels? - __tracebackhide__: bool = False - raise err - - return self._final_result_pld - - # TODO: factor this out into a `.highlevel` API-wrapper that uses - # a single `.open_context()` call underneath. - async def result( - self, - *args, - **kwargs, - ) -> Any|Exception: - typname: str = type(self).__name__ - log.warning( - f'`{typname}.result()` is DEPRECATED!\n' - f'\n' - f'Use `{typname}.wait_for_result()` instead!\n' - ) - return await self.wait_for_result( - *args, - **kwargs, - ) - async def _cancel_streams(self): # terminate all locally running async generator # IPC calls diff --git a/tractor/runtime/_supervise.py b/tractor/runtime/_supervise.py index fe28344f..2cb0600b 100644 --- a/tractor/runtime/_supervise.py +++ b/tractor/runtime/_supervise.py @@ -20,7 +20,6 @@ """ from contextlib import asynccontextmanager as acm from functools import partial -import inspect from typing import ( TYPE_CHECKING, ) @@ -232,14 +231,6 @@ class ActorNursery: # and syncing purposes to any actor opened nurseries. self._implicit_runtime_started: bool = False - # TODO, factor this into a .hilevel api! - # - # portals spawned with ``run_in_actor()`` are - # cancelled when their "main" result arrives. Reaped by - # `_reap_ria_portals()` at nursery-block exit now that - # the 2ndary `._ria_nursery` is gone (see issue #477). - self._cancel_after_result_on_exit: set = set() - # trio.Nursery-like cancel (request) statuses self._cancelled_caught: bool = False self._cancel_called: bool = False @@ -372,84 +363,6 @@ class ActorNursery: ) ) - # TODO: DEPRECATE THIS: - # -[x] impl instead as a hilevel wrapper on top of - # the lower level daemon-spawn + portal APIs - # |_ see `.to_actor.run()` (issue #477) which does - # `.start_actor()` + `Portal.run()` + a one-shot - # reap via `Portal.cancel_actor()`. - # -[ ] emit a `DeprecationWarning` here (requires - # migrating all in-repo usage first!) - # -[ ] use @api_frame on the wrapper - async def run_in_actor( - self, - - fn: typing.Callable, - *, - - name: str | None = None, - bind_addrs: UnwrappedAddress|None = None, - rpc_module_paths: list[str] | None = None, - enable_modules: list[str] | None = None, - loglevel: str | None = None, # set log level per subactor - infect_asyncio: bool = False, - inherit_parent_main: bool = True, - proc_kwargs: dict[str, typing.Any] | None = None, - - **kwargs, # explicit args to ``fn`` - - ) -> Portal: - ''' - Spawn a new actor, run a lone task, then terminate the actor and - return its result. - - Actors spawned using this method are kept alive at nursery teardown - until the task spawned by executing ``fn`` completes at which point - the actor is terminated. - - NOTE: prefer the (eventual) replacement API - `tractor.to_actor.run()` which delivers the same - one-shot semantics decoupled from this nursery's - internal spawn machinery; see issue #477. - - ''' - __runtimeframe__: int = 1 # noqa - mod_path: str = fn.__module__ - - if name is None: - # use the explicit function name if not provided - name = fn.__name__ - - proc_kwargs = dict(proc_kwargs or {}) - portal: Portal = await self.start_actor( - name, - enable_modules=[mod_path] + ( - enable_modules or rpc_module_paths or [] - ), - bind_addrs=bind_addrs, - loglevel=loglevel, - infect_asyncio=infect_asyncio, - inherit_parent_main=inherit_parent_main, - proc_kwargs=proc_kwargs - ) - - # XXX: don't allow stream funcs - if not ( - inspect.iscoroutinefunction(fn) and - not getattr(fn, '_tractor_stream_function', False) - ): - raise TypeError(f'{fn} must be an async function!') - - # this marks the actor to be cancelled after its portal result - # is retreived, see logic in `open_nursery()` below. - self._cancel_after_result_on_exit.add(portal) - await portal._submit_for_result( - mod_path, - fn.__name__, - **kwargs - ) - return portal - # @api_frame async def cancel( self, @@ -568,51 +481,6 @@ 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, @@ -627,11 +495,10 @@ async def _open_and_supervise_one_cancels_all_nursery( errors: dict[tuple[str, str], BaseException] = {} # The single "daemon actor" nursery into which ALL subactors - # are spawned — both `.start_actor()` daemons AND - # `.run_in_actor()` one-shots. The latter's result-reaping now - # runs via `_reap_ria_portals()` at block-exit rather than a - # 2ndary `._ria_nursery` (see the #477 removal); errors from - # this nursery bubble up to the caller. + # are spawned; one-shot (`to_actor.run()`) subactors are + # result-waited and reaped in their caller's own task-scope + # (see the #477 `.run_in_actor()`/`._ria_nursery` removal); + # errors from this nursery bubble up to the caller. async with ( collapse_eg(), trio.open_nursery() as da_nursery, @@ -655,11 +522,6 @@ 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) - # Single one-cancels-all handler for the (now single) # daemon nursery. Pre-#477 a 2ndary `._ria_nursery` # required a separate *outer* handler to catch errors @@ -733,46 +595,14 @@ 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, - ) - finally: - # No errors were raised while awaiting ".run_in_actor()" - # actors but those actors may have returned remote errors as - # results (meaning they errored remotely and have relayed - # those errors back to this parent actor). The errors are - # collected in ``errors`` so cancel all actors, summarize - # all errors and re-raise. + # an error was stashed by the handler above (or by + # a spawn task via the shared `errors` dict) so + # cancel any remaining subactors, summarize and + # re-raise. if errors: if an._children: with trio.CancelScope(shield=True): diff --git a/tractor/spawn/_mp.py b/tractor/spawn/_mp.py index 4a680cc6..90cc83c9 100644 --- a/tractor/spawn/_mp.py +++ b/tractor/spawn/_mp.py @@ -188,9 +188,7 @@ async def mp_proc( # 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. + # condition. await soft_kill( proc, proc_waiter, diff --git a/tractor/spawn/_spawn.py b/tractor/spawn/_spawn.py index c0218c30..501ee5a2 100644 --- a/tractor/spawn/_spawn.py +++ b/tractor/spawn/_spawn.py @@ -126,98 +126,6 @@ def try_set_start_method( return _ctx -async def exhaust_portal( - - portal: Portal, - actor: Actor - -) -> Any: - ''' - Pull final result from portal (assuming it has one). - - If the main task is an async generator do our best to consume - what's left of it. - ''' - __tracebackhide__ = True - try: - log.debug( - f'Waiting on final result from {actor.aid.uid}' - ) - - # XXX: streams should never be reaped here since they should - # always be established and shutdown using a context manager api - final: Any = await portal.wait_for_result() - - except ( - Exception, - BaseExceptionGroup, - ) as err: - # we reraise in the parent task via a ``BaseExceptionGroup`` - return err - - except trio.Cancelled as err: - # lol, of course we need this too ;P - # TODO: merge with above? - log.warning( - 'Cancelled portal result waiter task:\n' - f'uid: {portal.channel.aid}\n' - f'error: {err}\n' - ) - return err - - else: - log.debug( - f'Returning final result from portal:\n' - f'uid: {portal.channel.aid}\n' - f'result: {final}\n' - ) - return final - - -async def cancel_on_completion( - - portal: Portal, - actor: Actor, - errors: dict[tuple[str, str], Exception], - -) -> None: - ''' - Cancel actor gracefully once its "main" portal's - result arrives. - - Should only be called for actors spawned via the - `Portal.run_in_actor()` API. - - => and really this API will be deprecated and should be - re-implemented as a `.hilevel.one_shot_task_nursery()`..) - - ''' - # if this call errors we store the exception for later - # in ``errors`` which will be reraised inside - # an exception group and we still send out a cancel request - result: Any|Exception = await exhaust_portal( - portal, - actor, - ) - if isinstance(result, Exception): - errors[actor.aid.uid]: Exception = result - log.cancel( - 'Cancelling subactor runtime due to error:\n\n' - f'Portal.cancel_actor() => {portal.channel.aid}\n\n' - f'error: {result}\n' - ) - - else: - log.runtime( - 'Cancelling subactor gracefully:\n\n' - f'Portal.cancel_actor() => {portal.channel.aid}\n\n' - f'result: {result}\n' - ) - - # cancel the process now that we have a final result - await portal.cancel_actor() - - async def hard_kill( proc: trio.Process, @@ -461,8 +369,8 @@ async def new_proc( # NOTE: bottom-of-module to avoid a circular import since the -# backend submodules pull `cancel_on_completion`/`soft_kill`/ -# `hard_kill`/`proc_waiter` from this module. +# backend submodules pull `soft_kill`/`hard_kill`/`proc_waiter` +# from this module. from ._trio import trio_proc from ._mp import mp_proc diff --git a/tractor/spawn/_trio.py b/tractor/spawn/_trio.py index 53955f4f..072c0a6a 100644 --- a/tractor/spawn/_trio.py +++ b/tractor/spawn/_trio.py @@ -196,9 +196,7 @@ async def trio_proc( # 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. + # condition. await soft_kill( proc, trio.Process.wait, # XXX, uses `pidfd_open()` below. diff --git a/tractor/to_actor/__init__.py b/tractor/to_actor/__init__.py index bbec4133..4c39ec30 100644 --- a/tractor/to_actor/__init__.py +++ b/tractor/to_actor/__init__.py @@ -23,8 +23,8 @@ Adopts the "run it over there" parlance from analogous reuse) a subactor, schedule a single remote task, wait on its result and (when the call owns the subactor) reap it. -The "spiritual successor" to (and eventual replacement of) -the `ActorNursery.run_in_actor()` API; see +The "spiritual successor" to (and replacement of) the removed +legacy `ActorNursery.run_in_actor()` API; see https://github.com/goodboy/tractor/issues/477 ''' diff --git a/tractor/to_actor/_api.py b/tractor/to_actor/_api.py index fc15193f..07c40688 100644 --- a/tractor/to_actor/_api.py +++ b/tractor/to_actor/_api.py @@ -31,7 +31,7 @@ the lower level daemon-actor spawn + portal APIs, such that error collection and propagation happens in the *caller's task* (and thus whatever `trio` nursery/scope encloses it) instead of inside the actor-nursery's -spawn-machinery nurseries as with the (to be deprecated) +spawn-machinery nurseries as with the (now removed) legacy `ActorNursery.run_in_actor()` API. ''' @@ -151,9 +151,9 @@ async def run( the distributed-parallelism equivalent of `trio.to_thread.run_sync()`. - Unlike `ActorNursery.run_in_actor()` (which returns - a `Portal` whose result is only collected at - actor-nursery teardown) this is a plain "call and + Unlike the removed legacy `.run_in_actor()` (which + returned a `Portal` whose result was only collected + at actor-nursery teardown) this is a plain "call and wait" primitive: any remote error is raised HERE, in the caller's task. Concurrency is composed the usual `trio` way by scheduling multiple `run()` calls in