From 96a18382100ca63cca1baa84ba8ed8102309c488 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 18:58:43 -0400 Subject: [PATCH] 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`)) --- examples/a_trynamic_first_scene.py | 30 +++++-- examples/actor_spawning_and_causality.py | 12 ++- ...ctor_spawning_and_causality_with_daemon.py | 18 ++-- examples/asynchronous_generators.py | 27 ++++-- examples/full_fledged_streaming_service.py | 62 +++++++++---- examples/infected_asyncio_echo_server.py | 35 ++++++-- .../integration/mpi4py/inherit_parent_main.py | 28 +++--- examples/multiple_streams_one_portal.py | 42 ++++++--- examples/nested_actor_tree.py | 18 +++- .../parallelism/concurrent_actors_primes.py | 90 ++++++++++++++----- examples/parallelism/single_func.py | 16 ++-- examples/quick_cluster.py | 11 ++- examples/remote_error_propagation.py | 22 +++-- examples/rpc_bidir_streaming.py | 17 +++- examples/service_daemon_discovery.py | 10 ++- examples/service_discovery.py | 20 +++-- examples/streaming_broadcast_fanout.py | 13 ++- examples/trio/lockacquire_not_unmasked.py | 27 ++++-- examples/trio/send_chan_aclose_masks_beg.py | 42 ++++++--- examples/typed_payloads.py | 10 +++ examples/uds_transport_actor_tree.py | 11 ++- 21 files changed, 418 insertions(+), 143 deletions(-) diff --git a/examples/a_trynamic_first_scene.py b/examples/a_trynamic_first_scene.py index 0606e5f0..ba6d12b7 100644 --- a/examples/a_trynamic_first_scene.py +++ b/examples/a_trynamic_first_scene.py @@ -1,29 +1,40 @@ import trio import tractor -_this_module = __name__ -the_line = 'Hi my name is {}' +_this_module: str = __name__ +the_line: str = 'Hi my name is {}' -tractor.log.get_console_log("INFO") +tractor.log.get_console_log('INFO') async def hi() -> str: + ''' + Return a greeting naming the current actor. + + ''' return the_line.format(tractor.current_actor().name) async def say_hello(other_actor: str) -> str: + ''' + Ask another actor to return its greeting. + + ''' portal: tractor.Portal async with tractor.wait_for_actor(other_actor) as portal: return await portal.run(hi) 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"). - """ + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - print("Alright... Action!") + print('Alright... Action!') # both actors wait on (then dial!) the *other*, so each # must outlive both hellos: spawn as daemons, run the @@ -40,6 +51,10 @@ async def main() -> None: name: str, other_actor: str, ) -> None: + ''' + Print a greeting fetched through a named actor. + + ''' print( # RPC through an existing actor's `Portal`. await portals[name].run( @@ -48,13 +63,14 @@ async def main() -> None: ) ) + tn: trio.Nursery async with trio.open_nursery() as tn: tn.start_soon(run_and_print, 'donny', 'gretchen') tn.start_soon(run_and_print, 'gretchen', 'donny') 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__': diff --git a/examples/actor_spawning_and_causality.py b/examples/actor_spawning_and_causality.py index e5a5790f..8b3dd0ac 100644 --- a/examples/actor_spawning_and_causality.py +++ b/examples/actor_spawning_and_causality.py @@ -3,13 +3,19 @@ import tractor async def cellar_door() -> str: + ''' + Return a phrase from a spawned actor. + + ''' assert not tractor.is_root_process() - return "Dang that's beautiful" + return 'Dang that\'s beautiful' async def main() -> None: - """The main ``tractor`` routine. - """ + ''' + The main ``tractor`` routine. + + ''' # spawn a subactor, run ``cellar_door()`` as its lone task, # block until its result arrives and the subactor is reaped. print( diff --git a/examples/actor_spawning_and_causality_with_daemon.py b/examples/actor_spawning_and_causality_with_daemon.py index c75ab57f..166f08b0 100644 --- a/examples/actor_spawning_and_causality_with_daemon.py +++ b/examples/actor_spawning_and_causality_with_daemon.py @@ -3,15 +3,20 @@ import tractor 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. - """ + + ''' return 'have you ever seen a portal?' async def main() -> None: - """The main ``tractor`` routine. - """ + ''' + The main ``tractor`` routine. + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -24,9 +29,8 @@ async def main() -> None: # call the subactor a 2nd time print(await portal.run(movie_theatre_question)) - # the async with will block here indefinitely waiting - # for our actor "frank" to complete, but since it's an - # "outlive_main" actor it will never end until cancelled + # the async with will wait indefinitely for "frank" because + # its runtime remains active until explicitly cancelled await portal.cancel_actor() diff --git a/examples/asynchronous_generators.py b/examples/asynchronous_generators.py index 0202c98b..037230ff 100644 --- a/examples/asynchronous_generators.py +++ b/examples/asynchronous_generators.py @@ -1,20 +1,31 @@ -from typing import AsyncIterator from itertools import repeat +from typing import AsyncIterator import trio 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 - yield i + ''' + message: str + 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) async def main() -> None: + ''' + Print messages streamed from a subactor. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -24,10 +35,12 @@ async def main() -> None: # this async for loop streams values from the above # async generator running in a separate process + stream: tractor.MsgStream async with portal.open_stream_from(stream_forever) as stream: count: int = 0 - async for letter in stream: - print(letter) + message: str + async for message in stream: + print(message) count += 1 if count > 50: diff --git a/examples/full_fledged_streaming_service.py b/examples/full_fledged_streaming_service.py index de4d5286..87c777eb 100644 --- a/examples/full_fledged_streaming_service.py +++ b/examples/full_fledged_streaming_service.py @@ -1,4 +1,6 @@ import time +from typing import AsyncIterator + import trio import tractor from tractor import ( @@ -9,14 +11,19 @@ from tractor import ( # 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): yield i await trio.sleep(0.0001) # trigger scheduler # 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 a single set of values to the parent. @@ -25,6 +32,7 @@ async def aggregate(seed: int): an: ActorNursery async with tractor.open_nursery() as an: portals: list[Portal] = [] + i: int for i in range(1, 3): # fork/spawn call @@ -35,20 +43,35 @@ async def aggregate(seed: int): portals.append(portal) + send_chan: trio.MemorySendChannel[int] + recv_chan: trio.MemoryReceiveChannel[int] 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 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: # leverage trio's built-in backpressure 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 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 await send_chan.aclose() - unique_vals = set() + unique_vals: set[int] = set() async with recv_chan: + value: int async for value in recv_chan: if value not in unique_vals: unique_vals.add(value) @@ -72,11 +96,11 @@ async def aggregate(seed: int): assert value in unique_vals - print("FINISHED ITERATING in aggregator") + print('FINISHED ITERATING in aggregator') await an.cancel() - print("WAITING on `ActorNursery` to finish") - print("AGGREGATOR COMPLETE!") + print('WAITING on `ActorNursery` to finish') + print('AGGREGATOR COMPLETE!') async def main() -> list[int]: @@ -95,8 +119,8 @@ async def main() -> list[int]: # debug_mode=True, ) as an: - seed = int(1e3) - pre_start = time.time() + seed: int = int(1e3) + pre_start: float = time.time() portal: Portal = await an.start_actor( name='aggregator', @@ -109,23 +133,27 @@ async def main() -> list[int]: seed=seed, ) as stream: - start = time.time() + start: float = time.time() # 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] = [] + value: int async for value in stream: result_stream.append(value) cancelled: bool = await portal.cancel_actor() assert cancelled + stream_time: float = time.time() - start + total_time: float = time.time() - pre_start print( - f"STREAM TIME = {time.time() - start}\n" - f"STREAM + SPAWN TIME = {time.time() - pre_start}\n" + f'STREAM TIME = {stream_time}\n' + f'STREAM + SPAWN TIME = {total_time}\n' ) assert result_stream == list(range(seed)) return result_stream if __name__ == '__main__': - final_stream = trio.run(main) + final_stream: list[int] = trio.run(main) diff --git a/examples/infected_asyncio_echo_server.py b/examples/infected_asyncio_echo_server.py index e111459d..b570f522 100644 --- a/examples/infected_asyncio_echo_server.py +++ b/examples/infected_asyncio_echo_server.py @@ -13,7 +13,10 @@ import tractor async def aio_echo_server( chan: tractor.to_asyncio.LinkedTaskChannel, ) -> None: + ''' + Echo messages received through an asyncio task channel. + ''' # a first message must be sent **from** this ``asyncio`` # task or the ``trio`` side will never unblock from # ``tractor.to_asyncio.open_channel_from():`` @@ -29,8 +32,14 @@ async def aio_echo_server( async def trio_to_aio_echo_server( ctx: tractor.Context, ) -> None: + ''' + Bridge an actor stream to the asyncio echo server. + + ''' # this will block until the ``asyncio`` task sends a "first" # message. + chan: tractor.to_asyncio.LinkedTaskChannel + first: str async with tractor.to_asyncio.open_channel_from( aio_echo_server, ) as (chan, first): @@ -38,39 +47,49 @@ async def trio_to_aio_echo_server( assert first == 'start' await ctx.started(first) + stream: tractor.MsgStream async with ctx.open_stream() as stream: + msg: int async for msg in stream: await chan.send(msg) - out = await chan.receive() + out: int = await chan.receive() # echo back to parent actor-task await stream.send(out) async def main() -> None: + ''' + Run the infected asyncio echo-server example. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - p: tractor.Portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'aio_server', enable_modules=[__name__], infect_asyncio=True, ) - async with p.open_context( + ctx: tractor.Context + first: str + async with portal.open_context( trio_to_aio_echo_server, ) as (ctx, first): assert first == 'start' - count = 0 + count: int = 0 + stream: tractor.MsgStream async with ctx.open_stream() as stream: - delays = [] - send = time.time() + delays: list[float] = [] + send: float = time.time() await stream.send(count) + msg: int async for msg in stream: - recv = time.time() + recv: float = time.time() delays.append(recv - send) assert msg == count count += 1 @@ -81,7 +100,7 @@ async def main() -> None: break print(f'mean round trip rate (Hz): {1/mean(delays)}') - await p.cancel_actor() + await portal.cancel_actor() if __name__ == '__main__': diff --git a/examples/integration/mpi4py/inherit_parent_main.py b/examples/integration/mpi4py/inherit_parent_main.py index 741d8a6b..39d023aa 100644 --- a/examples/integration/mpi4py/inherit_parent_main.py +++ b/examples/integration/mpi4py/inherit_parent_main.py @@ -1,9 +1,10 @@ -""" +''' Integration test: spawning tractor actors from an MPI process. -When a parent is launched via ``mpirun``, Open MPI sets ``OMPI_*`` env -vars that bind ``MPI_Init`` to the ``orted`` daemon. Tractor children -inherit those env vars, so if ``inherit_parent_main=True`` (the default) +When a parent is launched via ``mpirun``, Open MPI sets +``OMPI_*`` env vars that bind ``MPI_Init`` to the ``orted`` +daemon. Tractor children inherit those env vars, so if +``inherit_parent_main=True`` (the default) the child re-executes ``__main__``, re-imports ``mpi4py``, and ``MPI_Init_thread`` fails because the child was never spawned by ``orted``:: @@ -12,13 +13,15 @@ the child re-executes ``__main__``, re-imports ``mpi4py``, and --> Returned value No permission (-17) instead of ORTE_SUCCESS 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:: mpirun --allow-run-as-root -np 1 python -m \ examples.integration.mpi4py.inherit_parent_main -""" + +''' from mpi4py import MPI @@ -30,8 +33,13 @@ from ._child import child_fn 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 async with tractor.open_nursery(start_method='trio') as an: @@ -43,9 +51,9 @@ async def main() -> None: inherit_parent_main=False, ) 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() -if __name__ == "__main__": +if __name__ == '__main__': trio.run(main) diff --git a/examples/multiple_streams_one_portal.py b/examples/multiple_streams_one_portal.py index 45272b6c..81a8a3bf 100644 --- a/examples/multiple_streams_one_portal.py +++ b/examples/multiple_streams_one_portal.py @@ -1,3 +1,5 @@ +from typing import AsyncIterator + import trio import tractor @@ -5,20 +7,30 @@ import tractor log = tractor.log.get_logger('multiportal') -async def stream_data(seed: int = 10): - log.info("Starting stream task") +async def stream_data(seed: int = 10) -> AsyncIterator[int]: + ''' + Stream a finite sequence of integers. + ''' + log.info('Starting stream task') + + i: int for i in range(seed): yield i await trio.sleep(0) # trigger scheduler async def stream_from_portal( - p: tractor.Portal, - consumed: list, + portal: tractor.Portal, + consumed: list[int], ) -> 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: if item in consumed: consumed.remove(item) @@ -27,24 +39,32 @@ async def stream_from_portal( async def main() -> None: + ''' + Consume two concurrent streams through one portal. + ''' an: tractor.ActorNursery 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', enable_modules=[__name__], ) - consumed: list = [] + consumed: list[int] = [] n: trio.Nursery async with trio.open_nursery() as n: - for i in range(2): - n.start_soon(stream_from_portal, p, consumed) + for _ in range(2): + n.start_soon( + stream_from_portal, + portal, + consumed, + ) - # both streaming consumer tasks have completed and so we should - # have nothing in our list thanks to single threadedness + # both streaming consumer tasks have completed and so we + # should have nothing in our list thanks to single + # threadedness assert not consumed await an.cancel() diff --git a/examples/nested_actor_tree.py b/examples/nested_actor_tree.py index cb45b22c..279e3a4d 100644 --- a/examples/nested_actor_tree.py +++ b/examples/nested_actor_tree.py @@ -53,15 +53,21 @@ async def fan_out_squares( ) # unblock the parent's `.open_context()` entry and # report which leaves came up. - await ctx.started( - [p.chan.aid.name for p in portals] - ) + leaf_names: list[str] = [ + portal.chan.aid.name + for portal in portals + ] + await ctx.started(leaf_names) squares: dict[int, int] = {} async def run_in_leaf( portal: tractor.Portal, x: int, ) -> None: + ''' + Run one square calculation in a leaf actor. + + ''' squares[x] = await portal.run( compute_square, x=x, @@ -85,12 +91,18 @@ async def fan_out_squares( async def main() -> None: + ''' + Run the nested actor-tree example. + + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( 'supervisor', enable_modules=[__name__], ) + ctx: tractor.Context + leaf_names: list[str] async with portal.open_context( fan_out_squares, vals=[1, 2, 3, 4], diff --git a/examples/parallelism/concurrent_actors_primes.py b/examples/parallelism/concurrent_actors_primes.py index e38adeee..9cd7b8e6 100644 --- a/examples/parallelism/concurrent_actors_primes.py +++ b/examples/parallelism/concurrent_actors_primes.py @@ -1,18 +1,23 @@ -""" +''' Demonstration of the prime number detector example from the ``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 is ``tractor``'s channels. -""" +''' from contextlib import ( asynccontextmanager as acm, aclosing, ) -from typing import Callable +from typing import ( + AsyncIterator, + Awaitable, + Callable, +) import itertools import math import time @@ -21,7 +26,12 @@ import tractor import trio -PRIMES = [ +type ActorMap = Callable[ + [Callable[[int], Awaitable[bool]], list[int]], + AsyncIterator[tuple[int, bool]], +] + +PRIMES: list[int] = [ 112272535095293, 112582705942171, 112272535095293, @@ -32,6 +42,10 @@ PRIMES = [ async def is_prime(n: int) -> bool: + ''' + Return whether ``n`` is prime. + + ''' if n < 2: return False if n == 2: @@ -47,23 +61,32 @@ async def is_prime(n: int) -> bool: @acm -async def worker_pool(workers: int = 4): - """Though it's a trivial special case for ``tractor``, the well +async def worker_pool( + 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 process pattern!" for most parallelism pilgrims. Yes, the workers stay alive (and ready for work) until you close the context. - """ + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: 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)) + i: int for i in range(workers): - # this starts a new sub-actor (process + trio runtime) and - # stores it's "portal" for later use to "submit jobs" (ugh). + # this starts a new sub-actor (process + trio + # runtime) and stores it's "portal" for later use to + # "submit jobs" (ugh). portals.append( await an.start_actor( f'worker_{i}', @@ -72,22 +95,36 @@ async def worker_pool(workers: int = 4): ) async def _map( - worker_func: Callable[[int], bool], - sequence: list[int] - ) -> list[bool]: + worker_func: Callable[[int], Awaitable[bool]], + sequence: list[int], + ) -> 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( - func: Callable, + func: Callable[[int], Awaitable[bool]], value: int, portal: tractor.Portal, - ): - await snd_chan.send((value, await portal.run(func, n=value))) + ) -> None: + ''' + 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 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( send_result, worker_func, @@ -107,20 +144,29 @@ async def worker_pool(workers: int = 4): async def main() -> None: + ''' + Report primality results from a pool of actors. + ''' + actor_map: ActorMap 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: + number: int + prime: bool async for number, prime in results: 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__': - start = time.time() + start: float = time.time() trio.run(main) - print(f'script took {time.time() - start} seconds') + elapsed: float = time.time() - start + print(f'script took {elapsed} seconds') diff --git a/examples/parallelism/single_func.py b/examples/parallelism/single_func.py index 9513505d..71458aa1 100644 --- a/examples/parallelism/single_func.py +++ b/examples/parallelism/single_func.py @@ -1,11 +1,11 @@ -""" +''' Run with a process monitor from a terminal using:: $TERM -e watch -n 0.1 "pstree -a $$" \ & python examples/parallelism/single_func.py \ && kill $! -""" +''' import os import tractor @@ -13,18 +13,24 @@ import trio 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 for _ in range(50000): - await trio.sleep(1/50000/50) + await trio.sleep(1 / 50000 / 50) return pid async def main() -> None: + ''' + Run ``burn_cpu()`` in the parent and a subactor. + ''' async with trio.open_nursery() as tn: # 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 pid: int = await tractor.to_actor.run(burn_cpu) - print(f"Collected subproc {pid}") + print(f'Collected subproc {pid}') if __name__ == '__main__': diff --git a/examples/quick_cluster.py b/examples/quick_cluster.py index 489912c1..be3597aa 100644 --- a/examples/quick_cluster.py +++ b/examples/quick_cluster.py @@ -1,10 +1,13 @@ - import trio import tractor 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}') await trio.sleep_forever() @@ -28,7 +31,9 @@ async def main() -> None: 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( portal.run, sleepy_jane, diff --git a/examples/remote_error_propagation.py b/examples/remote_error_propagation.py index 405c626b..fe399fda 100644 --- a/examples/remote_error_propagation.py +++ b/examples/remote_error_propagation.py @@ -3,17 +3,29 @@ import tractor async def assert_err() -> None: + ''' + Raise an assertion error in the current actor. + + ''' assert 0 async def main() -> None: + ''' + Propagate a failing one-shot task from a subactor. + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: real_actors: list[tractor.Portal] = [] + i: int for i in range(3): - real_actors.append(await an.start_actor( - f'actor_{i}', - enable_modules=[__name__], - )) + real_actors.append( + await an.start_actor( + f'actor_{i}', + enable_modules=[__name__], + ) + ) # run one one-shot task actor that will fail immediately; # its error raises right here in the caller's task.. @@ -28,4 +40,4 @@ if __name__ == '__main__': # also raises trio.run(main) except tractor.RemoteActorError: - print("Look Maa that actor failed hard, hehhh!") + print('Look Maa that actor failed hard, hehhh!') diff --git a/examples/rpc_bidir_streaming.py b/examples/rpc_bidir_streaming.py index 68c4f541..e71007bf 100644 --- a/examples/rpc_bidir_streaming.py +++ b/examples/rpc_bidir_streaming.py @@ -9,16 +9,19 @@ async def simple_rpc( data: int, ) -> 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 # ``trio.TaskStatus.started()`` await ctx.started(data + 1) + stream: tractor.MsgStream async with ctx.open_stream() as stream: - count = 0 + count: int = 0 + msg: str async for msg in stream: assert msg == 'ping' @@ -30,7 +33,11 @@ async def simple_rpc( async def main() -> None: + ''' + Exercise bidirectional streaming with a remote actor. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -39,6 +46,9 @@ async def main() -> None: ) # XXX: syntax requires py3.9 + ctx: tractor.Context + sent: int + stream: tractor.MsgStream async with ( portal.open_context( @@ -52,10 +62,11 @@ async def main() -> None: assert sent == 11 - count = 0 + count: int = 0 # receive msgs using async for style await stream.send('ping') + msg: str async for msg in stream: assert msg == 'pong' await stream.send('ping') diff --git a/examples/service_daemon_discovery.py b/examples/service_daemon_discovery.py index 0abd4987..5c5c6b03 100644 --- a/examples/service_daemon_discovery.py +++ b/examples/service_daemon_discovery.py @@ -35,11 +35,13 @@ async def client_task() -> None: ''' # a lookup miss yields `None` (not an error). - async with tractor.find_actor('no_such_svc') as portal: - assert portal is None + maybe_portal: tractor.Portal|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') # block until the service shows up in the registry, # then call into it through the delivered portal. + portal: tractor.Portal async with tractor.wait_for_actor('quote_svc') as portal: quote: float = await portal.run( get_quote, @@ -49,6 +51,10 @@ async def client_task() -> None: async def main() -> None: + ''' + Run a discoverable quote service and its client. + + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( diff --git a/examples/service_discovery.py b/examples/service_discovery.py index ef678951..3d30e800 100644 --- a/examples/service_discovery.py +++ b/examples/service_discovery.py @@ -1,22 +1,28 @@ import trio import tractor -tractor.log.get_console_log("INFO") +tractor.log.get_console_log('INFO') async def main(service_name: str) -> None: + ''' + Discover one actor and inspect its registrar connection. + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: await an.start_actor(service_name) - portal: tractor.Portal - async with tractor.get_registry() as portal: - print(f"Registrar is listening on {portal.channel}") + async with tractor.get_registry() as reg_portal: + print( + f'Registrar is listening on {reg_portal.channel}' + ) - sockaddr: tractor.Portal - async with tractor.wait_for_actor(service_name) as sockaddr: - print(f"my_service is found at {sockaddr}") + actor_portal: tractor.Portal + async with tractor.wait_for_actor( + service_name, + ) as actor_portal: + print(f'my_service is found at {actor_portal}') await an.cancel() diff --git a/examples/streaming_broadcast_fanout.py b/examples/streaming_broadcast_fanout.py index 15708795..c0ae45f1 100644 --- a/examples/streaming_broadcast_fanout.py +++ b/examples/streaming_broadcast_fanout.py @@ -28,6 +28,7 @@ async def tick_stream( # wait for the go-signal ensuring every parent-side # subscriber is attached before any tick is sent. assert await stream.receive() == 'go' + i: int for i in range(count): await stream.send(i) # falling out gracefully closes our stream side; @@ -37,15 +38,17 @@ async def tick_stream( async def consume( name: str, stream: tractor.MsgStream, - task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, + task_status: trio.TaskStatus[None] = trio.TASK_STATUS_IGNORED, ) -> None: ''' Consume a private broadcast-copy of the IPC stream. ''' + bcaster: tractor.trionics.BroadcastReceiver async with stream.subscribe() as bcaster: task_status.started() ticks: list[int] = [] + tick: int async for tick in bcaster: print(f'{name}: rx {tick}') ticks.append(tick) @@ -54,12 +57,19 @@ async def consume( async def main() -> None: + ''' + Fan one remote stream out to local subscribers. + + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( 'ticker', enable_modules=[__name__], ) + ctx: tractor.Context + first: int + stream: tractor.MsgStream async with ( portal.open_context( tick_stream, @@ -72,6 +82,7 @@ async def main() -> None: async with trio.open_nursery() as tn: # use `.start()` so each consumer is known # to be subscribed before the ticks flow. + i: int for i in range(3): await tn.start( consume, diff --git a/examples/trio/lockacquire_not_unmasked.py b/examples/trio/lockacquire_not_unmasked.py index 19077706..58fc681d 100644 --- a/examples/trio/lockacquire_not_unmasked.py +++ b/examples/trio/lockacquire_not_unmasked.py @@ -1,3 +1,4 @@ +from collections.abc import AsyncIterator from contextlib import ( asynccontextmanager as acm, ) @@ -16,7 +17,11 @@ _lock: trio.Lock|None = None @acm async def acquire_singleton_lock( -) -> None: +) -> AsyncIterator[trio.Lock]: + ''' + Acquire and yield the process-wide lock. + + ''' global _lock if _lock is None: log.info('Allocating LOCK') @@ -32,8 +37,15 @@ async def acquire_singleton_lock( async def hold_lock_forever( - task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, + task_status: trio.TaskStatus[ + trio.Lock, + ] = trio.TASK_STATUS_IGNORED, ) -> None: + ''' + Hold the singleton lock until cancellation. + + ''' + lock: trio.Lock async with ( tractor.trionics.maybe_raise_from_masking_exc(), acquire_singleton_lock() as lock, @@ -47,6 +59,11 @@ async def main( loglevel: str = 'info', debug_mode: bool = True, ) -> None: + ''' + Exercise lock acquisition while cancellation is masked. + + ''' + tn: trio.Nursery async with ( trio.open_nursery() as tn, @@ -58,7 +75,7 @@ async def main( from tractor.trionics import _taskc _taskc._mask_cases.clear() - _lock = await tn.start( + _held_lock: trio.Lock = await tn.start( hold_lock_forever, ) with trio.move_on_after(0.2): @@ -74,8 +91,8 @@ if __name__ == '__main__': tractor.log.get_console_log(level='info') for case in [True, False]: log.info( - f'\n' - f'------ RUNNING SCRIPT TRIAL ------\n' + '\n' + '------ RUNNING SCRIPT TRIAL ------\n' f'ignore_special_cases: {case!r}\n' ) trio.run(partial( diff --git a/examples/trio/send_chan_aclose_masks_beg.py b/examples/trio/send_chan_aclose_masks_beg.py index 971d8678..669c3dea 100644 --- a/examples/trio/send_chan_aclose_masks_beg.py +++ b/examples/trio/send_chan_aclose_masks_beg.py @@ -1,3 +1,4 @@ +from collections.abc import Iterator from contextlib import ( contextmanager as cm, # TODO, any diff in async case(s)?? @@ -17,7 +18,7 @@ log = tractor.log.get_logger( @cm def teardown_on_exc( raise_from_handler: bool = False, -): +) -> Iterator[None]: ''' You could also have a teardown handler which catches any exc and does some required teardown. In this case the problem is @@ -30,7 +31,7 @@ def teardown_on_exc( except BaseException as _berr: berr = _berr log.exception( - f'Handling termination teardown in child due to,\n' + 'Handling termination teardown in child due to,\n' f'{berr!r}\n' ) if raise_from_handler: @@ -54,14 +55,18 @@ def teardown_on_exc( async def finite_stream_to_rent( - tx: trio.abc.SendChannel, + tx: trio.abc.SendChannel[int], child_errors_mid_stream: bool, raise_unmasked: bool, task_status: trio.TaskStatus[ - trio.CancelScope, + trio.CancelScope|None, ] = trio.TASK_STATUS_IGNORED, -): +) -> None: + ''' + Stream values while reproducing exception masking on close. + + ''' async with ( # XXX without this unmasker the mid-streaming RTE is never # reported since it is masked by the `tx.aclose()` @@ -135,18 +140,25 @@ async def main( raise_unmasked: bool = False, loglevel: str = 'info', ) -> None: + ''' + Reproduce cancellation masking a child-stream failure. + + ''' tractor.log.get_console_log(level=loglevel) # the `.aclose()` being checkpoints on these # is the source of the problem.. + tx: trio.MemorySendChannel[int] + rx: trio.MemoryReceiveChannel[int] tx, rx = trio.open_memory_channel(1) + tn: trio.Nursery async with ( tractor.trionics.collapse_eg(), trio.open_nursery() as tn, rx as rx, ): - _child_cs = await tn.start( + _child_cs: trio.CancelScope|None = await tn.start( partial( finite_stream_to_rent, child_errors_mid_stream=child_errors_mid_stream, @@ -154,6 +166,7 @@ async def main( tx=tx, ) ) + msg: int async for msg in rx: log.debug( f'Rent rx {msg!r}\n' @@ -162,12 +175,13 @@ async def main( # simulate some external cancellation # request **JUST BEFORE** the child errors. if msg == 65: - log.cancel( - f'Cancelling parent on,\n' - f'msg={msg}\n' - f'\n' - f'Simulates OOB cancel request!\n' - ) + cancel_msg: str = ( + 'Cancelling parent on,\n' + 'msg={msg}\n' + '\n' + 'Simulates OOB cancel request!\n' + ).format(msg=msg) + log.cancel(cancel_msg) tn.cancel_scope.cancel() @@ -176,8 +190,8 @@ if __name__ == '__main__': tractor.log.get_console_log(level='info') for case in [True, False]: log.info( - f'\n' - f'------ RUNNING SCRIPT TRIAL ------\n' + '\n' + '------ RUNNING SCRIPT TRIAL ------\n' f'child_errors_midstream: {case!r}\n' ) try: diff --git a/examples/typed_payloads.py b/examples/typed_payloads.py index 58acada6..e70c075f 100644 --- a/examples/typed_payloads.py +++ b/examples/typed_payloads.py @@ -46,7 +46,9 @@ async def point_doubler( # 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)) + stream: tractor.MsgStream async with ctx.open_stream() as stream: + pt: Point async for pt in stream: # natively decoded to our struct type! assert type(pt) is Point @@ -59,12 +61,19 @@ async def point_doubler( async def main() -> None: + ''' + Exchange typed ``Point`` payloads with a subactor. + + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( 'point_doubler', enable_modules=[__name__], ) + ctx: tractor.Context + first: Point + stream: tractor.MsgStream async with ( portal.open_context( point_doubler, @@ -73,6 +82,7 @@ async def main() -> None: ): # the (validated) started-value from the child assert first == Point(x=0, y=0) + i: int for i in range(3): await stream.send(Point(x=i, y=i)) doubled: Point = await stream.receive() diff --git a/examples/uds_transport_actor_tree.py b/examples/uds_transport_actor_tree.py index 6555329d..dfcfd746 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -21,13 +21,17 @@ async def report_addr() -> str: Return this actor's own accept (bind) addr + pid. ''' - actor = tractor.current_actor() - addr: tuple = actor.accept_addr + actor: tractor.Actor = tractor.current_actor() + addr: tuple[str, str] = actor.accept_addr pid: int = os.getpid() return f'{actor.name}@{addr} pid={pid}' async def main() -> None: + ''' + Run a child actor over the UDS transport. + + ''' an: tractor.ActorNursery async with tractor.open_nursery( enable_transports=['uds'], @@ -52,7 +56,8 @@ async def main() -> None: ) # ask the child for its OWN distinct bind addr: another # 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()