Compare commits
No commits in common. "70f497bf5f0e211774183f08ea90757ae3295d4f" and "1b8ae7933ad0a0f011007861db0c0af49daab10e" have entirely different histories.
70f497bf5f
...
1b8ae7933a
|
|
@ -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.
|
|
||||||
|
|
@ -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.
|
|
||||||
|
|
@ -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.
|
|
||||||
|
|
@ -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.
|
|
||||||
|
|
@ -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
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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.
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
---------------------
|
---------------------
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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.
|
|
||||||
|
|
@ -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.
|
||||||
|
|
|
||||||
|
|
@ -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.
|
||||||
|
|
|
||||||
|
|
@ -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,
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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(
|
||||||
|
|
|
||||||
|
|
@ -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;
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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,
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue