7900: sync: add closed method to broadcast::WeakSender#60
7900: sync: add closed method to broadcast::WeakSender#60martin-augment wants to merge 2 commits intomasterfrom
closed method to broadcast::WeakSender#60Conversation
Summary of ChangesHello @martin-augment, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request enhances the Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
WalkthroughThe broadcast channel implementation is refactored to replace a single closure flag with two distinct flags: ✨ Finishing touches
🧪 Generate unit tests (beta)
Tip Issue Planner is now in beta. Read the docs and try it out! Share your feedback on Discord. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Code ReviewThank you for this PR! This adds the Positive Aspects✅ Better semantic clarity: Splitting ✅ Code reuse: Extracting ✅ Comprehensive test coverage: The updated tests thoroughly validate both ✅ Cooperative scheduling: The implementation correctly adds cooperative task yielding via ✅ Tracing support: The Issues Found🐛 Critical: Documentation Example Bug// In WeakSender::closed() documentation (lines 1137, 1141)
assert\!(tx.closed().now_or_never().is_none()); // ❌ Wrong
assert\!(tx.closed().now_or_never().is_some()); // ❌ WrongProblem: The example demonstrates Fix: Change to: assert\!(weak.closed().now_or_never().is_none());
assert\!(weak.closed().now_or_never().is_some());
|
| Category | Status |
|---|---|
| Code Quality | ✅ Good - clean refactoring |
| Best Practices | |
| Potential Bugs | 🐛 Documentation example bug |
| Performance | ✅ Good - adds cooperative scheduling |
| Security | ✅ No concerns |
| Test Coverage | ✅ Excellent |
Recommended Actions
- Fix the documentation example to use
weak.closed()instead oftx.closed() - Consider moving
async_trace_leaf()inside thecooperativeblock to match the pattern inwatch.rs - After fixes, the PR should be good to merge
Great work on this feature! The semantic clarity improvement is valuable, and the implementation is solid overall.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request adds a closed method to broadcast::WeakSender, which completes when all receivers have been dropped. To support this, the internal closed flag in the Tail struct has been split into closed_for_senders and closed_for_receivers for better clarity on the channel's state. The implementation looks solid and is well-tested. I have one minor suggestion to improve the documentation example.
| /// assert!(tx.closed().now_or_never().is_none()); | ||
| /// | ||
| /// assert_eq!(rx2.recv().await.unwrap(), 10); | ||
| /// drop(rx2); | ||
| /// assert!(tx.closed().now_or_never().is_some()); |
There was a problem hiding this comment.
The documentation example for WeakSender::closed incorrectly uses tx.closed() instead of weak.closed(). To accurately demonstrate the new method, the example should call closed() on the WeakSender instance.
| /// assert!(tx.closed().now_or_never().is_none()); | |
| /// | |
| /// assert_eq!(rx2.recv().await.unwrap(), 10); | |
| /// drop(rx2); | |
| /// assert!(tx.closed().now_or_never().is_some()); | |
| /// assert!(weak.closed().now_or_never().is_none()); | |
| /// | |
| /// assert_eq!(rx2.recv().await.unwrap(), 10); | |
| /// drop(rx2); | |
| /// assert!(weak.closed().now_or_never().is_some()); |
There was a problem hiding this comment.
value:useful; category:documentation; feedback: The Gemini AI reviewer is correct! The doc test is supposed to show that the WeakSender is not closed until the last channel receiver is dropped but instead it uses the strong Sender. Prevents releasing wrong documentation.
🤖 Augment PR SummarySummary: This PR adds a Changes:
🤖 Was this summary useful? React with 👍 or 👎 |
![augmentcode[bot]](https://avatars.githubusercontent.com/in/1027498?s=60&v=4){kind=small}
| /// | ||
| /// assert_eq!(rx1.recv().await.unwrap(), 10); | ||
| /// drop(rx1); | ||
| /// assert!(tx.closed().now_or_never().is_none()); |
There was a problem hiding this comment.
The WeakSender::closed docs example uses tx.closed() instead of weak.closed(), so it doesn't actually demonstrate the new API (and may mislead readers about usage when all strong senders are dropped). Consider updating the example to await/check weak.closed() specifically.
Severity: low
🤖 Was this useful? React with 👍 or 👎, or 🚀 if it prevented an incident/outage.
There was a problem hiding this comment.
value:useful; category:documentation; feedback: The Augment AI reviewer is correct! The doc test is supposed to show that the WeakSender is not closed until the last channel receiver is dropped but instead it uses the strong Sender. Prevents releasing wrong documentation.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@tokio/src/sync/broadcast.rs`:
- Around line 1118-1146: The doc example for WeakSender::closed() mistakenly
calls tx.closed() instead of demonstrating the weak handle; update the example
to call weak.closed() (the WeakSender returned by tx.downgrade()) in the two
assertions that currently use tx.closed() so the snippet exercises weak.closed()
as intended; keep the rest of the example (creating tx via broadcast::channel,
tx.downgrade(), recv and drops) unchanged.
🧹 Nitpick comments (1)
tokio/src/sync/broadcast.rs (1)
1599-1602: Setclosed_for_sendersbefore callingnotify_waiters()for defensive ordering.Currently
notify_waiters()fires on line 1600 before the flag is set on line 1601. This works because the tail lock is held throughout, so woken tasks block on the lock. However, the conventional pattern is state-then-notify, and reversing these two lines would make the code more robust against future refactoring that might release the lock between them.♻️ Suggested reorder
if remaining_rx == 0 { - self.shared.notify_last_rx_drop.notify_waiters(); tail.closed_for_senders = true; + self.shared.notify_last_rx_drop.notify_waiters(); }
| /// A future which completes when the number of [Receiver]s subscribed to this channel reaches | ||
| /// zero. | ||
| /// | ||
| /// # Examples | ||
| /// | ||
| /// ``` | ||
| /// use futures::FutureExt; | ||
| /// use tokio::sync::broadcast; | ||
| /// | ||
| /// # #[tokio::main(flavor = "current_thread")] | ||
| /// # async fn main() { | ||
| /// let (tx, mut rx1) = broadcast::channel::<u32>(16); | ||
| /// let mut rx2 = tx.subscribe(); | ||
| /// | ||
| /// let _ = tx.send(10); | ||
| /// let weak = tx.downgrade(); | ||
| /// | ||
| /// assert_eq!(rx1.recv().await.unwrap(), 10); | ||
| /// drop(rx1); | ||
| /// assert!(tx.closed().now_or_never().is_none()); | ||
| /// | ||
| /// assert_eq!(rx2.recv().await.unwrap(), 10); | ||
| /// drop(rx2); | ||
| /// assert!(tx.closed().now_or_never().is_some()); | ||
| /// # } | ||
| /// ``` | ||
| pub async fn closed(&self) { | ||
| self.shared.closed_for_senders().await; | ||
| } |
There was a problem hiding this comment.
Doc example never calls weak.closed() — it only exercises tx.closed().
This is the documentation for WeakSender::closed(), but the example creates weak via tx.downgrade() on line 1133 and never uses it. Lines 1137 and 1141 call tx.closed() instead. The example should demonstrate weak.closed().
📝 Suggested doc fix
/// let _ = tx.send(10);
/// let weak = tx.downgrade();
///
/// assert_eq!(rx1.recv().await.unwrap(), 10);
/// drop(rx1);
- /// assert!(tx.closed().now_or_never().is_none());
+ /// assert!(weak.closed().now_or_never().is_none());
///
/// assert_eq!(rx2.recv().await.unwrap(), 10);
/// drop(rx2);
- /// assert!(tx.closed().now_or_never().is_some());
+ /// assert!(weak.closed().now_or_never().is_some());📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| /// A future which completes when the number of [Receiver]s subscribed to this channel reaches | |
| /// zero. | |
| /// | |
| /// # Examples | |
| /// | |
| /// ``` | |
| /// use futures::FutureExt; | |
| /// use tokio::sync::broadcast; | |
| /// | |
| /// # #[tokio::main(flavor = "current_thread")] | |
| /// # async fn main() { | |
| /// let (tx, mut rx1) = broadcast::channel::<u32>(16); | |
| /// let mut rx2 = tx.subscribe(); | |
| /// | |
| /// let _ = tx.send(10); | |
| /// let weak = tx.downgrade(); | |
| /// | |
| /// assert_eq!(rx1.recv().await.unwrap(), 10); | |
| /// drop(rx1); | |
| /// assert!(tx.closed().now_or_never().is_none()); | |
| /// | |
| /// assert_eq!(rx2.recv().await.unwrap(), 10); | |
| /// drop(rx2); | |
| /// assert!(tx.closed().now_or_never().is_some()); | |
| /// # } | |
| /// ``` | |
| pub async fn closed(&self) { | |
| self.shared.closed_for_senders().await; | |
| } | |
| /// A future which completes when the number of [Receiver]s subscribed to this channel reaches | |
| /// zero. | |
| /// | |
| /// # Examples | |
| /// | |
| /// |
🤖 Prompt for AI Agents
In `@tokio/src/sync/broadcast.rs` around lines 1118 - 1146, The doc example for
WeakSender::closed() mistakenly calls tx.closed() instead of demonstrating the
weak handle; update the example to call weak.closed() (the WeakSender returned
by tx.downgrade()) in the two assertions that currently use tx.closed() so the
snippet exercises weak.closed() as intended; keep the rest of the example
(creating tx via broadcast::channel, tx.downgrade(), recv and drops) unchanged.
There was a problem hiding this comment.
value:useful; category:documentation; feedback: The CodeRabbit AI reviewer is correct! The doc test is supposed to show that the WeakSender is not closed until the last channel receiver is dropped but instead it uses the strong Sender. Prevents releasing wrong documentation.
value:useful; category:documentation; feedback: The Claude AI reviewer is correct! The doc test is supposed to show that the WeakSender is not closed until the last channel receiver is dropped but instead it uses the strong Sender. Prevents releasing wrong documentation. |
value:good-to-have; category:bug; feedback: The CodeRabbit AI reviewer is correct! The pattern is wrongly applied - first notify the subscribers and then change the state. It should be the opposite - first update the state and then notify. The acquired lock is still the same, so there is no behavior change. It will be semantically better. |
value:useful; category:bug; feedback: The CodeRabbit AI reviewer is correct! The call to async_trace_leaf() should be inside the cooperative() call even if it does not need the cooperative context. It may need to use it one day. Also it is always good to keep the code consistent. |
2 participants
7900: To review by AI