Repository: libp2p/go-libp2p
Pull request #2745 (2024-03-20 14:40:48)

webrtc: set sctp receive buffer size to 100kB
Based on numbers here: https://github.com/libp2p/go-libp2p/issues/1917#issuecomment-2007681530



Patch:
diff --git a/go.mod b/go.mod
index 1ba2f7ac..3582c5b0 100644
--- a/go.mod
+++ b/go.mod
@@ -50,7 +50,7 @@ require (
 	github.com/pion/webrtc/v3 v3.2.23
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.6.0
-	github.com/quic-go/quic-go v0.41.0
+	github.com/quic-go/quic-go v0.42.0
 	github.com/quic-go/webtransport-go v0.6.0
 	github.com/raulk/go-watchdog v1.3.0
 	github.com/stretchr/testify v1.8.4
diff --git a/go.sum b/go.sum
index 0137afb7..5d1648e6 100644
--- a/go.sum
+++ b/go.sum
@@ -357,8 +357,8 @@ github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qpack v0.4.0/go.mod h1:UZVnYIfi5GRk+zI9UMaCPsmZ2xKJP7XBUvVyT1Knj9A=
-github.com/quic-go/quic-go v0.41.0 h1:aD8MmHfgqTURWNJy48IYFg2OnxwHT3JL7ahGs73lb4k=
-github.com/quic-go/quic-go v0.41.0/go.mod h1:qCkNjqczPEvgsOnxZ0eCD14lv+B2LHlFAB++CNOh9hA=
+github.com/quic-go/quic-go v0.42.0 h1:uSfdap0eveIl8KXnipv9K7nlwZ5IqLlYOpJ58u5utpM=
+github.com/quic-go/quic-go v0.42.0/go.mod h1:132kz4kL3F9vxhW3CtQJLDVwcFe5wdWeJXXijhsO57M=
 github.com/quic-go/webtransport-go v0.6.0 h1:CvNsKqc4W2HljHJnoT+rMmbRJybShZ0YPFDD3NxaZLY=
 github.com/quic-go/webtransport-go v0.6.0/go.mod h1:9KjU4AEBqEQidGHNDkZrb8CAa1abRaosM2yGOyiikEc=
 github.com/raulk/go-watchdog v1.3.0 h1:oUmdlHxdkXRJlwfG0O9omj8ukerm8MEQavSiDTEtBsk=
@@ -606,6 +606,8 @@ golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181030000716-a0a13e073c7b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/p2p/host/basic/basic_host.go b/p2p/host/basic/basic_host.go
index 8e6e8efe..c148b528 100644
--- a/p2p/host/basic/basic_host.go
+++ b/p2p/host/basic/basic_host.go
@@ -414,7 +414,7 @@ func (h *BasicHost) newStreamHandler(s network.Stream) {
 			}
 			logf("protocol EOF: %s (took %s)", s.Conn().RemotePeer(), took)
 		} else {
-			log.Debugf("protocol mux failed: %s (took %s)", err, took)
+			log.Debugf("protocol mux failed: %s (took %s, id:%s, remote peer:%s, remote addr:%v)", err, took, s.ID(), s.Conn().RemotePeer(), s.Conn().RemoteMultiaddr())
 		}
 		s.Reset()
 		return
diff --git a/p2p/transport/quicreuse/config.go b/p2p/transport/quicreuse/config.go
index 45b1d4b8..92c881ef 100644
--- a/p2p/transport/quicreuse/config.go
+++ b/p2p/transport/quicreuse/config.go
@@ -1,7 +1,6 @@
 package quicreuse
 
 import (
-	"net"
 	"time"
 
 	"github.com/quic-go/quic-go"
@@ -12,12 +11,8 @@ var quicConfig = &quic.Config{
 	MaxIncomingUniStreams:      5,              // allow some unidirectional streams, in case we speak WebTransport
 	MaxStreamReceiveWindow:     10 * (1 << 20), // 10 MB
 	MaxConnectionReceiveWindow: 15 * (1 << 20), // 15 MB
-	RequireAddressValidation: func(net.Addr) bool {
-		// TODO(#1535): require source address validation when under load
-		return false
-	},
-	KeepAlivePeriod: 15 * time.Second,
-	Versions:        []quic.VersionNumber{quic.Version1},
+	KeepAlivePeriod:            15 * time.Second,
+	Versions:                   []quic.VersionNumber{quic.Version1},
 	// We don't use datagrams (yet), but this is necessary for WebTransport
 	EnableDatagrams: true,
 }
diff --git a/p2p/transport/quicreuse/connmgr.go b/p2p/transport/quicreuse/connmgr.go
index 0add5ddf..ec2b7b28 100644
--- a/p2p/transport/quicreuse/connmgr.go
+++ b/p2p/transport/quicreuse/connmgr.go
@@ -222,3 +222,7 @@ func (c *ConnManager) Close() error {
 	}
 	return c.reuseUDP4.Close()
 }
