Skip to content

Add has_receivers/has_senders and fix BroadcastChannel auto-close - #565

Open
shsms wants to merge 15 commits into
frequenz-floss:v1.x.xfrom
shsms:broadcast-channel-has-receivers
Open

shsms wants to merge 15 commits into
frequenz-floss:v1.x.xfrom
shsms:broadcast-channel-has-receivers

Conversation

@shsms

@shsms shsms commented Oct 20, 2026

Copy link
Copy Markdown
Contributor

BroadcastChannel (new in 1.12) had several gaps in how it closes itself, and producers had no way to tell whether anyone was still listening. This branch fixes the gaps and adds BroadcastSender.has_receivers and BroadcastReceiver.has_senders.

Changes

  • Closing a receiver now wakes a task that waits on it. Receivers wait on an Event that is replaced after each wake-up, instead of an asyncio.Condition.
  • Garbage-collected senders now count as closed (through weakref.finalize). The finalizer can run in any thread, or in the middle of an await, so it only updates the channel's state and wakes the receivers through call_soon_threadsafe(). aclose() still wakes them directly. Please look closely at _remove_sender(), _wake_receivers_soon(), BroadcastReceiver.ready() and their tests in tests/test_broadcast.py.
  • Garbage-collected receivers detach at once. The channel still closes on the next send() without receivers, as in 1.12, so "close the initial receiver, subscribe later" keeps working.
  • clone() refuses a closed sender or channel. subscribe() uses the limit and warn_on_overflow the channel was created with as its defaults.
  • BroadcastChannel no longer emits Broadcast's deprecation warning. The implementation moved into a private _Broadcast class. The Broadcast docstring moved with it, so git diff --color-moved shows the real changes.
  • Docs: the user guide, README, package overview and all docstring examples use BroadcastChannel, there is a new Oneshot page, and several false claims about closing are fixed.

Breaking

  • Garbage-collected broadcast senders count as closed: an auto-closing channel closes when its last sender is dropped, and sender_count no longer counts dropped senders.
  • BroadcastSender.clone() raises SenderClosedError or SenderError on a closed sender or channel.
  • BroadcastSender.subscribe() takes its limit and warn_on_overflow defaults from the channel.
  • For errors from a BroadcastChannel caused by a ChannelClosedError, isinstance(error.__cause__.channel, Broadcast) is now False.

@shsms
shsms requested a review from a team as a code owner October 20, 2026 16:08
@shsms
shsms requested review from Marenz and removed request for a team October 20, 2026 16:08
@github-actions github-actions Bot added part:docs Affects the documentation part:tests Affects the unit, integration and performance (benchmarks) tests part:channels Affects channels implementation part:synchronization Affects the synchronization of multiple sources (`select`, `merge`) part:core Affects the core types (`Sender`, `Receiver`, exceptions, etc.) part:utilities Affects the utility receivers (`Timer`, `Event`, `FileWatcher`) part:experimental Affects the experimental package labels Oct 20, 2026
@shsms
shsms marked this pull request as draft October 20, 2026 17:19
shsms added 15 commits October 23, 2026 13:29
`BroadcastReceiver.close()` is synchronous and couldn't notify the
channel's condition variable, so a task waiting in `ready()` hung until
the next message arrived or the channel was closed.  Receivers now wait
on an event that can be set from synchronous code.

Waiters left on a closed event loop are skipped, so they don't make
`close()` fail.  For the same reason, `send()` and `aclose()` no longer
raise `RuntimeError: Event loop is closed` because of them.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
A closed receiver with messages left in its buffer kept returning True
from `ready()`, while `consume()` raised `ReceiverStoppedError`.  So
`select()` yielded stopped selections for it forever, and a `filter()`
on it made `select()` fail.  A closed receiver is now never ready.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Senders that were garbage-collected without `aclose()` stayed in the
sender count, so their receivers waited forever.  Senders now detach
from the channel when they are closed or garbage-collected, and an
auto-closing channel, like `BroadcastChannel`, closes when the last one
is gone.

