fix: avoid race condition between pending frames and closing stream

[thomaseizinger]

Currently, we have a garbage_collect function that checks whether any of our streams have been dropped. This can cause a race condition where the channel between a Stream and the Connection still has pending frames for a stream but dropping a stream causes us to already send a FIN flag for the stream.

We fix this by maintaining a single channel for each stream. When a stream gets dropped, the Receiver becomes disconnected. We use this information to queue the correct frame (FIN vs RST) into the buffer. At this point, all previous frames have already been processed and the race condition is thus not present.

Additionally, this also allows us to implement Stream::poll_flush by forwarding to the underlying Sender. Note that at present day, this only checks whether there is space in the channel, not whether the items have been emitted by the Receiver.

We have a PR upstream that might fix this: rust-lang/futures-rs#2746

Fixes: #117.

[mxinden]

Would the following alternative approach as well be an option?

• Currently we have a single channel for all streams to send StreamCommand to the connection. How about a unique channel between each stream and its connection?
• Flushing a stream is simply a matter of calling Sink::poll_flush on the mpsc::Sender.
• Garbage collection can use the Poll::Ready(None) returned by a mpsc::Receiver from a stream as a signal instead of Arc::strong_count.

Given that the above would not require a new command variant and would not require sending Wakers over mpsc::channels, I consider it to be simpler.

@thomaseizinger what do you think? Am I missing something?

[thomaseizinger]

I like that idea! Much simpler. Will implement! :)

[thomaseizinger]

I like that idea! Much simpler. Will implement! :)

There is one caveat. poll_flush on Sender only checks whether we can send more items into the channel, not whether they have been taken out on the Receiver end. Thus, we need to set the capacity to 0 for this to work.

See rust-lang/futures-rs#2504.

[mxinden]

I like that idea! Much simpler. Will implement! :)

There is one caveat. poll_flush on Sender only checks whether we can send more items into the channel, not whether they have been taken out on the Receiver end. Thus, we need to set the capacity to 0 for this to work.

See rust-lang/futures-rs#2504.

As far as I can tell, this should be fine. We could even not call poll_flush at all. The receiver will still receive all messages and eventually receive Poll::Ready(None). See https://play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=def93d1497874d26f93dabf4a74b3feb.

[thomaseizinger]

I like that idea! Much simpler. Will implement! :)

There is one caveat. poll_flush on Sender only checks whether we can send more items into the channel, not whether they have been taken out on the Receiver end. Thus, we need to set the capacity to 0 for this to work.
See rust-lang/futures-rs#2504.

As far as I can tell, this should be fine. We could even not call poll_flush at all. The receiver will still receive all messages and eventually receive Poll::Ready(None). See play.rust-lang.org/?version=stable&mode=debug&edition=2021&gist=def93d1497874d26f93dabf4a74b3feb.

I am not sure what you are trying to say? The point of this PR is to have poll_flush park the task if the channel is not yet empty. How can we achieve that if we don't forward to the poll_flush call in Sender?

FWIW: I attempted to fix this upstream: rust-lang/futures-rs#2746

[thomaseizinger]

• Garbage collection can use the Poll::Ready(None) returned by a mpsc::Receiver from a stream as a signal instead of Arc::strong_count.

Ah I forgot about this. This is a bit trickier to implement because I currently use SelectAll which hides this detail from me.

[thomaseizinger]

@mxinden This is now available for review again. I implemented your suggestion. It is the much better design :)

[mxinden]

This looks good to me overall, though one suggestion around using a SelectAll.

[mxinden]

Polling each stream Receiver even though only one might be Poll::Ready seems wasteful. Would SelectAll not be an option? One would wrap each Receiver such that it returns the StreamId when the Receiver returns Poll::Ready(None). With that StreamId we can then call self.on_drop_stream(id).

(On consecutive polls the wrapper can return Poll::Ready(None) and thus it would be cleaned up by the SelectAll.)

[thomaseizinger]

I had a SelectAll based design before but dropped it because I couldn't detect None from the outside.

Detecting that requires another data structure for translating between the stream commands and the actual frames in connection (the wrapping you mentioned).

This also needs to work for the Cleanup and Close case.

I tried hiding it in a custom collection object but we need access to the stream IntMap upon drop so that needs even more refactoring but would be a clean design.

I can invest the time if you want but I don't think it is a quick 30min refactoring.

[thomaseizinger]

I was wrong, it was actually a super quick refactoring, now that I've seen how tokio's StreamMap works. They recommend an additional wrapper around Stream. The trick is to wrap the output of the Stream again such that you get a tuple of (key, Option<Item>) which allows you to detect closing of the stream.

[mxinden]

Thanks for the follow-up. Also great to be aware of StreamMap.

[mxinden]

Instead of using pin-project, we could as well implement Unpin for Stream.

impl Unpin for Stream {}

That said, I doubt there is any async Rust code-base without pin-project in its dependency tree already. Thus fine to include it here as well.