Randomize Stream::merge to improve the throughput. #503
Conversation
|
Hi, so I made 3 commits. The first one was a bit of a fail. Tests passed on my laoptop because this was an unstable function. The second commit is what I think it should look like: one function to not duplicate code, but I was getting an error saying that Duplicating the code in the two branches of the Do you know why the second commit might be failing? |
|
Hi, I fixed the deduplication of the code, so I think this is ready for merging. |
src/stream/stream/merge.rs
Outdated
| S: Stream<Item = I>, | ||
| { | ||
| match first.poll_next(cx) { | ||
| Poll::Ready(Some(item)) => Poll::Ready(Some(item)), |
There was a problem hiding this comment.
| Poll::Ready(Some(item)) => Poll::Ready(Some(item)), | |
| Poll::Ready(item) => Poll::Ready(item), |
I think either is fine, but I thought it was a little redundant.
However, the above changes will not match None, so the order must be changed.
Which do you think is better?
Poll::Ready(None) => second.poll_next(cx),
Poll::Ready(item) => Poll::Ready(item),There was a problem hiding this comment.
Hmmm yes, I will review this, better if we remove redundancy. I just used the code that was already in place, but this change looks better.
src/stream/stream/merge.rs
Outdated
| Poll::Ready(Some(item)) => Poll::Ready(Some(item)), | ||
| Poll::Ready(None) => second.poll_next(cx), | ||
| Poll::Pending => match second.poll_next(cx) { | ||
| Poll::Ready(Some(item)) => Poll::Ready(Some(item)), |
There was a problem hiding this comment.
| Poll::Ready(Some(item)) => Poll::Ready(Some(item)), | |
| Poll::Ready(item) => Poll::Ready(item), |
The same applies here
| lst.sort_unstable(); | ||
| assert_eq!(&lst, &[1u8, 2u8, 3u8]); |
src/stream/stream/mod.rs
Outdated
| let mut lst = Vec::new(); | ||
| assert_eq!(s.next().await, Some(1u8)); | ||
| assert_eq!(s.next().await, Some(2u8)); | ||
| assert_eq!(s.next().await, Some(3u8)); | ||
| assert_eq!(s.next().await, None); | ||
| while let Some(n) = s.next().await { | ||
| lst.push(n) | ||
| } |
There was a problem hiding this comment.
I think I can write like this.
use async_std::stream::FromStream;
let v = Vec::from_stream(stream).await;There was a problem hiding this comment.
Yes, this makes it more clear :)
|
I made the requested changes and reduced the number of match arms in the second match (two arms had the same value). |
This implements #490. Let me know of any improvements that can be done!