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

Merge the supervise error handlers into one 

Step B2 of the `._ria_nursery` removal (issue #477; see
`ai/conc-anal/ria_nursery_removal_plan.md`). With the 2ndary
nursery gone (step B), the two nested error handlers in
`_open_and_supervise_one_cancels_all_nursery` collapse to one,

- the outer `except (Exception, BaseExceptionGroup,
  trio.Cancelled)` existed to catch errors bubbling from the
  old `._ria_nursery.__aexit__` reaper-group; that nursery no
  longer exists.
- trace shows the outer handler's `raise` was already DEAD: the
  inner handler records `errors[uid]` as its first action, so
  `errors` is always non-empty by the time anything could reach
  the outer handler, and the `finally`'s raise-from-`errors`
  always superseded the outer `raise`.
- so fold both into a single `except BaseException as
  _scope_err` guarding the lone daemon nursery; the `finally`
  (unchanged) still raises the collected `errors` as a single
  exc or `BaseExceptionGroup`.
- drop the now-unused `outer_err`/`inner_err` locals.

Behaviour-preserving (net ~30 lines lighter); the big diff is
the one-level de-indent of the handler body. The two remaining
`maybe_wait_for_debugger()` guards collapse to the single
pre-teardown wait.

Prompt-IO: ai/prompt-io/claude/20260702T222544Z_9201a2ed_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
Gud Boi 2026-07-02 18:33:17 -04:00
parent 762deb1a13
commit 09bbe9701e
1 changed files with 138 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

408
tractor/runtime/_supervise.py
View File

