goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

Polish the non-debug docs examples 

Finish the Python style, typing and docstring pass across the
ordinary, parallelism, Trio and integration examples.

Preserve each demo's runtime behavior while tightening callable,
portal, stream and nursery annotations. Use the modern `.chan`
portal attr and current actor-lifecycle terminology.

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))
wkt/big_boi_docs_472_follow_ups
Gud Boi 2026-08-29 18:58:43 -04:00
parent 0be872ff97
commit 96a1838210
21 changed files with 418 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
examples/a_trynamic_first_scene.py
View File

@ -1,29 +1,40 @@
import trio import trio
import tractor import tractor
_this_module = __name__ _this_module: str = __name__
the_line = 'Hi my name is {}' the_line: str = 'Hi my name is {}'
tractor.log.get_console_log("INFO") tractor.log.get_console_log('INFO')
async def hi() -> str: async def hi() -> str:
'''
Return a greeting naming the current actor.
'''
return the_line.format(tractor.current_actor().name) return the_line.format(tractor.current_actor().name)
async def say_hello(other_actor: str) -> str: async def say_hello(other_actor: str) -> str:
'''
Ask another actor to return its greeting.
'''
portal: tractor.Portal portal: tractor.Portal
async with tractor.wait_for_actor(other_actor) as portal: async with tractor.wait_for_actor(other_actor) as portal:
return await portal.run(hi) return await portal.run(hi)
async def main() -> None: async def main() -> None:
"""Main tractor entry point, the "master" process (for now '''
Main tractor entry point, the "master" process (for now
acts as the "director"). acts as the "director").
"""
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
print("Alright... Action!") print('Alright... Action!')
# both actors wait on (then dial!) the *other*, so each # both actors wait on (then dial!) the *other*, so each
# must outlive both hellos: spawn as daemons, run the # must outlive both hellos: spawn as daemons, run the
@ -40,6 +51,10 @@ async def main() -> None:
name: str, name: str,
other_actor: str, other_actor: str,
) -> None: ) -> None:
'''
Print a greeting fetched through a named actor.
'''
print( print(
# RPC through an existing actor's `Portal`. # RPC through an existing actor's `Portal`.
await portals[name].run( await portals[name].run(
@ -48,13 +63,14 @@ async def main() -> None:
) )
) )
tn: trio.Nursery
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
tn.start_soon(run_and_print, 'donny', 'gretchen') tn.start_soon(run_and_print, 'donny', 'gretchen')
tn.start_soon(run_and_print, 'gretchen', 'donny') tn.start_soon(run_and_print, 'gretchen', 'donny')
await an.cancel() await an.cancel()
print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...") print('CUTTTT CUUTT CUT!!! Donny!! You\'re supposed to say...')
if __name__ == '__main__': if __name__ == '__main__':

12
examples/actor_spawning_and_causality.py
View File

@ -3,13 +3,19 @@ import tractor
async def cellar_door() -> str: async def cellar_door() -> str:
'''
Return a phrase from a spawned actor.
'''
assert not tractor.is_root_process() assert not tractor.is_root_process()
return "Dang that's beautiful" return 'Dang that\'s beautiful'
async def main() -> None: async def main() -> None:
"""The main ``tractor`` routine. '''
""" The main ``tractor`` routine.
'''
# spawn a subactor, run ``cellar_door()`` as its lone task, # spawn a subactor, run ``cellar_door()`` as its lone task,
# block until its result arrives and the subactor is reaped. # block until its result arrives and the subactor is reaped.
print( print(

18
examples/actor_spawning_and_causality_with_daemon.py
View File

@ -3,15 +3,20 @@ import tractor
async def movie_theatre_question() -> str: async def movie_theatre_question() -> str:
"""A question asked in a dark theatre, in a tangent '''
A question asked in a dark theatre, in a tangent
(errr, I mean different) process. (errr, I mean different) process.
"""
'''
return 'have you ever seen a portal?' return 'have you ever seen a portal?'
async def main() -> None: async def main() -> None:
"""The main ``tractor`` routine. '''
""" The main ``tractor`` routine.
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
@ -24,9 +29,8 @@ async def main() -> None:
# call the subactor a 2nd time # call the subactor a 2nd time
print(await portal.run(movie_theatre_question)) print(await portal.run(movie_theatre_question))
# the async with will block here indefinitely waiting # the async with will wait indefinitely for "frank" because
# for our actor "frank" to complete, but since it's an # its runtime remains active until explicitly cancelled
# "outlive_main" actor it will never end until cancelled
await portal.cancel_actor() await portal.cancel_actor()

27
examples/asynchronous_generators.py
View File

@ -1,20 +1,31 @@
from typing import AsyncIterator
from itertools import repeat from itertools import repeat
from typing import AsyncIterator
import trio import trio
import tractor import tractor
async def stream_forever() -> AsyncIterator[int]: async def stream_forever() -> AsyncIterator[str]:
'''
Stream the same message indefinitely.
for i in repeat("I can see these little future bubble things"): '''
# each yielded value is sent over the ``Channel`` to the parent actor message: str
yield i for message in repeat(
'I can see these little future bubble things',
):
# each yielded value is sent over the ``Channel`` to the
# parent actor
yield message
await trio.sleep(0.01) await trio.sleep(0.01)
async def main() -> None: async def main() -> None:
'''
Print messages streamed from a subactor.
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
@ -24,10 +35,12 @@ async def main() -> None:
# this async for loop streams values from the above # this async for loop streams values from the above
# async generator running in a separate process # async generator running in a separate process
stream: tractor.MsgStream
async with portal.open_stream_from(stream_forever) as stream: async with portal.open_stream_from(stream_forever) as stream:
count: int = 0 count: int = 0
async for letter in stream: message: str
print(letter) async for message in stream:
print(message)
count += 1 count += 1
if count > 50: if count > 50:

62
examples/full_fledged_streaming_service.py
View File

@ -1,4 +1,6 @@
import time import time
from typing import AsyncIterator
import trio import trio
import tractor import tractor
from tractor import ( from tractor import (
@ -9,14 +11,19 @@ from tractor import (
# this is the first 2 actors, streamer_1 and streamer_2 # this is the first 2 actors, streamer_1 and streamer_2
async def stream_data(seed: int): async def stream_data(seed: int) -> AsyncIterator[int]:
'''
Stream integers up to a seed value.
'''
i: int
for i in range(seed): for i in range(seed):
yield i yield i
await trio.sleep(0.0001) # trigger scheduler await trio.sleep(0.0001) # trigger scheduler
# this is the third actor; the aggregator # this is the third actor; the aggregator
async def aggregate(seed: int): async def aggregate(seed: int) -> AsyncIterator[int]:
''' '''
Ensure that the two streams we receive match but only stream Ensure that the two streams we receive match but only stream
a single set of values to the parent. a single set of values to the parent.
@ -25,6 +32,7 @@ async def aggregate(seed: int):
an: ActorNursery an: ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portals: list[Portal] = [] portals: list[Portal] = []
i: int
for i in range(1, 3): for i in range(1, 3):
# fork/spawn call # fork/spawn call
@ -35,20 +43,35 @@ async def aggregate(seed: int):
portals.append(portal) portals.append(portal)
send_chan: trio.MemorySendChannel[int]
recv_chan: trio.MemoryReceiveChannel[int]
send_chan, recv_chan = trio.open_memory_channel(500) send_chan, recv_chan = trio.open_memory_channel(500)
async def push_to_chan(portal: Portal, send_chan): async def push_to_chan(
portal: Portal,
send_chan: trio.MemorySendChannel[int],
) -> None:
'''
Forward one remote stream into a local channel.
'''
# TODO: https://github.com/goodboy/tractor/issues/207 # TODO: https://github.com/goodboy/tractor/issues/207
async with send_chan: async with send_chan:
async with portal.open_stream_from(stream_data, seed=seed) as stream: stream: MsgStream
async with portal.open_stream_from(
stream_data,
seed=seed,
) as stream:
value: int
async for value in stream: async for value in stream:
# leverage trio's built-in backpressure # leverage trio's built-in backpressure
await send_chan.send(value) await send_chan.send(value)
print(f"FINISHED ITERATING {portal.channel.uid}") uid: tuple[str, str] = portal.chan.uid
print(f'FINISHED ITERATING {uid}')
# spawn 2 trio tasks to collect streams and push to a local queue # spawn 2 trio tasks to collect streams and push to a local
# queue
n: trio.Nursery n: trio.Nursery
async with trio.open_nursery() as n: async with trio.open_nursery() as n:
@ -62,8 +85,9 @@ async def aggregate(seed: int):
# close this local task's reference to send side # close this local task's reference to send side
await send_chan.aclose() await send_chan.aclose()
unique_vals = set() unique_vals: set[int] = set()
async with recv_chan: async with recv_chan:
value: int
async for value in recv_chan: async for value in recv_chan:
if value not in unique_vals: if value not in unique_vals:
unique_vals.add(value) unique_vals.add(value)
@ -72,11 +96,11 @@ async def aggregate(seed: int):
assert value in unique_vals assert value in unique_vals
print("FINISHED ITERATING in aggregator") print('FINISHED ITERATING in aggregator')
await an.cancel() await an.cancel()
print("WAITING on `ActorNursery` to finish") print('WAITING on `ActorNursery` to finish')
print("AGGREGATOR COMPLETE!") print('AGGREGATOR COMPLETE!')
async def main() -> list[int]: async def main() -> list[int]:
@ -95,8 +119,8 @@ async def main() -> list[int]:
# debug_mode=True, # debug_mode=True,
) as an: ) as an:
seed = int(1e3) seed: int = int(1e3)
pre_start = time.time() pre_start: float = time.time()
portal: Portal = await an.start_actor( portal: Portal = await an.start_actor(
name='aggregator', name='aggregator',
@ -109,23 +133,27 @@ async def main() -> list[int]:
seed=seed, seed=seed,
) as stream: ) as stream:
start = time.time() start: float = time.time()
# the portal call returns exactly what you'd expect # the portal call returns exactly what you'd expect
# as if the remote "aggregate" function was called locally # as if the remote "aggregate" function was called
# locally
result_stream: list[int] = [] result_stream: list[int] = []
value: int
async for value in stream: async for value in stream:
result_stream.append(value) result_stream.append(value)
cancelled: bool = await portal.cancel_actor() cancelled: bool = await portal.cancel_actor()
assert cancelled assert cancelled
stream_time: float = time.time() - start
total_time: float = time.time() - pre_start
print( print(
f"STREAM TIME = {time.time() - start}\n" f'STREAM TIME = {stream_time}\n'
f"STREAM + SPAWN TIME = {time.time() - pre_start}\n" f'STREAM + SPAWN TIME = {total_time}\n'
) )
assert result_stream == list(range(seed)) assert result_stream == list(range(seed))
return result_stream return result_stream
if __name__ == '__main__': if __name__ == '__main__':
final_stream = trio.run(main) final_stream: list[int] = trio.run(main)

35
examples/infected_asyncio_echo_server.py
View File

@ -13,7 +13,10 @@ import tractor
async def aio_echo_server( async def aio_echo_server(
chan: tractor.to_asyncio.LinkedTaskChannel, chan: tractor.to_asyncio.LinkedTaskChannel,
) -> None: ) -> None:
'''
Echo messages received through an asyncio task channel.
'''
# a first message must be sent **from** this ``asyncio`` # a first message must be sent **from** this ``asyncio``
# task or the ``trio`` side will never unblock from # task or the ``trio`` side will never unblock from
# ``tractor.to_asyncio.open_channel_from():`` # ``tractor.to_asyncio.open_channel_from():``
@ -29,8 +32,14 @@ async def aio_echo_server(
async def trio_to_aio_echo_server( async def trio_to_aio_echo_server(
ctx: tractor.Context, ctx: tractor.Context,
) -> None: ) -> None:
'''
Bridge an actor stream to the asyncio echo server.
'''
# this will block until the ``asyncio`` task sends a "first" # this will block until the ``asyncio`` task sends a "first"
# message. # message.
chan: tractor.to_asyncio.LinkedTaskChannel
first: str
async with tractor.to_asyncio.open_channel_from( async with tractor.to_asyncio.open_channel_from(
aio_echo_server, aio_echo_server,
) as (chan, first): ) as (chan, first):
@ -38,39 +47,49 @@ async def trio_to_aio_echo_server(
assert first == 'start' assert first == 'start'
await ctx.started(first) await ctx.started(first)
stream: tractor.MsgStream
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
msg: int
async for msg in stream: async for msg in stream:
await chan.send(msg) await chan.send(msg)
out = await chan.receive() out: int = await chan.receive()
# echo back to parent actor-task # echo back to parent actor-task
await stream.send(out) await stream.send(out)
async def main() -> None: async def main() -> None:
'''
Run the infected asyncio echo-server example.
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
p: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'aio_server', 'aio_server',
enable_modules=[__name__], enable_modules=[__name__],
infect_asyncio=True, infect_asyncio=True,
) )
async with p.open_context( ctx: tractor.Context
first: str
async with portal.open_context(
trio_to_aio_echo_server, trio_to_aio_echo_server,
) as (ctx, first): ) as (ctx, first):
assert first == 'start' assert first == 'start'
count = 0 count: int = 0
stream: tractor.MsgStream
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
delays = [] delays: list[float] = []
send = time.time() send: float = time.time()
await stream.send(count) await stream.send(count)
msg: int
async for msg in stream: async for msg in stream:
recv = time.time() recv: float = time.time()
delays.append(recv - send) delays.append(recv - send)
assert msg == count assert msg == count
count += 1 count += 1
@ -81,7 +100,7 @@ async def main() -> None:
break break
print(f'mean round trip rate (Hz): {1/mean(delays)}') print(f'mean round trip rate (Hz): {1/mean(delays)}')
await p.cancel_actor() await portal.cancel_actor()
if __name__ == '__main__': if __name__ == '__main__':

28
examples/integration/mpi4py/inherit_parent_main.py
View File

@ -1,9 +1,10 @@
""" '''
Integration test: spawning tractor actors from an MPI process. Integration test: spawning tractor actors from an MPI process.
When a parent is launched via ``mpirun``, Open MPI sets ``OMPI_*`` env When a parent is launched via ``mpirun``, Open MPI sets
vars that bind ``MPI_Init`` to the ``orted`` daemon. Tractor children ``OMPI_*`` env vars that bind ``MPI_Init`` to the ``orted``
inherit those env vars, so if ``inherit_parent_main=True`` (the default) daemon. Tractor children inherit those env vars, so if
``inherit_parent_main=True`` (the default)
the child re-executes ``__main__``, re-imports ``mpi4py``, and the child re-executes ``__main__``, re-imports ``mpi4py``, and
``MPI_Init_thread`` fails because the child was never spawned by ``MPI_Init_thread`` fails because the child was never spawned by
``orted``:: ``orted``::
@ -12,13 +13,15 @@ the child re-executes ``__main__``, re-imports ``mpi4py``, and
--> Returned value No permission (-17) instead of ORTE_SUCCESS --> Returned value No permission (-17) instead of ORTE_SUCCESS
Passing ``inherit_parent_main=False`` and placing RPC functions in a Passing ``inherit_parent_main=False`` and placing RPC functions in a
separate importable module (``_child``) avoids the re-import entirely. separate importable module (``_child``) avoids the re-import
entirely.
Usage:: Usage::
mpirun --allow-run-as-root -np 1 python -m \ mpirun --allow-run-as-root -np 1 python -m \
examples.integration.mpi4py.inherit_parent_main examples.integration.mpi4py.inherit_parent_main
"""
'''
from mpi4py import MPI from mpi4py import MPI
@ -30,8 +33,13 @@ from ._child import child_fn
async def main() -> None: async def main() -> None:
rank = MPI.COMM_WORLD.Get_rank() '''
print(f"[parent] rank={rank} pid={os.getpid()}", flush=True) Spawn an MPI-safe child without replaying the parent main.
'''
rank: int = MPI.COMM_WORLD.Get_rank()
pid: int = os.getpid()
print(f'[parent] rank={rank} pid={pid}', flush=True)
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery(start_method='trio') as an: async with tractor.open_nursery(start_method='trio') as an:
@ -43,9 +51,9 @@ async def main() -> None:
inherit_parent_main=False, inherit_parent_main=False,
) )
result: str = await portal.run(child_fn) result: str = await portal.run(child_fn)
print(f"[parent] got: {result}", flush=True) print(f'[parent] got: {result}', flush=True)
await portal.cancel_actor() await portal.cancel_actor()
if __name__ == "__main__": if __name__ == '__main__':
trio.run(main) trio.run(main)

42
examples/multiple_streams_one_portal.py
View File

@ -1,3 +1,5 @@
from typing import AsyncIterator
import trio import trio
import tractor import tractor
@ -5,20 +7,30 @@ import tractor
log = tractor.log.get_logger('multiportal') log = tractor.log.get_logger('multiportal')
async def stream_data(seed: int = 10): async def stream_data(seed: int = 10) -> AsyncIterator[int]:
log.info("Starting stream task") '''
Stream a finite sequence of integers.
'''
log.info('Starting stream task')
i: int
for i in range(seed): for i in range(seed):
yield i yield i
await trio.sleep(0) # trigger scheduler await trio.sleep(0) # trigger scheduler
async def stream_from_portal( async def stream_from_portal(
p: tractor.Portal, portal: tractor.Portal,
consumed: list, consumed: list[int],
) -> None: ) -> None:
'''
Consume one stream and toggle each value in a shared list.
async with p.open_stream_from(stream_data) as stream: '''
stream: tractor.MsgStream
async with portal.open_stream_from(stream_data) as stream:
item: int
async for item in stream: async for item in stream:
if item in consumed: if item in consumed:
consumed.remove(item) consumed.remove(item)
@ -27,24 +39,32 @@ async def stream_from_portal(
async def main() -> None: async def main() -> None:
'''
Consume two concurrent streams through one portal.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery(loglevel='info') as an: async with tractor.open_nursery(loglevel='info') as an:
p: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'stream_boi', 'stream_boi',
enable_modules=[__name__], enable_modules=[__name__],
) )
consumed: list = [] consumed: list[int] = []
n: trio.Nursery n: trio.Nursery
async with trio.open_nursery() as n: async with trio.open_nursery() as n:
for i in range(2): for _ in range(2):
n.start_soon(stream_from_portal, p, consumed) n.start_soon(
stream_from_portal,
portal,
consumed,
)
# both streaming consumer tasks have completed and so we should # both streaming consumer tasks have completed and so we
# have nothing in our list thanks to single threadedness # should have nothing in our list thanks to single
# threadedness
assert not consumed assert not consumed
await an.cancel() await an.cancel()

18
examples/nested_actor_tree.py
View File

@ -53,15 +53,21 @@ async def fan_out_squares(
) )
# unblock the parent's `.open_context()` entry and # unblock the parent's `.open_context()` entry and
# report which leaves came up. # report which leaves came up.
await ctx.started( leaf_names: list[str] = [
[p.chan.aid.name for p in portals] portal.chan.aid.name
) for portal in portals
]
await ctx.started(leaf_names)
squares: dict[int, int] = {} squares: dict[int, int] = {}
async def run_in_leaf( async def run_in_leaf(
portal: tractor.Portal, portal: tractor.Portal,
x: int, x: int,
) -> None: ) -> None:
'''
Run one square calculation in a leaf actor.
'''
squares[x] = await portal.run( squares[x] = await portal.run(
compute_square, compute_square,
x=x, x=x,
@ -85,12 +91,18 @@ async def fan_out_squares(
async def main() -> None: async def main() -> None:
'''
Run the nested actor-tree example.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'supervisor', 'supervisor',
enable_modules=[__name__], enable_modules=[__name__],
) )
ctx: tractor.Context
leaf_names: list[str]
async with portal.open_context( async with portal.open_context(
fan_out_squares, fan_out_squares,
vals=[1, 2, 3, 4], vals=[1, 2, 3, 4],

90
examples/parallelism/concurrent_actors_primes.py
View File

@ -1,18 +1,23 @@
""" '''
Demonstration of the prime number detector example from the Demonstration of the prime number detector example from the
``concurrent.futures`` docs: ``concurrent.futures`` docs:
https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor-example https://docs.python.org/3/library/concurrent.futures.html\
#processpoolexecutor-example
This uses no extra threads, fancy semaphores or futures; all we need This uses no extra threads, fancy semaphores or futures; all we need
is ``tractor``'s channels. is ``tractor``'s channels.
""" '''
from contextlib import ( from contextlib import (
asynccontextmanager as acm, asynccontextmanager as acm,
aclosing, aclosing,
) )
from typing import Callable from typing import (
AsyncIterator,
Awaitable,
Callable,
)
import itertools import itertools
import math import math
import time import time
@ -21,7 +26,12 @@ import tractor
import trio import trio
PRIMES = [ type ActorMap = Callable[
[Callable[[int], Awaitable[bool]], list[int]],
AsyncIterator[tuple[int, bool]],
]
PRIMES: list[int] = [
112272535095293, 112272535095293,
112582705942171, 112582705942171,
112272535095293, 112272535095293,
@ -32,6 +42,10 @@ PRIMES = [
async def is_prime(n: int) -> bool: async def is_prime(n: int) -> bool:
'''
Return whether ``n`` is prime.
'''
if n < 2: if n < 2:
return False return False
if n == 2: if n == 2:
@ -47,23 +61,32 @@ async def is_prime(n: int) -> bool:
@acm @acm
async def worker_pool(workers: int = 4): async def worker_pool(
"""Though it's a trivial special case for ``tractor``, the well workers: int = 4,
) -> AsyncIterator[ActorMap]:
'''
Though it's a trivial special case for ``tractor``, the well
known "worker pool" seems to be the defacto "but, I want this known "worker pool" seems to be the defacto "but, I want this
process pattern!" for most parallelism pilgrims. process pattern!" for most parallelism pilgrims.
Yes, the workers stay alive (and ready for work) until you close Yes, the workers stay alive (and ready for work) until you close
the context. the context.
"""
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portals: list[tractor.Portal] = [] portals: list[tractor.Portal] = []
snd_chan: trio.MemorySendChannel[tuple[int, bool]]
recv_chan: trio.MemoryReceiveChannel[tuple[int, bool]]
snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES)) snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES))
i: int
for i in range(workers): for i in range(workers):
# this starts a new sub-actor (process + trio runtime) and # this starts a new sub-actor (process + trio
# stores it's "portal" for later use to "submit jobs" (ugh). # runtime) and stores it's "portal" for later use to
# "submit jobs" (ugh).
portals.append( portals.append(
await an.start_actor( await an.start_actor(
f'worker_{i}', f'worker_{i}',
@ -72,22 +95,36 @@ async def worker_pool(workers: int = 4):
) )
async def _map( async def _map(
worker_func: Callable[[int], bool], worker_func: Callable[[int], Awaitable[bool]],
sequence: list[int] sequence: list[int],
) -> list[bool]: ) -> AsyncIterator[tuple[int, bool]]:
'''
Dispatch values across workers and yield their results.
# define an async (local) task to collect results from workers '''
# define an async (local) task to collect results from
# workers
async def send_result( async def send_result(
func: Callable, func: Callable[[int], Awaitable[bool]],
value: int, value: int,
portal: tractor.Portal, portal: tractor.Portal,
): ) -> None:
await snd_chan.send((value, await portal.run(func, n=value))) '''
Run one remote worker call and send its result.
'''
result: bool = await portal.run(func, n=value)
await snd_chan.send((value, result))
tn: trio.Nursery tn: trio.Nursery
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
for value, portal in zip(sequence, itertools.cycle(portals)): value: int
portal: tractor.Portal
for value, portal in zip(
sequence,
itertools.cycle(portals),
):
tn.start_soon( tn.start_soon(
send_result, send_result,
worker_func, worker_func,
@ -107,20 +144,29 @@ async def worker_pool(workers: int = 4):
async def main() -> None: async def main() -> None:
'''
Report primality results from a pool of actors.
'''
actor_map: ActorMap
async with worker_pool() as actor_map: async with worker_pool() as actor_map:
start = time.time() start: float = time.time()
results: AsyncIterator[tuple[int, bool]]
async with aclosing(actor_map(is_prime, PRIMES)) as results: async with aclosing(actor_map(is_prime, PRIMES)) as results:
number: int
prime: bool
async for number, prime in results: async for number, prime in results:
print(f'{number} is prime: {prime}') print(f'{number} is prime: {prime}')
print(f'processing took {time.time() - start} seconds') elapsed: float = time.time() - start
print(f'processing took {elapsed} seconds')
if __name__ == '__main__': if __name__ == '__main__':
start = time.time() start: float = time.time()
trio.run(main) trio.run(main)
print(f'script took {time.time() - start} seconds') elapsed: float = time.time() - start
print(f'script took {elapsed} seconds')

14
examples/parallelism/single_func.py
View File

@ -1,11 +1,11 @@
""" '''
Run with a process monitor from a terminal using:: Run with a process monitor from a terminal using::
$TERM -e watch -n 0.1 "pstree -a $$" \ $TERM -e watch -n 0.1 "pstree -a $$" \
& python examples/parallelism/single_func.py \ & python examples/parallelism/single_func.py \
&& kill $! && kill $!
""" '''
import os import os
import tractor import tractor
@ -13,8 +13,11 @@ import trio
async def burn_cpu() -> int: async def burn_cpu() -> int:
'''
Burn CPU briefly and return the current process ID.
pid = os.getpid() '''
pid: int = os.getpid()
# burn a core @ ~ 50kHz # burn a core @ ~ 50kHz
for _ in range(50000): for _ in range(50000):
@ -24,7 +27,10 @@ async def burn_cpu() -> int:
async def main() -> None: async def main() -> None:
'''
Run ``burn_cpu()`` in the parent and a subactor.
'''
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
# burn rubber in the parent too # burn rubber in the parent too
@ -34,7 +40,7 @@ async def main() -> None:
# block on and collect its PID as the caller-side result # block on and collect its PID as the caller-side result
pid: int = await tractor.to_actor.run(burn_cpu) pid: int = await tractor.to_actor.run(burn_cpu)
print(f"Collected subproc {pid}") print(f'Collected subproc {pid}')
if __name__ == '__main__': if __name__ == '__main__':

11
examples/quick_cluster.py
View File

@ -1,10 +1,13 @@
import trio import trio
import tractor import tractor
async def sleepy_jane() -> None: async def sleepy_jane() -> None:
uid: tuple = tractor.current_actor().uid '''
Identify the current actor and sleep forever.
'''
uid: tuple[str, str] = tractor.current_actor().uid
print(f'Yo i am actor {uid}') print(f'Yo i am actor {uid}')
await trio.sleep_forever() await trio.sleep_forever()
@ -28,7 +31,9 @@ async def main() -> None:
trio.open_nursery() as tn, trio.open_nursery() as tn,
): ):
for (name, portal) in portal_map.items(): name: str
portal: tractor.Portal
for name, portal in portal_map.items():
tn.start_soon( tn.start_soon(
portal.run, portal.run,
sleepy_jane, sleepy_jane,

18
examples/remote_error_propagation.py
View File

@ -3,17 +3,29 @@ import tractor
async def assert_err() -> None: async def assert_err() -> None:
'''
Raise an assertion error in the current actor.
'''
assert 0 assert 0
async def main() -> None: async def main() -> None:
'''
Propagate a failing one-shot task from a subactor.
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
real_actors: list[tractor.Portal] = [] real_actors: list[tractor.Portal] = []
i: int
for i in range(3): for i in range(3):
real_actors.append(await an.start_actor( real_actors.append(
await an.start_actor(
f'actor_{i}', f'actor_{i}',
enable_modules=[__name__], enable_modules=[__name__],
)) )
)
# run one one-shot task actor that will fail immediately; # run one one-shot task actor that will fail immediately;
# its error raises right here in the caller's task.. # its error raises right here in the caller's task..
@ -28,4 +40,4 @@ if __name__ == '__main__':
# also raises # also raises
trio.run(main) trio.run(main)
except tractor.RemoteActorError: except tractor.RemoteActorError:
print("Look Maa that actor failed hard, hehhh!") print('Look Maa that actor failed hard, hehhh!')

17
examples/rpc_bidir_streaming.py
View File

@ -9,16 +9,19 @@ async def simple_rpc(
data: int, data: int,
) -> None: ) -> None:
'''Test a small ping-pong 2-way streaming server. '''
Test a small ping-pong 2-way streaming server.
''' '''
# signal to parent that we're up much like # signal to parent that we're up much like
# ``trio.TaskStatus.started()`` # ``trio.TaskStatus.started()``
await ctx.started(data + 1) await ctx.started(data + 1)
stream: tractor.MsgStream
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
count = 0 count: int = 0
msg: str
async for msg in stream: async for msg in stream:
assert msg == 'ping' assert msg == 'ping'
@ -30,7 +33,11 @@ async def simple_rpc(
async def main() -> None: async def main() -> None:
'''
Exercise bidirectional streaming with a remote actor.
'''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
@ -39,6 +46,9 @@ async def main() -> None:
) )
# XXX: syntax requires py3.9 # XXX: syntax requires py3.9
ctx: tractor.Context
sent: int
stream: tractor.MsgStream
async with ( async with (
portal.open_context( portal.open_context(
@ -52,10 +62,11 @@ async def main() -> None:
assert sent == 11 assert sent == 11
count = 0 count: int = 0
# receive msgs using async for style # receive msgs using async for style
await stream.send('ping') await stream.send('ping')
msg: str
async for msg in stream: async for msg in stream:
assert msg == 'pong' assert msg == 'pong'
await stream.send('ping') await stream.send('ping')

10
examples/service_daemon_discovery.py
View File

@ -35,11 +35,13 @@ async def client_task() -> None:
''' '''
# a lookup miss yields `None` (not an error). # a lookup miss yields `None` (not an error).
async with tractor.find_actor('no_such_svc') as portal: maybe_portal: tractor.Portal|None
assert portal is None async with tractor.find_actor('no_such_svc') as maybe_portal:
assert maybe_portal is None
print('client: "no_such_svc" is not registered') print('client: "no_such_svc" is not registered')
# block until the service shows up in the registry, # block until the service shows up in the registry,
# then call into it through the delivered portal. # then call into it through the delivered portal.
portal: tractor.Portal
async with tractor.wait_for_actor('quote_svc') as portal: async with tractor.wait_for_actor('quote_svc') as portal:
quote: float = await portal.run( quote: float = await portal.run(
get_quote, get_quote,
@ -49,6 +51,10 @@ async def client_task() -> None:
async def main() -> None: async def main() -> None:
'''
Run a discoverable quote service and its client.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(

20
examples/service_discovery.py
View File

@ -1,22 +1,28 @@
import trio import trio
import tractor import tractor
tractor.log.get_console_log("INFO") tractor.log.get_console_log('INFO')
async def main(service_name: str) -> None: async def main(service_name: str) -> None:
'''
Discover one actor and inspect its registrar connection.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
await an.start_actor(service_name) await an.start_actor(service_name)
portal: tractor.Portal async with tractor.get_registry() as reg_portal:
async with tractor.get_registry() as portal: print(
print(f"Registrar is listening on {portal.channel}") f'Registrar is listening on {reg_portal.channel}'
)
sockaddr: tractor.Portal actor_portal: tractor.Portal
async with tractor.wait_for_actor(service_name) as sockaddr: async with tractor.wait_for_actor(
print(f"my_service is found at {sockaddr}") service_name,
) as actor_portal:
print(f'my_service is found at {actor_portal}')
await an.cancel() await an.cancel()

13
examples/streaming_broadcast_fanout.py
View File

@ -28,6 +28,7 @@ async def tick_stream(
# wait for the go-signal ensuring every parent-side # wait for the go-signal ensuring every parent-side
# subscriber is attached before any tick is sent. # subscriber is attached before any tick is sent.
assert await stream.receive() == 'go' assert await stream.receive() == 'go'
i: int
for i in range(count): for i in range(count):
await stream.send(i) await stream.send(i)
# falling out gracefully closes our stream side; # falling out gracefully closes our stream side;
@ -37,15 +38,17 @@ async def tick_stream(
async def consume( async def consume(
name: str, name: str,
stream: tractor.MsgStream, stream: tractor.MsgStream,
task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, task_status: trio.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
) -> None: ) -> None:
''' '''
Consume a private broadcast-copy of the IPC stream. Consume a private broadcast-copy of the IPC stream.
''' '''
bcaster: tractor.trionics.BroadcastReceiver
async with stream.subscribe() as bcaster: async with stream.subscribe() as bcaster:
task_status.started() task_status.started()
ticks: list[int] = [] ticks: list[int] = []
tick: int
async for tick in bcaster: async for tick in bcaster:
print(f'{name}: rx {tick}') print(f'{name}: rx {tick}')
ticks.append(tick) ticks.append(tick)
@ -54,12 +57,19 @@ async def consume(
async def main() -> None: async def main() -> None:
'''
Fan one remote stream out to local subscribers.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'ticker', 'ticker',
enable_modules=[__name__], enable_modules=[__name__],
) )
ctx: tractor.Context
first: int
stream: tractor.MsgStream
async with ( async with (
portal.open_context( portal.open_context(
tick_stream, tick_stream,
@ -72,6 +82,7 @@ async def main() -> None:
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
# use `.start()` so each consumer is known # use `.start()` so each consumer is known
# to be subscribed before the ticks flow. # to be subscribed before the ticks flow.
i: int
for i in range(3): for i in range(3):
await tn.start( await tn.start(
consume, consume,

27
examples/trio/lockacquire_not_unmasked.py
View File

@ -1,3 +1,4 @@
from collections.abc import AsyncIterator
from contextlib import ( from contextlib import (
asynccontextmanager as acm, asynccontextmanager as acm,
) )
@ -16,7 +17,11 @@ _lock: trio.Lock|None = None
@acm @acm
async def acquire_singleton_lock( async def acquire_singleton_lock(
) -> None: ) -> AsyncIterator[trio.Lock]:
'''
Acquire and yield the process-wide lock.
'''
global _lock global _lock
if _lock is None: if _lock is None:
log.info('Allocating LOCK') log.info('Allocating LOCK')
@ -32,8 +37,15 @@ async def acquire_singleton_lock(
async def hold_lock_forever( async def hold_lock_forever(
task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, task_status: trio.TaskStatus[
trio.Lock,
] = trio.TASK_STATUS_IGNORED,
) -> None: ) -> None:
'''
Hold the singleton lock until cancellation.
'''
lock: trio.Lock
async with ( async with (
tractor.trionics.maybe_raise_from_masking_exc(), tractor.trionics.maybe_raise_from_masking_exc(),
acquire_singleton_lock() as lock, acquire_singleton_lock() as lock,
@ -47,6 +59,11 @@ async def main(
loglevel: str = 'info', loglevel: str = 'info',
debug_mode: bool = True, debug_mode: bool = True,
) -> None: ) -> None:
'''
Exercise lock acquisition while cancellation is masked.
'''
tn: trio.Nursery
async with ( async with (
trio.open_nursery() as tn, trio.open_nursery() as tn,
@ -58,7 +75,7 @@ async def main(
from tractor.trionics import _taskc from tractor.trionics import _taskc
_taskc._mask_cases.clear() _taskc._mask_cases.clear()
_lock = await tn.start( _held_lock: trio.Lock = await tn.start(
hold_lock_forever, hold_lock_forever,
) )
with trio.move_on_after(0.2): with trio.move_on_after(0.2):
@ -74,8 +91,8 @@ if __name__ == '__main__':
tractor.log.get_console_log(level='info') tractor.log.get_console_log(level='info')
for case in [True, False]: for case in [True, False]:
log.info( log.info(
f'\n' '\n'
f'------ RUNNING SCRIPT TRIAL ------\n' '------ RUNNING SCRIPT TRIAL ------\n'
f'ignore_special_cases: {case!r}\n' f'ignore_special_cases: {case!r}\n'
) )
trio.run(partial( trio.run(partial(

42
examples/trio/send_chan_aclose_masks_beg.py
View File

@ -1,3 +1,4 @@
from collections.abc import Iterator
from contextlib import ( from contextlib import (
contextmanager as cm, contextmanager as cm,
# TODO, any diff in async case(s)?? # TODO, any diff in async case(s)??
@ -17,7 +18,7 @@ log = tractor.log.get_logger(
@cm @cm
def teardown_on_exc( def teardown_on_exc(
raise_from_handler: bool = False, raise_from_handler: bool = False,
): ) -> Iterator[None]:
''' '''
You could also have a teardown handler which catches any exc and You could also have a teardown handler which catches any exc and
does some required teardown. In this case the problem is does some required teardown. In this case the problem is
@ -30,7 +31,7 @@ def teardown_on_exc(
except BaseException as _berr: except BaseException as _berr:
berr = _berr berr = _berr
log.exception( log.exception(
f'Handling termination teardown in child due to,\n' 'Handling termination teardown in child due to,\n'
f'{berr!r}\n' f'{berr!r}\n'
) )
if raise_from_handler: if raise_from_handler:
@ -54,14 +55,18 @@ def teardown_on_exc(
async def finite_stream_to_rent( async def finite_stream_to_rent(
tx: trio.abc.SendChannel, tx: trio.abc.SendChannel[int],
child_errors_mid_stream: bool, child_errors_mid_stream: bool,
raise_unmasked: bool, raise_unmasked: bool,
task_status: trio.TaskStatus[ task_status: trio.TaskStatus[
trio.CancelScope, trio.CancelScope|None,
] = trio.TASK_STATUS_IGNORED, ] = trio.TASK_STATUS_IGNORED,
): ) -> None:
'''
Stream values while reproducing exception masking on close.
'''
async with ( async with (
# XXX without this unmasker the mid-streaming RTE is never # XXX without this unmasker the mid-streaming RTE is never
# reported since it is masked by the `tx.aclose()` # reported since it is masked by the `tx.aclose()`
@ -135,18 +140,25 @@ async def main(
raise_unmasked: bool = False, raise_unmasked: bool = False,
loglevel: str = 'info', loglevel: str = 'info',
) -> None: ) -> None:
'''
Reproduce cancellation masking a child-stream failure.
'''
tractor.log.get_console_log(level=loglevel) tractor.log.get_console_log(level=loglevel)
# the `.aclose()` being checkpoints on these # the `.aclose()` being checkpoints on these
# is the source of the problem.. # is the source of the problem..
tx: trio.MemorySendChannel[int]
rx: trio.MemoryReceiveChannel[int]
tx, rx = trio.open_memory_channel(1) tx, rx = trio.open_memory_channel(1)
tn: trio.Nursery
async with ( async with (
tractor.trionics.collapse_eg(), tractor.trionics.collapse_eg(),
trio.open_nursery() as tn, trio.open_nursery() as tn,
rx as rx, rx as rx,
): ):
_child_cs = await tn.start( _child_cs: trio.CancelScope|None = await tn.start(
partial( partial(
finite_stream_to_rent, finite_stream_to_rent,
child_errors_mid_stream=child_errors_mid_stream, child_errors_mid_stream=child_errors_mid_stream,
@ -154,6 +166,7 @@ async def main(
tx=tx, tx=tx,
) )
) )
msg: int
async for msg in rx: async for msg in rx:
log.debug( log.debug(
f'Rent rx {msg!r}\n' f'Rent rx {msg!r}\n'
@ -162,12 +175,13 @@ async def main(
# simulate some external cancellation # simulate some external cancellation
# request **JUST BEFORE** the child errors. # request **JUST BEFORE** the child errors.
if msg == 65: if msg == 65:
log.cancel( cancel_msg: str = (
f'Cancelling parent on,\n' 'Cancelling parent on,\n'
f'msg={msg}\n' 'msg={msg}\n'
f'\n' '\n'
f'Simulates OOB cancel request!\n' 'Simulates OOB cancel request!\n'
) ).format(msg=msg)
log.cancel(cancel_msg)
tn.cancel_scope.cancel() tn.cancel_scope.cancel()
@ -176,8 +190,8 @@ if __name__ == '__main__':
tractor.log.get_console_log(level='info') tractor.log.get_console_log(level='info')
for case in [True, False]: for case in [True, False]:
log.info( log.info(
f'\n' '\n'
f'------ RUNNING SCRIPT TRIAL ------\n' '------ RUNNING SCRIPT TRIAL ------\n'
f'child_errors_midstream: {case!r}\n' f'child_errors_midstream: {case!r}\n'
) )
try: try:

10
examples/typed_payloads.py
View File

@ -46,7 +46,9 @@ async def point_doubler(
# now do it right; the parent receives this as the 2nd # now do it right; the parent receives this as the 2nd
# element of its `.open_context()` entry tuple. # element of its `.open_context()` entry tuple.
await ctx.started(Point(x=0, y=0)) await ctx.started(Point(x=0, y=0))
stream: tractor.MsgStream
async with ctx.open_stream() as stream: async with ctx.open_stream() as stream:
pt: Point
async for pt in stream: async for pt in stream:
# natively decoded to our struct type! # natively decoded to our struct type!
assert type(pt) is Point assert type(pt) is Point
@ -59,12 +61,19 @@ async def point_doubler(
async def main() -> None: async def main() -> None:
'''
Exchange typed ``Point`` payloads with a subactor.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal: tractor.Portal = await an.start_actor(
'point_doubler', 'point_doubler',
enable_modules=[__name__], enable_modules=[__name__],
) )
ctx: tractor.Context
first: Point
stream: tractor.MsgStream
async with ( async with (
portal.open_context( portal.open_context(
point_doubler, point_doubler,
@ -73,6 +82,7 @@ async def main() -> None:
): ):
# the (validated) started-value from the child # the (validated) started-value from the child
assert first == Point(x=0, y=0) assert first == Point(x=0, y=0)
i: int
for i in range(3): for i in range(3):
await stream.send(Point(x=i, y=i)) await stream.send(Point(x=i, y=i))
doubled: Point = await stream.receive() doubled: Point = await stream.receive()

11
examples/uds_transport_actor_tree.py
View File

@ -21,13 +21,17 @@ async def report_addr() -> str:
Return this actor's own accept (bind) addr + pid. Return this actor's own accept (bind) addr + pid.
''' '''
actor = tractor.current_actor() actor: tractor.Actor = tractor.current_actor()
addr: tuple = actor.accept_addr addr: tuple[str, str] = actor.accept_addr
pid: int = os.getpid() pid: int = os.getpid()
return f'{actor.name}@{addr} pid={pid}' return f'{actor.name}@{addr} pid={pid}'
async def main() -> None: async def main() -> None:
'''
Run a child actor over the UDS transport.
'''
an: tractor.ActorNursery an: tractor.ActorNursery
async with tractor.open_nursery( async with tractor.open_nursery(
enable_transports=['uds'], enable_transports=['uds'],
@ -52,7 +56,8 @@ async def main() -> None:
) )
# ask the child for its OWN distinct bind addr: another # ask the child for its OWN distinct bind addr: another
# socket-file path under the runtime dir. # socket-file path under the runtime dir.
print(f'child says: {await portal.run(report_addr)}') child_report: str = await portal.run(report_addr)
print(f'child says: {child_report}')
await portal.cancel_actor() await portal.cancel_actor()