Commit Graph

1145 Commits (28513bc6013e64c79c0d859ed65ff04e892b8ba3)

Author SHA1 Message Date
Tyler Goodlet 2745a2b1dc Solve first-recv-cancelled by recursive `.receive()` on wake 2021-09-02 21:12:54 -04:00
Tyler Goodlet 5881a82d2a Add a first receiver is cancelled test 2021-09-02 21:12:54 -04:00
Tyler Goodlet b7b489dd07 Drop shielded stream api usage 2021-09-02 21:12:54 -04:00
Tyler Goodlet d9bb52fe7b Store array `maxlen` in state singleton
The `collections.deque` takes care of array length truncation of values
for us implicitly but in the future we'll likely want this value exposed
to alternate array implementations. This patch is to provide for that as
well as make `mypy` happy since the `dequeu.maxlen` can also be `None`.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 9258f79510 Don't wake sibling bcast consumers on a cancelled call 2021-09-02 21:12:54 -04:00
Tyler Goodlet 5c6355062c Shorten sequence length for test speedup 2021-09-02 21:12:54 -04:00
Tyler Goodlet 44ef26bb18 Shorten default feeder mem chan size to 64 2021-09-02 21:12:54 -04:00
Tyler Goodlet d9e793d4ba Can't use built-in generics till 3.9... 2021-09-02 21:12:54 -04:00
Tyler Goodlet 7857a9ac6d Add `shield: bool` kwarg to `Portal.open_stream_from()` 2021-09-02 21:12:54 -04:00
Tyler Goodlet 5182ee7782 Add a "faster task is cancelled" test 2021-09-02 21:12:54 -04:00
Tyler Goodlet 39cf9af9fc Rename test module 2021-09-02 21:12:54 -04:00
Tyler Goodlet 63ec740e27 Add some bcaster ref sanity asserts around subscriptions 2021-09-02 21:12:54 -04:00
Tyler Goodlet 0d70e3081a Add laggy parent stream tests
Add a couple more tests to check that a parent and sub-task stream can
be lagged and recovered (depending on who's slower). Factor some of the
test machinery into a new ctx mngr to make it all happen.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 093e7d921c Instance ids are ints 2021-09-02 21:12:54 -04:00
Tyler Goodlet d7ad8982ff Add subscribe after close test 2021-09-02 21:12:54 -04:00
Tyler Goodlet bec3f5999d Drop uuid4 keys, raise closed error on subscription after close 2021-09-02 21:12:54 -04:00
Tyler Goodlet 2bad2bac50 Don't enable debug mode..it borks CI 2021-09-02 21:12:54 -04:00
Tyler Goodlet a4cb0ef21f Fix `.receive()` re-assignment, drop `.clone()` 2021-09-02 21:12:54 -04:00
Tyler Goodlet 236ed0b0dd Initial broadcaster tests including one to test our `MsgStream.subscribe()` api 2021-09-02 21:12:54 -04:00
Tyler Goodlet 346b5d2eda Blade runner it
Get rid of all the (requirements for) clones of the underlying
receivable. We can just use a uuid generated key for each instance
(thinking now this can probably just be `id(self)`). I'm fully convinced
now that channel cloning is only a source of confusion and anti-patterns
when we already have nurseries to define resource lifetimes. There is no
benefit in particular when you allocate subscriptions using a context
manager (not sure why `trio.open_memory_channel()` doesn't enforce
this).

Further refinements:
- add a `._closed` state that will error the receiver on reuse
- drop module script section;  it's been moved to a real test
- call the "receiver" duck-type stub a new name
2021-09-02 21:12:54 -04:00
Tyler Goodlet 6c17c7367a Store handle to underlying channel's `.receive()`
This allows for wrapping an existing stream by re-assigning its receive
method to the allocated broadcaster's `.receive()` so as to avoid
expecting any original consumer(s) of the stream to now know about the
broadcaster; this instead mutates the stream to delegate to the new
receive call behind the scenes any time `.subscribe()` is called.

Add a `typing.Protocol` for so called "cloneable channels" until we
decide/figure out a better keying system for each subscription and
mask all undesired typing failures.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 2d1c24112b Add subscription support to message streams
Add `ReceiveMsgStream.subscribe()` which allows allocating a broadcast
receiver around the stream for use by multiple actor-local consumer
tasks. Entering this context manager idempotently mutates the stream's
receive machinery which for now can not be undone. Move `.clone()` to
the receive stream type.

Resolves #204
2021-09-02 21:12:54 -04:00
Tyler Goodlet a12b1fc631 Drop optimization check, binance made its point 2021-09-02 21:12:54 -04:00
Tyler Goodlet ceed96aa3f Add common state delegate type for all consumers
For every set of broadcast receivers which pull from the same producer,
we need a singleton state for all of,
- subscriptions
- the sender ready event
- the queue

Add a `BroadcastState` dataclass for this and pass it to all
subscriptions. This makes the design much more like the built-in memory
channels which do something very similar with `MemoryChannelState`.

Use a `filter()` on the subs list in the sequence update step, plus some
other commented approaches we can try for speed.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 6e78bcf898 Facepalm: use single `_subs` per clone set 2021-09-02 21:12:54 -04:00
Tyler Goodlet 4ad75a3287 Obviously keying on tasks isn't going to work
Using the current task as a subscription key fails horribly as soon as
you hand off new subscription receiver to another task you've spawned..

Instead use the underlying ``trio.abc.ReceiveChannel.clone()`` as a key
(so i guess we're assuming cloning is supported by the underlying?)
which makes this all work just like default mem chans. As a bonus, now
we can just close the underlying rx (which may be a clone) on
`.aclose()` and everything should just work in terms of the underlying
channels lifetime (i think?).

Change `.subscribe()` to be async since the receive channel type
interface only expects `.aclose()` and it actually ends up being
nicer for 3.9+ style `async with` parentheses style anyway.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 64358f6525 Rename to broadcast mod, don't expect mem chan specifically 2021-09-02 21:12:54 -04:00
Tyler Goodlet 1af7dbb732 `Task` is hashable, so key on it 2021-09-02 21:12:54 -04:00
Tyler Goodlet 6a2c3da1bb Simplify api around receive channel
Buncha improvements:
- pass in the queue via constructor
- tracking over all underlying memory channel closure using cloning
- do it like `tokio` and set lagged consumers to the last sequence
  before raising
- copy the subs on first receiver wakeup for iteration instead of
  iterating the table directly (and being forced to skip the current
  tasks sequence increment)
- implement `.aclose()` to close the underlying clone for this task
- make `broadcast_receiver()` just take the recv chan since it doesn't
  need anything on the send side.
2021-09-02 21:12:54 -04:00
Tyler Goodlet 3817b4fb5e Ultra naive broadcast channel prototype 2021-09-02 21:12:54 -04:00
goodboy 87ce6c8eb3
Merge pull request #234 from goodboy/root_tty_hangs
Root tty hangs
2021-09-02 16:58:25 -04:00
Tyler Goodlet 76342ed0c5 Add news bit 2021-09-02 16:50:32 -04:00
Tyler Goodlet 3208b67f57 Drop shielding on root lock acquire; seems to prevent hangs 2021-09-02 16:23:38 -04:00
Tyler Goodlet 61d2307e52 Unlock pdb tty on all possible net faults 2021-09-02 16:23:38 -04:00
Tyler Goodlet 79f0d6fda0 Attempt to avoid pdb lockups on channel breakage
Always try to release the root tty lock on broken connection errors.
2021-09-02 16:23:10 -04:00
Tyler Goodlet 4f166500d0 Add return type to debugger factory 2021-09-02 16:22:59 -04:00
Tyler Goodlet d906c81f14 Export portal type at top level 2021-09-02 16:22:59 -04:00
Tyler Goodlet 68d56d5df0 Try not masking SIGINT in child processes 2021-09-02 16:22:59 -04:00
Tyler Goodlet 497fa72c96 Add a SIGINT handler that kills the process tree
We're not actually using this but it's for reference if we do end up
needing it.

The std lib's `pdb` internals override SIGINT handling whenever one
enters the debugger repl. Force a handler that kills the tree if SIGINT
is triggered from the root actor, otherwise ignore it since supervised
children should be managed already. This resolves an issue with guest
mode where `pdb` causes SIGINTs to be swallowed resulting in the host
loop never terminating the process tree.
2021-09-02 16:22:02 -04:00
goodboy e5845b5d36
Merge pull request #230 from goodboy/drop_stream_shielding
Drop stream shielding; it was from a legacy api design
2021-09-02 16:18:42 -04:00
goodboy 7e98afa685
Merge pull request #233 from goodboy/drop_py37
Drop py37
2021-09-02 15:00:12 -04:00
Tyler Goodlet 22a79219a1 Lol, guess windows needs the extra minutes 2021-09-02 08:35:31 -04:00
Tyler Goodlet 3919c9739e Make fragment a `.rst` 2021-09-02 08:35:25 -04:00
Tyler Goodlet 558c44fdbe Add newsfragment 2021-09-02 08:33:29 -04:00
Tyler Goodlet b4d95e9543 Update docs to new close semantics 2021-09-02 08:24:18 -04:00
Tyler Goodlet af85d35685 Drop stream shielding; it was from a legacy design
The whole origin was not having an explicit open/close semantic for
streams. We have that now so this internal mechanic isn't needed and
further our streams become more correct by having `.aclose()` be
independent of cancellation.
2021-09-02 08:24:18 -04:00
Tyler Goodlet b176628206 Drop 3.7 support from install script
Resolves #232
2021-09-02 07:51:33 -04:00
Tyler Goodlet 47a469484d Drop py3.7 from CI; cut run to 5mins 2021-09-02 07:51:33 -04:00
goodboy 07e43f88bf
Merge pull request #231 from goodboy/add_the_crier
Use `towncrier`
2021-09-02 07:50:15 -04:00
Tyler Goodlet a221949e8f Add small howto instructions 2021-09-02 07:40:43 -04:00