tokio-stream: add wrapper for broadcast and watch #3384

liufuyang · 2021-01-06T16:07:20Z

Signed-off-by: Fuyang Liu liufuyang@users.noreply.github.com

Motivation

Closes #3382

Solution

The basic sketch is going to be:

Make the constructor takes a broadcast receiver
Then it uses async-stream to turn that into an impl Stream
Still inside the constructor, call Box::pin on that stream, the put the pinned box into the struct
Then your impl Stream on Wrapper just forwards calls to poll_next to the pinned box

Darksonn · 2021-01-21T18:25:08Z

The Unpin bound can be removed once tokio-rs/async-stream#52 is released.

As for the Send/Sync bounds, I believe it would be sound to take them away with the right unsafe, but the question is if it's a good idea.

liufuyang · 2021-01-23T12:27:22Z

That sounds like some difficult question for me to answer :)

tokio-stream/src/wrappers/broadcast.rs

Darksonn · 2021-01-25T11:13:16Z

tokio-stream/src/wrappers/watch.rs

+pub struct WatchStream<T> {
+    inner: Pin<Box<dyn Stream<Item = ()>>>,
+    _marker: std::marker::PhantomData<T>,
+}


It seems like a problem to have the Stream return (). The user of the Stream would want the actual item, though I think we might have to clone it to do that?

🤔 I thought users would mostly use this to see whether something got arrived? (Maybe I am wrong...) Or maybe it is nice to have it return the last arrived value as well.

Or can I justyield rx.borrow() ? Instead of yield a ()? I am trying this and now having some difficulty to deal with the life-times.

Signed-off-by: Fuyang Liu <liufuyang@users.noreply.github.com>

Co-authored-by: Alice Ryhl <alice@ryhl.io>

* add watch * Fix up

tokio-stream/Cargo.toml

Darksonn · 2021-01-31T15:34:48Z

tokio-stream/src/wrappers/broadcast.rs

+    Ok((item, rx))
+}
+
+impl<T: Clone + Unpin + 'static + Send + Sync> BroadcastStream<T> {


With the new change, some of these bounds can be removed.

Darksonn · 2021-01-31T15:36:31Z

tokio-stream/src/wrappers/broadcast.rs

+/// [`tokio::sync::broadcast::Receiver`]: struct@tokio::sync::broadcast::Receiver
+/// [`Stream`]: trait@crate::Stream
+pub struct BroadcastStream<T> {
+    inner: ReusableBoxFuture<Result<(T, Receiver<T>), RecvError>>,


We need to return the receiver even if it emits an error, as the stream would otherwise not be usable again after returning an error.

Co-authored-by: Alice Ryhl <alice@ryhl.io>

Darksonn · 2021-01-31T16:27:46Z

tokio-stream/src/wrappers/broadcast.rs

+enum WrappedRecvError<T> {
+    Lagged(u64, Receiver<T>),
+    Closed,
+}
+
+async fn make_future<T: Clone>(
+    mut rx: Receiver<T>,
+) -> Result<(T, Receiver<T>), WrappedRecvError<T>> {
+    match rx.recv().await {
+        Ok(item) => Ok((item, rx)),
+        Err(RecvError::Lagged(n)) => Err(WrappedRecvError::Lagged(n, rx)),
+        Err(RecvError::Closed) => Err(WrappedRecvError::Closed),
+    }
+}


We should be able to simplify to the following.

Suggested change

enum WrappedRecvError<T> {

Lagged(u64, Receiver<T>),

Closed,

}

async fn make_future<T: Clone>(

mut rx: Receiver<T>,

) -> Result<(T, Receiver<T>), WrappedRecvError<T>> {

match rx.recv().await {

Ok(item) => Ok((item, rx)),

Err(RecvError::Lagged(n)) => Err(WrappedRecvError::Lagged(n, rx)),

Err(RecvError::Closed) => Err(WrappedRecvError::Closed),

}

}

async fn make_future<T: Clone>(

mut rx: Receiver<T>,

) -> (Result<T, RecvError<T>>, Receiver<T>) {

let result = rx.recv().await;

(result, rx)

}

That looks much better 😄

Darksonn · 2021-01-31T16:29:51Z

tokio-stream/src/wrappers/broadcast.rs

+        match ready!(self.inner.poll(cx)) {
+            Ok((item, rx)) => {
+                self.inner.set(make_future(rx));
+                Poll::Ready(Some(Ok(item)))
+            }
+            Err(err) => match err {
+                WrappedRecvError::Closed => Poll::Ready(None),
+                WrappedRecvError::Lagged(n, rx) => {
+                    self.inner.set(make_future(rx));
+                    Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(n))))
+                }
+            },
+        }


I want to call self.inner.set(make_future(rx)) in all three cases because then, if the user calls poll_next after WrappedRecvError::Closed, then we will just return Poll::Ready(None) again.

With this implementation, it will panic if polled again after the first WrappedRecvError::Closed.

You should be able to avoid duplicating the call three times by doing it before the match (but after the self.inner.poll call).

That is a good catch. Thank you 👍

Darksonn · 2021-01-31T17:56:37Z

tokio-stream/src/wrappers/broadcast.rs

+        match result {
+            Ok(item) => Poll::Ready(Some(Ok(item))),
+            Err(err) => match err {
+                RecvError::Closed => Poll::Ready(None),
+                RecvError::Lagged(n) => Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(n)))),
+            },
+        }


This could be simplified.

Suggested change

match result {

Ok(item) => Poll::Ready(Some(Ok(item))),

Err(err) => match err {

RecvError::Closed => Poll::Ready(None),

RecvError::Lagged(n) => Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(n)))),

},

}

match result {

Ok(item) => Poll::Ready(Some(Ok(item))),

Err(RecvError::Closed) => Poll::Ready(None),

Err(RecvError::Lagged(n)) => Poll::Ready(Some(Err(BroadcastStreamRecvError::Lagged(n)))),

}

Darksonn · 2021-01-31T17:57:25Z

tokio-stream/src/wrappers/broadcast.rs

+    }
+}
+
+impl<T: Clone> fmt::Debug for BroadcastStream<T> {


Suggested change

impl<T: Clone> fmt::Debug for BroadcastStream<T> {

impl<T> fmt::Debug for BroadcastStream<T> {

Darksonn · 2021-01-31T17:59:17Z

tokio-stream/src/wrappers/watch.rs

+async fn make_future<T: Clone + Send + Sync>(
+    mut rx: Receiver<T>,
+) -> Result<((), Receiver<T>), RecvError> {
+    let signal = rx.changed().await?;
+    Ok((signal, rx))
+}


Same change here.

Aha, sorry I forgot this one.

Darksonn

This seems good to me now!

## Issue Addressed resolves #2129 resolves #2099 addresses some of #1712 unblocks #2076 unblocks #2153 ## Proposed Changes - Updates all the dependencies mentioned in #2129, except for web3. They haven't merged their tokio 1.0 update because they are waiting on some dependencies of their own. Since we only use web3 in tests, I think updating it in a separate issue is fine. If they are able to merge soon though, I can update in this PR. - Updates `tokio_util` to 0.6.2 and `bytes` to 1.0.1. - We haven't made a discv5 release since merging tokio 1.0 updates so I'm using a commit rather than release atm. **Edit:** I think we should merge an update of `tokio_util` to 0.6.2 into discv5 before this release because it has panic fixes in `DelayQueue` --> PR in discv5: sigp/discv5#58 ## Additional Info tokio 1.0 changes that required some changes in lighthouse: - `interval.next().await.is_some()` -> `interval.tick().await` - `sleep` future is now `!Unpin` -> tokio-rs/tokio#3028 - `try_recv` has been temporarily removed from `mpsc` -> tokio-rs/tokio#3350 - stream features have moved to `tokio-stream` and `broadcast::Receiver::into_stream()` has been temporarily removed -> `tokio-rs/tokio#2870 - I've copied over the `BroadcastStream` wrapper from this PR, but can update to use `tokio-stream` once it's merged tokio-rs/tokio#3384 Co-authored-by: realbigsean <seananderson33@gmail.com>

taiki-e added A-tokio-stream Area: The tokio-stream crate M-stream Module: tokio/stream labels Jan 10, 2021

Darksonn reviewed Jan 23, 2021

View reviewed changes

liufuyang force-pushed the tokio-stream-3382 branch 2 times, most recently from 354649a to f1c0143 Compare January 24, 2021 10:08

Darksonn reviewed Jan 24, 2021

View reviewed changes

tokio-stream/src/wrappers/broadcast.rs Show resolved Hide resolved

tokio-stream/src/wrappers/broadcast.rs Outdated Show resolved Hide resolved

liufuyang changed the title ~~tokio-stream: add wrapper for broadcast~~ tokio-stream: add wrapper for broadcast and watch Jan 25, 2021

Darksonn reviewed Jan 25, 2021

View reviewed changes

realbigsean mentioned this pull request Jan 25, 2021

[Merged by Bors] - Update to tokio 1.1 sigp/lighthouse#2172

Closed

liufuyang force-pushed the tokio-stream-3382 branch from c6bc555 to 3957c05 Compare January 26, 2021 12:41

liufuyang and others added 11 commits January 31, 2021 14:39

tokio-stream: add wrapper for broadcast

eddbdaa

Signed-off-by: Fuyang Liu <liufuyang@users.noreply.github.com>

Fix up

f1094aa

Signed-off-by: Fuyang Liu <liufuyang@users.noreply.github.com>

Minor fixup - remove cfg_attr

6995914

Add #[derive(Debug)]

5229cad

Update tokio-stream/src/wrappers/broadcast.rs

c611d5c

Co-authored-by: Alice Ryhl <alice@ryhl.io>

return Lagged error

1977be3

expose BroadcastStreamRecvError

d9fd015

reformat

9513b71

add watch (#1)

ed00803

* add watch * Fix up

return cloned value when warpped recv changed

ef12286

remove dependency of async_stream

34eab73

liufuyang force-pushed the tokio-stream-3382 branch from 3957c05 to 34eab73 Compare January 31, 2021 15:28

Darksonn reviewed Jan 31, 2021

View reviewed changes

liufuyang and others added 3 commits January 31, 2021 16:51

Update tokio-stream/Cargo.toml

c6f9ee9

Co-authored-by: Alice Ryhl <alice@ryhl.io>

wrap the receiver in an inner error

35a8502

fix up

2cbf560

Darksonn reviewed Jan 31, 2021

View reviewed changes

simplify code

e2d8214

liufuyang added 2 commits January 31, 2021 18:36

alwasy refresh inner future

f7f2f00

add async-stream back for dev dep

fb244b1

Darksonn reviewed Jan 31, 2021

View reviewed changes

liufuyang added 2 commits January 31, 2021 19:09

simplify code

3b045dc

fix up

d44d54a

Darksonn approved these changes Jan 31, 2021

View reviewed changes

Darksonn merged commit 1c1e0e3 into tokio-rs:master Feb 5, 2021

Darksonn mentioned this pull request Feb 5, 2021

chore: prepare tokio-stream 0.1.3 #3507

Merged

liufuyang deleted the tokio-stream-3382 branch February 6, 2021 09:59

paulhauner mentioned this pull request Feb 10, 2021

Use tokio-stream BroadcastStream sigp/lighthouse#2189

Closed

dependabot bot mentioned this pull request Mar 8, 2021

Bump tokio from 0.2.25 to 1.2.0 technic/epg-server#135

Closed

dependabot bot mentioned this pull request Mar 17, 2021

Bump tokio from 0.2.23 to 1.2.0 predominant/habitat#3

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

tokio-stream: add wrapper for broadcast and watch #3384

tokio-stream: add wrapper for broadcast and watch #3384

liufuyang commented Jan 6, 2021

Darksonn commented Jan 21, 2021 •

edited

liufuyang commented Jan 23, 2021

Darksonn Jan 25, 2021

liufuyang Jan 25, 2021 •

edited

liufuyang Jan 26, 2021

Darksonn Jan 31, 2021

Darksonn Jan 31, 2021

Darksonn Jan 31, 2021

liufuyang Jan 31, 2021

liufuyang Jan 31, 2021

Darksonn Jan 31, 2021

liufuyang Jan 31, 2021

Darksonn Jan 31, 2021

Darksonn Jan 31, 2021

Darksonn Jan 31, 2021

liufuyang Jan 31, 2021

Darksonn left a comment

	impl<T: Clone> fmt::Debug for BroadcastStream<T> {
	impl<T> fmt::Debug for BroadcastStream<T> {

tokio-stream: add wrapper for broadcast and watch #3384

tokio-stream: add wrapper for broadcast and watch #3384

Conversation

liufuyang commented Jan 6, 2021

Motivation

Solution

Darksonn commented Jan 21, 2021 • edited

liufuyang commented Jan 23, 2021

Choose a reason for hiding this comment

liufuyang Jan 25, 2021 • edited

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

Choose a reason for hiding this comment

Darksonn left a comment

Choose a reason for hiding this comment

Darksonn commented Jan 21, 2021 •

edited

liufuyang Jan 25, 2021 •

edited