diff --git a/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md new file mode 100644 index 00000000..6304e7e8 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md @@ -0,0 +1,62 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: pr512-docs-review +timestamp: 2026-08-28T20:08:22Z +git_ref: wkt/big_boi_docs_472_follow_ups +scope: code +substantive: true +raw_file: 20260828T200822Z_0be872ff_prompt_io.raw.md +--- + +## Prompt + +Prime PR #512 for landing by applying the repository Python style and +typing policy across changed examples, reviewing and correcting the +result with delegated agents, validating it, and auditing documentation +changes since PR #460 for broader follow-up work. + +## Response summary + +Completed a branch-wide style, typing, and docstring pass over the +changed examples. Reworked the dedicated registrar example so it proves +external discovery, handles bind collisions, publishes readiness +atomically, and shuts down cleanly across platforms. Staged review moved +it under the discovery examples, introduced an async lifecycle manager, +and captured public-subsystem, Piker, and pytest follow-ups. Corrected +discovery and typed-messaging documentation after cross-checking runtime +behavior. +Created issue #514 for typed `Start` contracts and linked its planned +sender-side argument validation from the messaging guide. +Independent review passes were repeated until no actionable findings +remained. Static checks, Sphinx, runnable examples, Trio coverage, and +focused registrar checks passed; Ctrl-C debugger failures remain +unclassified after one case reproduced on untouched `main`. + +## Files changed + +- `examples/discovery/dedicated_registrar.py` - implement robust + standalone registrar discovery and teardown. +- `examples/**/*.py` - apply style, typing, docstring, and terminology + corrections to the 38 changed example modules. +- `docs/guide/discovery.rst` - document actual registrar and + multi-registrar behavior. +- `docs/guide/msging.rst` - document actual validation and codec-hook + boundaries. + +## Human edits + +The human selected PR #512, required the `/py-codestyle` policy and +typing improvements throughout the examples, required delegated review +agents to find additional fixes, authorized agents to own and push their +code through the proper review process, and requested a broad audit of +documentation changes since PR #460 rendered as an evolving Taken WIP. +The human then required the earlier `ParamSpec` proposal to be captured +as a dedicated GitHub issue and linked from the current docs patch. +The human also required the registrar demo to move under discovery, use +structural matching and an async context manager, explain socket ownership, +and preserve public API, Piker, and pytest-harness follow-ups. +The human also explicitly retracted messages intended for another +session; those messages did not direct the result. No direct manual +source edits were identified; the agent applied the human-directed +changes. diff --git a/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md new file mode 100644 index 00000000..554459f5 --- /dev/null +++ b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md @@ -0,0 +1,90 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-28T20:08:22Z +git_ref: wkt/big_boi_docs_472_follow_ups +diff_cmd: git diff 0be872ff..HEAD +--- + +## Prompt + +Prime PR #512 for landing. Apply the Python style policy throughout +the changed examples, improve typing, use review agents to find and +fix correctness issues, run the proper review process, push the agent's +own code, and then audit documentation changes since PR #460 for stale, +missing, defunct, or improvable coverage. + +## Response + +> `git diff 0be872ff..HEAD -- examples/` + +Restyled and typed all 38 changed example modules while preserving +intentional debugger failures and hanging reproducers. Added required +multiline docstrings, corrected async callable annotations, modernized +legacy API terminology, and preserved complete debugger process trees. + +Rebuilt `examples/discovery/dedicated_registrar.py` as a real +external-discovery example. A standalone registrar process now selects +and retries bounded ephemeral addresses, publishes readiness atomically, +rejects accidental registrar reuse, serves sibling service and client +actors, proves lookup used the registrar instead of a local-peer channel, +and performs bounded, validated, cross-platform shutdown. + +Staged review moved the demo into the discovery example group, converted +collision classification to structural pattern matching, documented the +selector-socket close/rebind race, and extracted process ownership into an +async context manager. A source TODO records the future public discovery +subsystem, Piker service-management lessons, and pytest registry-isolation +use case. + +> `git diff 0be872ff..HEAD -- docs/guide/discovery.rst` + +Corrected registrar, duplicate-name, and multi-registrar discovery +guidance, including configured-order and `None` placeholder behavior. + +> `git diff 0be872ff..HEAD -- docs/guide/msging.rst` + +Corrected typed-payload validation boundaries and separated working +task-scoped codec encoding from private per-dialog decoding and the +incomplete decorator hook parameters. + +## Review and validation + +Multiple independent reviews found and drove fixes for registrar +discovery validity, port-selection races, teardown, process diagnostics, +shutdown status, inaccurate discovery ordering, async callable typing, +missing docstrings, truncated debugger command diagrams, stale APIs, and +payload-error relay wording. + +Validation completed: + +- AST parsing, Ruff, 69-column checks, and required-docstring audit for + all 38 changed Python files. +- Sphinx HTML build succeeded. +- Documentation example harness: 24 passed. +- Trio coverage: 7 passed, 4 xfailed, 1 xpassed. +- Dedicated registrar direct run and focused harness test passed with a + clean child exit and no traceback. +- Debugger suite: 20 passed, 6 skipped, and 8 reproducible + `ctl-c=True` pexpect timeouts. One exact failure reproduced on + untouched `main`, confirming a baseline failure; its root cause + remains unclassified. No leaked actor processes remained. + +The post-PR-#460 audit identified follow-up work around advertised but +inert runtime selectors, platform/backend support, unresolved discovery +contracts, cached-context teardown, codec recipes, examples-as-tests +coverage, public API exports, broadcast factory contracts, stale examples, +README duplication, release notes, and process-title terminology. + +## Follow-up prompt + +Capture the previously proposed `ParamSpec`-based `Start` argument +validation work in a focused GitHub issue, then link it from the current +typed-messaging docs patch as planned sender-side checking. + +## Follow-up response + +Created https://github.com/goodboy/tractor/issues/514 to track deriving +typed `Start` contracts from endpoint signatures, preserving caller-facing +signatures, and validating arguments before sending where possible. Added +the issue link beside the guide's current `Start` validation boundary. diff --git a/docs/guide/discovery.rst b/docs/guide/discovery.rst index ea993b52..a14967e0 100644 --- a/docs/guide/discovery.rst +++ b/docs/guide/discovery.rst @@ -30,9 +30,9 @@ the registry tracks the live tree as it grows and shrinks. .. note:: Actor names are **not** enforced unique — the registry is keyed - by the full ``(name, uuid)`` pair. Name-based lookups simply - resolve to the *last* registered match, so if you boot five - actors all named ``'bob'``, you get the freshest ``'bob'`` B) + by the full ``(name, uuid)`` pair. A name lookup returns one + matching registration, but the API does not promise which match + wins. Use unique service names when selection matters. First boot: who's the registrar? -------------------------------- @@ -65,6 +65,43 @@ the one-and-only registrar; boot then fails loudly with a ``RuntimeError`` if some other process already bound the registry socket(s). +A dedicated registrar +--------------------- +That second rule — *"if a registrar answers, boot as a plain +root"* — is all you need to run the registry as its own +**standalone process**, decoupled from any app tree's root. In the +daemon process, enter ``open_root_actor()`` with an explicit +``registry_addrs`` and ``ensure_registry=True``; the latter makes +startup fail instead of silently joining a registrar that won the +address. Point each app tree at the address that daemon actually +bound: + +.. literalinclude:: ../../examples/discovery/dedicated_registrar.py + :caption: examples/discovery/dedicated_registrar.py + :language: python + +The example's selector socket binds but deliberately never listens. +It owns the kernel-selected local address only long enough to read it, +then closes so Tractor's actual listener can bind the same address. +This is not a socket transfer: the close/rebind handoff is non-atomic, +so the example retries with a fresh candidate only when registrar +startup reports that another process claimed the released address. +Retries are bounded, and other startup failures remain visible. It +publishes the selected address only after the actor context enters. +It also performs the lookup inside a separate ``client`` actor. The +service is its sibling, not its child, so the client has no spawn-time +service channel to satisfy the local-peer fast path. The +``query_actor()`` assertion verifies that a registrar portal handled +the lookup before ``find_actor()`` makes the service RPC. + +This is the "registrar as a subsystem, not the app root actor" +shape. Two caveats today (both tracked as #472 follow-ups): +``enable_transports`` is single-proto per runtime, so a registrar +can't yet serve multiple backends at once; and there's no way to +spawn a registrar as a *sub*-actor of a shared tree (only as its +own root), since ``start_actor()`` has no custom-``actor_cls`` +hook. + Looking up actors ----------------- @@ -92,13 +129,22 @@ Knobs worth knowing: - ``registry_addrs=[...]``: query specific (possibly multiple, possibly remote) registrars instead of your tree's default, -- ``only_first=False``: deliver a ``list[Portal]`` of *all* - matches found across the queried registrars instead of just the - first, +- ``only_first=True``: after all configured registrars are queried + concurrently, yield the result in the first ``registry_addrs`` + position. This is configured order, not first-reachable order, so + the result can be ``None`` even when a later registrar returned a + portal, -- ``raise_on_none=True``: raise a ``RuntimeError`` instead of - yielding ``None`` when no match is found — for when absence is - a hard error in your app. +- ``only_first=False``: when any query succeeds, yield an ordered + ``list[Portal | None]`` with one result per ``registry_addrs`` + position; misses remain ``None`` placeholders. When every query + misses, yield ``None`` instead of a list. This does not enumerate + every duplicate name in one registrar, + +- ``raise_on_none=True``: raise a ``RuntimeError`` when every + registrar query returns ``None``. With ``only_first=True`` it does + not raise merely because the first ordered result is ``None`` when + a later result is a portal. ``wait_for_actor()`` ******************** @@ -131,10 +177,11 @@ Yields a portal straight to the registrar actor itself — or a Fast paths and address preference --------------------------------- -Before doing any RPC to the registrar, every lookup first scans -the calling actor's *already-connected peers*: if you have a live -channel to an actor named ``name`` you get a portal over it -immediately, no registrar round-trip at all. +Before doing any RPC to the registrar, ``query_actor()``, +``wait_for_actor()``, and the default ``find_actor()`` lookup first +scan the calling actor's *already-connected peers*. If the caller +has a live channel to an actor named ``name``, it gets a portal over +that channel immediately, with no registrar round-trip. When a registry entry holds *multiple* addresses (a multihomed actor) the "best" one is chosen by locality: @@ -201,8 +248,8 @@ the existing registrar: Per the bootstrap rules above, if those addrs are absent this process becomes its own registrar root, so the same code works standalone and -as a tree-joiner. An occupied address that does not complete a Tractor -registrar handshake fails startup instead of being rebound. +as a tree-joiner. An occupied address that does not complete a +Tractor registrar handshake fails startup instead of being rebound. "Arbiter"? A legacy naming note ------------------------------- @@ -226,10 +273,11 @@ Very naive, very honest ----------------------- To be clear, this is a **very naive** discovery system: one -process-tree-local registrar holding a dict, no replication, no -re-election when it dies, no cross-host propagation. That's -intentional (for now); it covers the "wire up my services on this -host" case without dragging in a consensus protocol. +in-memory registrar holding a dict, no replication, no re-election +when it dies, and no automatic cross-host propagation. Separate +programs can use the same reachable registrar, as above, but must be +configured with its address. That's intentional (for now); it covers +the "wire up my services" case without a consensus protocol. On the roadmap (issue `#216`_ tracks a chunk of it): @@ -254,6 +302,7 @@ to hear from you. :class:`tractor.Registrar`. .. _gossip protocol: https://en.wikipedia.org/wiki/Gossip_protocol -.. _modern protocol: https://en.wikipedia.org/wiki/Rendezvous_protocol +.. _modern protocol: + https://en.wikipedia.org/wiki/Rendezvous_protocol .. _discovery: https://zguide.zeromq.org/docs/chapter8/#Discovery .. _#216: https://github.com/goodboy/tractor/issues/216 diff --git a/docs/guide/msging.rst b/docs/guide/msging.rst index 5c5a4d12..d0760eb9 100644 --- a/docs/guide/msging.rst +++ b/docs/guide/msging.rst @@ -153,14 +153,16 @@ the high-rate stream path. never even hits the wire. (You can opt out per-call with ``ctx.started(..., validate_pld_spec=False)`` if you measure a real cost.) -- ``Yield`` payloads are **never** checked inside - ``MsgStream.send()``; they're validated receiver-side on each - ``MsgStream.receive()``. A violation raises a ``MsgTypeError`` - in the receiver *and* relays an ``Error`` msg back so the - offending sender gets one raised too. -- the remaining control msgs (``Start``, ``Return``) are likewise - validated such that violations raise in the **sending** actor, - pointing the traceback at the code that actually goofed. +- ``Yield`` and ``Return`` payloads are not checked before sending; + they're decoded against the dialog's spec by the receiver. A + violation raises a ``MsgTypeError`` there and terminates that + dialog. The peer then observes the resulting protocol teardown; + it is not guaranteed to receive the same ``MsgTypeError``. +- ``Start`` arguments are dispatched through the RPC endpoint's + Python signature. They are not payloads covered by the dialog's + ``pld_spec``. A planned follow-up will derive a typed ``Start`` + contract from endpoint annotations and validate arguments + sender-side; see `#514`_. Anatomy of a ``MsgTypeError`` ----------------------------- @@ -177,21 +179,21 @@ a msg fails to decode against the active spec. The useful bits: ``.src_uid``, ``.ipc_msg`` and the fancy ``.pformat()`` tb-box rendering. -Practical reading guide: a *sender-side* MTE (``Started``, -``Return``) points straight at your offending ``await -ctx.started()`` or ``return`` statement, while a *receiver-side* -MTE (``Yield``) surfaces from the consumer's ``receive()`` call -with the relay copy delivered back to the producer. Either way -the failure is scoped to that one dialog; sibling contexts on the -same channel keep right on trucking. +Practical reading guide: a *sender-side* MTE for ``Started`` points +straight at the offending ``await ctx.started()`` call. A +*receiver-side* MTE for ``Yield`` or ``Return`` surfaces while the +peer decodes the payload. Either way the failure is scoped to that +one dialog; sibling contexts on the same channel keep right on +trucking. Custom wire types: ``mk_codec()`` and friends --------------------------------------------- msgspec covers a wide set of `builtin types`__ natively; for -anything else you teach the codec via extension hooks. The -easiest path is per-endpoint: ``@tractor.context()`` accepts -``enc_hook``/``dec_hook`` params right alongside ``pld_spec``. -For full control build and apply a codec yourself; encode-side: +anything else you teach the codec via extension hooks. The complete +public path currently available is task-scoped encoding: +``tractor.msg.mk_codec()`` builds a codec with an ``enc_hook``, and +``tractor.msg.apply_codec()`` installs it for the current task. To +build and apply that transport codec: __ https://jcristharif.com/msgspec/supported-types.html @@ -206,8 +208,9 @@ __ https://jcristharif.com/msgspec/supported-types.html with apply_codec(codec): # ContextVar-scoped override ... # msgs sent by this task now encode NSPs -and decode-side, scoped to an open context (note the import from -``tractor.msg._ops``, not yet re-exported): +The context manager which temporarily installs payload-decoder +settings on an open context is separate and still private (note +the ``tractor.msg._ops`` import): .. code:: python @@ -220,11 +223,15 @@ and decode-side, scoped to an open context (note the import from ): ... # this dialog's payloads decode as NSPs -``apply_codec()`` is ``ContextVar``-scoped: it overrides the -codec for the current task (and only that task), not the whole -process. For complete working flows, including hook pairing rules -and roundtrip cases, see ``tests/msg/test_ext_types_msgspec.py`` -and ``tests/msg/test_pldrx_limiting.py``. +``apply_codec()`` is ``ContextVar``-scoped: it overrides the codec +for the current task (and only that task), not the whole process. +``@tractor.context()`` accepts ``enc_hook`` and ``dec_hook`` +parameters, but their runtime wiring is not yet a symmetric, +end-to-end public hook pair: the encode hook is not consumed and +the decode hook is not applied on both peers. For the working flows +and their current boundaries, see +``tests/msg/test_ext_types_msgspec.py`` and +``tests/msg/test_pldrx_limiting.py``. The runtime dogfoods this pattern with :class:`tractor.msg.NamespacePath`: a ``str``-subtype shaped like @@ -238,9 +245,33 @@ Toward capability-based msging The ``pld_spec`` + codec-hook layer is the foundation for the long-game: **capability-based msging** where each dialog's type contract doubles as a capability grant, negotiated as part -of the protocol itself. That work is tracked in `#196`_ (with the -original typed-proto epic in `#36`_); if strongly-typed -distributed systems get you going, we'd love your input. +of the protocol itself. The epic is tracked in `#196`_ (evolving +the original typed-proto work in `#36`_), and the most recent +concrete step is `#365`_ — driving the whole ``pld_spec`` off +plain type-annotations (e.g. annotating a context's +``open_stream()`` with ``msgspec.Struct`` subtypes) instead of +explicit ``pld_spec=`` kwargs. + +You don't have to wait for that, though: the decorator-level +``@tractor.context(pld_spec=...)`` shown above is already the +*higher-level* way to pin a dialog's payload contract, while +``tractor.msg._ops.limit_plds()`` is the lower-level, per-block +escape hatch. Both are exercised end-to-end in +``tests/msg/test_pldrx_limiting.py`` and +``tests/msg/test_ext_types_msgspec.py``. + +The codec constructor and task-scoped override are public; the +per-dialog decoder override remains private, and the decorator hook +parameters remain incomplete. `#376`_ (from +`@guilledk `_, on the +`auto_codecs `_ +branch) instead drafts pair-building factories which derive +matching ``enc_hook``/``dec_hook`` functions and encoder/decoder +pairs from a type spec. That automation, not hook availability, +is the proposed long-term home for custom-type (de)serialization. + +If strongly-typed distributed systems get you going, we'd love +your input on any of the above. Where to next? -------------- @@ -258,3 +289,6 @@ Where to next? .. _(un)protocol: https://zguide.zeromq.org/docs/chapter7/#Unprotocols .. _#196: https://github.com/goodboy/tractor/issues/196 .. _#36: https://github.com/goodboy/tractor/issues/36 +.. _#365: https://github.com/goodboy/tractor/issues/365 +.. _#376: https://github.com/goodboy/tractor/pull/376 +.. _#514: https://github.com/goodboy/tractor/issues/514 diff --git a/examples/a_trynamic_first_scene.py b/examples/a_trynamic_first_scene.py index 85eb23e7..ba6d12b7 100644 --- a/examples/a_trynamic_first_scene.py +++ b/examples/a_trynamic_first_scene.py @@ -1,28 +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(): +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): +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(): - """Main tractor entry point, the "master" process (for now +async def main() -> None: + ''' + 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 @@ -39,6 +51,10 @@ async def main(): 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( @@ -47,13 +63,14 @@ async def main(): ) ) + 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 00dc645c..8b3dd0ac 100644 --- a/examples/actor_spawning_and_causality.py +++ b/examples/actor_spawning_and_causality.py @@ -2,14 +2,20 @@ import trio import tractor -async def cellar_door(): +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(): - """The main ``tractor`` routine. - """ +async def main() -> None: + ''' + 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 2e6824c9..166f08b0 100644 --- a/examples/actor_spawning_and_causality_with_daemon.py +++ b/examples/actor_spawning_and_causality_with_daemon.py @@ -2,19 +2,24 @@ import trio import tractor -async def movie_theatre_question(): - """A question asked in a dark theatre, in a tangent +async def movie_theatre_question() -> str: + ''' + 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(): - """The main ``tractor`` routine. - """ +async def main() -> None: + ''' + The main ``tractor`` routine. + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'frank', # enable the actor to run funcs from this current module enable_modules=[__name__], @@ -24,9 +29,8 @@ async def main(): # 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 237794a6..037230ff 100644 --- a/examples/asynchronous_generators.py +++ b/examples/asynchronous_generators.py @@ -1,33 +1,46 @@ -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(): +async def main() -> None: + ''' + Print messages streamed from a subactor. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'donny', enable_modules=[__name__], ) # 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 = 0 - async for letter in stream: - print(letter) + count: int = 0 + message: str + async for message in stream: + print(message) count += 1 if count > 50: diff --git a/examples/debugging/fast_error_in_root_after_spawn.py b/examples/debugging/fast_error_in_root_after_spawn.py index 5c2fdce2..fdafddef 100644 --- a/examples/debugging/fast_error_in_root_after_spawn.py +++ b/examples/debugging/fast_error_in_root_after_spawn.py @@ -13,7 +13,11 @@ import tractor @tractor.context async def sleep( ctx: tractor.Context, -): +) -> None: + ''' + Start a context after a brief initialization delay. + + ''' await trio.sleep(0.5) await ctx.started() await trio.sleep_forever() @@ -21,10 +25,13 @@ async def sleep( async def open_ctx( n: tractor.runtime._supervise.ActorNursery -): +) -> None: + ''' + Spawn a sleeper and open a context with it. + ''' # spawn both actors - portal = await n.start_actor( + portal: tractor.Portal = await n.start_actor( name='sleeper', enable_modules=[__name__], ) @@ -35,8 +42,11 @@ async def open_ctx( assert first is None -async def main(): +async def main() -> None: + ''' + Fail the root while a subactor context is still starting. + ''' async with tractor.open_nursery( debug_mode=True, loglevel='runtime', diff --git a/examples/debugging/multi_daemon_subactors.py b/examples/debugging/multi_daemon_subactors.py index e313803a..0c77ffed 100644 --- a/examples/debugging/multi_daemon_subactors.py +++ b/examples/debugging/multi_daemon_subactors.py @@ -1,9 +1,14 @@ +from collections.abc import AsyncIterator + import tractor import trio -async def breakpoint_forever(): - "Indefinitely re-enter debugger in child actor." +async def breakpoint_forever() -> AsyncIterator[str]: + ''' + Indefinitely re-enter debugger in child actor. + + ''' try: while True: yield 'yo' @@ -15,12 +20,15 @@ async def breakpoint_forever(): raise -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa -async def main(): +async def main() -> None: ''' Test breakpoint in a streaming actor. @@ -28,8 +36,14 @@ async def main(): async with tractor.open_nursery( debug_mode=True, ) as an: - p0 = await an.start_actor('bp_forever', enable_modules=[__name__]) - p1 = await an.start_actor('name_error', enable_modules=[__name__]) + p0: tractor.Portal = await an.start_actor( + 'bp_forever', + enable_modules=[__name__], + ) + p1: tractor.Portal = await an.start_actor( + 'name_error', + enable_modules=[__name__], + ) # retreive results async with p0.open_stream_from(breakpoint_forever) as stream: @@ -40,6 +54,7 @@ async def main(): except tractor.RemoteActorError as rae: assert rae.boxed_type is NameError + i: str async for i in stream: # a second time try the failing subactor and this tie diff --git a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py b/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py index 9929b498..6357b96f 100644 --- a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py +++ b/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py @@ -4,13 +4,19 @@ import trio import tractor -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa -async def breakpoint_forever(): - "Indefinitely re-enter debugger in child actor." +async def breakpoint_forever() -> None: + ''' + Indefinitely re-enter debugger in child actor. + + ''' while True: await tractor.pause() @@ -20,9 +26,13 @@ async def breakpoint_forever(): # await trio.sleep(0) -async def spawn_until(depth=0): - """"A nested nursery that triggers another ``NameError``. - """ +async def spawn_until( + depth: int = 0, +) -> None: + ''' + A nested nursery that triggers another ``NameError``. + + ''' async with ( tractor.open_nursery() as an, trio.open_nursery() as tn, @@ -37,7 +47,8 @@ async def spawn_until(depth=0): ) ) - # Let the background one-shot enter `breakpoint_forever()` + # Let the background one-shot enter + # `breakpoint_forever()` # before its sibling raises and cancellation propagates. await trio.sleep(0.5) # rx and propagate error from child @@ -48,9 +59,9 @@ async def spawn_until(depth=0): ) else: - # recusrive call to spawn another process branching layer of - # the tree; blocks (up) each level until the leaf's - # `name_error` relays through. + # recusrive call to spawn another process branching + # layer of the tree; blocks (up) each level until the + # leaf's `name_error` relays through. depth -= 1 await tractor.to_actor.run( partial( @@ -63,25 +74,35 @@ async def spawn_until(depth=0): # TODO: notes on the new boxed-relayed errors through proxy actors -async def main(): - """The main ``tractor`` routine. +async def main() -> None: + ''' + The main ``tractor`` routine. - The process tree should look as approximately as follows when the debugger - first engages: + The process tree should look approximately as follows when the + debugger first engages: python examples/debugging/multi_nested_subactors_bp_forever.py - ├─ python -m tractor._child --uid ('spawner1', '7eab8462 ...) - │ └─ python -m tractor._child --uid ('spawn_until_3', 'afcba7a8 ...) - │ └─ python -m tractor._child --uid ('spawn_until_2', 'd2433d13 ...) - │ └─ python -m tractor._child --uid ('spawn_until_1', '1df589de ...) - │ └─ python -m tractor._child --uid ('spawn_until_0', '3720602b ...) + ├─ python -m tractor._child --uid + │ ('spawner1', '7eab8462 ...') + │ └─ python -m tractor._child --uid + │ ('spawn_until_3', 'afcba7a8 ...') + │ └─ python -m tractor._child --uid + │ ('spawn_until_2', 'd2433d13 ...') + │ └─ python -m tractor._child --uid + │ ('spawn_until_1', '1df589de ...') + │ └─ python -m tractor._child --uid + │ ('spawn_until_0', '3720602b ...') │ - └─ python -m tractor._child --uid ('spawner0', '1d42012b ...) - └─ python -m tractor._child --uid ('spawn_until_2', '2877e155 ...) - └─ python -m tractor._child --uid ('spawn_until_1', '0502d786 ...) - └─ python -m tractor._child --uid ('spawn_until_0', 'de918e6d ...) + └─ python -m tractor._child --uid + ('spawner0', '1d42012b ...') + └─ python -m tractor._child --uid + ('spawn_until_2', '2877e155 ...') + └─ python -m tractor._child --uid + ('spawn_until_1', '0502d786 ...') + └─ python -m tractor._child --uid + ('spawn_until_0', 'de918e6d ...') - """ + ''' async with ( tractor.open_nursery( debug_mode=True, diff --git a/examples/debugging/multi_subactor_root_errors.py b/examples/debugging/multi_subactor_root_errors.py index 5aa3a4ff..a9d0debb 100644 --- a/examples/debugging/multi_subactor_root_errors.py +++ b/examples/debugging/multi_subactor_root_errors.py @@ -7,14 +7,19 @@ import trio import tractor -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa -async def spawn_error(): - """"A nested nursery that triggers another ``NameError``. - """ +async def spawn_error() -> None: + ''' + A nested nursery that triggers another ``NameError``. + + ''' async with tractor.open_nursery() as an: return await tractor.to_actor.run( name_error, @@ -23,8 +28,9 @@ async def spawn_error(): ) -async def main(): - """The main ``tractor`` routine. +async def main() -> None: + ''' + The main ``tractor`` routine. The process tree should look as approximately as follows: @@ -37,7 +43,8 @@ async def main(): - nested name_error sub-sub-actor - root actor should then fail on assert - program termination - """ + + ''' async with ( tractor.open_nursery( debug_mode=True, @@ -46,11 +53,11 @@ async def main(): trio.open_nursery() as tn, ): # spawn both actors.. - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'name_error', enable_modules=[__name__], ) - portal1 = await an.start_actor( + portal1: tractor.Portal = await an.start_actor( 'spawn_error', enable_modules=[__name__], ) diff --git a/examples/debugging/multi_subactors.py b/examples/debugging/multi_subactors.py index 63ab5404..efa8bfd1 100644 --- a/examples/debugging/multi_subactors.py +++ b/examples/debugging/multi_subactors.py @@ -1,22 +1,32 @@ +from collections.abc import Awaitable, Callable + import tractor import trio -async def breakpoint_forever(): - "Indefinitely re-enter debugger in child actor." +async def breakpoint_forever() -> None: + ''' + Indefinitely re-enter debugger in child actor. + + ''' while True: await trio.sleep(0.1) await tractor.pause() -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa -async def spawn_error(): - """"A nested nursery that triggers another ``NameError``. - """ +async def spawn_error() -> None: + ''' + A nested nursery that triggers another ``NameError``. + + ''' async with tractor.open_nursery() as an: return await tractor.to_actor.run( name_error, @@ -25,8 +35,9 @@ async def spawn_error(): ) -async def main(): - """The main ``tractor`` routine. +async def main() -> None: + ''' + The main ``tractor`` routine. The process tree should look as approximately as follows: @@ -35,7 +46,8 @@ async def main(): |-python -m tractor._child --uid ('bp_forever', '1f787a7e ...) `-python -m tractor._child --uid ('spawn_error', '52ee14a5 ...) `-python -m tractor._child --uid ('name_error', '3391222c ...) - """ + + ''' errors: list[BaseException] = [] async with tractor.open_nursery( @@ -43,12 +55,14 @@ async def main(): # loglevel='runtime', ) as an: - async def run_and_collect(fn): + async def run_and_collect( + fn: Callable[[], Awaitable[object]], + ) -> None: ''' One-shot whose (boxed) error is stashed instead of raised so a sibling's crash never cancels the others before they've had their own debugger sessions (the - "collect all errors" the legacy `run_in_actor()` API + 'collect all errors' the legacy `run_in_actor()` API did implicitly at nursery teardown). ''' diff --git a/examples/debugging/per_actor_debug.py b/examples/debugging/per_actor_debug.py index 189fb45e..8345a35f 100644 --- a/examples/debugging/per_actor_debug.py +++ b/examples/debugging/per_actor_debug.py @@ -1,19 +1,28 @@ import trio import tractor -async def die(): + +async def die() -> None: + ''' + Deliberately crash the calling actor. + + ''' raise RuntimeError -async def main(): +async def main() -> None: + ''' + Crash actors with different debugger settings concurrently. + + ''' async with tractor.open_nursery() as an: - debug_actor = await an.start_actor( + debug_actor: tractor.Portal = await an.start_actor( 'debugged_boi', enable_modules=[__name__], debug_mode=True, ) - crash_boi = await an.start_actor( + crash_boi: tractor.Portal = await an.start_actor( 'crash_boi', enable_modules=[__name__], # debug_mode=True, diff --git a/examples/debugging/pm_in_subactor.py b/examples/debugging/pm_in_subactor.py index a8f5048e..cc60347d 100644 --- a/examples/debugging/pm_in_subactor.py +++ b/examples/debugging/pm_in_subactor.py @@ -5,7 +5,7 @@ import tractor @tractor.context async def name_error( ctx: tractor.Context, -): +) -> None: ''' Raise a `NameError`, catch it and enter `.post_mortem()`, then expect the `._rpc._invoke()` crash handler to also engage. @@ -18,7 +18,7 @@ async def name_error( raise -async def main(): +async def main() -> None: ''' Test 3 `PdbREPL` entries: - one in the child due to manual `.post_mortem()`, @@ -49,7 +49,9 @@ async def main(): await tractor.post_mortem() raise else: - raise RuntimeError('IPC ctx should have remote errored!?') + raise RuntimeError( + 'IPC ctx should have remote errored!?' + ) if __name__ == '__main__': diff --git a/examples/debugging/root_actor_breakpoint.py b/examples/debugging/root_actor_breakpoint.py index 55b4ca56..35a029ec 100644 --- a/examples/debugging/root_actor_breakpoint.py +++ b/examples/debugging/root_actor_breakpoint.py @@ -2,8 +2,11 @@ import trio import tractor -async def main(): +async def main() -> None: + ''' + Pause in the root actor to exercise its debugger REPL. + ''' async with tractor.open_root_actor( debug_mode=True, ): diff --git a/examples/debugging/root_actor_error.py b/examples/debugging/root_actor_error.py index fab46335..1fae47b3 100644 --- a/examples/debugging/root_actor_error.py +++ b/examples/debugging/root_actor_error.py @@ -2,7 +2,11 @@ import trio import tractor -async def main(): +async def main() -> None: + ''' + Raise an assertion error from the debug-enabled root actor. + + ''' async with tractor.open_root_actor( debug_mode=True, ): diff --git a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py b/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py index 75e1c9a4..7fafa1bb 100644 --- a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py +++ b/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py @@ -4,14 +4,21 @@ import trio import tractor -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa -async def spawn_until(depth=0): - """"A nested nursery that triggers another ``NameError``. - """ +async def spawn_until( + depth: int = 0, +) -> None: + ''' + A nested nursery that triggers another ``NameError``. + + ''' async with tractor.open_nursery() as an: if depth < 1: await tractor.to_actor.run(name_error, an=an) @@ -27,18 +34,23 @@ async def spawn_until(depth=0): ) -async def main(): +async def main() -> None: ''' The process tree should look as approximately as follows when the debugger first engages: python examples/debugging/multi_nested_subactors_bp_forever.py - ├─ python -m tractor._child --uid ('spawner1', '7eab8462 ...) - │ └─ python -m tractor._child --uid ('spawn_until_0', '3720602b ...) - │ └─ python -m tractor._child --uid ('name_error', '505bf71d ...) + ├─ python -m tractor._child --uid + │ ('spawner1', '7eab8462 ...') + │ └─ python -m tractor._child --uid + │ ('spawn_until_0', '3720602b ...') + │ └─ python -m tractor._child --uid + │ ('name_error', '505bf71d ...') │ - └─ python -m tractor._child --uid ('spawner0', '1d42012b ...) - └─ python -m tractor._child --uid ('name_error', '6c2733b8 ...) + └─ python -m tractor._child --uid + ('spawner0', '1d42012b ...') + └─ python -m tractor._child --uid + ('name_error', '6c2733b8 ...') ''' async with ( diff --git a/examples/debugging/root_self_cancelled_w_error.py b/examples/debugging/root_self_cancelled_w_error.py index b3c15288..fb4b2ecb 100644 --- a/examples/debugging/root_self_cancelled_w_error.py +++ b/examples/debugging/root_self_cancelled_w_error.py @@ -2,7 +2,11 @@ import trio import tractor -async def main(): +async def main() -> None: + ''' + Enter shielded debugging after root cancellation, then fail. + + ''' async with tractor.open_root_actor( debug_mode=True, loglevel='cancel', @@ -18,16 +22,19 @@ async def main(): try: await tractor.pause() except trio.Cancelled as _taskc: - assert (root_cs := _root._root_tn.cancel_scope).cancel_called + root_cs: trio.CancelScope + assert ( + root_cs := _root._root_tn.cancel_scope + ).cancel_called # NOTE^^ above logic but inside `open_root_actor()` and # passed to the `shield=` expression is effectively what # we're testing here! await tractor.pause(shield=root_cs.cancel_called) - # XXX, if shield logic *is wrong* inside `open_root_actor()`'s - # crash-handler block this should never be interacted, - # instead `trio.Cancelled` would be bubbled up: the original - # BUG. + # XXX, if shield logic *is wrong* inside + # `open_root_actor()`'s crash-handler block this should never + # be interacted, instead `trio.Cancelled` would be bubbled + # up: the original BUG. assert 0 diff --git a/examples/debugging/root_timeout_while_child_crashed.py b/examples/debugging/root_timeout_while_child_crashed.py index 043cb5c7..8ca1a813 100644 --- a/examples/debugging/root_timeout_while_child_crashed.py +++ b/examples/debugging/root_timeout_while_child_crashed.py @@ -2,12 +2,15 @@ import trio import tractor -async def key_error(): - "Raise a ``NameError``" +async def key_error() -> None: + ''' + Raise a ``KeyError``. + + ''' return {}['doggy'] -async def main(): +async def main() -> None: ''' Root is fail-after-cancelled while blocking and child RPC fails simultaneously. @@ -21,7 +24,7 @@ async def main(): trio.open_nursery() as tn, ): # spawn the actor.. - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'key_error', enable_modules=[__name__], ) @@ -32,9 +35,9 @@ async def main(): # root blocks below. tn.start_soon(portal.run, key_error) - # XXX: originally a bug caused by this is where root would enter - # the debugger and clobber the tty used by the repl even though - # child should have it locked. + # XXX: originally a bug caused by this is where root would + # enter the debugger and clobber the tty used by the repl + # even though child should have it locked. with trio.fail_after(1): await trio.Event().wait() diff --git a/examples/debugging/shielded_pause.py b/examples/debugging/shielded_pause.py index 5a8c50e7..a145a3bb 100644 --- a/examples/debugging/shielded_pause.py +++ b/examples/debugging/shielded_pause.py @@ -3,8 +3,15 @@ import tractor async def cancellable_pause_loop( - task_status: trio.TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED -): + task_status: trio.TaskStatus[ + trio.CancelScope + ] = trio.TASK_STATUS_IGNORED, +) -> None: + ''' + Exercise shielded debugger pauses under cancellation. + + ''' + cs: trio.CancelScope with trio.CancelScope() as cs: task_status.started(cs) for _ in range(3): @@ -30,7 +37,11 @@ async def cancellable_pause_loop( await trio.lowlevel.checkpoint() -async def pm_on_cancelled(): +async def pm_on_cancelled() -> None: + ''' + Compare shielded and unshielded post-mortem entry. + + ''' async with trio.open_nursery() as tn: tn.cancel_scope.cancel() try: @@ -56,7 +67,7 @@ async def pm_on_cancelled(): async def cancelled_before_pause( -): +) -> None: ''' Verify that using a shielded pause works despite surrounding cancellation called state in the calling task. @@ -71,7 +82,11 @@ async def cancelled_before_pause( await pm_on_cancelled() -async def main(): +async def main() -> None: + ''' + Exercise shielded debugger entry in subactor and root tasks. + + ''' async with tractor.open_nursery( debug_mode=True, ) as an: diff --git a/examples/debugging/subactor_bp_in_ctx.py b/examples/debugging/subactor_bp_in_ctx.py index 36d4b2b3..3b3074c2 100644 --- a/examples/debugging/subactor_bp_in_ctx.py +++ b/examples/debugging/subactor_bp_in_ctx.py @@ -1,10 +1,15 @@ import platform +from collections.abc import AsyncIterator import tractor import trio -async def gen(): +async def gen() -> AsyncIterator[str]: + ''' + Yield values around debugger pauses. + + ''' yield 'yo' await tractor.pause() yield 'yo' @@ -15,11 +20,15 @@ async def gen(): async def just_bp( ctx: tractor.Context, ) -> None: + ''' + Pause repeatedly before deliberately breaking the context. + ''' await ctx.started() await tractor.pause() # TODO: bps and errors in this call.. + val: str async for val in gen(): print(val) @@ -34,15 +43,18 @@ async def just_bp( -async def main(): +async def main() -> None: + ''' + Run the breakpoint context over a supported transport. + ''' # !TODO, parametrize the --tpt-proto={key} with osenv vars just # like we do for loglevel/spawn-backend! # - [ ] run on both tpts for all such debugger tests? # - [ ] special skip for macos! # if platform.system() != 'Darwin': - tpt = 'uds' + tpt: str = 'uds' else: # XXX, precisely we can't use pytest's tmp-path generation # for tests.. apparently because: @@ -59,7 +71,7 @@ async def main(): enable_transports=[tpt], loglevel='devx', ) as an: - p = await an.start_actor( + p: tractor.Portal = await an.start_actor( 'bp_boi', enable_modules=[__name__], ) diff --git a/examples/debugging/subactor_breakpoint.py b/examples/debugging/subactor_breakpoint.py index e3a4f250..7b844f10 100644 --- a/examples/debugging/subactor_breakpoint.py +++ b/examples/debugging/subactor_breakpoint.py @@ -2,7 +2,7 @@ import trio import tractor -async def breakpoint_forever(): +async def breakpoint_forever() -> None: ''' Indefinitely re-enter debugger in child actor. @@ -12,8 +12,11 @@ async def breakpoint_forever(): await tractor.pause() -async def main(): +async def main() -> None: + ''' + Run a subactor that repeatedly pauses in the debugger. + ''' async with tractor.open_nursery( debug_mode=True, loglevel='cancel', diff --git a/examples/debugging/subactor_error.py b/examples/debugging/subactor_error.py index 95c1fe12..8ec65781 100644 --- a/examples/debugging/subactor_error.py +++ b/examples/debugging/subactor_error.py @@ -2,11 +2,19 @@ import trio import tractor -async def name_error(): +async def name_error() -> None: + ''' + Deliberately raise a ``NameError`` in a subactor. + + ''' getattr(doggypants) # noqa (on purpose) -async def main(): +async def main() -> None: + ''' + Surface a subactor `NameError` at the waiting root task. + + ''' async with tractor.open_nursery( debug_mode=True, ) as an: diff --git a/examples/discovery/dedicated_registrar.py b/examples/discovery/dedicated_registrar.py new file mode 100644 index 00000000..f4386f13 --- /dev/null +++ b/examples/discovery/dedicated_registrar.py @@ -0,0 +1,394 @@ +''' +Run a dedicated registrar in a standalone process. + +The service and discovery client are sibling actors. The client has +no pre-existing channel to the service, so its lookup must use the +external registrar instead of the local-peer fast path. + +''' +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager as acm +import errno +from pathlib import Path +import signal +import socket +import subprocess +import sys +import tempfile +import time + +import trio +import tractor + + +MAX_BIND_ATTEMPTS: int = 5 + + +def _is_addr_collision(exc: BaseException) -> bool: + ''' + Return whether registrar startup lost the selected TCP address. + + Tractor can notice the collision while probing the address or + later when its listener binds. Exception groups are retryable + only when every contained failure reports the same collision. + + ''' + match exc: + case BaseExceptionGroup(exceptions=exceptions): + return bool(exceptions) and all( + _is_addr_collision(child) + for child in exceptions + ) + + case OSError() as os_error: + return ( + os_error.errno in {errno.EADDRINUSE, 10048} + or getattr(os_error, 'winerror', None) == 10048 + ) + + case RuntimeError() as runtime_error: + message: str = str(runtime_error) + return ( + 'Registry address(es) are occupied' in message + or 'registry socket(s) already bound' in message + ) + + case _: + return False + + +def run_registrar(ready_path: str) -> None: + ''' + Serve as the required registrar and report its selected address. + + The kernel selects ephemeral loopback candidates in this process. + If another process claims a released candidate first, retry with + a fresh candidate up to `MAX_BIND_ATTEMPTS`. Other startup errors + and the final collision remain visible. `ensure_registry=True` + prevents silently joining a registrar that won the address. + + ''' + ready_file: Path = Path(ready_path) + + async def serve() -> None: + ''' + Open the registrar, publish readiness, and serve forever. + + ''' + for attempt in range(1, MAX_BIND_ATTEMPTS + 1): + # This selector socket reserves and reports a + # kernel-selected candidate; it never listens and is + # not transferred to Tractor. Closing it lets + # `open_root_actor()` create its own listener on the + # same addr. The close/rebind handoff is non-atomic, + # hence the bounded collision retries. + sock: socket.socket + with socket.socket( + socket.AF_INET, + socket.SOCK_STREAM, + ) as sock: + sock.bind(('127.0.0.1', 0)) + selected: tuple[str, int] = sock.getsockname() + registry_addr: tuple[str, int] = ( + selected[0], + selected[1], + ) + + try: + actor: tractor.Actor + async with tractor.open_root_actor( + name='dedicated_registrar', + registry_addrs=[registry_addr], + enable_transports=['tcp'], + enable_modules=[], + ensure_registry=True, + loglevel='error', + ) as actor: + if not actor.is_registrar: + raise RuntimeError( + 'daemon did not become registrar' + ) + + tmp_file: Path = ready_file.with_suffix('.tmp') + tmp_file.write_text( + str(registry_addr[1]), + encoding='ascii', + ) + tmp_file.replace(ready_file) + await trio.sleep_forever() + + except BaseException as exc: + if ( + not _is_addr_collision(exc) + or attempt == MAX_BIND_ATTEMPTS + ): + raise + await trio.sleep(.05 * attempt) + + try: + trio.run(serve) + except KeyboardInterrupt: + pass + + +def _registrar_command(ready_path: Path) -> list[str]: + ''' + Build a child command that loads without running `main()`. + + `runpy.run_path()` also works when the docs test copies and + renames this example before executing it. + + ''' + module_path: str = repr(str(Path(__file__).resolve())) + function_name: str = repr('run_registrar') + ready_arg: str = repr(str(ready_path)) + code: str = ( + f'import runpy; module = runpy.run_path({module_path}); ' + f'module[{function_name}]({ready_arg})' + ) + return [sys.executable, '-c', code] + + +def _wait_registrar_ready( + ready_path: Path, + proc: subprocess.Popen, + deadline: float = 10.0, +) -> tuple[str, int]: + ''' + Wait until the child has entered its registrar actor context. + + The child atomically publishes its selected port only after + `open_root_actor()` completes. Fail early if startup crashes. + + ''' + end: float = time.monotonic() + deadline + while time.monotonic() < end: + if proc.poll() is not None: + returncode: int|None = proc.returncode + raise RuntimeError( + f'registrar exited during startup: {returncode=}' + ) + + try: + port: int = int( + ready_path.read_text(encoding='ascii') + ) + except ( + OSError, + ValueError, + ): + time.sleep(.05) + continue + + if not 0 < port < 2**16: + raise RuntimeError(f'invalid registrar port: {port!r}') + if proc.poll() is not None: + raise RuntimeError( + 'registrar exited after reporting ready' + ) + return ('127.0.0.1', port) + + raise TimeoutError('registrar did not report ready') + + +def _stop_registrar( + proc: subprocess.Popen, + graceful_timeout: float = 5.0, +) -> None: + ''' + Stop and reap the registrar, escalating after a bounded wait. + + Windows children receive `CTRL_C_EVENT` in their new process + group; POSIX children receive `SIGINT`. A child that ignores + graceful shutdown is killed, and every path finishes with + `wait()`. A non-zero child exit remains visible to the caller. + + ''' + if proc.poll() is None: + graceful_signal: int = ( + signal.CTRL_C_EVENT + if sys.platform == 'win32' + else signal.SIGINT + ) + try: + proc.send_signal(graceful_signal) + except OSError: + if proc.poll() is None: + proc.terminate() + + try: + proc.wait(timeout=graceful_timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + + if proc.returncode: + raise RuntimeError( + 'registrar shutdown failed: ' + f'returncode={proc.returncode}' + ) + + +async def greet() -> str: + ''' + Return a greeting identifying the actor serving the RPC. + + ''' + actor_name: str = tractor.current_actor().name + return f'hello from {actor_name}!' + + +async def discover_and_greet( + registry_addr: tuple[str, int], +) -> tuple[str, str, str]: + ''' + Prove registrar lookup from a client without a service channel. + + The parent spawns this actor as `greeter`'s sibling. A non-`None` + registry portal from `query_actor()` proves that discovery did + not take the existing-peer fast path, which returns no registry + portal. + + ''' + service_addr: tuple[str, int]|None + registry_portal: tractor.Portal|None + async with tractor.query_actor( + 'greeter', + regaddr=registry_addr, + ) as (service_addr, registry_portal): + if registry_portal is None: + raise RuntimeError('lookup used a local service channel') + if service_addr is None: + raise RuntimeError('greeter was not registered') + + service_portal: tractor.Portal|None + async with tractor.find_actor( + 'greeter', + registry_addrs=[registry_addr], + ) as service_portal: + if service_portal is None: + raise RuntimeError('greeter disappeared before RPC') + greeting: str = await service_portal.run(greet) + + client_name: str = tractor.current_actor().name + return client_name, repr(service_addr), greeting + + +async def app(registry_addr: tuple[str, int]) -> None: + ''' + Use sibling service and client actors with an external registrar. + + Only the parent receives both spawn-time portals. The `client` + actor performs discovery in its own process and has no direct + `greeter` channel before the lookup. + + ''' + actor_nursery: tractor.ActorNursery + async with tractor.open_nursery( + registry_addrs=[registry_addr], + enable_transports=['tcp'], + ) as actor_nursery: + await actor_nursery.start_actor( + 'greeter', + enable_modules=[__name__], + ) + client_portal: tractor.Portal = ( + await actor_nursery.start_actor( + 'client', + enable_modules=[__name__], + ) + ) + result: tuple[str, str, str] = await client_portal.run( + discover_and_greet, + registry_addr=registry_addr, + ) + client_name: str + service_addr: str + greeting: str + ( + client_name, + service_addr, + greeting, + ) = result + print( + f'{client_name!r} found `greeter` through registrar ' + f'{registry_addr!r}; service address: {service_addr}\n' + f'{greeting}' + ) + await actor_nursery.cancel() + + +# TODO: Promote this lifecycle into an OTB `tractor.discovery` +# registrar subsystem. Reuse attach-or-create ownership from +# `piker.service.maybe_open_pikerd()` and named service supervision +# from `piker.service.Services`; replace the file readiness +# handshake, then use the API from `tractor._testing.pytest` to +# isolate remaining hard-coded `reg_addr` cases. +@acm +async def _open_registrar( +) -> AsyncIterator[tuple[str, int]]: + ''' + Start, publish, and reap one dedicated registrar process. + + The Windows child gets a distinct console process group so the + graceful control event targets it without interrupting this + process. + + ''' + temp_dir: str + with tempfile.TemporaryDirectory( + prefix='tractor-registrar-', + ) as temp_dir: + ready_path: Path = Path(temp_dir) / 'ready' + creationflags: int = ( + subprocess.CREATE_NEW_PROCESS_GROUP + if sys.platform == 'win32' + else 0 + ) + registrar: subprocess.Popen = subprocess.Popen( + _registrar_command(ready_path), + stdout=subprocess.DEVNULL, + creationflags=creationflags, + ) + primary_error: BaseException|None = None + try: + registry_addr: tuple[str, int] = _wait_registrar_ready( + ready_path, + registrar, + ) + print( + f'dedicated registrar ready at {registry_addr!r} ' + f'(pid {registrar.pid})' + ) + yield registry_addr + except BaseException as error: + primary_error = error + raise + finally: + try: + _stop_registrar(registrar) + except BaseException as cleanup_error: + if primary_error is None: + raise + cleanup_note: str = ( + 'registrar cleanup also failed: ' + f'{cleanup_error!r}' + ) + primary_error.add_note(cleanup_note) + print('dedicated registrar shut down') + + +async def main() -> None: + ''' + Run the external registrar and sibling discovery actors. + + ''' + registry_addr: tuple[str, int] + async with _open_registrar() as registry_addr: + await app(registry_addr) + + +if __name__ == '__main__': + trio.run(main) diff --git a/examples/full_fledged_streaming_service.py b/examples/full_fledged_streaming_service.py index 390a1b75..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): +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): +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,30 +32,47 @@ async def aggregate(seed): an: ActorNursery async with tractor.open_nursery() as an: portals: list[Portal] = [] + i: int for i in range(1, 3): # fork/spawn call - portal = await an.start_actor( + portal: Portal = await an.start_actor( name=f'streamer_{i}', enable_modules=[__name__], ) 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, 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: for portal in portals: @@ -61,8 +85,9 @@ async def aggregate(seed): # 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) @@ -71,11 +96,11 @@ async def aggregate(seed): 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]: @@ -94,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', @@ -108,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 e3ff2a09..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():`` @@ -28,9 +31,15 @@ async def aio_echo_server( @tractor.context 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(): +async def main() -> None: + ''' + Run the infected asyncio echo-server example. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - p = 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(): 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 60e30a95..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,21 +33,27 @@ 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: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'mpi-child', enable_modules=[child_fn.__module__], # Without this the child replays __main__, which # re-imports mpi4py and crashes on MPI_Init. inherit_parent_main=False, ) - result = await portal.run(child_fn) - print(f"[parent] got: {result}", flush=True) + result: str = await portal.run(child_fn) + 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 3e592a45..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,17 +7,30 @@ import tractor log = tractor.log.get_logger('multiportal') -async def stream_data(seed=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, consumed): +async def stream_from_portal( + 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) @@ -23,20 +38,33 @@ async def stream_from_portal(p, consumed): consumed.append(item) -async def main(): +async def main() -> None: + ''' + Consume two concurrent streams through one portal. + ''' + an: tractor.ActorNursery async with tractor.open_nursery(loglevel='info') as an: - p = await an.start_actor('stream_boi', enable_modules=[__name__]) + portal: tractor.Portal = await an.start_actor( + 'stream_boi', + enable_modules=[__name__], + ) - consumed = [] + 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 a0c66b88..279e3a4d 100644 --- a/examples/nested_actor_tree.py +++ b/examples/nested_actor_tree.py @@ -41,6 +41,7 @@ async def fan_out_squares( aggregated squares to our parent. ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portals: list[tractor.Portal] = [] for i in (1, 2): @@ -52,21 +53,28 @@ 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, ) # fan out one sub-RPC per input val, concurrently. + tn: trio.Nursery async with trio.open_nursery() as tn: for i, x in enumerate(vals): tn.start_soon( @@ -83,11 +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 = await an.start_actor( + 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 e5b32359..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, @@ -31,7 +41,11 @@ PRIMES = [ ] -async def is_prime(n): +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): @acm -async def worker_pool(workers=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 = [] + 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,17 +95,36 @@ async def worker_pool(workers=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 - async def send_result(func, value, portal): - await snd_chan.send((value, await portal.run(func, n=value))) + ''' + # define an async (local) task to collect results from + # workers + async def send_result( + func: Callable[[int], Awaitable[bool]], + value: int, + portal: tractor.Portal, + ) -> 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, @@ -101,21 +143,30 @@ async def worker_pool(workers=4): await an.cancel() -async def main(): +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 46fc4837..71458aa1 100644 --- a/examples/parallelism/single_func.py +++ b/examples/parallelism/single_func.py @@ -1,30 +1,36 @@ -""" +''' 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 import trio -async def burn_cpu(): +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(): +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 @@ -32,9 +38,9 @@ async def main(): # run the same func as the lone task in a subactor, # block on and collect its PID as the caller-side result - pid = 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__': diff --git a/examples/quick_cluster.py b/examples/quick_cluster.py index 3fa4ca2a..be3597aa 100644 --- a/examples/quick_cluster.py +++ b/examples/quick_cluster.py @@ -1,20 +1,24 @@ - 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() -async def main(): +async def main() -> None: ''' Spawn a flat actor cluster, with one process per detected core. ''' portal_map: dict[str, tractor.Portal] + tn: trio.Nursery # look at this hip new syntax! async with ( @@ -27,7 +31,9 @@ async def main(): 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 db9eb9b5..fe399fda 100644 --- a/examples/remote_error_propagation.py +++ b/examples/remote_error_propagation.py @@ -2,18 +2,30 @@ import trio import tractor -async def assert_err(): +async def assert_err() -> None: + ''' + Raise an assertion error in the current actor. + + ''' assert 0 -async def main(): +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 = [] + 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 eb4f03cd..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,15 +33,22 @@ 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 = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'rpc_server', enable_modules=[__name__], ) # 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 a9086b8c..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,13 +51,19 @@ 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 = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'quote_svc', enable_modules=[__name__], ) # run the client in a separate task which discovers # the daemon purely by its registered name. + tn: trio.Nursery async with trio.open_nursery() as tn: tn.start_soon(client_task) # explicit graceful teardown of the daemon. diff --git a/examples/service_discovery.py b/examples/service_discovery.py index 574ba019..5787359a 100644 --- a/examples/service_discovery.py +++ b/examples/service_discovery.py @@ -1,19 +1,29 @@ import trio import tractor -tractor.log.get_console_log("INFO") +tractor.log.get_console_log('INFO') -async def main(service_name): +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) - 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}' + ) - 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: + service_addr = actor_portal.chan.raddr + print(f'my_service is found at {service_addr}') await an.cancel() diff --git a/examples/streaming_broadcast_fanout.py b/examples/streaming_broadcast_fanout.py index 9f6900ce..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,11 +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 = await an.start_actor( + 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, @@ -67,9 +78,11 @@ async def main() -> None: ctx.open_stream() as stream, ): assert first == 5 + tn: trio.Nursery 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 2f979a00..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.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, @@ -46,7 +58,12 @@ async def main( ignore_special_cases: bool, 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 e7f895b7..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()` @@ -134,19 +139,26 @@ 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 c6352de5..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,11 +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 = await an.start_actor( + 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, @@ -72,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 93a17626..68b780ac 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -21,17 +21,22 @@ 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, int|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'], ) as an: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'uds_child', enable_modules=[__name__], ) @@ -51,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()