Added ringd actor to broker ring buf resources dynamically
Added ring pubsub based on ringd
Created tractor.linux submodule and moved eventfd stuff there
Implemented linux file descriptor ipc share async helpers
Type hint all ctx managers in _ringbuf.py
Remove unnecesary send lock on ring chan sender
Handle EOF on ring chan receiver
Rename ringbuf tests to make it less redundant
Add EOF signaling mechanism
Support proper `receive_some` end of stream semantics
Add StapledStream non-ipc test
Create MsgpackRBStream similar to MsgpackTCPStream for buffered whole-msg reads
Add EventFD.read cancellation on EventFD.close mechanism using cancel scope
Add test for eventfd cancellation
Improve and add docstrings
Add trio.StrictFIFOLock on sender.send_all
Support max_bytes argument on receive_some, keep track of write_ptr on receiver
Add max_bytes receive test test_ringbuf_max_bytes
Add docstrings to all ringbuf tests
Remove EFD_NONBLOCK support, not necesary anymore since we can use abandon_on_cancel=True on trio.to_thread.run_sync
Close eventfd's after usage on open_ringbuf
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.
Namely while what I was actually trying to solve was why
`TransportClosed` was getting raised from `Portal.cancel_actor()` but
still useful edge case auditing either way. Also opts into the
`debug_mode` fixture with apprope timeout adjustment B)
Just like we *were* for the `trio`-resource-errors it normally wraps
since we now also do the same wrapping in `MsgpackTransport.send()`
and we don't normally care to raise tpt-closure-errors on graceful actor
cancel requests.
Also, warn-report any non-tpt-closed low-level `trio` errors we haven't
yet re-wrapped (likely bc they haven't shown up).
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.
In `tests/test_advanced_faults.py` that is.
Since instead of zero-responses like we'd expect from a network-socket
we actually can get a few differences from the OS when "everything IPC
is known"
XD
Namely it's about underlying `trio` exceptions versus how we wrap them
and how we expect to box them. A `TransportClosed` boxing improvement
is coming in follow up btw to make this all work!
B)
Via a new accumulative `--tpt-proto` arg you can select which
`tpt_protos: list[str]`-fixture protocol keys will be delivered to
opting in tests!
B)
Also includes,
- CLI quote handling/stripping.
- default of 'tcp'.
- only support one selection per session at the moment (until we figure
out how we want to support multiples, either simultaneously or
sequentially).
- draft a (masked) dynamic-`metafunc` parametrization in the
`pytest_generate_tests()` hook.
- first proven and working use in the `test_advanced_faults`-suite (and
thus its underlying
`examples/advanced_faults/ipc_failure_during_stream.py` script)!
|_ actually needed this to prove that the suite only has 2 failures on
'uds' seemingly due to low-level `trio` error semantics translation
differences to do with with calling `socket.close()`..
On a very nearly related topic,
- draft an (also commented out) `set_script_runtime_args()` fixture idea
for a std way of `partial`-ling in runtime args to `examples/`
scripts-as-modules defining a `main()` which would proxy to
`tractor.open_nursery()`.
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.
Such that the "global-ish" setting (actor-local) is managed with the
others per actor-process and type it as a `Literal['tcp', 'uds']` of the
currently support protocol keys.
Here obvi `_tpt` is some kinda shorthand for "transport" and `_proto` is
for "protocol" Bp
Change imports and refs in all dependent modules.
Oh right, and disable UDS in `wrap_address()` for the moment while
i figure out how to avoid the unwrapped type collision..
There was a very strange legacy test
`test_spawning.test_local_arbiter_subactor_global_state` which was
causing unforseen hangs/errors on the UDS tpt and looking deeper this
test was already doing root-actor things that should never have been
valid XD
So rework that test to properly demonstrate something of value
(i guess..) and add a new suite which start more rigorously auditing our
`open_root_actor()` permitted usage.
For the old test,
- since the main point of this test seemed to be the ability to invoke
the same function in both the parent and child actor (using the very
legacy `ActorNursery.run_in_actor()`.. due to be deprecated) rename it
to `test_run_in_actor_same_func_in_child`,
- don't re-enter `.open_root_actor()` since that's invalid usage (tested
in new suite see below),
- adjust some `spawn()` arg/var naming and ensure we only return in the
child.
For the new suite add tests for,
- ensuring the implicit `open_root_actor()` call under `open_nursery()`.
- double open of `open_root_actor()` from within the same process tree
both from a root and sub.
Intro some new `_exceptions` used in the new suite,
- a top level `RuntimeFailure` for generically expressing faults not of
our own doing that prevent successful operation; this is what we now
(changed in this commit) raise on attempts to open a 2nd root.
- mk `ActorFailure` derive from the former; it's already used from
`._spawn` when subprocs fail to boot.
- aggregate the `MsgStream.aclose()` "reader tasks" stats content into a
common `message: str` before emit.
- tweak an `_rpc.process_messages()` emit per new `Channel.__repr__()`.
Throughout `_context` and `_spawn` where it causes no big disruption.
Still lots to work out for things like how to pass `--uid
<tuple-as-str>` to spawned subactors and whether we want a diff name for
the minimum `tuple` required to distinguish a subactor pre-process-ID
allocation by the OS.
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`.
The more I think about it, it seems @guille's orig approach of
unwrapping UDS socket-file addresses to strings (or `Path`) is making
the most sense. I had originally thought that pairing it with the
listening side's pid would add clarity (and it definitely does for
introspection/debug/logging) but since we don't end up passing that pid
to the eventual `.connect()` call on the client side, it doesn't make
much sense to wrap it for the wire just to discard.. Further, the
`tuple[str, int]` makes `wrap_address()` break for TCP since it will
always match on uds first.
So, on that note this patch refines a few things in prep for going back
to that original `UnwrappedAddress` as `str` type though longer run
i think the more "builtin approach" would be to add `msgspec` codec
hooks for these types to avoid all the `.wrap()`/`.unwrap()` calls
throughout the runtime.
Down-low deats,
- add `wrap_address()` doc string, detailed (todo) comments and handle
the `[None, None]` case that can come directly from
`._state._runtime_vars['_root_mailbox']`.
- buncha adjustments to `UDSAddress`,
- add a `filedir`, chng `filepath` -> `filename` and mk `maybe_pid` optional.
- the intent `filedir` is act as the equivalent of the host part in a network proto's
socket address and when it's null use the `.def_bindspace = get_rt_dir()`.
- always ensure the `filedir / filename` is an absolute path and
expose it as a new `.sockpath: Path` property.
- mk `.is_valid` actually verify the `.sockpath` is in the valid
`.bindspace: namely just checking it's in the expected dir.
- add pedantic `match:`ing to `.from_addr()` such that we error on
unexpected `type(addr)` inputs and otherwise parse any `sockpath:
Path` inputs using a new `unwrap_sockpath()` which simply splits an
abs file path to dir, file-name parts.
- `.unwrap()` now just `str`-ifies the `.sockpath: Path`
- adjust `.open/close_listener()` to use `.sockpath`.
Such that we can run (opting-in) tests on both TCP and UDS backends and
ensure the `reg_addr` fixture and various timeouts are adjusted
accordingly.
Impl deats,
- add a new `tpc_proto` CLI option and fixture to allow choosing which
"transport protocol" will be used in the test suites (either globally
or contextually).
- rm `_reg_addr` instead opting for a `_rando_port` which will only be
used for `reg_addr`s which are net-tpt-protos.
- rejig `reg_addr` fixture to set a ideally session-unique `testrun_reg_addr`
based on the `tpt_proto` setting making appropriate calls to `._addr`
APIs as needed.
- refine `daemon` fixture a bit with typing, `tpt_proto` timings, and
stderr capture.
- in `test_discovery` do a ton of type-annots, add `debug_mode` fixture
opt ins, augment `spawn_and_check_registry()` with `psutil.Process`
passing for introspection (when things go wrong..).
Call it `maybe_block_bp()` can wrap the `open_root_actor()` body with
it. Main reason is to guarantee we can bp inside actor runtime bootup as
needed when debugging internals! Prolly should factor this to another
module tho?
ALSO, ensure we RTE on recurrent entries to `open_root_actor()` from
within an existing tree! There was actually `test_spawning` test somehow
getting away with this!? Should never be possible or allowed!
And map `.__repr__/__str__` to it and add various new fields to fill it
out,
- drop `self.uid` as var and instead add `Actor._aid: Aid` and proxy to
it for the various `.name/.uid/.pid` properties as well as a new
`.aid` field.
|_ the `Aid.pid` addition is also included.
Other improvements,
- flip to a sync call to `Address.close_listener()`.
- track the `async_main()` parent task as `Actor._task`.
- add exception logging around failure to bind due to already-in-use
when calling `add.open_listener()` in `._stream_forever()`; sometimes
the error might be overridden by something else during the
runtime-failure unwind..
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.
Guillermo Rodriguez
Tyler Goodlet