From 34a2f0c1f3318093eb6bdfbc3a14440e9723533a Mon Sep 17 00:00:00 2001
From: Guillermo Rodriguez <guillermo@telos.net>
Date: Sun, 23 Mar 2025 00:14:04 -0300
Subject: [PATCH] Finally switch to using address protocol in all runtime
---
examples/service_discovery.py | 2 +-
tests/test_discovery.py | 6 +-
tests/test_inter_peer_cancellation.py | 2 +-
tests/test_local.py | 2 +-
tests/test_multi_program.py | 4 +-
tests/test_spawning.py | 2 +-
tractor/_addr.py | 301 ++++++++++++++++++++++++++
tractor/_child.py | 3 +-
tractor/_context.py | 11 +-
tractor/_discovery.py | 62 +++---
tractor/_entry.py | 7 +-
tractor/_root.py | 56 ++---
tractor/_runtime.py | 182 ++++++++--------
tractor/_spawn.py | 29 +--
tractor/_supervise.py | 19 +-
tractor/ipc/__init__.py | 5 +-
tractor/ipc/_chan.py | 32 +--
tractor/ipc/_ringbuf.py | 5 +-
tractor/ipc/_tcp.py | 33 +--
tractor/ipc/_transport.py | 35 ++-
tractor/ipc/_types.py | 57 ++---
tractor/ipc/_uds.py | 37 ++--
tractor/msg/types.py | 2 +-
23 files changed, 590 insertions(+), 304 deletions(-)
create mode 100644 tractor/_addr.py
diff --git a/examples/service_discovery.py b/examples/service_discovery.py
index a0f37b88..64697e5b 100644
--- a/examples/service_discovery.py
+++ b/examples/service_discovery.py
@@ -9,7 +9,7 @@ async def main(service_name):
async with tractor.open_nursery() as an:
await an.start_actor(service_name)
- async with tractor.get_registry('127.0.0.1', 1616) as portal:
+ async with tractor.get_registry(('127.0.0.1', 1616)) as portal:
print(f"Arbiter is listening on {portal.channel}")
async with tractor.wait_for_actor(service_name) as sockaddr:
diff --git a/tests/test_discovery.py b/tests/test_discovery.py
index 87455983..18b2aa1b 100644
--- a/tests/test_discovery.py
+++ b/tests/test_discovery.py
@@ -26,7 +26,7 @@ async def test_reg_then_unreg(reg_addr):
portal = await n.start_actor('actor', enable_modules=[__name__])
uid = portal.channel.uid
- async with tractor.get_registry(*reg_addr) as aportal:
+ async with tractor.get_registry(reg_addr) as aportal:
# this local actor should be the arbiter
assert actor is aportal.actor
@@ -160,7 +160,7 @@ async def spawn_and_check_registry(
async with tractor.open_root_actor(
registry_addrs=[reg_addr],
):
- async with tractor.get_registry(*reg_addr) as portal:
+ async with tractor.get_registry(reg_addr) as portal:
# runtime needs to be up to call this
actor = tractor.current_actor()
@@ -300,7 +300,7 @@ async def close_chans_before_nursery(
async with tractor.open_root_actor(
registry_addrs=[reg_addr],
):
- async with tractor.get_registry(*reg_addr) as aportal:
+ async with tractor.get_registry(reg_addr) as aportal:
try:
get_reg = partial(unpack_reg, aportal)
diff --git a/tests/test_inter_peer_cancellation.py b/tests/test_inter_peer_cancellation.py
index bac9a791..25935df2 100644
--- a/tests/test_inter_peer_cancellation.py
+++ b/tests/test_inter_peer_cancellation.py
@@ -871,7 +871,7 @@ async def serve_subactors(
)
await ipc.send((
peer.chan.uid,
- peer.chan.raddr,
+ peer.chan.raddr.unwrap(),
))
print('Spawner exiting spawn serve loop!')
diff --git a/tests/test_local.py b/tests/test_local.py
index ecdad5fe..c6f5047a 100644
--- a/tests/test_local.py
+++ b/tests/test_local.py
@@ -38,7 +38,7 @@ async def test_self_is_registered_localportal(reg_addr):
"Verify waiting on the arbiter to register itself using a local portal."
actor = tractor.current_actor()
assert actor.is_arbiter
- async with tractor.get_registry(*reg_addr) as portal:
+ async with tractor.get_registry(reg_addr) as portal:
assert isinstance(portal, tractor._portal.LocalPortal)
with trio.fail_after(0.2):
diff --git a/tests/test_multi_program.py b/tests/test_multi_program.py
index 860eeebb..b0b145ee 100644
--- a/tests/test_multi_program.py
+++ b/tests/test_multi_program.py
@@ -32,7 +32,7 @@ def test_abort_on_sigint(daemon):
@tractor_test
async def test_cancel_remote_arbiter(daemon, reg_addr):
assert not tractor.current_actor().is_arbiter
- async with tractor.get_registry(*reg_addr) as portal:
+ async with tractor.get_registry(reg_addr) as portal:
await portal.cancel_actor()
time.sleep(0.1)
@@ -41,7 +41,7 @@ async def test_cancel_remote_arbiter(daemon, reg_addr):
# no arbiter socket should exist
with pytest.raises(OSError):
- async with tractor.get_registry(*reg_addr) as portal:
+ async with tractor.get_registry(reg_addr) as portal:
pass
diff --git a/tests/test_spawning.py b/tests/test_spawning.py
index 99ec9abc..58aa955a 100644
--- a/tests/test_spawning.py
+++ b/tests/test_spawning.py
@@ -77,7 +77,7 @@ async def movie_theatre_question():
async def test_movie_theatre_convo(start_method):
"""The main ``tractor`` routine.
"""
- async with tractor.open_nursery() as n:
+ async with tractor.open_nursery(debug_mode=True) as n:
portal = await n.start_actor(
'frank',
diff --git a/tractor/_addr.py b/tractor/_addr.py
new file mode 100644
index 00000000..0944c89d
--- /dev/null
+++ b/tractor/_addr.py
@@ -0,0 +1,301 @@
+# tractor: structured concurrent "actors".
+# Copyright 2018-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+from __future__ import annotations
+import tempfile
+from uuid import uuid4
+from typing import (
+ Protocol,
+ ClassVar,
+ TypeVar,
+ Union,
+ Type
+)
+
+import trio
+from trio import socket
+
+
+NamespaceType = TypeVar('NamespaceType')
+AddressType = TypeVar('AddressType')
+StreamType = TypeVar('StreamType')
+ListenerType = TypeVar('ListenerType')
+
+
+class Address(Protocol[
+ NamespaceType,
+ AddressType,
+ StreamType,
+ ListenerType
+]):
+
+ name_key: ClassVar[str]
+ address_type: ClassVar[Type[AddressType]]
+
+ @property
+ def is_valid(self) -> bool:
+ ...
+
+ @property
+ def namespace(self) -> NamespaceType|None:
+ ...
+
+ @classmethod
+ def from_addr(cls, addr: AddressType) -> Address:
+ ...
+
+ def unwrap(self) -> AddressType:
+ ...
+
+ @classmethod
+ def get_random(cls, namespace: NamespaceType | None = None) -> Address:
+ ...
+
+ @classmethod
+ def get_root(cls) -> Address:
+ ...
+
+ def __repr__(self) -> str:
+ ...
+
+ def __eq__(self, other) -> bool:
+ ...
+
+ async def open_stream(self, **kwargs) -> StreamType:
+ ...
+
+ async def open_listener(self, **kwargs) -> ListenerType:
+ ...
+
+
+class TCPAddress(Address[
+ str,
+ tuple[str, int],
+ trio.SocketStream,
+ trio.SocketListener
+]):
+
+ name_key: str = 'tcp'
+ address_type: type = tuple[str, int]
+
+ def __init__(
+ self,
+ host: str,
+ port: int
+ ):
+ if (
+ not isinstance(host, str)
+ or
+ not isinstance(port, int)
+ ):
+ raise TypeError(f'Expected host {host} to be str and port {port} to be int')
+ self._host = host
+ self._port = port
+
+ @property
+ def is_valid(self) -> bool:
+ return self._port != 0
+
+ @property
+ def namespace(self) -> str:
+ return self._host
+
+ @classmethod
+ def from_addr(cls, addr: tuple[str, int]) -> TCPAddress:
+ return TCPAddress(addr[0], addr[1])
+
+ def unwrap(self) -> tuple[str, int]:
+ return self._host, self._port
+
+ @classmethod
+ def get_random(cls, namespace: str = '127.0.0.1') -> TCPAddress:
+ return TCPAddress(namespace, 0)
+
+ @classmethod
+ def get_root(cls) -> Address:
+ return TCPAddress('127.0.0.1', 1616)
+
+ def __repr__(self) -> str:
+ return f'{type(self)} @ {self.unwrap()}'
+
+ def __eq__(self, other) -> bool:
+ if not isinstance(other, TCPAddress):
+ raise TypeError(
+ f'Can not compare {type(other)} with {type(self)}'
+ )
+
+ return (
+ self._host == other._host
+ and
+ self._port == other._port
+ )
+
+ async def open_stream(self, **kwargs) -> trio.SocketStream:
+ stream = await trio.open_tcp_stream(
+ self._host,
+ self._port,
+ **kwargs
+ )
+ self._host, self._port = stream.socket.getsockname()[:2]
+ return stream
+
+ async def open_listener(self, **kwargs) -> trio.SocketListener:
+ listeners = await trio.open_tcp_listeners(
+ host=self._host,
+ port=self._port,
+ **kwargs
+ )
+ assert len(listeners) == 1
+ listener = listeners[0]
+ self._host, self._port = listener.socket.getsockname()[:2]
+ return listener
+
+
+class UDSAddress(Address[
+ None,
+ str,
+ trio.SocketStream,
+ trio.SocketListener
+]):
+
+ name_key: str = 'uds'
+ address_type: type = str
+
+ def __init__(
+ self,
+ filepath: str
+ ):
+ self._filepath = filepath
+
+ @property
+ def is_valid(self) -> bool:
+ return True
+
+ @property
+ def namespace(self) -> None:
+ return
+
+ @classmethod
+ def from_addr(cls, filepath: str) -> UDSAddress:
+ return UDSAddress(filepath)
+
+ def unwrap(self) -> str:
+ return self._filepath
+
+ @classmethod
+ def get_random(cls, _ns: None = None) -> UDSAddress:
+ return UDSAddress(f'{tempfile.gettempdir()}/{uuid4().sock}')
+
+ @classmethod
+ def get_root(cls) -> Address:
+ return UDSAddress('tractor.sock')
+
+ def __repr__(self) -> str:
+ return f'{type(self)} @ {self._filepath}'
+
+ def __eq__(self, other) -> bool:
+ if not isinstance(other, UDSAddress):
+ raise TypeError(
+ f'Can not compare {type(other)} with {type(self)}'
+ )
+
+ return self._filepath == other._filepath
+
+ async def open_stream(self, **kwargs) -> trio.SocketStream:
+ stream = await trio.open_tcp_stream(
+ self._filepath,
+ **kwargs
+ )
+ self._binded = True
+ return stream
+
+ async def open_listener(self, **kwargs) -> trio.SocketListener:
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ sock.bind(self._filepath)
+ sock.listen()
+ self._binded = True
+ return trio.SocketListener(sock)
+
+
+preferred_transport = 'tcp'
+
+
+_address_types = (
+ TCPAddress,
+ UDSAddress
+)
+
+
+_default_addrs: dict[str, Type[Address]] = {
+ cls.name_key: cls
+ for cls in _address_types
+}
+
+
+AddressTypes = Union[
+ tuple([
+ cls.address_type
+ for cls in _address_types
+ ])
+]
+
+
+_default_lo_addrs: dict[
+ str,
+ AddressTypes
+] = {
+ cls.name_key: cls.get_root().unwrap()
+ for cls in _address_types
+}
+
+
+def get_address_cls(name: str) -> Type[Address]:
+ return _default_addrs[name]
+
+
+def is_wrapped_addr(addr: any) -> bool:
+ return type(addr) in _address_types
+
+
+def wrap_address(addr: AddressTypes) -> Address:
+
+ if is_wrapped_addr(addr):
+ return addr
+
+ cls = None
+ match addr:
+ case str():
+ cls = UDSAddress
+
+ case tuple() | list():
+ cls = TCPAddress
+
+ case None:
+ cls = get_address_cls(preferred_transport)
+ addr = cls.get_root().unwrap()
+
+ case _:
+ raise TypeError(
+ f'Can not wrap addr {addr} of type {type(addr)}'
+ )
+
+ return cls.from_addr(addr)
+
+
+def default_lo_addrs(transports: list[str]) -> list[AddressTypes]:
+ return [
+ _default_lo_addrs[transport]
+ for transport in transports
+ ]
diff --git a/tractor/_child.py b/tractor/_child.py
index 4226ae90..69142889 100644
--- a/tractor/_child.py
+++ b/tractor/_child.py
@@ -31,8 +31,7 @@ def parse_uid(arg):
return str(name), str(uuid) # ensures str encoding
def parse_ipaddr(arg):
- host, port = literal_eval(arg)
- return (str(host), int(port))
+ return literal_eval(arg)
if __name__ == "__main__":
diff --git a/tractor/_context.py b/tractor/_context.py
index d93d7759..19f3daef 100644
--- a/tractor/_context.py
+++ b/tractor/_context.py
@@ -859,19 +859,10 @@ class Context:
@property
def dst_maddr(self) -> str:
chan: Channel = self.chan
- dst_addr, dst_port = chan.raddr
trans: MsgTransport = chan.transport
# cid: str = self.cid
# cid_head, cid_tail = cid[:6], cid[-6:]
- return (
- f'/ipv4/{dst_addr}'
- f'/{trans.name_key}/{dst_port}'
- # f'/{self.chan.uid[0]}'
- # f'/{self.cid}'
-
- # f'/cid={cid_head}..{cid_tail}'
- # TODO: ? not use this ^ right ?
- )
+ return trans.maddr
dmaddr = dst_maddr
diff --git a/tractor/_discovery.py b/tractor/_discovery.py
index f6f4b9d9..9258b3de 100644
--- a/tractor/_discovery.py
+++ b/tractor/_discovery.py
@@ -30,6 +30,12 @@ from contextlib import asynccontextmanager as acm
from tractor.log import get_logger
from .trionics import gather_contexts
from .ipc import _connect_chan, Channel
+from ._addr import (
+ AddressTypes,
+ Address,
+ preferred_transport,
+ wrap_address
+)
from ._portal import (
Portal,
open_portal,
@@ -48,11 +54,7 @@ log = get_logger(__name__)
@acm
-async def get_registry(
- host: str,
- port: int,
-
-) -> AsyncGenerator[
+async def get_registry(addr: AddressTypes) -> AsyncGenerator[
Portal | LocalPortal | None,
None,
]:
@@ -69,13 +71,13 @@ async def get_registry(
# (likely a re-entrant call from the arbiter actor)
yield LocalPortal(
actor,
- Channel((host, port))
+ await Channel.from_addr(addr)
)
else:
# TODO: try to look pre-existing connection from
# `Actor._peers` and use it instead?
async with (
- _connect_chan((host, port)) as chan,
+ _connect_chan(addr) as chan,
open_portal(chan) as regstr_ptl,
):
yield regstr_ptl
@@ -89,11 +91,10 @@ async def get_root(
# TODO: rename mailbox to `_root_maddr` when we finally
# add and impl libp2p multi-addrs?
- host, port = _runtime_vars['_root_mailbox']
- assert host is not None
+ addr = _runtime_vars['_root_mailbox']
async with (
- _connect_chan((host, port)) as chan,
+ _connect_chan(addr) as chan,
open_portal(chan, **kwargs) as portal,
):
yield portal
@@ -134,10 +135,10 @@ def get_peer_by_name(
@acm
async def query_actor(
name: str,
- regaddr: tuple[str, int]|None = None,
+ regaddr: AddressTypes|None = None,
) -> AsyncGenerator[
- tuple[str, int]|None,
+ AddressTypes|None,
None,
]:
'''
@@ -163,31 +164,31 @@ async def query_actor(
return
reg_portal: Portal
- regaddr: tuple[str, int] = regaddr or actor.reg_addrs[0]
- async with get_registry(*regaddr) as reg_portal:
+ regaddr: Address = wrap_address(regaddr) or actor.reg_addrs[0]
+ async with get_registry(regaddr) as reg_portal:
# TODO: return portals to all available actors - for now
# just the last one that registered
- sockaddr: tuple[str, int] = await reg_portal.run_from_ns(
+ addr: AddressTypes = await reg_portal.run_from_ns(
'self',
'find_actor',
name=name,
)
- yield sockaddr
+ yield addr
@acm
async def maybe_open_portal(
- addr: tuple[str, int],
+ addr: AddressTypes,
name: str,
):
async with query_actor(
name=name,
regaddr=addr,
- ) as sockaddr:
+ ) as addr:
pass
- if sockaddr:
- async with _connect_chan(sockaddr) as chan:
+ if addr:
+ async with _connect_chan(addr) as chan:
async with open_portal(chan) as portal:
yield portal
else:
@@ -197,7 +198,8 @@ async def maybe_open_portal(
@acm
async def find_actor(
name: str,
- registry_addrs: list[tuple[str, int]]|None = None,
+ registry_addrs: list[AddressTypes]|None = None,
+ enable_transports: list[str] = [preferred_transport],
only_first: bool = True,
raise_on_none: bool = False,
@@ -224,15 +226,15 @@ async def find_actor(
# XXX NOTE: make sure to dynamically read the value on
# every call since something may change it globally (eg.
# like in our discovery test suite)!
- from . import _root
+ from ._addr import default_lo_addrs
registry_addrs = (
_runtime_vars['_registry_addrs']
or
- _root._default_lo_addrs
+ default_lo_addrs(enable_transports)
)
maybe_portals: list[
- AsyncContextManager[tuple[str, int]]
+ AsyncContextManager[AddressTypes]
] = list(
maybe_open_portal(
addr=addr,
@@ -274,7 +276,7 @@ async def find_actor(
@acm
async def wait_for_actor(
name: str,
- registry_addr: tuple[str, int] | None = None,
+ registry_addr: AddressTypes | None = None,
) -> AsyncGenerator[Portal, None]:
'''
@@ -291,7 +293,7 @@ async def wait_for_actor(
yield peer_portal
return
- regaddr: tuple[str, int] = (
+ regaddr: AddressTypes = (
registry_addr
or
actor.reg_addrs[0]
@@ -299,8 +301,8 @@ async def wait_for_actor(
# TODO: use `.trionics.gather_contexts()` like
# above in `find_actor()` as well?
reg_portal: Portal
- async with get_registry(*regaddr) as reg_portal:
- sockaddrs = await reg_portal.run_from_ns(
+ async with get_registry(regaddr) as reg_portal:
+ addrs = await reg_portal.run_from_ns(
'self',
'wait_for_actor',
name=name,
@@ -308,8 +310,8 @@ async def wait_for_actor(
# get latest registered addr by default?
# TODO: offer multi-portal yields in multi-homed case?
- sockaddr: tuple[str, int] = sockaddrs[-1]
+ addr: AddressTypes = addrs[-1]
- async with _connect_chan(sockaddr) as chan:
+ async with _connect_chan(addr) as chan:
async with open_portal(chan) as portal:
yield portal
diff --git a/tractor/_entry.py b/tractor/_entry.py
index 8156d25f..1328aa45 100644
--- a/tractor/_entry.py
+++ b/tractor/_entry.py
@@ -37,6 +37,7 @@ from .log import (
from . import _state
from .devx import _debug
from .to_asyncio import run_as_asyncio_guest
+from ._addr import AddressTypes
from ._runtime import (
async_main,
Actor,
@@ -52,10 +53,10 @@ log = get_logger(__name__)
def _mp_main(
actor: Actor,
- accept_addrs: list[tuple[str, int]],
+ accept_addrs: list[AddressTypes],
forkserver_info: tuple[Any, Any, Any, Any, Any],
start_method: SpawnMethodKey,
- parent_addr: tuple[str, int] | None = None,
+ parent_addr: AddressTypes | None = None,
infect_asyncio: bool = False,
) -> None:
@@ -206,7 +207,7 @@ def nest_from_op(
def _trio_main(
actor: Actor,
*,
- parent_addr: tuple[str, int] | None = None,
+ parent_addr: AddressTypes | None = None,
infect_asyncio: bool = False,
) -> None:
diff --git a/tractor/_root.py b/tractor/_root.py
index 40682a7a..e9cac3f2 100644
--- a/tractor/_root.py
+++ b/tractor/_root.py
@@ -43,21 +43,18 @@ from .devx import _debug
from . import _spawn
from . import _state
from . import log
-from .ipc import _connect_chan
+from .ipc import (
+ _connect_chan,
+)
+from ._addr import (
+ AddressTypes,
+ wrap_address,
+ preferred_transport,
+ default_lo_addrs
+)
from ._exceptions import is_multi_cancelled
-# set at startup and after forks
-_default_host: str = '127.0.0.1'
-_default_port: int = 1616
-
-# default registry always on localhost
-_default_lo_addrs: list[tuple[str, int]] = [(
- _default_host,
- _default_port,
-)]
-
-
logger = log.get_logger('tractor')
@@ -66,10 +63,12 @@ async def open_root_actor(
*,
# defaults are above
- registry_addrs: list[tuple[str, int]]|None = None,
+ registry_addrs: list[AddressTypes]|None = None,
# defaults are above
- arbiter_addr: tuple[str, int]|None = None,
+ arbiter_addr: tuple[AddressTypes]|None = None,
+
+ enable_transports: list[str] = [preferred_transport],
name: str|None = 'root',
@@ -195,11 +194,9 @@ async def open_root_actor(
)
registry_addrs = [arbiter_addr]
- registry_addrs: list[tuple[str, int]] = (
- registry_addrs
- or
- _default_lo_addrs
- )
+ if not registry_addrs:
+ registry_addrs: list[AddressTypes] = default_lo_addrs(enable_transports)
+
assert registry_addrs
loglevel = (
@@ -248,10 +245,10 @@ async def open_root_actor(
enable_stack_on_sig()
# closed into below ping task-func
- ponged_addrs: list[tuple[str, int]] = []
+ ponged_addrs: list[AddressTypes] = []
async def ping_tpt_socket(
- addr: tuple[str, int],
+ addr: AddressTypes,
timeout: float = 1,
) -> None:
'''
@@ -284,10 +281,10 @@ async def open_root_actor(
for addr in registry_addrs:
tn.start_soon(
ping_tpt_socket,
- tuple(addr), # TODO: just drop this requirement?
+ addr,
)
- trans_bind_addrs: list[tuple[str, int]] = []
+ trans_bind_addrs: list[AddressTypes] = []
# Create a new local root-actor instance which IS NOT THE
# REGISTRAR
@@ -311,9 +308,12 @@ async def open_root_actor(
)
# DO NOT use the registry_addrs as the transport server
# addrs for this new non-registar, root-actor.
- for host, port in ponged_addrs:
- # NOTE: zero triggers dynamic OS port allocation
- trans_bind_addrs.append((host, 0))
+ for addr in ponged_addrs:
+ waddr = wrap_address(addr)
+ print(waddr)
+ trans_bind_addrs.append(
+ waddr.get_random(namespace=waddr.namespace)
+ )
# Start this local actor as the "registrar", aka a regular
# actor who manages the local registry of "mailboxes" of
@@ -322,7 +322,7 @@ async def open_root_actor(
# NOTE that if the current actor IS THE REGISTAR, the
# following init steps are taken:
- # - the tranport layer server is bound to each (host, port)
+ # - the tranport layer server is bound to each addr
# pair defined in provided registry_addrs, or the default.
trans_bind_addrs = registry_addrs
@@ -462,7 +462,7 @@ def run_daemon(
# runtime kwargs
name: str | None = 'root',
- registry_addrs: list[tuple[str, int]] = _default_lo_addrs,
+ registry_addrs: list[AddressTypes]|None = None,
start_method: str | None = None,
debug_mode: bool = False,
diff --git a/tractor/_runtime.py b/tractor/_runtime.py
index eaab31b6..e755d5ce 100644
--- a/tractor/_runtime.py
+++ b/tractor/_runtime.py
@@ -74,6 +74,12 @@ from tractor.msg import (
types as msgtypes,
)
from .ipc import Channel
+from ._addr import (
+ AddressTypes,
+ Address,
+ TCPAddress,
+ wrap_address,
+)
from ._context import (
mk_context,
Context,
@@ -179,11 +185,11 @@ class Actor:
enable_modules: list[str] = [],
uid: str|None = None,
loglevel: str|None = None,
- registry_addrs: list[tuple[str, int]]|None = None,
+ registry_addrs: list[AddressTypes]|None = None,
spawn_method: str|None = None,
# TODO: remove!
- arbiter_addr: tuple[str, int]|None = None,
+ arbiter_addr: AddressTypes|None = None,
) -> None:
'''
@@ -223,7 +229,7 @@ class Actor:
DeprecationWarning,
stacklevel=2,
)
- registry_addrs: list[tuple[str, int]] = [arbiter_addr]
+ registry_addrs: list[AddressTypes] = [arbiter_addr]
# marked by the process spawning backend at startup
# will be None for the parent most process started manually
@@ -257,6 +263,7 @@ class Actor:
] = {}
self._listeners: list[trio.abc.Listener] = []
+ self._listen_addrs: list[Address] = []
self._parent_chan: Channel|None = None
self._forkserver_info: tuple|None = None
@@ -269,13 +276,13 @@ class Actor:
# when provided, init the registry addresses property from
# input via the validator.
- self._reg_addrs: list[tuple[str, int]] = []
+ self._reg_addrs: list[AddressTypes] = []
if registry_addrs:
- self.reg_addrs: list[tuple[str, int]] = registry_addrs
+ self.reg_addrs: list[AddressTypes] = registry_addrs
_state._runtime_vars['_registry_addrs'] = registry_addrs
@property
- def reg_addrs(self) -> list[tuple[str, int]]:
+ def reg_addrs(self) -> list[AddressTypes]:
'''
List of (socket) addresses for all known (and contactable)
registry actors.
@@ -286,7 +293,7 @@ class Actor:
@reg_addrs.setter
def reg_addrs(
self,
- addrs: list[tuple[str, int]],
+ addrs: list[AddressTypes],
) -> None:
if not addrs:
log.warning(
@@ -295,16 +302,7 @@ class Actor:
)
return
- # always sanity check the input list since it's critical
- # that addrs are correct for discovery sys operation.
- for addr in addrs:
- if not isinstance(addr, tuple):
- raise ValueError(
- 'Expected `Actor.reg_addrs: list[tuple[str, int]]`\n'
- f'Got {addrs}'
- )
-
- self._reg_addrs = addrs
+ self._reg_addrs = addrs
async def wait_for_peer(
self,
@@ -1024,11 +1022,11 @@ class Actor:
async def _from_parent(
self,
- parent_addr: tuple[str, int]|None,
+ parent_addr: AddressTypes|None,
) -> tuple[
Channel,
- list[tuple[str, int]]|None,
+ list[AddressTypes]|None,
]:
'''
Bootstrap this local actor's runtime config from its parent by
@@ -1040,13 +1038,13 @@ class Actor:
# Connect back to the parent actor and conduct initial
# handshake. From this point on if we error, we
# attempt to ship the exception back to the parent.
- chan = await Channel.from_destaddr(parent_addr)
+ chan = await Channel.from_addr(wrap_address(parent_addr))
# TODO: move this into a `Channel.handshake()`?
# Initial handshake: swap names.
await self._do_handshake(chan)
- accept_addrs: list[tuple[str, int]]|None = None
+ accept_addrs: list[AddressTypes]|None = None
if self._spawn_method == "trio":
@@ -1063,7 +1061,7 @@ class Actor:
# if "trace"/"util" mode is enabled?
f'{pretty_struct.pformat(spawnspec)}\n'
)
- accept_addrs: list[tuple[str, int]] = spawnspec.bind_addrs
+ accept_addrs: list[AddressTypes] = spawnspec.bind_addrs
# TODO: another `Struct` for rtvs..
rvs: dict[str, Any] = spawnspec._runtime_vars
@@ -1170,8 +1168,7 @@ class Actor:
self,
handler_nursery: Nursery,
*,
- # (host, port) to bind for channel server
- listen_sockaddrs: list[tuple[str, int]]|None = None,
+ listen_addrs: list[AddressTypes]|None = None,
task_status: TaskStatus[Nursery] = trio.TASK_STATUS_IGNORED,
) -> None:
@@ -1183,37 +1180,39 @@ class Actor:
`.cancel_server()` is called.
'''
- if listen_sockaddrs is None:
- listen_sockaddrs = [(None, 0)]
+ if listen_addrs is None:
+ listen_addrs = [TCPAddress.get_random()]
+
+ else:
+ listen_addrs: list[Address] = [
+ wrap_address(a) for a in listen_addrs
+ ]
self._server_down = trio.Event()
try:
async with trio.open_nursery() as server_n:
+ listeners: list[trio.abc.Listener] = [
+ await addr.open_listener()
+ for addr in listen_addrs
+ ]
+ await server_n.start(
+ partial(
+ trio.serve_listeners,
+ handler=self._stream_handler,
+ listeners=listeners,
- for host, port in listen_sockaddrs:
- listeners: list[trio.abc.Listener] = await server_n.start(
- partial(
- trio.serve_tcp,
-
- handler=self._stream_handler,
- port=port,
- host=host,
-
- # NOTE: configured such that new
- # connections will stay alive even if
- # this server is cancelled!
- handler_nursery=handler_nursery,
- )
+ # NOTE: configured such that new
+ # connections will stay alive even if
+ # this server is cancelled!
+ handler_nursery=handler_nursery
)
- sockets: list[trio.socket] = [
- getattr(listener, 'socket', 'unknown socket')
- for listener in listeners
- ]
- log.runtime(
- 'Started TCP server(s)\n'
- f'|_{sockets}\n'
- )
- self._listeners.extend(listeners)
+ )
+ log.runtime(
+ 'Started server(s)\n'
+ '\n'.join([f'|_{addr}' for addr in listen_addrs])
+ )
+ self._listen_addrs.extend(listen_addrs)
+ self._listeners.extend(listeners)
task_status.started(server_n)
@@ -1576,26 +1575,21 @@ class Actor:
return False
@property
- def accept_addrs(self) -> list[tuple[str, int]]:
+ def accept_addrs(self) -> list[AddressTypes]:
'''
All addresses to which the transport-channel server binds
and listens for new connections.
'''
- # throws OSError on failure
- return [
- listener.socket.getsockname()
- for listener in self._listeners
- ] # type: ignore
+ return [a.unwrap() for a in self._listen_addrs]
@property
- def accept_addr(self) -> tuple[str, int]:
+ def accept_addr(self) -> AddressTypes:
'''
Primary address to which the IPC transport server is
bound and listening for new connections.
'''
- # throws OSError on failure
return self.accept_addrs[0]
def get_parent(self) -> Portal:
@@ -1667,7 +1661,7 @@ class Actor:
async def async_main(
actor: Actor,
- accept_addrs: tuple[str, int]|None = None,
+ accept_addrs: AddressTypes|None = None,
# XXX: currently ``parent_addr`` is only needed for the
# ``multiprocessing`` backend (which pickles state sent to
@@ -1676,7 +1670,7 @@ async def async_main(
# change this to a simple ``is_subactor: bool`` which will
# be False when running as root actor and True when as
# a subactor.
- parent_addr: tuple[str, int]|None = None,
+ parent_addr: AddressTypes|None = None,
task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
) -> None:
@@ -1766,7 +1760,7 @@ async def async_main(
partial(
actor._serve_forever,
service_nursery,
- listen_sockaddrs=accept_addrs,
+ listen_addrs=accept_addrs,
)
)
except OSError as oserr:
@@ -1782,7 +1776,7 @@ async def async_main(
raise
- accept_addrs: list[tuple[str, int]] = actor.accept_addrs
+ accept_addrs: list[AddressTypes] = actor.accept_addrs
# NOTE: only set the loopback addr for the
# process-tree-global "root" mailbox since
@@ -1790,9 +1784,8 @@ async def async_main(
# their root actor over that channel.
if _state._runtime_vars['_is_root']:
for addr in accept_addrs:
- host, _ = addr
- # TODO: generic 'lo' detector predicate
- if '127.0.0.1' in host:
+ waddr = wrap_address(addr)
+ if waddr == waddr.get_root():
_state._runtime_vars['_root_mailbox'] = addr
# Register with the arbiter if we're told its addr
@@ -1807,24 +1800,21 @@ async def async_main(
# only on unique actor uids?
for addr in actor.reg_addrs:
try:
- assert isinstance(addr, tuple)
- assert addr[1] # non-zero after bind
+ waddr = wrap_address(addr)
+ assert waddr.is_valid
except AssertionError:
await _debug.pause()
- async with get_registry(*addr) as reg_portal:
+ async with get_registry(addr) as reg_portal:
for accept_addr in accept_addrs:
-
- if not accept_addr[1]:
- await _debug.pause()
-
- assert accept_addr[1]
+ accept_addr = wrap_address(accept_addr)
+ assert accept_addr.is_valid
await reg_portal.run_from_ns(
'self',
'register_actor',
uid=actor.uid,
- sockaddr=accept_addr,
+ addr=accept_addr.unwrap(),
)
is_registered: bool = True
@@ -1951,12 +1941,13 @@ async def async_main(
):
failed: bool = False
for addr in actor.reg_addrs:
- assert isinstance(addr, tuple)
+ waddr = wrap_address(addr)
+ assert waddr.is_valid
with trio.move_on_after(0.5) as cs:
cs.shield = True
try:
async with get_registry(
- *addr,
+ addr,
) as reg_portal:
await reg_portal.run_from_ns(
'self',
@@ -2034,7 +2025,7 @@ class Arbiter(Actor):
self._registry: dict[
tuple[str, str],
- tuple[str, int],
+ AddressTypes,
] = {}
self._waiters: dict[
str,
@@ -2050,18 +2041,18 @@ class Arbiter(Actor):
self,
name: str,
- ) -> tuple[str, int]|None:
+ ) -> AddressTypes|None:
- for uid, sockaddr in self._registry.items():
+ for uid, addr in self._registry.items():
if name in uid:
- return sockaddr
+ return addr
return None
async def get_registry(
self
- ) -> dict[str, tuple[str, int]]:
+ ) -> dict[str, AddressTypes]:
'''
Return current name registry.
@@ -2081,7 +2072,7 @@ class Arbiter(Actor):
self,
name: str,
- ) -> list[tuple[str, int]]:
+ ) -> list[AddressTypes]:
'''
Wait for a particular actor to register.
@@ -2089,44 +2080,41 @@ class Arbiter(Actor):
registered.
'''
- sockaddrs: list[tuple[str, int]] = []
- sockaddr: tuple[str, int]
+ addrs: list[AddressTypes] = []
+ addr: AddressTypes
mailbox_info: str = 'Actor registry contact infos:\n'
- for uid, sockaddr in self._registry.items():
+ for uid, addr in self._registry.items():
mailbox_info += (
f'|_uid: {uid}\n'
- f'|_sockaddr: {sockaddr}\n\n'
+ f'|_addr: {addr}\n\n'
)
if name == uid[0]:
- sockaddrs.append(sockaddr)
+ addrs.append(addr)
- if not sockaddrs:
+ if not addrs:
waiter = trio.Event()
self._waiters.setdefault(name, []).append(waiter)
await waiter.wait()
for uid in self._waiters[name]:
if not isinstance(uid, trio.Event):
- sockaddrs.append(self._registry[uid])
+ addrs.append(self._registry[uid])
log.runtime(mailbox_info)
- return sockaddrs
+ return addrs
async def register_actor(
self,
uid: tuple[str, str],
- sockaddr: tuple[str, int]
-
+ addr: AddressTypes
) -> None:
uid = name, hash = (str(uid[0]), str(uid[1]))
- addr = (host, port) = (
- str(sockaddr[0]),
- int(sockaddr[1]),
- )
- if port == 0:
+ waddr: Address = wrap_address(addr)
+ if not waddr.is_valid:
+ # should never be 0-dynamic-os-alloc
await _debug.pause()
- assert port # should never be 0-dynamic-os-alloc
+
self._registry[uid] = addr
# pop and signal all waiter events
diff --git a/tractor/_spawn.py b/tractor/_spawn.py
index dc2429d9..d1eb7f37 100644
--- a/tractor/_spawn.py
+++ b/tractor/_spawn.py
@@ -46,6 +46,7 @@ from tractor._state import (
_runtime_vars,
)
from tractor.log import get_logger
+from tractor._addr import AddressTypes
from tractor._portal import Portal
from tractor._runtime import Actor
from tractor._entry import _mp_main
@@ -392,8 +393,8 @@ async def new_proc(
errors: dict[tuple[str, str], Exception],
# passed through to actor main
- bind_addrs: list[tuple[str, int]],
- parent_addr: tuple[str, int],
+ bind_addrs: list[AddressTypes],
+ parent_addr: AddressTypes,
_runtime_vars: dict[str, Any], # serialized and sent to _child
*,
@@ -431,8 +432,8 @@ async def trio_proc(
errors: dict[tuple[str, str], Exception],
# passed through to actor main
- bind_addrs: list[tuple[str, int]],
- parent_addr: tuple[str, int],
+ bind_addrs: list[AddressTypes],
+ parent_addr: AddressTypes,
_runtime_vars: dict[str, Any], # serialized and sent to _child
*,
infect_asyncio: bool = False,
@@ -520,15 +521,15 @@ async def trio_proc(
# send a "spawning specification" which configures the
# initial runtime state of the child.
- await chan.send(
- SpawnSpec(
- _parent_main_data=subactor._parent_main_data,
- enable_modules=subactor.enable_modules,
- reg_addrs=subactor.reg_addrs,
- bind_addrs=bind_addrs,
- _runtime_vars=_runtime_vars,
- )
+ sspec = SpawnSpec(
+ _parent_main_data=subactor._parent_main_data,
+ enable_modules=subactor.enable_modules,
+ reg_addrs=subactor.reg_addrs,
+ bind_addrs=bind_addrs,
+ _runtime_vars=_runtime_vars,
)
+ log.runtime(f'Sending spawn spec: {str(sspec)}')
+ await chan.send(sspec)
# track subactor in current nursery
curr_actor: Actor = current_actor()
@@ -638,8 +639,8 @@ async def mp_proc(
subactor: Actor,
errors: dict[tuple[str, str], Exception],
# passed through to actor main
- bind_addrs: list[tuple[str, int]],
- parent_addr: tuple[str, int],
+ bind_addrs: list[AddressTypes],
+ parent_addr: AddressTypes,
_runtime_vars: dict[str, Any], # serialized and sent to _child
*,
infect_asyncio: bool = False,
diff --git a/tractor/_supervise.py b/tractor/_supervise.py
index 052a5f4c..2a3842f7 100644
--- a/tractor/_supervise.py
+++ b/tractor/_supervise.py
@@ -28,7 +28,13 @@ import warnings
import trio
+
from .devx._debug import maybe_wait_for_debugger
+from ._addr import (
+ AddressTypes,
+ preferred_transport,
+ get_address_cls
+)
from ._state import current_actor, is_main_process
from .log import get_logger, get_loglevel
from ._runtime import Actor
@@ -47,8 +53,6 @@ if TYPE_CHECKING:
log = get_logger(__name__)
-_default_bind_addr: tuple[str, int] = ('127.0.0.1', 0)
-
class ActorNursery:
'''
@@ -130,8 +134,9 @@ class ActorNursery:
*,
- bind_addrs: list[tuple[str, int]] = [_default_bind_addr],
+ bind_addrs: list[AddressTypes]|None = None,
rpc_module_paths: list[str]|None = None,
+ enable_transports: list[str] = [preferred_transport],
enable_modules: list[str]|None = None,
loglevel: str|None = None, # set log level per subactor
debug_mode: bool|None = None,
@@ -156,6 +161,12 @@ class ActorNursery:
or get_loglevel()
)
+ if not bind_addrs:
+ bind_addrs: list[AddressTypes] = [
+ get_address_cls(transport).get_random().unwrap()
+ for transport in enable_transports
+ ]
+
# configure and pass runtime state
_rtv = _state._runtime_vars.copy()
_rtv['_is_root'] = False
@@ -224,7 +235,7 @@ class ActorNursery:
*,
name: str | None = None,
- bind_addrs: tuple[str, int] = [_default_bind_addr],
+ bind_addrs: AddressTypes|None = None,
rpc_module_paths: list[str] | None = None,
enable_modules: list[str] | None = None,
loglevel: str | None = None, # set log level per subactor
diff --git a/tractor/ipc/__init__.py b/tractor/ipc/__init__.py
index dd4d9e5a..9bd2240f 100644
--- a/tractor/ipc/__init__.py
+++ b/tractor/ipc/__init__.py
@@ -17,7 +17,6 @@ import platform
from ._transport import (
MsgTransportKey as MsgTransportKey,
- AddressType as AddressType,
MsgType as MsgType,
MsgTransport as MsgTransport,
MsgpackTransport as MsgpackTransport
@@ -27,10 +26,8 @@ from ._tcp import MsgpackTCPStream as MsgpackTCPStream
from ._uds import MsgpackUDSStream as MsgpackUDSStream
from ._types import (
- default_lo_addrs as default_lo_addrs,
- transport_from_destaddr as transport_from_destaddr,
+ transport_from_addr as transport_from_addr,
transport_from_stream as transport_from_stream,
- AddressTypes as AddressTypes
)
from ._chan import (
diff --git a/tractor/ipc/_chan.py b/tractor/ipc/_chan.py
index ee259371..93f17132 100644
--- a/tractor/ipc/_chan.py
+++ b/tractor/ipc/_chan.py
@@ -35,8 +35,12 @@ import trio
from tractor.ipc._transport import MsgTransport
from tractor.ipc._types import (
- transport_from_destaddr,
+ transport_from_addr,
transport_from_stream,
+)
+from tractor._addr import (
+ wrap_address,
+ Address,
AddressTypes
)
from tractor.log import get_logger
@@ -66,7 +70,6 @@ class Channel:
def __init__(
self,
- destaddr: AddressTypes|None = None,
transport: MsgTransport|None = None,
# TODO: optional reconnection support?
# auto_reconnect: bool = False,
@@ -81,8 +84,6 @@ class Channel:
# user in ``.from_stream()``.
self._transport: MsgTransport|None = transport
- self._destaddr = destaddr if destaddr else self._transport.raddr
-
# set after handshake - always uid of far end
self.uid: tuple[str, str]|None = None
@@ -121,13 +122,14 @@ class Channel:
)
@classmethod
- async def from_destaddr(
+ async def from_addr(
cls,
- destaddr: AddressTypes,
+ addr: AddressTypes,
**kwargs
) -> Channel:
- transport_cls = transport_from_destaddr(destaddr)
- transport = await transport_cls.connect_to(destaddr, **kwargs)
+ addr: Address = wrap_address(addr)
+ transport_cls = transport_from_addr(addr)
+ transport = await transport_cls.connect_to(addr, **kwargs)
log.transport(
f'Opened channel[{type(transport)}]: {transport.laddr} -> {transport.raddr}'
@@ -164,11 +166,11 @@ class Channel:
)
@property
- def laddr(self) -> tuple[str, int]|None:
+ def laddr(self) -> Address|None:
return self._transport.laddr if self._transport else None
@property
- def raddr(self) -> tuple[str, int]|None:
+ def raddr(self) -> Address|None:
return self._transport.raddr if self._transport else None
# TODO: something like,
@@ -205,7 +207,11 @@ class Channel:
# assert err
__tracebackhide__: bool = False
else:
- assert err.cid
+ try:
+ assert err.cid
+
+ except KeyError:
+ raise err
raise
@@ -332,14 +338,14 @@ class Channel:
@acm
async def _connect_chan(
- destaddr: AddressTypes
+ addr: AddressTypes
) -> typing.AsyncGenerator[Channel, None]:
'''
Create and connect a channel with disconnect on context manager
teardown.
'''
- chan = await Channel.from_destaddr(destaddr)
+ chan = await Channel.from_addr(addr)
yield chan
with trio.CancelScope(shield=True):
await chan.aclose()
diff --git a/tractor/ipc/_ringbuf.py b/tractor/ipc/_ringbuf.py
index 10975b7a..6a71a00a 100644
--- a/tractor/ipc/_ringbuf.py
+++ b/tractor/ipc/_ringbuf.py
@@ -183,6 +183,9 @@ class RingBuffSender(trio.abc.SendStream):
def wrap_fd(self) -> int:
return self._wrap_event.fd
+ async def _wait_wrap(self):
+ await self._wrap_event.read()
+
async def send_all(self, data: Buffer):
async with self._send_lock:
# while data is larger than the remaining buf
@@ -193,7 +196,7 @@ class RingBuffSender(trio.abc.SendStream):
self._shm.buf[self.ptr:] = data[:remaining]
# signal write and wait for reader wrap around
self._write_event.write(remaining)
- await self._wrap_event.read()
+ await self._wait_wrap()
# wrap around and trim already written bytes
self._ptr = 0
diff --git a/tractor/ipc/_tcp.py b/tractor/ipc/_tcp.py
index 4a69ebbd..a8008519 100644
--- a/tractor/ipc/_tcp.py
+++ b/tractor/ipc/_tcp.py
@@ -23,6 +23,7 @@ import trio
from tractor.msg import MsgCodec
from tractor.log import get_logger
+from tractor._addr import TCPAddress
from tractor.ipc._transport import MsgpackTransport
@@ -38,9 +39,8 @@ class MsgpackTCPStream(MsgpackTransport):
using the ``msgspec`` codec lib.
'''
- address_type = tuple[str, int]
+ address_type = TCPAddress
layer_key: int = 4
- name_key: str = 'tcp'
# def __init__(
# self,
@@ -55,19 +55,32 @@ class MsgpackTCPStream(MsgpackTransport):
# codec=codec
# )
+ @property
+ def maddr(self) -> str:
+ host, port = self.raddr.unwrap()
+ return (
+ f'/ipv4/{host}'
+ f'/{self.address_type.name_key}/{port}'
+ # f'/{self.chan.uid[0]}'
+ # f'/{self.cid}'
+
+ # f'/cid={cid_head}..{cid_tail}'
+ # TODO: ? not use this ^ right ?
+ )
+
def connected(self) -> bool:
return self.stream.socket.fileno() != -1
@classmethod
async def connect_to(
cls,
- destaddr: tuple[str, int],
+ destaddr: TCPAddress,
prefix_size: int = 4,
codec: MsgCodec|None = None,
**kwargs
) -> MsgpackTCPStream:
stream = await trio.open_tcp_stream(
- *destaddr,
+ *destaddr.unwrap(),
**kwargs
)
return MsgpackTCPStream(
@@ -87,14 +100,6 @@ class MsgpackTCPStream(MsgpackTransport):
lsockname = stream.socket.getsockname()
rsockname = stream.socket.getpeername()
return (
- tuple(lsockname[:2]),
- tuple(rsockname[:2]),
+ TCPAddress.from_addr(tuple(lsockname[:2])),
+ TCPAddress.from_addr(tuple(rsockname[:2])),
)
-
- @classmethod
- def get_random_addr(self) -> tuple[str, int]:
- return (None, 0)
-
- @classmethod
- def get_root_addr(self) -> tuple[str, int]:
- return ('127.0.0.1', 1616)
diff --git a/tractor/ipc/_transport.py b/tractor/ipc/_transport.py
index 55a218f7..8c35ffee 100644
--- a/tractor/ipc/_transport.py
+++ b/tractor/ipc/_transport.py
@@ -50,6 +50,7 @@ from tractor.msg import (
types as msgtypes,
pretty_struct,
)
+from tractor._addr import Address
log = get_logger(__name__)
@@ -62,12 +63,11 @@ MsgTransportKey = tuple[str, str]
# ?TODO? this should be our `Union[*msgtypes.__spec__]` alias now right..?
# => BLEH, except can't bc prots must inherit typevar or param-spec
# vars..
-AddressType = TypeVar('AddressType')
MsgType = TypeVar('MsgType')
@runtime_checkable
-class MsgTransport(Protocol[AddressType, MsgType]):
+class MsgTransport(Protocol[MsgType]):
#
# ^-TODO-^ consider using a generic def and indexing with our
# eventual msg definition/types?
@@ -75,10 +75,9 @@ class MsgTransport(Protocol[AddressType, MsgType]):
stream: trio.abc.Stream
drained: list[MsgType]
- address_type: ClassVar[Type[AddressType]]
+ address_type: ClassVar[Type[Address]]
codec_key: ClassVar[str]
- name_key: ClassVar[str]
# XXX: should this instead be called `.sendall()`?
async def send(self, msg: MsgType) -> None:
@@ -100,20 +99,24 @@ class MsgTransport(Protocol[AddressType, MsgType]):
@classmethod
def key(cls) -> MsgTransportKey:
- return cls.codec_key, cls.name_key
+ return cls.codec_key, cls.address_type.name_key
@property
- def laddr(self) -> AddressType:
+ def laddr(self) -> Address:
...
@property
- def raddr(self) -> AddressType:
+ def raddr(self) -> Address:
+ ...
+
+ @property
+ def maddr(self) -> str:
...
@classmethod
async def connect_to(
cls,
- destaddr: AddressType,
+ addr: Address,
**kwargs
) -> MsgTransport:
...
@@ -123,8 +126,8 @@ class MsgTransport(Protocol[AddressType, MsgType]):
cls,
stream: trio.abc.Stream
) -> tuple[
- AddressType, # local
- AddressType # remote
+ Address, # local
+ Address # remote
]:
'''
Return the `trio` streaming transport prot's addrs for both
@@ -133,14 +136,6 @@ class MsgTransport(Protocol[AddressType, MsgType]):
'''
...
- @classmethod
- def get_random_addr(self) -> AddressType:
- ...
-
- @classmethod
- def get_root_addr(self) -> AddressType:
- ...
-
class MsgpackTransport(MsgTransport):
@@ -447,9 +442,9 @@ class MsgpackTransport(MsgTransport):
return self._aiter_pkts
@property
- def laddr(self) -> AddressType:
+ def laddr(self) -> Address:
return self._laddr
@property
- def raddr(self) -> AddressType:
+ def raddr(self) -> Address:
return self._raddr
diff --git a/tractor/ipc/_types.py b/tractor/ipc/_types.py
index 92d3af91..3e0e43e5 100644
--- a/tractor/ipc/_types.py
+++ b/tractor/ipc/_types.py
@@ -13,49 +13,42 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
-from typing import Type, Union
+from typing import Type
import trio
import socket
-from ._transport import (
+from tractor._addr import Address
+from tractor.ipc._transport import (
MsgTransportKey,
MsgTransport
)
-from ._tcp import MsgpackTCPStream
-from ._uds import MsgpackUDSStream
+from tractor.ipc._tcp import MsgpackTCPStream
+from tractor.ipc._uds import MsgpackUDSStream
+# manually updated list of all supported msg transport types
_msg_transports = [
MsgpackTCPStream,
MsgpackUDSStream
]
-# manually updated list of all supported codec+transport types
-key_to_transport: dict[MsgTransportKey, Type[MsgTransport]] = {
+# convert a MsgTransportKey to the corresponding transport type
+_key_to_transport: dict[MsgTransportKey, Type[MsgTransport]] = {
cls.key(): cls
for cls in _msg_transports
}
-
-# all different address py types we use
-AddressTypes = Union[
- tuple([
- cls.address_type
- for cls in _msg_transports
- ])
-]
-
-
-default_lo_addrs: dict[MsgTransportKey, AddressTypes] = {
- cls.key(): cls.get_root_addr()
+# convert an Address wrapper to its corresponding transport type
+_addr_to_transport: dict[Type[Address], Type[MsgTransport]] = {
+ cls.address_type: cls
for cls in _msg_transports
}
-def transport_from_destaddr(
- destaddr: AddressTypes,
+def transport_from_addr(
+ addr: Address,
codec_key: str = 'msgpack',
) -> Type[MsgTransport]:
'''
@@ -63,23 +56,13 @@ def transport_from_destaddr(
corresponding `MsgTransport` type.
'''
- match destaddr:
- case str():
- return MsgpackUDSStream
+ try:
+ return _addr_to_transport[type(addr)]
- case tuple():
- if (
- len(destaddr) == 2
- and
- isinstance(destaddr[0], str)
- and
- isinstance(destaddr[1], int)
- ):
- return MsgpackTCPStream
-
- raise NotImplementedError(
- f'No known transport for address {destaddr}'
- )
+ except KeyError:
+ raise NotImplementedError(
+ f'No known transport for address {repr(addr)}'
+ )
def transport_from_stream(
@@ -113,4 +96,4 @@ def transport_from_stream(
key = (codec_key, transport)
- return _msg_transports[key]
+ return _key_to_transport[key]
diff --git a/tractor/ipc/_uds.py b/tractor/ipc/_uds.py
index eb2e7f32..ee147d42 100644
--- a/tractor/ipc/_uds.py
+++ b/tractor/ipc/_uds.py
@@ -18,13 +18,12 @@ Unix Domain Socket implementation of tractor.ipc._transport.MsgTransport protoco
'''
from __future__ import annotations
-import tempfile
-from uuid import uuid4
import trio
from tractor.msg import MsgCodec
from tractor.log import get_logger
+from tractor._addr import UDSAddress
from tractor.ipc._transport import MsgpackTransport
@@ -37,9 +36,8 @@ class MsgpackUDSStream(MsgpackTransport):
using the ``msgspec`` codec lib.
'''
- address_type = str
+ address_type = UDSAddress
layer_key: int = 7
- name_key: str = 'uds'
# def __init__(
# self,
@@ -54,19 +52,32 @@ class MsgpackUDSStream(MsgpackTransport):
# codec=codec
# )
+ @property
+ def maddr(self) -> str:
+ filepath = self.raddr.unwrap()
+ return (
+ f'/ipv4/localhost'
+ f'/{self.address_type.name_key}/{filepath}'
+ # f'/{self.chan.uid[0]}'
+ # f'/{self.cid}'
+
+ # f'/cid={cid_head}..{cid_tail}'
+ # TODO: ? not use this ^ right ?
+ )
+
def connected(self) -> bool:
return self.stream.socket.fileno() != -1
@classmethod
async def connect_to(
cls,
- filename: str,
+ addr: UDSAddress,
prefix_size: int = 4,
codec: MsgCodec|None = None,
**kwargs
) -> MsgpackUDSStream:
stream = await trio.open_unix_socket(
- filename,
+ addr.unwrap(),
**kwargs
)
return MsgpackUDSStream(
@@ -79,16 +90,8 @@ class MsgpackUDSStream(MsgpackTransport):
def get_stream_addrs(
cls,
stream: trio.SocketStream
- ) -> tuple[str, str]:
+ ) -> tuple[UDSAddress, UDSAddress]:
return (
- stream.socket.getsockname(),
- stream.socket.getpeername(),
+ UDSAddress.from_addr(stream.socket.getsockname()),
+ UDSAddress.from_addr(stream.socket.getsockname()),
)
-
- @classmethod
- def get_random_addr(self) -> str:
- return f'{tempfile.gettempdir()}/{uuid4()}.sock'
-
- @classmethod
- def get_root_addr(self) -> str:
- return 'tractor.sock'
diff --git a/tractor/msg/types.py b/tractor/msg/types.py
index 76d0bad6..3e58ae3a 100644
--- a/tractor/msg/types.py
+++ b/tractor/msg/types.py
@@ -46,8 +46,8 @@ from msgspec import (
from tractor.msg import (
pretty_struct,
)
-from tractor.ipc import AddressTypes
from tractor.log import get_logger
+from tractor._addr import AddressTypes
log = get_logger('tractor.msgspec')