Compare commits

..

11 Commits

Author SHA1 Message Date
Gud Boi 2a59cefbe8 Remove `run_in_actor()` + the ria reap cluster
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
2026-07-06 12:52:29 -04:00
Gud Boi 07e1669e6d Fix stale `@pub` docstring example in `experimental`
The `_pubsub.pub` decorator's usage example predates several API
generations: ancient positional-arg-order `run_in_actor()` (a
missing `await` too) plus the deprecated `portal.result()` — and
`run_in_actor()` never allowed streaming funcs anyway. Show the
canonical `start_actor()` + `Portal.open_stream_from()`
consumption instead (#477 removal sweep).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 12:31:27 -04:00
Gud Boi d6bed7c4a0 Port docs off `run_in_actor` + `Portal.wait_for_result`
The 8-page docs sweep of the #477 removal, ahead of the API's
excision,

- `start/quickstart.rst`: the first-actor-tree walkthrough now
  narrates the (migrated) `to_actor.run()` example — no portal
  in hand until the daemon section introduces `start_actor()`.
- `guide/spawning.rst`: the one-shot section becomes
  `to_actor.run()` (blocking call, placement opts, "built on the
  primitives" note); lifetime/teardown rules update — one-shots
  never make it to nursery exit since each is reaped inside its
  own call.
- `guide/rpc.rst`: the `wait_for_result()` section (an API that
  dies with the reap cluster, incl. the `NoResult` sentinel)
  becomes a `to_actor.run()` one-shot section.
- `api/core.rst`: drop `run_in_actor`/`wait_for_result` from the
  autodoc member lists, drop the `Portal.result()` deprecation
  note, add a "One-shot task actors" `tractor.to_actor.run`
  autodoc section.
- `guide/{asyncio,context,cancellation,parallelism}.rst`:
  mention swaps to the successor API.

Gate: `make -C docs html` builds clean; `to_actor.run` autodoc
renders in `api/core.html`.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 12:31:02 -04:00
Gud Boi a3057cb24f Port debugging examples off `run_in_actor`
The 8 `examples/debugging/` scripts driven by the pexpect'd
`test_debugger.py` REPL-flows (#477 removal),

- blocking one-shots (`subactor_error`, `subactor_breakpoint`,
  `shielded_pause`): straight `to_actor.run(fn, an=an)` — the
  boxed error/`BdbQuit` raises in the root's task.
- `multi_subactors`: introduces the "collect all errors" pattern
  — each one-shot catches + stashes its `RemoteActorError` (vs
  raising) so no child's crash cancels its siblings before
  they've had their own REPL sessions, then a
  `BaseExceptionGroup` of the lot raises at the end; preserves
  the legacy teardown-reap REPL flow exactly (28/28 debugger
  suite unchanged).
- `multi_nested_subactors_error_up_through_nurseries` +
  `root_cancelled_but_child_is_in_tty_lock`: recursive
  `spawn_until` levels each block on their one-shot child; the
  parallel spawner-trees run as bg task-nursery one-shots where
  the first tree's error cancels the other.
- `multi_subactor_root_errors` +
  `root_timeout_while_child_crashed`: `start_actor()` + bg
  `Portal.run()` tasks so the root's own error/timeout races the
  already-crashed children, same as before.
- `sync_bp`: TODO-comment x-ref update only.

`test_debugger.py`: the nested-nurseries test's final-output
patterns update to the new relay shape — the LAST-released leaf
REPL's error chain wins each level's relay-vs-cancel race and
relays as a `collapse_eg()`-annotated collapsed chain, while the
sibling tree is cancelled + absorbed. (The legacy teardown-reap
grouped BOTH the `name_error` and bp-quit chains — explaining
the previously-mysterious "extra" `src_uid`/`relay_uid` patterns
noted in the old TODO.)

Gate: `tests/devx/test_debugger.py` = 28 passed, 6 skipped —
identical to the pre-migration baseline.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 12:24:29 -04:00
Gud Boi d8af5f125a Port non-debugging examples off `run_in_actor`
4 example scripts of the #477 removal sweep, each exercised by
`test_docs_examples.py`,

- `actor_spawning_and_causality.py`: the simplest possible
  `to_actor.run()` demo — private call-scoped nursery, block on
  and print the one-shot's result.
- `remote_error_propagation.py`: blocking `to_actor.run(an=n)`
  raises the boxed `AssertionError` in the caller's task,
  cancelling the sibling daemons.
- `parallelism/single_func.py`: bg-burn a core in the parent via
  a local task-nursery while the one-shot burns (and returns
  from) a subactor.
- `a_trynamic_first_scene.py`: donny + gretchen wait on each
  *other* so their one-shots run concurrently in a local
  task-nursery against a shared `an` (mirrors the migrated
  `test_trynamic_trio`).

Gate: all 4 green via the example-runner suite.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 12:11:39 -04:00
Gud Boi cb6202e3cb Port `test_dynamic_pub_sub` off `run_in_actor`
The known-flaky dynamic pubsub test's 3 fire-and-forget spawn
sites (#477 removal),

- the forever-streaming `publisher` + N `consumer` one-shots now
  bg-schedule as `to_actor.run(fn, an=n)` tasks in a local `trio`
  task-nursery (`publisher`'s rendezvous name still derives from
  `fn.__name__`).
- the simulated user-cancel raise (`KeyboardInterrupt` /
  `TooSlowError` params) cancels the task-nursery, each one-shot
  reaping its subactor via `to_actor.run()`'s shielded
  `Portal.cancel_actor()`; `_run_and_match()`'s existing
  `BaseExceptionGroup.split()` walk covers the (possibly nested)
  relay shapes unchanged.
- spawns now issue concurrently rather than sequentially —
  comment on the fork-backend budget updated to match.

Gate: both params x4 runs green on `trio` + x1 on `mp_spawn`;
full module green.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 12:06:10 -04:00
Gud Boi f11754cecf Port SIGINT + sync-sleep cancel tests off `run_in_actor`
Final `test_cancellation.py` group of the `run_in_actor` removal
(#477) — cancel-mechanics tests, so clean conversions,

- `test_cancel_via_SIGINT_other_task`: the 3 keep-alive
  `run_in_actor(sleep_forever)` one-shots become plain
  `start_actor()` daemons (an idle daemon needs no "main" task,
  and no longer shares a single dup'd `namesucka` name).
- `spawn_sub_with_sync_blocking_task`: the middle layer's spawn
  becomes a blocking `to_actor.run(spin_for, an=an)` which parks
  awaiting the sync-sleeping grandchild's result until cancelled
  from above.
- `test_cancel_while_childs_child_in_sync_sleep`: the
  fire-and-forget middle-actor spawn becomes a bg
  `to_actor.run()` task in a local task-nursery; the root's
  `assert 0` cancels it, driving the same
  graceful-cancel-then-zombie-reap cascade on the sync-blocked
  grandchild. The `man_cancel_outer` xfail param is unchanged.

Zero live `run_in_actor()` call-sites remain in this suite.

Gate: full `test_cancellation.py` module green on both `trio`
(18p/1xf) + `mp_spawn` (18p/1xf).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 11:59:38 -04:00
Gud Boi fa8799d5bb Port `test_nested_multierrors` off `run_in_actor`
Third `test_cancellation.py` group of the `run_in_actor` removal
(#477),

- `spawn_and_error` fans out each level's erroring one-shots as
  concurrent `to_actor.run(fn, an=an)` tasks in a local `trio`
  task-nursery (recursing per spawner subactor), as does the
  test-body's top-level spawner loop.
- the deterministic exact-breadth nested-BEG shape dies with the
  legacy teardown-reap: each level now groups whatever subset of
  sub-tree errors relay before the first one's cancel wins, and
  a single-member group gets unwrapped by the runtime's own
  `collapse_eg()` at every actor boundary — so a fully-raced
  tree relays a bare `RemoteActorError` chain.
- loosen the shape walk accordingly: accept a lone
  `RemoteActorError` or a 1..breadth group whose members box
  `ExceptionGroup` (multi-relay), `AssertionError` (collapsed
  leaf chain), `RemoteActorError` (re-boxed collapsed chain) or
  `BaseExceptionGroup` (runtime reap-deadline `Cancelled`
  upgrade); fold the windows-only tolerances into the same walk.
- raced sibling `trio.Cancelled`s are now ABSORBED by the
  task-nursery instead of landing in the group, so the MTF
  shape-mismatch xfail should consistently xpass — note added to
  drop the marker once CI confirms.
- add an `else: pytest.fail()` so a silently-clean tree can no
  longer pass.

Gate: both depths green on `trio` (10 consecutive runs) +
`mp_spawn`.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 11:47:47 -04:00
Gud Boi 697c6152a6 Fix unbound `timeout` under non-trio/MTF backends
`test_nested_multierrors`'s backend/depth budget `match` only
carries arms for the `trio` + `main_thread_forkserver` spawn
backends, so running under any other (e.g. `mp_spawn`) leaves
`timeout` unbound and crashes with an `UnboundLocalError` at the
headroom-scaling below. Add default per-depth arms riding the MTF
budgets (same per-spawn round-trip cost class).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 11:40:39 -04:00
Gud Boi d01a21239c Port `test_some_cancels_all` off `run_in_actor`
Second `test_cancellation.py` group of the `run_in_actor` removal
(#477),

- one-shot subactors now run as concurrent `to_actor.run(fn,
  an=an)` tasks in a local `trio` task-nursery, so their errors
  raise WHILE the actor-nursery block is open (vs the legacy
  teardown-reap) and the first error cancels sibling one-shots.
- wrap the task-nursery in `collapse_eg()` so the deterministic
  single-error cases still surface a bare `RemoteActorError`.
- loosen the group-shape assertion: the relay-vs-cancel race
  populates anywhere from 1 to `num_actors` `RemoteActorError`s
  (the exact-`num_actors` BEG was `run_in_actor`'s
  reap-all-at-teardown); group members are always
  `RemoteActorError` now since sibling `trio.Cancelled`s are
  absorbed by the task-nursery.
- move the daemon-portal call loop inside the task-nursery body
  so the sleep-forever one-shot case is cancelled by the daemon
  error raise.
- rename the `*run_in_actor*` param ids to `*one_shot*`.

Gate: 6 passed on both `trio` + `mp_spawn` backends.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
2026-07-06 11:18:11 -04:00
Gud Boi ff3474728c Doc ria-reap hang fix + paused reaper re-scope
Append two sections to the ria-removal plan capturing the
2026-07-02 hang episode + the resulting design pivot.

Regression writeup: the full-suite hang on
`test_tractor_cancels_aio` root-caused to the step-A reaper
hoist (`5cd190c5`), not the B2 handler merge. The happy-path
`_reap_ria_portals()` parks unbounded on `wait_for_result()`
after a user `portal.cancel_actor()`; the old spawn-backend
reaper raced `soft_kill()`'s scope-cancel, the hoist dropped
it. Records the `proc.poll()` death-watch fix + why poll (not
the event `wait_func`) bc `soft_kill` already awaits
`proc.sentinel` (a 2nd `wait_readable` -> `BusyResourceError`).

Pause writeup: user's insight that the hoist landed in the
wrong scope — result-waiting belongs in the `to_actor`
one-shot scope (`_invoke_in_subactor()`), beside `an` + a
local task-nursery + cancel-scope, where bounding the wait is
trivial + the hang dissolves. So the poll fix is likely
SUPERSEDED (flagged do-not-land); the anti-hang guard commit
(`d1fb4a1a`) stays red-first per the failing-test convention.

(this patch was generated in some part by `claude-code` using
`claude-opus-4-8` (`anthropic`))
2026-07-06 10:59:20 -04:00
38 changed files with 690 additions and 781 deletions

View File

@ -276,6 +276,97 @@ Gate (box ran ~2.7x slow this session, load-induced
completion). RECOMMEND a clean full-suite run on a completion). RECOMMEND a clean full-suite run on a
normal-speed box before this merges. normal-speed box before this merges.
## Regression + fix: ria-reap hang (2026-07-02)
Human hit a full-suite hang on
`test_infected_asyncio.py::test_tractor_cancels_aio`. Bisected:
passes at pre-ria `a34aaf98` (0.59s), hangs at B2 `e617b498`
(90s+). Root-caused to the STEP-A reaper hoist (`5cd190c5`),
NOT B2 (`_reap_ria_portals` is byte-identical A->B2).
Bug: the test does `run_in_actor(asyncio_actor)` then a USER
`portal.cancel_actor()` and exits the block cleanly -> the
happy path's `await _reap_ria_portals()`, which waits UNBOUNDED
on `cancel_on_completion -> wait_for_result()`. The child was
cancelled out-of-band so no final result is relayed -> parked
forever. The OLD spawn-backend reaper was raced against
`soft_kill()` (per-child nursery `cancel_scope.cancel()` on
subproc death); the hoist dropped that race.
Fix: `_reap_ria_portals()` runs each `cancel_on_completion()`
in a local nursery alongside a `proc.poll()` death-watch that
cancels the parked reaper once the subproc exits — restoring
the old race, backend-agnostic (guarded by
`hasattr(proc, 'poll')` for a future `subint` handle).
Why POLL (`proc.poll()`) not the event-driven `wait_func`:
the mp waiter (`_spawn.proc_waiter`) does
`wait_readable(proc.sentinel)`, and `soft_kill()` is ALREADY
awaiting that same fd concurrently in the daemon nursery — a
2nd `wait_readable` on one fd raises `trio.BusyResourceError`.
(`trio.Process.wait()` IS multi-waiter-safe, but mp has no
async equivalent.) `proc.poll()` — the same liveness check
`soft_kill` itself falls back to — is the conflict-free common
denominator. Verified: poll-fix passes on BOTH trio and
mp_spawn.
Also added a per-test anti-hang guard: wrapped
`test_tractor_cancels_aio`'s `main()` in
`with trio.fail_after(9 * cpu_perf_headroom())` — the blessed
pattern (`pytest-timeout`'s global cap is intentionally off;
breaks trio under fork backends, see `pyproject` NOTE). So a
future recurrence FAILS FAST instead of hanging the suite.
(Several other tests in the file are still guardless —
`test_aio_simple_error`, `test_trio_error_cancels_intertask_chan`,
`test_aio_errors_and_channel_propagates_and_closes` — candidate
follow-up sweep.)
Lesson: the B2 focused gate OMITTED `test_infected_asyncio`
(and the full runs were clipped/slow), so the step-A hang
slipped through. Any future ria-touching change MUST gate
`test_infected_asyncio` explicitly.
Gate: `test_tractor_cancels_aio` green (trio 1.53s, mp 3.98s);
fix gate (`test_infected_asyncio test_cancellation test_to_actor
test_spawning`) = 74 passed, 3 xfailed, 0 failures.
## PAUSED (2026-07-02): re-assess the reaper's SCOPE
User's insight (compelling — likely the real root cause of
the hang, not just the missing proc-death race):
> the "hoisting" of 5cd190c5 was just not really done right
> — the hoist should have been into the `to_actor` scope,
> not `_supervise`.
The argument: `.run_in_actor()`'s result-waiting/reaping got
hoisted into `_supervise._reap_ria_portals` (nursery-machinery
scope), which has NO natural cancel-scope to bound a parked
`wait_for_result()` — hence the awkward proc-death race +
the poll-vs-`proc_waiter` dilemma. If the result-wait instead
lived in the `to_actor` one-shot scope
(`to_actor._invoke_in_subactor()`), it would sit right next to
the caller's `an` + a local `trio` task-nursery + cancel-scope
(the `trio.to_thread`-style model #477 actually wants) — so
bounding/cancelling the wait is trivial and the hang
dissolves from correct scoping rather than a bolt-on race.
Follow-on to re-evaluate on resume:
- should `_reap_ria_portals` exist AT ALL, or should
result-waiting move entirely into
`to_actor._invoke_in_subactor()`?
- reimplement legacy `run_in_actor()` on top of
`to_actor.run()` so `_reap_ria_portals` +
`_cancel_after_result_on_exit` can be DROPPED from
`_supervise` entirely (the true #477 simplification)?
- the poll-vs-event decision is MOOT under this re-scoping.
State at pause: `test_infected_asyncio` anti-hang guard
COMMITTED (`d1fb4a1a`, intentionally red w/o the fix — the
user's failing-test-first convention). The poll-based reap
fix in `_supervise.py` is UNCOMMITTED and likely SUPERSEDED
by the re-scoping — do NOT land it as-is.
## Verification gate ## Verification gate
- `tests/test_cancellation.py test_spawning.py test_local.py - `tests/test_cancellation.py test_spawning.py test_local.py

View File

@ -37,7 +37,6 @@ Spawning actors
.. autoclass:: ActorNursery .. autoclass:: ActorNursery
:members: start_actor, :members: start_actor,
run_in_actor,
cancel, cancel,
cancel_called, cancel_called,
cancelled_caught cancelled_caught
@ -47,10 +46,22 @@ Spawning actors
:meth:`ActorNursery.start_actor` (daemon actor + portal) is the :meth:`ActorNursery.start_actor` (daemon actor + portal) is the
blessed spawning primitive; pair it with blessed spawning primitive; pair it with
``Portal.open_context()`` for SC-linked remote tasks. ``Portal.open_context()`` for SC-linked remote tasks.
:meth:`ActorNursery.run_in_actor` is a *convenience* one-shot —
spawn, run a single task, auto-cancel after the result — slated One-shot task actors
to be rebuilt as a high-level wrapper, so don't design around --------------------
it as the core model.
.. autofunction:: tractor.to_actor.run
.. note::
:func:`tractor.to_actor.run` (parlance of
``trio.to_thread.run_sync()`` and friends) is the
*convenience* one-shot — spawn, run a single task, block on
its result, reap — built entirely on
:meth:`ActorNursery.start_actor` + :meth:`Portal.run` +
:meth:`Portal.cancel_actor`, so don't design around it as the
core model. It supersedes the removed (legacy, non-blocking)
``ActorNursery.run_in_actor()``.
.. deprecated:: 0.1.0a6 .. deprecated:: 0.1.0a6
@ -71,14 +82,12 @@ flowing back `exactly like trio`_.
:members: run, :members: run,
run_from_ns, run_from_ns,
open_stream_from, open_stream_from,
wait_for_result,
cancel_actor, cancel_actor,
chan chan
.. deprecated:: 0.1.0a6 .. deprecated:: 0.1.0a6
``Portal.result()`` warns; use :meth:`Portal.wait_for_result`. The str-form ``Portal.run('mod.path', 'fn_name')`` warns;
The str-form ``Portal.run('mod.path', 'fn_name')`` also warns;
pass a function *object* whose module is listed in the target's pass a function *object* whose module is listed in the target's
``enable_modules``. ``Portal.channel`` is the legacy spelling ``enable_modules``. ``Portal.channel`` is the legacy spelling
of :attr:`Portal.chan`. of :attr:`Portal.chan`.

View File

@ -76,8 +76,8 @@ Just flip the flag on :meth:`tractor.ActorNursery.start_actor`:
infect_asyncio=True, infect_asyncio=True,
) )
The one-shot convenience ``ActorNursery.run_in_actor()`` accepts The one-shot convenience ``tractor.to_actor.run()`` accepts the
the same flag. The ``to_asyncio`` APIs may **only** be called from same flag. The ``to_asyncio`` APIs may **only** be called from
tasks inside an infected actor; calling them anywhere else raises tasks inside an infected actor; calling them anywhere else raises
a loud ``RuntimeError``. You can introspect at runtime with a loud ``RuntimeError``. You can introspect at runtime with
``tractor.current_actor().is_infected_aio()``. ``tractor.current_actor().is_infected_aio()``.
@ -229,7 +229,7 @@ dialog, skip the channel ceremony and use
It schedules the fn as an ``asyncio.Task``, waits for completion It schedules the fn as an ``asyncio.Task``, waits for completion
and hands the return value back to ``trio``; think of it as the and hands the return value back to ``trio``; think of it as the
cross-loop sibling of ``ActorNursery.run_in_actor()``. Errors and cross-loop sibling of ``tractor.to_actor.run()``. Errors and
cancellation are translated exactly as for channels. cancellation are translated exactly as for channels.
Cross-loop errors and cancellation Cross-loop errors and cancellation

View File

@ -64,11 +64,13 @@ What's going on here?
- three healthy actors are spawned as daemons via - three healthy actors are spawned as daemons via
:meth:`tractor.ActorNursery.start_actor`; left alone they'd :meth:`tractor.ActorNursery.start_actor`; left alone they'd
happily idle forever, happily idle forever,
- a fourth actor runs ``assert_err()`` via ``.run_in_actor()`` and - a fourth actor runs ``assert_err()`` via a blocking
promptly trips its ``assert 0``, ``tractor.to_actor.run()`` one-shot and promptly trips its
``assert 0``,
- the resulting ``AssertionError`` ships back over IPC as a - the resulting ``AssertionError`` ships back over IPC as a
serialized error msg and re-raises *boxed* inside the nursery serialized error msg and re-raises *boxed* right at the call
block as a :class:`tractor.RemoteActorError`, inside the nursery block as a
:class:`tractor.RemoteActorError`,
- the nursery reacts like any ``trio`` nursery would: it cancels - the nursery reacts like any ``trio`` nursery would: it cancels
the three healthy siblings (graceful runtime-cancel requests, the three healthy siblings (graceful runtime-cancel requests,
acks awaited), reaps all four processes, then re-raises, acks awaited), reaps all four processes, then re-raises,

View File

@ -15,8 +15,8 @@ a single `structured concurrency`_ (SC) scope over IPC.
:alt: sequence diagram of the context handshake msg flow :alt: sequence diagram of the context handshake msg flow
Pretty much everything else is (or is slated to be) built on this Pretty much everything else is (or is slated to be) built on this
one primitive: ``ActorNursery.run_in_actor()`` is a convenience one primitive: ``tractor.to_actor.run()`` is a convenience for
for "spawn, open a context, await the result, tear down"; plain "spawn, run the lone task, await the result, tear down"; plain
``Portal.run()`` RPC is planned to be re-implemented on top of it; ``Portal.run()`` RPC is planned to be re-implemented on top of it;
the multi-process debugger's tree-wide REPL lock rides one. Grok the multi-process debugger's tree-wide REPL lock rides one. Grok
this page and the rest of the library reads as convenience this page and the rest of the library reads as convenience

View File

@ -119,15 +119,16 @@ Run a func in a process
Even a pool can be overkill; "run this one async func in a Even a pool can be overkill; "run this one async func in a
subprocess and give me the result" is a one-liner via subprocess and give me the result" is a one-liner via
:meth:`tractor.ActorNursery.run_in_actor`, :func:`tractor.to_actor.run`,
.. literalinclude:: ../../examples/parallelism/single_func.py .. literalinclude:: ../../examples/parallelism/single_func.py
:caption: examples/parallelism/single_func.py :caption: examples/parallelism/single_func.py
:language: python :language: python
``run_in_actor()`` is a *convenience wrapper* — spawn an actor, run ``to_actor.run()`` is a *convenience wrapper* — spawn an actor,
exactly one task in it, reap on result — not the core spawning run exactly one task in it, block on and return its result, reap
model (that's :meth:`tractor.ActorNursery.start_actor` plus — not the core spawning model (that's
:meth:`tractor.ActorNursery.start_actor` plus
:meth:`tractor.Portal.open_context`; see :doc:`/guide/context`). :meth:`tractor.Portal.open_context`; see :doc:`/guide/context`).
But for this fire-and-collect shape it's exactly the right amount But for this fire-and-collect shape it's exactly the right amount
of typing. of typing.

View File

@ -80,28 +80,30 @@ One special namespace exists: ``'self'`` resolves to the remote
how internal machinery (cancel requests, registry ops) travels; how internal machinery (cancel requests, registry ops) travels;
don't build your app on it. don't build your app on it.
One-shot results: ``wait_for_result()`` One-shot subactors: ``to_actor.run()``
--------------------------------------- --------------------------------------
A portal returned from When a subactor's *entire job* is a single function call, skip
:meth:`~tractor.ActorNursery.run_in_actor` has exactly one the portal plumbing with :func:`tractor.to_actor.run`: spawn,
"main" task running remotely; that task's ``return`` value is run the lone task, return its result and reap the process — all
delivered as the portal's *final result*: in one blocking call:
.. code:: python .. code:: python
portal = await an.run_in_actor(fib, n=10) final = await tractor.to_actor.run(fib, an=an, n=10)
final = await portal.wait_for_result()
Semantics worth knowing: Semantics worth knowing:
- it blocks until the remote task returns, re-raising any - it blocks until the remote task returns, re-raising any
remote error in the usual boxed form. remote error in the usual boxed form right in the calling
- once resolved it's idempotent: later calls return the same task.
cached value. - "placement" is composable: ``an=`` spawns from an existing
- a *daemon* portal (from ``start_actor()``) has no main task, actor-nursery, ``portal=`` reuses an already-running actor
so there's no final result to wait for: you'll get a warning (no spawn/reap, just a ``Portal.run()``), and passing
plus a ``NoResult`` sentinel. Results of individual daemon neither opens a private call-scoped nursery (booting the
calls come straight back from each ``await portal.run()``. runtime if needed).
- concurrency composes the plain ``trio`` way: schedule
multiple ``run()`` calls into a local task nursery (see
``examples/parallelism/to_actor_one_shots.py``).
Pure RPC daemons: ``run_daemon()`` Pure RPC daemons: ``run_daemon()``
---------------------------------- ----------------------------------

View File

@ -103,19 +103,22 @@ What's going on here?
on him **forever**. Daemon lifetimes are *yours* to end; that on him **forever**. Daemon lifetimes are *yours* to end; that
explicitness is the point. explicitness is the point.
``run_in_actor()``: quick one-shot parallelism ``to_actor.run()``: quick one-shot parallelism
---------------------------------------------- ----------------------------------------------
:meth:`~tractor.ActorNursery.run_in_actor` is the convenience :func:`tractor.to_actor.run` is the convenience wrapper: spawn
wrapper: spawn an actor, run exactly one async function in it, an actor, run exactly one async function in it, block on the
then reap the process as soon as the result arrives. result, then reap the process — the distributed sibling of
``trio.to_thread.run_sync()``.
.. code:: python .. code:: python
async with tractor.open_nursery() as an: async with (
portal = await an.run_in_actor(burn_cpu) tractor.open_nursery() as an,
trio.open_nursery() as tn,
):
# burn rubber in the parent too... # burn rubber in the parent too...
await burn_cpu() tn.start_soon(burn_cpu)
total = await portal.wait_for_result() total = await tractor.to_actor.run(burn_cpu, an=an)
A few details worth knowing: A few details worth knowing:
@ -124,18 +127,21 @@ A few details worth knowing:
- 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. - extra ``**kwargs`` are forwarded to the function itself.
- the child is *auto-cancelled* once its "main" result lands; - the call blocks until the result (or error) lands and the
at nursery exit these run-once children are always reaped child is *auto-cancelled* (reaped) right after — so remote
first (causality_ is paramount!). errors raise directly in your calling task (causality_ is
paramount!).
- "placement" composes: ``an=`` spawns from a caller-managed
actor-nursery, ``portal=`` reuses an already-running actor
(no spawn/reap), and passing neither opens a private
call-scoped nursery (booting the runtime if needed).
.. note:: .. note::
``run_in_actor()`` is a convenience, **not** the core model. ``to_actor.run()`` is a convenience, **not** the core model —
The source literally marks it for an eventual rebuild as it's built *entirely* on ``start_actor()`` + ``Portal.run()``
a thin "hilevel" wrapper on top of + ``Portal.cancel_actor()``. Teach your fingers to use it for
:meth:`~tractor.Portal.open_context` (the modern inter-actor quick fire-and-collect parallelism — think a per-function
task API). Teach your fingers to use it for quick
fire-and-collect parallelism — think a per-function
trio-parallel_ style one-shot — and reach for trio-parallel_ style one-shot — and reach for
``start_actor()`` + ``open_context()`` for anything ``start_actor()`` + ``open_context()`` for anything
long-lived, stateful or streaming long-lived, stateful or streaming
@ -145,9 +151,9 @@ Actor lifetimes and teardown order
---------------------------------- ----------------------------------
So we have two lifetime flavors: So we have two lifetime flavors:
- **run-once** (``run_in_actor()``): 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. arrives back in the (blocking) call.
- **daemon** (``start_actor()``): lives until *someone* cancels - **daemon** (``start_actor()``): lives until *someone* cancels
it — an explicit ``await portal.cancel_actor()``, a bulk it — an explicit ``await portal.cancel_actor()``, a bulk
``await an.cancel()``, or the one-cancels-all strategy kicking ``await an.cancel()``, or the one-cancels-all strategy kicking
@ -155,11 +161,12 @@ So we have two lifetime flavors:
On a clean exit of the nursery block the teardown order is: On a clean exit of the nursery block the teardown order is:
1. the nursery waits on every run-once actor's final result; 1. one-shot actors never make it to nursery exit: each is
any errors from these are raised immediately so your code reaped inside its own ``to_actor.run()`` call, any error
(acting as supervisor) gets first crack at handling them. raising immediately in the calling task so your code
2. then it waits on daemon actors — **indefinitely**. If you (acting as supervisor) gets first crack at handling it.
spawned a daemon, you own its lifetime. 2. the nursery then waits on daemon actors — **indefinitely**.
If you spawned a daemon, you own its lifetime.
When a child *is* cancelled, teardown is graceful-first per SC When a child *is* cancelled, teardown is graceful-first per SC
discipline: the runtime sends an IPC cancel request and gives discipline: the runtime sends an IPC cancel request and gives

View File

@ -43,24 +43,20 @@ Run it::
What's going on here? What's going on here?
- ``trio.run(main)`` starts the **root actor**; the ``tractor`` - ``trio.run(main)`` starts the **root actor**; the ``tractor``
runtime boots *implicitly* inside ``tractor.open_nursery()`` runtime boots *implicitly* inside ``tractor.to_actor.run()``
whenever it isn't already up. No special entrypoint, no whenever it isn't already up. No special entrypoint, no
framework takeover - it's just a ``trio`` app, framework takeover - it's just a ``trio`` app,
- inside ``main()`` a *subactor* is spawned via - inside ``main()`` a *subactor* is spawned via
``ActorNursery.run_in_actor()`` and told to run exactly one ``tractor.to_actor.run()`` and told to run exactly one
function: ``cellar_door()``, function: ``cellar_door()``,
- you get back a ``Portal``: your handle for invoking tasks in
the new process's (separate!) memory domain. We lean on it
much harder in the next section,
- the subactor, *some_linguist*, boots a fresh ``trio.run()`` in - the subactor, *some_linguist*, boots a fresh ``trio.run()`` in
a **new process** and executes ``cellar_door()`` as its *main a **new process** and executes ``cellar_door()`` as its *main
task* (note the child proving it is *not* the root with task* (note the child proving it is *not* the root with
``tractor.is_root_process()``), then ships the return value ``tractor.is_root_process()``), then ships the return value
back over IPC, back over IPC,
- the parent grabs that *final result* with - the call *blocks* until that final result arrives, then
``await portal.wait_for_result()``, much like you'd expect returns it - causality is preserved: your task only proceeds
from a "future" - except causality is preserved: the nursery once the child is *done*, dead, and reaped.
block only exits once the child is *done*, dead, and reaped.
.. margin:: Just need a worker pool? .. margin:: Just need a worker pool?
@ -71,19 +67,22 @@ What's going on here?
.. note:: .. note::
``run_in_actor()`` is the *convenience* wrapper: one-shot ``to_actor.run()`` (parlance of ``trio.to_thread`` and
friends) is the *convenience* wrapper: one-shot
spawn-run-reap semantics for when a subactor's entire job is spawn-run-reap semantics for when a subactor's entire job is
a single function call. The core primitives are a single function call. The core primitives are
``ActorNursery.start_actor()`` (next up) paired with ``ActorNursery.start_actor()`` (next up) — which hands you
a ``Portal``, your handle for invoking tasks in the new
process's (separate!) memory domain — paired with
``Portal.open_context()`` for full, SC-linked cross-actor ``Portal.open_context()`` for full, SC-linked cross-actor
dialogs - see :doc:`/guide/context`. dialogs - see :doc:`/guide/context`.
Daemon actors and RPC Daemon actors and RPC
--------------------- ---------------------
A ``run_in_actor()``-spawned actor terminates when its main task A ``to_actor.run()`` one-shot subactor terminates when its lone
returns. But often you want long-lived *daemon* actors instead: task returns. But often you want long-lived *daemon* actors
spawned once, then serving (allowlisted) RPC requests until told instead: spawned once, then serving (allowlisted) RPC requests
otherwise. That's ``start_actor()``: until told otherwise. That's ``start_actor()``:
.. literalinclude:: ../../examples/actor_spawning_and_causality_with_daemon.py .. literalinclude:: ../../examples/actor_spawning_and_causality_with_daemon.py
:caption: examples/actor_spawning_and_causality_with_daemon.py :caption: examples/actor_spawning_and_causality_with_daemon.py
@ -91,9 +90,9 @@ otherwise. That's ``start_actor()``:
Two lifetime rules to internalize: Two lifetime rules to internalize:
- a ``run_in_actor()`` actor lives exactly as long as its main - a ``to_actor.run()`` one-shot actor lives exactly as long as
task; the nursery waits for that function (and thus the its lone task; the call blocks until that function (and thus
process) to complete before unblocking, the process) completes,
- a ``start_actor()`` actor *lives forever* - an RPC daemon the - a ``start_actor()`` actor *lives forever* - an RPC daemon the
nursery will happily wait on **indefinitely** - until some nursery will happily wait on **indefinitely** - until some
task explicitly cancels it via ``Portal.cancel_actor()`` (as task explicitly cancels it via ``Portal.cancel_actor()`` (as

View File

@ -17,27 +17,38 @@ async def say_hello(other_actor):
return await portal.run(hi) return await portal.run(hi)
async def run_and_print(
an: tractor.ActorNursery,
name: str,
other_actor: str,
):
print(
await tractor.to_actor.run(
say_hello,
an=an,
name=name,
# arguments are always named
other_actor=other_actor,
)
)
async def main(): async def main():
"""Main tractor entry point, the "master" process (for now """Main tractor entry point, the "master" process (for now
acts as the "director"). acts as the "director").
""" """
async with tractor.open_nursery() as n: async with (
tractor.open_nursery() as an,
trio.open_nursery() as tn,
):
print("Alright... Action!") print("Alright... Action!")
donny = await n.run_in_actor( # both actors wait on the *other* to register so their
say_hello, # one-shots must run concurrently.
name='donny', tn.start_soon(run_and_print, an, 'donny', 'gretchen')
# arguments are always named tn.start_soon(run_and_print, an, 'gretchen', 'donny')
other_actor='gretchen',
) print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...")
gretchen = await n.run_in_actor(
say_hello,
name='gretchen',
other_actor='donny',
)
print(await gretchen.wait_for_result())
print(await donny.wait_for_result())
print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...")
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -10,17 +10,14 @@ async def cellar_door():
async def main(): async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
""" """
async with tractor.open_nursery() as n: # spawn a subactor, run ``cellar_door()`` as its lone task,
# block until its result arrives and the subactor is reaped.
portal = await n.run_in_actor( print(
await tractor.to_actor.run(
cellar_door, cellar_door,
name='some_linguist', name='some_linguist',
) )
)
# The ``async with`` will unblock here since the 'some_linguist'
# actor has completed its main task ``cellar_door``.
print(await portal.wait_for_result())
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -1,3 +1,5 @@
from functools import partial
import trio import trio
import tractor import tractor
@ -21,25 +23,36 @@ async def breakpoint_forever():
async def spawn_until(depth=0): async def spawn_until(depth=0):
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
async with tractor.open_nursery() as n: async with (
tractor.open_nursery() as an,
trio.open_nursery() as tn,
):
if depth < 1: if depth < 1:
await n.run_in_actor(breakpoint_forever) tn.start_soon(
partial(
p = await n.run_in_actor( tractor.to_actor.run,
name_error, breakpoint_forever,
name='name_error' an=an,
)
) )
await trio.sleep(0.5) await trio.sleep(0.5)
# rx and propagate error from child # rx and propagate error from child
await p.result() await tractor.to_actor.run(
name_error,
an=an,
name='name_error',
)
else: else:
# recusrive call to spawn another process branching layer of # recusrive call to spawn another process branching layer of
# the tree # the tree; blocks (up) each level until the leaf's
# `name_error` relays through.
depth -= 1 depth -= 1
await n.run_in_actor( await tractor.to_actor.run(
spawn_until, spawn_until,
an=an,
depth=depth, depth=depth,
name=f'spawn_until_{depth}', name=f'spawn_until_{depth}',
) )
@ -65,35 +78,34 @@ async def main():
python -m tractor._child --uid ('spawn_until_0', 'de918e6d ...) python -m tractor._child --uid ('spawn_until_0', 'de918e6d ...)
""" """
async with tractor.open_nursery( async with (
debug_mode=True, tractor.open_nursery(
loglevel='pdb', debug_mode=True,
) as n: loglevel='pdb',
) as an,
# spawn both actors trio.open_nursery() as tn,
portal = await n.run_in_actor( ):
spawn_until, # spawn both spawner trees as concurrent one-shots; the
depth=3, # first tree's (relayed) error cancels the other.
name='spawner0', tn.start_soon(
partial(
tractor.to_actor.run,
spawn_until,
an=an,
depth=3,
name='spawner0',
)
) )
portal1 = await n.run_in_actor( tn.start_soon(
spawn_until, partial(
depth=4, tractor.to_actor.run,
name='spawner1', spawn_until,
an=an,
depth=4,
name='spawner1',
)
) )
# TODO: test this case as well where the parent don't see
# the sub-actor errors by default and instead expect a user
# ctrl-c to kill the root.
with trio.move_on_after(3):
await trio.sleep_forever()
# gah still an issue here.
await portal.result()
# should never get here
await portal1.result()
if __name__ == '__main__': if __name__ == '__main__':
trio.run(main) trio.run(main)

View File

@ -16,11 +16,11 @@ async def spawn_error():
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal = await n.run_in_actor( return await tractor.to_actor.run(
name_error, name_error,
an=n,
name='name_error_1', name='name_error_1',
) )
return await portal.result()
async def main(): async def main():
@ -38,29 +38,36 @@ async def main():
- root actor should then fail on assert - root actor should then fail on assert
- program termination - program termination
""" """
async with tractor.open_nursery( async with (
debug_mode=True, tractor.open_nursery(
loglevel='devx', debug_mode=True,
) as n: loglevel='devx',
) as an,
trio.open_nursery() as tn,
):
# spawn both actors..
portal = await an.start_actor(
'name_error',
enable_modules=[__name__],
)
portal1 = await an.start_actor(
'spawn_error',
enable_modules=[__name__],
)
# spawn both actors # ..and bg-schedule their erroring tasks.
portal = await n.run_in_actor( tn.start_soon(portal.run, name_error)
name_error, tn.start_soon(portal1.run, spawn_error)
name='name_error',
) # yield to the bg tasks so both RPC requests are
portal1 = await n.run_in_actor( # submitted (and start crashing) before the root's own
spawn_error, # error below (the legacy `run_in_actor()` submitted
name='spawn_error', # in-line with each spawn).
) await trio.sleep(0.5)
# trigger a root actor error # trigger a root actor error
assert 0 assert 0
# attempt to collect results (which raises error in parent)
# still has some issues where the parent seems to get stuck
await portal.result()
await portal1.result()
if __name__ == '__main__': if __name__ == '__main__':
trio.run(main) trio.run(main)

View File

@ -18,11 +18,11 @@ async def spawn_error():
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal = await n.run_in_actor( return await tractor.to_actor.run(
name_error, name_error,
an=n,
name='name_error_1', name='name_error_1',
) )
return await portal.result()
async def main(): async def main():
@ -36,17 +36,39 @@ async def main():
`-python -m tractor._child --uid ('spawn_error', '52ee14a5 ...) `-python -m tractor._child --uid ('spawn_error', '52ee14a5 ...)
`-python -m tractor._child --uid ('name_error', '3391222c ...) `-python -m tractor._child --uid ('name_error', '3391222c ...)
""" """
errors: list[BaseException] = []
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,
# loglevel='runtime', # loglevel='runtime',
) as n: ) as an:
# Spawn both actors, don't bother with collecting results async def run_and_collect(fn):
# (would result in a different debugger outcome due to parent's '''
# cancellation). One-shot whose (boxed) error is stashed instead of
await n.run_in_actor(breakpoint_forever) raised so a sibling's crash never cancels the others
await n.run_in_actor(name_error) before they've had their own debugger sessions (the
await n.run_in_actor(spawn_error) "collect all errors" the legacy `run_in_actor()` API
did implicitly at nursery teardown).
'''
try:
await tractor.to_actor.run(fn, an=an)
except tractor.RemoteActorError as rae:
errors.append(rae)
# Spawn all one-shot task actors, collecting (vs.
# raising) their errors.
async with trio.open_nursery() as tn:
tn.start_soon(run_and_collect, breakpoint_forever)
tn.start_soon(run_and_collect, name_error)
tn.start_soon(run_and_collect, spawn_error)
if errors:
raise BaseExceptionGroup(
'multi_subactors errored!',
errors,
)
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -1,3 +1,5 @@
from functools import partial
import trio import trio
import tractor import tractor
@ -10,14 +12,14 @@ async def name_error():
async def spawn_until(depth=0): async def spawn_until(depth=0):
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
async with tractor.open_nursery() as n: async with tractor.open_nursery() as an:
if depth < 1: if depth < 1:
# await n.run_in_actor('breakpoint_forever', breakpoint_forever) await tractor.to_actor.run(name_error, an=an)
await n.run_in_actor(name_error)
else: else:
depth -= 1 depth -= 1
await n.run_in_actor( await tractor.to_actor.run(
spawn_until, spawn_until,
an=an,
depth=depth, depth=depth,
name=f'spawn_until_{depth}', name=f'spawn_until_{depth}',
) )
@ -37,28 +39,33 @@ async def main():
python -m tractor._child --uid ('name_error', '6c2733b8 ...) python -m tractor._child --uid ('name_error', '6c2733b8 ...)
''' '''
async with tractor.open_nursery( async with (
debug_mode=True, tractor.open_nursery(
enable_transports=['uds'], # TODO, apss this via osenv? debug_mode=True,
loglevel='devx', # XXX, required for test! enable_transports=['uds'], # TODO, apss this via osenv?
) as n: loglevel='devx', # XXX, required for test!
) as an,
trio.open_nursery() as tn,
):
# spawn the deeper tree in the bg..
tn.start_soon(
partial(
tractor.to_actor.run,
spawn_until,
an=an,
depth=1,
name='spawner1',
)
)
# spawn both actors # ..while blocking on the shallow (faster to fail) tree
portal = await n.run_in_actor( # whose propagated error triggers nursery cancellation.
await tractor.to_actor.run(
spawn_until, spawn_until,
an=an,
depth=0, depth=0,
name='spawner0', name='spawner0',
) )
portal1 = await n.run_in_actor(
spawn_until,
depth=1,
name='spawner1',
)
# nursery cancellation should be triggered due to propagated
# error from child.
await portal.result()
await portal1.result()
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -13,17 +13,24 @@ async def main():
simultaneously. simultaneously.
''' '''
async with tractor.open_nursery( async with (
debug_mode=True, tractor.open_nursery(
# loglevel='debug' # ?XXX required? debug_mode=True,
) as n: # loglevel='debug' # ?XXX required?
) as n,
# spawn both actors trio.open_nursery() as tn,
portal = await n.run_in_actor(key_error) ):
# spawn the actor..
portal = await n.start_actor(
'key_error',
enable_modules=[__name__],
)
print( print(
f'Child is up @ {portal.chan.aid.reprol()}' f'Child is up @ {portal.chan.aid.reprol()}'
) )
# ..then schedule its erroring task in the bg while the
# root blocks below.
tn.start_soon(portal.run, key_error)
# XXX: originally a bug caused by this is where root would enter # XXX: originally a bug caused by this is where root would enter
# the debugger and clobber the tty used by the repl even though # the debugger and clobber the tty used by the repl even though

View File

@ -75,10 +75,10 @@ async def main():
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,
) as n: ) as n:
portal: tractor.Portal = await n.run_in_actor( await tractor.to_actor.run(
cancelled_before_pause, cancelled_before_pause,
an=n,
) )
await portal.wait_for_result()
# ensure the same works in the root actor! # ensure the same works in the root actor!
await pm_on_cancelled() await pm_on_cancelled()

View File

@ -19,10 +19,12 @@ async def main():
loglevel='cancel', loglevel='cancel',
) as n: ) as n:
portal = await n.run_in_actor( # parks awaiting a result which only arrives once the
# user quits (`BdbQuit`s) the child's REPL loop.
await tractor.to_actor.run(
breakpoint_forever, breakpoint_forever,
an=n,
) )
await portal.wait_for_result()
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -12,16 +12,12 @@ async def main():
) as an: ) as an:
# TODO: ideally the REPL arrives at this frame in the parent, # TODO: ideally the REPL arrives at this frame in the parent,
# ABOVE the @api_frame of `Portal.run_in_actor()` (which # ABOVE the @api_frame of `to_actor.run()` ..
# should eventually not even be a portal method ... XD)
# await tractor.pause() # await tractor.pause()
p: tractor.Portal = await an.run_in_actor(name_error)
# with this style, should raise on this line # the one-shot blocks on the subactor's result so the
await p.wait_for_result() # boxed `NameError` raises right here.
await tractor.to_actor.run(name_error, an=an)
# with this alt style should raise at `open_nusery()`
# return await p.wait_for_result()
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -90,7 +90,7 @@ async def main() -> None:
# TODO: 3 sub-actor usage cases: # TODO: 3 sub-actor usage cases:
# -[x] via a `.open_context()` # -[x] via a `.open_context()`
# -[ ] via a `.run_in_actor()` call # -[ ] via a `to_actor.run()` call
# -[ ] via a `.run()` # -[ ] via a `.run()`
# -[ ] via a `.to_thread.run_sync()` in subactor # -[ ] via a `.to_thread.run_sync()` in subactor
async with p.open_context( async with p.open_context(

View File

@ -25,17 +25,15 @@ async def burn_cpu():
async def main(): async def main():
async with tractor.open_nursery() as n: async with trio.open_nursery() as tn:
portal = await n.run_in_actor(burn_cpu) # burn rubber in the parent too
tn.start_soon(burn_cpu)
# burn rubber in the parent too # run the same func as the lone task in a subactor,
await burn_cpu() # block on (and collect) its result
pid = await tractor.to_actor.run(burn_cpu)
# wait on result from target function
pid = await portal.wait_for_result()
# end of nursery block
print(f"Collected subproc {pid}") print(f"Collected subproc {pid}")

View File

@ -15,11 +15,12 @@ async def main():
enable_modules=[__name__], enable_modules=[__name__],
)) ))
# start one actor that will fail immediately # run one one-shot task actor that will fail immediately;
await n.run_in_actor(assert_err) # its error raises right here in the caller's task..
await tractor.to_actor.run(assert_err, an=n)
# should error here with a ``RemoteActorError`` containing # ..as a ``RemoteActorError`` containing an ``AssertionError``
# an ``AssertionError`` and all the other actors have been cancelled # and all the other actors have been cancelled
if __name__ == '__main__': if __name__ == '__main__':

View File

@ -849,43 +849,36 @@ def test_multi_nested_subactors_error_through_nurseries(
break break
# boxed source errors # boxed source errors
#
# NB post-#477 (`to_actor.run()` one-shots in local
# task-nurseries) the final relay is the LAST-released
# (leaf) REPL's error chain: it wins each level's
# relay-vs-cancel race so every level's single-member
# group gets unwrapped by the runtime's `collapse_eg()`
# (annotated at each actor boundary) while the sibling
# tree ('spawner1') is cancelled + absorbed. The legacy
# `run_in_actor()` teardown-reap instead grouped BOTH the
# `name_error` and bp-quit chains into the final dump
# (the previously-unexplained "extra" patterns).
expect_patts: list[str] = [ expect_patts: list[str] = [
"NameError: name 'doggypants' is not defined",
"tractor._exceptions.RemoteActorError:", "tractor._exceptions.RemoteActorError:",
"('name_error'",
# first level subtrees # each level's unwrapped-single-member-group
# "tractor._exceptions.RemoteActorError: ('spawner0'", # annotation + the first-level subtree's boundary
"src_uid=('spawner0'", # footer.
"( ^^^ this exc was collapsed from a group ^^^ )",
# "tractor._exceptions.RemoteActorError: ('spawner1'", "------ ('spawner0'",
# propagation of errors up through nested subtrees
# "tractor._exceptions.RemoteActorError: ('spawn_until_0'",
# "tractor._exceptions.RemoteActorError: ('spawn_until_1'",
# "tractor._exceptions.RemoteActorError: ('spawn_until_2'",
# ^-NOTE-^ old RAE repr, new one is below with a field
# showing the src actor's uid.
"src_uid=('spawn_until_2'",
] ]
# XXX, I HAVE NO IDEA why these patts only show on the
# `trio`-spawner but it seems to have something to do with
# what gets dumped in prior-prompt latches somehow??
# TODO for claude, explain and or work through how this is
# happening but ONLY WHEN RUN FROM THE TEST, bc when i try to
# run the test script manually the correct output ALWAYS seems
# to be in the last `str(child.before.decode())` output !?!?
if ( if (
not is_forking_spawner not is_forking_spawner
and and
last_send_char == 'q' last_send_char == 'q'
): ):
expect_patts += [ expect_patts += [
# expect the pdb-quit exc. # expect the pdb-quit exc relayed from the leaf's
# bp-loop child.
"bdb.BdbQuit", "bdb.BdbQuit",
# BUT WHY these dude!? "src_uid=('breakpoint_forever'",
"src_uid=('spawn_until_0'",
"relay_uid=('spawn_until_1'",
] ]
assert_before( assert_before(

View File

@ -3,6 +3,7 @@ Advanced streaming patterns using bidirectional streams and contexts.
''' '''
from collections import Counter from collections import Counter
from functools import partial
import itertools import itertools
import platform import platform
from typing import Type from typing import Type
@ -173,8 +174,8 @@ def test_dynamic_pub_sub(
# test. Picked backend-aware: under `trio` backend spawn is # test. Picked backend-aware: under `trio` backend spawn is
# cheap (~1s for `cpus` actors) but fork-based backends pay # cheap (~1s for `cpus` actors) but fork-based backends pay
# a per-spawn cost (forkserver round-trip + IPC peer-handshake) # a per-spawn cost (forkserver round-trip + IPC peer-handshake)
# that can stack up over `cpus - 1` sequential `n.run_in_actor()` # that can stack up over the `cpus - 1` one-shot
# calls — especially on UDS under cross-pytest contention # (`to_actor.run()`) spawns — especially on UDS under cross-pytest contention
# (#451 / #452). 4s was flaking right at the edge under fork # (#451 / #452). 4s was flaking right at the edge under fork
# backends — bumped to 8s with diag-snapshot-on-timeout via # backends — bumped to 8s with diag-snapshot-on-timeout via
# `fail_after_w_trace` so a borderline run still fails loud # `fail_after_w_trace` so a borderline run still fails loud
@ -214,33 +215,55 @@ def test_dynamic_pub_sub(
f'enter `fail_after_w_trace({fail_after_s})` scope' f'enter `fail_after_w_trace({fail_after_s})` scope'
) )
try: try:
async with tractor.open_nursery( async with (
registry_addrs=[reg_addr], tractor.open_nursery(
debug_mode=debug_mode, registry_addrs=[reg_addr],
) as n: debug_mode=debug_mode,
) as n,
# bg-schedules the forever-streaming
# one-shots below; the user-cancel raise
# cancels them all, each reaping its
# subactor via `to_actor.run()`'s
# (shielded) `Portal.cancel_actor()`.
trio.open_nursery() as tn,
):
test_log.cancel( test_log.cancel(
'test_dynamic_pub_sub: ' 'test_dynamic_pub_sub: '
'actor nursery opened' 'actor nursery opened'
) )
# name of this actor will be same as target func # name of this actor will be same as target func
await n.run_in_actor(publisher) tn.start_soon(
partial(
tractor.to_actor.run,
publisher,
an=n,
)
)
for i, sub in zip( for i, sub in zip(
range(cpus - 2), range(cpus - 2),
itertools.cycle(_registry.keys()) itertools.cycle(_registry.keys())
): ):
await n.run_in_actor( tn.start_soon(
consumer, partial(
name=f'consumer_{sub}', tractor.to_actor.run,
subs=[sub], consumer,
an=n,
name=f'consumer_{sub}',
subs=[sub],
)
) )
# make one dynamic subscriber # make one dynamic subscriber
await n.run_in_actor( tn.start_soon(
consumer, partial(
name='consumer_dynamic', tractor.to_actor.run,
subs=list(_registry.keys()), consumer,
an=n,
name='consumer_dynamic',
subs=list(_registry.keys()),
)
) )
# block until "cancelled by user" # block until "cancelled by user"

View File

@ -2,6 +2,7 @@
Cancellation and error propagation Cancellation and error propagation
""" """
from functools import partial
import os import os
import signal import signal
import platform import platform
@ -16,7 +17,10 @@ from tractor._testing import (
tractor_test, tractor_test,
) )
from tractor._testing.trace import FailAfterWTraceFactory from tractor._testing.trace import FailAfterWTraceFactory
from tractor.trionics import gather_contexts from tractor.trionics import (
collapse_eg,
gather_contexts,
)
from .conftest import no_windows from .conftest import no_windows
@ -305,30 +309,30 @@ async def test_cancel_infinite_streamer(
@pytest.mark.parametrize( @pytest.mark.parametrize(
'num_actors_and_errs', 'num_actors_and_errs',
[ [
# daemon actors sit idle while single task actors error out # daemon actors sit idle while one-shot task actors error out
(1, tractor.RemoteActorError, AssertionError, (assert_err, {}), None), (1, tractor.RemoteActorError, AssertionError, (assert_err, {}), None),
(2, BaseExceptionGroup, AssertionError, (assert_err, {}), None), (2, BaseExceptionGroup, AssertionError, (assert_err, {}), None),
(3, BaseExceptionGroup, AssertionError, (assert_err, {}), None), (3, BaseExceptionGroup, AssertionError, (assert_err, {}), None),
# 1 daemon actor errors out while single task actors sleep forever # 1 daemon actor errors out while one-shot task actors sleep forever
(3, tractor.RemoteActorError, AssertionError, (sleep_forever, {}), (3, tractor.RemoteActorError, AssertionError, (sleep_forever, {}),
(assert_err, {}, True)), (assert_err, {}, True)),
# daemon actors error out after brief delay while single task # daemon actors error out after brief delay while one-shot task
# actors complete quickly # actors complete quickly
(3, tractor.RemoteActorError, AssertionError, (3, tractor.RemoteActorError, AssertionError,
(do_nuthin, {}), (assert_err, {'delay': 1}, True)), (do_nuthin, {}), (assert_err, {'delay': 1}, True)),
# daemon complete quickly delay while single task # daemon complete quickly delay while one-shot task
# actors error after brief delay # actors error after brief delay
(3, BaseExceptionGroup, AssertionError, (3, BaseExceptionGroup, AssertionError,
(assert_err, {'delay': 1}), (do_nuthin, {}, False)), (assert_err, {'delay': 1}), (do_nuthin, {}, False)),
], ],
ids=[ ids=[
'1_run_in_actor_fails', '1_one_shot_fails',
'2_run_in_actors_fail', '2_one_shots_fail',
'3_run_in_actors_fail', '3_one_shots_fail',
'1_daemon_actors_fail', '1_daemon_actors_fail',
'1_daemon_actors_fail_all_run_in_actors_dun_quick', '1_daemon_actors_fail_all_one_shots_dun_quick',
'no_daemon_actors_fail_all_run_in_actors_sleep_then_fail', 'no_daemon_actors_fail_all_one_shots_sleep_then_fail',
], ],
) )
@tractor_test( @tractor_test(
@ -347,12 +351,22 @@ async def test_some_cancels_all(
This is the first and only supervisory strategy at the moment. This is the first and only supervisory strategy at the moment.
One-shot subactors run as concurrent `to_actor.run()` tasks
in a local task-nursery so their errors raise WHILE the
actor-nursery block is still open (vs the legacy
`run_in_actor()` teardown-reap); the first error cancels the
sibling one-shots (whose `trio.Cancelled`s the task-nursery
absorbs) so the group shape is 1..num_actors
`RemoteActorError`s depending on relay-vs-cancel timing
with `collapse_eg()` unwrapping the deterministic
single-error cases to a bare `RemoteActorError`.
''' '''
( (
num_actors, num_actors,
first_err, first_err,
err_type, err_type,
ria_func, one_shot_func,
da_func, da_func,
) = num_actors_and_errs ) = num_actors_and_errs
try: try:
@ -366,51 +380,64 @@ async def test_some_cancels_all(
enable_modules=[__name__], enable_modules=[__name__],
)) ))
func, kwargs = ria_func func, kwargs = one_shot_func
riactor_portals = [] async with (
for i in range(num_actors): collapse_eg(),
# start actor(s) that will fail immediately trio.open_nursery() as tn,
riactor_portals.append( ):
await an.run_in_actor( for i in range(num_actors):
func, # schedule one-shot task actor(s); errors
name=f'actor_{i}', # raise into this task-nursery scope.
**kwargs tn.start_soon(
partial(
tractor.to_actor.run,
func,
an=an,
name=f'actor_{i}',
**kwargs,
)
) )
)
if da_func: if da_func:
func, kwargs, expect_error = da_func func, kwargs, expect_error = da_func
for portal in dactor_portals: for portal in dactor_portals:
# if this function fails then we should error here # if this function fails then we should error
# and the nursery should teardown all other actors # here and the nursery should teardown all
try: # other actors
await portal.run(func, **kwargs) try:
await portal.run(func, **kwargs)
except tractor.RemoteActorError as err: except tractor.RemoteActorError as err:
assert err.boxed_type == err_type assert err.boxed_type == err_type
# we only expect this first error to propogate # we only expect this first error to propogate
# (all other daemons are cancelled before they # (all other daemons are cancelled before they
# can be scheduled) # can be scheduled)
num_actors = 1 num_actors = 1
# reraise so nursery teardown is triggered # reraise so nursery teardown is triggered
raise raise
else: else:
if expect_error: if expect_error:
pytest.fail( pytest.fail(
"Deamon call should fail at checkpoint?") "Deamon call should fail at checkpoint?")
# should error here with a ``RemoteActorError`` or ``MultiError`` # should error here with a `RemoteActorError` or a beg of them
except first_err as _err: except (
BaseExceptionGroup,
tractor.RemoteActorError,
) as _err:
err = _err err = _err
if isinstance(err, BaseExceptionGroup): if isinstance(err, BaseExceptionGroup):
assert len(err.exceptions) == num_actors # only the concurrent multi-error cases can group; the
# relay-vs-cancel race means anywhere from 1 (all
# siblings cancelled before relaying) up to all
# `num_actors` errors may populate the group.
assert first_err is BaseExceptionGroup
assert 1 <= len(err.exceptions) <= num_actors
for exc in err.exceptions: for exc in err.exceptions:
if isinstance(exc, tractor.RemoteActorError): assert isinstance(exc, tractor.RemoteActorError)
assert exc.boxed_type == err_type assert exc.boxed_type == err_type
else: else:
assert isinstance(exc, trio.Cancelled)
elif isinstance(err, tractor.RemoteActorError):
assert err.boxed_type == err_type assert err.boxed_type == err_type
assert an.cancel_called is True assert an.cancel_called is True
@ -423,8 +450,20 @@ async def spawn_and_error(
breadth: int, breadth: int,
depth: int, depth: int,
) -> None: ) -> None:
'''
Recursively spawn a breadth-wide level of erroring one-shot
subactors as concurrent `to_actor.run()` tasks; the leaf level
errors ~simultaneously and each level's task-nursery groups
whatever `RemoteActorError`s relay before the first one's
cancel wins, boxing the (`ExceptionGroup`-shaped) group into
this actor's own relayed error.
'''
name = tractor.current_actor().name name = tractor.current_actor().name
async with tractor.open_nursery() as nursery: async with (
tractor.open_nursery() as an,
trio.open_nursery() as tn,
):
for i in range(breadth): for i in range(breadth):
if depth > 0: if depth > 0:
@ -444,7 +483,14 @@ async def spawn_and_error(
kwargs = { kwargs = {
'name': f'{name}_errorer_{i}', 'name': f'{name}_errorer_{i}',
} }
await nursery.run_in_actor(*args, **kwargs) tn.start_soon(
partial(
tractor.to_actor.run,
*args,
an=an,
**kwargs,
)
)
# NOTE: `main_thread_forkserver` capture-fd hang class is no # NOTE: `main_thread_forkserver` capture-fd hang class is no
@ -486,7 +532,11 @@ async def test_nested_multierrors(
depth: int, depth: int,
): ):
''' '''
Test that failed actor sets are wrapped in `BaseExceptionGroup`s. Test that a nested tree of concurrently failing one-shot
subactors tears down cleanly, relaying (whatever subset of)
the leaf `AssertionError`s (that win the per-level
relay-vs-cancel race) re-boxed/grouped at each actor
boundary.
Parametrized over recursion `depth {1, 3}`: Parametrized over recursion `depth {1, 3}`:
@ -536,6 +586,13 @@ async def test_nested_multierrors(
# fork-spawn jitter + UDS-contention widens both `t1` and # fork-spawn jitter + UDS-contention widens both `t1` and
# `t2` further. # `t2` further.
# #
# NB post-#477 (`to_actor.run()` fan-out in a local
# task-nursery) a race-tripped sibling's `Cancelled` is
# ABSORBED by the task-nursery instead of landing in the
# group — the raced case now shows as a *smaller* BEG, so
# this marker should consistently `xpass`; drop it once CI
# confirms.
#
# With `strict=False` the clean-cascade cases (most # With `strict=False` the clean-cascade cases (most
# depth=1 runs, rare depth=3 runs) report as `xpassed` # depth=1 runs, rare depth=3 runs) report as `xpassed`
# while the race-tripped cases report as `xfailed` — # while the race-tripped cases report as `xfailed` —
@ -620,6 +677,14 @@ async def test_nested_multierrors(
timeout = 16 timeout = 16
case ('main_thread_forkserver', 3): case ('main_thread_forkserver', 3):
timeout = 30 timeout = 30
# any other fork-based backend (`mp_spawn` et al) pays
# the same per-spawn round-trip costs as MTF so rides
# its budgets; without a default arm `timeout` is left
# unbound -> `UnboundLocalError` at the scaling below.
case (_, 1):
timeout = 16
case (_, 3):
timeout = 30
# inflate the budget by the throttle headroom probed above so # inflate the budget by the throttle headroom probed above so
# a slow box doesn't masquerade as a deadline regression. # a slow box doesn't masquerade as a deadline regression.
@ -632,67 +697,82 @@ async def test_nested_multierrors(
async with fail_after_w_trace(timeout): async with fail_after_w_trace(timeout):
try: try:
async with tractor.open_nursery() as nursery: async with (
tractor.open_nursery() as nursery,
trio.open_nursery() as tn,
):
for i in range(subactor_breadth): for i in range(subactor_breadth):
await nursery.run_in_actor( tn.start_soon(
spawn_and_error, partial(
name=f'spawner_{i}', tractor.to_actor.run,
breadth=subactor_breadth, spawn_and_error,
depth=depth, an=nursery,
name=f'spawner_{i}',
breadth=subactor_breadth,
depth=depth,
)
) )
except BaseExceptionGroup as err: except (
assert len(err.exceptions) == subactor_breadth BaseExceptionGroup,
for subexc in err.exceptions: tractor.RemoteActorError,
) as err:
# verify first level actor errors are wrapped as remote # group membership is bounded by the relay-vs-cancel
if _friggin_windows: # race: the first spawner-tree's error cancels its
# siblings, whose own errors only group when relayed
# first; a fully-raced tree even collapses (via the
# runtime's own `collapse_eg()` unwrapping each level's
# single-member group) to a bare `RemoteActorError`
# re-boxing the leaf `AssertionError` at every actor
# boundary. The deterministic exact-breadth nested-BEG
# was the legacy `run_in_actor()` reap-all-at-teardown.
subexcs: list[BaseException] = (
err.exceptions
if isinstance(err, BaseExceptionGroup)
else [err]
)
assert 1 <= len(subexcs) <= subactor_breadth
for subexc in subexcs:
if (
_friggin_windows
and
isinstance(subexc, trio.Cancelled)
):
# windows is often too slow and cancellation seems # windows is often too slow and cancellation seems
# to happen before an actor is spawned # to happen before an actor is spawned
if isinstance(subexc, trio.Cancelled): continue
continue
elif isinstance(subexc, tractor.RemoteActorError): assert isinstance(subexc, tractor.RemoteActorError)
# on windows it seems we can't exactly be sure wtf
# will happen..
assert subexc.boxed_type in (
tractor.RemoteActorError,
trio.Cancelled,
BaseExceptionGroup,
)
elif isinstance(subexc, BaseExceptionGroup): accepted: tuple[Type[BaseException], ...] = (
for subsub in subexc.exceptions: # ≥2 sub-tree errors relayed before the
# cancel-cascade won → grouped per-level.
if subsub in (tractor.RemoteActorError,): ExceptionGroup,
subsub = subsub.boxed_type # every level collapsed down to its lone
# relayed (leaf) error.
assert type(subsub) in ( AssertionError,
trio.Cancelled, # a mid-level spawner relays an
BaseExceptionGroup, # already-boxed (collapsed) leaf chain,
) # re-boxing the `RemoteActorError` itself.
else: tractor.RemoteActorError,
assert isinstance(subexc, tractor.RemoteActorError) # under heavy load a runtime-internal reap
# deadline can inject a `trio.Cancelled`
if depth > 0 and subactor_breadth > 1: # into a child's group before relay (the
# XXX not sure what's up with this.. # same class the depth=3 throttle-xfail
# on windows sometimes spawning is just too slow and # covers) upgrading it from an
# we get back the (sent) cancel signal instead # `ExceptionGroup`.
if _friggin_windows: BaseExceptionGroup,
if isinstance(subexc, tractor.RemoteActorError): )
assert subexc.boxed_type in ( if _friggin_windows:
BaseExceptionGroup, # on windows it seems we can't exactly be
tractor.RemoteActorError # sure wtf will happen..
) accepted += (
else: trio.Cancelled,
assert isinstance(subexc, BaseExceptionGroup)
else:
assert subexc.boxed_type is ExceptionGroup
else:
assert subexc.boxed_type in (
tractor.RemoteActorError,
trio.Cancelled
) )
assert subexc.boxed_type in accepted
else:
pytest.fail(
'Should have raised a (grouped) `RemoteActorError`?'
)
@no_windows @no_windows
@ -757,11 +837,13 @@ def test_cancel_via_SIGINT_other_task(
): ):
async with tractor.open_nursery( async with tractor.open_nursery(
registry_addrs=[reg_addr], registry_addrs=[reg_addr],
) as tn: ) as an:
# just keep a set of (daemon) subactors alive for the
# SIGINT to cancel (was 3 `run_in_actor(sleep_forever)`
# one-shots — a daemon needs no "main" task to idle).
for i in range(3): for i in range(3):
await tn.run_in_actor( await an.start_actor(
sleep_forever, f'namesucka_{i}',
name='namesucka',
) )
task_status.started() task_status.started()
await trio.sleep_forever() await trio.sleep_forever()
@ -802,8 +884,11 @@ async def spin_for(period=3):
async def spawn_sub_with_sync_blocking_task(): async def spawn_sub_with_sync_blocking_task():
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
print('starting sync blocking subactor..\n') print('starting sync blocking subactor..\n')
await an.run_in_actor( # one-shot: parks HERE awaiting the sync-sleeping
# grandchild's result until cancelled from above.
await tractor.to_actor.run(
spin_for, spin_for,
an=an,
name='sleeper', name='sleeper',
) )
print('exiting first subactor layer..\n') print('exiting first subactor layer..\n')
@ -909,10 +994,18 @@ def test_cancel_while_childs_child_in_sync_sleep(
debug_mode=debug_mode, debug_mode=debug_mode,
registry_addrs=[reg_addr], registry_addrs=[reg_addr],
) as an, ) as an,
trio.open_nursery() as tn,
): ):
await an.run_in_actor( # bg one-shot: parks on the middle actor's result
spawn_sub_with_sync_blocking_task, # (itself parked on the sync-sleeping grandchild)
name='sync_blocking_sub', # until the `assert 0` below cancels this scope.
tn.start_soon(
partial(
tractor.to_actor.run,
spawn_sub_with_sync_blocking_task,
an=an,
name='sync_blocking_sub',
)
) )
await trio.sleep(1) await trio.sleep(1)

View File

@ -131,9 +131,12 @@ def test_ringbuf(
child_read_shm, child_read_shm,
**common_kwargs, **common_kwargs,
total_bytes=total_bytes, 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 send_p.cancel_actor()
await recv_p.cancel_actor() await recv_p.cancel_actor()

View File

@ -1,8 +1,8 @@
''' '''
`tractor.to_actor`: one-shot single-remote-task API suite. `tractor.to_actor`: one-shot single-remote-task API suite.
Verifies the "spiritual successor" to (and eventual Verifies the "spiritual successor" to (and replacement of)
replacement of) `ActorNursery.run_in_actor()`; see the removed legacy `ActorNursery.run_in_actor()`; see
https://github.com/goodboy/tractor/issues/477 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 A remote task error is raised directly in the
caller's task as a boxed `RemoteActorError` instead caller's task as a boxed `RemoteActorError` instead
of surfacing at actor-nursery teardown as with the 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: with pytest.raises(RemoteActorError) as excinfo:

View File

@ -780,7 +780,7 @@ class Context:
# `Portal.open_context()` has been opened since it's # `Portal.open_context()` has been opened since it's
# assumed that other portal APIs like, # assumed that other portal APIs like,
# - `Portal.run()`, # - `Portal.run()`,
# - `ActorNursery.run_in_actor()` # - `to_actor.run()`
# do their own error checking at their own call points and # do their own error checking at their own call points and
# result processing. # result processing.

View File

@ -1161,10 +1161,6 @@ class TransportClosed(Exception):
) )
class NoResult(RuntimeError):
"No final result is expected for this actor"
class ModuleNotExposed(ModuleNotFoundError): class ModuleNotExposed(ModuleNotFoundError):
"The requested module is not exposed for RPC" "The requested module is not exposed for RPC"

View File

@ -221,16 +221,19 @@ def pub(
import tractor import tractor
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal = n.run_in_actor( portal = await n.start_actor(
'publisher', # actor name 'publisher', # actor name
partial( # func to execute in it enable_modules=[__name__],
)
async with portal.open_stream_from(
partial( # func to execute in it
pub_service, pub_service,
topics=('clicks', 'users'), topics=('clicks', 'users'),
task_name='source1', task_name='source1',
) )
) ) as stream:
async for value in await portal.result(): async for value in stream:
print(f"Subscriber received {value}") print(f"Subscriber received {value}")
Here, you don't need to provide the ``ctx`` argument since the Here, you don't need to provide the ``ctx`` argument since the

View File

@ -304,12 +304,11 @@ class Start(
It is called by all the following public APIs: It is called by all the following public APIs:
- `ActorNursery.run_in_actor()` - `to_actor.run()`
- `Portal.run()` - `Portal.run()`
`|_.run_from_ns()` `|_.run_from_ns()`
`|_.open_stream_from()` `|_.open_stream_from()`
`|_._submit_for_result()`
- `Context.open_context()` - `Context.open_context()`

View File

@ -50,13 +50,11 @@ from ..ipc import Channel
from ..log import get_logger from ..log import get_logger
from ..msg import ( from ..msg import (
# Error, # Error,
PayloadMsg,
NamespacePath, NamespacePath,
Return, Return,
) )
from .._exceptions import ( from .._exceptions import (
ActorTooSlowError, ActorTooSlowError,
NoResult,
TransportClosed, TransportClosed,
) )
from .._context import ( from .._context import (
@ -102,14 +100,6 @@ class Portal:
) -> None: ) -> None:
self._chan: Channel = channel 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() self._streams: set[MsgStream] = set()
# TODO, this should be PRIVATE (and never used publicly)! since it's just # TODO, this should be PRIVATE (and never used publicly)! since it's just
@ -137,102 +127,6 @@ class Portal:
) )
return self.chan 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): async def _cancel_streams(self):
# terminate all locally running async generator # terminate all locally running async generator
# IPC calls # IPC calls

View File

@ -20,7 +20,6 @@
""" """
from contextlib import asynccontextmanager as acm from contextlib import asynccontextmanager as acm
from functools import partial from functools import partial
import inspect
from typing import ( from typing import (
TYPE_CHECKING, TYPE_CHECKING,
) )
@ -232,14 +231,6 @@ 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, 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 # trio.Nursery-like cancel (request) statuses
self._cancelled_caught: bool = False self._cancelled_caught: bool = False
self._cancel_called: 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 # @api_frame
async def cancel( async def cancel(
self, self,
@ -568,51 +481,6 @@ class ActorNursery:
self._join_procs.set() 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 @acm
async def _open_and_supervise_one_cancels_all_nursery( async def _open_and_supervise_one_cancels_all_nursery(
actor: Actor, actor: Actor,
@ -627,11 +495,10 @@ async def _open_and_supervise_one_cancels_all_nursery(
errors: dict[tuple[str, str], BaseException] = {} errors: dict[tuple[str, str], BaseException] = {}
# The single "daemon actor" nursery into which ALL subactors # The single "daemon actor" nursery into which ALL subactors
# are spawned — both `.start_actor()` daemons AND # are spawned; one-shot (`to_actor.run()`) subactors are
# `.run_in_actor()` one-shots. The latter's result-reaping now # result-waited and reaped in their caller's own task-scope
# runs via `_reap_ria_portals()` at block-exit rather than a # (see the #477 `.run_in_actor()`/`._ria_nursery` removal);
# 2ndary `._ria_nursery` (see the #477 removal); errors from # errors from this nursery bubble up to the caller.
# this nursery bubble up to the caller.
async with ( async with (
collapse_eg(), collapse_eg(),
trio.open_nursery() as da_nursery, trio.open_nursery() as da_nursery,
@ -655,11 +522,6 @@ async def _open_and_supervise_one_cancels_all_nursery(
) )
an._join_procs.set() 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) # Single one-cancels-all handler for the (now single)
# daemon nursery. Pre-#477 a 2ndary `._ria_nursery` # daemon nursery. Pre-#477 a 2ndary `._ria_nursery`
# required a separate *outer* handler to catch errors # 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 # cancel all subactors
await an.cancel() 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()" # an error was stashed by the handler above (or by
# actors but those actors may have returned remote errors as # a spawn task via the shared `errors` dict) so
# results (meaning they errored remotely and have relayed # cancel any remaining subactors, summarize and
# those errors back to this parent actor). The errors are # re-raise.
# collected in ``errors`` so cancel all actors, summarize
# all errors and re-raise.
if errors: if errors:
if an._children: if an._children:
with trio.CancelScope(shield=True): with trio.CancelScope(shield=True):

View File

@ -188,9 +188,7 @@ async def mp_proc(
# This is a "soft" (cancellable) join/reap which # This is a "soft" (cancellable) join/reap which
# will remote cancel the actor on a ``trio.Cancelled`` # will remote cancel the actor on a ``trio.Cancelled``
# condition. Any `.run_in_actor()` result-reaping # condition.
# happens up in the `ActorNursery` machinery (see
# `_supervise._reap_ria_portals()`), NOT here.
await soft_kill( await soft_kill(
proc, proc,
proc_waiter, proc_waiter,

View File

@ -126,98 +126,6 @@ def try_set_start_method(
return _ctx 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( async def hard_kill(
proc: trio.Process, proc: trio.Process,
@ -461,8 +369,8 @@ async def new_proc(
# NOTE: bottom-of-module to avoid a circular import since the # NOTE: bottom-of-module to avoid a circular import since the
# backend submodules pull `cancel_on_completion`/`soft_kill`/ # backend submodules pull `soft_kill`/`hard_kill`/`proc_waiter`
# `hard_kill`/`proc_waiter` from this module. # from this module.
from ._trio import trio_proc from ._trio import trio_proc
from ._mp import mp_proc from ._mp import mp_proc

View File

@ -196,9 +196,7 @@ async def trio_proc(
# This is a "soft" (cancellable) join/reap which # This is a "soft" (cancellable) join/reap which
# will remote cancel the actor on a ``trio.Cancelled`` # will remote cancel the actor on a ``trio.Cancelled``
# condition. Any `.run_in_actor()` result-reaping # condition.
# happens up in the `ActorNursery` machinery (see
# `_supervise._reap_ria_portals()`), NOT here.
await soft_kill( await soft_kill(
proc, proc,
trio.Process.wait, # XXX, uses `pidfd_open()` below. trio.Process.wait, # XXX, uses `pidfd_open()` below.

View File

@ -23,8 +23,8 @@ Adopts the "run it over there" parlance from analogous
reuse) a subactor, schedule a single remote task, wait on reuse) a subactor, schedule a single remote task, wait on
its result and (when the call owns the subactor) reap it. its result and (when the call owns the subactor) reap it.
The "spiritual successor" to (and eventual replacement of) The "spiritual successor" to (and replacement of) the removed
the `ActorNursery.run_in_actor()` API; see legacy `ActorNursery.run_in_actor()` API; see
https://github.com/goodboy/tractor/issues/477 https://github.com/goodboy/tractor/issues/477
''' '''

View File

@ -31,7 +31,7 @@ the lower level daemon-actor spawn + portal APIs,
such that error collection and propagation happens in the such that error collection and propagation happens in the
*caller's task* (and thus whatever `trio` nursery/scope *caller's task* (and thus whatever `trio` nursery/scope
encloses it) instead of inside the actor-nursery's 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. `ActorNursery.run_in_actor()` API.
''' '''
@ -151,9 +151,9 @@ async def run(
the distributed-parallelism equivalent of the distributed-parallelism equivalent of
`trio.to_thread.run_sync()`. `trio.to_thread.run_sync()`.
Unlike `ActorNursery.run_in_actor()` (which returns Unlike the removed legacy `.run_in_actor()` (which
a `Portal` whose result is only collected at returned a `Portal` whose result was only collected
actor-nursery teardown) this is a plain "call and at actor-nursery teardown) this is a plain "call and
wait" primitive: any remote error is raised HERE, in wait" primitive: any remote error is raised HERE, in
the caller's task. Concurrency is composed the usual the caller's task. Concurrency is composed the usual
`trio` way by scheduling multiple `run()` calls in `trio` way by scheduling multiple `run()` calls in