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

Drop the vestigial `._ria_nursery` 

Step B of the `._ria_nursery` removal (issue #477; see
`ai/conc-anal/ria_nursery_removal_plan.md`). With step A having
rerouted `.run_in_actor()` children onto the daemon nursery,
the 2ndary "run-in-actor" nursery spawns nothing and its stored
ref is never read — pure dead weight,

- collapse the inner `async with trio.open_nursery() as
  ria_nursery` layer in
  `_open_and_supervise_one_cancels_all_nursery`; `da_nursery` is
  now the single nursery for ALL subactors.
- `ActorNursery.__init__` loses the `ria_nursery` param + the
  `self._ria_nursery` attr; `start_actor()` loses its `nursery=`
  escape-hatch (spawns via `self._da_nursery` directly).
- `._cancel_after_result_on_exit` stays — still the ria-child
  discriminator for `_reap_ria_portals()`.

Behavior-preserving: a zero-task `trio.open_nursery()` only adds
a checkpoint. The two error handlers are KEPT (now nested under
the single nursery); merging them changes error/cancel
propagation and is deferred to its own PR (TODO left at the
outer `except`).

Prompt-IO: ai/prompt-io/claude/20260702T172233Z_5cd190c5_prompt_io.md

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
drop_ria_nursery
Gud Boi 2026-07-02 17:47:46 -04:00
parent d2e812fba5
commit 9201a2ed53
1 changed files with 126 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

278
tractor/runtime/_supervise.py
View File

