forked from goodboy/tractor
Added named arguments to child init, and now passing less of them.
parent
e5dbf14ec3
commit
ef053eb070
|
@ -171,11 +171,6 @@ class Actor:
|
||||||
_root_nursery: trio.Nursery
|
_root_nursery: trio.Nursery
|
||||||
_server_nursery: trio.Nursery
|
_server_nursery: trio.Nursery
|
||||||
|
|
||||||
# marked by the process spawning backend at startup
|
|
||||||
# will be None for the parent most process started manually
|
|
||||||
# by the user (currently called the "arbiter")
|
|
||||||
_spawn_method: Optional[str] = None
|
|
||||||
|
|
||||||
# Information about `__main__` from parent
|
# Information about `__main__` from parent
|
||||||
_parent_main_data: Dict[str, str]
|
_parent_main_data: Dict[str, str]
|
||||||
|
|
||||||
|
@ -187,6 +182,7 @@ class Actor:
|
||||||
uid: str = None,
|
uid: str = None,
|
||||||
loglevel: str = None,
|
loglevel: str = None,
|
||||||
arbiter_addr: Optional[Tuple[str, int]] = None,
|
arbiter_addr: Optional[Tuple[str, int]] = None,
|
||||||
|
spawn_method: Optional[str] = None
|
||||||
) -> None:
|
) -> None:
|
||||||
"""This constructor is called in the parent actor **before** the spawning
|
"""This constructor is called in the parent actor **before** the spawning
|
||||||
phase (aka before a new process is executed).
|
phase (aka before a new process is executed).
|
||||||
|
@ -212,6 +208,11 @@ class Actor:
|
||||||
self.loglevel = loglevel
|
self.loglevel = loglevel
|
||||||
self._arb_addr = arbiter_addr
|
self._arb_addr = arbiter_addr
|
||||||
|
|
||||||
|
# marked by the process spawning backend at startup
|
||||||
|
# will be None for the parent most process started manually
|
||||||
|
# by the user (currently called the "arbiter")
|
||||||
|
self._spawn_method = spawn_method
|
||||||
|
|
||||||
self._peers: defaultdict = defaultdict(list)
|
self._peers: defaultdict = defaultdict(list)
|
||||||
self._peer_connected: dict = {}
|
self._peer_connected: dict = {}
|
||||||
self._no_more_peers = trio.Event()
|
self._no_more_peers = trio.Event()
|
||||||
|
@ -552,8 +553,8 @@ class Actor:
|
||||||
A "root-most" (or "top-level") nursery for this actor is opened here
|
A "root-most" (or "top-level") nursery for this actor is opened here
|
||||||
and when cancelled effectively cancels the actor.
|
and when cancelled effectively cancels the actor.
|
||||||
"""
|
"""
|
||||||
arbiter_addr = arbiter_addr or self._arb_addr
|
|
||||||
registered_with_arbiter = False
|
registered_with_arbiter = False
|
||||||
|
arbiter_addr = arbiter_addr or self._arb_addr
|
||||||
try:
|
try:
|
||||||
async with trio.open_nursery() as nursery:
|
async with trio.open_nursery() as nursery:
|
||||||
self._root_nursery = nursery
|
self._root_nursery = nursery
|
||||||
|
@ -578,9 +579,13 @@ class Actor:
|
||||||
|
|
||||||
if self._spawn_method == "trio":
|
if self._spawn_method == "trio":
|
||||||
# recieve additional init params
|
# recieve additional init params
|
||||||
self._parent_main_data = await chan.recv()
|
parent_data = await chan.recv()
|
||||||
self.rpc_module_paths = await chan.recv()
|
for attr, value in parent_data.items():
|
||||||
self.statespace = await chan.recv()
|
setattr(self, attr, value)
|
||||||
|
|
||||||
|
# update local arbiter_addr var
|
||||||
|
if "_arb_addr" in parent_data:
|
||||||
|
arbiter_addr = self._arb_addr
|
||||||
|
|
||||||
except OSError: # failed to connect
|
except OSError: # failed to connect
|
||||||
log.warning(
|
log.warning(
|
||||||
|
|
|
@ -2,52 +2,49 @@ import sys
|
||||||
import trio
|
import trio
|
||||||
import argparse
|
import argparse
|
||||||
|
|
||||||
|
from ast import literal_eval
|
||||||
|
|
||||||
from ._actor import Actor
|
from ._actor import Actor
|
||||||
from ._entry import _trio_main
|
from ._entry import _trio_main
|
||||||
|
|
||||||
|
|
||||||
"""This is the "bootloader" for actors started using the native trio backend
|
"""This is the "bootloader" for actors started using the native trio backend.
|
||||||
added in #128
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
def parse_uid(arg):
|
||||||
|
uid = literal_eval(arg)
|
||||||
|
assert len(uid) == 2
|
||||||
|
assert isinstance(uid[0], str)
|
||||||
|
assert isinstance(uid[1], str)
|
||||||
|
return uid
|
||||||
|
|
||||||
|
def parse_ipaddr(arg):
|
||||||
|
addr = literal_eval(arg)
|
||||||
|
assert len(addr) == 2
|
||||||
|
assert isinstance(addr[0], str)
|
||||||
|
assert isinstance(addr[1], int)
|
||||||
|
return addr
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|
||||||
parser = argparse.ArgumentParser()
|
parser = argparse.ArgumentParser()
|
||||||
|
|
||||||
parser.add_argument("name")
|
parser.add_argument("--uid", type=parse_uid)
|
||||||
parser.add_argument("uid")
|
parser.add_argument("--loglevel", type=str)
|
||||||
parser.add_argument("loglevel")
|
parser.add_argument("--parent_addr", type=parse_ipaddr)
|
||||||
parser.add_argument("bind_addr")
|
|
||||||
parser.add_argument("parent_addr")
|
|
||||||
parser.add_argument("arbiter_addr")
|
|
||||||
|
|
||||||
args = parser.parse_args()
|
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(
|
subactor = Actor(
|
||||||
args.name,
|
args.uid[0],
|
||||||
uid=args.uid,
|
uid=args.uid[1],
|
||||||
loglevel=loglevel,
|
loglevel=args.loglevel,
|
||||||
arbiter_addr=arbiter_addr
|
spawn_method="trio"
|
||||||
)
|
)
|
||||||
subactor._spawn_method = "trio"
|
|
||||||
|
|
||||||
_trio_main(
|
_trio_main(
|
||||||
subactor,
|
subactor,
|
||||||
bind_addr,
|
("127.0.0.1", 0),
|
||||||
parent_addr=parent_addr
|
parent_addr=args.parent_addr
|
||||||
)
|
)
|
|
@ -160,23 +160,28 @@ async def cancel_on_completion(
|
||||||
async def spawn_subactor(
|
async def spawn_subactor(
|
||||||
subactor: 'Actor',
|
subactor: 'Actor',
|
||||||
accept_addr: Tuple[str, int],
|
accept_addr: Tuple[str, int],
|
||||||
parent_addr: Tuple[str, int] = None
|
parent_addr: Optional[Tuple[str, int]] = None
|
||||||
):
|
):
|
||||||
async with await trio.open_process(
|
|
||||||
[
|
spawn_cmd = [
|
||||||
sys.executable,
|
sys.executable,
|
||||||
"-m",
|
"-m",
|
||||||
# Hardcode this (instead of using ``_child.__name__`` to avoid a
|
# Hardcode this (instead of using ``_child.__name__`` to avoid a
|
||||||
# double import warning: https://stackoverflow.com/a/45070583
|
# double import warning: https://stackoverflow.com/a/45070583
|
||||||
"tractor._child",
|
"tractor._child",
|
||||||
subactor.name,
|
"--uid",
|
||||||
subactor.uid[1],
|
str(subactor.uid),
|
||||||
subactor.loglevel or "None",
|
"--parent_addr",
|
||||||
f"{accept_addr[0]}:{accept_addr[1]}",
|
str(parent_addr)
|
||||||
f"{parent_addr[0]}:{parent_addr[1]}",
|
]
|
||||||
f"{subactor._arb_addr[0]}:{subactor._arb_addr[1]}"
|
|
||||||
|
if subactor.loglevel:
|
||||||
|
spawn_cmd += [
|
||||||
|
"--loglevel",
|
||||||
|
subactor.loglevel
|
||||||
]
|
]
|
||||||
) as proc:
|
|
||||||
|
async with await trio.open_process(spawn_cmd) as proc:
|
||||||
yield proc
|
yield proc
|
||||||
|
|
||||||
|
|
||||||
|
@ -218,9 +223,12 @@ async def new_proc(
|
||||||
subactor, proc, portal)
|
subactor, proc, portal)
|
||||||
|
|
||||||
# send additional init params
|
# send additional init params
|
||||||
await chan.send(subactor._parent_main_data)
|
await chan.send({
|
||||||
await chan.send(subactor.rpc_module_paths)
|
"_parent_main_data": subactor._parent_main_data,
|
||||||
await chan.send(subactor.statespace)
|
"rpc_module_paths": subactor.rpc_module_paths,
|
||||||
|
"statespace": subactor.statespace,
|
||||||
|
"_arb_addr": subactor._arb_addr
|
||||||
|
})
|
||||||
|
|
||||||
task_status.started(portal)
|
task_status.started(portal)
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue