WHATWG streams for data channel messages

There is a long standing issue with large data channel messages in the existing API of RTCDataChannel. If I want to send a single message containing 1 GiB of data (for example a large file), I have to have this gigantic message in memory at the time of sending. If I receive a 1 GiB message, this message is slowly being reassembled until it's in memory and then handed out to the application. That creates backpressure and the like.

My idea is to resolve this by extending the RTCDataChannel API in the following way:

Sending

Add a .createWritableStream method which returns a WritableStream instance. With the associated WritableStreamWriter instance, one can add chunks by calling .write on the writer. Once the writer is closed, the message is considered complete.

Receiving

If .binaryType is set to stream, the event raised for onmessage contains a ReadableStream instance that is being created when the first chunk is being received. Once the whole message has been read, the reader will return EOF on a .read call (as specified by the streams API).

Edit: What should happen when a string is being received will need to be discussed.

In the meeting, I think there was a slight confusion about what streaming API I meant. Basically, I propose two streaming APIs that use WHATWG streams:

WHATWG streams for the QuicStream API is being discussed here: https://github.com/w3c/webrtc-quic/issues/2
WHATWG streams for the existing RTCDataChannel API. This is what this issue is for. And with the above description it should be clear that for RTCDataChannel there would have to be a stream for each message as data channels transfer datagrams and not a sequence of bytes. This is the point I wanted to make during the meeting. I hope this clarifies it a bit. :)

This is a good idea.

You could define a data channel as a stream of streams, but that might be a little more difficult than the simple addition suggested here. I would try to find a way to model this more on fetch API model if possible, but I'd need to look carefully at the API to decide if that were feasible or just more disruptive.

You can use Blobs to avoid the memory commitment for big things, but that's not a great solution.

See the fetch Body mixin for a model that might be an improvement, at least for receiving.

@annevk

cc @annevk @ricea @domenic (as you might be interested on commenting this)

This would be super-great; I know developers who do both Node.js and Web RTC development, and their frustration in not having a streaming model for RTC data channels is part of what originally motivated the streams work.

The hardest part of designing this is the interaction with existing methods of reading/writing. That is, the most developer-friendly interface is (IMO) dataChannel.writable + dataChannel.readable, i.e. properties that always exist. See https://streams.spec.whatwg.org/#example-both. The issue then becomes figuring out what happens to your existing API when someone acquires a reader or writer from the stream. If possible, just making those APIs not function (e.g. never deliver events, or throw errors/reject promises) is the simplest way to do that.

Let me know how I can help; this would be very exciting to get working. I don't know much about RTC data channels at the moment myself, but perhaps I should read up.

@martinthomson

That is, the most developer-friendly interface is (IMO) dataChannel.writable + dataChannel.readable, i.e. properties that always exist.

This sounds like you'd also like to wrap data channels, so handing out the messages (not the content) could also be done using the streams API? So, basically you'd have a stream that hands out messages where each message itself is a byte stream? I think this is what @martinthomson mentioned as well and it's a cool idea but I think we should start with using the streams API for the messages' content for now. (Wrapping data channels in such a way would make a neat JS lib, though.)

The issue then becomes figuring out what happens to your existing API when someone acquires a reader or writer from the stream.

Good point. This is what I think would be feasible:

Make .send and .createWritableStream raise an exception as long as there's an existing WritableStream instance that's not closed or aborted.
Either make .binaryType read-only once the first chunk has been received or make it read-only as long as a ReadableStream instance exists for an incoming message.

@domenic Maybe you can also comment on @martinthomson's idea regarding the use of fetch Body mixin as I'm not really familiar with it. If answering that would require understanding data channels a little more, send me a mail.

Also, to clarify if it's not clear:

The example you've mentioned (https://streams.spec.whatwg.org/#example-both) sends a WebSocket message for each chunk and when the stream is at EOF the WS connection is being closed. So, if I had one stream associated to a file, I can send that file and then the stream is closed.

Whereas I'm proposing an API where a RTCDataChannel message remains just a message but each message is a stream. When the stream for a specific message is at EOF, the message is complete. This makes it possible to send more than one message. So, if I had one stream associated to a file, I can send that file as one message.

@lgrahl

@lgrahl, a stream of streams only really makes sense if we can resolve the issue of whether messages on data channels are strictly ordered.

Actually, belay that, lack of sleep. I assume that we would only be able to produce a stream of streams if the data channel were ordered. Though perhaps it's OK if there is no requirement for messages to be complete before the next message is delivered. There might be some caveats regarding back pressure in that case though.

