WIP: Add resubscribe functionality to broadcast::Receiver

[estk]

Add subscribe method to broadcast::Receiver with the same functionality as the clone impl removed in #3020.

[Darksonn]

Why not just implement Clone in a way that isn't broken?

[uklotzde]

Though some kind of resubscribe method that produces an independent, fresh receiver as if we have subscribed at this moment at the sender is still a valid use case.

Not sure how to name such a method appropriately? resubscribe()?

[Darksonn]

I would consider a Clone impl more important than such a method. If you want to discuss that further, you should probably open an issue to discuss, rather than discussing on this PR.

[estk]

impl Clone it is

[estk]

there go

[Darksonn]

Can you add some tests for this?

[estk]

@Darksonn wdyt?

[estk]

@Darksonn in case you missed the last mention

[Darksonn]

You are not currently updating this counter:

tokio/tokio/src/sync/broadcast.rs

Lines 326 to 334 in 06e413c

	/// Slot in the buffer.
	struct Slot<T> {
	/// Remaining number of receivers that are expected to see this value.
	///
	/// When this goes to zero, the value is released.
	///
	/// An atomic is used as it is mutated concurrently with the slot read lock
	/// acquired.
	rem: AtomicUsize,

Currently that counter can never increase. It's unclear to me whether simply adding to it can result in some kind of race where the value is lost due to a concurrent call to recv. Additionally, if you manage to find a way to fix this, I would like to see a test that fails with the current code but doesn't fail with the fix.

[estk]

@Darksonn thanks for the review, I guess I didn't realize what I was getting myself into 🤷‍♂️. I agree with @uklotzde that we should probably also have a Receiver::resubscribe() method, in practice it's not a problem since Sender: Clone.

I've added some tests and below I've laid out my reasoning why incrementing Slot::rem is safe. Lock-free algorithms is not something I have much experience with so I would greatly appreciate a careful review.

At first I came to the same conclusion as you, that incrementing slot.rem is not safe on account of the fact that a concurrent call to recv may drop the slot value since the fetch_add from impl Clone for Receiver may happen between fetch_sub and "drop value" in impl Drop for RecvGuard. However I now think the compiler prevents this.

We are concerned with a data-race on Slot::rem. The methods that manipulate it are:

1. Sender::send2
2. Receiver::clone
3. RecvGuard::drop

Consider their interactions:

1 and 3 are pre-existing so I am going to ignore how they may affect each other.
1 and 2 are unable to manipulate the tail slot at the same time due to the tail lock. Sends which happen before or after result in predictable behavior.
2 and 3 are not protected by a mutex, we consider the interactions below.

We obtain a RecvGuard, drop then Receiver::clone: we don't care about the slot valued dropped.
We obtain a RecvGuard Receiver::clone then drop(guard): the rem has already been incremented, no drop.

Now, can RecvGuard::drop and Receiver::clone execute concurrently. No, consider the following: Receiver::recv_ref is the only method which produces a RecvGuard. Receiver::try_recv, Recv::poll, and Receiver::drop are the only methods which call recv_ref. These methods immediately drop or clone then drop the RecvGuard. All of these methods obtain a &mut Receiver before calling recv_ref and critically &mut 'receiver: 'recv_guard. Therefore Receiver::clone is prevented by rustc from being called concurrently with RecvGuard::drop.

[estk]

working on fixing the loom test

[estk]

@Darksonn Okay so I think I've fixed my loom test, glad I wrote that one haha. I was wondering, do you think I should write a compile_fail test that verifies my assertion that RecvGuard::drop and Receiver::clone may not be called concurrently? Would that address your concern around "some kind of race where the value is lost due to a concurrent call to recv"?

[Darksonn]

That RecvGuard::drop can only be reached through a mutable references is enough. You don't have to write compile-fail tests for that. The compiler will definitely guarantee that mutable references are exclusive.

Unfortunately, I probably wont have time to review this in detail until next week, but I think I have been convinced that elements that the new receiver needs wont be dropped since the existing receiver also needs to see them, and can't consume them while the new receiver is being created.

[estk]

@Darksonn bump

[estk]

@Darksonn fixed that

[carllerche]

My hesitation w/ this PR and the reason I did not add Receiver::clone() initially is that clone() here is O(N) which is not the case for any other channel handle. This would be an unexpected behavior.

Because of this, if the clone behavior is required, I suggest providing it as an explicit method that can be documented as having the O(N) caveat.

[Darksonn]

I see. I didn't consider that. Going back to a resubscribe that doesn't keep the position might be a better option.

[estk]

@carllerche @Darksonn It seems like both functionalities would be useful. But I also did not consider that the O(n)ness would be an issue here. Just let me know what makes sense and I'll be happy to adjust accordingly.

[estk]

Alright, I've just added both methods here for discussion. When you all decide if you'd like one or both or in separate PRs just let me know and I'll be more than happy to update the PR. Additionally please feel free to just copy what I have and adjust as necessary.

[Darksonn]

My current thought is to just provide a resubscribe method. We may be able to fix it so we can have a cheap Clone in the future, in which case we would want a Clone impl instead.

[Darksonn]

The documentation here would need to be improved.

[Darksonn]

Isn't this clearer?

Suggested change

	/// assert_neq!(rx.recv().await.unwrap(), rx2.recv().await.unwrap()); // 1 != 2
	/// assert_eq!(rx2.recv().await.unwrap(), 2);
	/// assert_eq!(rx.recv().await.unwrap(), 1);

[estk]

ya agreed

[estk]

@Darksonn those bsd failures seem unrelated, any idea what's going on there?

[Darksonn]

You need to merge master to get the fixes. It's because of the 1.61 release of Rust.

[estk]

@Darksonn

[Darksonn]

Thanks.

[jeremija]

My hesitation w/ this PR and the reason I did not add Receiver::clone() initially is that clone() here is O(N) which is not the case for any other channel handle. This would be an unexpected behavior.

Because of this, if the clone behavior is required, I suggest providing it as an explicit method that can be documented as having the O(N) caveat.

@carllerche sorry for commenting on an old thread, but would you be able to explain how would Clone be O(N) since the current implementation of resubscribe() seems to be O(1) because shared is behind an Arc:

    pub fn resubscribe(&self) -> Self {
        let shared = self.shared.clone();
        new_receiver(shared)
    }

/// Create a new `Receiver` which reads starting from the tail.
fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
    let mut tail = shared.tail.lock();

    if tail.rx_cnt == MAX_RECEIVERS {
        panic!("max receivers");
    }

    tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");

    let next = tail.pos;

    drop(tail);

    Receiver { shared, next }
}

and I think it would be just a matter of setting next to the beginning if Clone were to be implemented. Or am I missing something here? TIA

[Darksonn]

Each item contains a counter for the number of receivers that need to see it. You would have to increment all of those.

[jeremija]

Ah I see, thanks for explaining!