Compare commits

..

No commits in common. "70f497bf5f0e211774183f08ea90757ae3295d4f" and "1b8ae7933ad0a0f011007861db0c0af49daab10e" have entirely different histories.

20 changed files with 120 additions and 734 deletions

View File

@ -1,65 +0,0 @@
---
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

@ -1,41 +0,0 @@
---
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

@ -1,57 +0,0 @@
---
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

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

View File

@ -62,10 +62,7 @@ one kwarg away,
.. code:: python .. code:: python
async with tractor.open_actor_cluster( async with tractor.open_actor_cluster(
modules=[ modules=['mylib.workers'],
'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
@ -74,11 +71,9 @@ 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 cancellation-linked ``tractor.to_actor.run(..., portal=portal)`` for linked one-shot calls,
one-shot tasks in an existing worker (include or — for a persistent bidirectional dialog per worker — concurrently
``tractor.to_actor.MODULE`` in ``modules``; the cluster still owns enter N ``portal.open_context()`` blocks with
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()`` uses it for a linked one primitive: ``tractor.to_actor.run()`` is a convenience for
one-shot task, spawning and reaping an actor only when no ``portal=`` "spawn, run the lone task, await the result, tear down"; plain
is supplied; plain ``Portal.run()`` RPC is planned to be ``Portal.run()`` RPC is planned to be re-implemented on top of it;
re-implemented on top of it; the multi-process debugger's tree-wide the multi-process debugger's tree-wide REPL lock rides one. Grok
REPL lock rides one. Grok this page and the rest of the library reads this page and the rest of the library reads as convenience
as convenience wrappers B) wrappers B)
The endpoint contract The endpoint contract
--------------------- ---------------------

View File

@ -82,9 +82,10 @@ don't build your app on it.
One-shot subactors: ``to_actor.run()`` One-shot subactors: ``to_actor.run()``
-------------------------------------- --------------------------------------
When the call should own a fresh subactor whose entire job is one When a subactor's *entire job* is a single function call, skip
function call, :func:`tractor.to_actor.run` spawns it, runs the task, the portal plumbing with :func:`tractor.to_actor.run`: spawn,
returns its result and reaps the process — all in one blocking call: run the lone task, return its result and reap the process — all
in one blocking call:
.. code:: python .. code:: python
@ -100,37 +101,16 @@ 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 also determines process ownership: ``an=`` spawns and - "placement" is composable: ``an=`` spawns from an existing
reaps a fresh child in an existing actor nursery, while passing actor-nursery, ``portal=`` reuses an already-running actor
neither does the same in a private call-scoped nursery (booting (no spawn/reap, just a linked
the runtime if needed). ``portal=`` instead runs one linked task :meth:`~tractor.Portal.open_context` call; see the
in an existing actor; it neither spawns nor reaps that actor, so :doc:`context guide </guide/context>`), and passing neither
the portal's owner remains responsible for its lifetime. opens a private call-scoped nursery (booting the runtime if needed).
- concurrency composes the plain ``trio`` way: schedule - concurrency composes the plain ``trio`` way: schedule
multiple ``run()`` calls into a local task nursery (see multiple ``run()`` calls into a local task nursery (see
``examples/parallelism/to_actor_one_shots.py``). ``examples/parallelism/to_actor_one_shots.py``).
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
---------------------------------------------- ----------------------------------------------
Without ``portal=``, :func:`tractor.to_actor.run` is the convenience :func:`tractor.to_actor.run` is the convenience wrapper: spawn
wrapper: spawn an actor, run exactly one async function in it, block an actor, run exactly one async function in it, block on the
on the result, then reap the process — the distributed sibling of result, then reap the process — the distributed sibling of
``trio.to_thread.run_sync()``. ``trio.to_thread.run_sync()``.
.. code:: python .. code:: python
@ -126,10 +126,6 @@ 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.
@ -137,23 +133,18 @@ 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 a call-owned child from an - "placement" composes: ``an=`` spawns from a caller-managed
existing actor nursery, while passing neither opens a private actor-nursery, ``portal=`` reuses an already-running actor
call-scoped nursery. ``portal=`` instead reuses an existing actor: (no spawn/reap), and passing neither opens a private
the call scopes only its linked remote task, neither spawns nor call-scoped nursery (booting the runtime if needed).
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. For actor-owning placements it combines model — it's built *entirely* on
:meth:`~tractor.ActorNursery.start_actor`, a linked :meth:`~tractor.ActorNursery.start_actor` plus a linked
:meth:`~tractor.Portal.open_context` call, and per-child :meth:`~tractor.Portal.open_context` call and per-child
cancellation/reaping. With ``portal=`` it uses only the linked cancellation/reaping. Teach your fingers to use it for quick
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
@ -162,25 +153,25 @@ A few details worth knowing:
Actor lifetimes and teardown order Actor lifetimes and teardown order
---------------------------------- ----------------------------------
There are two actor-lifetime flavors: So we have two lifetime flavors:
- **call-owned one-shot** (``to_actor.run()`` without ``portal=``): - **one-shot** (``to_actor.run()``): lives exactly as long as
spawned for one task, then cancelled and joined before ``run()`` its single task; reaped the moment its result (or error)
returns its result or raises its error. arrives back in the (blocking) call.
- **caller-owned daemon** (:meth:`~tractor.ActorNursery.start_actor`), - **daemon** (:meth:`~tractor.ActorNursery.start_actor`): lives
including an actor later reused through until *someone* cancels it — an explicit
``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. call-owned actors do not survive their own ``to_actor.run()`` 1. one-shot actors never make it to nursery exit: each is
calls; each is reaped before its call returns. reaped inside its own ``to_actor.run()`` call, any error
2. the nursery waits on caller-owned daemon actors raising immediately in the calling task so your code
**indefinitely**. If you spawned one, you own its lifetime. (acting as supervisor) gets first crack at handling it.
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,16 +43,15 @@ 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 this ``tractor.to_actor.run()`` runtime boots *implicitly* inside ``tractor.to_actor.run()``
call because neither ``an=`` nor ``portal=`` was supplied. No whenever it isn't already up. No special entrypoint, no
special entrypoint, no framework takeover - it's just a ``trio`` framework takeover - it's just a ``trio`` app,
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 linked a **new process** and executes ``cellar_door()`` as its *main
one-shot 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 call *blocks* until that final result arrives, then - the call *blocks* until that final result arrives, then
@ -68,10 +67,10 @@ What's going on here?
.. note:: .. note::
Without ``portal=``, ``to_actor.run()`` (parlance of ``to_actor.run()`` (parlance of ``trio.to_thread`` and
``trio.to_thread`` and friends) is the *convenience* wrapper: friends) is the *convenience* wrapper: one-shot
one-shot spawn-run-reap semantics for when a subactor's entire spawn-run-reap semantics for when a subactor's entire job is
job is a single function call. The core primitives are 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
@ -80,10 +79,10 @@ What's going on here?
Daemon actors and RPC Daemon actors and RPC
--------------------- ---------------------
A subactor spawned by ``to_actor.run()`` terminates after its lone A ``to_actor.run()`` one-shot subactor terminates when its lone
task 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,17 +90,14 @@ otherwise. That's ``start_actor()``:
Two lifetime rules to internalize: Two lifetime rules to internalize:
- a subactor spawned and owned by ``to_actor.run()`` is cancelled - a ``to_actor.run()`` one-shot actor lives exactly as long as
and reaped before the call returns its result or raises its error, its lone task; the call blocks until that function (and thus
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

@ -1,3 +0,0 @@
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,10 +15,7 @@ from unittest.mock import Mock
import pytest import pytest
import trio import trio
from trio.testing import ( from trio.testing import wait_all_tasks_blocked
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
@ -137,65 +134,6 @@ 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,21 +8,17 @@ 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
@ -258,186 +254,6 @@ 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,7 +310,6 @@ class Channel:
payload: Any, payload: Any,
hide_tb: bool = False, hide_tb: bool = False,
send_deadline: float = float('inf'),
) -> None: ) -> None:
''' '''
@ -321,9 +320,6 @@ 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:
@ -334,17 +330,10 @@ class Channel:
f'{pformat(payload)}\n' f'{pformat(payload)}\n'
) )
# assert self._transport # but why typing? # assert self._transport # but why typing?
if send_deadline == float('inf'):
await self._transport.send( await self._transport.send(
payload, payload,
hide_tb=hide_tb, 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,7 +439,6 @@ 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:
''' '''
@ -448,10 +447,6 @@ 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
@ -518,26 +513,12 @@ 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.
# #
# Ordinary sends may delay cancellation while a peer is # This can delay cancellation while a peer is not
# not reading. Actor-wide cancel requests pass their own # reading; peer/channel teardown must close the stream
# deadline so this operation can close a stalled stream. # to unblock a permanently stalled socket write.
with trio.CancelScope( with trio.CancelScope(shield=True):
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,53 +319,39 @@ 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:
with trio.move_on_at(cancel_deadline) as cs: # send cancel cmd - might not get response
# 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,
) )
cancel_ack_received = True return True
# Preserve shielded actor teardown, then immediately # `move_on_after` fired — peer didn't ack within
# 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 cs.cancelled_caught: if (
if raise_on_timeout: cs.cancelled_caught
and
raise_on_timeout
):
raise ActorTooSlowError( raise ActorTooSlowError(
f'Peer {peer_id} did not ack its ' f'Peer {peer_id} did not ack its '
f'`Actor.cancel()` RPC within bounded wait ' f'`Actor.cancel()` RPC within bounded wait '
f'of {cancel_timeout!r}s' f'of {cancel_timeout!r}s'
) )
# Legacy fire-and-forget callers decide whether to # legacy fire-and-forget path: log + return False so
# escalate the missed acknowledgement themselves. # the caller can decide whether to escalate.
#
# NOTE, we also land here in the (unexpected) case where
# the shielded `move_on_after` block exits WITHOUT
# `return True` and WITHOUT the deadline firing — prefer
# a soft `False` over an `assert`-crash mid-teardown.
log.debug( log.debug(
f'May have failed to cancel peer?\n' f'May have failed to cancel peer?\n'
f'\n' f'\n'
@ -373,8 +359,6 @@ class Portal:
) )
return False return False
return cancel_ack_received
except TransportClosed as tpt_err: except TransportClosed as tpt_err:
ipc_borked_report: str = ( ipc_borked_report: str = (
f'IPC for actor already closed/broken?\n\n' f'IPC for actor already closed/broken?\n\n'
@ -395,24 +379,16 @@ 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}'
@ -423,7 +399,6 @@ 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,11 +793,6 @@ 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
@ -850,13 +845,7 @@ class Actor:
) )
start_published: bool = False start_published: bool = False
try: try:
if send_deadline == float('inf'):
await chan.send(msg) 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,29 +327,6 @@ 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.
@ -442,12 +419,6 @@ 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,22 +138,12 @@ async def mp_proc(
# daemon=True, # daemon=True,
name=name, name=name,
) )
# `multiprocessing` only (since no async interface): publish the
# process and its reap coordination before start so cancellation # `multiprocessing` only (since no async interface):
# can own every subsequently started child. # register the process before start in case we get a cancel
( # request before the actor has fully spawned - then we can wait
reap_request, # for it to fully come up before sending a cancel 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():
@ -180,6 +170,9 @@ 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,6 +39,7 @@ 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
@ -130,23 +131,6 @@ 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
@ -177,6 +161,9 @@ 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,