@ -199,7 +199,6 @@ class ActorNursery:
self, self,
# TODO: maybe def these as fields of a struct looking type? # TODO: maybe def these as fields of a struct looking type?
actor: Actor, actor: Actor,
ria_nursery: trio.Nursery,
da_nursery: trio.Nursery, da_nursery: trio.Nursery,
errors: dict[tuple[str, str], BaseException], errors: dict[tuple[str, str], BaseException],
@ -233,14 +232,12 @@ class ActorNursery:
# and syncing purposes to any actor opened nurseries. # and syncing purposes to any actor opened nurseries.
self._implicit_runtime_started: bool = False self._implicit_runtime_started: bool = False
# TODO: remove the `.run_in_actor()` API and thus this 2ndary
# nursery when that API get's moved outside this primitive!
self._ria_nursery = ria_nursery
# TODO, factor this into a .hilevel api! # TODO, factor this into a .hilevel api!
# #
# portals spawned with ``run_in_actor()`` are # portals spawned with ``run_in_actor()`` are
# cancelled when their "main" result arrives # 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() self._cancel_after_result_on_exit: set = set()
# trio.Nursery-like cancel (request) statuses # trio.Nursery-like cancel (request) statuses
@ -298,11 +295,6 @@ class ActorNursery:
debug_mode: bool|None = None, debug_mode: bool|None = None,
infect_asyncio: bool = False, infect_asyncio: bool = False,
inherit_parent_main: bool = True, inherit_parent_main: bool = True,
# TODO: ideally we can rm this once we no longer have
# a `._ria_nursery` since the dependent APIs have been
# removed!
nursery: trio.Nursery|None = None,
proc_kwargs: dict[str, typing.Any] | None = None, proc_kwargs: dict[str, typing.Any] | None = None,
) -> Portal: ) -> Portal:
@ -364,10 +356,8 @@ class ActorNursery:
# start a task to spawn a process # start a task to spawn a process
# blocks until process has been started and a portal setup # blocks until process has been started and a portal setup
nursery: trio.Nursery = nursery or self._da_nursery
# XXX: the type ignore is actually due to a `mypy` bug # XXX: the type ignore is actually due to a `mypy` bug
return await nursery.start( # type: ignore return await self._da_nursery.start( # type: ignore
partial( partial(
_spawn.new_proc, _spawn.new_proc,
name, name,
@ -639,161 +629,145 @@ async def _open_and_supervise_one_cancels_all_nursery(
# the collection of errors retreived from spawned sub-actors # the collection of errors retreived from spawned sub-actors
errors: dict[tuple[str, str], BaseException] = {} errors: dict[tuple[str, str], BaseException] = {}
# This is the outermost level "deamon actor" nursery. It is awaited # The single "daemon actor" nursery into which ALL subactors
# **after** the below inner "run in actor nursery". This allows for # are spawned — both `.start_actor()` daemons AND
# handling errors that are generated by the inner nursery in # `.run_in_actor()` one-shots. The latter's result-reaping now
# a supervisor strategy **before** blocking indefinitely to wait for # runs via `_reap_ria_portals()` at block-exit rather than a
# actors spawned in "daemon mode" (aka started using # 2ndary `._ria_nursery` (see the #477 removal); errors from
# `ActorNursery.start_actor()`). # this nursery bubble up to the caller.
# errors from this daemon actor nursery bubble up to caller
async with ( async with (
collapse_eg(), collapse_eg(),
trio.open_nursery() as da_nursery, trio.open_nursery() as da_nursery,
): ):
an = ActorNursery(
actor,
da_nursery,
errors
)
try: try:
# This is the inner level "run in actor" nursery. It is try:
# awaited first since actors spawned in this way (using # spawning of actors happens in the caller's scope
# `ActorNusery.run_in_actor()`) are expected to only # after we yield upwards
# return a single result and then complete (i.e. be canclled yield an
# gracefully). Errors collected from these actors are
# immediately raised for handling by a supervisor strategy. # When we didn't error in the caller's scope,
# As such if the strategy propagates any error(s) upwards # signal all process-monitor-tasks to conduct
# the above "daemon actor" nursery will be notified. # the "hard join phase".
async with ( log.runtime(
collapse_eg(), 'Waiting on subactors to complete:\n'
trio.open_nursery() as ria_nursery, f'>}} {len(an._children)}\n'
):
an = ActorNursery(
actor,
ria_nursery,
da_nursery,
errors
) )
try: an._join_procs.set()
# spawning of actors happens in the caller's scope
# after we yield upwards
yield an
# When we didn't error in the caller's scope, # collect results (and errors) from all
# signal all process-monitor-tasks to conduct # `.run_in_actor()` children then cancel
# the "hard join phase". # each, one reaper task per child.
log.runtime( await _reap_ria_portals(an, errors)
'Waiting on subactors to complete:\n'
f'>}} {len(an._children)}\n'
)
an._join_procs.set()
# collect results (and errors) from all except BaseException as _inner_err:
# `.run_in_actor()` children then cancel inner_err = _inner_err
# each, one reaper task per child. errors[actor.aid.uid] = inner_err
await _reap_ria_portals(an, errors)
except BaseException as _inner_err: # If we error in the root but the debugger is
inner_err = _inner_err # engaged we don't want to prematurely kill (and
errors[actor.aid.uid] = inner_err # thus clobber access to) the local tty since it
# will make the pdb repl unusable.
# Instead try to wait for pdb to be released before
# tearing down.
await debug.maybe_wait_for_debugger(
child_in_debug=an._at_least_one_child_in_debug
)
# If we error in the root but the debugger is # if the caller's scope errored then we activate our
# engaged we don't want to prematurely kill (and # one-cancels-all supervisor strategy (don't
# thus clobber access to) the local tty since it # worry more are coming).
# will make the pdb repl unusable. an._join_procs.set()
# Instead try to wait for pdb to be released before
# tearing down.
await debug.maybe_wait_for_debugger(
child_in_debug=an._at_least_one_child_in_debug
)
# if the caller's scope errored then we activate our # XXX NOTE XXX: hypothetically an error could
# one-cancels-all supervisor strategy (don't # be raised and then a cancel signal shows up
# worry more are coming). # slightly after in which case the `else:`
an._join_procs.set() # block here might not complete? For now,
# shield both.
with trio.CancelScope(shield=True):
etype: type = type(inner_err)
if etype in (
trio.Cancelled,
KeyboardInterrupt,
) or (
is_multi_cancelled(inner_err)
):
log.cancel(
f'Actor-nursery cancelled by {etype}\n\n'
# XXX NOTE XXX: hypothetically an error could f'{current_actor().aid.uid}\n'
# be raised and then a cancel signal shows up f' |_{an}\n\n'
# slightly after in which case the `else:`
# block here might not complete? For now,
# shield both.
with trio.CancelScope(shield=True):
etype: type = type(inner_err)
if etype in (
trio.Cancelled,
KeyboardInterrupt,
) or (
is_multi_cancelled(inner_err)
):
log.cancel(
f'Actor-nursery cancelled by {etype}\n\n'
f'{current_actor().aid.uid}\n' # TODO: show tb str?
f' |_{an}\n\n' # f'{tb_str}'
)
elif etype in {
ContextCancelled,
}:
log.cancel(
'Actor-nursery caught remote cancellation\n'
'\n'
f'{inner_err.tb_str}'
)
else:
log.exception(
'Nursery errored with:\n'
# TODO: show tb str? # TODO: same thing as in
# f'{tb_str}' # `._invoke()` to compute how to
) # place this div-line in the
elif etype in { # middle of the above msg
ContextCancelled, # content..
}: # -[ ] prolly helper-func it too
log.cancel( # in our `.log` module..
'Actor-nursery caught remote cancellation\n' # '------ - ------'
'\n' )
f'{inner_err.tb_str}'
)
else:
log.exception(
'Nursery errored with:\n'
# TODO: same thing as in # snapshot `.run_in_actor()` children
# `._invoke()` to compute how to # BEFORE cancelling: each backend
# place this div-line in the # spawn-task pops its `._children`
# middle of the above msg # entry as the proc gets reaped.
# content.. ria_children: list = [
# -[ ] prolly helper-func it too (portal, subactor)
# in our `.log` module.. for subactor, _, portal
# '------ - ------' in an._children.values()
) if portal in
an._cancel_after_result_on_exit
]
# cancel all subactors
await an.cancel()
# snapshot `.run_in_actor()` children # then collect any already-relayed
# BEFORE cancelling: each backend # results/errors from ria children.
# spawn-task pops its `._children` # Tightly bounded: anything
# entry as the proc gets reaped. # collectable is already queued in
ria_children: list = [ # the local ctx (relayed BEFORE the
(portal, subactor) # cancel above); a child hard-killed
for subactor, _, portal # without relaying just parks its
in an._children.values() # reaper which then self-cleans (a
if portal in # `trio.Cancelled` result is never
an._cancel_after_result_on_exit # stashed), mirroring the old
] # backend-side reaper-vs-`soft_kill`
# cancel all subactors # cancel race.
await an.cancel() with trio.move_on_after(0.5):
await _reap_ria_portals(
an,
errors,
ria_children=ria_children,
)
# then collect any already-relayed # Safety-net handler for anything escaping the inner
# results/errors from ria children. # `except BaseException` above (e.g. a `trio.Cancelled`
# Tightly bounded: anything # during its non-shielded debugger-wait, or a failure
# collectable is already queued in # inside the shielded teardown itself). Post-#477 the
# the local ctx (relayed BEFORE the # 2ndary `._ria_nursery` (and its dedicated handler) is
# cancel above); a child hard-killed # gone; this now guards the single daemon nursery scope.
# without relaying just parks its # TODO: can this be merged into the inner handler now that
# reaper which then self-cleans (a # there's only one nursery? (higher-risk, own PR)
# `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()``
# nursery. Ideally we can drop this entirely in the future as
# the whole ``.run_in_actor()`` API should be built "on top of"
# this lower level spawn-request-cancel "daemon actor" API where
# a local in-actor task nursery is used with one-to-one task
# + `await Portal.run()` calls and the results/errors are
# handled directly (inline) and errors by the local nursery.
except ( except (
Exception, Exception,
BaseExceptionGroup, BaseExceptionGroup,