Repository navigation
Conversation
shsms
marked this pull request as draft
October 20, 2026 17:19
`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
force-pushed
the
broadcast-channel-has-receivers
branch
from
October 23, 2026 11:37
64050f2 to
e723c15
Compare
shsms
marked this pull request as ready for review
October 23, 2026 12:16
Marenz
approved these changes
Oct 23, 2026
Marenz
left a comment
Contributor
There was a problem hiding this comment.
🤖 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:
- The PR description says "The channel still closes on the next
send()without receivers, as in 1.12". The code does the opposite:send()raisesNoReceiversErrorand 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 raisedSenderErrorwith aChannelClosedErrorcause). BroadcastReceiver.close()wakes receivers directly (_broadcast.py:756), while the sender finalizer usescall_soon_threadsafe(). With asyncio debug off, callingclose()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()._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._Broadcastis private, so there is no public type left to checkChannelClosedError.channelagainst. Maybe say so in the release notes.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 addsBroadcastSender.has_receiversandBroadcastReceiver.has_senders.Changes
Eventthat is replaced after each wake-up, instead of anasyncio.Condition.weakref.finalize). The finalizer can run in any thread, or in the middle of anawait, so it only updates the channel's state and wakes the receivers throughcall_soon_threadsafe().aclose()still wakes them directly. Please look closely at_remove_sender(),_wake_receivers_soon(),BroadcastReceiver.ready()and their tests intests/test_broadcast.py.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 thelimitandwarn_on_overflowthe channel was created with as its defaults.BroadcastChannelno longer emitsBroadcast's deprecation warning. The implementation moved into a private_Broadcastclass. TheBroadcastdocstring moved with it, sogit diff --color-movedshows the real changes.BroadcastChannel, there is a new Oneshot page, and several false claims about closing are fixed.Breaking
sender_countno longer counts dropped senders.BroadcastSender.clone()raisesSenderClosedErrororSenderErroron a closed sender or channel.BroadcastSender.subscribe()takes itslimitandwarn_on_overflowdefaults from the channel.BroadcastChannelcaused by aChannelClosedError,isinstance(error.__cause__.channel, Broadcast)is nowFalse.