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()