goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 10 Packages Projects Releases Wiki Activity

Drop 'IPC' prefix from `._server` types 

We already have the `.ipc` sub-pkg name so it seems a bit
redundant/noisy for a namespace path Bp

Leave an alias for the `Server` rn since it's already used in a few
other internal mods.. will likely rename later if everyone is cool with
it..
enable_tpts
Tyler Goodlet 2025-06-17 23:33:58 -04:00
parent b71afdc615
commit bff32b0ad7
2 changed files with 23 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/ipc/test_server.py
View File

@ -49,7 +49,7 @@ def test_basic_ipc_server(
) )
assert server._no_more_peers.is_set() assert server._no_more_peers.is_set()
eps: list[ipc.IPCEndpoint] = await server.listen_on( eps: list[ipc._server.Endpoint] = await server.listen_on(
accept_addrs=[rando_addr], accept_addrs=[rando_addr],
stream_handler_nursery=None, stream_handler_nursery=None,
) )

40
tractor/ipc/_server.py
View File

@ -289,7 +289,7 @@ async def maybe_wait_on_canced_subs(
# #
# -[x] maybe change to mod-func and rename for implied # -[x] maybe change to mod-func and rename for implied
# multi-transport semantics? # multi-transport semantics?
# -[ ] register each stream/tpt/chan with the owning `IPCEndpoint` # -[ ] register each stream/tpt/chan with the owning `Endpoint`
# so that we can query per tpt all peer contact infos? # so that we can query per tpt all peer contact infos?
# |_[ ] possibly provide a global viewing via a # |_[ ] possibly provide a global viewing via a
# `collections.ChainMap`? # `collections.ChainMap`?
@ -309,7 +309,7 @@ async def handle_stream_from_peer(
any `IPCServer.listen_on()` passed `stream_handler_tn: Nursery` any `IPCServer.listen_on()` passed `stream_handler_tn: Nursery`
such that it is invoked as, such that it is invoked as,
IPCEndpoint.stream_handler_tn.start_soon( Endpoint.stream_handler_tn.start_soon(
handle_stream, handle_stream,
stream, stream,
) )
@ -577,7 +577,7 @@ async def handle_stream_from_peer(
# finally block closure # finally block closure
class IPCEndpoint(Struct): class Endpoint(Struct):
''' '''
An instance of an IPC "bound" address where the lifetime of the An instance of an IPC "bound" address where the lifetime of the
"ability to accept connections" (from clients) and then handle "ability to accept connections" (from clients) and then handle
@ -636,7 +636,7 @@ class IPCEndpoint(Struct):
) )
class IPCServer(Struct): class Server(Struct):
_parent_tn: Nursery _parent_tn: Nursery
_stream_handler_tn: Nursery _stream_handler_tn: Nursery
# level-triggered sig for whether "no peers are currently # level-triggered sig for whether "no peers are currently
@ -644,7 +644,7 @@ class IPCServer(Struct):
# initialized with `.is_set() == True`. # initialized with `.is_set() == True`.
_no_more_peers: trio.Event _no_more_peers: trio.Event
_endpoints: list[IPCEndpoint] = [] _endpoints: list[Endpoint] = []
# connection tracking & mgmt # connection tracking & mgmt
_peers: defaultdict[ _peers: defaultdict[
@ -659,10 +659,10 @@ class IPCServer(Struct):
# syncs for setup/teardown sequences # syncs for setup/teardown sequences
_shutdown: trio.Event|None = None _shutdown: trio.Event|None = None
# TODO, maybe just make `._endpoints: list[IPCEndpoint]` and # TODO, maybe just make `._endpoints: list[Endpoint]` and
# provide dict-views onto it? # provide dict-views onto it?
# @property # @property
# def addrs2eps(self) -> dict[Address, IPCEndpoint]: # def addrs2eps(self) -> dict[Address, Endpoint]:
# ... # ...
@property @property
@ -708,7 +708,7 @@ class IPCServer(Struct):
await self._shutdown.wait() await self._shutdown.wait()
else: else:
tpt_protos: list[str] = [] tpt_protos: list[str] = []
ep: IPCEndpoint ep: Endpoint
for ep in self._endpoints: for ep in self._endpoints:
tpt_protos.append(ep.addr.proto_key) tpt_protos.append(ep.addr.proto_key)
@ -790,7 +790,7 @@ class IPCServer(Struct):
def epsdict(self) -> dict[ def epsdict(self) -> dict[
Address, Address,
IPCEndpoint, Endpoint,
]: ]:
return { return {
ep.addr: ep ep.addr: ep
@ -804,7 +804,7 @@ class IPCServer(Struct):
return ev.is_set() return ev.is_set()
def pformat(self) -> str: def pformat(self) -> str:
eps: list[IPCEndpoint] = self._endpoints eps: list[Endpoint] = self._endpoints
state_repr: str = ( state_repr: str = (
f'{len(eps)!r} IPC-endpoints active' f'{len(eps)!r} IPC-endpoints active'
@ -835,13 +835,13 @@ class IPCServer(Struct):
# TODO? maybe allow shutting down a `.listen_on()`s worth of # TODO? maybe allow shutting down a `.listen_on()`s worth of
# listeners by cancelling the corresponding # listeners by cancelling the corresponding
# `IPCEndpoint._listen_tn` only ? # `Endpoint._listen_tn` only ?
# -[ ] in theory you could use this to # -[ ] in theory you could use this to
# "boot-and-wait-for-reconnect" of all current and connecting # "boot-and-wait-for-reconnect" of all current and connecting
# peers? # peers?
# |_ would require that the stream-handler is intercepted so we # |_ would require that the stream-handler is intercepted so we
# can intercept every `MsgTransport` (stream) and track per # can intercept every `MsgTransport` (stream) and track per
# `IPCEndpoint` likely? # `Endpoint` likely?
# #
# async def unlisten( # async def unlisten(
# self, # self,
@ -854,7 +854,7 @@ class IPCServer(Struct):
*, *,
accept_addrs: list[tuple[str, int|str]]|None = None, accept_addrs: list[tuple[str, int|str]]|None = None,
stream_handler_nursery: Nursery|None = None, stream_handler_nursery: Nursery|None = None,
) -> list[IPCEndpoint]: ) -> list[Endpoint]:
''' '''
Start `SocketListeners` (i.e. bind and call `socket.listen()`) Start `SocketListeners` (i.e. bind and call `socket.listen()`)
for all IPC-transport-protocol specific `Address`-types for all IPC-transport-protocol specific `Address`-types
@ -888,7 +888,7 @@ class IPCServer(Struct):
f'Binding to endpoints for,\n' f'Binding to endpoints for,\n'
f'{accept_addrs}\n' f'{accept_addrs}\n'
) )
eps: list[IPCEndpoint] = await self._parent_tn.start( eps: list[Endpoint] = await self._parent_tn.start(
partial( partial(
_serve_ipc_eps, _serve_ipc_eps,
server=self, server=self,
@ -904,7 +904,7 @@ class IPCServer(Struct):
self._endpoints.extend(eps) self._endpoints.extend(eps)
# XXX, just a little bit of sanity # XXX, just a little bit of sanity
group_tn: Nursery|None = None group_tn: Nursery|None = None
ep: IPCEndpoint ep: Endpoint
for ep in eps: for ep in eps:
if ep.addr not in self.addrs: if ep.addr not in self.addrs:
breakpoint() breakpoint()
@ -917,6 +917,10 @@ class IPCServer(Struct):
return eps return eps
# alias until we decide on final naming
IPCServer = Server
async def _serve_ipc_eps( async def _serve_ipc_eps(
*, *,
server: IPCServer, server: IPCServer,
@ -941,12 +945,12 @@ async def _serve_ipc_eps(
listen_tn: Nursery listen_tn: Nursery
async with trio.open_nursery() as listen_tn: async with trio.open_nursery() as listen_tn:
eps: list[IPCEndpoint] = [] eps: list[Endpoint] = []
# XXX NOTE, required to call `serve_listeners()` below. # XXX NOTE, required to call `serve_listeners()` below.
# ?TODO, maybe just pass `list(eps.values()` tho? # ?TODO, maybe just pass `list(eps.values()` tho?
listeners: list[trio.abc.Listener] = [] listeners: list[trio.abc.Listener] = []
for addr in listen_addrs: for addr in listen_addrs:
ep = IPCEndpoint( ep = Endpoint(
addr=addr, addr=addr,
listen_tn=listen_tn, listen_tn=listen_tn,
stream_handler_tn=stream_handler_tn, stream_handler_tn=stream_handler_tn,
@ -1010,7 +1014,7 @@ async def _serve_ipc_eps(
finally: finally:
if eps: if eps:
addr: Address addr: Address
ep: IPCEndpoint ep: Endpoint
for addr, ep in server.epsdict().items(): for addr, ep in server.epsdict().items():
ep.close_listener() ep.close_listener()
server._endpoints.remove(ep) server._endpoints.remove(ep)