Compare commits
3 Commits
65bf9df5ba
...
a34aaf98d2
| Author | SHA1 | Date |
|---|---|---|
|
|
a34aaf98d2 | |
|
|
6b16d0d282 | |
|
|
63df5534bb |
|
|
@ -0,0 +1,79 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5
|
||||||
|
service: claude
|
||||||
|
session: f6c84722-471a-4458-9a80-e453fea9029f
|
||||||
|
timestamp: 2026-07-02T15:42:55Z
|
||||||
|
git_ref: 65bf9df5
|
||||||
|
scope: code
|
||||||
|
substantive: true
|
||||||
|
raw_file: 20260702T154255Z_65bf9df5_prompt_io.raw.md
|
||||||
|
---
|
||||||
|
|
||||||
|
## Prompt
|
||||||
|
|
||||||
|
Driver prompt file `ai/prompt-io/prompts/issue_477.md`:
|
||||||
|
|
||||||
|
> attempt to resolve
|
||||||
|
> https://github.com/goodboy/tractor/issues/477
|
||||||
|
> do it with /open-wkt.
|
||||||
|
|
||||||
|
(plus a hard stop-for-human-review deadline of 12:50PM
|
||||||
|
EST the same day)
|
||||||
|
|
||||||
|
Issue #477 asks to factor `ActorNursery.run_in_actor()`
|
||||||
|
(and possibly `Portal.run()`) out of the nursery
|
||||||
|
internals into a new `tractor.to_actor` wrapper
|
||||||
|
subpackage of "higher level one shot" single-remote-task
|
||||||
|
APIs, adopting the `trio.to_thread`/`anyio.to_process`
|
||||||
|
parlance, so that error collection/propagation moves up
|
||||||
|
into the caller's local `trio` scope and the nursery's
|
||||||
|
spawn machinery can eventually drop the
|
||||||
|
`._ria_nursery` coupling.
|
||||||
|
|
||||||
|
## Response summary
|
||||||
|
|
||||||
|
First-cut `tractor.to_actor` subpkg delivering the
|
||||||
|
one-shot API composed purely from the existing
|
||||||
|
daemon-spawn + portal primitives (`start_actor()` +
|
||||||
|
`Portal.run()` + `Portal.cancel_actor()`), leaving the
|
||||||
|
legacy `.run_in_actor()` machinery untouched (formal
|
||||||
|
deprecation deferred until in-repo usage migrates):
|
||||||
|
|
||||||
|
- `to_actor.run(fn, **fn_kwargs) -> Any`: spawn a
|
||||||
|
subactor, schedule `fn` as its lone remote task, wait
|
||||||
|
on and return its result, ALWAYS reaping the subactor
|
||||||
|
(shield-safe `finally`). Remote errors raise in the
|
||||||
|
caller's task as boxed `RemoteActorError`s.
|
||||||
|
- placement variants: `portal=` reuses a running actor
|
||||||
|
(no spawn/reap), `an=` spawns from a caller-managed
|
||||||
|
actor-nursery, neither opens a call-scoped private
|
||||||
|
`open_nursery()` (implicitly booting the runtime,
|
||||||
|
configurable via `runtime_kwargs`).
|
||||||
|
- fail-fast validation before any spawn: non-streaming
|
||||||
|
async fn required; `portal=`/`an=` mutually
|
||||||
|
exclusive; `runtime_kwargs` rejected alongside any
|
||||||
|
placement opt.
|
||||||
|
- `run_in_actor()` TODO/docstring now cross-reference
|
||||||
|
the successor API.
|
||||||
|
|
||||||
|
## Files changed
|
||||||
|
|
||||||
|
- `tractor/to_actor/__init__.py` — new subpkg,
|
||||||
|
re-exports `run`
|
||||||
|
- `tractor/to_actor/_api.py` — `run()` +
|
||||||
|
`_invoke_in_subactor()` + `_validate_one_shot_fn()`
|
||||||
|
- `tractor/__init__.py` — top-level `to_actor`
|
||||||
|
re-export
|
||||||
|
- `tractor/runtime/_supervise.py` — comment/docstring
|
||||||
|
pointers from `run_in_actor()` to the successor
|
||||||
|
- `tests/test_to_actor.py` — 11-test suite covering
|
||||||
|
all placement variants, error relay, the concurrent
|
||||||
|
worker-pool-ish pattern and arg validation
|
||||||
|
- `examples/parallelism/to_actor_one_shots.py` —
|
||||||
|
runnable demo (auto-collected by
|
||||||
|
`test_docs_examples.py`)
|
||||||
|
|
||||||
|
## Human edits
|
||||||
|
|
||||||
|
None yet — pending human review (work paused before the
|
||||||
|
12:50PM EST deadline per the driver prompt).
|
||||||
|
|
@ -0,0 +1,100 @@
|
||||||
|
---
|
||||||
|
model: claude-fable-5
|
||||||
|
service: claude
|
||||||
|
timestamp: 2026-07-02T15:42:55Z
|
||||||
|
git_ref: 65bf9df5
|
||||||
|
diff_cmd: git diff main..wkt/to_actor_subpkg
|
||||||
|
---
|
||||||
|
|
||||||
|
# Raw AI output (diff-ref mode)
|
||||||
|
|
||||||
|
All generated code is committed on the
|
||||||
|
`wkt/to_actor_subpkg` branch; per diff-ref mode each
|
||||||
|
file's verbatim content is reachable via the pointers
|
||||||
|
below rather than duplicated here.
|
||||||
|
|
||||||
|
## Generated files
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- tractor/to_actor/__init__.py`
|
||||||
|
|
||||||
|
New subpackage init: module docstring establishing the
|
||||||
|
`trio.to_thread`/`anyio.to_process` "run it over there"
|
||||||
|
parlance for actors, plus the single public re-export
|
||||||
|
`run as run` from `._api`.
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- tractor/to_actor/_api.py`
|
||||||
|
|
||||||
|
The one-shot invocation impl, composed entirely from the
|
||||||
|
lower level daemon-spawn + portal primitives as
|
||||||
|
prescribed by issue #477:
|
||||||
|
|
||||||
|
- `_validate_one_shot_fn()`: the `Portal.run()`
|
||||||
|
non-streaming-async-fn constraint checked up-front,
|
||||||
|
before any subactor is spawned.
|
||||||
|
- `_invoke_in_subactor()`: `an.start_actor()` ->
|
||||||
|
`Portal.run()` -> always-reap via
|
||||||
|
`Portal.cancel_actor()` in a `finally` (the cancel
|
||||||
|
req's bounded wait is internally shielded so the reap
|
||||||
|
also runs under caller-scope cancellation).
|
||||||
|
- `run()`: the public API. Placement options:
|
||||||
|
`portal=` (reuse a running actor, no spawn/reap),
|
||||||
|
`an=` (spawn from a caller-managed nursery), or
|
||||||
|
neither (private `open_nursery()` scoped to the call,
|
||||||
|
implicitly booting the runtime when needed, tunable
|
||||||
|
via pass-through `runtime_kwargs`). Spawn opts mirror
|
||||||
|
`ActorNursery.start_actor()`; `**fn_kwargs` are
|
||||||
|
relayed to the remote task. Errors raise in the
|
||||||
|
caller's task as boxed `RemoteActorError`s.
|
||||||
|
`runtime_kwargs` alongside any placement opt is a
|
||||||
|
hard `ValueError`, never silently ignored.
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- tractor/__init__.py`
|
||||||
|
|
||||||
|
Top-level `from . import to_actor as to_actor`
|
||||||
|
re-export.
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- tractor/runtime/_supervise.py`
|
||||||
|
|
||||||
|
Comment/docstring-only: the `run_in_actor()` deprecation
|
||||||
|
TODO now points at the implemented `.to_actor.run()`
|
||||||
|
successor (checkbox ticked) and the method docstring
|
||||||
|
gains a NOTE steering users to the new API; remaining
|
||||||
|
TODO items are the `DeprecationWarning` emission +
|
||||||
|
in-repo usage migration.
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- tests/test_to_actor.py`
|
||||||
|
|
||||||
|
11-test suite: private-nursery one-shot, implicit
|
||||||
|
runtime boot via `runtime_kwargs`, remote-error relay to
|
||||||
|
the caller's task (bare + caller-managed nursery),
|
||||||
|
caller-nursery spawn, portal reuse w/o implicit reap,
|
||||||
|
the concurrent worker-pool-ish pattern (local `trio`
|
||||||
|
nursery x shared `an`), and the four validation
|
||||||
|
rejections (sync fn, async-gen fn, `portal`+`an`
|
||||||
|
combo, `runtime_kwargs`+placement combo).
|
||||||
|
|
||||||
|
> `git diff main..wkt/to_actor_subpkg -- examples/parallelism/to_actor_one_shots.py`
|
||||||
|
|
||||||
|
Runnable example (auto-collected by
|
||||||
|
`test_docs_examples.py`): the fully-implicit one-shot
|
||||||
|
plus the concurrent worker-pool-ish prime-check pattern
|
||||||
|
against a shared caller-managed actor-nursery.
|
||||||
|
|
||||||
|
## Test runs (verbatim)
|
||||||
|
|
||||||
|
```
|
||||||
|
tests/test_to_actor.py .......... [100%]
|
||||||
|
============= 10 passed in 4.29s =============
|
||||||
|
```
|
||||||
|
|
||||||
|
Regression subset for touched modules
|
||||||
|
(`test_local.py test_rpc.py test_spawning.py
|
||||||
|
test_cancellation.py`):
|
||||||
|
|
||||||
|
```
|
||||||
|
38 passed, 1 xfailed, 24 warnings in 80.71s (0:01:20)
|
||||||
|
```
|
||||||
|
|
||||||
|
(warnings are pre-existing stdlib `os.fork()`
|
||||||
|
DeprecationWarnings from the mp spawn backends, not
|
||||||
|
introduced by this change)
|
||||||
|
|
@ -0,0 +1,7 @@
|
||||||
|
NOTE: you MUST pause this work at 12:50PM EST (BEFORE your weekly
|
||||||
|
limit reset) for review by a human!
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
attempt to resolve https://github.com/goodboy/tractor/issues/477
|
||||||
|
do it with /open-wkt.
|
||||||
|
|
@ -0,0 +1,83 @@
|
||||||
|
'''
|
||||||
|
`tractor.to_actor.run()`: one-shot single-task subactor
|
||||||
|
invocation, the SC-parallelism sibling of
|
||||||
|
`trio.to_thread.run_sync()` (and `anyio.to_process`).
|
||||||
|
|
||||||
|
Each call spawns a subactor, schedules the async fn as
|
||||||
|
its lone remote task, waits on the result and reaps the
|
||||||
|
subactor. Concurrency composes the plain `trio` way:
|
||||||
|
schedule multiple one-shot calls in a local task nursery
|
||||||
|
against a shared actor-nursery; any remote error raises
|
||||||
|
directly in the task which scheduled it.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import math
|
||||||
|
|
||||||
|
import tractor
|
||||||
|
import trio
|
||||||
|
|
||||||
|
|
||||||
|
async def is_prime(
|
||||||
|
n: int,
|
||||||
|
) -> bool:
|
||||||
|
if n < 2:
|
||||||
|
return False
|
||||||
|
if n == 2:
|
||||||
|
return True
|
||||||
|
if n % 2 == 0:
|
||||||
|
return False
|
||||||
|
|
||||||
|
sqrt_n = int(math.floor(math.sqrt(n)))
|
||||||
|
for i in range(3, sqrt_n + 1, 2):
|
||||||
|
if n % i == 0:
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
|
||||||
|
# fully implicit one-shot: boots the actor-runtime,
|
||||||
|
# spawns a subactor, runs the task, reaps the
|
||||||
|
# subactor, tears the runtime back down.
|
||||||
|
assert await tractor.to_actor.run(
|
||||||
|
is_prime,
|
||||||
|
n=2,
|
||||||
|
)
|
||||||
|
|
||||||
|
# the "worker-pool-ish" pattern from the original
|
||||||
|
# `concurrent.futures` example: one subactor per
|
||||||
|
# input, all concurrent, results and errors
|
||||||
|
# collected by caller-side tasks.
|
||||||
|
results: dict[int, bool] = {}
|
||||||
|
|
||||||
|
async def check(
|
||||||
|
an: tractor.ActorNursery,
|
||||||
|
n: int,
|
||||||
|
i: int,
|
||||||
|
) -> None:
|
||||||
|
results[n] = await tractor.to_actor.run(
|
||||||
|
is_prime,
|
||||||
|
an=an,
|
||||||
|
name=f'prime_checker_{i}',
|
||||||
|
n=n,
|
||||||
|
)
|
||||||
|
|
||||||
|
inputs: list[int] = [
|
||||||
|
7,
|
||||||
|
8,
|
||||||
|
3691,
|
||||||
|
3693,
|
||||||
|
]
|
||||||
|
async with (
|
||||||
|
tractor.open_nursery() as an,
|
||||||
|
trio.open_nursery() as tn,
|
||||||
|
):
|
||||||
|
for i, n in enumerate(inputs):
|
||||||
|
tn.start_soon(check, an, n, i)
|
||||||
|
|
||||||
|
for n, prime in sorted(results.items()):
|
||||||
|
print(f'{n} is prime: {prime}')
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
|
|
@ -0,0 +1,271 @@
|
||||||
|
'''
|
||||||
|
`tractor.to_actor`: one-shot single-remote-task API suite.
|
||||||
|
|
||||||
|
Verifies the "spiritual successor" to (and eventual
|
||||||
|
replacement of) `ActorNursery.run_in_actor()`; see
|
||||||
|
https://github.com/goodboy/tractor/issues/477
|
||||||
|
|
||||||
|
'''
|
||||||
|
from functools import partial
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
from tractor import (
|
||||||
|
RemoteActorError,
|
||||||
|
to_actor,
|
||||||
|
)
|
||||||
|
from tractor._testing import tractor_test
|
||||||
|
|
||||||
|
|
||||||
|
async def add_one(
|
||||||
|
n: int,
|
||||||
|
) -> int:
|
||||||
|
return n + 1
|
||||||
|
|
||||||
|
|
||||||
|
async def raise_value_error() -> None:
|
||||||
|
raise ValueError('kaboom')
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_one_shot_in_private_nursery(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
No `an`/`portal` provided: a private actor-nursery
|
||||||
|
is opened (and torn down) scoped to just the call.
|
||||||
|
|
||||||
|
'''
|
||||||
|
assert await to_actor.run(
|
||||||
|
add_one,
|
||||||
|
n=1,
|
||||||
|
) == 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_one_shot_boots_implicit_runtime(
|
||||||
|
reg_addr: tuple,
|
||||||
|
start_method: str,
|
||||||
|
loglevel: str,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Outside any actor-runtime `to_actor.run()` boots one
|
||||||
|
implicitly (just like bare `open_nursery()` usage)
|
||||||
|
configured via pass-through `runtime_kwargs`.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async def main() -> None:
|
||||||
|
assert tractor.current_actor(
|
||||||
|
err_on_no_runtime=False,
|
||||||
|
) is None
|
||||||
|
result = await to_actor.run(
|
||||||
|
add_one,
|
||||||
|
n=41,
|
||||||
|
runtime_kwargs=dict(
|
||||||
|
registry_addrs=[reg_addr],
|
||||||
|
start_method=start_method,
|
||||||
|
loglevel=loglevel,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
assert result == 42
|
||||||
|
|
||||||
|
trio.run(main)
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_remote_error_relayed_to_caller_task(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
A remote task error is raised directly in the
|
||||||
|
caller's task as a boxed `RemoteActorError` instead
|
||||||
|
of surfacing at actor-nursery teardown as with the
|
||||||
|
legacy `.run_in_actor()` API.
|
||||||
|
|
||||||
|
'''
|
||||||
|
with pytest.raises(RemoteActorError) as excinfo:
|
||||||
|
await to_actor.run(raise_value_error)
|
||||||
|
|
||||||
|
assert excinfo.value.boxed_type is ValueError
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_spawn_from_caller_nursery(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Pass a caller-managed `an: ActorNursery` for the
|
||||||
|
spawn; the subactor is still one-shot reaped by the
|
||||||
|
time the call returns.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
assert await to_actor.run(
|
||||||
|
add_one,
|
||||||
|
an=an,
|
||||||
|
n=10,
|
||||||
|
) == 11
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_remote_error_from_caller_nursery(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
With a caller-managed `an` the remote error also
|
||||||
|
surfaces in the caller's task, INSIDE the nursery
|
||||||
|
block, allowing inline (supervision-style) handling.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
with pytest.raises(RemoteActorError) as excinfo:
|
||||||
|
await to_actor.run(
|
||||||
|
raise_value_error,
|
||||||
|
an=an,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert excinfo.value.boxed_type is ValueError
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_reuse_existing_actor_via_portal(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
Pass `portal=` to schedule the one-shot task in an
|
||||||
|
already-running actor; no spawn, no implicit reap.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portal: tractor.Portal = await an.start_actor(
|
||||||
|
'one_shot_worker',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
for i in range(3):
|
||||||
|
assert await to_actor.run(
|
||||||
|
add_one,
|
||||||
|
portal=portal,
|
||||||
|
n=i,
|
||||||
|
) == i + 1
|
||||||
|
|
||||||
|
# still alive: caller owns the actor's lifetime.
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
@tractor_test
|
||||||
|
async def test_concurrent_one_shots_from_task_nursery(
|
||||||
|
start_method: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
):
|
||||||
|
'''
|
||||||
|
The worker-pool-ish pattern from #477: concurrency
|
||||||
|
is composed with a plain (caller-side) `trio` task
|
||||||
|
nursery scheduling multiple one-shot calls against
|
||||||
|
a shared caller-managed actor-nursery; error
|
||||||
|
collection thus lives entirely in caller-code.
|
||||||
|
|
||||||
|
'''
|
||||||
|
results: dict[int, int] = {}
|
||||||
|
|
||||||
|
async def one_shot(
|
||||||
|
an: tractor.ActorNursery,
|
||||||
|
i: int,
|
||||||
|
) -> None:
|
||||||
|
results[i] = await to_actor.run(
|
||||||
|
add_one,
|
||||||
|
an=an,
|
||||||
|
name=f'one_shot_{i}',
|
||||||
|
n=i,
|
||||||
|
)
|
||||||
|
|
||||||
|
async with (
|
||||||
|
tractor.open_nursery() as an,
|
||||||
|
trio.open_nursery() as tn,
|
||||||
|
):
|
||||||
|
for i in range(4):
|
||||||
|
tn.start_soon(one_shot, an, i)
|
||||||
|
|
||||||
|
assert results == {
|
||||||
|
i: i + 1 for i in range(4)
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_rejects_sync_fn():
|
||||||
|
'''
|
||||||
|
Non-async callables error BEFORE any spawn (or even
|
||||||
|
runtime-boot) happens.
|
||||||
|
|
||||||
|
'''
|
||||||
|
def not_async() -> None:
|
||||||
|
...
|
||||||
|
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
trio.run(
|
||||||
|
partial(
|
||||||
|
to_actor.run,
|
||||||
|
not_async,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rejects_streaming_fn():
|
||||||
|
'''
|
||||||
|
Async-gen (streaming) fns are not one-shot-able,
|
||||||
|
same constraint as `Portal.run()`.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async def agen():
|
||||||
|
yield 1
|
||||||
|
|
||||||
|
with pytest.raises(TypeError):
|
||||||
|
trio.run(
|
||||||
|
partial(
|
||||||
|
to_actor.run,
|
||||||
|
agen,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rejects_portal_and_an_combo():
|
||||||
|
'''
|
||||||
|
`portal=` and `an=` are mutually exclusive
|
||||||
|
placement options.
|
||||||
|
|
||||||
|
'''
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
trio.run(
|
||||||
|
partial(
|
||||||
|
to_actor.run,
|
||||||
|
add_one,
|
||||||
|
portal=object(),
|
||||||
|
an=object(),
|
||||||
|
n=1,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rejects_runtime_kwargs_with_placement():
|
||||||
|
'''
|
||||||
|
`runtime_kwargs` only applies when the call opens
|
||||||
|
its own private actor-nursery; passing it alongside
|
||||||
|
a placement opt is an error, never silently
|
||||||
|
ignored.
|
||||||
|
|
||||||
|
'''
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
trio.run(
|
||||||
|
partial(
|
||||||
|
to_actor.run,
|
||||||
|
add_one,
|
||||||
|
an=object(),
|
||||||
|
runtime_kwargs=dict(
|
||||||
|
loglevel='cancel',
|
||||||
|
),
|
||||||
|
n=1,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
@ -62,6 +62,7 @@ from .devx import (
|
||||||
post_mortem as post_mortem,
|
post_mortem as post_mortem,
|
||||||
)
|
)
|
||||||
from . import msg as msg
|
from . import msg as msg
|
||||||
|
from . import to_actor as to_actor
|
||||||
from ._root import (
|
from ._root import (
|
||||||
run_daemon as run_daemon,
|
run_daemon as run_daemon,
|
||||||
open_root_actor as open_root_actor,
|
open_root_actor as open_root_actor,
|
||||||
|
|
|
||||||
|
|
@ -383,12 +383,13 @@ class ActorNursery:
|
||||||
)
|
)
|
||||||
|
|
||||||
# TODO: DEPRECATE THIS:
|
# TODO: DEPRECATE THIS:
|
||||||
# -[ ] impl instead as a hilevel wrapper on
|
# -[x] impl instead as a hilevel wrapper on top of
|
||||||
# top of a `@context` style invocation.
|
# the lower level daemon-spawn + portal APIs
|
||||||
# |_ dynamic @context decoration on child side
|
# |_ see `.to_actor.run()` (issue #477) which does
|
||||||
# |_ implicit `Portal.open_context() as (ctx, first):`
|
# `.start_actor()` + `Portal.run()` + a one-shot
|
||||||
# and `return first` on parent side.
|
# reap via `Portal.cancel_actor()`.
|
||||||
# |_ mention how it's similar to `trio-parallel` API?
|
# -[ ] emit a `DeprecationWarning` here (requires
|
||||||
|
# migrating all in-repo usage first!)
|
||||||
# -[ ] use @api_frame on the wrapper
|
# -[ ] use @api_frame on the wrapper
|
||||||
async def run_in_actor(
|
async def run_in_actor(
|
||||||
self,
|
self,
|
||||||
|
|
@ -416,6 +417,11 @@ class ActorNursery:
|
||||||
until the task spawned by executing ``fn`` completes at which point
|
until the task spawned by executing ``fn`` completes at which point
|
||||||
the actor is terminated.
|
the actor is terminated.
|
||||||
|
|
||||||
|
NOTE: prefer the (eventual) replacement API
|
||||||
|
`tractor.to_actor.run()` which delivers the same
|
||||||
|
one-shot semantics decoupled from this nursery's
|
||||||
|
internal spawn machinery; see issue #477.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
__runtimeframe__: int = 1 # noqa
|
__runtimeframe__: int = 1 # noqa
|
||||||
mod_path: str = fn.__module__
|
mod_path: str = fn.__module__
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,33 @@
|
||||||
|
# tractor: distributed structured concurrency.
|
||||||
|
# Copyright 2018-eternity Tyler Goodlet.
|
||||||
|
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU Affero General Public License as published by
|
||||||
|
# the Free Software Foundation, either version 3 of the License, or
|
||||||
|
# (at your option) any later version.
|
||||||
|
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU Affero General Public License for more details.
|
||||||
|
|
||||||
|
# You should have received a copy of the GNU Affero General Public License
|
||||||
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
'''
|
||||||
|
`tractor.to_actor`: high-level "one-shot" remote-task APIs.
|
||||||
|
|
||||||
|
Adopts the "run it over there" parlance from analogous
|
||||||
|
(sibling-library) APIs like `trio.to_thread` and
|
||||||
|
`anyio.to_process` but for SC-supervised actors: spawn (or
|
||||||
|
reuse) a subactor, schedule a single remote task, wait on
|
||||||
|
its result and (when the call owns the subactor) reap it.
|
||||||
|
|
||||||
|
The "spiritual successor" to (and eventual replacement of)
|
||||||
|
the `ActorNursery.run_in_actor()` API; see
|
||||||
|
https://github.com/goodboy/tractor/issues/477
|
||||||
|
|
||||||
|
'''
|
||||||
|
from ._api import (
|
||||||
|
run as run,
|
||||||
|
)
|
||||||
|
|
@ -0,0 +1,226 @@
|
||||||
|
# tractor: distributed structured concurrency.
|
||||||
|
# Copyright 2018-eternity Tyler Goodlet.
|
||||||
|
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU Affero General Public License as published by
|
||||||
|
# the Free Software Foundation, either version 3 of the License, or
|
||||||
|
# (at your option) any later version.
|
||||||
|
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU Affero General Public License for more details.
|
||||||
|
|
||||||
|
# You should have received a copy of the GNU Affero General Public License
|
||||||
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
'''
|
||||||
|
One-shot remote-task invocation built on spawn-and-portal
|
||||||
|
primitives.
|
||||||
|
|
||||||
|
Implemented (as prescribed by #477) entirely "on top of"
|
||||||
|
the lower level daemon-actor spawn + portal APIs,
|
||||||
|
|
||||||
|
- `ActorNursery.start_actor()` for (daemon-style) subactor
|
||||||
|
spawning,
|
||||||
|
- `Portal.run()` for scheduling the lone remote task and
|
||||||
|
waiting on its result,
|
||||||
|
- `Portal.cancel_actor()` for reaping the subactor once
|
||||||
|
that result (or error) arrives,
|
||||||
|
|
||||||
|
such that error collection and propagation happens in the
|
||||||
|
*caller's task* (and thus whatever `trio` nursery/scope
|
||||||
|
encloses it) instead of inside the actor-nursery's
|
||||||
|
spawn-machinery nurseries as with the (to be deprecated)
|
||||||
|
`ActorNursery.run_in_actor()` API.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from __future__ import annotations
|
||||||
|
import inspect
|
||||||
|
from typing import (
|
||||||
|
Any,
|
||||||
|
Callable,
|
||||||
|
TYPE_CHECKING,
|
||||||
|
)
|
||||||
|
|
||||||
|
from ..runtime._supervise import (
|
||||||
|
ActorNursery,
|
||||||
|
open_nursery,
|
||||||
|
)
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from ..discovery._addr import UnwrappedAddress
|
||||||
|
from ..runtime._portal import Portal
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_one_shot_fn(
|
||||||
|
fn: Callable,
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Ensure `fn` is a non-streaming async function, raise
|
||||||
|
a `TypeError` otherwise.
|
||||||
|
|
||||||
|
The same constraint enforced by `Portal.run()` but
|
||||||
|
checked up-front, BEFORE any subactor is spawned.
|
||||||
|
|
||||||
|
'''
|
||||||
|
if not (
|
||||||
|
inspect.iscoroutinefunction(fn)
|
||||||
|
and
|
||||||
|
not getattr(
|
||||||
|
fn,
|
||||||
|
'_tractor_stream_function',
|
||||||
|
False,
|
||||||
|
)
|
||||||
|
):
|
||||||
|
raise TypeError(
|
||||||
|
f'{fn!r} must be a non-streaming async '
|
||||||
|
f'function!'
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _invoke_in_subactor(
|
||||||
|
an: ActorNursery,
|
||||||
|
fn: Callable,
|
||||||
|
name: str,
|
||||||
|
spawn_kwargs: dict[str, Any],
|
||||||
|
fn_kwargs: dict[str, Any],
|
||||||
|
) -> Any:
|
||||||
|
'''
|
||||||
|
Spawn a (daemon) subactor via `an.start_actor()`,
|
||||||
|
schedule `fn` as its lone remote task via
|
||||||
|
`Portal.run()` and, ALWAYS, reap the subactor once
|
||||||
|
that task's result (or error) has been delivered.
|
||||||
|
|
||||||
|
'''
|
||||||
|
portal: Portal = await an.start_actor(
|
||||||
|
name,
|
||||||
|
**spawn_kwargs,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
return await portal.run(
|
||||||
|
fn,
|
||||||
|
**fn_kwargs,
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
# one-shot semantics: the subactor's lifetime is
|
||||||
|
# bound to its lone task's completion; the
|
||||||
|
# cancel-req's bounded wait is shielded
|
||||||
|
# internally (see `Portal.cancel_actor()`) so
|
||||||
|
# this reap also runs when the caller's scope
|
||||||
|
# was itself cancelled.
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
async def run(
|
||||||
|
fn: Callable,
|
||||||
|
*,
|
||||||
|
|
||||||
|
# actor "placement": reuse an already-running peer
|
||||||
|
# via its `portal`, spawn a fresh subactor from
|
||||||
|
# a caller-managed `an: ActorNursery`, or, when
|
||||||
|
# neither is provided, open a private actor-nursery
|
||||||
|
# (implicitly booting the actor-runtime as needed)
|
||||||
|
# scoped to just this call.
|
||||||
|
portal: Portal|None = None,
|
||||||
|
an: ActorNursery|None = None,
|
||||||
|
|
||||||
|
# subactor spawn opts passed (mostly) verbatim to
|
||||||
|
# `ActorNursery.start_actor()`; unused when `portal`
|
||||||
|
# is provided.
|
||||||
|
name: str|None = None,
|
||||||
|
bind_addrs: list[UnwrappedAddress]|None = None,
|
||||||
|
enable_modules: list[str]|None = None,
|
||||||
|
loglevel: str|None = None,
|
||||||
|
debug_mode: bool|None = None,
|
||||||
|
infect_asyncio: bool = False,
|
||||||
|
inherit_parent_main: bool = True,
|
||||||
|
proc_kwargs: dict[str, Any]|None = None,
|
||||||
|
|
||||||
|
# passed verbatim to the private `open_nursery()`
|
||||||
|
# (and in turn any implicit `open_root_actor()`)
|
||||||
|
# when NO `an`/`portal` is provided.
|
||||||
|
runtime_kwargs: dict[str, Any]|None = None,
|
||||||
|
|
||||||
|
**fn_kwargs, # explicit (keyword) args to `fn`
|
||||||
|
|
||||||
|
) -> Any:
|
||||||
|
'''
|
||||||
|
Run the async `fn` as the lone task in a (new)
|
||||||
|
subactor, block waiting on its result and return it;
|
||||||
|
the distributed-parallelism equivalent of
|
||||||
|
`trio.to_thread.run_sync()`.
|
||||||
|
|
||||||
|
Unlike `ActorNursery.run_in_actor()` (which returns
|
||||||
|
a `Portal` whose result is only collected at
|
||||||
|
actor-nursery teardown) this is a plain "call and
|
||||||
|
wait" primitive: any remote error is raised HERE, in
|
||||||
|
the caller's task. Concurrency is composed the usual
|
||||||
|
`trio` way by scheduling multiple `run()` calls in
|
||||||
|
a local task nursery, ideally against a shared
|
||||||
|
caller-managed `an: ActorNursery` (see the test
|
||||||
|
suite for the canonical worker-pool-ish pattern).
|
||||||
|
|
||||||
|
'''
|
||||||
|
__runtimeframe__: int = 1 # noqa
|
||||||
|
_validate_one_shot_fn(fn)
|
||||||
|
|
||||||
|
if (
|
||||||
|
runtime_kwargs
|
||||||
|
and
|
||||||
|
(
|
||||||
|
an is not None
|
||||||
|
or
|
||||||
|
portal is not None
|
||||||
|
)
|
||||||
|
):
|
||||||
|
raise ValueError(
|
||||||
|
'`runtime_kwargs` only applies when this '
|
||||||
|
'call opens its own private actor-nursery '
|
||||||
|
'(no `an`/`portal` provided)!'
|
||||||
|
)
|
||||||
|
|
||||||
|
if portal is not None:
|
||||||
|
if an is not None:
|
||||||
|
raise ValueError(
|
||||||
|
'Pass at most ONE of `portal` or `an`, '
|
||||||
|
'not both!'
|
||||||
|
)
|
||||||
|
return await portal.run(
|
||||||
|
fn,
|
||||||
|
**fn_kwargs,
|
||||||
|
)
|
||||||
|
|
||||||
|
name: str = name or fn.__name__
|
||||||
|
spawn_kwargs: dict[str, Any] = dict(
|
||||||
|
enable_modules=(
|
||||||
|
[fn.__module__]
|
||||||
|
+
|
||||||
|
(enable_modules or [])
|
||||||
|
),
|
||||||
|
bind_addrs=bind_addrs,
|
||||||
|
loglevel=loglevel,
|
||||||
|
debug_mode=debug_mode,
|
||||||
|
infect_asyncio=infect_asyncio,
|
||||||
|
inherit_parent_main=inherit_parent_main,
|
||||||
|
proc_kwargs=proc_kwargs,
|
||||||
|
)
|
||||||
|
if an is not None:
|
||||||
|
return await _invoke_in_subactor(
|
||||||
|
an,
|
||||||
|
fn,
|
||||||
|
name,
|
||||||
|
spawn_kwargs,
|
||||||
|
fn_kwargs,
|
||||||
|
)
|
||||||
|
|
||||||
|
async with open_nursery(
|
||||||
|
**(runtime_kwargs or {}),
|
||||||
|
) as an:
|
||||||
|
return await _invoke_in_subactor(
|
||||||
|
an,
|
||||||
|
fn,
|
||||||
|
name,
|
||||||
|
spawn_kwargs,
|
||||||
|
fn_kwargs,
|
||||||
|
)
|
||||||
Loading…
Reference in New Issue