Move peer-tracking attrs from Actor -> IPCServer

Namely transferring the `Actor` peer-`Channel` tracking attrs,
- `._peers` which maps the uids to client channels (with duplicates
  apparently..)
- the `._peer_connected: dict[tuple[str, str], trio.Event]` child-peer
  syncing table mostly used by parent actors to wait on sub's to connect
  back during spawn.
- the `._no_more_peers = trio.Event()` level triggered state signal.

Further we move over with some minor reworks,
- `.wait_for_peer()` verbatim (adjusting all dependants).
- factor the no-more-peers shielded wait branch-block out of
  the end of `async_main()` into 2 new server meths,
  * `.has_peers()` with optional chan-connected checking flag.
  * `.wait_for_no_more_peers()` which *just* does the
    maybe-shielded `._no_more_peers.wait()`

Author: goodboy

tests/test_spawning.py

@@ -57,7 +57,11 @@ async def spawn(
             )
 
             assert len(an._children) == 1
-            assert portal.channel.uid in tractor.current_actor()._peers
+            assert (
+                portal.channel.uid
+                in
+                tractor.current_actor().ipc_server._peers
+            )
 
             # get result from child subactor
             result = await portal.result()

tractor/_discovery.py

@@ -48,6 +48,7 @@
 
 if TYPE_CHECKING:
     from ._runtime import Actor
+    from .ipc._server import IPCServer
 
 
 log = get_logger(__name__)
@@ -79,7 +80,7 @@ async def get_registry(
         )
     else:
         # TODO: try to look pre-existing connection from
