Redefine Stream/Sink/AsyncRead/AsyncWrite/etc on top of Future

[Matthias247]

I think in the async/await world it might be preferable to redefine the traits and types like the ones mentioned in the title on top of Futures, and e.g. move from poll_something() signatures to methods that return Futures.

E.g.

trait Stream<T> {
    type Item: Future<Item=T>;
    fn read(&mut self) -> Self::Item;
    // or something along the following, whatever works best
    fn read(&mut self) -> impl Future<Item = T>
}

The important reason for this, that it should be possible to implement those types on top of Futures as primitives, which isn't possible today.

As a motivating example I'm thinking about a MPMC channel, which can be built around some async primitives as shown below:

struct ChannelImpl<T> {
    state: AsyncMutex<Queue<T>>;
    has_space: AsyncManualResetEvent;
    has_element: AsyncManualResetEvent;
}

impl ChannelImpl<T> {
    async fn write(&mut self, item: T) {
        loop {
            await!(self.has_space.wait());
            let mut queue = await!(self.state.lock());
            if queue.free_space() == 0 {
                continue;
            }
            queue.push(item);
            break;
       }
       self.has_element.set();
    }

    async fn read(&mut self) -> T {
        let mut result: Option<T> = None;
        loop {
            await!(self.has_element.wait());
            let mut queue = await!(self.state.lock());
            if queue.len() == 0 {
                continue;
            }
            result = Some(queue.pop());
            break;
       }
       self.has_space.set();
       result.unwrap()
    }
}

I'm not yet sure if the current state of async functions and traits allow all this (please educate me!), but I think this should be one of the goals that async programming in Rust should enable.

This kind of implementation is however currently not wrappable in a Stream or Sink implementation.
We can't stuff this kind of code inside a poll_xyz method of a Stream/etc, since it requires temporary Futures that all carry their own poll state to be instantiated in the task. Those Futures must maintain their state between calls to poll(). E.g. polling an asynchronous mutex the first time creates a registration in the waiting list, and dropping it cancels the registration. If we would try to use an asynchronous mutex inside a poll_next method, we must always need to drop it before poll_next() returns, which would cancel the registration and never allow use to retrieve the mutex.

I think the more general observation might be, that Futures allow for persisting some additional temporary state per operation, and provide additional lifecycle hooks per operation (Future gets polled for the first time, and Future gets dropped) compared to the current Stream definitions.

Another side effect is that currently Streams/etc must all support pinning. In Streams that return Futures that wouldn't be the case, since the Futures get pinned. They might indirectly get pinned because the respective Futures might reference the Stream through a reference and a certain lifetime, but that relationship is already directly checked through Rusts normal means, and e.g. doesn't put special requirements regarding pinning on the implementation of the type. And often we would Streams being moveable after they have been partly consumed (e.g. a Tcp Socket), which is not allowed if they are pinned. Of course some of them might be Unpin, but this kind of change would allow moves again for all Streams.

[Nemo157]

Having the core traits return futures directly are probably blocked on GATs (or all the way to async methods, although I’m hopeful that there will be good cross-compatibility between async methods and their underlying GAT representation). While it is possible to have a stream that can produce the futures for more than one item at a time, I think the common case is that the stream can only produce a future for the next item. For example a trait like

trait Stream {
    type Output;
    type Next<'a>: Future<Item = Option<Self::Output>> + 'a;

    fn next(self: Pin<&mut self>) -> Self::Next<'_>;
}

I reintroduced Pin to the method as I have a usecase where the stream must be pinned from the first call to next until it is dropped, doing the setup and teardown of the immovable part on every call to next would be too expensive. I’m not sure how common this might be, so I guess I could impl Stream for Pin<&mut MyStream> instead and eat the cost of a double indirection if pinning is dropped from the Stream API. Maybe if we have some way to construct streams via async fn this would be common enough to keep it as part of the trait.

[Matthias247]

Having the core traits return futures directly are probably blocked on GATs (or all the way to async methods, although I’m hopeful that there will be good cross-compatibility between async methods and their underlying GAT representation).

I was afraid of that. I don't know yet what GATs exactly are but will try read up on them. Can you elaborate how a signature for a stream would need to look like in order to make it implementable by future composition as outlined above?

I would say as long as that is not possible, we would maybe face the general issue that it's not possible to define abstractions around types that are implemented by async functions. And e.g if I built an async powered HTTP client, I can't built an interface/trait around it that I use for unit-testing an dependency-injection in other code.

While it is possible to have a stream that can produce the futures for more than one item at a time, I think the common case is that the stream can only produce a future for the next item.

You are right, most streams will only support producing a single item at a time. Supporting more would require internal queuing (which can easily be implemented through AsyncMutex in some situations, but is hard for manually implemented Streams). E.g. if we look at other implementations glib only does support one concurrent outstanding operation per IO type. WinAPI/IOCP supports more (that's what the term Overlapped tries to describe), but there's barely a need for it.

I think that aspect could be most cleanly solved by the type system, if reading from the stream produces a Future and consumes the stream until the operation had been finished - where we get it back. Allowing the creation of multiple futures, and then either blocking in their implementation (via AsyncMutex) or reporting an error if no concurrent operations are allowed is another option.

I reintroduced Pin to the method as I have a usecase where the stream must be pinned from the first call to next until it is dropped, doing the setup and teardown of the immovable part on every call to next would be too expensive. I’m not sure how common this might be, so I guess I could impl Stream for Pin<&mut MyStream> instead and eat the cost of a double indirection if pinning is dropped from the Stream API

I can see that some Streams must have stable addresses, but I'm not sure whether that fact needs to be propagated or is an implementation detail. E.g many streams will be be implemented through Arc<Mutex<InnerStream>>, which effectively pins the inner stream, but that doesn't leak into the API and still makes moveable. What's your use-case for this?

[Nemo157]

GATs are generic associated types, that's the feature that allows parameterizing the Next associated type by the lifetime 'a in my example.

I think that aspect could be most cleanly solved by the type system, if reading from the stream produces a Future and consumes the stream until the operation had been finished - where we get it back.

This is where having GATs is useful, rather than consuming the stream and receiving it back once complete you can borrow the stream, and have that borrow last until the future is complete/dropped.

What's your use-case for this?

My use case was reading a stream of values from a radio peripheral in a microcontroller. By using pinning the stream can have an internal buffer that is written to by the radio using DMA, without pinning this buffer would need to be externally allocated and passed in as a reference (or heap allocated, but this was for a no_std project).

[Matthias247]

This is where having GATs is useful, rather than consuming the stream and receiving it back once complete you can borrow the stream, and have that borrow last until the future is complete/dropped.

Ah, that makes sense. So the following wouldn't work, because without GAT we can't put the lifetime on the item?

trait Stream {
    type Output;
    type Next: Future<Item = Self::Output>;

    fn next(self: &'a mut) -> Self::Next<'a>; // Or is that Self::Next + 'a ?
}

After reading GATs I understand it makes lifetimes more flexible, but I wasn't sure whether something is possible without them. It should be OK for this use-case if the lifetime of the returned Future is bound to the stream.

My use case was reading a stream of values from a radio peripheral in a microcontroller. By using pinning the stream can have an internal buffer that is written to by the radio using DMA, without pinning this buffer would need to be externally allocated and passed in as a reference (or heap allocated, but this was for a no_std project).

Ah, cool idea to represent it as a stream. I think however your requirement is more the typical pain point of an embedded system: One shouldn't dynamically allocate, and therefore ideally all buffers/etc. should be contained inside the type.

However without e.g. const generics that's already painful, and even with I don't find the embedding approach super appealing (there's an endless forwarding of type parameters).
I guess I would go with either the externally allocated route (which gives the stream a lifetime, but that can be 'static for a global buffer). Or maybe cleaner: Create a simple bump allocator for the platform, and change the strategy from "don't allocate at all" to "only allocate at startup".

[Matthias247]

I tried things out. I seems to be only possible to define this at the moment:

trait Stream<'a> {
    type Output;
    type Next: Future<Item = Self::Output> + 'a;

    fn next(&'a mut self) -> Self::Next;
}

which is arguably not nice due to the lifetime which needlessly is on the trait.
Haven't yet tried out whether that's implementable through async function composition.

[aturon]

cc @withoutboats @cramertj

[cramertj]

I discussed this with @erickt elsewhere and explained my thoughts. As the ecosystem stands, it's not possible to make this work without making all of these traits non-object-safe, so I'm not interested in pursuing this at the moment.

[seanmonstar]

One thing that come to mind with regards to AsyncRead/AsyncWrite, if the poll_read/poll_write methods are removed, and there are only async fns, then we can't really wait on both the read and the write future, since both will want to mutably borrow the transport and not release the borrow until complete.

[Matthias247]

@cramertj: would you mind sharing these publicly at some point? I guess that would be interesting for others too. The implications on the ecosystem are totally understood. Although the current changes from 0.1 already are lots of those smaller renaming things. And this would mostly be another. We can still have the old things around, with adapters to the new variants. This direction is possible to bridge. However it’s not possible to move future based thingies into the current signatures without costs (eg extra allocations)

@seanmonstar: I think this should still be possible as all types that support read and write should also support a split method for getting individual halves in order to make them more flexible. Then you can get a future from the reader and one from the writer.
In lots of Protocol implementations i would want to give ownership of the writer to some writing task, and keep the reader in the reading task. Then I need split anyway.

[cramertj]

@Matthias247 Oh sorry, I don't have a large writeup or anything, I just mean't we had discussed it. If there were a way to return a dyn<size_of::<usize>> Future<Output = io::Result<()>> (dynamic future with maximum one-pointer size) I'd support that, but that feature doesn't exist yet.

[Matthias247]

Ok. Let's focus on this issue, instead of spreading the discussion out.

I think e.g. AsyncWrite should be de defined along:

trait AsyncWrite {
    type Error;
    type WriteFut<'a>: Future<Output = Result<usize, Self::Error>> + 'a;

    fn write(&mut self, bytes: &[u8]) -> Self:: WriteFut <'_>;
}

which is mostly a copy of the well-known synchronous variant, and behaves just like it when combined with await.

The obvious problem is that this won't work yet without GATs. For Streams/Sinks we can work around this by requiring them to be implemented through Arcs, which at least for std environments isn't a big issue. That removes the lifetime from the returned Futures.

I am aware that the inability to create trait objects is an issue. But not being able to compose Futures and other async traits seems like an even bigger one to me currently.

Just as a motivating example: I have written a completely async HTTP/2 library a while ago, which I still think has a reasonably simple API and implementation. The API provides users of the library some byte stream based interfaces, which are similar to AsyncWrite/Read: https://github.com/Matthias247/http2dotnet/blob/master/Http2/ByteStreams.cs

The implementation of this interface is heavily based on top of Future/Task composition, as it can be seen here: https://github.com/Matthias247/http2dotnet/blob/master/Http2/StreamImpl.cs#L520-L591

Writing data includes blocking on async mutexes for synchronization of long-lived tasks, sending data asynchronously to a background task which performs all IO multiplexing and more async waiting until the data has been actually flushed. This can't really be done with the the current Rust traits. It seems like it would be back to manually implementing state-machines for all those operations.

Therefore I would really like to see a mechanism where we can implement those common async base traits on top of Futures - ideally in a zero cost fashion. Since it seems like this isn't yet possible I'm totally ok with having some intermediate solutions in between. But I wouldn't like the whole ecosystem to commit already on those. Or even to build a completely separate and incompatible world on top of those.

[seanmonstar]

@Matthias247 The "Ext" traits allow you to easily treat read and write as Futures. Changing the core traits to remove their poll_* methods and only return a Future has two issues that I can see:

1. It's pretty common to make boxed trait objects of IO types, like Box<AsyncWrite>. With those associated types, this becomes a whole lot harder. Essentially, you end up needing to require boxing the future as well, to get some sort of Box<AsyncWrite<WriteFut = Box<Future<..>>>.

2. As I mentioned before, async fns end up causing a borrow on &mut self. This prevents one from being able to wait on both a read and write future at the same time. It's possible to get that back by putting the IO object inside an Arc with some AsyncMutex, but that is no longer a "zero cost fashion".

Given that the style of writing code that you want want is already possible thanks to the "Ext" traits, it seems a shame to remove the ability to have zero cost async read + write.