Rust async support: WIT stream<T> -> Rust stream<Vec<T>>

[mariopasquali]

I am studying the new async support and trying to figure out the rationale behind the WIT stream<T> translating into Rust stream<Vec<T>> instead of stream<T>.

A common use of async streams in WIT is for sockets and http, where a stream of bytes (chunks) is needed. In this case stream<u8> covers the need, but this could also be achieved by defining a stream<list<T>> in WIT which would translate to stream<Vec<u8>> in Rust. On the other hand, a stream of complex records which should be processed individually cannot be expressed with a WIT stream, as this would result in a stream of complex records chunks.

Wouldn't it be more universal to define as WIT stream<T> as a Rust stream<T>, to cover all use cases?

[cpetig]

This is a good question, I simply accepted the implementation as is. I asked about some other details and always learned that a lot of thought had been put into the design.

My guess is that it avoids allocations by reading and writing multiple objects at a time - the ABI representation is for sure a stream<u8> and the Vec is only present on API level.

@dicej would be able to answer this question with more expertise.

[dicej]

Component model streams permit the guest to send and receive arbitrary numbers of items per call to stream.write and stream.read, respectively. For example, a single stream.write call could be used to send 4096 T items to a stream<T>, i.e. 4096 u8s, 4096 tuple<u8, string>s, 4096 list<list<u32>>s etc. Representing the writable end of a stream<T> as a Rust Sink<T> in that scenario would require 4096 calls to SinkExt::send instead of the single call required for Sink<Vec<T>>, and a naive implementation of SinkExt<T>::send would require 4096 calls to stream.write (although we could buffer the items internally and only call stream.write on flush). Likewise for Stream<T> vs. Stream<Vec<T>> on the readable end (although again we could make use of an internal buffer to limit the number of stream.read calls required).

I'm open to using Sink<T>/Stream<T> instead of Sink<Vec<T>>/Stream<Vec<T>>, but it feels more awkward to use, especially in the common case of stream<u8> where the application already has a buffer of bytes and would like to send them all at once rather than byte-by-byte.

On the other hand, a stream of complex records which should be processed individually cannot be expressed with a WIT stream, as this would result in a stream of complex records chunks.

Wouldn't it be more universal to define as WIT stream as a Rust stream, to cover all use cases?

I believe the current approach does cover all use cases, although I agree that it's awkward to force apps which really only care to send or receive a single item at a time to wrap output in a Vec on the write end and iterate over a Vec on the read end.

One possible solution here would be to specialize code generation for stream<u8> to produce AsyncWrite/AsyncRead implementations instead of Sink/Stream implementations, and then do as you suggest for all other stream<T>. That still leaves the question of how to handle e.g. stream<u32>, where apps might reasonably want to send/receive more than one value at a time. We could even consider adding a generate! option allowing the application to specify which style they want.

[mariopasquali]

Thank you very much for the feedback!

My understanding from this feedback is that the intent of the Component Model stream<T> is to express a stream of unitary T (not a stream of Vec<T>), and that the choice of mapping a WIT stream<T> to a Rust stream<Vec<T>> is for efficiency reasons, not to actually express a stream of Vec<T> (which could be sought as frames or packets). This is already a useful clarification as I was confused about the actual intent.

This means that there is no guarantee on how the stream reader side will provide the elements based on how these are sent on the stream writer side. So if a writer would write 200 items, 50 items and 6 items, the reader side could provide these in any way, there is no assumption that the items shall be provided in the same chunks as by the writer. In this example the reader could get 200 / 50 / 6 items (same as writer) but could also get a single chunk of 256 items, or any other possible combinations. Is this correct?

In my application, my actual need is a "stream of frames", so on the Rust side, I would like having a stream<Vec<u8>> where each read or write operation would be for one frame. It looks like I should declare a stream<list<u8>> in WIT then, but this will result in a stream<Vec<Vec<u8>>> in Rust, which isn't exactly what I would like in this case.

I am unsure about the best approach to improve the situation for application of unitary T streams, but I do confirm that having to use a stream<Vec<Vec<u8>>> in my application would be a little awkward :)

[dicej]

So if a writer would write 200 items, 50 items and 6 items, the reader side could provide these in any way, there is no assumption that the items shall be provided in the same chunks as by the writer. In this example the reader could get 200 / 50 / 6 items (same as writer) but could also get a single chunk of 256 items, or any other possible combinations. Is this correct?