-        # `Actor._peers` and use it instead?
+        # `IPCServer._peers` and use it instead?
         async with (
             _connect_chan(addr) as chan,
             open_portal(chan) as regstr_ptl,
@@ -111,14 +112,15 @@ def get_peer_by_name(
 ) -> list[Channel]|None:  # at least 1
     '''
     Scan for an existing connection (set) to a named actor
-    and return any channels from `Actor._peers`.
+    and return any channels from `IPCServer._peers: dict`.
 
     This is an optimization method over querying the registrar for
     the same info.
 
     '''
     actor: Actor = current_actor()
-    to_scan: dict[tuple, list[Channel]] = actor._peers.copy()
+    server: IPCServer = actor.ipc_server
+    to_scan: dict[tuple, list[Channel]] = server._peers.copy()
     pchan: Channel|None = actor._parent_chan
     if pchan:
         to_scan[pchan.uid].append(pchan)

tractor/_runtime.py

@@ -40,9 +40,7 @@
 from contextlib import (
     ExitStack,
 )
-from collections import defaultdict
 from functools import partial
-from itertools import chain
 import importlib
 import importlib.util
 import os
@@ -76,6 +74,7 @@
 )
 from .ipc import (
     Channel,
+    # IPCServer,  # causes cycles atm..
     _server,
 )
 from ._addr import (
@@ -156,7 +155,6 @@ def is_registrar(self) -> bool:
     _root_n: Nursery|None = None
     _service_n: Nursery|None = None
 
-    # XXX moving to IPCServer!
     _ipc_server: _server.IPCServer|None = None
 
     @property
@@ -246,14 +244,6 @@ def __init__(
         # by the user (currently called the "arbiter")
         self._spawn_method: str = spawn_method
 
-        self._peers: defaultdict[
-            str,  # uaid
-            list[Channel],  # IPC conns from peer
-        ] = defaultdict(list)
-        self._peer_connected: dict[tuple[str, str], trio.Event] = {}
-        self._no_more_peers = trio.Event()
-        self._no_more_peers.set()
-
         # RPC state
         self._ongoing_rpc_tasks = trio.Event()
         self._ongoing_rpc_tasks.set()
@@ -338,7 +328,12 @@ def pformat(self) -> str:
         parent_uid: tuple|None = None
         if rent_chan := self._parent_chan:
             parent_uid = rent_chan.uid
-        peers: list[tuple] = list(self._peer_connected)
+
+        peers: list = []
+        server: _server.IPCServer = self.ipc_server
+        if server:
+            peers: list[tuple] = list(server._peer_connected)
+
         fmtstr: str = (
             f' |_id: {self.aid!r}\n'
             # f"   aid{ds}{self.aid!r}\n"
@@ -394,25 +389,6 @@ def reg_addrs(
 
         self._reg_addrs = addrs
 
-    async def wait_for_peer(
-        self,
-        uid: tuple[str, str],
-
-    ) -> tuple[trio.Event, Channel]:
-        '''
-        Wait for a connection back from a (spawned sub-)actor with
-        a `uid` using a `trio.Event` for sync.
-
-        '''
-        log.debug(f'Waiting for peer {uid!r} to connect')
-        event = self._peer_connected.setdefault(uid, trio.Event())
-        await event.wait()
-        log.debug(f'{uid!r} successfully connected back to us')
-        return (
-            event,
-            self._peers[uid][-1],
-        )
-
     def load_modules(
         self,
         # debug_mode: bool = False,
@@ -724,7 +700,7 @@ async def _from_parent(
             )
             assert isinstance(chan, Channel)
 
-            # Initial handshake: swap names.
+            # init handshake: swap actor-IDs.
             await chan._do_handshake(aid=self.aid)
 
             accept_addrs: list[UnwrappedAddress]|None = None
@@ -1620,16 +1596,18 @@ async def async_main(
                     )
 
         # Ensure all peers (actors connected to us as clients) are finished
-        if not actor._no_more_peers.is_set():
-            if any(
-                chan.connected() for chan in chain(*actor._peers.values())
-            ):
-                teardown_report += (
-                    f'-> Waiting for remaining peers {actor._peers} to clear..\n'
-                )
-                log.runtime(teardown_report)
-                with CancelScope(shield=True):
-                    await actor._no_more_peers.wait()
+        if (
+            (ipc_server := actor.ipc_server)
+            and
+            ipc_server.has_peers(check_chans=True)
+        ):
+            teardown_report += (
+                f'-> Waiting for remaining peers {ipc_server._peers} to clear..\n'
+            )
+            log.runtime(teardown_report)
+            await ipc_server.wait_for_no_more_peers(
+                shield=True,
+            )
 
         teardown_report += (
             '-> All peer channels are complete\n'

tractor/_spawn.py

@@ -58,9 +58,11 @@
 
 
 if TYPE_CHECKING:
+    from ipc import IPCServer
     from ._supervise import ActorNursery
     ProcessType = TypeVar('ProcessType', mp.Process, trio.Process)
 
+
 log = get_logger('tractor')
 
 # placeholder for an mp start context if so using that backend
@@ -481,6 +483,7 @@ async def trio_proc(
 
     cancelled_during_spawn: bool = False
     proc: trio.Process|None = None
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         try:
             proc: trio.Process = await trio.lowlevel.open_process(spawn_cmd, **proc_kwargs)
@@ -492,7 +495,7 @@ async def trio_proc(
             # wait for actor to spawn and connect back to us
             # channel should have handshake completed by the
             # local actor by the time we get a ref to it
-            event, chan = await actor_nursery._actor.wait_for_peer(
+            event, chan = await ipc_server.wait_for_peer(
                 subactor.uid
             )
 
@@ -724,11 +727,12 @@ async def mp_proc(
 
     log.runtime(f"Started {proc}")
 
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         # wait for actor to spawn and connect back to us
         # channel should have handshake completed by the
         # local actor by the time we get a ref to it
-        event, chan = await actor_nursery._actor.wait_for_peer(
+        event, chan = await ipc_server.wait_for_peer(
             subactor.uid,
         )
 

tractor/_supervise.py

@@ -53,6 +53,9 @@
 
 if TYPE_CHECKING:
     import multiprocessing as mp
+    # from .ipc._server import IPCServer
+    from .ipc import IPCServer
+
 
 log = get_logger(__name__)
 
@@ -315,6 +318,9 @@ async def cancel(
         children: dict = self._children
         child_count: int = len(children)
         msg: str = f'Cancelling actor nursery with {child_count} children\n'
+
+        server: IPCServer = self._actor.ipc_server
+
         with trio.move_on_after(3) as cs:
             async with trio.open_nursery(
                 strict_exception_groups=False,
@@ -337,7 +343,7 @@ async def cancel(
 
                     else:
                         if portal is None:  # actor hasn't fully spawned yet
-                            event = self._actor._peer_connected[subactor.uid]
+                            event: trio.Event = server._peer_connected[subactor.uid]
                             log.warning(
                                 f"{subactor.uid} never 't finished spawning?"
                             )
@@ -353,7 +359,7 @@ async def cancel(
                             if portal is None:
                                 # cancelled while waiting on the event
                                 # to arrive
-                                chan = self._actor._peers[subactor.uid][-1]
+                                chan = server._peers[subactor.uid][-1]
                                 if chan:
                                     portal = Portal(chan)
                                 else:  # there's no other choice left

tractor/devx/_debug.py

@@ -92,7 +92,11 @@
 if TYPE_CHECKING:
     from trio.lowlevel import Task
     from threading import Thread
-    from tractor.ipc import Channel
+    from tractor.ipc import (
+        Channel,
+        IPCServer,
+        # _server,  # TODO? export at top level?
+    )
     from tractor._runtime import (
         Actor,
     )
@@ -1434,6 +1438,7 @@ def any_connected_locker_child() -> bool:
 
     '''
     actor: Actor = current_actor()
+    server: IPCServer = actor.ipc_server
 
     if not is_root_process():
         raise InternalError('This is a root-actor only API!')
@@ -1443,7 +1448,7 @@ def any_connected_locker_child() -> bool:
         and
         (uid_in_debug := ctx.chan.uid)
     ):
-        chans: list[tractor.Channel] = actor._peers.get(
+        chans: list[tractor.Channel] = server._peers.get(
             tuple(uid_in_debug)
         )
         if chans: