Add 5 examples filling doc-tutorial gaps
New runnable (and so CI-auto-tested) scripts backing the new docs guides, each following the `test_docs_examples.py` runner conventions, - `typed_payloads.py`: `@tractor.context(pld_spec=)` typed `started()`/stream roundtrip + a deliberate `MsgTypeError` catch demoing send-side validation, - `nested_actor_tree.py`: 3-level tree w/ fan-out rpc through a mid-tier supervisor actor, - `service_daemon_discovery.py`: registered daemon located via `find_actor()`/`wait_for_actor()` sans any spawn-portal ref, - `uds_transport_actor_tree.py`: `enable_transports=['uds']` tree printing the filesystem sockaddr + kernel peer-pid creds, - `streaming_broadcast_fanout.py`: one ipc stream fanned out to N local tasks via `MsgStream.subscribe()`. All gaps were mined from in-code TODOs + the docs recon pass; see #175 for the orig tutorial wishlist. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-codedocs_vibed_for_serious
parent
f11a74c15d
commit
759d46dd17
|
|
@ -0,0 +1,105 @@
|
||||||
|
'''
|
||||||
|
Demonstrate a (3-level) nested actor tree where one RPC from
|
||||||
|
the root fans out through a mid-tier 'supervisor' actor to
|
||||||
|
2 'leaf' worker actors and an aggregate result is relayed
|
||||||
|
back up.
|
||||||
|
|
||||||
|
The process tree should look approximately like:
|
||||||
|
|
||||||
|
python examples/nested_actor_tree.py
|
||||||
|
`-python -m tractor._child --uid ('supervisor', '7c9b1039 ..)
|
||||||
|
|-python -m tractor._child --uid ('leaf_1', '92d62f50 ..)
|
||||||
|
`-python -m tractor._child --uid ('leaf_2', 'de91fdf5 ..)
|
||||||
|
|
||||||
|
Teardown runs inside-out: the supervisor cancels its leaves
|
||||||
|
first, then the root cancels the supervisor; watch the
|
||||||
|
prints to see the ordering.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
|
||||||
|
async def compute_square(x: int) -> int:
|
||||||
|
'''
|
||||||
|
Tiny "work unit" run inside a leaf actor.
|
||||||
|
|
||||||
|
'''
|
||||||
|
name: str = tractor.current_actor().name
|
||||||
|
print(f'{name}: squaring {x}')
|
||||||
|
return x * x
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context
|
||||||
|
async def fan_out_squares(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
vals: list[int],
|
||||||
|
) -> list[int]:
|
||||||
|
'''
|
||||||
|
Spawn a (nested) pair of leaf actors, fan the input vals
|
||||||
|
out across them round-robin style, then return the
|
||||||
|
aggregated squares to our parent.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portals: list[tractor.Portal] = []
|
||||||
|
for i in (1, 2):
|
||||||
|
portals.append(
|
||||||
|
await an.start_actor(
|
||||||
|
f'leaf_{i}',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
# unblock the parent's `.open_context()` entry and
|
||||||
|
# report which leaves came up.
|
||||||
|
await ctx.started(
|
||||||
|
[p.chan.aid.name for p in portals]
|
||||||
|
)
|
||||||
|
squares: dict[int, int] = {}
|
||||||
|
|
||||||
|
async def run_in_leaf(
|
||||||
|
portal: tractor.Portal,
|
||||||
|
x: int,
|
||||||
|
) -> None:
|
||||||
|
squares[x] = await portal.run(
|
||||||
|
compute_square,
|
||||||
|
x=x,
|
||||||
|
)
|
||||||
|
|
||||||
|
# fan out one sub-RPC per input val, concurrently.
|
||||||
|
async with trio.open_nursery() as tn:
|
||||||
|
for i, x in enumerate(vals):
|
||||||
|
tn.start_soon(
|
||||||
|
run_in_leaf,
|
||||||
|
portals[i % len(portals)],
|
||||||
|
x,
|
||||||
|
)
|
||||||
|
# graceful inside-out teardown: leaves go first!
|
||||||
|
for portal in portals:
|
||||||
|
leaf_name: str = portal.chan.aid.name
|
||||||
|
print(f'supervisor: cancelling {leaf_name}')
|
||||||
|
await portal.cancel_actor()
|
||||||
|
return [squares[x] for x in vals]
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portal = await an.start_actor(
|
||||||
|
'supervisor',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
async with portal.open_context(
|
||||||
|
fan_out_squares,
|
||||||
|
vals=[1, 2, 3, 4],
|
||||||
|
) as (ctx, leaf_names):
|
||||||
|
print(f'root: supervisor spawned {leaf_names}')
|
||||||
|
squares: list[int] = await ctx.wait_for_result()
|
||||||
|
assert squares == [1, 4, 9, 16]
|
||||||
|
print(f'root: aggregate result {squares}')
|
||||||
|
print('root: cancelling supervisor')
|
||||||
|
await portal.cancel_actor()
|
||||||
|
print('root: tree torn down, what zombies?')
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
|
|
@ -0,0 +1,68 @@
|
||||||
|
'''
|
||||||
|
Demonstrate the "service daemon" pattern: a named,
|
||||||
|
long-lived actor spawned via `ActorNursery.start_actor()`
|
||||||
|
which any other task can locate through the registrar using
|
||||||
|
`tractor.find_actor()` / `tractor.wait_for_actor()` - no
|
||||||
|
spawn-portal required - and RPC into directly.
|
||||||
|
|
||||||
|
Teardown is explicit and graceful via `portal.cancel_actor()`
|
||||||
|
once the clients are done.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
_quotes: dict[str, float] = {
|
||||||
|
'btcusdt': 66_000.5,
|
||||||
|
'ethusdt': 3_500.25,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
async def get_quote(sym: str) -> float:
|
||||||
|
'''
|
||||||
|
Look up the "current" quote for a symbol.
|
||||||
|
|
||||||
|
'''
|
||||||
|
name: str = tractor.current_actor().name
|
||||||
|
print(f'{name}: serving quote for {sym!r}')
|
||||||
|
return _quotes[sym]
|
||||||
|
|
||||||
|
|
||||||
|
async def client_task() -> None:
|
||||||
|
'''
|
||||||
|
Locate the quote service by name and RPC it; note no
|
||||||
|
spawn-nursery/portal reference is ever passed in here!
|
||||||
|
|
||||||
|
'''
|
||||||
|
# a lookup miss yields `None` (not an error).
|
||||||
|
async with tractor.find_actor('no_such_svc') as portal:
|
||||||
|
assert portal is None
|
||||||
|
print('client: "no_such_svc" is not registered')
|
||||||
|
# block until the service shows up in the registry,
|
||||||
|
# then call into it through the delivered portal.
|
||||||
|
async with tractor.wait_for_actor('quote_svc') as portal:
|
||||||
|
quote: float = await portal.run(
|
||||||
|
get_quote,
|
||||||
|
sym='btcusdt',
|
||||||
|
)
|
||||||
|
print(f'client: got btcusdt quote {quote}')
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portal = await an.start_actor(
|
||||||
|
'quote_svc',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
# run the client in a separate task which discovers
|
||||||
|
# the daemon purely by its registered name.
|
||||||
|
async with trio.open_nursery() as tn:
|
||||||
|
tn.start_soon(client_task)
|
||||||
|
# explicit graceful teardown of the daemon.
|
||||||
|
print('root: cancelling quote_svc')
|
||||||
|
await portal.cancel_actor()
|
||||||
|
print('root: service shut down cleanly')
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
|
|
@ -0,0 +1,84 @@
|
||||||
|
'''
|
||||||
|
Demonstrate fanning out ONE inter-actor `MsgStream` to N
|
||||||
|
local (parent-side) trio tasks using `MsgStream.subscribe()`:
|
||||||
|
each subscriber gets its own `BroadcastReceiver` copy of
|
||||||
|
every msg from the single underlying IPC stream.
|
||||||
|
|
||||||
|
The child waits for a 'go' msg so that all subscribers are
|
||||||
|
guaranteed-attached before the first tick is sent; when the
|
||||||
|
child's stream closes each subscriber's `async for` ends
|
||||||
|
cleanly.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context
|
||||||
|
async def tick_stream(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
count: int,
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Send `count` "ticks" once the parent says go.
|
||||||
|
|
||||||
|
'''
|
||||||
|
await ctx.started(count)
|
||||||
|
async with ctx.open_stream() as stream:
|
||||||
|
# wait for the go-signal ensuring every parent-side
|
||||||
|
# subscriber is attached before any tick is sent.
|
||||||
|
assert await stream.receive() == 'go'
|
||||||
|
for i in range(count):
|
||||||
|
await stream.send(i)
|
||||||
|
# falling out gracefully closes our stream side;
|
||||||
|
# all parent-side subscribers see end-of-channel.
|
||||||
|
|
||||||
|
|
||||||
|
async def consume(
|
||||||
|
name: str,
|
||||||
|
stream: tractor.MsgStream,
|
||||||
|
task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED,
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Consume a private broadcast-copy of the IPC stream.
|
||||||
|
|
||||||
|
'''
|
||||||
|
async with stream.subscribe() as bcaster:
|
||||||
|
task_status.started()
|
||||||
|
ticks: list[int] = []
|
||||||
|
async for tick in bcaster:
|
||||||
|
print(f'{name}: rx {tick}')
|
||||||
|
ticks.append(tick)
|
||||||
|
# EVERY subscriber gets its own full copy B)
|
||||||
|
print(f'{name}: stream ended, got {ticks}')
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portal = await an.start_actor(
|
||||||
|
'ticker',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
async with (
|
||||||
|
portal.open_context(
|
||||||
|
tick_stream,
|
||||||
|
count=5,
|
||||||
|
) as (ctx, first),
|
||||||
|
ctx.open_stream() as stream,
|
||||||
|
):
|
||||||
|
assert first == 5
|
||||||
|
async with trio.open_nursery() as tn:
|
||||||
|
# use `.start()` so each consumer is known
|
||||||
|
# to be subscribed before the ticks flow.
|
||||||
|
for i in range(3):
|
||||||
|
await tn.start(
|
||||||
|
consume,
|
||||||
|
f'sub_{i}',
|
||||||
|
stream,
|
||||||
|
)
|
||||||
|
await stream.send('go')
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
|
|
@ -0,0 +1,85 @@
|
||||||
|
'''
|
||||||
|
Demonstrate "typed messaging": applying a `msgspec.Struct`
|
||||||
|
payload-type-spec to an IPC context via
|
||||||
|
`@tractor.context(pld_spec=...)`.
|
||||||
|
|
||||||
|
The child's `ctx.started()` value is stringently (round-trip)
|
||||||
|
validated against the spec *on the send side*, so a mistyped
|
||||||
|
payload raises a `tractor.MsgTypeError` before it ever hits
|
||||||
|
the wire; stream payloads are checked on `receive()` and
|
||||||
|
decode natively to the struct type.
|
||||||
|
|
||||||
|
'''
|
||||||
|
from msgspec import Struct
|
||||||
|
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
|
||||||
|
class Point(Struct):
|
||||||
|
'''
|
||||||
|
A simple 2D-coordinate msg-payload type.
|
||||||
|
|
||||||
|
'''
|
||||||
|
x: int
|
||||||
|
y: int
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context(pld_spec=Point|None)
|
||||||
|
async def point_doubler(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Stream back each received `Point` with doubled fields.
|
||||||
|
|
||||||
|
'''
|
||||||
|
# deliberately send a non-`Point` as our started-value;
|
||||||
|
# the send-side pld-spec validation catches it locally
|
||||||
|
# BEFORE anything is shipped over IPC.
|
||||||
|
try:
|
||||||
|
await ctx.started('this is no Point..')
|
||||||
|
except tractor.MsgTypeError:
|
||||||
|
print(
|
||||||
|
'child: just as planned, a `str` payload failed '
|
||||||
|
'the `Point|None` pld-spec B)'
|
||||||
|
)
|
||||||
|
# now do it right; the parent receives this as the 2nd
|
||||||
|
# element of its `.open_context()` entry tuple.
|
||||||
|
await ctx.started(Point(x=0, y=0))
|
||||||
|
async with ctx.open_stream() as stream:
|
||||||
|
async for pt in stream:
|
||||||
|
# natively decoded to our struct type!
|
||||||
|
assert type(pt) is Point
|
||||||
|
await stream.send(
|
||||||
|
Point(
|
||||||
|
x=pt.x * 2,
|
||||||
|
y=pt.y * 2,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery() as an:
|
||||||
|
portal = await an.start_actor(
|
||||||
|
'point_doubler',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
async with (
|
||||||
|
portal.open_context(
|
||||||
|
point_doubler,
|
||||||
|
) as (ctx, first),
|
||||||
|
ctx.open_stream() as stream,
|
||||||
|
):
|
||||||
|
# the (validated) started-value from the child
|
||||||
|
assert first == Point(x=0, y=0)
|
||||||
|
for i in range(3):
|
||||||
|
await stream.send(Point(x=i, y=i))
|
||||||
|
doubled: Point = await stream.receive()
|
||||||
|
assert doubled == Point(x=i * 2, y=i * 2)
|
||||||
|
print(f'parent: rx {doubled}')
|
||||||
|
# explicitly teardown the daemon-actor
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
|
|
@ -0,0 +1,53 @@
|
||||||
|
'''
|
||||||
|
Demonstrate an actor tree which talks over unix-domain-socket
|
||||||
|
(UDS) transport instead of the default TCP: pass
|
||||||
|
`enable_transports=['uds']` when opening the root and every
|
||||||
|
subactor inherits the preference.
|
||||||
|
|
||||||
|
The child's channel address is a filesystem socket path (no
|
||||||
|
TCP port in sight!) and, as a kernel-provided bonus, the
|
||||||
|
peer's pid is exchanged for free via `SO_PEERCRED`.
|
||||||
|
|
||||||
|
'''
|
||||||
|
import os
|
||||||
|
|
||||||
|
import trio
|
||||||
|
import tractor
|
||||||
|
|
||||||
|
|
||||||
|
async def report_addr() -> str:
|
||||||
|
'''
|
||||||
|
Return this actor's own accept (bind) addr + pid.
|
||||||
|
|
||||||
|
'''
|
||||||
|
actor = tractor.current_actor()
|
||||||
|
addr: tuple = actor.accept_addr
|
||||||
|
pid: int = os.getpid()
|
||||||
|
return f'{actor.name}@{addr} pid={pid}'
|
||||||
|
|
||||||
|
|
||||||
|
async def main() -> None:
|
||||||
|
async with tractor.open_nursery(
|
||||||
|
enable_transports=['uds'],
|
||||||
|
) as an:
|
||||||
|
portal = await an.start_actor(
|
||||||
|
'uds_child',
|
||||||
|
enable_modules=[__name__],
|
||||||
|
)
|
||||||
|
# the channel's remote addr is a `UDSAddress`: a
|
||||||
|
# filesystem socket path, NOT a (host, port) pair!
|
||||||
|
raddr = portal.chan.raddr
|
||||||
|
assert raddr.proto_key == 'uds'
|
||||||
|
print(
|
||||||
|
f'portal chan tpt proto: {raddr.proto_key!r}\n'
|
||||||
|
f'portal chan sock file: {raddr.sockpath}\n'
|
||||||
|
f'kernel-reported peer pid: {raddr.maybe_pid}\n'
|
||||||
|
)
|
||||||
|
# ask the child for its own bind addr: also a
|
||||||
|
# socket-file path under the runtime dir.
|
||||||
|
print(f'child says: {await portal.run(report_addr)}')
|
||||||
|
await portal.cancel_actor()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
trio.run(main)
|
||||||
Loading…
Reference in New Issue