WebRTC Direct transport implementation

[marten-seemann]

No description provided.

[justin0mcateer]

We are interested in using this. Is this just waiting to be reviewed and merged, or is there more work to do?

[marten-seemann]

@justin0mcateer It’s not working 100% correctly yet. As you can see, all the tests are failing. We expect to make some progress in the next couple of weeks.

Can you tell us some more about your use case?

[filefilegoadmin]

@marten-seemann Is this PR going to be merged for the Aug 28th release?
We plan to experiment video streaming with libp2p

[marten-seemann]

@marten-seemann Is this PR going to be merged for the Aug 28th release? We plan to experiment video streaming with libp2p

Hopefully, if we get it to work properly by that date. Are you planning to use the WebRTC transport?

[filefilegoadmin]

@marten-seemann Is this PR going to be merged for the Aug 28th release? We plan to experiment video streaming with libp2p

Hopefully, if we get it to work properly by that date. Are you planning to use the WebRTC transport?

Currently using tcp for transport. We would like to experiment with a feature to play videos using webrtc.

[marten-seemann]

There's a flaky deadline test: https://github.com/libp2p/go-libp2p/actions/runs/5982859079/job/16232739645?pr=2337

[MarcoPolo]

I've debugged the interop issue. The problem is we close the stream right after sending the echo back. In Pion (maybe an SCTP thing?) closing a channel resets the stream. So the write isn't acknowledged by the peer.

We can fix this by waiting. Here I'm waiting for the buffered amount to be 0. There might be a better way here.

diff --git a/p2p/transport/webrtc/stream.go b/p2p/transport/webrtc/stream.go
index b86128dc..24c69a0a 100644
--- a/p2p/transport/webrtc/stream.go
+++ b/p2p/transport/webrtc/stream.go
@@ -192,6 +192,10 @@ func (s *stream) maybeDeclareStreamDone() {
 		(s.receiveState == receiveStateReset || s.receiveState == receiveStateDataRead) &&
 		len(s.controlMsgQueue) == 0 {
 		_ = s.SetReadDeadline(time.Now().Add(-1 * time.Hour)) // pion ignores zero times
+		s.dataChannel.SetBufferedAmountLowThreshold(0)
+		ch := make(chan struct{})
+		s.dataChannel.OnBufferedAmountLow(func() { ch <- struct{}{} })
+		<-ch
 		_ = s.dataChannel.Close()
 		// TODO: write for the spawned reader to return
 		s.onDone()

[MarcoPolo]

Nice! let's get this in!

[marten-seemann]

I've debugged the interop issue. The problem is we close the stream right after sending the echo back. In Pion (maybe an SCTP thing?) closing a channel resets the stream. So the write isn't acknowledged by the peer.

We can fix this by waiting. Here I'm waiting for the buffered amount to be 0. There might be a better way here.

diff --git a/p2p/transport/webrtc/stream.go b/p2p/transport/webrtc/stream.go
index b86128dc..24c69a0a 100644
--- a/p2p/transport/webrtc/stream.go
+++ b/p2p/transport/webrtc/stream.go
@@ -192,6 +192,10 @@ func (s *stream) maybeDeclareStreamDone() {
 		(s.receiveState == receiveStateReset || s.receiveState == receiveStateDataRead) &&
 		len(s.controlMsgQueue) == 0 {
 		_ = s.SetReadDeadline(time.Now().Add(-1 * time.Hour)) // pion ignores zero times
+		s.dataChannel.SetBufferedAmountLowThreshold(0)
+		ch := make(chan struct{})
+		s.dataChannel.OnBufferedAmountLow(func() { ch <- struct{}{} })
+		<-ch
 		_ = s.dataChannel.Close()
 		// TODO: write for the spawned reader to return
 		s.onDone()

This fix doesn't work. First of all, it's racy: SetBufferedAmountLowThreshold and OnBufferedAmountLow would need to happen atomically. I don't think this is the problem though, the transport tests still fail when I try to work around this:

@@ -192,6 +193,16 @@ func (s *stream) maybeDeclareStreamDone() {
                (s.receiveState == receiveStateReset || s.receiveState == receiveStateDataRead) &&
                len(s.controlMsgQueue) == 0 {
                _ = s.SetReadDeadline(time.Now().Add(-1 * time.Hour)) // pion ignores zero times
+               s.dataChannel.SetBufferedAmountLowThreshold(0)
+               ch := make(chan struct{}, 1)
+               s.dataChannel.OnBufferedAmountLow(func() { ch <- struct{}{} })
+               if s.dataChannel.BufferedAmount() > 0 {
+                       select {
+                       case <-ch:
+                       case <-time.After(5 * time.Second):
+                               fmt.Println("sending close took too long", s.dataChannel.BufferedAmount())
+                       }
+               }
                _ = s.dataChannel.Close()
                // TODO: write for the spawned reader to return
                s.onDone()

Lots of test cases run into the 5s timeout, with a non-zero buffered amount.

[BigLep]

It's been a couple of weeks since saw activity here. Is the interop issue what we're working on here?

[marten-seemann]

@BigLep libp2p/test-plans#279 is what has slowed us down for ~2 weeks. It's resolved now. I'll update this issue with a status update tomorrow.

[marten-seemann]

I wrote up libp2p/specs#575 explaining the problem, and implemented the workaround described there. This should allow us to merge this PR and get WebRTC support out. It's limited to 512 streams per WebRTC connection, which is pretty terrible, but probably enough for most use cases.

[marten-seemann]

CI is happy with this PR. I am not really happy, but we need to get this out somehow. Merging.