sync: add broadcast channel #1943

carllerche · 2019-12-11T06:30:09Z

Adds a broadcast channel implementation. A broadcast channel is a
multi-producer, multi-consumer channel where each consumer receives a
clone of every value sent. This is useful for implementing pub / sub
style patterns.

Implemented as a ring buffer, a Vec of the specified capacity is
allocated on initialization of the channel. Values are pushed into
slots.

When the channel is full, a send overwrites the oldest value. Receivers
detect this and return an error on the next call to receive. This
prevents unbounded buffering and does not make the channel vulnerable to
the slowest consumer.

This is ready for review, but should not be merged until it has some
application validation.

Closes: #1585

Adds a broadcast channel implementation. A broadcast channel is a multi-producer, multi-consumer channel where each consumer receives a clone of every value sent. This is useful for implementing pub / sub style patterns. Implemented as a ring buffer, a Vec of the specified capacity is allocated on initialization of the channel. Values are pushed into slots. When the channel is full, a send overwrites the oldest value. Receivers detect this and return an error on the next call to receive. This prevents unbounded buffering and does not make the channel vulnerable to the slowest consumer. Closes: #1585

carllerche · 2019-12-11T06:39:30Z

One question is if Sender::send should return the number of active receivers at the time of send.

Matthias247 · 2019-12-11T08:10:48Z

+1 on the bounded implementation! That part it the hardest for any "broadcast" like implementation.

API-wise you could cross-check against futures-intrusive::channel::StateBroadcastChannel, which is pretty much the same thing - but with a channel size preconfigured to 1, so it will only retain the latest element. This is the "latest state", which consumers are typically interested in. Actually it could be made configurable there too.

Anyway, what future-intrusive does is having the items IDs inside the API. Every successful receive gives you the ID of the next item to receive. Here it seems like you capture it in the Stream. Which obviously has the benefit of staying more true to the Stream API. With explicit IDs

users could see how many updates they missed. But you probably could (or should!) provide this error in the Err variant of receive.
can stop and resume reads from the channel without having to persist a Future or the channel directly. Just persisting the index is good enough. But to be fair I didn't really had a use-case for this yet.

jonhoo · 2019-12-11T16:54:13Z

cc @habnabit

tokio/src/sync/broadcast.rs

`send` now returns the number of receivers subscribed at the time `send` is called. The `Lagged` error now includes the number of skipped messages.

carllerche · 2019-12-11T21:52:15Z

@Matthias247

I added the number of skipped messages to the Lagged error.

I also return from send the number of receivers subscribed at that moment.

Co-Authored-By: Eliza Weisman <[email protected]>

hawkw · 2019-12-11T22:54:48Z

tokio/src/sync/broadcast.rs

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


nit/tioli: could be

assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");

jebrosen · 2019-12-17T17:56:28Z

I've been trying this out in a demo application of mine (commit at https://git.jebrosen.com/jeb/rocket-rooms/commit/e3fa0b106566a55a9bbcb82854b6be71cd254fff) and I really like it so far. It's as good or better than my naive wrapper over Watch was in terms of throughput - not sure exactly how much, because my test harness is a bit disappointing right now.

carllerche · 2019-12-17T19:21:27Z

@jebrosen thanks for giving it a run, I tried it a bit locally and also works for me. I'm happy merging this if others are 👍

carllerche · 2019-12-18T06:31:22Z

@LucioFranco @jonhoo @hawkw i need a +1 on this to merge :)

LucioFranco · 2019-12-18T16:01:31Z

tokio/src/sync/broadcast.rs

+
+/// Receiving-half of the [`broadcast`] channel.
+///
+/// May not be used concurrently. Messages may be retrieved using


Suggested change

/// May not be used concurrently. Messages may be retrieved using

/// Should not be used concurrently. Messages may be retrieved using

I think this sounds a bit better.

"should not" seems like a suggestion and "may not" is prohibited?

Ah then you want Must not

Don't ask my why I know this is here but https://www.faa.gov/air_traffic/publications/atpubs/atc_html/chap1_section_2.html is a good reference :)

I personally find all of these wordings confusing. "May not / must not / should not" reads like something I have to keep in mind when I use the API otherwise some bad thing will happen, but the only consequence I can see of using this concurrently is that rustc will complain because all of the (public) methods take &mut self. std::sync::mpsc::Receiver uses the wording "This half can only be owned by one thread" which I'm also not quite happy with but IMO is less open to confusion.

LucioFranco · 2019-12-18T16:10:00Z

tokio/src/sync/broadcast.rs

+    ///     assert_eq!(20, value);
+    /// }
+    /// ```
+    pub fn subscribe(&self) -> Receiver<T> {


is there value in having this versus the way we normally do it with a mpsc? May make sense to align the apis

Subscribing starts at the head of the list. A receiver may currently be pointing to any index. A clone() would then not return an "identical" item. I thought it didn't match.

Ah I see, Im not gonna reread the docs but can we make sure this is documented somewhere. This point is totally fair.

Could we maybe add a method to Reciever that can get you one that starts at the head?

I would opt to punt on that until there is a clear case where that is helpful. I'm not sure what it would be named or if it makes sense 🤷

You know me I like to punt 👍

LucioFranco · 2019-12-18T16:12:58Z

tokio/src/sync/broadcast.rs

+    /// internal cursor is updated to point to the oldest value still held by
+    /// the channel. A subsequent call to [`try_recv`] will return this value
+    /// **unless** it has been since overwritten.
+    ///


missing doc on TryRecvError::Empty

LucioFranco

At a high level this LGTM, hopping loom covers most of the cases 😄

carllerche requested review from hawkw, LucioFranco and seanmonstar December 11, 2019 06:30

hawkw reviewed Dec 11, 2019

View reviewed changes

tokio/src/sync/broadcast.rs Outdated Show resolved Hide resolved

tokio/src/sync/broadcast.rs Outdated Show resolved Hide resolved

jebrosen reviewed Dec 11, 2019

View reviewed changes

tokio/src/sync/broadcast.rs Outdated Show resolved Hide resolved

tweak send and lagged API

d7b090b

`send` now returns the number of receivers subscribed at the time `send` is called. The `Lagged` error now includes the number of skipped messages.

carllerche and others added 6 commits December 11, 2019 13:52

Update tokio/src/sync/broadcast.rs

fbd4b5d

Co-Authored-By: Eliza Weisman <[email protected]>

Update tokio/src/sync/broadcast.rs

a747465

Co-Authored-By: Eliza Weisman <[email protected]>

tweaks

5726623

more tweaks

b58035c

tweak

7d68ab5

fmt

4664466

hawkw reviewed Dec 11, 2019

View reviewed changes

LucioFranco reviewed Dec 18, 2019

View reviewed changes

LucioFranco approved these changes Dec 18, 2019

View reviewed changes

carllerche added 2 commits December 18, 2019 08:43

doc tweak

2b60470

Merge remote-tracking branch 'origin/master' into broadcast

9c056c7

carllerche merged commit 7c010ed into master Dec 18, 2019

carllerche deleted the broadcast branch January 10, 2020 19:30

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

sync: add broadcast channel #1943

sync: add broadcast channel #1943

carllerche commented Dec 11, 2019 •

edited

Loading

carllerche commented Dec 11, 2019

Matthias247 commented Dec 11, 2019

jonhoo commented Dec 11, 2019

carllerche commented Dec 11, 2019

hawkw Dec 11, 2019

jebrosen commented Dec 17, 2019

carllerche commented Dec 17, 2019

carllerche commented Dec 18, 2019

LucioFranco Dec 18, 2019

carllerche Dec 18, 2019

LucioFranco Dec 18, 2019

LucioFranco Dec 18, 2019

jebrosen Dec 18, 2019

LucioFranco Dec 18, 2019

carllerche Dec 18, 2019

LucioFranco Dec 18, 2019

LucioFranco Dec 18, 2019

carllerche Dec 18, 2019

LucioFranco Dec 18, 2019

LucioFranco Dec 18, 2019

LucioFranco left a comment

	/// May not be used concurrently. Messages may be retrieved using
	/// Should not be used concurrently. Messages may be retrieved using

sync: add broadcast channel #1943

sync: add broadcast channel #1943

Conversation

carllerche commented Dec 11, 2019 • edited Loading

carllerche commented Dec 11, 2019

Matthias247 commented Dec 11, 2019

jonhoo commented Dec 11, 2019

carllerche commented Dec 11, 2019

Choose a reason for hiding this comment

jebrosen commented Dec 17, 2019

carllerche commented Dec 17, 2019

carllerche commented Dec 18, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LucioFranco left a comment

Choose a reason for hiding this comment

carllerche commented Dec 11, 2019 •

edited

Loading