From 2a407be532ec7db4174f5a677d6c71810934ba25 Mon Sep 17 00:00:00 2001 From: Guillermo Rodriguez Date: Mon, 27 Jul 2020 14:55:37 -0300 Subject: [PATCH] Now passing additional initialization parameters through channel early after handshake. --- tractor/_actor.py | 5 +++++ tractor/_child.py | 51 +++++++++++++++++++++++++++++++++++++++++++++-- tractor/_entry.py | 21 ++++++++++++------- tractor/_spawn.py | 34 +++++++++++++++++-------------- 4 files changed, 87 insertions(+), 24 deletions(-) diff --git a/tractor/_actor.py b/tractor/_actor.py index 81f05cd..b52ef18 100644 --- a/tractor/_actor.py +++ b/tractor/_actor.py @@ -575,6 +575,11 @@ class Actor: await chan.connect() # initial handshake, report who we are, who they are await self._do_handshake(chan) + + self._parent_main_data = await chan.recv() + self.rpc_module_paths = await chan.recv() + self.statespace = await chan.recv() + except OSError: # failed to connect log.warning( f"Failed to connect to parent @ {parent_addr}," diff --git a/tractor/_child.py b/tractor/_child.py index b4d1d60..64c3054 100644 --- a/tractor/_child.py +++ b/tractor/_child.py @@ -1,6 +1,53 @@ import sys import trio -import cloudpickle +import argparse + +from ._actor import Actor +from ._entry import _trio_main + + +"""This is the "bootloader" for actors started using the native trio backend +added in #128 +""" + if __name__ == "__main__": - trio.run(cloudpickle.load(sys.stdin.buffer)) + + parser = argparse.ArgumentParser() + + parser.add_argument("name") + parser.add_argument("uid") + parser.add_argument("loglevel") + parser.add_argument("bind_addr") + parser.add_argument("parent_addr") + parser.add_argument("arbiter_addr") + + args = parser.parse_args() + + bind_addr = args.bind_addr.split(":") + bind_addr = (bind_addr[0], int(bind_addr[1])) + + parent_addr = args.parent_addr.split(":") + parent_addr = (parent_addr[0], int(parent_addr[1])) + + arbiter_addr = args.arbiter_addr.split(":") + arbiter_addr = (arbiter_addr[0], int(arbiter_addr[1])) + + if args.loglevel == "None": + loglevel = None + else: + loglevel = args.loglevel + + subactor = Actor( + args.name, + uid=args.uid, + loglevel=loglevel, + arbiter_addr=arbiter_addr + ) + subactor._spawn_method = "trio" + + _trio_main( + subactor, + bind_addr, + parent_addr=parent_addr + ) \ No newline at end of file diff --git a/tractor/_entry.py b/tractor/_entry.py index 1c26065..81b28d7 100644 --- a/tractor/_entry.py +++ b/tractor/_entry.py @@ -51,24 +51,31 @@ def _mp_main( log.info(f"Actor {actor.uid} terminated") -async def _trio_main( +def _trio_main( actor: 'Actor', accept_addr: Tuple[str, int], parent_addr: Tuple[str, int] = None ) -> None: """Entry point for a `trio_run_in_process` subactor. - - Here we don't need to call `trio.run()` since trip does that as - part of its subprocess startup sequence. """ if actor.loglevel is not None: log.info( f"Setting loglevel for {actor.uid} to {actor.loglevel}") get_console_log(actor.loglevel) - log.info(f"Started new trio process for {actor.uid}") + log.info( + f"Started {actor.uid}") _state._current_actor = actor - await actor._async_main(accept_addr, parent_addr=parent_addr) - log.info(f"Actor {actor.uid} terminated") + log.debug(f"parent_addr is {parent_addr}") + trio_main = partial( + actor._async_main, + accept_addr, + parent_addr=parent_addr + ) + try: + trio.run(trio_main) + except KeyboardInterrupt: + pass # handle it the same way trio does? + log.info(f"Actor {actor.uid} terminated") \ No newline at end of file diff --git a/tractor/_spawn.py b/tractor/_spawn.py index 8078c03..49baed6 100644 --- a/tractor/_spawn.py +++ b/tractor/_spawn.py @@ -10,7 +10,6 @@ from typing import Any, Dict, Optional from functools import partial import trio -import cloudpickle from trio_typing import TaskStatus from async_generator import aclosing, asynccontextmanager @@ -158,9 +157,11 @@ async def cancel_on_completion( @asynccontextmanager -async def run_in_process(subactor, async_fn, *args, **kwargs): - encoded_job = cloudpickle.dumps(partial(async_fn, *args, **kwargs)) - +async def spawn_subactor( + subactor: 'Actor', + accept_addr: Tuple[str, int], + parent_addr: Tuple[str, int] = None +): async with await trio.open_process( [ sys.executable, @@ -168,15 +169,14 @@ async def run_in_process(subactor, async_fn, *args, **kwargs): # Hardcode this (instead of using ``_child.__name__`` to avoid a # double import warning: https://stackoverflow.com/a/45070583 "tractor._child", - # This is merely an identifier for debugging purposes when - # viewing the process tree from the OS - str(subactor.uid), - ], - stdin=subprocess.PIPE, + subactor.name, + subactor.uid[1], + subactor.loglevel or "None", + f"{accept_addr[0]}:{accept_addr[1]}", + f"{parent_addr[0]}:{parent_addr[1]}", + f"{subactor._arb_addr[0]}:{subactor._arb_addr[1]}" + ] ) as proc: - - # send func object to call in child - await proc.stdin.send_all(encoded_job) yield proc @@ -201,9 +201,7 @@ async def new_proc( async with trio.open_nursery() as nursery: if use_trio_run_in_process or _spawn_method == 'trio': - async with run_in_process( - subactor, - _trio_main, + async with spawn_subactor( subactor, bind_addr, parent_addr, @@ -218,6 +216,12 @@ async def new_proc( portal = Portal(chan) actor_nursery._children[subactor.uid] = ( subactor, proc, portal) + + # send additional init params + await chan.send(subactor._parent_main_data) + await chan.send(subactor.rpc_module_paths) + await chan.send(subactor.statespace) + task_status.started(portal) # wait for ActorNursery.wait() to be called