Compare commits

..

No commits in common. "56fbed0be19d278c69f8044d69d9dafebf1cbea5" and "65bf9df5bad583bb30e09b2659781370268a6b64" have entirely different histories.

40 changed files with 71 additions and 282 deletions

View File

@ -62,28 +62,6 @@ the one-and-only registrar; boot then fails loudly with a
``RuntimeError`` if some other process already bound the registry ``RuntimeError`` if some other process already bound the registry
socket(s). 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. Boot
a bare ``tractor.run_daemon([], registry_addrs=[...])`` (a root
actor that does nothing but hold the registry), point your app
tree at the same ``registry_addrs``, and every actor discovers
through that *external* registrar instead of a tree-local one:
.. literalinclude:: ../../examples/dedicated_registrar.py
:caption: examples/dedicated_registrar.py
:language: python
This is the "registrar as a subsystem, not the 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 Looking up actors
----------------- -----------------

View File

@ -238,31 +238,9 @@ Toward capability-based msging
The ``pld_spec`` + codec-hook layer is the foundation for the The ``pld_spec`` + codec-hook layer is the foundation for the
long-game: **capability-based msging** where each dialog's long-game: **capability-based msging** where each dialog's
type contract doubles as a capability grant, negotiated as part type contract doubles as a capability grant, negotiated as part
of the protocol itself. The epic is tracked in `#196`_ (evolving of the protocol itself. That work is tracked in `#196`_ (with the
the original typed-proto work in `#36`_), and the most recent original typed-proto epic in `#36`_); if strongly-typed
concrete step is `#365`_ — driving the whole ``pld_spec`` off distributed systems get you going, we'd love your input.
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``.
On the codec-hook side, the ``enc_hook``/``dec_hook`` pair is
today only reachable via ``tractor.msg._ops``; a public *factory*
API for them is drafted in `#376`_ (from
`@guilledk <https://github.com/guilledk>`_, on the
`auto_codecs <https://github.com/goodboy/tractor/tree/auto_codecs>`_
branch) — the likely 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? Where to next?
-------------- --------------
@ -280,5 +258,3 @@ Where to next?
.. _(un)protocol: https://zguide.zeromq.org/docs/chapter7/#Unprotocols .. _(un)protocol: https://zguide.zeromq.org/docs/chapter7/#Unprotocols
.. _#196: https://github.com/goodboy/tractor/issues/196 .. _#196: https://github.com/goodboy/tractor/issues/196
.. _#36: https://github.com/goodboy/tractor/issues/36 .. _#36: https://github.com/goodboy/tractor/issues/36
.. _#365: https://github.com/goodboy/tractor/issues/365
.. _#376: https://github.com/goodboy/tractor/pull/376

View File

@ -8,31 +8,29 @@ the_line = 'Hi my name is {}'
tractor.log.get_console_log("INFO") tractor.log.get_console_log("INFO")
async def hi() -> str: async def hi():
return the_line.format(tractor.current_actor().name) return the_line.format(tractor.current_actor().name)
async def say_hello(other_actor: str) -> str: async def say_hello(other_actor):
portal: tractor.Portal
async with tractor.wait_for_actor(other_actor) as portal: async with tractor.wait_for_actor(other_actor) as portal:
return await portal.run(hi) return await portal.run(hi)
async def main() -> None: async def main():
"""Main tractor entry point, the "master" process (for now """Main tractor entry point, the "master" process (for now
acts as the "director"). acts as the "director").
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
print("Alright... Action!") print("Alright... Action!")
donny: tractor.Portal = await n.run_in_actor( donny = await n.run_in_actor(
say_hello, say_hello,
name='donny', name='donny',
# arguments are always named # arguments are always named
other_actor='gretchen', other_actor='gretchen',
) )
gretchen: tractor.Portal = await n.run_in_actor( gretchen = await n.run_in_actor(
say_hello, say_hello,
name='gretchen', name='gretchen',
other_actor='donny', other_actor='donny',

View File

@ -2,18 +2,17 @@ import trio
import tractor import tractor
async def cellar_door() -> str: async def cellar_door():
assert not tractor.is_root_process() assert not tractor.is_root_process()
return "Dang that's beautiful" return "Dang that's beautiful"
async def main() -> None: async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal: tractor.Portal = await n.run_in_actor( portal = await n.run_in_actor(
cellar_door, cellar_door,
name='some_linguist', name='some_linguist',
) )

View File

@ -2,20 +2,19 @@ import trio
import tractor import tractor
async def movie_theatre_question() -> str: async def movie_theatre_question():
"""A question asked in a dark theatre, in a tangent """A question asked in a dark theatre, in a tangent
(errr, I mean different) process. (errr, I mean different) process.
""" """
return 'have you ever seen a portal?' return 'have you ever seen a portal?'
async def main() -> None: async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal: tractor.Portal = await n.start_actor( portal = await n.start_actor(
'frank', 'frank',
# enable the actor to run funcs from this current module # enable the actor to run funcs from this current module
enable_modules=[__name__], enable_modules=[__name__],

View File

@ -13,12 +13,11 @@ async def stream_forever() -> AsyncIterator[int]:
await trio.sleep(0.01) await trio.sleep(0.01)
async def main() -> None: async def main():
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal: tractor.Portal = await n.start_actor( portal = await n.start_actor(
'donny', 'donny',
enable_modules=[__name__], enable_modules=[__name__],
) )
@ -26,7 +25,7 @@ async def main() -> None:
# this async for loop streams values from the above # this async for loop streams values from the above
# async generator running in a separate process # async generator running in a separate process
async with portal.open_stream_from(stream_forever) as stream: async with portal.open_stream_from(stream_forever) as stream:
count: int = 0 count = 0
async for letter in stream: async for letter in stream:
print(letter) print(letter)
count += 1 count += 1

View File

@ -35,7 +35,7 @@ async def open_ctx(
assert first is None assert first is None
async def main() -> None: async def main():
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,

View File

@ -20,7 +20,7 @@ async def name_error():
getattr(doggypants) # noqa getattr(doggypants) # noqa
async def main() -> None: async def main():
''' '''
Test breakpoint in a streaming actor. Test breakpoint in a streaming actor.

View File

@ -21,7 +21,6 @@ async def breakpoint_forever():
async def spawn_until(depth=0): async def spawn_until(depth=0):
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
if depth < 1: if depth < 1:
@ -47,7 +46,7 @@ async def spawn_until(depth=0):
# TODO: notes on the new boxed-relayed errors through proxy actors # TODO: notes on the new boxed-relayed errors through proxy actors
async def main() -> None: async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
The process tree should look as approximately as follows when the debugger The process tree should look as approximately as follows when the debugger

View File

@ -15,7 +15,6 @@ async def name_error():
async def spawn_error(): async def spawn_error():
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal = await n.run_in_actor( portal = await n.run_in_actor(
name_error, name_error,
@ -24,7 +23,7 @@ async def spawn_error():
return await portal.result() return await portal.result()
async def main() -> None: async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
The process tree should look as approximately as follows: The process tree should look as approximately as follows:

View File

@ -17,7 +17,6 @@ async def name_error():
async def spawn_error(): async def spawn_error():
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal = await n.run_in_actor( portal = await n.run_in_actor(
name_error, name_error,
@ -26,7 +25,7 @@ async def spawn_error():
return await portal.result() return await portal.result()
async def main() -> None: async def main():
"""The main ``tractor`` routine. """The main ``tractor`` routine.
The process tree should look as approximately as follows: The process tree should look as approximately as follows:

View File

@ -5,8 +5,7 @@ async def die():
raise RuntimeError raise RuntimeError
async def main() -> None: async def main():
tn: tractor.ActorNursery
async with tractor.open_nursery() as tn: async with tractor.open_nursery() as tn:
debug_actor = await tn.start_actor( debug_actor = await tn.start_actor(

View File

@ -18,7 +18,7 @@ async def name_error(
raise raise
async def main() -> None: async def main():
''' '''
Test 3 `PdbREPL` entries: Test 3 `PdbREPL` entries:
- one in the child due to manual `.post_mortem()`, - one in the child due to manual `.post_mortem()`,

View File

@ -2,7 +2,7 @@ import trio
import tractor import tractor
async def main() -> None: async def main():
async with tractor.open_root_actor( async with tractor.open_root_actor(
debug_mode=True, debug_mode=True,

View File

@ -2,7 +2,7 @@ import trio
import tractor import tractor
async def main() -> None: async def main():
async with tractor.open_root_actor( async with tractor.open_root_actor(
debug_mode=True, debug_mode=True,
): ):

View File

@ -10,7 +10,6 @@ async def name_error():
async def spawn_until(depth=0): async def spawn_until(depth=0):
""""A nested nursery that triggers another ``NameError``. """"A nested nursery that triggers another ``NameError``.
""" """
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
if depth < 1: if depth < 1:
# await n.run_in_actor('breakpoint_forever', breakpoint_forever) # await n.run_in_actor('breakpoint_forever', breakpoint_forever)
@ -24,7 +23,7 @@ async def spawn_until(depth=0):
) )
async def main() -> None: async def main():
''' '''
The process tree should look as approximately as follows when the The process tree should look as approximately as follows when the
debugger first engages: debugger first engages:

View File

@ -2,7 +2,7 @@ import trio
import tractor import tractor
async def main() -> None: async def main():
async with tractor.open_root_actor( async with tractor.open_root_actor(
debug_mode=True, debug_mode=True,
loglevel='cancel', loglevel='cancel',

View File

@ -7,7 +7,7 @@ async def key_error():
return {}['doggy'] return {}['doggy']
async def main() -> None: async def main():
''' '''
Root is fail-after-cancelled while blocking and child RPC fails Root is fail-after-cancelled while blocking and child RPC fails
simultaneously. simultaneously.

View File

@ -71,7 +71,7 @@ async def cancelled_before_pause(
await pm_on_cancelled() await pm_on_cancelled()
async def main() -> None: async def main():
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,
) as n: ) as n:

View File

@ -34,7 +34,7 @@ async def just_bp(
async def main() -> None: async def main():
# !TODO, parametrize the --tpt-proto={key} with osenv vars just # !TODO, parametrize the --tpt-proto={key} with osenv vars just
# like we do for loglevel/spawn-backend! # like we do for loglevel/spawn-backend!

View File

@ -12,7 +12,7 @@ async def breakpoint_forever():
await tractor.pause() await tractor.pause()
async def main() -> None: async def main():
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,

View File

@ -6,7 +6,7 @@ async def name_error():
getattr(doggypants) # noqa (on purpose) getattr(doggypants) # noqa (on purpose)
async def main() -> None: async def main():
async with tractor.open_nursery( async with tractor.open_nursery(
debug_mode=True, debug_mode=True,
) as an: ) as an:

View File

@ -1,122 +0,0 @@
'''
Run a *dedicated* registrar as its own standalone process decoupled
from your app's root actor — and discover a service *through* it.
Normally the registrar **is** the root actor of your tree. Here we
instead boot a separate `tractor.run_daemon([], registry_addrs=[...])`
process whose *sole* job is to be the registry, then point our app
tree at it via `registry_addrs`. Because a registrar is already
reachable at that addr, our app's root actor does NOT become one — it
registers with (and discovers through) the external daemon. That's the
"registrar as a subsystem, not the root actor" pattern.
NB: `enable_transports` is single-proto per-runtime today (see
`tractor._root`), so this demos one transport; a genuinely
multi-backend registrar (and spawning one as a *sub*actor of a shared
tree) are future runtime work see the #472 follow-ups.
'''
from contextlib import suppress
import signal
import socket
import subprocess
import sys
import time
import trio
import tractor
# the fixed addr the dedicated registrar binds and everyone points at.
REG_ADDR: tuple[str, int] = ('127.0.0.1', 1717)
def _wait_registrar_ready(
addr: tuple[str, int],
proc: subprocess.Popen,
deadline: float = 10.0,
) -> None:
'''
Active-poll the registrar's bind addr until it accepts a
connection (proving it's booted + listening), bailing early if
the daemon proc dies during startup.
'''
end: float = time.monotonic() + deadline
while time.monotonic() < end:
if proc.poll() is not None:
raise RuntimeError(
f'registrar died on startup (rc={proc.returncode})'
)
with suppress(OSError):
with socket.create_connection(addr, timeout=0.1):
return
time.sleep(0.05)
raise TimeoutError(f'registrar never came up @ {addr}')
async def greet() -> str:
'''A trivial service task any peer can RPC by name.'''
return f'hello from {tractor.current_actor().name}!'
async def app() -> None:
'''
Point our app tree at the EXTERNAL registrar (not its own root)
via `registry_addrs`, register a named service, then discover +
RPC it purely by name.
'''
an: tractor.ActorNursery
async with tractor.open_nursery(
registry_addrs=[REG_ADDR],
enable_transports=['tcp'],
) as an:
# this subactor registers with the DEDICATED registrar @
# REG_ADDR (our root is a plain peer, not the registry).
await an.start_actor(
'greeter',
enable_modules=[__name__],
)
# discover it *through the external registrar*, by name only.
portal: tractor.Portal
async with tractor.wait_for_actor('greeter') as portal:
print(f'found `greeter` via dedicated registrar @ {REG_ADDR}')
print(await portal.run(greet))
await an.cancel()
def main() -> None:
# boot the dedicated registrar as its own process/tree: an empty
# `enable_modules` `run_daemon()` is just a root actor that does
# nothing but hold + serve the registry.
code: str = (
'import tractor; '
f'tractor.run_daemon([], registry_addrs={[REG_ADDR]!r}, '
"enable_transports=['tcp'], loglevel='error')"
)
registrar: subprocess.Popen = subprocess.Popen(
[sys.executable, '-c', code],
# the registry is a quiet background service; hush its logs +
# expected SIGINT-teardown traceback so the demo output stays
# focused on the discovery flow.
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
try:
_wait_registrar_ready(REG_ADDR, registrar)
print(
f'dedicated registrar up @ {REG_ADDR} '
f'(pid {registrar.pid})'
)
trio.run(app)
finally:
# graceful SIGINT teardown of the standalone registrar.
registrar.send_signal(signal.SIGINT)
with suppress(subprocess.TimeoutExpired):
registrar.wait(timeout=10)
print('dedicated registrar shut down')
if __name__ == '__main__':
main()

View File

@ -9,14 +9,14 @@ from tractor import (
# this is the first 2 actors, streamer_1 and streamer_2 # this is the first 2 actors, streamer_1 and streamer_2
async def stream_data(seed: int): async def stream_data(seed):
for i in range(seed): for i in range(seed):
yield i yield i
await trio.sleep(0.0001) # trigger scheduler await trio.sleep(0.0001) # trigger scheduler
# this is the third actor; the aggregator # this is the third actor; the aggregator
async def aggregate(seed: int): async def aggregate(seed):
''' '''
Ensure that the two streams we receive match but only stream Ensure that the two streams we receive match but only stream
a single set of values to the parent. a single set of values to the parent.
@ -28,7 +28,7 @@ async def aggregate(seed: int):
for i in range(1, 3): for i in range(1, 3):
# fork/spawn call # fork/spawn call
portal: Portal = await an.start_actor( portal = await an.start_actor(
name=f'streamer_{i}', name=f'streamer_{i}',
enable_modules=[__name__], enable_modules=[__name__],
) )
@ -37,7 +37,7 @@ async def aggregate(seed: int):
send_chan, recv_chan = trio.open_memory_channel(500) send_chan, recv_chan = trio.open_memory_channel(500)
async def push_to_chan(portal: Portal, send_chan): async def push_to_chan(portal, send_chan):
# TODO: https://github.com/goodboy/tractor/issues/207 # TODO: https://github.com/goodboy/tractor/issues/207
async with send_chan: async with send_chan:
@ -49,7 +49,6 @@ async def aggregate(seed: int):
print(f"FINISHED ITERATING {portal.channel.uid}") print(f"FINISHED ITERATING {portal.channel.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: async with trio.open_nursery() as n:
for portal in portals: for portal in portals:

View File

@ -28,7 +28,7 @@ async def aio_echo_server(
@tractor.context @tractor.context
async def trio_to_aio_echo_server( async def trio_to_aio_echo_server(
ctx: tractor.Context, ctx: tractor.Context,
) -> None: ):
# this will block until the ``asyncio`` task sends a "first" # this will block until the ``asyncio`` task sends a "first"
# message. # message.
async with tractor.to_asyncio.open_channel_from( async with tractor.to_asyncio.open_channel_from(
@ -48,11 +48,10 @@ async def trio_to_aio_echo_server(
await stream.send(out) await stream.send(out)
async def main() -> None: async def main():
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
p: tractor.Portal = await n.start_actor( p = await n.start_actor(
'aio_server', 'aio_server',
enable_modules=[__name__], enable_modules=[__name__],
infect_asyncio=True, infect_asyncio=True,

View File

@ -33,16 +33,15 @@ async def main() -> None:
rank = MPI.COMM_WORLD.Get_rank() rank = MPI.COMM_WORLD.Get_rank()
print(f"[parent] rank={rank} pid={os.getpid()}", flush=True) print(f"[parent] rank={rank} pid={os.getpid()}", flush=True)
an: tractor.ActorNursery
async with tractor.open_nursery(start_method='trio') as an: async with tractor.open_nursery(start_method='trio') as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'mpi-child', 'mpi-child',
enable_modules=[child_fn.__module__], enable_modules=[child_fn.__module__],
# Without this the child replays __main__, which # Without this the child replays __main__, which
# re-imports mpi4py and crashes on MPI_Init. # re-imports mpi4py and crashes on MPI_Init.
inherit_parent_main=False, inherit_parent_main=False,
) )
result: str = await portal.run(child_fn) result = await portal.run(child_fn)
print(f"[parent] got: {result}", flush=True) print(f"[parent] got: {result}", flush=True)
await portal.cancel_actor() await portal.cancel_actor()

View File

@ -5,7 +5,7 @@ import tractor
log = tractor.log.get_logger('multiportal') log = tractor.log.get_logger('multiportal')
async def stream_data(seed: int = 10): async def stream_data(seed=10):
log.info("Starting stream task") log.info("Starting stream task")
for i in range(seed): for i in range(seed):
@ -13,10 +13,7 @@ async def stream_data(seed: int = 10):
await trio.sleep(0) # trigger scheduler await trio.sleep(0) # trigger scheduler
async def stream_from_portal( async def stream_from_portal(p, consumed):
p: tractor.Portal,
consumed: list,
) -> None:
async with p.open_stream_from(stream_data) as stream: async with p.open_stream_from(stream_data) as stream:
async for item in stream: async for item in stream:
@ -26,19 +23,14 @@ async def stream_from_portal(
consumed.append(item) consumed.append(item)
async def main() -> None: async def main():
an: tractor.ActorNursery
async with tractor.open_nursery(loglevel='info') as an: async with tractor.open_nursery(loglevel='info') as an:
p: tractor.Portal = await an.start_actor( p = await an.start_actor('stream_boi', enable_modules=[__name__])
'stream_boi',
enable_modules=[__name__],
)
consumed: list = [] consumed = []
n: trio.Nursery
async with trio.open_nursery() as n: async with trio.open_nursery() as n:
for i in range(2): for i in range(2):
n.start_soon(stream_from_portal, p, consumed) n.start_soon(stream_from_portal, p, consumed)

View File

@ -41,7 +41,6 @@ async def fan_out_squares(
aggregated squares to our parent. aggregated squares to our parent.
''' '''
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portals: list[tractor.Portal] = [] portals: list[tractor.Portal] = []
for i in (1, 2): for i in (1, 2):
@ -68,7 +67,6 @@ async def fan_out_squares(
) )
# fan out one sub-RPC per input val, concurrently. # fan out one sub-RPC per input val, concurrently.
tn: trio.Nursery
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
for i, x in enumerate(vals): for i, x in enumerate(vals):
tn.start_soon( tn.start_soon(
@ -85,9 +83,8 @@ async def fan_out_squares(
async def main() -> None: async def main() -> None:
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'supervisor', 'supervisor',
enable_modules=[__name__], enable_modules=[__name__],
) )

View File

@ -31,7 +31,7 @@ PRIMES = [
] ]
async def is_prime(n: int) -> bool: async def is_prime(n):
if n < 2: if n < 2:
return False return False
if n == 2: if n == 2:
@ -47,7 +47,7 @@ async def is_prime(n: int) -> bool:
@acm @acm
async def worker_pool(workers: int = 4): async def worker_pool(workers=4):
"""Though it's a trivial special case for ``tractor``, the well """Though it's a trivial special case for ``tractor``, the well
known "worker pool" seems to be the defacto "but, I want this known "worker pool" seems to be the defacto "but, I want this
process pattern!" for most parallelism pilgrims. process pattern!" for most parallelism pilgrims.
@ -55,10 +55,9 @@ async def worker_pool(workers: int = 4):
Yes, the workers stay alive (and ready for work) until you close Yes, the workers stay alive (and ready for work) until you close
the context. the context.
""" """
tn: tractor.ActorNursery
async with tractor.open_nursery() as tn: async with tractor.open_nursery() as tn:
portals: list[tractor.Portal] = [] portals = []
snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES)) snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES))
for i in range(workers): for i in range(workers):
@ -78,14 +77,9 @@ async def worker_pool(workers: int = 4):
) -> list[bool]: ) -> list[bool]:
# define an async (local) task to collect results from workers # define an async (local) task to collect results from workers
async def send_result( async def send_result(func, value, portal):
func: Callable,
value: int,
portal: tractor.Portal,
):
await snd_chan.send((value, await portal.run(func, n=value))) await snd_chan.send((value, await portal.run(func, n=value)))
n: trio.Nursery
async with trio.open_nursery() as n: async with trio.open_nursery() as n:
for value, portal in zip(sequence, itertools.cycle(portals)): for value, portal in zip(sequence, itertools.cycle(portals)):
@ -107,7 +101,7 @@ async def worker_pool(workers: int = 4):
await tn.cancel() await tn.cancel()
async def main() -> None: async def main():
async with worker_pool() as actor_map: async with worker_pool() as actor_map:

View File

@ -12,7 +12,7 @@ import tractor
import trio import trio
async def burn_cpu() -> int: async def burn_cpu():
pid = os.getpid() pid = os.getpid()
@ -23,18 +23,17 @@ async def burn_cpu() -> int:
return os.getpid() return os.getpid()
async def main() -> None: async def main():
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal: tractor.Portal = await n.run_in_actor(burn_cpu) portal = await n.run_in_actor(burn_cpu)
# burn rubber in the parent too # burn rubber in the parent too
await burn_cpu() await burn_cpu()
# wait on result from target function # wait on result from target function
pid: int = await portal.wait_for_result() pid = await portal.wait_for_result()
# end of nursery block # end of nursery block
print(f"Collected subproc {pid}") print(f"Collected subproc {pid}")

View File

@ -9,13 +9,12 @@ async def sleepy_jane() -> None:
await trio.sleep_forever() await trio.sleep_forever()
async def main() -> None: async def main():
''' '''
Spawn a flat actor cluster, with one process per detected core. Spawn a flat actor cluster, with one process per detected core.
''' '''
portal_map: dict[str, tractor.Portal] portal_map: dict[str, tractor.Portal]
tn: trio.Nursery
# look at this hip new syntax! # look at this hip new syntax!
async with ( async with (

View File

@ -2,14 +2,13 @@ import trio
import tractor import tractor
async def assert_err() -> None: async def assert_err():
assert 0 assert 0
async def main() -> None: async def main():
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
real_actors: list[tractor.Portal] = [] real_actors = []
for i in range(3): for i in range(3):
real_actors.append(await n.start_actor( real_actors.append(await n.start_actor(
f'actor_{i}', f'actor_{i}',

View File

@ -31,10 +31,9 @@ async def simple_rpc(
async def main() -> None: async def main() -> None:
n: tractor.ActorNursery
async with tractor.open_nursery() as n: async with tractor.open_nursery() as n:
portal: tractor.Portal = await n.start_actor( portal = await n.start_actor(
'rpc_server', 'rpc_server',
enable_modules=[__name__], enable_modules=[__name__],
) )

View File

@ -49,15 +49,13 @@ async def client_task() -> None:
async def main() -> None: async def main() -> None:
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'quote_svc', 'quote_svc',
enable_modules=[__name__], enable_modules=[__name__],
) )
# run the client in a separate task which discovers # run the client in a separate task which discovers
# the daemon purely by its registered name. # the daemon purely by its registered name.
tn: trio.Nursery
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
tn.start_soon(client_task) tn.start_soon(client_task)
# explicit graceful teardown of the daemon. # explicit graceful teardown of the daemon.

View File

@ -4,17 +4,14 @@ import tractor
tractor.log.get_console_log("INFO") tractor.log.get_console_log("INFO")
async def main(service_name: str) -> None: async def main(service_name):
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
await an.start_actor(service_name) await an.start_actor(service_name)
portal: tractor.Portal
async with tractor.get_registry() as portal: async with tractor.get_registry() as portal:
print(f"Registrar is listening on {portal.channel}") print(f"Registrar is listening on {portal.channel}")
sockaddr: tractor.Portal
async with tractor.wait_for_actor(service_name) as sockaddr: async with tractor.wait_for_actor(service_name) as sockaddr:
print(f"my_service is found at {sockaddr}") print(f"my_service is found at {sockaddr}")

View File

@ -54,9 +54,8 @@ async def consume(
async def main() -> None: async def main() -> None:
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'ticker', 'ticker',
enable_modules=[__name__], enable_modules=[__name__],
) )
@ -68,7 +67,6 @@ async def main() -> None:
ctx.open_stream() as stream, ctx.open_stream() as stream,
): ):
assert first == 5 assert first == 5
tn: trio.Nursery
async with trio.open_nursery() as tn: async with trio.open_nursery() as tn:
# use `.start()` so each consumer is known # use `.start()` so each consumer is known
# to be subscribed before the ticks flow. # to be subscribed before the ticks flow.

View File

@ -32,8 +32,8 @@ async def acquire_singleton_lock(
async def hold_lock_forever( async def hold_lock_forever(
task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, task_status=trio.TASK_STATUS_IGNORED
) -> None: ):
async with ( async with (
tractor.trionics.maybe_raise_from_masking_exc(), tractor.trionics.maybe_raise_from_masking_exc(),
acquire_singleton_lock() as lock, acquire_singleton_lock() as lock,
@ -46,7 +46,7 @@ async def main(
ignore_special_cases: bool, ignore_special_cases: bool,
loglevel: str = 'info', loglevel: str = 'info',
debug_mode: bool = True, debug_mode: bool = True,
) -> None: ):
async with ( async with (
trio.open_nursery() as tn, trio.open_nursery() as tn,

View File

@ -134,7 +134,7 @@ async def main(
raise_unmasked: bool = False, raise_unmasked: bool = False,
loglevel: str = 'info', loglevel: str = 'info',
) -> None: ):
tractor.log.get_console_log(level=loglevel) tractor.log.get_console_log(level=loglevel)
# the `.aclose()` being checkpoints on these # the `.aclose()` being checkpoints on these

View File

@ -59,9 +59,8 @@ async def point_doubler(
async def main() -> None: async def main() -> None:
an: tractor.ActorNursery
async with tractor.open_nursery() as an: async with tractor.open_nursery() as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'point_doubler', 'point_doubler',
enable_modules=[__name__], enable_modules=[__name__],
) )

View File

@ -27,11 +27,10 @@ async def report_addr() -> str:
async def main() -> None: async def main() -> None:
an: tractor.ActorNursery
async with tractor.open_nursery( async with tractor.open_nursery(
enable_transports=['uds'], enable_transports=['uds'],
) as an: ) as an:
portal: tractor.Portal = await an.start_actor( portal = await an.start_actor(
'uds_child', 'uds_child',
enable_modules=[__name__], enable_modules=[__name__],
) )