Use linked contexts in `to_actor.run()`

Pass target inputs positionally and normalize every retained
`functools.partial()` layer, including Python 3.14 Placeholder
binding. Validate the complete target signature before startup.

Route each ordinary async fn through a static `@context` endpoint
so remote results, errors and caller cancellation remain linked.
Send namespace and function components separately, then resolve
through `Actor._get_rpc_func()` so the RPC module allowlist remains
authoritative. Retain client-created `NamespacePath` refs so
`to_tuple()` does not re-import their callable.

Owned actors enable the endpoint's `__name__` directly. Keep
`to_actor.MODULE` as the importer-facing alias used by caller-owned
portals while retaining the target module's authorization boundary.

Cover all placement modes, nested partials, argument collisions,
linked cancellation, remote errors and authorization failures.

Caught-during: review remediation
Found-via: `/run-tests` portal cancellation regression

Review: PR #481 (opencode)
https://github.com/goodboy/tractor/pull/481#pullrequestreview-4956692120

Prompt-IO: ai/prompt-io/opencode/20260818T193005Z_bf06b4f8_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
wkt/to_actor_subpkg
Gud Boi 2026-08-18 21:44:04 -04:00
parent 99be161ec0
commit fe0a724d10
8 changed files with 628 additions and 37 deletions

View File

@ -0,0 +1,52 @@
---
model: openai/gpt-5.6-sol
service: opencode
session: ses_3e4c90d3eafeqHEtRYSIHgHhpA
timestamp: 2026-08-18T19:30:05Z
git_ref: bf06b4f8
scope: code
substantive: true
raw_file: 20260818T193005Z_bf06b4f8_prompt_io.raw.md
---
## Prompt
Replace abandoned `Portal.run()` one-shots with a static linked-context
endpoint. Follow Trio positional-call semantics, use partials for target
keywords, preserve Python 3.14 Placeholder behavior, keep target lookup
behind the RPC allowlist and support private, nursery and portal
placement.
## Response summary
Use `Portal.open_context()` and `Context.wait_for_result()` for one-shot
tasks. Normalize every partial layer, validate signatures locally and
send target namespace/function components separately to the authorized
remote resolver. Retain the client-side function in its `NamespacePath`
so `to_tuple()` does not re-import it. Owned actors enable the declaring
`_api.__name__` directly; caller-owned portals opt in through the public
`to_actor.MODULE` alias.
## Files changed
- `tractor/to_actor/_api.py` - implement linked one-shot calls.
- `tractor/to_actor/__init__.py` - export `MODULE`.
- `tractor/msg/ptr.py` - retain refs created by `from_ref()`.
- `tests/test_to_actor.py` - cover the public API and authorization.
- `examples/parallelism/to_actor_one_shots.py` - use positional inputs.
## Human edits
The human rejected nested target-kwargs configuration and selected
Trio-style positional inputs plus `functools.partial()`. During staged
review the human required a Python 3.14 compatibility comment rather
than removing Placeholder support, requested separate namespace and
function inputs, preserved `_get_rpc_func(ns: str, funcname: str)`
authorization, renamed `RPC_MODULE` to `MODULE`, rejected global module
exposure and deferred speculative nursery/module-list helpers to the
`open_taskman()` design line. The human also required this public API
to land only after its lower-level safety dependencies. In final staged
review, the human required `_invoke_from_portal()` to use
`NamespacePath.to_tuple()` with the already-held function ref and
required internal actor setup to use `_api.__name__` directly, keeping
`to_actor.MODULE` solely as the public importer-facing alias.

View File

@ -0,0 +1,24 @@
---
model: openai/gpt-5.6-sol
service: opencode
timestamp: 2026-08-18T19:30:05Z
git_ref: bf06b4f8
diff_cmd: git diff HEAD~1..HEAD
---
Implement `to_actor.run()` with Trio-style positional target arguments,
`functools.partial` keyword and Python 3.14 Placeholder binding, and a
static context endpoint that links remote results, errors and caller
cancellation.
> `git diff HEAD~1..HEAD -- tractor/to_actor/_api.py tractor/to_actor/__init__.py`
Resolve target functions through `Actor._get_rpc_func()` so module
authorization remains authoritative. Automatically expose the helper
module for actors owned by `to_actor.run()` and document explicit
exposure for a caller-owned portal.
> `git diff HEAD~1..HEAD -- tests/test_to_actor.py examples/parallelism/to_actor_one_shots.py`
Cover placement modes, argument binding, nested partials, caller-linked
cancellation, remote errors and module authorization.

