Compare commits

...

4 Commits

Author SHA1 Message Date
Gud Boi 70f497bf5f Correct `to_actor.run()` target guidance
Target validation moved to a follow-up branch, so the guide should not
claim unstable callable forms are rejected before actor startup.

Describe module-global functions and `functools.partial()` wrappers
as portable stable-address forms without promising absent enforcement.

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-21 01:52:24 -04:00
Gud Boi dc52ecb96b Close late `ActorNursery` registration race
A child could pass the early `.start_actor()` guard, miss the
`.cancel()` child snapshot and register afterward. Its monitor
inherited a reap request without runtime cancellation and could wait
forever.

- publish child/reap events before sampling `_cancel_called`
- make MP abort before `proc.start()` when cancellation won
- kill a Trio child opened after cancellation won registration
- reject starts begun after nursery cancellation is already visible
- add deterministic registration and MP no-start regressions
- drop the touched Trio backend's stale `get_runtime_vars` import

Prompt-IO: ai/prompt-io/opencode/20260821T040803Z_3c1bbe73_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-21 01:20:08 -04:00
Gud Boi 3c1bbe7373 Bound `Portal.cancel_actor()` frame sends
A cancel RPC could stall forever in complete-frame transport
shielding before the peer received it, bypassing the outer ack
timeout and blocking graceful supervision.

- thread one absolute deadline from `Portal.cancel_actor()` through
  the private `Start` publication path
- force-close a partial-frame stream before releasing its send lock
- keep ordinary sends unbounded and preserve pending cancellation
- document the current `Start -> StartAck -> CancelAck` exchange and
  link the dedicated `Cancel` msg follow-up in #506
- cover partial publication and the shared send/ack timeout budget

Prompt-IO: ai/prompt-io/opencode/20260821T023537Z_ae6f2ac3_prompt_io.md

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-20 22:52:20 -04:00
Gud Boi ae6f2ac35c Clarify `to_actor.run()` ownership
The guides described every placement as spawn-run-reap and omitted
the trampoline allowlist required when reusing an existing actor.

- distinguish call-owned children from caller-owned portal actors
- document stable module-global target addresses and allowlists
- add the #477 feature news fragment

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))
2026-08-20 17:21:03 -04:00
20 changed files with 735 additions and 121 deletions

View File

@ -0,0 +1,65 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
timestamp: 2026-08-21T02:35:37Z
git_ref: ae6f2ac3
scope: code
substantive: true
raw_file: 20260821T023537Z_ae6f2ac3_prompt_io.raw.md
---
## Prompt
Simplify bounded actor cancellation by passing an explicit absolute
deadline from `Portal.cancel_actor()` through `_run_from_ns()`,
`Actor.start_remote_task()`, and `Channel.send()` into
`MsgpackTransport.send()`. Avoid a `ContextVar`, watcher tasks, shared
status, coalescing, and waiter state. After tracing the current
`Start -> StartAck -> CancelAck` transaction, rename the local result to
`cancel_ack_received`, document its exact semantics, and link a focused
follow-up for a dedicated `Cancel -> CancelAck` protocol.
## Response summary
Threaded one absolute Trio deadline through the existing private
actor-cancel RPC path. The transport retains complete-frame shielding
for ordinary sends, while a cancel-control send that overruns its
deadline force-closes the potentially corrupted stream before releasing
the send lock. The outer actor-cancel scope uses the same deadline for
ack waiting and redelivers pending caller cancellation afterward.
Renamed the completion flag to `cancel_ack_received` and documented that
the current private call consumes `StartAck`, then receives a real
`CancelAck` after `Actor.cancel()` completes; this does not establish
that the OS process exited. Added a source TODO linking issue #506 for
the future first-class `Cancel -> CancelAck` transaction.
Focused transport and actor-cancel verification passed all four tests.
## Files changed
- `tractor/runtime/_portal.py` - own the absolute deadline, accurately
record ack receipt, and link the dedicated cancellation protocol.
- `tractor/runtime/_runtime.py` - forward the optional deadline for the
exact private `Start` publication.
- `tractor/ipc/_chan.py` - pass the operation-specific deadline to the
transport without changing ordinary sends.
- `tractor/ipc/_transport.py` - bound the shielded frame publication and
close a partial-frame stream before unlocking it.
- `tests/ipc/test_each_tpt.py` - cover deadline expiry after a partial
frame prefix reaches the stream.
- `tests/test_to_actor.py` - prove actor-cancel publication and ack
waiting share one absolute timeout budget.
## Human edits
The human rejected the initial watcher-task, shared `_SendStatus`, cancel
coalescing, and per-waiter design as unnecessary complexity. They also
rejected `ContextVar` propagation in favor of explicit functional
threading, selected a single absolute deadline for publication and ack
waiting, and required item 2 to remain separate from the item-3 child
reaping work. After reviewing the result, they requested the precise
`cancel_ack_received` name, a detailed protocol-trace comment, a focused
follow-up issue, and a linked source TODO. No direct source-line edits
were made by the human.

View File