+
+func (c *ConnManager) ClientConfig() *quic.Config {
+	return c.clientConfig
+}
diff --git a/p2p/transport/webrtc/listener.go b/p2p/transport/webrtc/listener.go
index cbf9f8fb..1834fc81 100644
--- a/p2p/transport/webrtc/listener.go
+++ b/p2p/transport/webrtc/listener.go
@@ -209,6 +209,10 @@ func (l *listener) setupConnection(
 	// in a release.
 	settingEngine.SetReceiveMTU(udpmux.ReceiveBufSize)
 	settingEngine.DetachDataChannels()
+	settingEngine.SetSCTPMaxReceiveBufferSize(sctpReceiveBufferSize)
+	if err := scope.ReserveMemory(sctpReceiveBufferSize, network.ReservationPriorityMedium); err != nil {
+		return nil, err
+	}
 
 	w, err = newWebRTCConnection(settingEngine, l.config)
 	if err != nil {
diff --git a/p2p/transport/webrtc/stream.go b/p2p/transport/webrtc/stream.go
index 135940e4..733c2347 100644
--- a/p2p/transport/webrtc/stream.go
+++ b/p2p/transport/webrtc/stream.go
@@ -17,24 +17,19 @@ import (
 const (
 	// maxMessageSize is the maximum message size of the Protobuf message we send / receive.
 	maxMessageSize = 16384
-	// Pion SCTP association has an internal receive buffer of 1MB (roughly, 1MB per connection).
-	// We can change this value in the SettingEngine before creating the peerconnection.
-	// https://github.com/pion/webrtc/blob/v3.1.49/sctptransport.go#L341
-	maxBufferedAmount = 2 * maxMessageSize
+	// maxSendBuffer is the maximum data we enqueue on the underlying data channel for writes.
+	// The underlying SCTP layer has an unbounded buffer for writes. We limit the amount enqueued
+	// per stream is limited to avoid a single stream monopolizing the entire connection.
+	maxSendBuffer = 2 * maxMessageSize
+	// sendBufferLowThreshold is the threshold below which we write more data on the underlying
+	// data channel. We want a notification as soon as we can write 1 full sized message.
+	sendBufferLowThreshold = maxSendBuffer - maxMessageSize
 	// maxTotalControlMessagesSize is the maximum total size of all control messages we will
 	// write on this stream.
 	// 4 control messages of size 10 bytes + 10 bytes buffer. This number doesn't need to be
 	// exact. In the worst case, we enqueue these many bytes more in the webrtc peer connection
 	// send queue.
 	maxTotalControlMessagesSize = 50
-	// bufferedAmountLowThreshold and maxBufferedAmount are bound
-	// to a stream but congestion control is done on the whole
-	// SCTP association. This means that a single stream can monopolize
-	// the complete congestion control window (cwnd) if it does not
-	// read stream data and it's remote continues to send. We can
-	// add messages to the send buffer once there is space for 1 full
-	// sized message.
-	bufferedAmountLowThreshold = maxBufferedAmount / 2
 
 	// Proto overhead assumption is 5 bytes
 	protoOverhead = 5
@@ -120,7 +115,7 @@ func newStream(
 	}
 	// released when the controlMessageReader goroutine exits
 	s.controlMessageReaderDone.Add(1)
-	s.dataChannel.SetBufferedAmountLowThreshold(bufferedAmountLowThreshold)
+	s.dataChannel.SetBufferedAmountLowThreshold(sendBufferLowThreshold)
 	s.dataChannel.OnBufferedAmountLow(func() {
 		s.notifyWriteStateChanged()
 
diff --git a/p2p/transport/webrtc/stream_write.go b/p2p/transport/webrtc/stream_write.go
index 82d4ac28..0510c96c 100644
--- a/p2p/transport/webrtc/stream_write.go
+++ b/p2p/transport/webrtc/stream_write.go
@@ -112,9 +112,9 @@ func (s *stream) SetWriteDeadline(t time.Time) error {
 
 func (s *stream) availableSendSpace() int {
 	buffered := int(s.dataChannel.BufferedAmount())
-	availableSpace := maxBufferedAmount - buffered
+	availableSpace := maxSendBuffer - buffered
 	if availableSpace+maxTotalControlMessagesSize < 0 { // this should never happen, but better check
-		log.Errorw("data channel buffered more data than the maximum amount", "max", maxBufferedAmount, "buffered", buffered)
+		log.Errorw("data channel buffered more data than the maximum amount", "max", maxSendBuffer, "buffered", buffered)
 	}
 	return availableSpace
 }
diff --git a/p2p/transport/webrtc/transport.go b/p2p/transport/webrtc/transport.go
index 20bfdf98..2309ac4d 100644
--- a/p2p/transport/webrtc/transport.go
+++ b/p2p/transport/webrtc/transport.go
@@ -79,6 +79,8 @@ const (
 	DefaultDisconnectedTimeout = 20 * time.Second
 	DefaultFailedTimeout       = 30 * time.Second
 	DefaultKeepaliveTimeout    = 15 * time.Second
+
+	sctpReceiveBufferSize = 100_000
 )
 
 type WebRTCTransport struct {
@@ -314,6 +316,10 @@ func (t *WebRTCTransport) dial(ctx context.Context, scope network.ConnManagement
 	// If you run pion on a system with only the loopback interface UP,
 	// it will not connect to anything.
 	settingEngine.SetIncludeLoopbackCandidate(true)
+	settingEngine.SetSCTPMaxReceiveBufferSize(sctpReceiveBufferSize)
+	if err := scope.ReserveMemory(sctpReceiveBufferSize, network.ReservationPriorityMedium); err != nil {
+		return nil, err
+	}
 
 	w, err = newWebRTCConnection(settingEngine, t.webrtcConfig)
 	if err != nil {
diff --git a/p2p/transport/webtransport/transport.go b/p2p/transport/webtransport/transport.go
index 957d2ff1..26c87434 100644
--- a/p2p/transport/webtransport/transport.go
+++ b/p2p/transport/webtransport/transport.go
@@ -207,6 +207,7 @@ func (t *transport) dial(ctx context.Context, addr ma.Multiaddr, url, sni string
 			Dial: func(ctx context.Context, addr string, tlsCfg *tls.Config, cfg *quic.Config) (quic.EarlyConnection, error) {
 				return conn.(quic.EarlyConnection), nil
 			},
+			QuicConfig: t.connManager.ClientConfig().Clone(),
 		},
 	}
 	rsp, sess, err := dialer.Dial(ctx, url, nil)


Test patch:
diff --git a/p2p/test/transport/rcmgr_test.go b/p2p/test/transport/rcmgr_test.go
index 20f34de7..9a58a344 100644
--- a/p2p/test/transport/rcmgr_test.go
+++ b/p2p/test/transport/rcmgr_test.go
@@ -86,6 +86,10 @@ func TestResourceManagerIsUsed(t *testing.T) {
 						}
 						return nil
 					})
+					if tc.Name == "WebRTC" {
+						// webrtc receive buffer is a fix sized buffer allocated up front
+						connScope.EXPECT().ReserveMemory(gomock.Any(), gomock.Any())
+					}
 					connScope.EXPECT().Done().MinTimes(1)
 
 					var allStreamsDone sync.WaitGroup
diff --git a/test-plans/go.mod b/test-plans/go.mod
index 267a4134..c21cf255 100644
--- a/test-plans/go.mod
+++ b/test-plans/go.mod
@@ -87,7 +87,7 @@ require (
 	github.com/prometheus/common v0.47.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/quic-go/qpack v0.4.0 // indirect
-	github.com/quic-go/quic-go v0.41.0 // indirect
+	github.com/quic-go/quic-go v0.42.0 // indirect
 	github.com/quic-go/webtransport-go v0.6.0 // indirect
 	github.com/raulk/go-watchdog v1.3.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
diff --git a/test-plans/go.sum b/test-plans/go.sum
index 5ee58c67..c31561eb 100644
--- a/test-plans/go.sum
+++ b/test-plans/go.sum
@@ -309,8 +309,8 @@ github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qpack v0.4.0/go.mod h1:UZVnYIfi5GRk+zI9UMaCPsmZ2xKJP7XBUvVyT1Knj9A=
-github.com/quic-go/quic-go v0.41.0 h1:aD8MmHfgqTURWNJy48IYFg2OnxwHT3JL7ahGs73lb4k=
-github.com/quic-go/quic-go v0.41.0/go.mod h1:qCkNjqczPEvgsOnxZ0eCD14lv+B2LHlFAB++CNOh9hA=
+github.com/quic-go/quic-go v0.42.0 h1:uSfdap0eveIl8KXnipv9K7nlwZ5IqLlYOpJ58u5utpM=
+github.com/quic-go/quic-go v0.42.0/go.mod h1:132kz4kL3F9vxhW3CtQJLDVwcFe5wdWeJXXijhsO57M=
 github.com/quic-go/webtransport-go v0.6.0 h1:CvNsKqc4W2HljHJnoT+rMmbRJybShZ0YPFDD3NxaZLY=
 github.com/quic-go/webtransport-go v0.6.0/go.mod h1:9KjU4AEBqEQidGHNDkZrb8CAa1abRaosM2yGOyiikEc=
 github.com/raulk/go-watchdog v1.3.0 h1:oUmdlHxdkXRJlwfG0O9omj8ukerm8MEQavSiDTEtBsk=
@@ -534,6 +534,8 @@ golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
+golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181030000716-a0a13e073c7b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=