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.
Such that re-wrapping/raising from a low-level `trio` resource error is
simpler and includes the `.src_exc` in the `__repr__()` and
`.message/.args` rendered at higher layers (like from `Channel` and
`._rpc` machinery).
Impl deats,
- mainly leverages packing in a new cls-method `.repr_src_exc() -> str:`
repr of the underlying error before an optional `body: str` all as
handled by the previously augmented `.pformat()`'s delegation to
`pformat_exc()`.
- change `.src_exc` to be a property around a renamed `._src_exc`.
But wait, why?
- use it inside `MsgpackTransport.send()` to rewrap any
`trio.BrokenResourceError`s so we always see the underlying
`trio`-src-exc just like in the `.recv()._iter_packets()` handlers.
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.
That is moving from `._addr`,
- `TCPAddress` to `.ipc._tcp`
- `UDSAddress` to `.ipc._uds`
Obviously this requires adjusting a buncha stuff in `._addr` to avoid
import cycles (the original reason the module was not also included in
the new `.ipc` subpkg) including,
- avoiding "unnecessary" imports of `[Unwrapped]Address` in various modules.
* since `Address` is a protocol and the main point is that it **does
not need to be inherited** per
(https://typing.python.org/en/latest/spec/protocol.html#terminology)
thus I removed the need for it in both transport submods.
* and `UnwrappedAddress` is a type alias for tuples.. so we don't
really always need to be importing it since it also kinda obfuscates
what the underlying pairs are.
- not exporting everything in submods at the `.ipc` top level and
importing from specific submods by default.
- only importing various types under a `if typing.TYPE_CHECKING:` guard
as needed.
By borrowing from the implementation of `RemoteActorError.pformat()`
which is now factored into a new `.devx.pformat_exc()` and re-used for
both error types while maintaining the same func-sig. Obviously delegate
`RemoteActorError.pformat()` to the new helper accordingly and keeping
the prior `body` generation from `.devx.pformat_boxed_tb()` as before.
The new helper allows for,
- passing any of a `header|message|body: str` which are all combined in
that order in the final output.
- getting the `exc.message` as the default `message` part.
- generating an objecty-looking "type-name" header to be rendered by
default when `header` is not overridden.
- "first-line-of `message`" processing which we split-off and then
re-inject as a `f'<{type(exc).__name__}( {first} )>'` top line header.
- an optional `tail: str = '>'` to "close the object"-look only added
when `with_type_header: bool = True`.
Adjustments to `TransportClosed` around this include,
- replacing the init `cause` arg for a `src_exc` which is now always
assigned to a same named instance var.
- displaying that new `.src_exc` in the `body: str` arg to the
`.devx.pformat.pformat_exc()` call so you can always see the
underlying (normally `trio`) source error.
- just make it inherit from `Exception` not `trio.BrokenResourceError`
to avoid handlers catching `TransportClosed` as the former
particularly in testing when we want to sometimes to distinguish them.
Since in hindsight the real analog of a net-proto's "bindspace"
(normally its routing layer's addresses-port-set) is more akin to the
"location in the file-system" for a UDS socket file (aka the file's
parent directory) determines whether or not the "port" (aka it's
file-name) collides with any other.
So the `._filedir: Path` is like the allocated "address" and,
the `._filename: Path|str` is basically the "port",
at least in my mind.. Bp
Thinking about fs dirs like a "host address" means you can get
essentially the same benefits/behaviour of say an (ip)
addresses-port-space but using the (current process-namespace's)
filesys-tree. Note that for UDS sockets in particular the
network-namespace is what would normally isolate so called "abstract
sockets" (i.e. UDS sockets that do NOT use file-paths by setting `struct
sockaddr_un.sun_path = 'abstract', see `man unix`); using directories is
even easier and definitely more explicit/readable/immediately-obvious as
a human-user.
As such this reworks all the necessary `UDSAddress` meths,
- `.unwrap()` now returns a `tuple(str(._filedir, str(._filename))`,
- `wrap_address()` now matches UDS on a 2nd tuple `str()` element,
- `.get_root()` no longer passes `maybe_pid`.
AND adjusts `MsgpackUDSStream` to,
- use the new `unwrap_sockpath()` on the `socket.get[sock/peer]name()`
output before passing directly as `UDSAddress.__init__(filedir, filename)`
instead of via `.from_addr()`.
- also pass `maybe_pid`s to init since no longer included in the
unwrapped-type form.
Finally.. i've been meaning todo this for ages since the
actor-id-swap-as-handshake is better layered as part of the IPC msg-ing
machinery and then let's us encapsulate the connection-time-assignment
of a remote peer's `Aid` as a new `Channel.aid: Aid`. For now we
continue to offer the `.uid: tuple[str, str]` attr (by delegating to the
`.uid` field) since there's still a few things relying on it in the
runtime and ctx layers
Nice bonuses from this,
- it's very easy to get the peer's `Aid.pid: int` from anywhere in an
IPC ctx by just reading it from the chan.
- we aren't saving more then the wire struct-msg received.
Also add deprecation warnings around usage to get us moving on porting
the rest of consuming runtime code to the new attr!
For the case where there's clearly no socket file created/bound
obviously the `trio.socket.connect()` call will raise
`FileNotFoundError`, so just translate this to
a builtin-`ConnectionError` at the transport layer so we can report the
guilty `UDSAddress`.
Namely reducing the duplication of class-fields and `TypeVar`s used
for parametrizing the `Address` protocol type,
- drop all of the `TypeVar` types and just stick with all concrete addrs
types inheriting from `Address` only.
- rename `Address.name_key` -> `.proto_key`.
- rename `Address.address_type` -> `.unwrapped_type`
- rename `.namespace` -> `.bindspace` to better reflect that this "part"
of the address represents the possible "space for binding endpoints".
|_ also linux already uses "namespace" to mean the `netns` and i'd
prefer to stick with their semantics for that.
- add `TCPAddress/UDSAddress.def_bindspace` values.
- drop commented `.open_stream()` method; never used.
- simplify `UnwrappedAdress` to just a `tuple` of union types.
- add logging to `USDAddress.open_listener()` for now.
- adjust `tractor.ipc/_uds/tcp` transport to use new addr field names.
Much like we already do in the `._iter_packets()` async-generator which
delivers to `.recv()` and `async for`, handle the `''[Errno 32] Broken
pipe'` case that can show up with unix-domain-socket usage.
Seems like the cause is due to how fast the socket can be torn down
during a registry addr channel ping where,
- the sending side can break the connection faster then the pong side
can prep its handshake msg,
- the pong side tries to send it's handshake pkt via
`.SocketStream.send_all()` after the breakage and then raises
`trio.BrokenResourceError`.
Much like how `Context` has been implemented, try to give tons of high
level details on all the lower level encapsulated primitives, namely the
`.msgstream/.transport` and any useful runtime state.
B)
Impl deats,
- adjust `.from_addr()` to only call `._addr.wrap_address()` when we
detect `addr` is unwrapped.
- add another `log.runtime()` using the new `.__repr__()` in
`Channel.from_addr()`.
- change to `UnwrappedAddress` as in prior commits.
Such that any UDS socket pair is represented (and with the recent
updates to) a `USDAddress` via a similar pair-`tuple[str, int]` as TCP
sockets, a pair of the `.filepath: Path` & the peer proc's `.pid: int`
which we read from the underlying `socket.socket` using
`.set/getsockopt()` calls
Impl deats,
- using the Linux specific APIs, we add a `get_peer_info()` which reads
the `(pid, uid, gid)` using the `SOL_SOCKET` and `SOL_PEECRED` opts to
`sock.getsockopt()`.
|_ this presumes the client has been correspondingly configured to
deliver the creds via a `sock.setsockopt(SOL_SOCKET, SO_PASSCRED,
1)` call - this required us to override `trio.open_unix_socket()`.
- override `trio.open_unix_socket()` as per the above bullet to ensure
connecting peers always transmit "credentials" options info to the
listener.
- update `.get_stream_addrs()` to always call `get_peer_info()` and
extract the peer's pid for the `raddr` and use `os.getpid()` for
`laddr` (obvi).
|_ as part of the new impl also `log.info()` the creds-info deats and
socket-file path.
|_ handle the oddity where it depends which of `.getpeername()` or
`.getsockname()` will return the file-path; i think it's to do with
who is client vs. server?
Related refinements,
- set `.layer_key: int = 4` for the "transport layer" ;)
- tweak some typing and multi-line unpacking in `.ipc/_tcp`.
Tyler Goodlet
Guillermo Rodriguez