Not exactly. streams are unbuffered, meaning that when the writer writes 200 items, that write will block until the reader reads at least one item. There's currently no way for the writer to write the next 50 items until the reader has finished reading the first 200 since there's nowhere to put them, i.e. no buffer on the host side to stash them in. So there's no way for the reader to get a single chunk of 256 items all at once unless the writer writes at least 256 items all at once.

In my application, my actual need is a "stream of frames", so on the Rust side, I would like having a stream<Vec> where each read or write operation would be for one frame. It looks like I should declare a stream<list> in WIT then, but this will result in a stream<Vec<Vec>> in Rust, which isn't exactly what I would like in this case.

Yeah, I agree that's awkward. Probably the quickest way to address that would be to add a wit_bindgen::generate! option like I suggested above which indicates the app wants Sink<T>/Stream<T> implementations generated for the stream<T> WIT type rather than Sink<Vec<T>>/Stream<Vec<T>>.

[mariopasquali]

Not exactly. streams are unbuffered, meaning that when the writer writes 200 items, that write will block until the reader reads at least one item. There's currently no way for the writer to write the next 50 items until the reader has finished reading the first 200 since there's nowhere to put them, i.e. no buffer on the host side to stash them in. So there's no way for the reader to get a single chunk of 256 items all at once unless the writer writes at least 256 items all at once.

Is this a requirement of the Component Model spec, the current implementation of the wit-bindgen client code, or the assumed implementation of the host? (hopefully my question isn't too obvious!)

Yeah, I agree that's awkward. Probably the quickest way to address that would be to add a wit_bindgen::generate! option like I suggested above which indicates the app wants Sink<T>/Stream<T> implementations generated for the stream<T> WIT type rather than Sink<Vec<T>>/Stream<Vec<T>>.

I don't have the big picture to suggest what would be the right approach, so happy with anything you feel reasonable. An approach I could suggest, that wouldn't cover all cases but would at least cover mine, is to generate a stream<T> in Rust if a stream<list<T>> is declared in WIT. Maybe that could be default overridable in a wit_bindgen::generate! option later on, or maybe another better approach could be identified.

[dicej]

Not exactly. streams are unbuffered, meaning that when the writer writes 200 items, that write will block until the reader reads at least one item. There's currently no way for the writer to write the next 50 items until the reader has finished reading the first 200 since there's nowhere to put them, i.e. no buffer on the host side to stash them in. So there's no way for the reader to get a single chunk of 256 items all at once unless the writer writes at least 256 items all at once.

Is this a requirement of the Component Model spec, the current implementation of the wit-bindgen client code, or the assumed implementation of the host? (hopefully my question isn't too obvious!)

It's (a consequence of) the requirements of the spec. See https://github.com/WebAssembly/component-model/blob/main/design/mvp/CanonicalABI.md#-canon-streamfuturereadwrite, especially this part:

Introducing the copy function in chunks, copy first checks that the Waitable at index i is the right type and that there is not already a copy in progress. (In the future, this restriction could be relaxed, allowing a finite number of pipelined reads or writes.)

So there can only be one outstanding write to a given stream at a time, and the reader can only receive at most the number of items the writer provided for that write. Per the quote above, we could relax the one-write-at-a-time restriction later, but currently it's enforced by both the spec and the implementation.

[dicej]

An approach I could suggest, that wouldn't cover all cases but would at least cover mine, is to generate a stream<T> in Rust if a stream<list<T>> is declared in WIT. Maybe that could be default overridable in a wit_bindgen::generate! option later on, or maybe another better approach could be identified.

I'd be concerned about surprising and confusing users who aren't expecting that kind of specialization.

Another option to address your case would be to create newtype wrappers for the StreamWriter and StreamReader structs like this:

struct StreamSingleWriter<T>(StreamWriter<T>);

impl<T> Sink<T> for StreamSingleWriter<T> {
    // Implement `start_send` etc. by wrapping the single item in a `Vec`
    // ...
}

struct StreamSingleReader<T>(StreamReader<T>, Vec<T>);

impl<T> Stream<T> for StreamSingleReader<T> {
    // Implement `poll_next` by reading a `Vec`, storing it in `self` and draining it item by item
    // ...
}

[mariopasquali]

Thank you for the feedback.

There are indeed various ways to handle this on the client side. I will work this out on my side for now, and happy if at some point more flexibility in the generated code will be available.