View File

@ -41,7 +41,7 @@ async def main() -> None:
# subactor, tears the runtime back down. # subactor, tears the runtime back down.
assert await tractor.to_actor.run( assert await tractor.to_actor.run(
is_prime, is_prime,
n=2, 2,
) )
# the "worker-pool-ish" pattern from the original # the "worker-pool-ish" pattern from the original
@ -57,9 +57,9 @@ async def main() -> None:
) -> None: ) -> None:
results[n] = await tractor.to_actor.run( results[n] = await tractor.to_actor.run(
is_prime, is_prime,
n,
an=an, an=an,
name=f'prime_checker_{i}', name=f'prime_checker_{i}',
n=n,
) )
inputs: list[int] = [ inputs: list[int] = [

View File

@ -7,6 +7,7 @@ https://github.com/goodboy/tractor/issues/477
''' '''
from functools import partial from functools import partial
from pathlib import Path
import pytest import pytest
import trio import trio
@ -16,6 +17,9 @@ from tractor import (
to_actor, to_actor,
) )
from tractor._testing import tractor_test from tractor._testing import tractor_test
from tractor.msg import ptr as msgptr
from tractor.msg.ptr import NamespacePath
from tractor.to_actor import _api as to_actor_api
async def add_one( async def add_one(
@ -28,6 +32,98 @@ async def raise_value_error() -> None:
raise ValueError('kaboom') raise ValueError('kaboom')
async def echo_control_names(
value: int,
/,
*,
name: str,
portal: str,
an: str,
runtime_kwargs: str,
) -> dict[str, int|str]:
return {
'value': value,
'name': name,
'portal': portal,
'an': an,
'runtime_kwargs': runtime_kwargs,
}
async def mark_task_cancellation(
started_path: str,
cancelled_path: str,
) -> None:
Path(started_path).touch()
try:
await trio.sleep_forever()
finally:
Path(cancelled_path).touch()
async def echo_startup_control(
_cancel_on_startup: str,
) -> str:
return _cancel_on_startup
async def collect_args(
*args: object,
) -> tuple[object, ...]:
return args
async def collect_call(
*args: object,
**kwargs: object,
) -> tuple[tuple[object, ...], dict[str, object]]:
return args, kwargs
def _non_registration_contexts(
actor: tractor.Actor,
) -> dict[tuple, str]:
return {
key: str(ctx._nsf)
for key, ctx in actor._contexts.items()
if str(ctx._nsf) != (
'tractor.discovery._registry:'
'Registrar.register_actor'
)
}
def test_namespace_path_retains_target_ref(
monkeypatch: pytest.MonkeyPatch,
):
'''
Reuse the client-side target ref when splitting its namespace path.
`NamespacePath.from_ref()` previously discarded `add_one`, so
`to_tuple()` imported and resolved the just-created string again.
Replacing `resolve_name()` with a failure proves the retained ref
supplies the tuple without a redundant lookup. The public module
alias assertion also keeps internal `_api.__name__` authoritative.
'''
target = NamespacePath.from_ref(add_one)
def fail_resolve(name: str) -> object:
raise AssertionError(f'unexpected lookup for {name!r}')
monkeypatch.setattr(
msgptr,
'resolve_name',
fail_resolve,
)
assert target.to_tuple() == (
add_one.__module__,
add_one.__name__,
)
assert to_actor.MODULE == to_actor_api.__name__
assert not hasattr(to_actor_api, 'MODULE')
@tractor_test @tractor_test
async def test_one_shot_in_private_nursery( async def test_one_shot_in_private_nursery(
start_method: str, start_method: str,
@ -40,7 +136,7 @@ async def test_one_shot_in_private_nursery(
''' '''
assert await to_actor.run( assert await to_actor.run(
add_one, add_one,
n=1, 1,
) == 2 ) == 2
@ -61,7 +157,7 @@ def test_one_shot_boots_implicit_runtime(
) is None ) is None
result = await to_actor.run( result = await to_actor.run(
add_one, add_one,
n=41, 41,
runtime_kwargs=dict( runtime_kwargs=dict(
registry_addrs=[reg_addr], registry_addrs=[reg_addr],
start_method=start_method, start_method=start_method,
@ -110,8 +206,8 @@ async def test_spawn_from_caller_nursery(
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
assert await to_actor.run( assert await to_actor.run(
add_one, add_one,
10,
an=an, an=an,
n=10,
) == 11 ) == 11
assert not an._children assert not an._children
@ -152,8 +248,8 @@ async def test_cancel_ack_failure_hard_reaps_child(
with trio.fail_after(5): with trio.fail_after(5):
assert await to_actor.run( assert await to_actor.run(
add_one, add_one,
20,
an=an, an=an,
n=20,
) == 21 ) == 21
assert not an._children assert not an._children
@ -213,19 +309,35 @@ async def test_reuse_existing_actor_via_portal(
Pass `portal=` to schedule the one-shot task in an Pass `portal=` to schedule the one-shot task in an
already-running actor; no spawn, no implicit reap. already-running actor; no spawn, no implicit reap.
The low-level `Portal.run_from_ns()` assertion also proves its
target kwargs remain separate from the private startup-cancel
policy used by context cleanup.
''' '''
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
actor = tractor.current_actor()
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'one_shot_worker', 'one_shot_worker',
enable_modules=[__name__], enable_modules=[
__name__,
to_actor.MODULE,
],
) )
contexts_before = _non_registration_contexts(actor)
for i in range(3): for i in range(3):
assert await to_actor.run( assert await to_actor.run(
add_one, add_one,
i,
portal=portal, portal=portal,
n=i,
) == i + 1 ) == i + 1
assert await portal.run_from_ns(
__name__,
'echo_startup_control',
_cancel_on_startup='target_value',
) == 'target_value'
assert _non_registration_contexts(actor) == contexts_before
# still alive: caller owns the actor's lifetime. # still alive: caller owns the actor's lifetime.
await portal.cancel_actor() await portal.cancel_actor()
@ -251,9 +363,9 @@ async def test_concurrent_one_shots_from_task_nursery(
) -> None: ) -> None:
results[i] = await to_actor.run( results[i] = await to_actor.run(
add_one, add_one,
i,
an=an, an=an,
name=f'one_shot_{i}', name=f'one_shot_{i}',
n=i,
) )
async with ( async with (
@ -304,6 +416,83 @@ def test_rejects_streaming_fn():
) )
def test_partial_placeholder_normalization(
monkeypatch: pytest.MonkeyPatch,
):
'''
Preserve Python 3.14 `functools.partial` placeholder semantics.
The test environment runs Python 3.13, so this installs an identity
sentinel matching Python 3.14's `functools.Placeholder` API.
Interleaved placeholders prove call-time positional arguments are
merged in order. Undersupply and a mismatched final target
signature both fail locally before actor runtime startup.
'''
placeholder = object()
monkeypatch.setattr(
to_actor_api.functools,
'Placeholder',
placeholder,
raising=False,
)
fn = partial(
collect_args,
placeholder,
2,
placeholder,
)
normalized_fn, args, kwargs = to_actor_api._normalize_call(
fn,
(1, 3, 4),
)
assert normalized_fn is collect_args
assert args == (1, 2, 3, 4)
assert kwargs == {}
with pytest.raises(TypeError, match='Not enough positional'):
to_actor_api._normalize_call(fn, (1,))
with pytest.raises(TypeError, match='too many positional'):
to_actor_api._normalize_call(
partial(add_one, 1),
(2,),
)
def test_nested_partial_normalization():
'''
Flatten every retained `functools.partial` layer before RPC.
CPython normally combines nested partials, but preserves the inner
object when it has instance attributes. Unwrapping only the outer
layer left a non-namespace-addressable partial as the RPC target.
The custom attribute triggers that retained shape; the assertions
prove positional ordering and outer-keyword precedence match a
direct nested-partial call.
'''
inner = partial(
collect_call,
1,
label='inner',
)
inner.note = 'retain this partial layer'
outer = partial(
inner,
2,
label='outer',
)
fn, args, kwargs = to_actor_api._normalize_call(
outer,
(3,),
)
assert fn is collect_call
assert args == (1, 2, 3)
assert kwargs == {'label': 'outer'}
def test_rejects_portal_and_an_combo(): def test_rejects_portal_and_an_combo():
''' '''
`portal=` and `an=` are mutually exclusive `portal=` and `an=` are mutually exclusive
@ -315,9 +504,9 @@ def test_rejects_portal_and_an_combo():
partial( partial(
to_actor.run, to_actor.run,
add_one, add_one,
1,
portal=object(), portal=object(),
an=object(), an=object(),
n=1,
) )
) )
@ -335,10 +524,179 @@ def test_rejects_runtime_kwargs_with_placement():
partial( partial(
to_actor.run, to_actor.run,
add_one, add_one,
1,
an=object(), an=object(),
runtime_kwargs=dict( runtime_kwargs=dict(
loglevel='cancel', loglevel='cancel',
), ),
n=1,
) )
) )
@tractor_test
async def test_trio_style_args_and_partial_kwargs(
start_method: str,
debug_mode: bool,
):
'''
Forward positional args and partial-bound keyword arguments.
The original API captured every keyword matching an actor
control, so ordinary target parameters such as `name`, `portal`,
`an` and `runtime_kwargs` could not be called. This test uses a
positional-only target argument plus all colliding keyword names.
Binding the target keywords with `functools.partial()` proves the
Trio-style calling convention keeps target inputs separate from
actor controls.
'''
fn = partial(
echo_control_names,
name='target_name',
portal='target_portal',
an='target_an',
runtime_kwargs='target_runtime_kwargs',
)
async with tractor.open_nursery() as an:
result = await to_actor.run(
fn,
42,
an=an,
name='actor_name',
)
assert result == {
'value': 42,
'name': 'target_name',
'portal': 'target_portal',
'an': 'target_an',
'runtime_kwargs': 'target_runtime_kwargs',
}
@tractor_test
async def test_portal_task_cancelled_with_local_caller(
tmp_path: Path,
start_method: str,
debug_mode: bool,
):
'''
Couple a reused portal's remote task to its local caller.
The former `Portal.run()` path abandoned its remote task when the
local `to_actor.run()` caller was cancelled. The target writes
one file after starting and another from its cancellation
`finally`. Cancelling the local task nursery and observing the
second file proves `Portal.open_context()` propagated
cancellation before the caller exited. A subsequent call proves
the caller-owned actor was not cancelled with that task.
'''
started_path = tmp_path / 'started'
cancelled_path = tmp_path / 'cancelled'
async with tractor.open_nursery() as an:
actor = tractor.current_actor()
portal: tractor.Portal = await an.start_actor(
'context_worker',
enable_modules=[
__name__,
to_actor.MODULE,
],
)
contexts_before = _non_registration_contexts(actor)
async with trio.open_nursery() as tn:
tn.start_soon(
partial(
to_actor.run,
mark_task_cancellation,
str(started_path),
str(cancelled_path),
portal=portal,
),
)
with trio.fail_after(5):
while not started_path.exists():
await trio.sleep(0.01)
tn.cancel_scope.cancel()
assert cancelled_path.exists()
assert _non_registration_contexts(actor) == contexts_before
assert await to_actor.run(
add_one,
1,
portal=portal,
) == 2
assert _non_registration_contexts(actor) == contexts_before
await portal.cancel_actor()
@tractor_test
async def test_context_trampoline_preserves_module_allowlist(
start_method: str,
debug_mode: bool,
):
'''
Keep target resolution behind the actor's RPC module allowlist.
Loading the target with `NamespacePath.load_ref()` would silently
bypass the actor's existing module-exposure boundary. This actor
exposes only the trusted trampoline, not the test module; the
boxed `ModuleNotExposed` proves the trampoline delegates target
resolution to `Actor._get_rpc_func()`.
'''
async with tractor.open_nursery() as an:
actor = tractor.current_actor()
portal: tractor.Portal = await an.start_actor(
'restricted_context_worker',
enable_modules=[to_actor.MODULE],
)
contexts_before = _non_registration_contexts(actor)
with pytest.raises(RemoteActorError) as excinfo:
await to_actor.run(
add_one,
1,
portal=portal,
)
assert excinfo.value.boxed_type is tractor.ModuleNotExposed
assert _non_registration_contexts(actor) == contexts_before
await portal.cancel_actor()
@tractor_test
async def test_portal_requires_context_trampoline(
start_method: str,
debug_mode: bool,
):
'''
Require explicit trampoline exposure on a caller-owned actor.
Automatically exposing the module in every actor weakens the RPC
allowlist for actors that never use `to_actor.run()`. A portal to
such an actor instead fails with the usual `ModuleNotExposed`,
naming the module callers must opt into.
'''
async with tractor.open_nursery() as an:
actor = tractor.current_actor()
portal: tractor.Portal = await an.start_actor(
'no_context_trampoline_worker',
enable_modules=[__name__],
)
contexts_before = _non_registration_contexts(actor)
with pytest.raises(RemoteActorError) as excinfo:
await to_actor.run(
add_one,
1,
portal=portal,
)
err = excinfo.value
assert err.boxed_type is tractor.ModuleNotExposed
assert to_actor.MODULE in str(err)
assert _non_registration_contexts(actor) == contexts_before
await portal.cancel_actor()

View File

@ -125,7 +125,9 @@ class NamespacePath(str):
) -> NamespacePath: ) -> NamespacePath:
fqnp: tuple[str, str] = cls._mk_fqnp(ref) fqnp: tuple[str, str] = cls._mk_fqnp(ref)
return cls(':'.join(fqnp)) nsp = cls(':'.join(fqnp))
nsp._ref = ref
return nsp
def to_tuple( def to_tuple(
self, self,

View File

@ -600,14 +600,21 @@ class Actor:
# - cancel_rpc_tasks(), # - cancel_rpc_tasks(),
# - _cancel_task(), # - _cancel_task(),
# #
def _get_rpc_func(self, ns, funcname): def _get_rpc_func(
self,
ns: str,
funcname: str,
):
''' '''
Try to lookup and return a target RPC func from the Try to lookup and return a target RPC func from the
post-fork enabled module set. post-fork enabled module set.
''' '''
try: try:
return getattr(self._mods[ns], funcname) return getattr(
self._mods[ns],
funcname,
)
except KeyError as err: except KeyError as err:
mne = ModuleNotExposed(*err.args) mne = ModuleNotExposed(*err.args)

View File

@ -22,12 +22,18 @@ Adopts the "run it over there" parlance from analogous
`anyio.to_process` but for SC-supervised actors: spawn (or `anyio.to_process` but for SC-supervised actors: spawn (or
reuse) a subactor, schedule a single remote task, wait on reuse) a subactor, schedule a single remote task, wait on
its result and (when the call owns the subactor) reap it. its result and (when the call owns the subactor) reap it.
Target arguments follow Trio's positional convention; use
`functools.partial()` to bind target keyword arguments.
The "spiritual successor" to (and eventual replacement of) The "spiritual successor" to (and eventual replacement of)
the `ActorNursery.run_in_actor()` API; see the `ActorNursery.run_in_actor()` API; see
https://github.com/goodboy/tractor/issues/477 https://github.com/goodboy/tractor/issues/477
''' '''
from . import _api as _api
from ._api import ( from ._api import (
run as run, run as run,
) )
MODULE: str = _api.__name__

View File

@ -23,8 +23,8 @@ the lower level daemon-actor spawn + portal APIs,
- `ActorNursery.start_actor()` for (daemon-style) subactor - `ActorNursery.start_actor()` for (daemon-style) subactor
spawning, spawning,
- `Portal.run()` for scheduling the lone remote task and - `Portal.open_context()` for scheduling the lone remote
waiting on its result, task with linked cancellation and waiting on its result,
- `Portal.cancel_actor()` for reaping the subactor once - `Portal.cancel_actor()` for reaping the subactor once
that result (or error) arrives, that result (or error) arrives,
@ -36,13 +36,24 @@ spawn-machinery nurseries as with the (to be deprecated)
''' '''
from __future__ import annotations from __future__ import annotations
import functools
import inspect import inspect
from typing import ( from typing import (
Any, Any,
Awaitable,
Callable, Callable,
TYPE_CHECKING, TYPE_CHECKING,
TypeVar,
TypeVarTuple,
Unpack,
) )
from .._context import (
Context,
context,
)
from ..msg.ptr import NamespacePath
from ..runtime._state import current_actor
from ..runtime._supervise import ( from ..runtime._supervise import (
ActorNursery, ActorNursery,
open_nursery, open_nursery,
@ -53,6 +64,10 @@ if TYPE_CHECKING:
from ..runtime._portal import Portal from ..runtime._portal import Portal
ArgsT = TypeVarTuple('ArgsT')
RetT = TypeVar('RetT')
def _validate_one_shot_fn( def _validate_one_shot_fn(
fn: Callable, fn: Callable,
) -> None: ) -> None:
@ -60,7 +75,7 @@ def _validate_one_shot_fn(
Ensure `fn` is a non-streaming async function, raise Ensure `fn` is a non-streaming async function, raise
a `TypeError` otherwise. a `TypeError` otherwise.
The same constraint enforced by `Portal.run()` but The same constraint enforced by `Portal.open_context()` but
checked up-front, BEFORE any subactor is spawned. checked up-front, BEFORE any subactor is spawned.
''' '''
@ -79,18 +94,127 @@ def _validate_one_shot_fn(
) )
def _normalize_call(
fn: Callable,
args: tuple[Any, ...],
) -> tuple[
Callable,
tuple[Any, ...],
dict[str, Any],
]:
'''
Normalize Trio-style positional and partial-bound arguments.
Actor calls must send a namespace-addressable base function and
serializable inputs to another process, so decompose partials and
validate their complete call signature before runtime startup.
'''
kwargs: dict[str, Any] = {}
while isinstance(fn, functools.partial):
partial_args: tuple[Any, ...] = fn.args
# `functools.Placeholder` was added in Python 3.14. Drop
# this `getattr()` guard once 3.14 is the minimum version.
placeholder = getattr(
functools,
'Placeholder',
None,
)
if (
placeholder is not None
and
any(
arg is placeholder
for arg in partial_args
)
):
call_args = iter(args)
merged_args: list[Any] = []
for arg in partial_args:
if arg is placeholder:
try:
arg = next(call_args)
except StopIteration:
raise TypeError(
'Not enough positional arguments to '
'fill `functools.Placeholder`s'
) from None
merged_args.append(arg)
merged_args.extend(call_args)
args = tuple(merged_args)
else:
args = partial_args + args
partial_kwargs = dict(fn.keywords or {})
partial_kwargs.update(kwargs)
kwargs = partial_kwargs
fn = fn.func
_validate_one_shot_fn(fn)
inspect.signature(fn).bind(*args, **kwargs)
return fn, args, kwargs
@context
async def _invoke_one_shot(
ctx: Context,
namespace: str,
funcname: str,
args: list[Any],
kwargs: dict[str, Any],
) -> Any:
'''
Invoke an ordinary async function inside a linked IPC context.
'''
# Do not use `NamespacePath.load_ref()` here: target resolution
# must remain behind the actor's RPC module allowlist.
fn: Callable = current_actor()._get_rpc_func(
namespace,
funcname,
)
_validate_one_shot_fn(fn)
await ctx.started()
return await fn(*args, **kwargs)
async def _invoke_from_portal(
portal: Portal,
fn: Callable,
args: tuple[Any, ...],
kwargs: dict[str, Any],
) -> Any:
'''
Run `fn` through the context-linked one-shot endpoint.
'''
namespace, funcname = NamespacePath.from_ref(fn).to_tuple()
async with portal.open_context(
_invoke_one_shot,
namespace=namespace,
funcname=funcname,
args=list(args),
kwargs=kwargs,
) as (ctx, _):
return await ctx.wait_for_result()
async def _invoke_in_subactor( async def _invoke_in_subactor(
an: ActorNursery, an: ActorNursery,
fn: Callable, fn: Callable,
args: tuple[Any, ...],
kwargs: dict[str, Any],
name: str, name: str,
spawn_kwargs: dict[str, Any], spawn_kwargs: dict[str, Any],
fn_kwargs: dict[str, Any],
) -> Any: ) -> Any:
''' '''
Spawn a (daemon) subactor via `an.start_actor()`, Spawn a (daemon) subactor via `an.start_actor()`,
schedule `fn` as its lone remote task via schedule `fn` as its context-linked lone remote task and,
`Portal.run()` and, ALWAYS, reap the subactor once ALWAYS, reap the subactor once that task's result (or error)
that task's result (or error) has been delivered. has been delivered.
''' '''
portal: Portal = await an.start_actor( portal: Portal = await an.start_actor(
@ -98,9 +222,11 @@ async def _invoke_in_subactor(
**spawn_kwargs, **spawn_kwargs,
) )
try: try:
return await portal.run( return await _invoke_from_portal(
portal,
fn, fn,
**fn_kwargs, args,
kwargs,
) )
finally: finally:
# Cancel and join this child before returning. The nursery # Cancel and join this child before returning. The nursery
@ -110,8 +236,8 @@ async def _invoke_in_subactor(
async def run( async def run(
fn: Callable, fn: Callable[[Unpack[ArgsT]], Awaitable[RetT]],
*, *args: Unpack[ArgsT],
# actor "placement": reuse an already-running peer # actor "placement": reuse an already-running peer
# via its `portal`, spawn a fresh subactor from # via its `portal`, spawn a fresh subactor from
@ -139,15 +265,21 @@ async def run(
# when NO `an`/`portal` is provided. # when NO `an`/`portal` is provided.
runtime_kwargs: dict[str, Any]|None = None, runtime_kwargs: dict[str, Any]|None = None,
**fn_kwargs, # explicit (keyword) args to `fn` ) -> RetT:
) -> Any:
''' '''
Run the async `fn` as the lone task in a (new) Run the async `fn(*args)` as the lone task in a (new)
subactor, block waiting on its result and return it; subactor, block waiting on its result and return it; the
the distributed-parallelism equivalent of distributed-parallelism equivalent of
`trio.to_thread.run_sync()`. `trio.to_thread.run_sync()`.
As with Trio's API, target arguments are positional. Use
`functools.partial()` to bind target keyword arguments; all
keyword arguments accepted here configure actor placement or
spawning. A caller-supplied `portal` must address an actor started
with both `tractor.to_actor.MODULE` and the target function's
module in its `enable_modules` list. Calls that spawn their own
actor add the trampoline module automatically.
Unlike `ActorNursery.run_in_actor()` (which returns Unlike `ActorNursery.run_in_actor()` (which returns
a `Portal` whose result is only collected at a `Portal` whose result is only collected at
actor-nursery teardown) this is a plain "call and actor-nursery teardown) this is a plain "call and
@ -160,7 +292,7 @@ async def run(
''' '''
__runtimeframe__: int = 1 # noqa __runtimeframe__: int = 1 # noqa
_validate_one_shot_fn(fn) fn, args, kwargs = _normalize_call(fn, args)
if ( if (
runtime_kwargs runtime_kwargs
@ -183,15 +315,22 @@ async def run(
'Pass at most ONE of `portal` or `an`, ' 'Pass at most ONE of `portal` or `an`, '
'not both!' 'not both!'
) )
return await portal.run( return await _invoke_from_portal(
portal,
fn, fn,
**fn_kwargs, args,
kwargs,
) )
name: str = name or fn.__name__ name: str = name or fn.__name__
spawn_kwargs: dict[str, Any] = dict( spawn_kwargs: dict[str, Any] = dict(
enable_modules=( enable_modules=(
[fn.__module__] [
# The public `to_actor.MODULE` alias is only for
# callers configuring an existing actor.
__name__,
fn.__module__,
]
+ +
(enable_modules or []) (enable_modules or [])
), ),
@ -206,18 +345,21 @@ async def run(
return await _invoke_in_subactor( return await _invoke_in_subactor(
an, an,
fn, fn,
args,
kwargs,
name, name,
spawn_kwargs, spawn_kwargs,
fn_kwargs,
) )
an: ActorNursery
async with open_nursery( async with open_nursery(
**(runtime_kwargs or {}), **(runtime_kwargs or {}),
) as an: ) as an:
return await _invoke_in_subactor( return await _invoke_in_subactor(
an, an,
fn, fn,
args,
kwargs,
name, name,
spawn_kwargs, spawn_kwargs,
fn_kwargs,
) )