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,
|
||||
)
|
||||
from . import msg as msg
|
||||
from . import to_actor as to_actor
|
||||
from ._root import (
|
||||
run_daemon as run_daemon,
|
||||
open_root_actor as open_root_actor,
|
||||
|
|
|
|||
|
|
@ -383,12 +383,13 @@ class ActorNursery:
|
|||
)
|
||||
|
||||
# TODO: DEPRECATE THIS:
|
||||
# -[ ] impl instead as a hilevel wrapper on
|
||||
# top of a `@context` style invocation.
|
||||
# |_ dynamic @context decoration on child side
|
||||
# |_ implicit `Portal.open_context() as (ctx, first):`
|
||||
# and `return first` on parent side.
|
||||
# |_ mention how it's similar to `trio-parallel` API?
|
||||
# -[x] impl instead as a hilevel wrapper on top of
|
||||
# the lower level daemon-spawn + portal APIs
|
||||
# |_ see `.to_actor.run()` (issue #477) which does
|
||||
# `.start_actor()` + `Portal.run()` + a one-shot
|
||||
# reap via `Portal.cancel_actor()`.
|
||||
# -[ ] emit a `DeprecationWarning` here (requires
|
||||
# migrating all in-repo usage first!)
|
||||
# -[ ] use @api_frame on the wrapper
|
||||
async def run_in_actor(
|
||||
self,
|
||||
|
|
@ -416,6 +417,11 @@ class ActorNursery:
|
|||
until the task spawned by executing ``fn`` completes at which point
|
||||
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
|
||||
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