The garbage collector can detach a sender in any thread, or in the
middle of `Event.wait()`.  So the detach only updates the channel's
state, and wakes up the receivers through `call_soon_threadsafe()` on
the event loop they wait on.  An explicit `aclose()` still wakes them up
at once.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Like tokio's broadcast channel, a send to an auto-closing broadcast
channel without receivers still fails, but now leaves the channel open,
so new receivers can still be subscribed.  `send()` raises a
`NoReceiversError` and drops the message.  With `resend_latest=True`,
new receivers get the last message that was sent successfully instead.
Such a channel now closes itself only when its senders are gone.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Garbage-collected receivers stayed in the channel until the next
`send()` swept them out.  They now detach as soon as they are collected,
so the channel doesn't keep dead receivers between sends.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Unlike `send()` and `subscribe()`, `clone()` worked on a closed sender
or channel.  On a closed channel, it handed out senders that could never
send.  It now raises like `send()` and `subscribe()`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
`limit` and `warn_on_overflow` were documented as applying to the
channel's receivers, but only the initial receiver got them;
`subscribe()` always fell back to 50 and `True`.  They are now the
defaults for `subscribe()`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Lets producers stop when nobody is listening anymore, and consumers tell
whether more messages can still arrive, without having to send or wait
on the channel.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
`BroadcastChannel` was built on the deprecated `Broadcast` class, so
creating one warned users to "use BroadcastChannel instead".  The
implementation now lives in a private class that both use, and only
`Broadcast` is deprecated.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
About auto-closing, the docs only said that closing all senders closes
all receivers, and what a send without receivers does.  They now also
cover garbage-collected senders and receivers, what each side sees, and
`has_receivers`/`has_senders`.

The multi-sender example is fixed too: it claimed to use a small buffer
to make the senders block, it caught a `ReceiverStoppedError` that
`async for` never raises, and its receivers slept once after the loop
instead of after each message.

The first example now stops when its sender is closed.  The
`BroadcastSender` and `BroadcastReceiver` docs now point to
`BroadcastChannel`, the `resend_latest` docs point to `subscribe()`, and
`Broadcast` is marked as deprecated.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
The `aclose()` docs said that receiving after the buffer is drained
returns `None`, and the `BroadcastReceiver.close()` docs said that the
buffer can still be drained.  Both now say when a `ReceiverStoppedError`
is raised.  The receiver guide and `Receiver.close()` said every closed
receiver drains its buffer first; they now say this depends on the
receiver.

The multi-sender example of the deprecated `Broadcast` had the same
mistakes as the `BroadcastChannel` one, and it never closed its channel,
so it never ended.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
The user guide, package overview and README still presented the
deprecated `Broadcast` and `Anycast` classes.  They now show
`BroadcastChannel` and mark `Anycast` as deprecated, and a new page
covers `OneshotChannel`.  The README showcase now closes its data
senders, and its receiver stops when the data channels close.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
The examples used the deprecated `Anycast` and `Broadcast` classes.
Also fix some examples with missing imports, and the latest value cache
examples, which checked the cache before its task could run.  The
`Anycast` docs now point to `BroadcastChannel`.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
A `Pipe` whose receiver stopped, or whose send failed, ended but kept
its sender open, so the receivers of the sender's channel could wait
forever.  Since a broadcast channel now stays open without receivers,
receivers subscribed after a failed send waited forever too.  And the
pipe's receiver stayed attached to its source channel, so
`has_receivers` stayed `True` there.

Close the sender in both cases, and the receiver too when the send
failed.  `stop()` still leaves both open.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
@shsms
shsms force-pushed the broadcast-channel-has-receivers branch from 64050f2 to e723c15 Compare October 23, 2026 11:37
@shsms
shsms marked this pull request as ready for review October 23, 2026 12:16

@Marenz Marenz left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 LGTM. This fixes real bugs, and the tests cover the hard cases (GC in another thread, closed loops, double detach). The release notes are accurate. ci_checks_max and pytest_min pass locally (152 tests).

Nits:

  1. The PR description says "The channel still closes on the next send() without receivers, as in 1.12". The code does the opposite: send() raises NoReceiversError and keeps the channel open (_broadcast.py:499), as commit 8278c97 and the release notes say. Please fix the description and add this to the "Breaking" list (before, that send closed the channel and raised SenderError with a ChannelClosedError cause).
  2. BroadcastReceiver.close() wakes receivers directly (_broadcast.py:756), while the sender finalizer uses call_soon_threadsafe(). With asyncio debug off, calling close() from another thread leaves a waiting task stuck without any error; the thread test only covers debug mode. A docstring note is probably enough, or route it through _wake_receivers_soon().
  3. _sender_count -= 1 (_broadcast.py:182) runs in the finalizer, which can be any thread, and is not atomic. Worth a comment that only the wake-up is thread-safe.
  4. _Broadcast is private, so there is no public type left to check ChannelClosedError.channel against. Maybe say so in the release notes.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

part:channels Affects channels implementation part:core Affects the core types (`Sender`, `Receiver`, exceptions, etc.) part:docs Affects the documentation part:experimental Affects the experimental package part:synchronization Affects the synchronization of multiple sources (`select`, `merge`) part:tests Affects the unit, integration and performance (benchmarks) tests part:utilities Affects the utility receivers (`Timer`, `Event`, `FileWatcher`)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants