Resubscribing to a closed broadcast receiver will hang on a call to `recv` #4814

nathaniel-daniel · 2022-07-07T19:32:25Z

Version
tokio v1.19.2, tokio master (4daeea8)

Platform
Windows 10 64 bit

Description
Attempting to resubscribe to a closed broadcast receiver will hang on calls to recv.

I tried this code:
Cargo.toml:

[package]
name = "tokio-broadcast-bug"
version = "0.0.0"
edition = "2021"

[dependencies]
tokio = { version = "1.19.2", features = ["full"] }

main.rs:

#[tokio::main]
async fn main() {
    let (tx, rx) = tokio::sync::broadcast::channel::<u32>(4);
    drop(tx);

    let mut rx_clone = rx.resubscribe();
    drop(rx);

    loop {
        match rx_clone.recv().await {
            Ok(msg) => {
                println!("{}", msg);
            }
            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                println!("Closed");
                break;
            }
            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                println!("Lagged by {n} messages");
            }
        }
    }

    println!("Done");
}

I expected to see this happen:
The loop should exit.

Instead, this happened:
The program hangs indefinitely.

Furthermore, replacing the loop with a call to try_recv yields an Empty error instead of a Closed error.

The text was updated successfully, but these errors were encountered:

Darksonn · 2022-07-09T15:41:24Z

That certainly does sound like a bug.

nathaniel-daniel · 2022-07-12T04:21:02Z

I don't really understand everything about this channel, but shouldn't new_receiver update the next field similarly to how recv_ref does so here, by including an adjust value to account for the close message? Making this change seems to fix my simple example, but I don't know if it works in a larger project or if it introduces any other bugs; I was hoping someone more knowledgeable could take a look.

Darksonn · 2022-07-12T08:12:22Z

The index not being adjusted for the close message does indeed appear to be the correct answer.

The broadcast channel allows multiple senders to send messages to multiple receivers, where each receiver receives messages starting from when it subscribes. After all senders are dropped, the receivers will continue to receive all waiting messages in the buffer and then receive a `Closed` error. To mark that a channel has closed, it stores two closed flags, one on the channel level and another in the buffer slot *after* the last used slot (this may also be the earliest entry being kept for lagged receivers, see tokio-rs#2425). However, we don't need both closed flags, keeping the channel level closed flag is sufficient. Without the slot level closed flag, each receiver receives each message until it is up to date and for that receiver the channel is empty. Then, the actual return message is chosen depending on the channel level closed flag; if the channel is NOT closed, then `Empty` is returned, if the channel is closed then `Closed` is returned instead. With the modified logic, there is no longer a need to append a closed token to the internal buffer (by setting the slot level closed flag on the next slot). This fixes the off by one error described in tokio-rs#4814, which caused a receiver which was created after the channel was already closed to get `Empty` from `try_recv` (or hang forever when calling `recv`) instead of receiving `Closed`. As a bonus, we save a single bool on each buffer slot.

The broadcast channel allows multiple senders to send messages to multiple receivers, where each receiver receives messages starting from when it subscribes. After all senders are dropped, the receivers will continue to receive all waiting messages in the buffer and then receive a `Closed` error. To mark that a channel has closed, it stores two closed flags, one on the channel level and another in the buffer slot *after* the last used slot (this may also be the earliest entry being kept for lagged receivers, see tokio-rs#2425). However, we don't need both closed flags, keeping the channel level closed flag is sufficient. Without the slot level closed flag, each receiver receives each message until it is up to date and for that receiver the channel is empty. Then, the actual return message is chosen depending on the channel level closed flag; if the channel is NOT closed, then `Empty` is returned, if the channel is closed then `Closed` is returned instead. With the modified logic, there is no longer a need to append a closed token to the internal buffer (by setting the slot level closed flag on the next slot). This fixes the off by one error described in tokio-rs#4814, which caused a receiver which was created after the channel was already closed to get `Empty` from `try_recv` (or hang forever when calling `recv`) instead of receiving `Closed`. As a bonus, we save a single bool on each buffer slot. Refs: tokio-rs#4814

The broadcast channel allows multiple senders to send messages to multiple receivers, where each receiver receives messages starting from when it subscribes. After all senders are dropped, the receivers will continue to receive all waiting messages in the buffer and then receive a `Closed` error. To mark that a channel has closed, it stores two closed flags, one on the channel level and another in the buffer slot *after* the last used slot (this may also be the earliest entry being kept for lagged receivers, see tokio-rs#2425). However, we don't need both closed flags, keeping the channel level closed flag is sufficient. Without the slot level closed flag, each receiver receives each message until it is up to date and for that receiver the channel is empty. Then, the actual return message is chosen depending on the channel level closed flag; if the channel is NOT closed, then `Empty` is returned, if the channel is closed then `Closed` is returned instead. With the modified logic, there is no longer a need to append a closed token to the internal buffer (by setting the slot level closed flag on the next slot). This fixes the off by one error described in tokio-rs#4814, which caused a receiver which was created after the channel was already closed to get `Empty` from `try_recv` (or hang forever when calling `recv`) instead of receiving `Closed`. As a bonus, we save a single `bool` on each buffer slot. Refs: tokio-rs#4814

hds · 2022-07-26T16:17:56Z

I came up with a solution for this issue which is a bit more drastic that I originally anticipated (I'm removing closed flag in the slot and just relying on the channel level closed flag).

@nathaniel-daniel Since it was your observation that the next flag on the resubscribed receiver that was incorrect, perhaps you'd be interested in looking at the solution? I probably wouldn't have arrived at it without your comments!

The broadcast channel allows multiple senders to send messages to multiple receivers, where each receiver receives messages starting from when it subscribes. After all senders are dropped, the receivers will continue to receive all waiting messages in the buffer and then receive a `Closed` error. To mark that a channel has closed, it stores two closed flags, one on the channel level and another in the buffer slot *after* the last used slot (this may also be the earliest entry being kept for lagged receivers, see #2425). However, we don't need both closed flags, keeping the channel level closed flag is sufficient. Without the slot level closed flag, each receiver receives each message until it is up to date and for that receiver the channel is empty. Then, the actual return message is chosen depending on the channel level closed flag; if the channel is NOT closed, then `Empty` is returned, if the channel is closed then `Closed` is returned instead. With the modified logic, there is no longer a need to append a closed token to the internal buffer (by setting the slot level closed flag on the next slot). This fixes the off by one error described in #4814, which caused a receiver which was created after the channel was already closed to get `Empty` from `try_recv` (or hang forever when calling `recv`) instead of receiving `Closed`. As a bonus, we save a single `bool` on each buffer slot. Refs: #4814

nathaniel-daniel · 2022-08-11T03:37:04Z

Sorry for the rate reply, I've been busy. Thanks for the fix @hds! I just tested and it seems to work great on a simple example and a small sized project I'm working on. I was writing up a PR to do essentially what you did, but it stalled as I tried to figure out the ramifications of removing the double closed flags, performance or otherwise; I couldn't figure out why the two closed flags were even introduced in the first place. Thanks for the PR!

nathaniel-daniel added A-tokio Area: The main tokio crate C-bug Category: This is a bug. labels Jul 7, 2022

Darksonn added the M-sync Module: tokio/sync label Jul 7, 2022

hds mentioned this issue Jul 26, 2022

sync: remove broadcast channel slot level closed flag #4867

Merged

Darksonn closed this as completed in #4867 Aug 10, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Resubscribing to a closed broadcast receiver will hang on a call to `recv` #4814

Resubscribing to a closed broadcast receiver will hang on a call to `recv` #4814

nathaniel-daniel commented Jul 7, 2022 •

edited

Loading

Darksonn commented Jul 9, 2022

nathaniel-daniel commented Jul 12, 2022

Darksonn commented Jul 12, 2022

hds commented Jul 26, 2022

nathaniel-daniel commented Aug 11, 2022

Resubscribing to a closed broadcast receiver will hang on a call to recv #4814

Resubscribing to a closed broadcast receiver will hang on a call to recv #4814

Comments

nathaniel-daniel commented Jul 7, 2022 • edited Loading

Darksonn commented Jul 9, 2022

nathaniel-daniel commented Jul 12, 2022

Darksonn commented Jul 12, 2022

hds commented Jul 26, 2022

nathaniel-daniel commented Aug 11, 2022

Resubscribing to a closed broadcast receiver will hang on a call to `recv` #4814

Resubscribing to a closed broadcast receiver will hang on a call to `recv` #4814

nathaniel-daniel commented Jul 7, 2022 •

edited

Loading