Stream::merge does not end prematurely if one stream is delayed

[matklad]

r? @yoshuawuyts

The bug was that if left returned Poll::Pending, the right could have returnrf Poll::Ready(None), and we would consider the stream as finished, although we haven't really polled the left part thoroughly enough.

I also don't understand why we need cx.waker().wake_by_ref(); here: we are returning Poll::Ready, and calling wake_by_ref is only required if you return Pending?

[matklad]

Question (unrelated to PR mostly): is it a contract of Stream trait that, after returning Poll::Ready(None) once, all subsequent calls to poll_next should return Poll::Ready(None) as well? I think both initial and this version of merge assume this behavior, but it is not documented. I think merge doesn't neccaary have to require this contract from substreams, if we change it's internal state to

enum MergeState {
    Both(L, R),
    Left(L),
    Right(R),
}

[ghost]

Does this sleep have any significance? Perhaps it'd be safer to pick a larger duration, maybe 500ms?

[matklad]

without this sleep, the test can pass by accident, as we would not poll a Poll::Pending out of the channel. Basically, we need to ensure that the t tasks is polled before we do a send via channel. Bumped the dealay to 500ms: we can trivially dial it back if we find test-suite to be too long!

[matklad]

to be more precise, the unfortunate timing here can't lead to false positive: if the code is correct, the test will always pass. However, timing can lead to a false-negative, where a buggy code would pass the test.

I've considered making this deterministic by writing a stupid Stream which answers with Pending the first time it is polled and with Ready(Some(92)) the second time. In the end, I've decided to go with channels + sleep to simulate a similar effect: writing a stream impl manually can easily introduce errors of it's own.

[matklad]

Hm, judging by the StreamExt::fuse method, we don't actually have this guarantee. so we should call .fuse ourselves int the Merge ctor

[yoshuawuyts]

Hm, judging by the StreamExt::fuse method, we don't actually have this guarantee. so we should call .fuse ourselves int the Merge ctor

Took some time to think about this, and you're entirely right. Phew, you just caught a whole lot of bugs here. Thanks so much for this patch!

[matklad]

@yoshuawuyts i’ve even found one (debatable though) in std::iter::Chain: rust-lang/rust#66031 as a result of yesterday’s thinking about cancellation. Still, haven’t figured our cancellation itself :D

[matklad]

added fusing as well

[yoshuawuyts]

The implementation looks good to me, thanks so much @matklad!

@stjepang it seems your concerns have been addressed too, can you verify this is the case?

[ghost]

Yes, very happy with this :)