If A is sent before B, but the first octets of B are delivered first, we would manifest a new message on the top-level stream. But A might be completed prior to completing B. For instance, the stream might deliver B2, B1, A1, A2, A3, B3. B1 and B2 would have to be reordered by the browser for delivery (messages themselves are always ordered internally), but it has no obligation to hold B until A arrives. In an ordered stream, the browser would have to hold all of B until A arrives.

(It seems like there's a memory exhaustion attack somewhere in there, but I'm sure our SCTP friends will point out the natural defense that applies.)

I'm getting confused by the term stream of streams. Maybe we can call this a sequence of data channel messages where each message is a stream of bytes?

SCTP (with the ndata extension) can interleave messages on different streams but not on the same stream. That means messages on a data channel are strictly ordered (even though messages can be abandoned when using partially reliable delivery - this doesn't affect order).

I'm adding some examples here, so it's more clear how my idea would look like from a user's perspective:

Today's API

Peer A's code:

const dc = pc.createDataChannel(...);
dc.send(largeFile);

largeFile may be a Blob in which case the data doesn't have to be in memory at the time of sending (IIRC, please correct me if I'm wrong) or an ArrayBuffer in which case the data will need to be in memory.

Peer B's code:

const dc = pc.createDataChannel(...);
dc.binaryType = 'blob'; // ... or 'arraybuffer', it doesn't really matter here
dc.onmessage = (event) => {
    // Note: Ignoring the string case here.
    // ... write 'event.data' to disk
};

When the event has been raised, the large file's data has been transmitted entirely which can take a very long time. Also, the data will be in memory at that point.

Streamed (assuming the file is not a stream)

Peer A's code:

const dc = pc.createDataChannel(...);
const writableStream = dc.createWritableStream();
const writer = writableStream.getWriter();

while (true) {
    await writer.ready;
    // Note: Once a WritableStreamBYOBWriter exists, we could reuse the same buffer
    // over and over again.
    const length = Math.min(
        myPreferredChunkSize, remainingLength, writer.desiredSize)
    const buffer = new Uint8Array(length);
    // ... copy the file's chunk here into 'buffer'
    //    (or create a view if you already have the chunk in memory)
    writer.write(buffer);
    // ... continue until all chunks have been written and then break
}
await writer.close();

(Edit: I fixed the example above to not await the call to .write(...) but rather await the .ready attribute on the writable stream writer.)

If something goes wrong, you'd call writer.abort() which would abort sending the message. (We would have to discuss what .abort triggers but it should probably close the channel.)

Peer B's code:

const dc = pc.createDataChannel(...);
dc.binaryType = 'stream';
// We only need one small buffer per channel
const buffer = new Uint8Array(myPreferredChunkSize);
let view = new DataView(buffer.buffer, 0);

// Note: The function is declared async!
dc.onmessage = async (event) => {
    // Note: Ignoring the string case here.
    const readableStream = event.data;
    // You could also use the default reader but then the reader would not
    // copy directly into our buffer.
    const reader = readableStream.getReader({ mode: 'byob' });
    while (true) {
        const chunk = await reader.read(view);
        if (chunk.done) {
            break;
        } else {
            view = chunk.value; // We reclaim the buffer here
            // ... write the view's data to disk
        }
    }
};

The reader can call reader.cancel() to abort receiving data. (We would have to discuss what .cancel triggers but it should probably close the channel.)

This of course has the advantage that the file's data doesn't need to be in memory at once for both peers and that backpressure is being taken care of. Furthermore, writing to disk can be started with the first chunk received.

Streamed (assuming the file is also a stream)

So, the last example looked a lot more complicated than what we have now. But if the file could be read from and written to in form of a stream, it becomes pretty easy.

Peer A's code:

const file = ...;
const dc = pc.createDataChannel(...);
const writableStream = dc.createWritableStream();
await file.readableStream.pipeTo(writableStream);

Peer B's code:

const file = ...;
const dc = pc.createDataChannel(...);
dc.binaryType = 'stream';

// Note: The function is declared async!
dc.onmessage = async (event) => {
    // Note: Ignoring the string case here.
    const readableStream = event.data;
    await readableStream.pipeTo(file.writableStream);
};

Combination

Of course, peer A and B can just as well use the existing API on one side. It's entirely up to them.

There are several other ways to use WHATWG stream reader/writer instances. I've only shown two of them.

Comments? :)

@lgrahl

@lgrahl This looks good. I think in general it's better to have reusable wrappers for things like file APIs so that the simpler and easier pipeTo() style can be used.

@lgrahl

@lgrahl The Streams API provides an example of creating readable streams by wrapping a WebSocket. Yet it appears that no text was required in the WebSockets API to enable this.

So is this something we actually need to change the WebRTC 1.0 document to enable?

