As much as is possible given we currently do some graceful
cancellation join-waiting on any connected sub-actors whenever an active
`local_nursery: AcrtorNursery` in the post-rpc teardown sequence of
`handle_stream_from_peer()` is detected. In such cases we try to allow
the higher level inter-actor (task) context(s) to fully cancelled-ack
before conducting IPC machinery shutdown.
The main immediate motivation for all this is to support unit testing
the `.ipc._server` APIs but in the future may be useful for anyone
wanting to use our modular IPC transport layer sin-"actors".
Impl deats,
- drop passing an `actor: Actor` ref from as many routines in
`.ipc._server` as possible instead opting to use
`._state.current_actor()` where abs needed; thus the fns dropping an
`actor` input param are:
- `open_ipc_server()`
- `IPCServer.listen_on()`
- `._serve_ipc_eps()`
- `.handle_stream_from_peer()`
- factor the above mentioned graceful remote-cancel-ack waiting into
a new `maybe_wait_on_canced_subs()` which is called from
`handle_stream_from_peer()` and delivers a
maybe-`local_nursery: ActorNursery` for downstream logic; it's this
new fn which primarily still needs to call `current_actor()`.
- in `handle_stream_from_peer()` also use `current_actor()` to check if
a handshake is needed (or if it was called as part of some
actor-runtime-less operation like our unit test suite!).
- also don't pass an `actor` to `._rpc.process_messages()` see how-n-why
below..
Surrounding ipc-server client/caller adjustments,
- `._rpc.process_messages()` no longer takes an `actor` input and
now calls `current_actor()` instead.
- `._portal.open_portal()` is adjusted to ^.
- `._runtime.async_main()` is adjusted to the `.ipc._server`'s removal
of `actor` ref passing.
Also,
- drop some server `log.info()`s to `.runtime()`
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()`
Call it `handle_stream_from_peer()` and bind in the `actor: Actor` via
a `handler=partial()` to `trio.serve_listeners()`.
With this (minus the `Actor._peers/._peer_connected/._no_more_peers`
attrs ofc) we get nearly full separation of IPC-connection-processing
(concerns) from `Actor` state. Thus it's a first look at modularizing
the low-level runtime into isolated subsystems which will hopefully
improve the entire code base's grok-ability and ease any new feature
design discussions especially pertaining to introducing and/or
composing-together any new transport protocols.
Primarily moving the `Actor._serve_forever()`-task-as-method and
supporting actor-instance attributes to a new `.ipo._server` sub-mod
which now encapsulates,
- the coupling various `trio.Nursery`s (and their independent lifetime mgmt)
to different `trio.serve_listener()`s tasks and `SocketStream`
handler scopes.
- `Address` and `SocketListener` mgmt and tracking through the idea of
an "IPC endpoint": each "bound-and-active instance" of a served-listener
for some (varied transport protocol's socket) address.
- start and shutdown of the entire server's lifetime via an `@acm`.
- delegation of starting/stopping tpt-protocol-specific `trio.abc.Listener`s
to the corresponding `.ipc._<proto_key>` sub-module (newly defined
mod-top-level instead of `Address` method) `start/close_listener()`
funcs.
Impl details of the `.ipc._server` sub-sys,
- add new `IPCServer`, allocated with `open_ipc_server()`, and which
encapsulates starting multiple-transport-proto-`trio.abc.Listener`s
from an input set of `._addr.Address`s using,
|_`IPCServer.listen_on()` which internally spawns tasks that delegate to a new
`_serve_ipc_eps()`, a rework of what was (effectively)
`Actor._serve_forever()` and which now,
* allocates a new `IPCEndpoint`-struct (see below) for each
address-listener pair alongside the specified
listener-serving/stream-handling `trio.Nursery`s provided by the
caller.
* starts and stops each transport (socket's) listener by calling
`IPCEndpoint.start/close_listener()` which in turn delegates to
the underlying `inspect.getmodule(IPCEndpoint.addr)` backend tpt
module's equivalent impl.
* tracks all created endpoints in a `._endpoints: list[IPCEndpoint]`
which is further exposed through public properties for
introspection of served transport-protocols and their addresses.
|_`IPCServer._[parent/stream_handler]_tn: Nursery`s which are either
allocated (in which case, as the same instance) or provided by the
caller of `open_ipc_server()` such that the same nursery-cancel-scope
controls offered by `trio.serve_listeners(handler_nursery=)` are
offered where the `._parent_tn` is used to spawn `_serve_ipc_eps()`
tasks, and `._stream_handler_tn` is passed verbatim as `handler_nursery`.
- a new `IPCEndpoint`-struct (as mentioned) which wraps each
transport-proto's address + listener + allocated-supervising-nursery
to encapsulate the "lifetime of a server IPC endpoint" such that
eventually we can track and managed per-protocol/address/`.listen_on()`-call
scoped starts/stops/restarts for the purposes of filtering/banning
peer traffic.
|_ also included is an unused `.peer_tpts` table which we can
hopefully use to replace `Actor._peers` in a `Channel`-tracking
transport-proto-aware way!
Surrounding changes to `.ipc.*` primitives to match,
- make `[TCP|UDS]Address` types `msgspec.Struct(frozen=True)` and thus
drop any-and-all `addr._host =` style mutation throughout.
|_ as such also drop their `.__init__()` and `.__eq__()` meths.
|_ UDS tweaks to field names and thus `.__repr__()`.
- move `[TCP|UDS]Address.[start/close]_listener()` meths to be mod-level
equiv `start|close_listener()` funcs.
- just hard code the `.ipc._types._key_to_transport/._addr_to_transport`
table entries instead of all the prior fancy dynamic class property
reading stuff (remember, "explicit is better then implicit").
Modified in `._runtime.Actor` internals,
- drop the `._serve_forever()` and `.cancel_server()`, methods and
`._server_down` waiting logic from `.cancel_soon()`
- add `.[_]ipc_server` which is opened just after the `._service_n` and
delegate to it for any equivalent publicly exposed instance
attributes/properties.
Tyler Goodlet