@ -0,0 +1,41 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-21T02:35:37Z
git_ref: ae6f2ac3
diff_cmd: git diff HEAD~1..HEAD
---
Replace the actor-cancel timeout watcher/status experiment with one
explicit absolute deadline threaded through the existing private call
path. Do not use a `ContextVar`, shared result state, waiter
coalescing, or polling tasks.
> `git diff HEAD~1..HEAD -- tractor/runtime/_portal.py`
`Portal.cancel_actor()` computes one absolute deadline and uses it for
both `Start` frame publication and the subsequent cancel-ack wait.
> `git diff HEAD~1..HEAD -- tractor/runtime/_runtime.py`
> `git diff HEAD~1..HEAD -- tractor/ipc/_chan.py`
The private RPC path forwards the operation-specific deadline. Lower
layers preserve the ordinary infinite-deadline call shape.
> `git diff HEAD~1..HEAD -- tractor/ipc/_transport.py`
`MsgpackTransport.send()` applies the deadline inside its complete-frame
shield. If the deadline expires after partial publication, it closes
the unusable stream before releasing the send lock.
> `git diff HEAD~1..HEAD -- tests/ipc/test_each_tpt.py`
> `git diff HEAD~1..HEAD -- tests/test_to_actor.py`
Focused regressions prove a partial-frame timeout closes the stream and
that actor-cancel publication and acknowledgement share one budget.
The implementation removes the earlier `_SendStatus`, watcher task,
coalescing, shared cancel result, and per-waiter state. Four focused
transport and actor-cancel tests pass.

View File

@ -0,0 +1,57 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
timestamp: 2026-08-21T04:08:03Z
git_ref: 3c1bbe73
scope: code
substantive: true
raw_file: 20260821T040803Z_3c1bbe73_prompt_io.raw.md
---
## Prompt
Continue PR #481 review cleanup one item at a time. Compare the
remaining child-registration/reaping experiment against PR #484,
identify the next valid behavior, and generate an exact commit plan for
that boundary without committing automatically.
## Response summary
Separated late-child admission from the broader bounded-reap rewrite.
Added one synchronous `ActorNursery._register_child()` publication step
that stores the child and its reap events before sampling nursery
cancellation. This closes both scheduler orderings around
`ActorNursery.cancel()`'s child snapshot.
The MP backend registers immediately before synchronous process startup
and refuses to start when cancellation already owns the child. The Trio
backend registers immediately after `open_process()` and kills that
already-created process when registration observes cancellation. An
early `start_actor()` guard rejects calls begun after cancellation is
already visible.
Deterministic tests cover the nursery registration ordering and the MP
no-start invariant. Comparison with PR #484 confirmed that its retained
generic nursery/backends do not close this race.
## Files changed
- `tractor/runtime/_supervise.py` - atomically publish child ownership
and reject actor starts after nursery cancellation.
- `tractor/spawn/_mp.py` - register before synchronous process startup
and abort a cancellation-owned child.
- `tractor/spawn/_trio.py` - register immediately after process creation,
kill a cancellation-owned child, and remove its stale unused import.
- `tests/test_to_actor.py` - cover late registration and MP startup
suppression.
## Human edits
The human required review extras to be handled one item and one
behavioral commit at a time, with each item compared against PR #484
before acceptance. That direction split this late-registration fix from
the original broad experiment's bounded post-ack reaping,
`ActorNursery.cancel()` hard-reap rewrite, and debugger/error behavior.
The human accepted the narrower late-registration boundary by requesting
its commit plan. No direct source-line edits were made by the human.

View File

@ -0,0 +1,48 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-21T04:08:03Z
git_ref: 3c1bbe73
diff_cmd: git diff HEAD~1..HEAD
---
Compare the remaining child-registration and reaping experiment with
PR #484, then identify the next review item without changing code.
The next item is the late-child admission race. A spawn can pass
`ActorNursery.start_actor()`'s early cancellation check, then be absent
from `ActorNursery.cancel()`'s child snapshot and register afterward.
The existing reap-request latch releases its monitor but does not send
runtime cancellation, so the monitor can wait forever for a still-live
process.
> `git diff HEAD~1..HEAD -- tractor/runtime/_supervise.py`
`ActorNursery._register_child()` publishes the child, installs its reap
events, and samples `ActorNursery._cancel_called` without a checkpoint.
The two scheduler orderings are then complete: registration first puts
the child in the cancel snapshot, while cancellation first makes the
backend abort the late registration.
> `git diff HEAD~1..HEAD -- tractor/spawn/_mp.py`
The multiprocessing backend registers immediately before `proc.start()`
and refuses to start a process already owned by nursery cancellation.
There is no Trio checkpoint between registration and process startup.
> `git diff HEAD~1..HEAD -- tractor/spawn/_trio.py`
The Trio backend registers immediately after `open_process()` and kills
the newly opened process if cancellation won the registration race. Its
stale unused `get_runtime_vars` import is removed so the touched module
remains lint-clean.
> `git diff HEAD~1..HEAD -- tests/test_to_actor.py`
Deterministic regressions prove late registration observes cancellation
and that the MP backend never starts a process after cancellation owns
its registration.
PR #484 retains the affected generic nursery and spawn-backend paths and
does not close this race. Keep this fix in PR #481 as its own commit;
review bounded post-`CancelAck` reaping separately.

View File

@ -54,15 +54,16 @@ One-shot task actors
.. note:: .. note::
:func:`tractor.to_actor.run` (parlance of Without ``portal=``, :func:`tractor.to_actor.run` (parlance of
``trio.to_thread.run_sync()`` and friends) is the ``trio.to_thread.run_sync()`` and friends) is the convenience
*convenience* one-shot — spawn, run a single task, block on one-shot: spawn, run one task, block on its result and reap. It
its result, reap — built entirely on combines :meth:`ActorNursery.start_actor`, a linked
:meth:`ActorNursery.start_actor`, a linked :meth:`Portal.open_context` call and per-child reaping. With
:meth:`Portal.open_context` call and per-child cancellation/reaping, ``portal=`` it owns only the linked task and leaves the existing
so don't design around it as the core model. It supersedes the actor's lifetime to the portal owner; that actor must expose both
legacy, non-blocking ``ActorNursery.run_in_actor()`` retained only the target module and ``tractor.to_actor.MODULE``. It supersedes
for compatibility until its removal in PR #484. the legacy, non-blocking ``ActorNursery.run_in_actor()`` retained
only for compatibility until its removal in PR #484.
.. deprecated:: 0.1.0a6 .. deprecated:: 0.1.0a6