@ -90,7 +90,7 @@ async def _try_cancel_then_kill(
Sends a graceful actor-runtime cancel-RPC via Sends a graceful actor-runtime cancel-RPC via
`Portal.cancel_actor(raise_on_timeout=True)`. If the bounded-wait `Portal.cancel_actor(raise_on_timeout=True)`. If the bounded-wait
expires before the peer ack's, `ActorTooSlowError` is raised and expires before the peer ack's, `ActorTooSlowError` is raised and
we escalate via `proc.kill()` per SC-discipline: we escalate via `proc.terminate()` (SIGTERM) per SC-discipline:
graceful cancel-req -> bounded wait -> hard-kill graceful cancel-req -> bounded wait -> hard-kill
@ -102,9 +102,11 @@ async def _try_cancel_then_kill(
the wider write-up. the wider write-up.
''' '''
# XXX, delay hard-kill escalation while any debugger guard # XXX, do NOT escalate to `proc.terminate()` while ANY of
# below is active. Killing the sub immediately would tear down # the following are true — SIGTERM-ing a sub would tear
# its tree and clobber an actor proxying a REPL session: # down its sub-tree including any descendant proxying
# stdio to/from a REPL-locked actor, clobbering the user's
# debug session:
# #
# - `Lock.ctx_in_debug is not None`: most precise — some # - `Lock.ctx_in_debug is not None`: most precise — some
# actor in the tree is currently REPL-locked. Set in the # actor in the tree is currently REPL-locked. Set in the
@ -120,7 +122,7 @@ async def _try_cancel_then_kill(
# child. # child.
# #
# - `debug_mode_active`: this nursery has at least one # - `debug_mode_active`: this nursery has at least one
# child started with an explicit `debug_mode=True` arg # child started with an explicit `debug_mode=` arg
# (`ActorNursery._at_least_one_child_in_debug`). Catches # (`ActorNursery._at_least_one_child_in_debug`). Catches
# the case where root is NOT in debug-mode but a # the case where root is NOT in debug-mode but a
# nursery-direct child opted in. # nursery-direct child opted in.
@ -130,57 +132,36 @@ async def _try_cancel_then_kill(
# mutated by per-child `debug_mode=True`). ORing covers # mutated by per-child `debug_mode=True`). ORing covers
# every flavor without false-positively skipping # every flavor without false-positively skipping
# legitimate hard-kill paths in non-debug trees. # legitimate hard-kill paths in non-debug trees.
debug_protected: bool = ( if (
debug.Lock.ctx_in_debug is not None debug.Lock.ctx_in_debug is not None
or or
_state._runtime_vars.get('_debug_mode', False) _state._runtime_vars.get('_debug_mode', False)
or or
debug_mode_active debug_mode_active
) ):
await portal.cancel_actor()
return
try: try:
cancelled: bool = await portal.cancel_actor( await portal.cancel_actor(raise_on_timeout=True)
raise_on_timeout=not debug_protected,
)
if not cancelled:
if debug_protected:
await debug.maybe_wait_for_debugger(
child_in_debug=(
debug_mode_active
or
debug.Lock.ctx_in_debug is not None
),
header_msg=(
'Delaying subproc hard-reap while '
'debugger locked..\n'
),
)
peer_id: str = portal.channel.aid.reprol()
raise ActorTooSlowError(
f'Peer {peer_id} disconnected before '
f'acknowledging its `Actor.cancel()` RPC'
)
except ActorTooSlowError as too_slow: except ActorTooSlowError as too_slow:
log.error( log.error(
f'Cancel-ack TIMED OUT for sub-actor\n' f'Cancel-ack TIMED OUT for sub-actor\n'
f' uid: {subactor.aid.reprol()!r}\n' f' uid: {subactor.aid.reprol()!r}\n'
f' reason: {too_slow}\n' f' reason: {too_slow}\n'
f'-> escalating to `proc.kill()` (hard-reap)\n' f'-> escalating to `proc.terminate()` (hard-kill)\n'
) )
# XXX, the `subint` backend stores an `int` interp-id in the # XXX, the `subint` backend stores an `int` interp-id in the
# `proc` slot (not a `Process`), so it has no `.kill()`. # `proc` slot (not a `Process`), so it has no `.terminate()`.
# Guard here so a cancel-ack timeout doesn't `AttributeError` # Guard here so a cancel-ack timeout doesn't `AttributeError`
# once that backend lands; its hard-kill path is a TODO. # once that backend lands; its hard-kill path is a TODO.
if hasattr(proc, 'kill'): if hasattr(proc, 'terminate'):
if proc.poll() is None: proc.terminate()
proc.kill()
else: else:
log.error( log.error(
f'Cannot hard-kill sub-actor — backend proc-handle ' f'Cannot hard-kill sub-actor — backend proc-handle '
f'{proc!r} ({type(proc).__name__!r}) has no ' f'{proc!r} ({type(proc).__name__!r}) has no '
f'`.kill()`!\n' f'`.terminate()`!\n'
f' uid: {subactor.aid.reprol()!r}\n' f' uid: {subactor.aid.reprol()!r}\n'
f'TODO: per-backend cancel-escalation.\n' f'TODO: per-backend cancel-escalation.\n'
) )
@ -238,14 +219,6 @@ class ActorNursery:
] = {} ] = {}
self._join_procs = trio.Event() self._join_procs = trio.Event()
self._child_reap_requests: dict[
tuple[str, str],
trio.Event,
] = {}
self._child_reaped: dict[
tuple[str, str],
trio.Event,
] = {}
self._at_least_one_child_in_debug: bool = False self._at_least_one_child_in_debug: bool = False
self.errors = errors self.errors = errors
self._scope_error: BaseException|None = None self._scope_error: BaseException|None = None
@ -308,79 +281,6 @@ class ActorNursery:
# self._cancelled_caught # self._cancelled_caught
) )
def _register_child_reap(
self,
uid: tuple[str, str],
) -> tuple[trio.Event, trio.Event]:
'''
Register a child monitor's process-reap events.
'''
reap_request = trio.Event()
reaped = trio.Event()
self._child_reap_requests[uid] = reap_request
self._child_reaped[uid] = reaped
if self._join_procs.is_set():
reap_request.set()
return reap_request, reaped
def _request_reap_all(self) -> None:
'''
Release every child monitor into its process-join phase.
'''
self._join_procs.set()
for reap_request in tuple(
self._child_reap_requests.values()
):
reap_request.set()
def _mark_child_reaped(
self,
uid: tuple[str, str],
) -> None:
'''
Publish completed child-process teardown to its waiter.
'''
self._children.pop(uid, None)
self._child_reap_requests.pop(uid, None)
reaped: trio.Event|None = self._child_reaped.pop(
uid,
None,
)
if reaped is not None:
reaped.set()
async def _cancel_and_reap_child(
self,
portal: Portal,
) -> None:
'''
Cancel, join and unregister one nursery-owned child.
'''
uid: tuple[str, str] = portal.channel.aid.uid
child_entry = self._children.get(uid)
if child_entry is None:
return
subactor, proc, _ = child_entry
reap_request: trio.Event = self._child_reap_requests[uid]
reaped: trio.Event = self._child_reaped[uid]
with trio.CancelScope(shield=True):
try:
await _try_cancel_then_kill(
portal,
proc,
subactor,
self._at_least_one_child_in_debug,
)
finally:
reap_request.set()
await reaped.wait()
async def start_actor( async def start_actor(
self, self,
name: str, name: str,
@ -425,7 +325,7 @@ class ActorNursery:
# allow setting debug policy per actor # allow setting debug policy per actor
if debug_mode is not None: if debug_mode is not None:
_rtv['_debug_mode'] = debug_mode _rtv['_debug_mode'] = debug_mode
self._at_least_one_child_in_debug |= debug_mode self._at_least_one_child_in_debug = True
enable_modules = list(enable_modules or []) enable_modules = list(enable_modules or [])
proc_kwargs = dict(proc_kwargs or {}) proc_kwargs = dict(proc_kwargs or {})
@ -572,7 +472,7 @@ class ActorNursery:
# TODO: impl a repr for spawn more compact # TODO: impl a repr for spawn more compact
# then `._children`.. # then `._children`..
children: tuple = tuple(self._children.values()) children: dict = self._children
child_count: int = len(children) child_count: int = len(children)
msg: str = f'Cancelling actor nursery with {child_count} children\n' msg: str = f'Cancelling actor nursery with {child_count} children\n'
@ -591,7 +491,7 @@ class ActorNursery:
subactor, subactor,
proc, proc,
portal, portal,
) in children: ) in children.values():
# TODO: are we ever even going to use this or # TODO: are we ever even going to use this or
# is the spawning backend responsible for such # is the spawning backend responsible for such
@ -611,9 +511,7 @@ class ActorNursery:
await event.wait() await event.wait()
# channel/portal should now be up # channel/portal should now be up
_, _, portal = self._children[ _, _, portal = children[subactor.aid.uid]
subactor.aid.uid
]
# XXX should be impossible to get here # XXX should be impossible to get here
# unless method was called from within # unless method was called from within
@ -660,14 +558,14 @@ class ActorNursery:
subactor, subactor,
proc, proc,
portal, portal,
) in children: ) in children.values():
log.warning(f"Hard killing process {proc}") log.warning(f"Hard killing process {proc}")
proc.terminate() proc.terminate()
else: else:
self._cancelled_caught self._cancelled_caught
# mark ourselves as having (tried to have) cancelled all subactors # mark ourselves as having (tried to have) cancelled all subactors
self._request_reap_all() self._join_procs.set()
async def _reap_ria_portals( async def _reap_ria_portals(
@ -725,9 +623,6 @@ async def _open_and_supervise_one_cancels_all_nursery(
# normally don't need to show user by default # normally don't need to show user by default
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
outer_err: BaseException|None = None
inner_err: BaseException|None = None
# 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] = {}
@ -747,156 +642,129 @@ async def _open_and_supervise_one_cancels_all_nursery(
errors errors
) )
try: try:
try: # spawning of actors happens in the caller's scope
# spawning of actors happens in the caller's scope # after we yield upwards
# after we yield upwards yield an
yield an
# When we didn't error in the caller's scope, # When we didn't error in the caller's scope,
# signal all process-monitor-tasks to conduct # signal all process-monitor-tasks to conduct
# the "hard join phase". # the "hard join phase".
log.runtime( log.runtime(
'Waiting on subactors to complete:\n' 'Waiting on subactors to complete:\n'
f'>}} {len(an._children)}\n' f'>}} {len(an._children)}\n'
) )
an._request_reap_all() an._request_reap_all()
# collect results (and errors) from all # collect results (and errors) from all
# `.run_in_actor()` children then cancel # `.run_in_actor()` children then cancel
# each, one reaper task per child. # each, one reaper task per child.
await _reap_ria_portals(an, errors) await _reap_ria_portals(an, errors)
except BaseException as _inner_err: # Single one-cancels-all handler for the (now single)
inner_err = _inner_err # daemon nursery. Pre-#477 a 2ndary `._ria_nursery`
errors[actor.aid.uid] = inner_err # required a separate *outer* handler to catch errors
# bubbling from its task-reaping `__aexit__`; with that
# nursery gone this lone handler covers every scope
# error. NB: we deliberately do NOT re-raise here — the
# `finally` below raises the collected `errors` (as a
# single exc or `BaseExceptionGroup`), which already
# superseded the old outer handler's `raise` anyway
# since `errors` is populated (below) before any await.
except BaseException as _scope_err:
an._scope_error = _scope_err
errors[actor.aid.uid] = _scope_err
# If we error in the root but the debugger is # If we error in the root but the debugger is
# engaged we don't want to prematurely kill (and # engaged we don't want to prematurely kill (and
# thus clobber access to) the local tty since it # thus clobber access to) the local tty since it
# will make the pdb repl unusable. # will make the pdb repl unusable.
# Instead try to wait for pdb to be released before # Instead try to wait for pdb to be released before
# tearing down. # 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
# one-cancels-all supervisor strategy (don't
# worry more are coming).
an._request_reap_all()
# XXX NOTE XXX: hypothetically an error could
# be raised and then a cancel signal shows up
# 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'
f' |_{an}\n\n'
# TODO: show tb str?
# 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: same thing as in
# `._invoke()` to compute how to
# place this div-line in the
# middle of the above msg
# content..
# -[ ] prolly helper-func it too
# in our `.log` module..
# '------ - ------'
)
# 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,
)
# Safety-net handler for anything escaping the inner
# `except BaseException` above (e.g. a `trio.Cancelled`
# during its non-shielded debugger-wait, or a failure
# inside the shielded teardown itself). Post-#477 the
# 2ndary `._ria_nursery` (and its dedicated handler) is
# gone; this now guards the single daemon nursery scope.
# TODO: can this be merged into the inner handler now that
# there's only one nursery? (higher-risk, own PR)
except (
Exception,
BaseExceptionGroup,
trio.Cancelled
) as _outer_err:
outer_err = _outer_err
an._scope_error = outer_err or inner_err
# XXX: yet another guard before allowing the cancel
# sequence in case a (single) child is in debug.
await debug.maybe_wait_for_debugger( await debug.maybe_wait_for_debugger(
child_in_debug=an._at_least_one_child_in_debug child_in_debug=an._at_least_one_child_in_debug
) )
# If actor-local error was raised while waiting on # if the caller's scope errored then we activate our
# ".run_in_actor()" actors then we also want to cancel all # one-cancels-all supervisor strategy (don't
# remaining sub-actors (due to our lone strategy: # worry more are coming).
# one-cancels-all). an._request_reap_all()
if an._children:
log.cancel( # XXX NOTE XXX: hypothetically an error could
'Actor-nursery cancelling due error type:\n' # be raised and then a cancel signal shows up
f'{outer_err}\n' # slightly after in which case the `else:`
) # block here might not complete? For now,
with trio.CancelScope(shield=True): # shield both.
await an.cancel() with trio.CancelScope(shield=True):
raise etype: type = type(_scope_err)
if etype in (
trio.Cancelled,
KeyboardInterrupt,
) or (
is_multi_cancelled(_scope_err)
):
log.cancel(
f'Actor-nursery cancelled by {etype}\n\n'
f'{current_actor().aid.uid}\n'
f' |_{an}\n\n'
# TODO: show tb str?
# f'{tb_str}'
)
elif etype in {
ContextCancelled,
}:
log.cancel(
'Actor-nursery caught remote cancellation\n'
'\n'
f'{_scope_err.tb_str}'
)
else:
log.exception(
'Nursery errored with:\n'
# TODO: same thing as in
# `._invoke()` to compute how to
# place this div-line in the
# middle of the above msg
# content..
# -[ ] prolly helper-func it too
# in our `.log` module..
# '------ - ------'
)
# 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: finally:
# No errors were raised while awaiting ".run_in_actor()" # No errors were raised while awaiting ".run_in_actor()"