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

View File

@ -62,10 +62,7 @@ one kwarg away,
.. code:: python
async with tractor.open_actor_cluster(
modules=[
'mylib.workers',
tractor.to_actor.MODULE,
],
modules=['mylib.workers'],
count=4,
names=['scout', 'miner', 'smelter', 'smith'],
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:
``portal.run()`` for bare one-shot RPCs (as in the demo),
``tractor.to_actor.run(..., portal=portal)`` for cancellation-linked
one-shot tasks in an existing worker (include
``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.to_actor.run(..., portal=portal)`` for linked one-shot calls,
or — for a persistent bidirectional dialog per worker — concurrently
enter N ``portal.open_context()`` blocks with
``tractor.trionics.gather_contexts()``; see :doc:`/guide/context`
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
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-shot task, spawning and reaping an actor only when no ``portal=``
is supplied; plain ``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 this page and the rest of the library reads
as convenience wrappers B)
one primitive: ``tractor.to_actor.run()`` is a convenience for
"spawn, run the lone task, await the result, tear down"; plain
``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
this page and the rest of the library reads as convenience
wrappers B)
The endpoint contract
---------------------

View File

@ -82,9 +82,10 @@ don't build your app on it.
One-shot subactors: ``to_actor.run()``
--------------------------------------
When the call should own a fresh subactor whose entire job is one
function call, :func:`tractor.to_actor.run` spawns it, runs the task,
returns its result and reaps the process — all in one blocking call:
When a subactor's *entire job* is a single function call, skip
the portal plumbing with :func:`tractor.to_actor.run`: spawn,
run the lone task, return its result and reap the process — all
in one blocking call:
.. code:: python
@ -100,37 +101,16 @@ Semantics worth knowing:
- it blocks until the remote task returns, re-raising any
remote error in the usual boxed form right in the calling
task.
- placement also determines process ownership: ``an=`` spawns and
reaps a fresh child in an existing actor nursery, while passing
neither does the same in a private call-scoped nursery (booting
the runtime if needed). ``portal=`` instead runs one linked task
in an existing actor; it neither spawns nor reaps that actor, so
the portal's owner remains responsible for its lifetime.
- "placement" is composable: ``an=`` spawns from an existing
actor-nursery, ``portal=`` reuses an already-running actor
(no spawn/reap, just a linked
:meth:`~tractor.Portal.open_context` call; see the
:doc:`context guide </guide/context>`), and passing neither
opens a private call-scoped nursery (booting the runtime if needed).
- concurrency composes the plain ``trio`` way: schedule
multiple ``run()`` calls into a local task nursery (see
``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()``
----------------------------------
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
----------------------------------------------
Without ``portal=``, :func:`tractor.to_actor.run` is the convenience
wrapper: spawn an actor, run exactly one async function in it, block
on the result, then reap the process — the distributed sibling of
:func:`tractor.to_actor.run` is the convenience wrapper: spawn
an actor, run exactly one async function in it, block on the
result, then reap the process — the distributed sibling of
``trio.to_thread.run_sync()``.
.. code:: python
@ -126,10 +126,6 @@ A few details worth knowing:
``name='something_cuter'``.
- the function's module is auto-added to the child's
``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()``
to bind target keyword arguments. Keywords passed directly to
``run()`` configure actor placement and spawning.
@ -137,23 +133,18 @@ A few details worth knowing:
child is *auto-cancelled* (reaped) right after — so remote
errors raise directly in your calling task (causality_ is
paramount!).
- "placement" composes: ``an=`` spawns a call-owned child from an
existing actor nursery, while passing neither opens a private
call-scoped nursery. ``portal=`` instead reuses an existing actor:
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``.
- "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::
:func:`tractor.to_actor.run` is a convenience, **not** the core
model. For actor-owning placements it combines
:meth:`~tractor.ActorNursery.start_actor`, a linked
:meth:`~tractor.Portal.open_context` call, and per-child
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
model — it's built *entirely* on
:meth:`~tractor.ActorNursery.start_actor` plus a linked
:meth:`~tractor.Portal.open_context` call and per-child
cancellation/reaping. Teach your fingers to use it for quick
fire-and-collect parallelism — think a per-function trio-parallel_
style one-shot — and reach for
:meth:`~tractor.ActorNursery.start_actor` plus
@ -162,25 +153,25 @@ A few details worth knowing:
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=``):
spawned for one task, then cancelled and joined before ``run()``
returns its result or raises its error.
- **caller-owned daemon** (:meth:`~tractor.ActorNursery.start_actor`),
including an actor later reused through
``to_actor.run(..., portal=portal)``: lives until *someone*
cancels it via an explicit
- **one-shot** (``to_actor.run()``): lives exactly as long as
its single task; reaped the moment its result (or error)
arrives back in the (blocking) call.
- **daemon** (:meth:`~tractor.ActorNursery.start_actor`): lives
until *someone* cancels it — an explicit
:meth:`~tractor.Portal.cancel_actor`, a bulk
:meth:`~tractor.ActorNursery.cancel`, or the one-cancels-all
strategy kicking in on error.
On a clean exit of the nursery block the teardown order is:
1. call-owned actors do not survive their own ``to_actor.run()``
calls; each is reaped before its call returns.
2. the nursery waits on caller-owned daemon actors
**indefinitely**. If you spawned one, you own its lifetime.
1. one-shot actors never make it to nursery exit: each is
reaped inside its own ``to_actor.run()`` call, any error
raising immediately in the calling task so your code
(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
discipline: the runtime sends an IPC cancel request and gives

View File

@ -43,16 +43,15 @@ Run it::
What's going on here?
- ``trio.run(main)`` starts the **root actor**; the ``tractor``
runtime boots *implicitly* inside this ``tractor.to_actor.run()``
call because neither ``an=`` nor ``portal=`` was supplied. No
special entrypoint, no framework takeover - it's just a ``trio``
app,
runtime boots *implicitly* inside ``tractor.to_actor.run()``
whenever it isn't already up. No special entrypoint, no
framework takeover - it's just a ``trio`` app,
- inside ``main()`` a *subactor* is spawned via
``tractor.to_actor.run()`` and told to run exactly one
function: ``cellar_door()``,
- the subactor, *some_linguist*, boots a fresh ``trio.run()`` in
a **new process** and executes ``cellar_door()`` as its linked
one-shot task (note the child proving it is *not* the root with
a **new process** and executes ``cellar_door()`` as its *main
task* (note the child proving it is *not* the root with
``tractor.is_root_process()``), then ships the return value
back over IPC,
- the call *blocks* until that final result arrives, then
@ -68,10 +67,10 @@ What's going on here?
.. note::
Without ``portal=``, ``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 a single function call. The core primitives are
``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
a single function call. The core primitives are
:meth:`~tractor.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
@ -80,10 +79,10 @@ What's going on here?
Daemon actors and RPC
---------------------
A subactor spawned by ``to_actor.run()`` terminates after its lone
task returns. But often you want long-lived *daemon* actors instead:
spawned once, then serving (allowlisted) RPC requests until told
otherwise. That's ``start_actor()``:
A ``to_actor.run()`` one-shot subactor terminates when its lone
task returns. But often you want long-lived *daemon* actors
instead: spawned once, then serving (allowlisted) RPC requests
until told otherwise. That's ``start_actor()``:
.. literalinclude:: ../../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:
- a subactor spawned and owned by ``to_actor.run()`` is cancelled
and reaped before the call returns its result or raises its error,
- a ``to_actor.run()`` one-shot actor lives exactly as long as
its lone task; the call blocks until that function (and thus
the process) completes,
- a ``start_actor()`` actor *lives forever* - an RPC daemon the
nursery will happily wait on **indefinitely** - until some
task explicitly cancels it via ``Portal.cancel_actor()`` (as
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::
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 trio
from trio.testing import (
MockClock,
wait_all_tasks_blocked,
)
from trio.testing import wait_all_tasks_blocked
import tractor
from tractor import Actor
from tractor.discovery import _addr
@ -137,65 +134,6 @@ def test_cancelled_transport_send_completes_frame():
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():
'''
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 pathlib import Path
from types import SimpleNamespace
import pytest
import trio
from trio.testing import MockClock
import tractor
from tractor import (
RemoteActorError,
to_actor,
)
from tractor._testing import tractor_test
from tractor._exceptions import ActorTooSlowError
from tractor.msg import ptr as msgptr
from tractor.msg.ptr import NamespacePath
from tractor.spawn import _mp as mp_spawn
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
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():
'''
Preserve a nursery-wide reap request across child startup.

View File

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

View File

@ -439,7 +439,6 @@ class MsgpackTransport(MsgTransport):
strict_types: bool = True,
hide_tb: bool = True,
send_deadline: float = float('inf'),
) -> None:
'''
@ -448,10 +447,6 @@ class MsgpackTransport(MsgTransport):
If `strict_types == True` then a `MsgTypeError` will be raised on any
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
@ -518,26 +513,12 @@ class MsgpackTransport(MsgTransport):
# the frame is complete, the explicit checkpoint
# immediately delivers any pending cancellation.
#
# Ordinary sends may delay cancellation while a peer is
# not reading. Actor-wide cancel requests pass their own
# deadline so this operation can close a stalled stream.
with trio.CancelScope(
deadline=send_deadline,
shield=True,
) as send_cs:
# This can delay cancellation while a peer is not
# reading; peer/channel teardown must close the stream
# to unblock a permanently stalled socket write.
with trio.CancelScope(shield=True):
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()
return None

View File

@ -319,61 +319,45 @@ class Portal:
or
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:
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
await self._run_from_ns(
await self.run_from_ns(
'self',
'cancel',
kwargs={},
cancel_on_startup=False,
send_deadline=cancel_deadline,
)
cancel_ack_received = True
return True
# 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
# `move_on_after` fired — peer didn't ack within
# bounded window. Behaviour depends on
# `raise_on_timeout`:
if cs.cancelled_caught:
if raise_on_timeout:
raise ActorTooSlowError(
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 callers decide whether to
# escalate the missed acknowledgement themselves.
log.debug(
f'May have failed to cancel peer?\n'
f'\n'
f'c)=?> {peer_id}\n'
if (
cs.cancelled_caught
and
raise_on_timeout
):
raise ActorTooSlowError(
f'Peer {peer_id} did not ack its '
f'`Actor.cancel()` RPC within bounded wait '
f'of {cancel_timeout!r}s'
)
return False
return cancel_ack_received
# legacy fire-and-forget path: log + return False so
# 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(
f'May have failed to cancel peer?\n'
f'\n'
f'c)=?> {peer_id}\n'
)
return False
except TransportClosed as tpt_err:
ipc_borked_report: str = (
@ -395,24 +379,16 @@ class Portal:
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(
self,
namespace_path: str,
function_name: str,
kwargs: dict[str, Any],
cancel_on_startup: bool = True,
send_deadline: float = float('inf'),
) -> Any:
'''
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(
f'{namespace_path}:{function_name}'
@ -423,7 +399,6 @@ class Portal:
kwargs=kwargs,
portal=self,
cancel_on_startup=cancel_on_startup,
send_deadline=send_deadline,
)
try:
return await ctx._pld_rx.recv_pld(

View File

@ -793,11 +793,6 @@ class Actor:
ack_timeout: float = float('inf'),
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:
'''
Send a `'cmd'` msg to a remote actor, which requests the
@ -850,13 +845,7 @@ class Actor:
)
start_published: bool = False
try:
if send_deadline == float('inf'):
await chan.send(msg)
else:
await chan.send(
msg,
send_deadline=send_deadline,
)
await chan.send(msg)
start_published = True
# NOTE wait on first `StartAck` response msg and validate;

View File

@ -327,29 +327,6 @@ class ActorNursery:
reap_request.set()
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:
'''
Release every child monitor into its process-join phase.
@ -442,12 +419,6 @@ class ActorNursery:
'''
__runtimeframe__: int = 1 # noqa
if self._cancel_called:
raise RuntimeError(
'Cannot start an actor in a cancelling '
'`ActorNursery`'
)
loglevel: str = (
loglevel
or self._actor.loglevel

View File

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

View File

@ -39,6 +39,7 @@ from tractor.runtime._state import (
current_actor,
is_root_process,
debug_mode,
get_runtime_vars,
)
from tractor.log import get_logger
from tractor.discovery._addr import UnwrappedAddress
@ -130,23 +131,6 @@ async def trio_proc(
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
# channel should have handshake completed by the
# local actor by the time we get a ref to it
@ -177,6 +161,9 @@ async def trio_proc(
assert proc
portal = Portal(chan)
reap_request, _ = actor_nursery._register_child_reap(
subactor.aid.uid,
)
actor_nursery._children[subactor.aid.uid] = (
subactor,
proc,