The mentioned example for WebSocket has several requirements not applicable to data channels and some problems (which the authors of the streams API couldn't address because the WebSocket API is just how it is):

It requires a reliable and ordered transport and therefore would break partially reliable, unreliable and unordered data channels.
It violates the purpose of a message-oriented protocol and breaks it down to a pure byte stream (well, rather chunk stream) protocol. The point of message-oriented protocols is that no framing mechanism in the upper layer is required. To elaborate: If I use a message-oriented protocol to send a file, I just send that as one message. And then I send another file in another message. But as this has been broken down to a byte stream now, I need to add a framing protocol that recognises the end of the first file before I can send another one.
This is not a byte stream but rather a stream of chunks. Thus, if the sender sends me 64 KiB chunks, the receiver will get 64 KiB chunks. Whereas with byte streams this is up to what either side prefers (and also requires less copying with BYOB readers because the underlying implementation can copy directly into that buffer if the reader is fast enough).
(There's no backpressure signal for the sender. Adding this in braces because the data channel API has events for that.)

(And yeah, I think the API proposed here would also make sense for WebSocket if feasible.)

Edit: I forgot to answer your question.

So is this something we actually need to change the WebRTC 1.0 document to enable?

Yes, the first two points aren't fixable without changes to the spec. (One could argue about the third point.)

Reminding everyone of the -rtcweb-transports draft:

For data transport over the WebRTC data channel
[I-D.ietf-rtcweb-data-channel], WebRTC endpoints MUST support SCTP
over DTLS over ICE. This encapsulation is specified in
[I-D.ietf-tsvwg-sctp-dtls-encaps]. Negotiation of this transport in
SDP is defined in [I-D.ietf-mmusic-sctp-sdp]. The SCTP extension for
NDATA, [I-D.ietf-tsvwg-sctp-ndata], MUST be supported.

FYI I'm planning to create a PR for this but it will take some time. :)

I've also been meaning to put together an extension spec for this, but have been occupied with other matters thus far. Hope to get to it at some point

How this API would be applicable to unreliable DC?

My wild guess is that either when the SCTP transports reach the maxPacketLifeTime or maxRetransmits limits, an exception on the sender will be raised in the writable stream ready/send, and also on the receiver read stream when the abandon SACK is received? Would it be possible to abort\abandon an ongoing message?

Also, I assume that write/recv chunks boundaries are not preservesd(which is fine), right?

Your wild guess is correct. See the class definition for WritableStreamDefaultWriter.

Also, I assume that write/recv chunks boundaries are not preservesd(which is fine), right?

Yep.

Async iteration has landed to the stream specs:
https://streams.spec.whatwg.org/#rs-asynciterator

So it could be possible to write the following code:

const dc = pc.createDataChannel(...);
dc.binaryType = 'stream';
// We only need one small buffer per channel
const buffer = new Uint8Array(myPreferredChunkSize);
let view = new DataView(buffer.buffer, 0);

// Note: The function is declared async!
dc.onmessage = async (event) => {
    for await (const chunk of event.data) {
       view = chunk.value; // We reclaim the buffer here
            // ... write the view's data to disk
    }
};`

Here is a recent proposal along these lines for WebSockets: https://groups.google.com/a/chromium.org/forum/#!topic/blink-dev/X7rWpAkMCyg

Moved to the WebRTC-NV Use Case repo: w3c/webrtc-nv-use-cases#44

feat(transfer): 数据面加流控窗口；协议升 v3 并保留 v2 回退

Navigation Menu

Repository navigation

WHATWG streams for data channel messages #1732

Sending

Receiving

Today's API

Streamed (assuming the file is not a stream)

Streamed (assuming the file is also a stream)

Combination

22 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

WHATWG streams for data channel messages #1732

Description

Sending

Receiving

Activity

martinthomson commented on Jan 11, 2018

martinthomson commented on Jan 11, 2018

lgrahl commented on Jan 11, 2018

domenic commented on Jan 11, 2018

lgrahl commented on Jan 11, 2018

lgrahl commented on Jan 11, 2018

martinthomson commented on Jan 12, 2018

martinthomson commented on Jan 12, 2018

lgrahl commented on Jan 12, 2018

lgrahl commented on Jan 13, 2018

Today's API

Streamed (assuming the file is not a stream)

Streamed (assuming the file is also a stream)

Combination

ricea commented on Jan 17, 2018

aboba commented on Jan 17, 2018

lgrahl commented on Jan 17, 2018

22 remaining items

alvestrand commented on Feb 1, 2018

lgrahl commented on Mar 13, 2018

jan-ivar commented on Oct 9, 2018

murillo128 commented on Jan 28, 2019

lgrahl commented on Jan 28, 2019

murillo128 commented on Feb 6, 2019

aboba commented on Aug 13, 2019

aboba commented on Aug 13, 2019

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions