WIP: Add stream method to channel

[jean-airoldie]

This method returns a type which implements Future::Stream.

[jean-airoldie]

Hum, seems like I ran rustfmt which explains most of the modificiations. Is there a specific reason why the repo is not rustfmt'ed?

[Matthias247]

Hi! Thanks for looking into this - this is definitely something super-cool to have! I recently thought about it myself, and basically about the same kind of implementation (store the last Future inside the Stream). This should then also enable the use of those channels as trait objects, which is very good to have.

There is question about the logical behavior mentioned in the review: I'm currently not sure if storing the Future for a longer amount of time could have any unexpected side-effect (other receivers can't progress). I would need to read some code again for this.

The general concept and pinning looks reasonable from my side. But maybe @Nemo157 who looked before at Future/Stream conversions could take a short peek and give an opinion on whether it looks correct. It might also be easier if this would start using one of the pin projection crates. But since those are also not yet used inside the remaining codebase that is certainly not a must-have for this change.

Hum, seems like I ran rustfmt which explains most of the modificiations. Is there a specific reason why the repo is not rustfmt'ed?

No. The only reason is I didn't spend the effort on figuring out how to use it when I wrote it. I'm definitely good with having things formatted. However it might be a bit nicer to keep formatting and new content in separate CRs, to make things easier to review.

[Matthias247]

That definitely seems the way to go for a Future -> Stream adapter. From a pinning point of view it looks correct, since the Stream also needs to be pinned.

What I'm however not yet sure is whether it's ok from a logical point of view. If a user would start polling on the Stream but not getting an item, the Future would still be stuck in a signaled state in the Stream. If that prevents other waiters on the Stream from getting woken up it might be problematic. I think I would need to review the acutal wakeup logic to understand the impact - e.g. whether the Channel kept track of Futures it notified but which did not yet pick up an item. The Mutex implementation definitely does this in order to guarantee fairness. The Channel might need for rendevousz semantics?

It had been a while since I last looked into it. Will do on the weekend.

[jean-airoldie]

So a scenario would be user create a stream, polls the stream once, which returns Poll::NotReady then drops the stream future. The inner future is stuck in the stream and might be blocking subsequent futures.

[Matthias247]

Right. This can also happen with the Future based API. But there you are a lot more likely to see that an old Future is still around - especially since you can't move the Futures. In the Streams it is somewhat hidden.

But now since I wrote that: At least the Stream also needs to be pinned and can not be moved - so the user moving around the Stream without knowing that some registration is still active is not possible.

[jean-airoldie]

I'll write a separate PR for rustfmt and i'll add a travis.yml which will automatically run it. Then you can tweak the linter to you liking.

[jean-airoldie]

If you merge said rustfmt PR i'll rebase against it so that the current PR becomes more readable.

[Matthias247]

I have reviewed the code again. As mentioned, there could be an issue when someone starts reading from the Stream but then doesn't complete the read (doesn't extract a value). When the read is started the Future is registered in the Channel in the waiters list. When someone wants to send a value, the laster waiter in the waiter list is signalled here. This could be e.g. the Future which is stored in the Stream. Other tasks are not signalled.

That means e.g.

• if the channel capacity is 1
• 2 tasks are waiting on the channel
• the first (Stream) task gets signalled, but doesn't read from the channel anymore

then the second task will wait for the notification until the Stream gets dropped (which leads to waking up another waiter).

If during the situation another producer would send something to the channel, it would

• insert that sender into the wait list
• wakeup the old second task

Then that task would actually extract the old value, and the value from the second sender is shifted into the channel buffer.

Therefore it seems like we might only lose the buffer and not run into a permanent deadlock state, but I'm still not 100% sure. Need to think further about other edge-cases.

In order to avoid the issue totally each send could wake-up all stored receivers instead of only a single one. However that would introduce a lot of churn on normally used channels with multiple producers and consumers. E.g. if there are 5 consumers

• they all would register at the channel and wait
• a single send would unblock all of them
• only consumer gets the value
• at least 4 consumers register themself again at the channel

which means lots of churn in the executor. Therefore I'm not too keen on implementing a change to wake-up all receivers.

[jean-airoldie]

Lets say we did would run into a permanent deadlock, couldn't we circumvent it with a drop implementation so that the future cleans up after himself? For instance, by retrieving the value on drop by polling one last time?

[Matthias247]

Where would you drop? The ChannelReceive/SendFuture which you store in the Stream already implements Drop and wakes up another waiter if it had been notified, but the notification not been used. So if you drop the stream everything is good again.

The concern here is mainly that there is some hidden state in the Stream which might cause some none obvious behavior for some users. When someone uses the pure Future based APIs it is a bit more obvious, since the Future is still visible / in scope. When that gets dropped also everything is good again.

[jean-airoldie]

I did some very basic testing in e96b7ff and it seems like for the low level future API its not a big deal. From what I understand a forgotten future (send & receive) will eat up one wakeup, but everything works as normal after that. The values stored in the buffer won't be affected so its basically like a wakeup was lost in the aether.

If a receive future is forgotten, the next time a receive future is called, the issue is fixed because the buf contains an element. The same goes of a send receive.

[Nemo157]

so its basically like a wakeup was lost in the aether.

Lost wakeups can lead to deadlocks if the task is not concurrently doing anything else or never encounters a spurious wakeup. That could be allowable though, leaking a future/stream instead of dropping it already sort of violates the expected usage.

I'll try and take a look at this in a bit.

[Matthias247]

I think I now have better understanding on what will happen regarding the lost wakeups: Every lost waker will cause one sender to be blocked. However on each consecutive send() operation another receiver will be woken up. Independent of whether the sent element went into the internal buffer or the sender got attached to the waiter list. That will unblock a waiting receiver.

That means in my understanding the maximum amount each non-used wakeup would cause an exactly on element in the channel being not used, but successive reads and writes should be succesful. That means a latency of 1 element would be introduced into the system. I think that might be acceptable in typical MPMC workflows - where elements are unrelated and latency is likely unpredictable. In single consumer workflows this should never happen, since that would mean the only consumer would not read from the channel. For

To answer the question from yesterday why we don't run into the debug assert: The receivers always read from elements in the buffer. While reading one element from the buffer, they then also copy one element from the oldest send waiter into the buffer in order to unblock one more sender => There is no direct handover from a sender to the receiver. The buffer is always used in between.

[jean-airoldie]

That means a latency of 1 element would be introduced into the system. I think that might be acceptable in typical MPMC workflows

For a continuous workflow this is acceptable indeed. For temporary workflow, one of the sender would close the channel after its done which would wakes up all waiters.

[Matthias247]

Just some minor comment improvements. Maybe you can merge those

[jean-airoldie]

I was away for a while and I forgot about this PR.

What do you think would be necessary before this can be merged? What kind of additional tests do you think would be needed?

[Matthias247]

Sorry for the late feedback. I convinced myself that this is good to have that way. If we just degenerate on buffers instead of locking it seems acceptable. And @Nemo157's comment that even locking might be acceptable due to the fact that not using a Stream might be treated as misbehavior is interesting. Although I prefer to let people not run into surprises - which then lead to "why doesn't that work" tickets posted :-)

One random thought I had was whether the Stream support could be behind a feature flag. That way we could increase the chance that people read something about the feature in a readme instead of just trying to use APIs. But on the other hand just having a feature for this API seems weird. Maybe I would rather have one which disables all dependencies on future-core.

Another random thought I had is that the Stream support could also be implemented in another fashion: Instead of storing a Future in the Stream it could also directly store a ListNode. That would make it probably a tiny bit more compact. But on the other hand delegating it to the Future internally seems like less code to maintain, so I guess I have a preference for the path taken.

[jean-airoldie]

It would be interesting it we could add a compiler warning if the user misuses the stream. Kinda like #[must_use] but instead generates a warning when a stream is not driven to completion.

edit: It probably impossible to do that statically.

[Matthias247]

I was away for a while and I forgot about this PR.

What do you think would be necessary before this can be merged? What kind of additional tests do you think would be needed?

Ah, good point around testing. I just reviewed again what this has and have added a few suggestions based on it. Apart from that I think just checking what the other channel tests have, and whether we miss an importatant edge condition would be good. But since we just delegate to the existing stream most things should be pretty well covered.

[Matthias247]

It would be interesting it we could add a compiler warning if the user misuses the stream. Kinda like #[must_use] but instead generates a warning when a stream is not driven to completion.

That sounds cool. But I have no idea how, since it's a dynamic property. It would somehow need to track that an operation returned Pending and the object wasn't used from then on until Drop. And event that is a legitimate use-cases inside select! : The operation could just have been cancelled because it wasn't required to run to completion. Therefore time plays a major role in the decision.

I guess runtime checks could be used: When the object last retuns Poll store a timestamp. On drop() compare the timestamps and if a "long" timeframe has exceeded emit a warning (or debug_assert!). But no idea for compile-time check possibilities.

[jean-airoldie]

I also added some very vague doc concerning the SharedStream & ChannelStream to warn the user against potential lost wake-up notifications.

[Matthias247]

I merged this now. Thanks for working on it!