Multihomed transport (server) addrs 🕶️

examples/full_fledged_streaming_service.py

@@ -65,21 +65,28 @@ async def push_to_chan(portal, send_chan):
     print("AGGREGATOR COMPLETE!")
 
 
-# this is the main actor and *arbiter*
-async def main():
-    # a nursery which spawns "actors"
-    async with tractor.open_nursery(
-        arbiter_addr=('127.0.0.1', 1616)
-    ) as nursery:
+async def main() -> list[int]:
+    '''
+    This is the "root" actor's main task's entrypoint.
+
+    By default (and if not otherwise specified) that root process
+    also acts as a "registry actor" / "registrar" on the localhost
+    for the purposes of multi-actor "service discovery".
+
+    '''
+    # yes, a nursery which spawns `trio`-"actors" B)
+    nursery: tractor.ActorNursery
+    async with tractor.open_nursery() as nursery:
 
         seed = int(1e3)
         pre_start = time.time()
 
-        portal = await nursery.start_actor(
+        portal: tractor.Portal = await nursery.start_actor(
             name='aggregator',
             enable_modules=[__name__],
         )
 
+        stream: tractor.MsgStream
         async with portal.open_stream_from(
             aggregate,
             seed=seed,

tests/conftest.py

@@ -29,7 +29,7 @@ async def test_whatever():
 
     If fixtures:
 
-        - ``arb_addr`` (a socket addr tuple where arbiter is listening)
+        - ``reg_addr`` (a socket addr tuple where arbiter is listening)
         - ``loglevel`` (logging level passed to tractor internals)
         - ``start_method`` (subprocess spawning backend)
 
@@ -40,16 +40,16 @@ async def test_whatever():
     def wrapper(
         *args,
         loglevel=None,
-        arb_addr=None,
+        reg_addr=None,
         start_method=None,
         **kwargs
     ):
         # __tracebackhide__ = True
 
-        if 'arb_addr' in inspect.signature(fn).parameters:
+        if 'reg_addr' in inspect.signature(fn).parameters:
             # injects test suite fixture value to test as well
             # as `run()`
-            kwargs['arb_addr'] = arb_addr
+            kwargs['reg_addr'] = reg_addr
 
         if 'loglevel' in inspect.signature(fn).parameters:
             # allows test suites to define a 'loglevel' fixture
@@ -71,7 +71,7 @@ def wrapper(
             async def _main():
                 async with tractor.open_root_actor(
                     # **kwargs,
-                    arbiter_addr=arb_addr,
+                    registry_addrs=[reg_addr] if reg_addr else None,
                     loglevel=loglevel,
                     start_method=start_method,
 
@@ -92,9 +92,6 @@ async def _main():
     return wrapper
 
 
-_arb_addr = '127.0.0.1', random.randint(1000, 9999)
-
-
 # Sending signal.SIGINT on subprocess fails on windows. Use CTRL_* alternatives
 if platform.system() == 'Windows':
     _KILL_SIGNAL = signal.CTRL_BREAK_EVENT
@@ -173,9 +170,23 @@ def ci_env() -> bool:
     return _ci_env
 
 
+# choose randomly at import time
+_reg_addr: tuple[str, int] = (
+    '127.0.0.1',
+    random.randint(1000, 9999),
+)
+
+
 @pytest.fixture(scope='session')
-def arb_addr():
-    return _arb_addr
+def reg_addr() -> tuple[str, int]:
+
+    # globally override the runtime to the per-test-session-dynamic
+    # addr so that all tests never conflict with any other actor
+    # tree using the default.
+    from tractor import _root
+    _root._default_lo_addrs = [_reg_addr]
+
+    return _reg_addr
 
 
 def pytest_generate_tests(metafunc):
@@ -216,30 +227,35 @@ def sig_prog(proc, sig):
 def daemon(
     loglevel: str,
     testdir,
-    arb_addr: tuple[str, int],
+    reg_addr: tuple[str, int],
 ):
     '''
-    Run a daemon actor as a "remote arbiter".
+    Run a daemon root actor as a separate actor-process tree and
+    "remote registrar" for discovery-protocol related tests.
 
     '''
     if loglevel in ('trace', 'debug'):
-        # too much logging will lock up the subproc (smh)
-        loglevel = 'info'
-
-    cmdargs = [
-        sys.executable, '-c',
-        "import tractor; tractor.run_daemon([], registry_addr={}, loglevel={})"
-        .format(
-            arb_addr,
-            "'{}'".format(loglevel) if loglevel else None)
+        # XXX: too much logging will lock up the subproc (smh)
+        loglevel: str = 'info'
+
+    code: str = (
+            "import tractor; "
+            "tractor.run_daemon([], registry_addrs={reg_addrs}, loglevel={ll})"
+    ).format(
+        reg_addrs=str([reg_addr]),
+        ll="'{}'".format(loglevel) if loglevel else None,
+    )
+    cmd: list[str] = [
+        sys.executable,
+        '-c', code,
     ]
-    kwargs = dict()
+    kwargs = {}
     if platform.system() == 'Windows':
         # without this, tests hang on windows forever
         kwargs['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP
 
     proc = testdir.popen(
-        cmdargs,
+        cmd,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         **kwargs,

tests/test_cancellation.py

@@ -47,7 +47,7 @@ async def do_nuthin():
     ],
     ids=['no_args', 'unexpected_args'],
 )
-def test_remote_error(arb_addr, args_err):
+def test_remote_error(reg_addr, args_err):
     """Verify an error raised in a subactor that is propagated
     to the parent nursery, contains the underlying boxed builtin
     error type info and causes cancellation and reraising all the
@@ -57,7 +57,7 @@ def test_remote_error(arb_addr, args_err):
 
     async def main():
         async with tractor.open_nursery(
-            arbiter_addr=arb_addr,
+            registry_addrs=[reg_addr],
         ) as nursery:
 
             # on a remote type error caused by bad input args
@@ -97,15 +97,15 @@ async def main():
             assert exc.type == errtype
 
 
-def test_multierror(arb_addr):
+def test_multierror(reg_addr):
     '''
     Verify we raise a ``BaseExceptionGroup`` out of a nursery where
     more then one actor errors.
 
     '''
     async def main():
         async with tractor.open_nursery(
-            arbiter_addr=arb_addr,
+            registry_addrs=[reg_addr],
         ) as nursery:
 
             await nursery.run_in_actor(assert_err, name='errorer1')
@@ -130,14 +130,14 @@ async def main():
 @pytest.mark.parametrize(
     'num_subactors', range(25, 26),
 )
-def test_multierror_fast_nursery(arb_addr, start_method, num_subactors, delay):
+def test_multierror_fast_nursery(reg_addr, start_method, num_subactors, delay):
     """Verify we raise a ``BaseExceptionGroup`` out of a nursery where
     more then one actor errors and also with a delay before failure
     to test failure during an ongoing spawning.
     """
     async def main():
         async with tractor.open_nursery(
-            arbiter_addr=arb_addr,
+            registry_addrs=[reg_addr],
         ) as nursery:
 
             for i in range(num_subactors):
@@ -175,15 +175,20 @@ async def do_nothing():
 
 
 @pytest.mark.parametrize('mechanism', ['nursery_cancel', KeyboardInterrupt])
-def test_cancel_single_subactor(arb_addr, mechanism):
-    """Ensure a ``ActorNursery.start_actor()`` spawned subactor
+def test_cancel_single_subactor(reg_addr, mechanism):
+    '''
+    Ensure a ``ActorNursery.start_actor()`` spawned subactor
     cancels when the nursery is cancelled.
-    """
+
+    '''
     async def spawn_actor():
-        """Spawn an actor that blocks indefinitely.
-        """
+        '''
+        Spawn an actor that blocks indefinitely then cancel via
+        either `ActorNursery.cancel()` or an exception raise.
+
+        '''
         async with tractor.open_nursery(
-            arbiter_addr=arb_addr,
+            registry_addrs=[reg_addr],
         ) as nursery:
 
             portal = await nursery.start_actor(

tests/test_child_manages_service_nursery.py

@@ -141,7 +141,7 @@ async def open_actor_local_nursery(
 )
 def test_actor_managed_trio_nursery_task_error_cancels_aio(
     asyncio_mode: bool,
-    arb_addr
+    reg_addr: tuple,
 ):
     '''
     Verify that a ``trio`` nursery created managed in a child actor