extend udpReadBufferSize to WebRTC UDP sockets (#5130)

this allows to decrease packet losses without touching system parameters.

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/asticode/go-astits v1.14.0
 	github.com/bluenviron/gohlslib/v2 v2.2.3
 	github.com/bluenviron/gortmplib v0.1.1
-	github.com/bluenviron/gortsplib/v5 v5.1.1
+	github.com/bluenviron/gortsplib/v5 v5.1.2-0.20251026101219-59474a8fa3ab
 	github.com/bluenviron/mediacommon/v2 v2.5.1
 	github.com/datarhei/gosrt v0.9.0
 	github.com/fsnotify/fsnotify v1.9.0
@@ -33,6 +33,7 @@ require (
 	github.com/pion/rtcp v1.2.16
 	github.com/pion/rtp v1.8.23
 	github.com/pion/sdp/v3 v3.0.16
+	github.com/pion/transport/v3 v3.0.8
 	github.com/pion/webrtc/v4 v4.1.6
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/crypto v0.43.0
@@ -80,7 +81,6 @@ require (
 	github.com/pion/sctp v1.8.40 // indirect
 	github.com/pion/srtp/v3 v3.0.8 // indirect
 	github.com/pion/stun/v3 v3.0.0 // indirect
-	github.com/pion/transport/v3 v3.0.8 // indirect
 	github.com/pion/turn/v4 v4.1.1 // indirect
 	github.com/pjbgf/sha1cd v0.3.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

go.sum

@@ -37,8 +37,8 @@ github.com/bluenviron/gohlslib/v2 v2.2.3 h1:1R/Jnh1kNR9UB09KAX6xjS2GcdKFRLuPd9wM
 github.com/bluenviron/gohlslib/v2 v2.2.3/go.mod h1:z4Viks+Mdgcl7OcOVJ1fgSmuUwCCJBxYJPLN49n7Vnw=
 github.com/bluenviron/gortmplib v0.1.1 h1:pmR6qfPcJJmE17lWQ/bpuBFZtgGnMrN8KdFj1Gl/ZoQ=
 github.com/bluenviron/gortmplib v0.1.1/go.mod h1:XWy2YzbTP1XEEZ8232OG7I1MSwubsbDRKDNhXGgS2kg=
-github.com/bluenviron/gortsplib/v5 v5.1.1 h1:GC4sfMFBfx7dNARfSS8m8k3hz37TZ9V4hEMus/LPlQ4=
-github.com/bluenviron/gortsplib/v5 v5.1.1/go.mod h1:+4E4JNF7dpDu8LgssZu9fB3Ndh6FNbvGYMKOKR/wvvI=
+github.com/bluenviron/gortsplib/v5 v5.1.2-0.20251026101219-59474a8fa3ab h1:9QH6j4y2FMo299Rz/hX5jrhI+THhWgZ/oSiEmwSPavk=
+github.com/bluenviron/gortsplib/v5 v5.1.2-0.20251026101219-59474a8fa3ab/go.mod h1:+4E4JNF7dpDu8LgssZu9fB3Ndh6FNbvGYMKOKR/wvvI=
 github.com/bluenviron/mediacommon/v2 v2.5.1 h1:qB2fb5c0xyl5OB2gfSfulpEJn7Cdm3vI2n8wjiLMxKI=
 github.com/bluenviron/mediacommon/v2 v2.5.1/go.mod h1:zy1fODPuS/kBd93ftgJS1Jhvjq7LFWfAo32KP7By9AE=
 github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ=

internal/core/core.go

@@ -596,6 +596,7 @@ func (p *Core) createResources(initial bool) error {
 			TrustedProxies:        p.conf.WebRTCTrustedProxies,
 			ReadTimeout:           p.conf.ReadTimeout,
 			WriteTimeout:          p.conf.WriteTimeout,
+			UDPReadBufferSize:     p.conf.UDPReadBufferSize,
 			LocalUDPAddress:       p.conf.WebRTCLocalUDPAddress,
 			LocalTCPAddress:       p.conf.WebRTCLocalTCPAddress,
 			IPsFromInterfaces:     p.conf.WebRTCIPsFromInterfaces,
@@ -877,6 +878,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 		!reflect.DeepEqual(newConf.WebRTCTrustedProxies, p.conf.WebRTCTrustedProxies) ||
 		newConf.ReadTimeout != p.conf.ReadTimeout ||
 		newConf.WriteTimeout != p.conf.WriteTimeout ||
+		newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize ||
 		newConf.WebRTCLocalUDPAddress != p.conf.WebRTCLocalUDPAddress ||
 		newConf.WebRTCLocalTCPAddress != p.conf.WebRTCLocalTCPAddress ||
 		newConf.WebRTCIPsFromInterfaces != p.conf.WebRTCIPsFromInterfaces ||

internal/protocols/udp/udp.go

@@ -150,24 +150,11 @@ func CreateConn(u *url.URL, udpReadBufferSize int) (net.Conn, error) {
 	}
 
 	if udpReadBufferSize != 0 {
-		err = pc.SetReadBuffer(udpReadBufferSize)
+		err = readbuffer.SetReadBuffer(pc, udpReadBufferSize)
 		if err != nil {
 			pc.Close()
 			return nil, err
 		}
-
-		var v int
-		v, err = readbuffer.ReadBuffer(pc)
-		if err != nil {
-			pc.Close()
-			return nil, err
-		}
-
-		if v != udpReadBufferSize {
-			pc.Close()
-			return nil, fmt.Errorf("unable to set read buffer size to %v, check that the operating system allows that",
-				udpReadBufferSize)
-		}
 	}
 
 	return &udpConn{pc: pc, sourceIP: sourceIP}, nil

internal/protocols/webrtc/net.go

@@ -0,0 +1,41 @@
+package webrtc
+
+import (
+	"net"
+
+	"github.com/bluenviron/gortsplib/v5/pkg/readbuffer"
+	"github.com/pion/transport/v3"
+	"github.com/pion/transport/v3/stdnet"
+)
+
+type webrtcNet struct {
+	udpReadBufferSize int
+
+	*stdnet.Net
+}
+
+func (n *webrtcNet) initialize() error {
+	var err error
+	n.Net, err = stdnet.NewNet()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (n *webrtcNet) ListenUDP(network string, laddr *net.UDPAddr) (transport.UDPConn, error) {
+	conn, err := n.Net.ListenUDP(network, laddr)
+	if err != nil {
+		return nil, err
+	}
+
+	if n.udpReadBufferSize != 0 {
+		err = readbuffer.SetReadBuffer(conn.(*net.UDPConn), n.udpReadBufferSize)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return conn, nil
+}

internal/protocols/webrtc/peer_connection.go

@@ -132,6 +132,7 @@ type trackRecvPair struct {
 
 // PeerConnection is a wrapper around webrtc.PeerConnection.
 type PeerConnection struct {
+	UDPReadBufferSize     uint
 	LocalRandomUDP        bool
 	ICEUDPMux             ice.UDPMux
 	ICETCPMux             *TCPMuxWrapper
@@ -194,6 +195,15 @@ func (co *PeerConnection) Start() error {
 
 	settingsEngine.SetSTUNGatherTimeout(time.Duration(co.STUNGatherTimeout))
 
+	webrtcNet := &webrtcNet{
+		udpReadBufferSize: int(co.UDPReadBufferSize),
+	}
+	err := webrtcNet.initialize()
+	if err != nil {
+		return err
+	}
+	settingsEngine.SetNet(webrtcNet)
+
 	mediaEngine := &webrtc.MediaEngine{}
 
 	if co.Publish {
@@ -226,7 +236,7 @@ func (co *PeerConnection) Start() error {
 				codecType = webrtc.RTPCodecTypeAudio
 			}
 
-			err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{
+			err = mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{
 				RTPCodecCapability: tr.Caps,
 				PayloadType:        webrtc.PayloadType(96 + i),
 			}, codecType)
@@ -238,7 +248,7 @@ func (co *PeerConnection) Start() error {
 		// When video is not used, a track must not be added but a codec has to present.
 		// Otherwise audio is muted on Firefox and Chrome.
 		if !videoSetupped {
-			err := mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{
+			err = mediaEngine.RegisterCodec(webrtc.RTPCodecParameters{
 				RTPCodecCapability: webrtc.RTPCodecCapability{
 					MimeType:  webrtc.MimeTypeVP8,
 					ClockRate: 90000,
@@ -251,14 +261,14 @@ func (co *PeerConnection) Start() error {
 		}
 	} else {
 		for _, codec := range incomingVideoCodecs {
-			err := mediaEngine.RegisterCodec(codec, webrtc.RTPCodecTypeVideo)
+			err = mediaEngine.RegisterCodec(codec, webrtc.RTPCodecTypeVideo)
 			if err != nil {
 				return err
 			}
 		}
 
 		for _, codec := range incomingAudioCodecs {
-			err := mediaEngine.RegisterCodec(codec, webrtc.RTPCodecTypeAudio)
+			err = mediaEngine.RegisterCodec(codec, webrtc.RTPCodecTypeAudio)
 			if err != nil {
 				return err
 			}
@@ -267,7 +277,7 @@ func (co *PeerConnection) Start() error {
 
 	interceptorRegistry := &interceptor.Registry{}
 
-	err := registerInterceptors(
+	err = registerInterceptors(
 		mediaEngine,
 		interceptorRegistry,
 		func(s *statsInterceptor) {

internal/protocols/whip/client.go

@@ -26,11 +26,12 @@ const (
 
 // Client is a WHIP client.
 type Client struct {
-	URL            *url.URL
-	Publish        bool
-	OutgoingTracks []*webrtc.OutgoingTrack
-	HTTPClient     *http.Client
-	Log            logger.Writer
+	URL               *url.URL
+	Publish           bool
+	OutgoingTracks    []*webrtc.OutgoingTrack
+	HTTPClient        *http.Client
+	UDPReadBufferSize uint
+	Log               logger.Writer
 
 	pc               *webrtc.PeerConnection
 	patchIsSupported bool
@@ -44,6 +45,7 @@ func (c *Client) Initialize(ctx context.Context) error {
 	}
 
 	c.pc = &webrtc.PeerConnection{
+		UDPReadBufferSize:  c.UDPReadBufferSize,
 		LocalRandomUDP:     true,
 		ICEServers:         iceServers,
 		IPsFromInterfaces:  true,

internal/servers/webrtc/server.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pion/logging"
 	pwebrtc "github.com/pion/webrtc/v4"
 
+	"github.com/bluenviron/gortsplib/v5/pkg/readbuffer"
 	"github.com/bluenviron/mediamtx/internal/conf"
 	"github.com/bluenviron/mediamtx/internal/defs"
 	"github.com/bluenviron/mediamtx/internal/externalcmd"
@@ -193,6 +194,7 @@ type Server struct {
 	TrustedProxies        conf.IPNetworks
 	ReadTimeout           conf.Duration
 	WriteTimeout          conf.Duration
+	UDPReadBufferSize     uint
 	LocalUDPAddress       string
 	LocalTCPAddress       string
 	IPsFromInterfaces     bool
@@ -272,6 +274,17 @@ func (s *Server) Initialize() error {
 			ctxCancel()
 			return err
 		}
+
+		if s.UDPReadBufferSize != 0 {
+			err = readbuffer.SetReadBuffer(s.udpMuxLn.(*net.UDPConn), int(s.UDPReadBufferSize))
+			if err != nil {
+				s.udpMuxLn.Close()
+				s.httpServer.close()
+				ctxCancel()
+				return err
+			}
+		}
+
 		s.iceUDPMux = pwebrtc.NewICEUDPMux(webrtcNilLogger, s.udpMuxLn)
 	}
 
@@ -285,6 +298,7 @@ func (s *Server) Initialize() error {
 			ctxCancel()
 			return err
 		}
+
 		s.iceTCPMux = &webrtc.TCPMuxWrapper{
 			Mux: pwebrtc.NewICETCPMux(webrtcNilLogger, s.tcpMuxLn, 8),
 			Ln:  s.tcpMuxLn,
@@ -336,6 +350,7 @@ outer:
 		select {
 		case req := <-s.chNewSession:
 			sx := &session{
+				udpReadBufferSize:     s.UDPReadBufferSize,
 				parentCtx:             s.ctx,
 				ipsFromInterfaces:     s.IPsFromInterfaces,
 				ipsFromInterfacesList: s.IPsFromInterfacesList,

internal/servers/webrtc/session.go

@@ -41,6 +41,7 @@ type sessionParent interface {
 }
 
 type session struct {
+	udpReadBufferSize     uint
 	parentCtx             context.Context
 	ipsFromInterfaces     bool
 	ipsFromInterfacesList []string
@@ -158,6 +159,7 @@ func (s *session) runPublish() (int, error) {
 	}
 
 	pc := &webrtc.PeerConnection{
+		UDPReadBufferSize:     s.udpReadBufferSize,
 		ICEUDPMux:             s.iceUDPMux,
 		ICETCPMux:             s.iceTCPMux,
 		ICEServers:            iceServers,
@@ -302,6 +304,7 @@ func (s *session) runRead() (int, error) {
 	}
 
 	pc := &webrtc.PeerConnection{
+		UDPReadBufferSize:     s.udpReadBufferSize,
 		ICEUDPMux:             s.iceUDPMux,
 		ICETCPMux:             s.iceTCPMux,
 		ICEServers:            iceServers,

internal/staticsources/handler.go

@@ -142,8 +142,9 @@ func (s *Handler) Initialize() {
 	case strings.HasPrefix(s.Conf.Source, "whep://") ||
 		strings.HasPrefix(s.Conf.Source, "wheps://"):
 		s.instance = &sswebrtc.Source{
-			ReadTimeout: s.ReadTimeout,
-			Parent:      s,
+			ReadTimeout:       s.ReadTimeout,
+			UDPReadBufferSize: s.UDPReadBufferSize,
+			Parent:            s,
 		}
 
 	case strings.HasPrefix(s.Conf.Source, "udp+rtp://") ||

internal/staticsources/webrtc/source.go

@@ -27,8 +27,9 @@ type parent interface {
 
 // Source is a WebRTC static source.
 type Source struct {
-	ReadTimeout conf.Duration
-	Parent      parent
+	ReadTimeout       conf.Duration
+	UDPReadBufferSize uint
+	Parent            parent
 }
 
 // Log implements logger.Writer.
@@ -58,7 +59,8 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 			Timeout:   time.Duration(s.ReadTimeout),
 			Transport: tr,
 		},
-		Log: s,
+		UDPReadBufferSize: s.UDPReadBufferSize,
+		Log:               s,
 	}
 	err = client.Initialize(params.Context)
 	if err != nil {