Compare commits
29 Commits
481ba00332
...
b645f7fa8c
| Author | SHA1 | Date |
|---|---|---|
|
|
b645f7fa8c | |
|
|
f8488401be | |
|
|
2fdf53dd9f | |
|
|
13b47d07c6 | |
|
|
47bd1a2eb6 | |
|
|
a15c0ccf3e | |
|
|
d737158e2a | |
|
|
52b953fce8 | |
|
|
e26a6160c7 | |
|
|
e468a81b27 | |
|
|
0ac9fa1c0c | |
|
|
321c72e950 | |
|
|
02a3e8de1c | |
|
|
1d0cb0d20c | |
|
|
a9b16d9936 | |
|
|
09341c2014 | |
|
|
e016709041 | |
|
|
040fac3bdb | |
|
|
8bb950ac45 | |
|
|
29057adc89 | |
|
|
55c2f9b1c1 | |
|
|
38fdfb47a0 | |
|
|
19c9eaa32a | |
|
|
c6c6cd0303 | |
|
|
80d86818ee | |
|
|
250396d7ac | |
|
|
e71c6ae588 | |
|
|
19c4b3dbd9 | |
|
|
ed216172ba |
|
|
@ -0,0 +1,56 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-5.6-sol
|
||||||
|
service: opencode
|
||||||
|
session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
|
||||||
|
timestamp: 2026-08-19T18:46:40Z
|
||||||
|
git_ref: 481ba003
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260819T184640Z_481ba003_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Rebase PR #484 onto final PR #481, migrate every affected one-shot call
|
||||||
|
to the new positional target API and continue through downstream tests,
|
||||||
|
examples and documentation review.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
Converted stale target keyword calls to target partials so previously
|
||||||
|
named inputs remain explicit while placement/runtime controls stay
|
||||||
|
direct. Updated error expectations for local signature validation and
|
||||||
|
linked remote error propagation, then corrected docs which still
|
||||||
|
described the removed one-shot implementation. Linked spawning and
|
||||||
|
context lifecycle prose to the corresponding API methods and detailed
|
||||||
|
context guide.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `docs/api/core.rst` - describe linked one-shot context execution.
|
||||||
|
- `docs/guide/rpc.rst` - update placement and target call semantics.
|
||||||
|
- `docs/guide/spawning.rst` - document positional target inputs.
|
||||||
|
- `examples/debugging/multi_nested_subactors_error_up_through_nurseries.py` - migrate nested actor target inputs.
|
||||||
|
- `examples/debugging/root_cancelled_but_child_is_in_tty_lock.py` - preserve named recursive target inputs with partials.
|
||||||
|
- `tests/test_advanced_streaming.py` - migrate streaming target inputs.
|
||||||
|
- `tests/test_cancellation.py` - migrate calls and tighten errors.
|
||||||
|
- `tests/test_infected_asyncio.py` - bind asyncio target options.
|
||||||
|
- `tests/test_rpc.py` - migrate RPC target argument binding.
|
||||||
|
- `tests/test_runtime.py` - preserve named runtime target inputs.
|
||||||
|
- `tests/test_spawning.py` - preserve named spawning target inputs.
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
The human selected the stack order and final PR #481 base, asked the
|
||||||
|
agent to continue after each diagnostic step and required a complete
|
||||||
|
commit plan after independently force-pushing the rebased history.
|
||||||
|
After reviewing the migration, the human required every formerly named
|
||||||
|
target input to remain visibly named through `functools.partial()`
|
||||||
|
rather than becoming positional. These were human-directed agent edits;
|
||||||
|
the human also required plain `start_actor()` and `open_context()`
|
||||||
|
references in the spawning and RPC guides to link to their API methods
|
||||||
|
and the detailed context guide, then clarified that `to_actor.run()`
|
||||||
|
already uses the full context API while `Portal.run()` should share
|
||||||
|
linked lifecycle machinery without necessarily delegating through
|
||||||
|
`Portal.open_context()` or adding a `Started` message. The human made
|
||||||
|
no direct source-line edits.
|
||||||
|
|
@ -0,0 +1,30 @@
|
||||||
|
---
|
||||||
|
model: openai/gpt-5.6-sol
|
||||||
|
service: opencode
|
||||||
|
timestamp: 2026-08-19T18:46:40Z
|
||||||
|
git_ref: 481ba003
|
||||||
|
diff_cmd: git diff HEAD~1..HEAD
|
||||||
|
---
|
||||||
|
|
||||||
|
Migrate PR #484's downstream one-shot calls to PR #481's final
|
||||||
|
`tractor.to_actor.run()` contract after the stack rebase.
|
||||||
|
|
||||||
|
> `git diff HEAD~1..HEAD -- docs examples tests`
|
||||||
|
|
||||||
|
Pass target arguments positionally and bind target keyword-only inputs
|
||||||
|
with `functools.partial()`. Keep placement and runtime controls as
|
||||||
|
direct `to_actor.run()` keywords. Update the invalid-target-argument
|
||||||
|
test to expect local signature binding before actor startup and require
|
||||||
|
direct `RemoteActorError` propagation from linked one-shots.
|
||||||
|
|
||||||
|
Update API and guide prose to describe positional target inputs,
|
||||||
|
linked `Portal.open_context()` execution and per-child reaping instead
|
||||||
|
of the removed `Portal.run()` and target-`**kwargs` conventions.
|
||||||
|
|
||||||
|
Verification:
|
||||||
|
|
||||||
|
- core and migrated runtime batches: `97 passed`
|
||||||
|
- discovery and related lifecycle batch: `33 passed, 1 skipped`
|
||||||
|
- changed executable examples: `9 passed`
|
||||||
|
- mapped debugger cases: `12 passed, 6 skipped`
|
||||||
|
- Ruff, compilation and `git diff --check`: clean
|
||||||
|
|
@ -58,10 +58,10 @@ One-shot task actors
|
||||||
``trio.to_thread.run_sync()`` and friends) is the
|
``trio.to_thread.run_sync()`` and friends) is the
|
||||||
*convenience* one-shot — spawn, run a single task, block on
|
*convenience* one-shot — spawn, run a single task, block on
|
||||||
its result, reap — built entirely on
|
its result, reap — built entirely on
|
||||||
:meth:`ActorNursery.start_actor` + :meth:`Portal.run` +
|
:meth:`ActorNursery.start_actor`, a linked
|
||||||
:meth:`Portal.cancel_actor`, so don't design around it as the
|
:meth:`Portal.open_context` call and per-child cancellation/reaping,
|
||||||
core model. It supersedes the removed (legacy, non-blocking)
|
so don't design around it as the core model. It supersedes the
|
||||||
``ActorNursery.run_in_actor()``.
|
removed (legacy, non-blocking) ``ActorNursery.run_in_actor()``.
|
||||||
|
|
||||||
.. deprecated:: 0.1.0a6
|
.. deprecated:: 0.1.0a6
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -89,7 +89,12 @@ in one blocking call:
|
||||||
|
|
||||||
.. code:: python
|
.. code:: python
|
||||||
|
|
||||||
final = await tractor.to_actor.run(fib, an=an, n=10)
|
from functools import partial
|
||||||
|
|
||||||
|
final = await tractor.to_actor.run(
|
||||||
|
partial(fib, n=10),
|
||||||
|
an=an,
|
||||||
|
)
|
||||||
|
|
||||||
Semantics worth knowing:
|
Semantics worth knowing:
|
||||||
|
|
||||||
|
|
@ -98,9 +103,10 @@ Semantics worth knowing:
|
||||||
task.
|
task.
|
||||||
- "placement" is composable: ``an=`` spawns from an existing
|
- "placement" is composable: ``an=`` spawns from an existing
|
||||||
actor-nursery, ``portal=`` reuses an already-running actor
|
actor-nursery, ``portal=`` reuses an already-running actor
|
||||||
(no spawn/reap, just a ``Portal.run()``), and passing
|
(no spawn/reap, just a linked
|
||||||
neither opens a private call-scoped nursery (booting the
|
:meth:`~tractor.Portal.open_context` call; see the
|
||||||
runtime if needed).
|
:doc:`context guide </guide/context>`), and passing neither
|
||||||
|
opens a private call-scoped nursery (booting the runtime if needed).
|
||||||
- concurrency composes the plain ``trio`` way: schedule
|
- concurrency composes the plain ``trio`` way: schedule
|
||||||
multiple ``run()`` calls into a local task nursery (see
|
multiple ``run()`` calls into a local task nursery (see
|
||||||
``examples/parallelism/to_actor_one_shots.py``).
|
``examples/parallelism/to_actor_one_shots.py``).
|
||||||
|
|
@ -149,7 +155,8 @@ call tears down the entire sub-tree — SC, transitively.
|
||||||
|
|
||||||
When to graduate to ``Context``
|
When to graduate to ``Context``
|
||||||
-------------------------------
|
-------------------------------
|
||||||
``portal.run()`` is great for one-shot, request-response calls.
|
The :meth:`~tractor.Portal.run` method is great for one-shot,
|
||||||
|
request-response calls.
|
||||||
Reach for :meth:`~tractor.Portal.open_context` with an
|
Reach for :meth:`~tractor.Portal.open_context` with an
|
||||||
``@tractor.context`` endpoint as soon as you want:
|
``@tractor.context`` endpoint as soon as you want:
|
||||||
|
|
||||||
|
|
@ -162,10 +169,15 @@ Reach for :meth:`~tractor.Portal.open_context` with an
|
||||||
:meth:`~tractor.Portal.cancel_actor` nukes the **entire**
|
:meth:`~tractor.Portal.cancel_actor` nukes the **entire**
|
||||||
remote runtime and its process.
|
remote runtime and its process.
|
||||||
|
|
||||||
In fact the source plans for ``Portal.run()`` itself to be
|
:func:`tractor.to_actor.run` already enters the full
|
||||||
rebuilt on top of ``open_context()`` — contexts *are* the core
|
:meth:`~tractor.Portal.open_context` lifecycle. The older
|
||||||
inter-actor protocol. Take the full tour in
|
:meth:`~tractor.Portal.run` path instead uses the ``Context`` returned
|
||||||
:doc:`/guide/context`.
|
by the lower-level ``Actor.start_remote_task()`` directly, avoiding a
|
||||||
|
``Started`` handshake but owning less lifecycle machinery. A follow-up
|
||||||
|
should factor their shared linked-task lifecycle without requiring
|
||||||
|
``Portal.run()`` to delegate through the public context API or add
|
||||||
|
another wire message. Take the full tour in
|
||||||
|
:doc:`the context guide </guide/context>`.
|
||||||
|
|
||||||
.. seealso::
|
.. seealso::
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -91,17 +91,17 @@ somebody-ing:
|
||||||
|
|
||||||
What's going on here?
|
What's going on here?
|
||||||
|
|
||||||
- ``start_actor('frank', enable_modules=[__name__])`` forks off
|
- :meth:`~tractor.ActorNursery.start_actor` forks off
|
||||||
a new process, boots a ``tractor`` runtime inside it, and
|
a new process, boots a ``tractor`` runtime inside it, and
|
||||||
allows it to serve functions from the current module (see the
|
allows it to serve functions from the current module (see the
|
||||||
allowlist section below).
|
allowlist section below).
|
||||||
- each ``await portal.run(...)`` schedules a *new* task in
|
- each :meth:`~tractor.Portal.run` call schedules a *new* task in
|
||||||
frank's task tree and waits on its result — the full RPC story
|
frank's task tree and waits on its result — the full RPC story
|
||||||
lives in :doc:`/guide/rpc`.
|
lives in :doc:`/guide/rpc`.
|
||||||
- frank has no main task to complete, so without the final
|
- frank has no main task to complete, so without the final
|
||||||
``await portal.cancel_actor()`` the nursery block would wait
|
:meth:`~tractor.Portal.cancel_actor` call the nursery block would
|
||||||
on him **forever**. Daemon lifetimes are *yours* to end; that
|
wait on him **forever**. Daemon lifetimes are *yours* to end;
|
||||||
explicitness is the point.
|
that explicitness is the point.
|
||||||
|
|
||||||
``to_actor.run()``: quick one-shot parallelism
|
``to_actor.run()``: quick one-shot parallelism
|
||||||
----------------------------------------------
|
----------------------------------------------
|
||||||
|
|
@ -126,7 +126,9 @@ A few details worth knowing:
|
||||||
``name='something_cuter'``.
|
``name='something_cuter'``.
|
||||||
- the function's module is auto-added to the child's
|
- the function's module is auto-added to the child's
|
||||||
``enable_modules`` allowlist.
|
``enable_modules`` allowlist.
|
||||||
- extra ``**kwargs`` are forwarded to the function itself.
|
- target arguments are positional; use ``functools.partial()``
|
||||||
|
to bind target keyword arguments. Keywords passed directly to
|
||||||
|
``run()`` configure actor placement and spawning.
|
||||||
- the call blocks until the result (or error) lands and the
|
- the call blocks until the result (or error) lands and the
|
||||||
child is *auto-cancelled* (reaped) right after — so remote
|
child is *auto-cancelled* (reaped) right after — so remote
|
||||||
errors raise directly in your calling task (causality_ is
|
errors raise directly in your calling task (causality_ is
|
||||||
|
|
@ -138,14 +140,16 @@ A few details worth knowing:
|
||||||
|
|
||||||
.. note::
|
.. note::
|
||||||
|
|
||||||
``to_actor.run()`` is a convenience, **not** the core model —
|
:func:`tractor.to_actor.run` is a convenience, **not** the core
|
||||||
it's built *entirely* on ``start_actor()`` + ``Portal.run()``
|
model — it's built *entirely* on
|
||||||
+ ``Portal.cancel_actor()``. Teach your fingers to use it for
|
:meth:`~tractor.ActorNursery.start_actor` plus a linked
|
||||||
quick fire-and-collect parallelism — think a per-function
|
:meth:`~tractor.Portal.open_context` call and per-child
|
||||||
trio-parallel_ style one-shot — and reach for
|
cancellation/reaping. Teach your fingers to use it for quick
|
||||||
``start_actor()`` + ``open_context()`` for anything
|
fire-and-collect parallelism — think a per-function trio-parallel_
|
||||||
long-lived, stateful or streaming
|
style one-shot — and reach for
|
||||||
(:doc:`/guide/context`).
|
:meth:`~tractor.ActorNursery.start_actor` plus
|
||||||
|
:meth:`~tractor.Portal.open_context` for anything long-lived,
|
||||||
|
stateful or streaming; see :doc:`/guide/context`.
|
||||||
|
|
||||||
Actor lifetimes and teardown order
|
Actor lifetimes and teardown order
|
||||||
----------------------------------
|
----------------------------------
|
||||||
|
|
@ -154,10 +158,11 @@ So we have two lifetime flavors:
|
||||||
- **one-shot** (``to_actor.run()``): lives exactly as long as
|
- **one-shot** (``to_actor.run()``): lives exactly as long as
|
||||||
its single task; reaped the moment its result (or error)
|
its single task; reaped the moment its result (or error)
|
||||||
arrives back in the (blocking) call.
|
arrives back in the (blocking) call.
|
||||||
- **daemon** (``start_actor()``): lives until *someone* cancels
|
- **daemon** (:meth:`~tractor.ActorNursery.start_actor`): lives
|
||||||
it — an explicit ``await portal.cancel_actor()``, a bulk
|
until *someone* cancels it — an explicit
|
||||||
``await an.cancel()``, or the one-cancels-all strategy kicking
|
:meth:`~tractor.Portal.cancel_actor`, a bulk
|
||||||
in on error.
|
:meth:`~tractor.ActorNursery.cancel`, or the one-cancels-all
|
||||||
|
strategy kicking in on error.
|
||||||
|
|
||||||
On a clean exit of the nursery block the teardown order is:
|
On a clean exit of the nursery block the teardown order is:
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -51,9 +51,11 @@ async def spawn_until(depth=0):
|
||||||
# `name_error` relays through.
|
# `name_error` relays through.
|
||||||
depth -= 1
|
depth -= 1
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=depth,
|
depth=depth,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name=f'spawn_until_{depth}',
|
name=f'spawn_until_{depth}',
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -90,18 +92,22 @@ async def main():
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=3,
|
depth=3,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name='spawner0',
|
name='spawner0',
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=4,
|
depth=4,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name='spawner1',
|
name='spawner1',
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -18,9 +18,11 @@ async def spawn_until(depth=0):
|
||||||
else:
|
else:
|
||||||
depth -= 1
|
depth -= 1
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=depth,
|
depth=depth,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name=f'spawn_until_{depth}',
|
name=f'spawn_until_{depth}',
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -51,9 +53,11 @@ async def main():
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=1,
|
depth=1,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name='spawner1',
|
name='spawner1',
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
@ -61,9 +65,11 @@ async def main():
|
||||||
# ..while blocking on the shallow (faster to fail) tree
|
# ..while blocking on the shallow (faster to fail) tree
|
||||||
# whose propagated error triggers nursery cancellation.
|
# whose propagated error triggers nursery cancellation.
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
spawn_until,
|
spawn_until,
|
||||||
an=an,
|
|
||||||
depth=0,
|
depth=0,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name='spawner0',
|
name='spawner0',
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -248,10 +248,12 @@ def test_dynamic_pub_sub(
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
consumer,
|
consumer,
|
||||||
|
subs=[sub],
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
name=f'consumer_{sub}',
|
name=f'consumer_{sub}',
|
||||||
subs=[sub],
|
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -259,10 +261,12 @@ def test_dynamic_pub_sub(
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
consumer,
|
consumer,
|
||||||
|
subs=list(_registry.keys()),
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
name='consumer_dynamic',
|
name='consumer_dynamic',
|
||||||
subs=list(_registry.keys()),
|
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -104,7 +104,7 @@ async def do_nuthin():
|
||||||
[
|
[
|
||||||
# expected to be thrown in assert_err
|
# expected to be thrown in assert_err
|
||||||
({}, AssertionError),
|
({}, AssertionError),
|
||||||
# argument mismatch raised in _invoke()
|
# argument mismatch rejected locally before spawn
|
||||||
({'unexpected': 10}, TypeError)
|
({'unexpected': 10}, TypeError)
|
||||||
],
|
],
|
||||||
ids=['no_args', 'unexpected_args'],
|
ids=['no_args', 'unexpected_args'],
|
||||||
|
|
@ -130,48 +130,34 @@ def test_remote_error(
|
||||||
|
|
||||||
# `to_actor.run()` blocks on the one-shot's result and
|
# `to_actor.run()` blocks on the one-shot's result and
|
||||||
# raises the remote error directly here in the caller's
|
# raises the remote error directly here in the caller's
|
||||||
# task (a bad-arg `TypeError` likewise relays as a
|
# task. Invalid target args fail local signature binding
|
||||||
# `RemoteActorError`).
|
# before any one-shot actor is spawned.
|
||||||
try:
|
try:
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
assert_err,
|
partial(assert_err, **args),
|
||||||
an=an,
|
an=an,
|
||||||
name='errorer',
|
name='errorer',
|
||||||
**args
|
|
||||||
)
|
)
|
||||||
except tractor.RemoteActorError as err:
|
except tractor.RemoteActorError as err:
|
||||||
assert err.boxed_type == errtype
|
assert err.boxed_type == errtype
|
||||||
print("Look Maa that actor failed hard, hehh")
|
print("Look Maa that actor failed hard, hehh")
|
||||||
raise
|
raise
|
||||||
|
|
||||||
# ensure boxed errors
|
# Invalid args never cross the process boundary.
|
||||||
if args:
|
if args:
|
||||||
with pytest.raises(tractor.RemoteActorError) as excinfo:
|
with pytest.raises(errtype):
|
||||||
|
trio.run(main)
|
||||||
|
|
||||||
|
else:
|
||||||
|
# The linked one-shot raises the child's boxed error
|
||||||
|
# directly in this caller task.
|
||||||
|
with pytest.raises(
|
||||||
|
tractor.RemoteActorError,
|
||||||
|
) as excinfo:
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
assert excinfo.value.boxed_type == errtype
|
assert excinfo.value.boxed_type == errtype
|
||||||
|
|
||||||
else:
|
|
||||||
# the root task will also error on the `Portal.result()`
|
|
||||||
# call so we expect an error from there AND the child.
|
|
||||||
# |_ tho seems like on new `trio` this doesn't always
|
|
||||||
# happen?
|
|
||||||
with pytest.raises((
|
|
||||||
BaseExceptionGroup,
|
|
||||||
tractor.RemoteActorError,
|
|
||||||
)) as excinfo:
|
|
||||||
trio.run(main)
|
|
||||||
|
|
||||||
# ensure boxed errors are `errtype`
|
|
||||||
err: BaseException = excinfo.value
|
|
||||||
if isinstance(err, BaseExceptionGroup):
|
|
||||||
suberrs: list[BaseException] = err.exceptions
|
|
||||||
else:
|
|
||||||
suberrs: list[BaseException] = [err]
|
|
||||||
|
|
||||||
for exc in suberrs:
|
|
||||||
assert exc.boxed_type == errtype
|
|
||||||
|
|
||||||
|
|
||||||
def test_multierror(
|
def test_multierror(
|
||||||
reg_addr: tuple[str, int],
|
reg_addr: tuple[str, int],
|
||||||
|
|
@ -391,10 +377,9 @@ async def test_some_cancels_all(
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
func,
|
partial(func, **kwargs),
|
||||||
an=an,
|
an=an,
|
||||||
name=f'actor_{i}',
|
name=f'actor_{i}',
|
||||||
**kwargs,
|
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -469,12 +454,14 @@ async def spawn_and_error(
|
||||||
if depth > 0:
|
if depth > 0:
|
||||||
|
|
||||||
args = (
|
args = (
|
||||||
|
partial(
|
||||||
spawn_and_error,
|
spawn_and_error,
|
||||||
|
breadth=breadth,
|
||||||
|
depth=depth - 1,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
kwargs = {
|
kwargs = {
|
||||||
'name': f'spawner_{i}_depth_{depth}',
|
'name': f'spawner_{i}_depth_{depth}',
|
||||||
'breadth': breadth,
|
|
||||||
'depth': depth - 1,
|
|
||||||
}
|
}
|
||||||
else:
|
else:
|
||||||
args = (
|
args = (
|
||||||
|
|
@ -705,11 +692,13 @@ async def test_nested_multierrors(
|
||||||
tn.start_soon(
|
tn.start_soon(
|
||||||
partial(
|
partial(
|
||||||
tractor.to_actor.run,
|
tractor.to_actor.run,
|
||||||
|
partial(
|
||||||
spawn_and_error,
|
spawn_and_error,
|
||||||
an=an,
|
|
||||||
name=f'spawner_{i}',
|
|
||||||
breadth=subactor_breadth,
|
breadth=subactor_breadth,
|
||||||
depth=depth,
|
depth=depth,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
|
name=f'spawner_{i}',
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
except (
|
except (
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,7 @@ The hipster way to force SC onto the stdlib's "async": 'infection mode'.
|
||||||
import asyncio
|
import asyncio
|
||||||
import builtins
|
import builtins
|
||||||
from contextlib import ExitStack
|
from contextlib import ExitStack
|
||||||
# from functools import partial
|
from functools import partial
|
||||||
import itertools
|
import itertools
|
||||||
import importlib
|
import importlib
|
||||||
import os
|
import os
|
||||||
|
|
@ -209,10 +209,12 @@ def test_aio_simple_error(
|
||||||
debug_mode=debug_mode,
|
debug_mode=debug_mode,
|
||||||
) as an:
|
) as an:
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
|
partial(
|
||||||
asyncio_actor,
|
asyncio_actor,
|
||||||
an=an,
|
|
||||||
target='sleep_and_err',
|
target='sleep_and_err',
|
||||||
expect_err='AssertionError',
|
expect_err='AssertionError',
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -455,10 +457,14 @@ def test_aio_cancelled_from_aio_causes_trio_cancelled(
|
||||||
# relays the remote error here in the caller's task.
|
# relays the remote error here in the caller's task.
|
||||||
with trio.fail_after(1 + delay):
|
with trio.fail_after(1 + delay):
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
|
partial(
|
||||||
asyncio_actor,
|
asyncio_actor,
|
||||||
an=an,
|
|
||||||
target='aio_cancel',
|
target='aio_cancel',
|
||||||
expect_err='tractor.to_asyncio.AsyncioCancelled',
|
expect_err=(
|
||||||
|
'tractor.to_asyncio.AsyncioCancelled'
|
||||||
|
),
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -663,10 +669,12 @@ def test_basic_interloop_channel_stream(
|
||||||
) as an:
|
) as an:
|
||||||
# should raise RAE diectly
|
# should raise RAE diectly
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
|
fan_out=fan_out,
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
fan_out=fan_out,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
@ -682,9 +690,11 @@ def test_trio_error_cancels_intertask_chan(
|
||||||
) as an:
|
) as an:
|
||||||
# should trigger remote actor error
|
# should trigger remote actor error
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
an=an,
|
|
||||||
trio_raise_err=True,
|
trio_raise_err=True,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -719,9 +729,11 @@ def test_trio_closes_early_causes_aio_checkpoint_raise(
|
||||||
# should raise RAE diectly
|
# should raise RAE diectly
|
||||||
print('waiting on final infected subactor result..')
|
print('waiting on final infected subactor result..')
|
||||||
res: None = await to_actor.run(
|
res: None = await to_actor.run(
|
||||||
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
an=an,
|
|
||||||
trio_exit_early=True,
|
trio_exit_early=True,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
assert res is None
|
assert res is None
|
||||||
|
|
@ -770,11 +782,13 @@ def test_aio_exits_early_relays_AsyncioTaskExited(
|
||||||
# should raise RAE diectly
|
# should raise RAE diectly
|
||||||
print('waiting on final infected subactor result..')
|
print('waiting on final infected subactor result..')
|
||||||
res: None = await to_actor.run(
|
res: None = await to_actor.run(
|
||||||
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
an=an,
|
|
||||||
infect_asyncio=True,
|
|
||||||
trio_exit_early=False,
|
trio_exit_early=False,
|
||||||
aio_exit_early=True,
|
aio_exit_early=True,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
assert res is None
|
assert res is None
|
||||||
print(f'infected subactor returned result: {res!r}\n')
|
print(f'infected subactor returned result: {res!r}\n')
|
||||||
|
|
@ -809,9 +823,11 @@ def test_aio_errors_and_channel_propagates_and_closes(
|
||||||
) as an:
|
) as an:
|
||||||
# should trigger RAE directly, not an eg.
|
# should trigger RAE directly, not an eg.
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
an=an,
|
|
||||||
aio_raise_err=True,
|
aio_raise_err=True,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
infect_asyncio=True,
|
infect_asyncio=True,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,7 @@ related API and error checks.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
import itertools
|
import itertools
|
||||||
|
from functools import partial
|
||||||
from unittest.mock import (
|
from unittest.mock import (
|
||||||
AsyncMock,
|
AsyncMock,
|
||||||
Mock,
|
Mock,
|
||||||
|
|
@ -239,19 +240,21 @@ def test_rpc_errors(
|
||||||
actor = tractor.current_actor()
|
actor = tractor.current_actor()
|
||||||
assert actor.is_registrar
|
assert actor.is_registrar
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
sleep_back_actor,
|
sleep_back_actor,
|
||||||
an=an,
|
|
||||||
actor_name=subactor_requests_to,
|
actor_name=subactor_requests_to,
|
||||||
|
func_name=funcname,
|
||||||
|
func_defined=bool(func_defined),
|
||||||
|
exposed_mods=exposed_mods,
|
||||||
|
reg_addr=reg_addr,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
|
|
||||||
name='subactor',
|
name='subactor',
|
||||||
|
|
||||||
# function from the local exposed module space
|
# Function from the local exposed module space the
|
||||||
# the subactor will invoke when it RPCs back to this actor
|
# subactor invokes when it RPCs back to this actor.
|
||||||
func_name=funcname,
|
|
||||||
exposed_mods=exposed_mods,
|
|
||||||
func_defined=True if func_defined else False,
|
|
||||||
enable_modules=subactor_exposed_mods,
|
enable_modules=subactor_exposed_mods,
|
||||||
reg_addr=reg_addr,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def run():
|
def run():
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@
|
||||||
Verifying internal runtime state and undocumented extras.
|
Verifying internal runtime state and undocumented extras.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
|
from functools import partial
|
||||||
import os
|
import os
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -84,10 +85,12 @@ async def test_lifetime_stack_wipes_tmpfile(
|
||||||
loglevel=loglevel,
|
loglevel=loglevel,
|
||||||
) as an:
|
) as an:
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
crash_and_clean_tmpdir,
|
crash_and_clean_tmpdir,
|
||||||
an=an,
|
|
||||||
tmp_file_path=path,
|
tmp_file_path=path,
|
||||||
error=error_in_child,
|
error=error_in_child,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
)
|
)
|
||||||
except (
|
except (
|
||||||
tractor.RemoteActorError,
|
tractor.RemoteActorError,
|
||||||
|
|
|
||||||
|
|
@ -51,18 +51,18 @@ async def spawn(
|
||||||
# recursively spawn this same `spawn()` fn as the lone
|
# recursively spawn this same `spawn()` fn as the lone
|
||||||
# task of a one-shot child subactor and get its result.
|
# task of a one-shot child subactor and get its result.
|
||||||
result = await tractor.to_actor.run(
|
result = await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
spawn,
|
spawn,
|
||||||
|
should_be_root=False,
|
||||||
|
data=data_to_pass_down,
|
||||||
|
reg_addr=reg_addr,
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
|
|
||||||
# spawning args
|
# spawning args
|
||||||
name='sub-actor',
|
name='sub-actor',
|
||||||
enable_modules=[__name__],
|
enable_modules=[__name__],
|
||||||
|
|
||||||
# passed to a subactor-recursive RPC invoke
|
|
||||||
# of this same `spawn()` fn.
|
|
||||||
should_be_root=False,
|
|
||||||
data=data_to_pass_down,
|
|
||||||
reg_addr=reg_addr,
|
|
||||||
)
|
)
|
||||||
assert result == 10
|
assert result == 10
|
||||||
return result
|
return result
|
||||||
|
|
@ -152,9 +152,11 @@ async def test_most_beautiful_word(
|
||||||
debug_mode=debug_mode,
|
debug_mode=debug_mode,
|
||||||
) as an:
|
) as an:
|
||||||
res: Any = await tractor.to_actor.run(
|
res: Any = await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
cellar_door,
|
cellar_door,
|
||||||
an=an,
|
|
||||||
return_value=return_value,
|
return_value=return_value,
|
||||||
|
),
|
||||||
|
an=an,
|
||||||
name='some_linguist',
|
name='some_linguist',
|
||||||
)
|
)
|
||||||
assert res == return_value
|
assert res == return_value
|
||||||
|
|
@ -204,10 +206,12 @@ def test_loglevel_propagated_to_subactor(
|
||||||
|
|
||||||
) as an:
|
) as an:
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
check_loglevel,
|
check_loglevel,
|
||||||
|
level=level,
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
loglevel=level,
|
loglevel=level,
|
||||||
level=level,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
@ -273,19 +277,23 @@ def test_to_actor_run_can_skip_parent_main_inheritance(
|
||||||
|
|
||||||
# Default: child receives parent __main__ bootstrap data
|
# Default: child receives parent __main__ bootstrap data
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
check_parent_main_inheritance,
|
check_parent_main_inheritance,
|
||||||
|
expect_inherited=True,
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
name='replaying-parent-main',
|
name='replaying-parent-main',
|
||||||
expect_inherited=True,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
# Opt-out: child gets no parent __main__ data
|
# Opt-out: child gets no parent __main__ data
|
||||||
await tractor.to_actor.run(
|
await tractor.to_actor.run(
|
||||||
|
partial(
|
||||||
check_parent_main_inheritance,
|
check_parent_main_inheritance,
|
||||||
|
expect_inherited=False,
|
||||||
|
),
|
||||||
an=an,
|
an=an,
|
||||||
name='isolated-parent-main',
|
name='isolated-parent-main',
|
||||||
inherit_parent_main=False,
|
inherit_parent_main=False,
|
||||||
expect_inherited=False,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
|
||||||
|
|
@ -89,7 +89,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.terminate()` (SIGTERM) per SC-discipline:
|
we escalate via `proc.kill()` per SC-discipline:
|
||||||
|
|
||||||
graceful cancel-req -> bounded wait -> hard-kill
|
graceful cancel-req -> bounded wait -> hard-kill
|
||||||
|
|
||||||
|
|
@ -101,11 +101,9 @@ async def _try_cancel_then_kill(
|
||||||
the wider write-up.
|
the wider write-up.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# XXX, do NOT escalate to `proc.terminate()` while ANY of
|
# XXX, delay hard-kill escalation while any debugger guard
|
||||||
# the following are true — SIGTERM-ing a sub would tear
|
# below is active. Killing the sub immediately would tear down
|
||||||
# down its sub-tree including any descendant proxying
|
# its tree and clobber an actor proxying a REPL session:
|
||||||
# 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
|
||||||
|
|
@ -121,7 +119,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=` arg
|
# child started with an explicit `debug_mode=True` 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.
|
||||||
|
|
@ -131,36 +129,57 @@ 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.
|
||||||
if (
|
debug_protected: bool = (
|
||||||
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:
|
||||||
await portal.cancel_actor(raise_on_timeout=True)
|
cancelled: bool = await portal.cancel_actor(
|
||||||
|
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.terminate()` (hard-kill)\n'
|
f'-> escalating to `proc.kill()` (hard-reap)\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 `.terminate()`.
|
# `proc` slot (not a `Process`), so it has no `.kill()`.
|
||||||
# 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, 'terminate'):
|
if hasattr(proc, 'kill'):
|
||||||
proc.terminate()
|
if proc.poll() is None:
|
||||||
|
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'`.terminate()`!\n'
|
f'`.kill()`!\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'
|
||||||
)
|
)
|
||||||
|
|
@ -218,6 +237,14 @@ 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
|
||||||
|
|
@ -272,6 +299,79 @@ 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,
|
||||||
|
|
@ -316,7 +416,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 = True
|
self._at_least_one_child_in_debug |= debug_mode
|
||||||
|
|
||||||
enable_modules = list(enable_modules or [])
|
enable_modules = list(enable_modules or [])
|
||||||
proc_kwargs = dict(proc_kwargs or {})
|
proc_kwargs = dict(proc_kwargs or {})
|
||||||
|
|
@ -385,7 +485,7 @@ class ActorNursery:
|
||||||
|
|
||||||
# TODO: impl a repr for spawn more compact
|
# TODO: impl a repr for spawn more compact
|
||||||
# then `._children`..
|
# then `._children`..
|
||||||
children: dict = self._children
|
children: tuple = tuple(self._children.values())
|
||||||
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'
|
||||||
|
|
||||||
|
|
@ -404,7 +504,7 @@ class ActorNursery:
|
||||||
subactor,
|
subactor,
|
||||||
proc,
|
proc,
|
||||||
portal,
|
portal,
|
||||||
) in children.values():
|
) in children:
|
||||||
|
|
||||||
# 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
|
||||||
|
|
@ -424,7 +524,9 @@ class ActorNursery:
|
||||||
await event.wait()
|
await event.wait()
|
||||||
|
|
||||||
# channel/portal should now be up
|
# channel/portal should now be up
|
||||||
_, _, portal = children[subactor.aid.uid]
|
_, _, portal = self._children[
|
||||||
|
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
|
||||||
|
|
@ -471,14 +573,14 @@ class ActorNursery:
|
||||||
subactor,
|
subactor,
|
||||||
proc,
|
proc,
|
||||||
portal,
|
portal,
|
||||||
) in children.values():
|
) in children:
|
||||||
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._join_procs.set()
|
self._request_reap_all()
|
||||||
|
|
||||||
|
|
||||||
@acm
|
@acm
|
||||||
|
|
@ -646,15 +748,13 @@ async def open_nursery(
|
||||||
Create and yield a new ``ActorNursery`` to be used for spawning
|
Create and yield a new ``ActorNursery`` to be used for spawning
|
||||||
structured concurrent subactors.
|
structured concurrent subactors.
|
||||||
|
|
||||||
When an actor is spawned a new trio task is started which
|
When an actor is spawned a new trio task invokes one of the
|
||||||
invokes one of the process spawning backends to create and start
|
process spawning backends to create and start a new subprocess.
|
||||||
a new subprocess. These tasks are started by one of two nurseries
|
These tasks are started in the supervisor's process nursery.
|
||||||
detailed below. The reason for spawning processes from within
|
Spawning from a task is required because ``trio_run_in_process``
|
||||||
a new task is because ``trio_run_in_process`` itself creates a new
|
creates an internal nursery which the opening task **must** close;
|
||||||
internal nursery and the same task that opens a nursery **must**
|
this also makes each task's cancellation scope correspond to its
|
||||||
close it. It turns out this approach is probably more correct
|
spawned subactor.
|
||||||
anyway since it is more clear from the following nested nurseries
|
|
||||||
which cancellation scopes correspond to each spawned subactor set.
|
|
||||||
|
|
||||||
'''
|
'''
|
||||||
__tracebackhide__: bool = hide_tb
|
__tracebackhide__: bool = hide_tb
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue