diff --git a/tests/discovery/test_daemon_fixture.py b/tests/discovery/test_daemon_fixture.py index d8899e42..84a80ca3 100644 --- a/tests/discovery/test_daemon_fixture.py +++ b/tests/discovery/test_daemon_fixture.py @@ -1,6 +1,10 @@ ''' Discovery daemon fixture regressions. +This module imports private helpers from the sibling +`tests.discovery.conftest` plugin to exercise that fixture machinery +directly, rather than testing a production `tractor` API. + ''' from unittest.mock import ( call, diff --git a/tests/discovery/test_tpt_bind_addrs.py b/tests/discovery/test_tpt_bind_addrs.py index ca65587e..538b4e8b 100644 --- a/tests/discovery/test_tpt_bind_addrs.py +++ b/tests/discovery/test_tpt_bind_addrs.py @@ -2,7 +2,8 @@ `open_root_actor(tpt_bind_addrs=...)` test suite. Verify all three runtime code paths for explicit IPC-server -bind-address selection in `_root.py`: +bind-address selection in `_root.py` and registry probing in +`discovery._api`: 1. Non-registrar, no explicit bind -> random addrs from registry proto 2. Registrar, no explicit bind -> binds to registry_addrs @@ -19,11 +20,12 @@ from unittest.mock import ( import pytest import trio import tractor -from tractor import _root +from tractor.discovery import _api from tractor.discovery._addr import ( wrap_address, ) from tractor.discovery._multiaddr import mk_maddr +from tractor.ipc import _connect_chan from tractor._testing.addr import get_rando_addr @@ -68,11 +70,11 @@ def test_registry_probe_retries_transient_handshake( closed.append(chan) sleep = AsyncMock() - monkeypatch.setattr(_root, '_connect_chan', connect_chan) - monkeypatch.setattr(_root.trio, 'sleep', sleep) + monkeypatch.setattr(_api, '_connect_chan', connect_chan) + monkeypatch.setattr(_api.trio, 'sleep', sleep) async def main(): - status = await _root._probe_registry( + status = await _api._probe_registry( addr=wrap_address(('127.0.0.1', 1616)), timeout=.3, attempt_timeout=.1, @@ -114,7 +116,7 @@ def test_probe_channel_close_is_bounded( async def main(): with trio.fail_after(.5): - async with _root._connect_chan( + async with _connect_chan( ('127.0.0.1', 1616), close_timeout=.01, ): @@ -193,7 +195,7 @@ def test_registry_probe_preserves_no_peers_state( actor = tractor.current_actor() server = actor.ipc_server - probe_status = await _root._probe_registry( + probe_status = await _api._probe_registry( addr=wrap_address(reg_addr), ) assert probe_status == 'registrar' diff --git a/tractor/_root.py b/tractor/_root.py index 28696918..a598526b 100644 --- a/tractor/_root.py +++ b/tractor/_root.py @@ -31,7 +31,6 @@ import sys from typing import ( Any, Callable, - Literal, ) import warnings @@ -48,9 +47,7 @@ from .devx import ( from .spawn import _spawn from .runtime import _state from . import log -from .ipc import ( - _connect_chan, -) +from .discovery._api import _probe_registry_addrs from .discovery._addr import ( Address, UnwrappedAddress, @@ -64,9 +61,7 @@ from .trionics import ( ) from ._exceptions import ( RuntimeFailure, - TransportClosed, ) -from .msg.types import Aid logger = log.get_logger('tractor') @@ -86,68 +81,6 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = ( ) -async def _probe_registry( - addr: Address, - timeout: float = 3, - attempt_timeout: float = 1, - max_attempts: int = 3, - retry_delay: float = .05, - close_timeout: float = .2, -) -> Literal[ - 'absent', - 'occupied', - 'registrar', -]: - ''' - Confirm an address serves the Tractor actor handshake. - - Connection and handshake work share `timeout`; each attempt gets - `attempt_timeout`. Shielded cleanup may add up to `close_timeout` - per attempted channel. - - ''' - connected_once: bool = False - with trio.move_on_after(timeout): - for attempt in range(max_attempts): - try: - with trio.move_on_after(attempt_timeout) as attempt_cs: - async with _connect_chan( - addr.unwrap(), - close_timeout=close_timeout, - ) as chan: - connected_once = True - peer_aid: Aid = await chan._do_handshake( - aid=Aid( - name='registry-probe', - uuid=mk_uuid(), - pid=os.getpid(), - is_probe=True, - ), - timeout=attempt_timeout, - ) - if peer_aid.is_registrar is not False: - return 'registrar' - return 'occupied' - - if attempt_cs.cancelled_caught: - if not connected_once: - return 'absent' - - except OSError: - return ( - 'occupied' - if connected_once - else 'absent' - ) - except TransportClosed: - pass - - if attempt + 1 < max_attempts: - await trio.sleep(retry_delay * (attempt + 1)) - - return 'occupied' - - # TODO: stick this in a `@acm` defined in `devx.debug`? # -[ ] also maybe consider making this a `wrapt`-deco to # save an indent level? @@ -516,46 +449,12 @@ async def open_root_actor( from .devx._stackscope import enable_stack_on_sig enable_stack_on_sig() - # closed into below ping task-func - ponged_addrs: list[Address] = [] - occupied_addrs: list[Address] = [] - - async def ping_tpt_socket( - addr: Address, - timeout: float = 3, - ) -> None: - ''' - Probe with a bounded Tractor actor handshake. - - Classify the address as a registrar, occupied by a - non-registrar, or absent. - - ''' - probe_status = await _probe_registry( - addr=addr, - timeout=timeout, - ) - if probe_status == 'registrar': - ponged_addrs.append(addr) - elif probe_status == 'occupied': - occupied_addrs.append(addr) - else: - # ?TODO, make this a "discovery" log level? - logger.info( - f'No root-actor registry found @ {addr!r}\n' - ) - - # !TODO, this is basically just another (abstract) - # happy-eyeballs, so we should try for formalize it somewhere - # in a `.[_]discovery` ya? - # - async with trio.open_nursery() as tn: - for uw_addr in uw_reg_addrs: - addr: Address = wrap_address(uw_addr) - tn.start_soon( - ping_tpt_socket, - addr, - ) + ponged_addrs: list[Address] + occupied_addrs: list[Address] + ( + ponged_addrs, + occupied_addrs, + ) = await _probe_registry_addrs(uw_reg_addrs) if ( not ponged_addrs diff --git a/tractor/discovery/_api.py b/tractor/discovery/_api.py index ec559baa..1691e2c8 100644 --- a/tractor/discovery/_api.py +++ b/tractor/discovery/_api.py @@ -21,14 +21,18 @@ management of (service) actors. """ from __future__ import annotations import ipaddress +import os import socket from typing import ( AsyncGenerator, AsyncContextManager, + Literal, TYPE_CHECKING, ) from contextlib import asynccontextmanager as acm +import trio + from tractor.log import get_logger from ..trionics import ( gather_contexts, @@ -40,6 +44,7 @@ from ..ipc._uds import UDSAddress from ._addr import ( UnwrappedAddress, Address, + mk_uuid, wrap_address, ) from ..runtime._portal import ( @@ -52,6 +57,7 @@ from ..runtime._state import ( _runtime_vars, _def_tpt_proto, ) +from ..msg.types import Aid if TYPE_CHECKING: from ..runtime._runtime import Actor @@ -60,6 +66,115 @@ if TYPE_CHECKING: log = get_logger() +async def _probe_registry( + addr: Address, + timeout: float = 3, + attempt_timeout: float = 1, + max_attempts: int = 3, + retry_delay: float = .05, + close_timeout: float = .2, +) -> Literal[ + 'absent', + 'occupied', + 'registrar', +]: + ''' + Confirm an address serves the Tractor actor handshake. + + Connection and handshake work share `timeout`; each attempt gets + `attempt_timeout`. Shielded cleanup may add up to `close_timeout` + per attempted channel. + + ''' + from .._exceptions import TransportClosed + + connected_once: bool = False + with trio.move_on_after(timeout): + for attempt in range(max_attempts): + try: + with trio.move_on_after(attempt_timeout) as attempt_cs: + async with _connect_chan( + addr.unwrap(), + close_timeout=close_timeout, + ) as chan: + connected_once = True + peer_aid: Aid = await chan._do_handshake( + aid=Aid( + name='registry-probe', + uuid=mk_uuid(), + pid=os.getpid(), + is_probe=True, + ), + timeout=attempt_timeout, + ) + if peer_aid.is_registrar is not False: + return 'registrar' + return 'occupied' + + if attempt_cs.cancelled_caught: + if not connected_once: + return 'absent' + + except OSError: + return ( + 'occupied' + if connected_once + else 'absent' + ) + except TransportClosed: + pass + + if attempt + 1 < max_attempts: + await trio.sleep(retry_delay * (attempt + 1)) + + return 'occupied' + + +async def _probe_registry_addrs( + addrs: list[UnwrappedAddress], + timeout: float = 3, +) -> tuple[ + list[Address], + list[Address], +]: + ''' + Concurrently classify candidate registrar addresses. + + Return confirmed registrar addresses followed by addresses occupied + by non-registrar or unresponsive Tractor peers. + + ''' + registrar_addrs: list[Address] = [] + occupied_addrs: list[Address] = [] + + async def probe_addr(addr: Address) -> None: + probe_status = await _probe_registry( + addr=addr, + timeout=timeout, + ) + if probe_status == 'registrar': + registrar_addrs.append(addr) + elif probe_status == 'occupied': + occupied_addrs.append(addr) + else: + # ?TODO, make this a "discovery" log level? + log.info( + f'No root-actor registry found @ {addr!r}\n' + ) + + async with trio.open_nursery() as nursery: + for unwrapped_addr in addrs: + nursery.start_soon( + probe_addr, + wrap_address(unwrapped_addr), + ) + + return ( + registrar_addrs, + occupied_addrs, + ) + + def _is_local_addr(addr: Address) -> bool: ''' Determine whether `addr` is reachable on the