Compare commits
No commits in common. "56fbed0be19d278c69f8044d69d9dafebf1cbea5" and "65bf9df5bad583bb30e09b2659781370268a6b64" have entirely different histories.
56fbed0be1
...
65bf9df5ba
|
|
@ -62,28 +62,6 @@ 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. 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
|
||||
-----------------
|
||||
|
||||
|
|
|
|||
|
|
@ -238,31 +238,9 @@ 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. 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``.
|
||||
|
||||
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.
|
||||
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.
|
||||
|
||||
Where to next?
|
||||
--------------
|
||||
|
|
@ -280,5 +258,3 @@ 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
|
||||
|
|
|
|||
|
|
@ -8,31 +8,29 @@ the_line = 'Hi my name is {}'
|
|||
tractor.log.get_console_log("INFO")
|
||||
|
||||
|
||||
async def hi() -> str:
|
||||
async def hi():
|
||||
return the_line.format(tractor.current_actor().name)
|
||||
|
||||
|
||||
async def say_hello(other_actor: str) -> str:
|
||||
portal: tractor.Portal
|
||||
async def say_hello(other_actor):
|
||||
async with tractor.wait_for_actor(other_actor) as portal:
|
||||
return await portal.run(hi)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""Main tractor entry point, the "master" process (for now
|
||||
acts as the "director").
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
print("Alright... Action!")
|
||||
|
||||
donny: tractor.Portal = await n.run_in_actor(
|
||||
donny = await n.run_in_actor(
|
||||
say_hello,
|
||||
name='donny',
|
||||
# arguments are always named
|
||||
other_actor='gretchen',
|
||||
)
|
||||
gretchen: tractor.Portal = await n.run_in_actor(
|
||||
gretchen = await n.run_in_actor(
|
||||
say_hello,
|
||||
name='gretchen',
|
||||
other_actor='donny',
|
||||
|
|
|
|||
|
|
@ -2,18 +2,17 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def cellar_door() -> str:
|
||||
async def cellar_door():
|
||||
assert not tractor.is_root_process()
|
||||
return "Dang that's beautiful"
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""The main ``tractor`` routine.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal: tractor.Portal = await n.run_in_actor(
|
||||
portal = await n.run_in_actor(
|
||||
cellar_door,
|
||||
name='some_linguist',
|
||||
)
|
||||
|
|
|
|||
|
|
@ -2,20 +2,19 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def movie_theatre_question() -> str:
|
||||
async def movie_theatre_question():
|
||||
"""A question asked in a dark theatre, in a tangent
|
||||
(errr, I mean different) process.
|
||||
"""
|
||||
return 'have you ever seen a portal?'
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""The main ``tractor`` routine.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal: tractor.Portal = await n.start_actor(
|
||||
portal = await n.start_actor(
|
||||
'frank',
|
||||
# enable the actor to run funcs from this current module
|
||||
enable_modules=[__name__],
|
||||
|
|
|
|||
|
|
@ -13,12 +13,11 @@ async def stream_forever() -> AsyncIterator[int]:
|
|||
await trio.sleep(0.01)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal: tractor.Portal = await n.start_actor(
|
||||
portal = await n.start_actor(
|
||||
'donny',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -26,7 +25,7 @@ async def main() -> None:
|
|||
# this async for loop streams values from the above
|
||||
# async generator running in a separate process
|
||||
async with portal.open_stream_from(stream_forever) as stream:
|
||||
count: int = 0
|
||||
count = 0
|
||||
async for letter in stream:
|
||||
print(letter)
|
||||
count += 1
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ async def open_ctx(
|
|||
assert first is None
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ async def name_error():
|
|||
getattr(doggypants) # noqa
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
'''
|
||||
Test breakpoint in a streaming actor.
|
||||
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@ async def breakpoint_forever():
|
|||
async def spawn_until(depth=0):
|
||||
""""A nested nursery that triggers another ``NameError``.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
if depth < 1:
|
||||
|
||||
|
|
@ -47,7 +46,7 @@ async def spawn_until(depth=0):
|
|||
|
||||
|
||||
# TODO: notes on the new boxed-relayed errors through proxy actors
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""The main ``tractor`` routine.
|
||||
|
||||
The process tree should look as approximately as follows when the debugger
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@ async def name_error():
|
|||
async def spawn_error():
|
||||
""""A nested nursery that triggers another ``NameError``.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
portal = await n.run_in_actor(
|
||||
name_error,
|
||||
|
|
@ -24,7 +23,7 @@ async def spawn_error():
|
|||
return await portal.result()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""The main ``tractor`` routine.
|
||||
|
||||
The process tree should look as approximately as follows:
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@ async def name_error():
|
|||
async def spawn_error():
|
||||
""""A nested nursery that triggers another ``NameError``.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
portal = await n.run_in_actor(
|
||||
name_error,
|
||||
|
|
@ -26,7 +25,7 @@ async def spawn_error():
|
|||
return await portal.result()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
"""The main ``tractor`` routine.
|
||||
|
||||
The process tree should look as approximately as follows:
|
||||
|
|
|
|||
|
|
@ -5,8 +5,7 @@ async def die():
|
|||
raise RuntimeError
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
tn: tractor.ActorNursery
|
||||
async def main():
|
||||
async with tractor.open_nursery() as tn:
|
||||
|
||||
debug_actor = await tn.start_actor(
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ async def name_error(
|
|||
raise
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
'''
|
||||
Test 3 `PdbREPL` entries:
|
||||
- one in the child due to manual `.post_mortem()`,
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
async with tractor.open_root_actor(
|
||||
debug_mode=True,
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
async with tractor.open_root_actor(
|
||||
debug_mode=True,
|
||||
):
|
||||
|
|
|
|||
|
|
@ -10,7 +10,6 @@ async def name_error():
|
|||
async def spawn_until(depth=0):
|
||||
""""A nested nursery that triggers another ``NameError``.
|
||||
"""
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
if depth < 1:
|
||||
# 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
|
||||
debugger first engages:
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
async with tractor.open_root_actor(
|
||||
debug_mode=True,
|
||||
loglevel='cancel',
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ async def key_error():
|
|||
return {}['doggy']
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
'''
|
||||
Root is fail-after-cancelled while blocking and child RPC fails
|
||||
simultaneously.
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ async def cancelled_before_pause(
|
|||
await pm_on_cancelled()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
) as n:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# like we do for loglevel/spawn-backend!
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ async def breakpoint_forever():
|
|||
await tractor.pause()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ async def name_error():
|
|||
getattr(doggypants) # noqa (on purpose)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
async with tractor.open_nursery(
|
||||
debug_mode=True,
|
||||
) as an:
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -9,14 +9,14 @@ from tractor import (
|
|||
|
||||
|
||||
# this is the first 2 actors, streamer_1 and streamer_2
|
||||
async def stream_data(seed: int):
|
||||
async def stream_data(seed):
|
||||
for i in range(seed):
|
||||
yield i
|
||||
await trio.sleep(0.0001) # trigger scheduler
|
||||
|
||||
|
||||
# this is the third actor; the aggregator
|
||||
async def aggregate(seed: int):
|
||||
async def aggregate(seed):
|
||||
'''
|
||||
Ensure that the two streams we receive match but only stream
|
||||
a single set of values to the parent.
|
||||
|
|
@ -28,7 +28,7 @@ async def aggregate(seed: int):
|
|||
for i in range(1, 3):
|
||||
|
||||
# fork/spawn call
|
||||
portal: Portal = await an.start_actor(
|
||||
portal = await an.start_actor(
|
||||
name=f'streamer_{i}',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -37,7 +37,7 @@ async def aggregate(seed: int):
|
|||
|
||||
send_chan, recv_chan = trio.open_memory_channel(500)
|
||||
|
||||
async def push_to_chan(portal: Portal, send_chan):
|
||||
async def push_to_chan(portal, send_chan):
|
||||
|
||||
# TODO: https://github.com/goodboy/tractor/issues/207
|
||||
async with send_chan:
|
||||
|
|
@ -49,7 +49,6 @@ async def aggregate(seed: int):
|
|||
print(f"FINISHED ITERATING {portal.channel.uid}")
|
||||
|
||||
# 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:
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ async def aio_echo_server(
|
|||
@tractor.context
|
||||
async def trio_to_aio_echo_server(
|
||||
ctx: tractor.Context,
|
||||
) -> None:
|
||||
):
|
||||
# this will block until the ``asyncio`` task sends a "first"
|
||||
# message.
|
||||
async with tractor.to_asyncio.open_channel_from(
|
||||
|
|
@ -48,11 +48,10 @@ async def trio_to_aio_echo_server(
|
|||
await stream.send(out)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
p: tractor.Portal = await n.start_actor(
|
||||
p = await n.start_actor(
|
||||
'aio_server',
|
||||
enable_modules=[__name__],
|
||||
infect_asyncio=True,
|
||||
|
|
|
|||
|
|
@ -33,16 +33,15 @@ async def main() -> None:
|
|||
rank = MPI.COMM_WORLD.Get_rank()
|
||||
print(f"[parent] rank={rank} pid={os.getpid()}", flush=True)
|
||||
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery(start_method='trio') as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
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: str = await portal.run(child_fn)
|
||||
result = await portal.run(child_fn)
|
||||
print(f"[parent] got: {result}", flush=True)
|
||||
await portal.cancel_actor()
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ import tractor
|
|||
log = tractor.log.get_logger('multiportal')
|
||||
|
||||
|
||||
async def stream_data(seed: int = 10):
|
||||
async def stream_data(seed=10):
|
||||
log.info("Starting stream task")
|
||||
|
||||
for i in range(seed):
|
||||
|
|
@ -13,10 +13,7 @@ async def stream_data(seed: int = 10):
|
|||
await trio.sleep(0) # trigger scheduler
|
||||
|
||||
|
||||
async def stream_from_portal(
|
||||
p: tractor.Portal,
|
||||
consumed: list,
|
||||
) -> None:
|
||||
async def stream_from_portal(p, consumed):
|
||||
|
||||
async with p.open_stream_from(stream_data) as stream:
|
||||
async for item in stream:
|
||||
|
|
@ -26,19 +23,14 @@ async def stream_from_portal(
|
|||
consumed.append(item)
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery(loglevel='info') as an:
|
||||
|
||||
p: tractor.Portal = await an.start_actor(
|
||||
'stream_boi',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
p = await an.start_actor('stream_boi', enable_modules=[__name__])
|
||||
|
||||
consumed: list = []
|
||||
consumed = []
|
||||
|
||||
n: trio.Nursery
|
||||
async with trio.open_nursery() as n:
|
||||
for i in range(2):
|
||||
n.start_soon(stream_from_portal, p, consumed)
|
||||
|
|
|
|||
|
|
@ -41,7 +41,6 @@ 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):
|
||||
|
|
@ -68,7 +67,6 @@ async def fan_out_squares(
|
|||
)
|
||||
|
||||
# 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(
|
||||
|
|
@ -85,9 +83,8 @@ async def fan_out_squares(
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
portal = await an.start_actor(
|
||||
'supervisor',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ PRIMES = [
|
|||
]
|
||||
|
||||
|
||||
async def is_prime(n: int) -> bool:
|
||||
async def is_prime(n):
|
||||
if n < 2:
|
||||
return False
|
||||
if n == 2:
|
||||
|
|
@ -47,7 +47,7 @@ async def is_prime(n: int) -> bool:
|
|||
|
||||
|
||||
@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
|
||||
known "worker pool" seems to be the defacto "but, I want this
|
||||
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
|
||||
the context.
|
||||
"""
|
||||
tn: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as tn:
|
||||
|
||||
portals: list[tractor.Portal] = []
|
||||
portals = []
|
||||
snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES))
|
||||
|
||||
for i in range(workers):
|
||||
|
|
@ -78,14 +77,9 @@ async def worker_pool(workers: int = 4):
|
|||
) -> list[bool]:
|
||||
|
||||
# define an async (local) task to collect results from workers
|
||||
async def send_result(
|
||||
func: Callable,
|
||||
value: int,
|
||||
portal: tractor.Portal,
|
||||
):
|
||||
async def send_result(func, value, portal):
|
||||
await snd_chan.send((value, await portal.run(func, n=value)))
|
||||
|
||||
n: trio.Nursery
|
||||
async with trio.open_nursery() as n:
|
||||
|
||||
for value, portal in zip(sequence, itertools.cycle(portals)):
|
||||
|
|
@ -107,7 +101,7 @@ async def worker_pool(workers: int = 4):
|
|||
await tn.cancel()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
async with worker_pool() as actor_map:
|
||||
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ import tractor
|
|||
import trio
|
||||
|
||||
|
||||
async def burn_cpu() -> int:
|
||||
async def burn_cpu():
|
||||
|
||||
pid = os.getpid()
|
||||
|
||||
|
|
@ -23,18 +23,17 @@ async def burn_cpu() -> int:
|
|||
return os.getpid()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
|
||||
n: tractor.ActorNursery
|
||||
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
|
||||
await burn_cpu()
|
||||
|
||||
# wait on result from target function
|
||||
pid: int = await portal.wait_for_result()
|
||||
pid = await portal.wait_for_result()
|
||||
|
||||
# end of nursery block
|
||||
print(f"Collected subproc {pid}")
|
||||
|
|
|
|||
|
|
@ -9,13 +9,12 @@ async def sleepy_jane() -> None:
|
|||
await trio.sleep_forever()
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
async def main():
|
||||
'''
|
||||
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 (
|
||||
|
|
|
|||
|
|
@ -2,14 +2,13 @@ import trio
|
|||
import tractor
|
||||
|
||||
|
||||
async def assert_err() -> None:
|
||||
async def assert_err():
|
||||
assert 0
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
n: tractor.ActorNursery
|
||||
async def main():
|
||||
async with tractor.open_nursery() as n:
|
||||
real_actors: list[tractor.Portal] = []
|
||||
real_actors = []
|
||||
for i in range(3):
|
||||
real_actors.append(await n.start_actor(
|
||||
f'actor_{i}',
|
||||
|
|
|
|||
|
|
@ -31,10 +31,9 @@ async def simple_rpc(
|
|||
|
||||
async def main() -> None:
|
||||
|
||||
n: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as n:
|
||||
|
||||
portal: tractor.Portal = await n.start_actor(
|
||||
portal = await n.start_actor(
|
||||
'rpc_server',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -49,15 +49,13 @@ async def client_task() -> None:
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -4,17 +4,14 @@ import tractor
|
|||
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:
|
||||
await an.start_actor(service_name)
|
||||
|
||||
portal: tractor.Portal
|
||||
async with tractor.get_registry() as portal:
|
||||
print(f"Registrar is listening on {portal.channel}")
|
||||
|
||||
sockaddr: tractor.Portal
|
||||
async with tractor.wait_for_actor(service_name) as sockaddr:
|
||||
print(f"my_service is found at {sockaddr}")
|
||||
|
||||
|
|
|
|||
|
|
@ -54,9 +54,8 @@ async def consume(
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
portal = await an.start_actor(
|
||||
'ticker',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
@ -68,7 +67,6 @@ 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.
|
||||
|
|
|
|||
|
|
@ -32,8 +32,8 @@ async def acquire_singleton_lock(
|
|||
|
||||
|
||||
async def hold_lock_forever(
|
||||
task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED,
|
||||
) -> None:
|
||||
task_status=trio.TASK_STATUS_IGNORED
|
||||
):
|
||||
async with (
|
||||
tractor.trionics.maybe_raise_from_masking_exc(),
|
||||
acquire_singleton_lock() as lock,
|
||||
|
|
@ -46,7 +46,7 @@ async def main(
|
|||
ignore_special_cases: bool,
|
||||
loglevel: str = 'info',
|
||||
debug_mode: bool = True,
|
||||
) -> None:
|
||||
):
|
||||
async with (
|
||||
trio.open_nursery() as tn,
|
||||
|
||||
|
|
|
|||
|
|
@ -134,7 +134,7 @@ async def main(
|
|||
|
||||
raise_unmasked: bool = False,
|
||||
loglevel: str = 'info',
|
||||
) -> None:
|
||||
):
|
||||
tractor.log.get_console_log(level=loglevel)
|
||||
|
||||
# the `.aclose()` being checkpoints on these
|
||||
|
|
|
|||
|
|
@ -59,9 +59,8 @@ async def point_doubler(
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery() as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
portal = await an.start_actor(
|
||||
'point_doubler',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
|
|||
|
|
@ -27,11 +27,10 @@ async def report_addr() -> str:
|
|||
|
||||
|
||||
async def main() -> None:
|
||||
an: tractor.ActorNursery
|
||||
async with tractor.open_nursery(
|
||||
enable_transports=['uds'],
|
||||
) as an:
|
||||
portal: tractor.Portal = await an.start_actor(
|
||||
portal = await an.start_actor(
|
||||
'uds_child',
|
||||
enable_modules=[__name__],
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in New Issue