View File

@ -62,7 +62,10 @@ one kwarg away,
.. code:: python .. code:: python
async with tractor.open_actor_cluster( async with tractor.open_actor_cluster(
modules=['mylib.workers'], modules=[
'mylib.workers',
tractor.to_actor.MODULE,
],
count=4, count=4,
names=['scout', 'miner', 'smelter', 'smith'], names=['scout', 'miner', 'smelter', 'smith'],
debug_mode=True, # whole-fleet crash-to-REPL debug_mode=True, # whole-fleet crash-to-REPL
@ -71,9 +74,11 @@ one kwarg away,
From here the composition patterns are the usual ``tractor`` fare: From here the composition patterns are the usual ``tractor`` fare:
``portal.run()`` for bare one-shot RPCs (as in the demo), ``portal.run()`` for bare one-shot RPCs (as in the demo),
``tractor.to_actor.run(..., portal=portal)`` for linked one-shot calls, ``tractor.to_actor.run(..., portal=portal)`` for cancellation-linked
or — for a persistent bidirectional dialog per worker — concurrently one-shot tasks in an existing worker (include
enter N ``portal.open_context()`` blocks with ``tractor.to_actor.MODULE`` in ``modules``; the cluster still owns
the worker's lifetime), or — for a persistent bidirectional dialog
per worker — concurrently enter N ``portal.open_context()`` blocks with
``tractor.trionics.gather_contexts()``; see :doc:`/guide/context` ``tractor.trionics.gather_contexts()``; see :doc:`/guide/context`
for that whole layer. for that whole layer.

View File

@ -15,12 +15,12 @@ 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: ``tractor.to_actor.run()`` is a convenience for one primitive: ``tractor.to_actor.run()`` uses it for a linked
"spawn, run the lone task, await the result, tear down"; plain one-shot task, spawning and reaping an actor only when no ``portal=``
``Portal.run()`` RPC is planned to be re-implemented on top of it; is supplied; plain ``Portal.run()`` RPC is planned to be
the multi-process debugger's tree-wide REPL lock rides one. Grok re-implemented on top of it; the multi-process debugger's tree-wide
this page and the rest of the library reads as convenience REPL lock rides one. Grok this page and the rest of the library reads
wrappers B) as convenience wrappers B)
The endpoint contract The endpoint contract
--------------------- ---------------------

View File

@ -82,10 +82,9 @@ don't build your app on it.
One-shot subactors: ``to_actor.run()`` One-shot subactors: ``to_actor.run()``
-------------------------------------- --------------------------------------
When a subactor's *entire job* is a single function call, skip When the call should own a fresh subactor whose entire job is one
the portal plumbing with :func:`tractor.to_actor.run`: spawn, function call, :func:`tractor.to_actor.run` spawns it, runs the task,
run the lone task, return its result and reap the process — all returns its result and reaps the process — all in one blocking call:
in one blocking call:
.. code:: python .. code:: python
@ -101,16 +100,37 @@ 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 right in the calling remote error in the usual boxed form right in the calling
task. task.
- "placement" is composable: ``an=`` spawns from an existing - placement also determines process ownership: ``an=`` spawns and
actor-nursery, ``portal=`` reuses an already-running actor reaps a fresh child in an existing actor nursery, while passing
(no spawn/reap, just a linked neither does the same in a private call-scoped nursery (booting
:meth:`~tractor.Portal.open_context` call; see the the runtime if needed). ``portal=`` instead runs one linked task
:doc:`context guide </guide/context>`), and passing neither in an existing actor; it neither spawns nor reaps that actor, so
opens a private call-scoped nursery (booting the runtime if needed). the portal's owner remains responsible for its lifetime.
- concurrency composes the plain ``trio`` way: schedule - concurrency composes the plain ``trio`` way: schedule
multiple ``run()`` calls into a local task nursery (see multiple ``run()`` calls into a local task nursery (see
``examples/parallelism/to_actor_one_shots.py``). ``examples/parallelism/to_actor_one_shots.py``).
A reused actor must expose both the target module and the
``to_actor`` context trampoline:
.. code:: python
async with tractor.open_nursery() as an:
portal = await an.start_actor(
'worker',
enable_modules=[
__name__,
tractor.to_actor.MODULE,
],
)
try:
final = await tractor.to_actor.run(
partial(fib, n=10),
portal=portal,
)
finally:
await portal.cancel_actor()
Pure RPC daemons: ``run_daemon()`` Pure RPC daemons: ``run_daemon()``
---------------------------------- ----------------------------------
When a process's *only* job is to sit at the root of its own When a process's *only* job is to sit at the root of its own

View File

@ -105,9 +105,9 @@ What's going on here?
``to_actor.run()``: quick one-shot parallelism ``to_actor.run()``: quick one-shot parallelism
---------------------------------------------- ----------------------------------------------
:func:`tractor.to_actor.run` is the convenience wrapper: spawn Without ``portal=``, :func:`tractor.to_actor.run` is the convenience
an actor, run exactly one async function in it, block on the wrapper: spawn an actor, run exactly one async function in it, block
result, then reap the process — the distributed sibling of on the result, then reap the process — the distributed sibling of
``trio.to_thread.run_sync()``. ``trio.to_thread.run_sync()``.
.. code:: python .. code:: python
@ -126,6 +126,10 @@ A few details worth knowing:
``name='something_cuter'``. ``name='something_cuter'``.
- the function's module is auto-added to the child's - the function's module is auto-added to the child's
``enable_modules`` allowlist. ``enable_modules`` allowlist.
- targets cross IPC as ``module:name`` references, so portable calls
use module-global async functions or ``functools.partial`` objects
wrapping them. Nested functions, methods and callable objects do not
provide that stable address.
- target arguments are positional; use ``functools.partial()`` - target arguments are positional; use ``functools.partial()``
to bind target keyword arguments. Keywords passed directly to to bind target keyword arguments. Keywords passed directly to
``run()`` configure actor placement and spawning. ``run()`` configure actor placement and spawning.
@ -133,18 +137,23 @@ A few details worth knowing:
child is *auto-cancelled* (reaped) right after — so remote child is *auto-cancelled* (reaped) right after — so remote
errors raise directly in your calling task (causality_ is errors raise directly in your calling task (causality_ is
paramount!). paramount!).
- "placement" composes: ``an=`` spawns from a caller-managed - "placement" composes: ``an=`` spawns a call-owned child from an
actor-nursery, ``portal=`` reuses an already-running actor existing actor nursery, while passing neither opens a private
(no spawn/reap), and passing neither opens a private call-scoped nursery. ``portal=`` instead reuses an existing actor:
call-scoped nursery (booting the runtime if needed). the call scopes only its linked remote task, neither spawns nor
reaps the actor, and leaves its lifetime with the portal's owner.
That actor must expose both the target module and
``tractor.to_actor.MODULE``.
.. note:: .. note::
:func:`tractor.to_actor.run` is a convenience, **not** the core :func:`tractor.to_actor.run` is a convenience, **not** the core
model — it's built *entirely* on model. For actor-owning placements it combines
:meth:`~tractor.ActorNursery.start_actor` plus a linked :meth:`~tractor.ActorNursery.start_actor`, a linked
:meth:`~tractor.Portal.open_context` call and per-child :meth:`~tractor.Portal.open_context` call, and per-child
cancellation/reaping. Teach your fingers to use it for quick cancellation/reaping. With ``portal=`` it uses only the linked
context call and leaves the existing actor's lifetime untouched.
Teach your fingers to use it for quick
fire-and-collect parallelism — think a per-function trio-parallel_ fire-and-collect parallelism — think a per-function trio-parallel_
style one-shot — and reach for style one-shot — and reach for
:meth:`~tractor.ActorNursery.start_actor` plus :meth:`~tractor.ActorNursery.start_actor` plus
@ -153,25 +162,25 @@ A few details worth knowing:
Actor lifetimes and teardown order Actor lifetimes and teardown order
---------------------------------- ----------------------------------
So we have two lifetime flavors: There are two actor-lifetime flavors:
- **one-shot** (``to_actor.run()``): lives exactly as long as - **call-owned one-shot** (``to_actor.run()`` without ``portal=``):
its single task; reaped the moment its result (or error) spawned for one task, then cancelled and joined before ``run()``
arrives back in the (blocking) call. returns its result or raises its error.
- **daemon** (:meth:`~tractor.ActorNursery.start_actor`): lives - **caller-owned daemon** (:meth:`~tractor.ActorNursery.start_actor`),
until *someone* cancels it — an explicit including an actor later reused through
``to_actor.run(..., portal=portal)``: lives until *someone*
cancels it via an explicit
:meth:`~tractor.Portal.cancel_actor`, a bulk :meth:`~tractor.Portal.cancel_actor`, a bulk
:meth:`~tractor.ActorNursery.cancel`, or the one-cancels-all :meth:`~tractor.ActorNursery.cancel`, or the one-cancels-all
strategy kicking in on error. strategy kicking in on error.
On a clean exit of the nursery block the teardown order is: On a clean exit of the nursery block the teardown order is:
1. one-shot actors never make it to nursery exit: each is 1. call-owned actors do not survive their own ``to_actor.run()``
reaped inside its own ``to_actor.run()`` call, any error calls; each is reaped before its call returns.
raising immediately in the calling task so your code 2. the nursery waits on caller-owned daemon actors
(acting as supervisor) gets first crack at handling it. **indefinitely**. If you spawned one, 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,15 +43,16 @@ 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.to_actor.run()`` runtime boots *implicitly* inside this ``tractor.to_actor.run()``
whenever it isn't already up. No special entrypoint, no call because neither ``an=`` nor ``portal=`` was supplied. No
framework takeover - it's just a ``trio`` app, special entrypoint, no framework takeover - it's just a ``trio``
app,
- inside ``main()`` a *subactor* is spawned via - inside ``main()`` a *subactor* is spawned via
``tractor.to_actor.run()`` and told to run exactly one ``tractor.to_actor.run()`` and told to run exactly one
function: ``cellar_door()``, function: ``cellar_door()``,
- 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 linked
task* (note the child proving it is *not* the root with one-shot 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 call *blocks* until that final result arrives, then - the call *blocks* until that final result arrives, then
@ -67,10 +68,10 @@ What's going on here?
.. note:: .. note::
``to_actor.run()`` (parlance of ``trio.to_thread`` and Without ``portal=``, ``to_actor.run()`` (parlance of
friends) is the *convenience* wrapper: one-shot ``trio.to_thread`` and friends) is the *convenience* wrapper:
spawn-run-reap semantics for when a subactor's entire job is one-shot spawn-run-reap semantics for when a subactor's entire
a single function call. The core primitives are job is a single function call. The core primitives are
:meth:`~tractor.ActorNursery.start_actor` (next up) — which :meth:`~tractor.ActorNursery.start_actor` (next up) — which
hands you a ``Portal``, your handle for invoking tasks in the hands you a ``Portal``, your handle for invoking tasks in the
new process's (separate!) memory domain — paired with new process's (separate!) memory domain — paired with
@ -79,10 +80,10 @@ What's going on here?
Daemon actors and RPC Daemon actors and RPC
--------------------- ---------------------
A ``to_actor.run()`` one-shot subactor terminates when its lone A subactor spawned by ``to_actor.run()`` terminates after its lone
task returns. But often you want long-lived *daemon* actors task returns. But often you want long-lived *daemon* actors instead:
instead: spawned once, then serving (allowlisted) RPC requests spawned once, then serving (allowlisted) RPC requests until told
until told otherwise. That's ``start_actor()``: 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
@ -90,14 +91,17 @@ until told otherwise. That's ``start_actor()``:
Two lifetime rules to internalize: Two lifetime rules to internalize:
- a ``to_actor.run()`` one-shot actor lives exactly as long as - a subactor spawned and owned by ``to_actor.run()`` is cancelled
its lone task; the call blocks until that function (and thus and reaped before the call returns its result or raises its error,
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
above), or its parent nursery is cancelled wholesale. above), or its parent nursery is cancelled wholesale.
Passing ``portal=`` is different: the call owns only the linked
remote task. It neither spawns nor reaps the existing actor; the
portal's owner must end that actor's lifetime.
.. tip:: .. tip::
Want your *entire program* to just be a long-lived RPC Want your *entire program* to just be a long-lived RPC

View File

@ -0,0 +1,3 @@
Add ``tractor.to_actor.run()`` for Trio-style one-shot async calls in
new or existing actors, with caller-scoped result/error propagation,
linked cancellation, and deterministic reaping of call-owned children.

View File

@ -15,7 +15,10 @@ from unittest.mock import Mock
import pytest import pytest
import trio import trio
from trio.testing import wait_all_tasks_blocked from trio.testing import (
MockClock,
wait_all_tasks_blocked,
)
import tractor import tractor
from tractor import Actor from tractor import Actor
from tractor.discovery import _addr from tractor.discovery import _addr
@ -134,6 +137,65 @@ def test_cancelled_transport_send_completes_frame():
trio.run(main) trio.run(main)
def test_transport_send_deadline_closes_partial_frame():
'''
Bound one shielded frame without exposing a corrupt stream.
Ordinary cancellation cannot interrupt complete-frame publication.
Actor-wide cancellation instead passes its absolute deadline into
this operation. The fake stream writes a partial header and stalls;
when the send's own deadline fires, the transport must close the
stream before releasing its shared send lock and report the channel
unusable.
'''
class StalledStream:
def __init__(self) -> None:
self.closed = False
self.wire = bytearray()
async def send_all(
self,
data: bytes,
) -> None:
self.wire.extend(data[:2])
await trio.sleep_forever()
async def aclose(self) -> None:
self.closed = True
async def main() -> None:
stream = StalledStream()
transport = object.__new__(MsgpackTransport)
transport.stream = stream
transport._send_lock = trio.StrictFIFOLock()
msg = tractor.msg.Start(
ns=__name__,
func='add_one',
kwargs={'n': 1},
uid=('root', 'test'),
cid='deadline-send',
)
with pytest.raises(
tractor.TransportClosed,
match='frame publication exceeded',
):
await transport.send(
msg,
send_deadline=1,
)
assert stream.closed
assert len(stream.wire) == 2
assert not transport._send_lock.locked()
trio.run(
main,
clock=MockClock(autojump_threshold=0),
)
def test_cancelled_transport_send_preserves_cancellation(): def test_cancelled_transport_send_preserves_cancellation():
''' '''
Prefer sender cancellation when teardown closes the stream. Prefer sender cancellation when teardown closes the stream.

View File

@ -8,17 +8,21 @@ https://github.com/goodboy/tractor/issues/477
''' '''
from functools import partial from functools import partial
from pathlib import Path from pathlib import Path
from types import SimpleNamespace
import pytest import pytest
import trio import trio
from trio.testing import MockClock
import tractor import tractor
from tractor import ( from tractor import (
RemoteActorError, RemoteActorError,
to_actor, to_actor,
) )
from tractor._testing import tractor_test from tractor._testing import tractor_test
from tractor._exceptions import ActorTooSlowError
from tractor.msg import ptr as msgptr from tractor.msg import ptr as msgptr
from tractor.msg.ptr import NamespacePath from tractor.msg.ptr import NamespacePath
from tractor.spawn import _mp as mp_spawn
from tractor.to_actor import _api as to_actor_api from tractor.to_actor import _api as to_actor_api
@ -254,6 +258,186 @@ async def test_cancel_ack_failure_hard_reaps_child(
assert not an._children assert not an._children
def test_cancel_actor_timeout_closes_blocked_send():
'''
Thread one absolute cancel deadline into shielded frame publication.
The cancel RPC's outer timeout cannot penetrate a complete-frame
shield. The fake private RPC applies the forwarded send deadline to
its own shielded wait, then checkpoints into the outer scope. A
bounded `ActorTooSlowError` and the recorded absolute deadline prove
publication and acknowledgement share one timeout budget.
'''
class ConnectedChannel:
def __init__(self) -> None:
self._cancel_called = False
self.aid = tractor.msg.Aid(
name='blocked_peer',
uuid='test',
)
def connected(self) -> bool:
return True
async def main() -> None:
channel = ConnectedChannel()
portal = object.__new__(tractor.Portal)
portal._chan = channel
deadlines: list[float] = []
async def blocked_cancel(
namespace: str,
function: str,
kwargs: dict[str, object],
cancel_on_startup: bool,
send_deadline: float,
) -> None:
assert (namespace, function) == ('self', 'cancel')
assert kwargs == {}
assert not cancel_on_startup
deadlines.append(send_deadline)
with trio.CancelScope(
deadline=send_deadline,
shield=True,
):
await trio.sleep_forever()
await trio.lowlevel.checkpoint_if_cancelled()
portal._run_from_ns = blocked_cancel
with pytest.raises(ActorTooSlowError):
await portal.cancel_actor(
timeout=1,
raise_on_timeout=True,
)
assert deadlines == [1.]
trio.run(
main,
clock=MockClock(autojump_threshold=0),
)
def _mock_actor_nursery() -> tractor.ActorNursery:
an = object.__new__(tractor.ActorNursery)
an._children = {}
an._join_procs = trio.Event()
an._child_reap_requests = {}
an._child_reaped = {}
an._at_least_one_child_in_debug = False
an._cancel_called = False
return an
def test_late_child_registration_observes_cancel():
'''
Make registration atomically observe nursery cancellation.
`ActorNursery.cancel()` previously snapshotted `_children` before
its next checkpoint. A process monitor registering after that
snapshot received a reap request but no runtime cancellation, then
waited forever for natural exit. Publishing the child and its reap
events together returns cancellation ownership to the late monitor.
'''
an = _mock_actor_nursery()
an._cancel_called = True
aid = tractor.msg.Aid(
name='late_child',
uuid='test',
)
subactor = SimpleNamespace(aid=aid)
proc = object()
(
reap_request,
reaped,
cancel_during_registration,
) = an._register_child(
subactor,
proc,
None,
)
assert cancel_during_registration
assert an._children[aid.uid] == (
subactor,
proc,
None,
)
assert an._child_reap_requests[aid.uid] is reap_request
assert an._child_reaped[aid.uid] is reaped
def test_mp_late_registration_never_starts_process(
monkeypatch: pytest.MonkeyPatch,
):
'''
Refuse to start an MP child already owned by nursery cancellation.
A concurrent `ActorNursery.cancel()` can publish cancellation after
`start_actor()` checks its flag but before the MP backend registers
its process. The fake registration reports that exact schedule.
Proving `FakeProcess.start()` is never called prevents a child from
starting after it was omitted from the cancellation snapshot.
'''
class FakeProcess:
started: bool = False
def start(self) -> None:
self.started = True
process = FakeProcess()
class FakeContext:
def get_start_method(self) -> str:
return 'spawn'
def Process(self, **kwargs: object) -> FakeProcess:
assert kwargs
return process
nursery = SimpleNamespace(
_register_child=lambda *args: (
trio.Event(),
trio.Event(),
True,
),
)
subactor = SimpleNamespace(
aid=tractor.msg.Aid(
name='late_mp_child',
uuid='test',
),
)
monkeypatch.setattr(
mp_spawn._spawn,
'_ctx',
FakeContext(),
)
with pytest.raises(
RuntimeError,
match='nursery began cancelling',
):
trio.run(
partial(
mp_spawn.mp_proc,
name='late_mp_child',
actor_nursery=nursery,
subactor=subactor,
errors={},
bind_addrs=[],
parent_addr=SimpleNamespace(),
_runtime_vars={},
)
)
assert not process.started
def test_late_child_reap_registration_is_released(): def test_late_child_reap_registration_is_released():
''' '''
Preserve a nursery-wide reap request across child startup. Preserve a nursery-wide reap request across child startup.

View File

@ -310,6 +310,7 @@ class Channel:
payload: Any, payload: Any,
hide_tb: bool = False, hide_tb: bool = False,
send_deadline: float = float('inf'),
) -> None: ) -> None:
''' '''
@ -320,6 +321,9 @@ class Channel:
expected-graceful cases, normally ephemercal expected-graceful cases, normally ephemercal
(re/dis)connects. (re/dis)connects.
`send_deadline` is an absolute Trio clock deadline forwarded
only to transports that support bounded frame publication.
''' '''
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
try: try:
@ -330,10 +334,17 @@ class Channel:
f'{pformat(payload)}\n' f'{pformat(payload)}\n'
) )
# assert self._transport # but why typing? # assert self._transport # but why typing?
await self._transport.send( if send_deadline == float('inf'):
payload, await self._transport.send(
hide_tb=hide_tb, payload,
) hide_tb=hide_tb,
)
else:
await self._transport.send(
payload,
hide_tb=hide_tb,
send_deadline=send_deadline,
)
except ( except (
BaseException, BaseException,
MsgTypeError, MsgTypeError,

View File

@ -439,6 +439,7 @@ class MsgpackTransport(MsgTransport):
strict_types: bool = True, strict_types: bool = True,
hide_tb: bool = True, hide_tb: bool = True,
send_deadline: float = float('inf'),
) -> None: ) -> None:
''' '''
@ -447,6 +448,10 @@ class MsgpackTransport(MsgTransport):
If `strict_types == True` then a `MsgTypeError` will be raised on any If `strict_types == True` then a `MsgTypeError` will be raised on any
invalid msg type invalid msg type
`send_deadline` bounds publication of this complete frame. A
timeout destroys the stream because a partial prefix may have
reached the wire.
''' '''
__tracebackhide__: bool = hide_tb __tracebackhide__: bool = hide_tb
@ -513,12 +518,26 @@ class MsgpackTransport(MsgTransport):
# the frame is complete, the explicit checkpoint # the frame is complete, the explicit checkpoint
# immediately delivers any pending cancellation. # immediately delivers any pending cancellation.
# #
# This can delay cancellation while a peer is not # Ordinary sends may delay cancellation while a peer is
# reading; peer/channel teardown must close the stream # not reading. Actor-wide cancel requests pass their own
# to unblock a permanently stalled socket write. # deadline so this operation can close a stalled stream.
with trio.CancelScope(shield=True): with trio.CancelScope(
deadline=send_deadline,
shield=True,
) as send_cs:
await self.stream.send_all(size + bytes_data) await self.stream.send_all(size + bytes_data)
if send_cs.cancelled_caught:
# This frame may be partial. Destroy the stream
# before releasing `_send_lock` so no later sender
# can append bytes to a corrupted frame.
await trio.aclose_forcefully(self.stream)
await trio.lowlevel.checkpoint_if_cancelled()
raise TransportClosed(
'IPC frame publication exceeded its '
f'deadline of {send_deadline!r}'
)
await trio.lowlevel.checkpoint_if_cancelled() await trio.lowlevel.checkpoint_if_cancelled()
return None return None

View File

@ -319,45 +319,61 @@ class Portal:
or or
self.cancel_timeout self.cancel_timeout
) )
cancel_deadline: float = (
trio.current_time()
+
cancel_timeout
)
# NOTE: Actor-runtime cancellation currently rides the normal
# RPC envelope:
#
# `Start(self.cancel)` -> `StartAck` -> `CancelAck`.
#
# `Actor.start_remote_task()` consumes the `StartAck`, then
# `._run_from_ns()` returns only after `PldRx.recv_pld()`
# decodes the final `CancelAck`. Thus this flag means that ack
# reached this portal after the peer's `Actor.cancel()` routine
# completed; it does not prove the peer OS process has exited.
# A dedicated `Cancel` request msg can eventually replace the
# internal `Start` RPC envelope and its extra `StartAck`.
cancel_ack_received: bool = False
try: try:
# send cancel cmd - might not get response with trio.move_on_at(cancel_deadline) as cs:
# XXX: sure would be nice to make this work with
# a proper shield
with trio.move_on_after(cancel_timeout) as cs:
cs.shield: bool = True cs.shield: bool = True
await self.run_from_ns( await self._run_from_ns(
'self', 'self',
'cancel', 'cancel',
kwargs={},
cancel_on_startup=False,
send_deadline=cancel_deadline,
) )
return True cancel_ack_received = True
# `move_on_after` fired — peer didn't ack within # Preserve shielded actor teardown, then immediately
# redeliver any cancellation pending from an outer scope.
await trio.lowlevel.checkpoint_if_cancelled()
# `move_on_at` fired — peer didn't ack within
# bounded window. Behaviour depends on # bounded window. Behaviour depends on
# `raise_on_timeout`: # `raise_on_timeout`:
if ( if cs.cancelled_caught:
cs.cancelled_caught if raise_on_timeout:
and raise ActorTooSlowError(
raise_on_timeout f'Peer {peer_id} did not ack its '
): f'`Actor.cancel()` RPC within bounded wait '
raise ActorTooSlowError( f'of {cancel_timeout!r}s'
f'Peer {peer_id} did not ack its ' )
f'`Actor.cancel()` RPC within bounded wait '
f'of {cancel_timeout!r}s'
)
# legacy fire-and-forget path: log + return False so # Legacy fire-and-forget callers decide whether to
# the caller can decide whether to escalate. # escalate the missed acknowledgement themselves.
# log.debug(
# NOTE, we also land here in the (unexpected) case where f'May have failed to cancel peer?\n'
# the shielded `move_on_after` block exits WITHOUT f'\n'
# `return True` and WITHOUT the deadline firing — prefer f'c)=?> {peer_id}\n'
# a soft `False` over an `assert`-crash mid-teardown. )
log.debug( return False
f'May have failed to cancel peer?\n'
f'\n' return cancel_ack_received
f'c)=?> {peer_id}\n'
)
return False
except TransportClosed as tpt_err: except TransportClosed as tpt_err:
ipc_borked_report: str = ( ipc_borked_report: str = (
@ -379,16 +395,24 @@ class Portal:
return False return False
# TODO: Replace actor-runtime cancellation's internal
# `Start -> StartAck -> CancelAck` RPC with a dedicated
# `Cancel -> CancelAck` transaction:
# https://github.com/goodboy/tractor/issues/506
async def _run_from_ns( async def _run_from_ns(
self, self,
namespace_path: str, namespace_path: str,
function_name: str, function_name: str,
kwargs: dict[str, Any], kwargs: dict[str, Any],
cancel_on_startup: bool = True, cancel_on_startup: bool = True,
send_deadline: float = float('inf'),
) -> Any: ) -> Any:
''' '''
Run a namespace target with local startup policy controls. Run a namespace target with local startup policy controls.
`send_deadline` bounds only publication of the `Start` frame;
the caller owns any larger RPC/acknowledgement deadline.
''' '''
nsf = NamespacePath( nsf = NamespacePath(
f'{namespace_path}:{function_name}' f'{namespace_path}:{function_name}'
@ -399,6 +423,7 @@ class Portal:
kwargs=kwargs, kwargs=kwargs,
portal=self, portal=self,
cancel_on_startup=cancel_on_startup, cancel_on_startup=cancel_on_startup,
send_deadline=send_deadline,
) )
try: try:
return await ctx._pld_rx.recv_pld( return await ctx._pld_rx.recv_pld(

View File

@ -793,6 +793,11 @@ class Actor:
ack_timeout: float = float('inf'), ack_timeout: float = float('inf'),
cancel_on_startup: bool = True, cancel_on_startup: bool = True,
# Optional absolute deadline for publishing this exact `Start`
# frame. Used by actor-wide cancel RPCs whose outer timeout
# cannot penetrate complete-frame transport shielding.
send_deadline: float = float('inf'),
) -> Context: ) -> Context:
''' '''
Send a `'cmd'` msg to a remote actor, which requests the Send a `'cmd'` msg to a remote actor, which requests the
@ -845,7 +850,13 @@ class Actor:
) )
start_published: bool = False start_published: bool = False
try: try:
await chan.send(msg) if send_deadline == float('inf'):
await chan.send(msg)
else:
await chan.send(
msg,
send_deadline=send_deadline,
)
start_published = True start_published = True
# NOTE wait on first `StartAck` response msg and validate; # NOTE wait on first `StartAck` response msg and validate;

View File

@ -327,6 +327,29 @@ class ActorNursery:
reap_request.set() reap_request.set()
return reap_request, reaped return reap_request, reaped
def _register_child(
self,
subactor: Actor,
proc: 'ProcessType',
portal: Portal|None,
) -> tuple[trio.Event, trio.Event, bool]:
'''
Atomically publish one child and its reap coordination.
'''
uid: tuple[str, str] = subactor.aid.uid
self._children[uid] = (
subactor,
proc,
portal,
)
reap_request, reaped = self._register_child_reap(uid)
return (
reap_request,
reaped,
self._cancel_called,
)
def _request_reap_all(self) -> None: def _request_reap_all(self) -> None:
''' '''
Release every child monitor into its process-join phase. Release every child monitor into its process-join phase.
@ -419,6 +442,12 @@ class ActorNursery:
''' '''
__runtimeframe__: int = 1 # noqa __runtimeframe__: int = 1 # noqa
if self._cancel_called:
raise RuntimeError(
'Cannot start an actor in a cancelling '
'`ActorNursery`'
)
loglevel: str = ( loglevel: str = (
loglevel loglevel
or self._actor.loglevel or self._actor.loglevel

View File

@ -138,12 +138,22 @@ async def mp_proc(
# daemon=True, # daemon=True,
name=name, name=name,
) )
# `multiprocessing` only (since no async interface): publish the
# `multiprocessing` only (since no async interface): # process and its reap coordination before start so cancellation
# register the process before start in case we get a cancel # can own every subsequently started child.
# request before the actor has fully spawned - then we can wait (
# for it to fully come up before sending a cancel request reap_request,
actor_nursery._children[subactor.aid.uid] = (subactor, proc, None) _,
cancel_during_registration,
) = actor_nursery._register_child(
subactor,
proc,
None,
)
if cancel_during_registration:
raise RuntimeError(
'Actor registered after its nursery began cancelling'
)
proc.start() proc.start()
if not proc.is_alive(): if not proc.is_alive():
@ -170,9 +180,6 @@ async def mp_proc(
# any process we may have started. # any process we may have started.
portal = Portal(chan) portal = Portal(chan)
reap_request, _ = actor_nursery._register_child_reap(
subactor.aid.uid,
)
actor_nursery._children[subactor.aid.uid] = (subactor, proc, portal) actor_nursery._children[subactor.aid.uid] = (subactor, proc, portal)
# unblock parent task # unblock parent task

View File

@ -39,7 +39,6 @@ from tractor.runtime._state import (
current_actor, current_actor,
is_root_process, is_root_process,
debug_mode, debug_mode,
get_runtime_vars,
) )
from tractor.log import get_logger from tractor.log import get_logger
from tractor.discovery._addr import UnwrappedAddress from tractor.discovery._addr import UnwrappedAddress
@ -131,6 +130,23 @@ async def trio_proc(
f' |_{proc}\n' f' |_{proc}\n'
) )
(
reap_request,
_,
cancel_during_registration,
) = actor_nursery._register_child(
subactor,
proc,
None,
)
if cancel_during_registration:
cancelled_during_spawn = True
proc.kill()
raise RuntimeError(
'Actor registered after its nursery began '
'cancelling'
)
# wait for actor to spawn and connect back to us # wait for actor to spawn and connect back to us
# channel should have handshake completed by the # channel should have handshake completed by the
# local actor by the time we get a ref to it # local actor by the time we get a ref to it
@ -161,9 +177,6 @@ async def trio_proc(
assert proc assert proc
portal = Portal(chan) portal = Portal(chan)
reap_request, _ = actor_nursery._register_child_reap(
subactor.aid.uid,
)
actor_nursery._children[subactor.aid.uid] = ( actor_nursery._children[subactor.aid.uid] = (
subactor, subactor,
proc, proc,