From adc4a6ceb6d6afbc63bd4756f4a04f1b71f73da8 Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Wed, 29 Oct 2025 11:28:22 +0100 Subject: [PATCH] add udpReadBufferSize parameter (#5129) this allows to set a global UDP read buffer, applied to every UDP socket. --- api/openapi.yaml | 17 ++---------- docs/2-usage/26-decrease-packet-loss.md | 37 ++++--------------------- internal/conf/conf.go | 3 +- internal/conf/path.go | 6 ++-- internal/core/core.go | 18 ++++++++++-- internal/core/path.go | 2 ++ internal/core/path_manager.go | 2 ++ internal/staticsources/handler.go | 20 +++++++------ internal/staticsources/mpegts/source.go | 12 ++++++-- internal/staticsources/rtp/source.go | 12 ++++++-- internal/staticsources/rtsp/source.go | 16 +++++++---- mediamtx.yml | 24 +++------------- 12 files changed, 78 insertions(+), 91 deletions(-) diff --git a/api/openapi.yaml b/api/openapi.yaml index c2b48edd..0428214a 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -77,6 +77,9 @@ components: udpMaxPayloadSize: type: integer format: int64 + udpReadBufferSize: + type: integer + format: int64 runOnConnect: type: string runOnConnectRestart: @@ -225,9 +228,6 @@ components: type: array items: type: string - rtspUDPReadBufferSize: - type: integer - format: int64 # RTMP server rtmp: @@ -393,21 +393,10 @@ components: type: string rtspRangeStart: type: string - rtspUDPReadBufferSize: - type: integer - format: int64 - - # MPEG-TS source - mpegtsUDPReadBufferSize: - type: integer - format: int64 # RTP source rtpSDP: type: string - rtpUDPReadBufferSize: - type: integer - format: int64 # Redirect source sourceRedirect: diff --git a/docs/2-usage/26-decrease-packet-loss.md b/docs/2-usage/26-decrease-packet-loss.md index b3f2d92d..70284e5f 100644 --- a/docs/2-usage/26-decrease-packet-loss.md +++ b/docs/2-usage/26-decrease-packet-loss.md @@ -1,47 +1,20 @@ # Decrease packet loss -MediaMTX is meant for routing live streams, and makes use of a series of protocol which try to minimize latency at cost of losing packets in transmit. In particular, most protocols are built on UDP, which is an "unrealiable transport", specifically picked because it allows to drop late packets in case of network congestions, preserving the real-time aspect of streams. +MediaMTX is meant for routing live streams, and makes use of a series of protocol which try to preserve the real-time aspect of streams and minimize latency at cost of losing packets in transmit. In particular, most protocols are built on UDP, which is an "unrealiable transport", specifically picked because it allows to drop late packets in case of network congestions. These packet losses are usually detected and printed in MediaMTX logs. -If you need to improve the stream reliability and decrease packet losses, the first thing to do is to check whether the network between the MediaMTX instance and the intended publishers and readers has sufficient bandwidth for transmitting the media stream. Most of the times, packet losses are caused by a network which is not fit for this scope. This limitation can be overcome by either recompressing the stream with a lower bitrate, or by changing network. +If you need to improve the stream reliability and decrease packet losses, the first thing to do is to check whether the network between the MediaMTX instance and the intended publishers and readers has sufficient bandwidth for transmitting the media stream. Most of the times, packet losses are caused by a network which is not fit for this scope. This limitation can be overcome by either recompressing the stream with a lower bitrate, or by upgrading the network physical infrastructure (routers, cables, Wi-Fi, firewalls, topology, etc). There are however some parameters that can be tuned to improve the situation, at cost of increasing RAM consumption: -- When publishing a stream with RTSP, MPEG-TS or RTP, packets might get discarded by the server because the UDP read buffer size is too small. Try increasing the UDP read buffer size: +- When publishing a stream with a UDP-based protocol (currently RTSP, MPEG-TS, RTP, SRT, WebRTC), packets might get discarded by the server because the read buffer size of UDP sockets is too small. It can be increased with this parameter: ```yml - rtspUDPReadBufferSize: 1000000 + udpReadBufferSize: 1000000 ``` - If the source of the stream is a camera: - - ```yml - paths: - test: - source: rtsp://.. - rtspUDPReadBufferSize: 1000000 - ``` - - There are similar options for the MPEG-TS and RTP protocol: - - ```yml - paths: - test: - source: udp+mpegts://.. - mpegtsUDPReadBufferSize: 1000000 - ``` - - or: - - ```yml - paths: - test: - source: udp+rtp://.. - rtpUDPReadBufferSize: 1000000 - ``` - - All these options require the `net.core.rmem_max` system parameter to be equal or greater than `rtspUDPReadBufferSize`: + The `udpReadBufferSize` parameter requires the `net.core.rmem_max` system parameter to be equal or greater than it. It can be set with this command: ```sh sudo sysctl net.core.rmem_max=100000000 diff --git a/internal/conf/conf.go b/internal/conf/conf.go index 66565dcf..baa1b02f 100644 --- a/internal/conf/conf.go +++ b/internal/conf/conf.go @@ -159,6 +159,7 @@ type Conf struct { ReadBufferCount *int `json:"readBufferCount,omitempty"` // deprecated WriteQueueSize int `json:"writeQueueSize"` UDPMaxPayloadSize int `json:"udpMaxPayloadSize"` + UDPReadBufferSize uint `json:"udpReadBufferSize"` RunOnConnect string `json:"runOnConnect"` RunOnConnectRestart bool `json:"runOnConnectRestart"` RunOnDisconnect string `json:"runOnDisconnect"` @@ -235,7 +236,7 @@ type Conf struct { RTSPServerCert string `json:"rtspServerCert"` AuthMethods *RTSPAuthMethods `json:"authMethods,omitempty"` // deprecated RTSPAuthMethods RTSPAuthMethods `json:"rtspAuthMethods"` - RTSPUDPReadBufferSize uint `json:"rtspUDPReadBufferSize"` + RTSPUDPReadBufferSize *uint `json:"rtspUDPReadBufferSize,omitempty"` // deprecated // RTMP server RTMP bool `json:"rtmp"` diff --git a/internal/conf/path.go b/internal/conf/path.go index 5d3dd159..fec6c86b 100644 --- a/internal/conf/path.go +++ b/internal/conf/path.go @@ -152,14 +152,14 @@ type Path struct { SourceAnyPortEnable *bool `json:"sourceAnyPortEnable,omitempty"` // deprecated RTSPRangeType RTSPRangeType `json:"rtspRangeType"` RTSPRangeStart string `json:"rtspRangeStart"` - RTSPUDPReadBufferSize uint `json:"rtspUDPReadBufferSize"` + RTSPUDPReadBufferSize *uint `json:"rtspUDPReadBufferSize,omitempty"` // deprecated // MPEG-TS source - MPEGTSUDPReadBufferSize uint `json:"mpegtsUDPReadBufferSize"` + MPEGTSUDPReadBufferSize *uint `json:"mpegtsUDPReadBufferSize,omitempty"` // deprecated // RTP source RTPSDP string `json:"rtpSDP"` - RTPUDPReadBufferSize uint `json:"rtpUDPReadBufferSize"` + RTPUDPReadBufferSize *uint `json:"rtpUDPReadBufferSize,omitempty"` // deprecated // Redirect source SourceRedirect string `json:"sourceRedirect"` diff --git a/internal/core/core.go b/internal/core/core.go index 85f558fd..6cc67c48 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -399,6 +399,7 @@ func (p *Core) createResources(initial bool) error { readTimeout: p.conf.ReadTimeout, writeTimeout: p.conf.WriteTimeout, writeQueueSize: p.conf.WriteQueueSize, + udpReadBufferSize: p.conf.UDPReadBufferSize, rtpMaxPayloadSize: rtpMaxPayloadSize, pathConfs: p.conf.Paths, externalCmdPool: p.externalCmdPool, @@ -415,10 +416,15 @@ func (p *Core) createResources(initial bool) error { _, useUDP := p.conf.RTSPTransports[gortsplib.ProtocolUDP] _, useMulticast := p.conf.RTSPTransports[gortsplib.ProtocolUDPMulticast] + udpReadBufferSize := p.conf.UDPReadBufferSize + if p.conf.RTSPUDPReadBufferSize != nil { + udpReadBufferSize = *p.conf.RTSPUDPReadBufferSize + } + i := &rtsp.Server{ Address: p.conf.RTSPAddress, AuthMethods: p.conf.RTSPAuthMethods, - UDPReadBufferSize: p.conf.RTSPUDPReadBufferSize, + UDPReadBufferSize: udpReadBufferSize, ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, WriteQueueSize: p.conf.WriteQueueSize, @@ -456,10 +462,15 @@ func (p *Core) createResources(initial bool) error { _, useUDP := p.conf.RTSPTransports[gortsplib.ProtocolUDP] _, useMulticast := p.conf.RTSPTransports[gortsplib.ProtocolUDPMulticast] + udpReadBufferSize := p.conf.UDPReadBufferSize + if p.conf.RTSPUDPReadBufferSize != nil { + udpReadBufferSize = *p.conf.RTSPUDPReadBufferSize + } + i := &rtsp.Server{ Address: p.conf.RTSPSAddress, AuthMethods: p.conf.RTSPAuthMethods, - UDPReadBufferSize: p.conf.RTSPUDPReadBufferSize, + UDPReadBufferSize: udpReadBufferSize, ReadTimeout: p.conf.ReadTimeout, WriteTimeout: p.conf.WriteTimeout, WriteQueueSize: p.conf.WriteQueueSize, @@ -749,6 +760,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.WriteQueueSize != p.conf.WriteQueueSize || + newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize || newConf.UDPMaxPayloadSize != p.conf.UDPMaxPayloadSize || newConf.RTSPEncryption != p.conf.RTSPEncryption || closeMetrics || @@ -764,6 +776,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTSPAddress != p.conf.RTSPAddress || !reflect.DeepEqual(newConf.RTSPAuthMethods, p.conf.RTSPAuthMethods) || newConf.RTSPUDPReadBufferSize != p.conf.RTSPUDPReadBufferSize || + newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.WriteQueueSize != p.conf.WriteQueueSize || @@ -787,6 +800,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.RTSPSAddress != p.conf.RTSPSAddress || !reflect.DeepEqual(newConf.RTSPAuthMethods, p.conf.RTSPAuthMethods) || newConf.RTSPUDPReadBufferSize != p.conf.RTSPUDPReadBufferSize || + newConf.UDPReadBufferSize != p.conf.UDPReadBufferSize || newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.WriteQueueSize != p.conf.WriteQueueSize || diff --git a/internal/core/path.go b/internal/core/path.go index ed6e376b..bc07f04a 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -70,6 +70,7 @@ type path struct { readTimeout conf.Duration writeTimeout conf.Duration writeQueueSize int + udpReadBufferSize uint rtpMaxPayloadSize int conf *conf.Path name string @@ -174,6 +175,7 @@ func (pa *path) run() { ReadTimeout: pa.readTimeout, WriteTimeout: pa.writeTimeout, WriteQueueSize: pa.writeQueueSize, + UDPReadBufferSize: pa.udpReadBufferSize, RTPMaxPayloadSize: pa.rtpMaxPayloadSize, Matches: pa.matches, PathManager: pa.parent, diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index 13ae9255..2d246ccd 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -76,6 +76,7 @@ type pathManager struct { readTimeout conf.Duration writeTimeout conf.Duration writeQueueSize int + udpReadBufferSize uint rtpMaxPayloadSize int pathConfs map[string]*conf.Path externalCmdPool *externalcmd.Pool @@ -439,6 +440,7 @@ func (pm *pathManager) createPath( readTimeout: pm.readTimeout, writeTimeout: pm.writeTimeout, writeQueueSize: pm.writeQueueSize, + udpReadBufferSize: pm.udpReadBufferSize, rtpMaxPayloadSize: pm.rtpMaxPayloadSize, conf: pathConf, name: name, diff --git a/internal/staticsources/handler.go b/internal/staticsources/handler.go index ddfe97c8..351ccb97 100644 --- a/internal/staticsources/handler.go +++ b/internal/staticsources/handler.go @@ -67,6 +67,7 @@ type Handler struct { ReadTimeout conf.Duration WriteTimeout conf.Duration WriteQueueSize int + UDPReadBufferSize uint RTPMaxPayloadSize int Matches []string PathManager handlerPathManager @@ -101,10 +102,11 @@ func (s *Handler) Initialize() { strings.HasPrefix(s.Conf.Source, "rtsp+ws://") || strings.HasPrefix(s.Conf.Source, "rtsps+ws://"): s.instance = &ssrtsp.Source{ - ReadTimeout: s.ReadTimeout, - WriteTimeout: s.WriteTimeout, - WriteQueueSize: s.WriteQueueSize, - Parent: s, + ReadTimeout: s.ReadTimeout, + WriteTimeout: s.WriteTimeout, + WriteQueueSize: s.WriteQueueSize, + UDPReadBufferSize: s.UDPReadBufferSize, + Parent: s, } case strings.HasPrefix(s.Conf.Source, "rtmp://") || @@ -126,8 +128,9 @@ func (s *Handler) Initialize() { strings.HasPrefix(s.Conf.Source, "udp+mpegts://") || strings.HasPrefix(s.Conf.Source, "unix+mpegts://"): s.instance = &ssmpegts.Source{ - ReadTimeout: s.ReadTimeout, - Parent: s, + ReadTimeout: s.ReadTimeout, + UDPReadBufferSize: s.UDPReadBufferSize, + Parent: s, } case strings.HasPrefix(s.Conf.Source, "srt://"): @@ -146,8 +149,9 @@ func (s *Handler) Initialize() { case strings.HasPrefix(s.Conf.Source, "udp+rtp://") || strings.HasPrefix(s.Conf.Source, "unix+rtp://"): s.instance = &ssrtp.Source{ - ReadTimeout: s.ReadTimeout, - Parent: s, + ReadTimeout: s.ReadTimeout, + UDPReadBufferSize: s.UDPReadBufferSize, + Parent: s, } case s.Conf.Source == "rpiCamera": diff --git a/internal/staticsources/mpegts/source.go b/internal/staticsources/mpegts/source.go index 95b4995b..4e41d0e1 100644 --- a/internal/staticsources/mpegts/source.go +++ b/internal/staticsources/mpegts/source.go @@ -27,8 +27,9 @@ type parent interface { // Source is a MPEG-TS static source. type Source struct { - ReadTimeout conf.Duration - Parent parent + ReadTimeout conf.Duration + UDPReadBufferSize uint + Parent parent } // Log implements logger.Writer. @@ -55,7 +56,12 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { } default: - nc, err = udp.CreateConn(u, int(params.Conf.MPEGTSUDPReadBufferSize)) + udpReadBufferSize := s.UDPReadBufferSize + if params.Conf.MPEGTSUDPReadBufferSize != nil { + udpReadBufferSize = *params.Conf.MPEGTSUDPReadBufferSize + } + + nc, err = udp.CreateConn(u, int(udpReadBufferSize)) if err != nil { return err } diff --git a/internal/staticsources/rtp/source.go b/internal/staticsources/rtp/source.go index fada03c8..ac23aeb7 100644 --- a/internal/staticsources/rtp/source.go +++ b/internal/staticsources/rtp/source.go @@ -29,8 +29,9 @@ type parent interface { // Source is a RTP static source. type Source struct { - ReadTimeout conf.Duration - Parent parent + ReadTimeout conf.Duration + UDPReadBufferSize uint + Parent parent } // Log implements logger.Writer. @@ -69,7 +70,12 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { } default: - nc, err = udp.CreateConn(u, int(params.Conf.RTPUDPReadBufferSize)) + udpReadBufferSize := s.UDPReadBufferSize + if params.Conf.RTPUDPReadBufferSize != nil { + udpReadBufferSize = *params.Conf.RTPUDPReadBufferSize + } + + nc, err = udp.CreateConn(u, int(udpReadBufferSize)) if err != nil { return err } diff --git a/internal/staticsources/rtsp/source.go b/internal/staticsources/rtsp/source.go index 4e0091a3..edfadbb0 100644 --- a/internal/staticsources/rtsp/source.go +++ b/internal/staticsources/rtsp/source.go @@ -71,10 +71,11 @@ type parent interface { // Source is a RTSP static source. type Source struct { - ReadTimeout conf.Duration - WriteTimeout conf.Duration - WriteQueueSize int - Parent parent + ReadTimeout conf.Duration + WriteTimeout conf.Duration + WriteQueueSize int + UDPReadBufferSize uint + Parent parent } // Log implements logger.Writer. @@ -145,6 +146,11 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { return err } + udpReadBufferSize := s.UDPReadBufferSize + if params.Conf.RTSPUDPReadBufferSize != nil { + udpReadBufferSize = *params.Conf.RTSPUDPReadBufferSize + } + c := &gortsplib.Client{ Scheme: scheme, Host: u.Host, @@ -154,7 +160,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error { ReadTimeout: time.Duration(s.ReadTimeout), WriteTimeout: time.Duration(s.WriteTimeout), WriteQueueSize: s.WriteQueueSize, - UDPReadBufferSize: int(params.Conf.RTSPUDPReadBufferSize), + UDPReadBufferSize: int(udpReadBufferSize), AnyPortEnable: params.Conf.RTSPAnyPort, OnRequest: func(req *base.Request) { s.Log(logger.Debug, "[c->s] %v", req) diff --git a/mediamtx.yml b/mediamtx.yml index 10917031..6bc9d9f0 100644 --- a/mediamtx.yml +++ b/mediamtx.yml @@ -25,6 +25,10 @@ writeQueueSize: 512 # Maximum size of outgoing UDP packets. # This can be decreased to avoid fragmentation on networks with a low UDP MTU. udpMaxPayloadSize: 1472 +# Size of the read buffer of every UDP socket. +# This can be increased to decrease packet losses. +# It defaults to the default value of the operating system. +udpReadBufferSize: 0 # Command to run when a client connects to the server. # This is terminated with SIGINT when a client disconnects from the server. @@ -277,10 +281,6 @@ rtspServerCert: server.crt # Authentication methods. Available are "basic" and "digest". # "digest" doesn't provide any additional security and is available for compatibility only. rtspAuthMethods: [basic] -# Size of the UDP buffer of the RTSP server. -# This can be increased to mitigate packet losses. -# It defaults to the default value of the operating system. -rtspUDPReadBufferSize: 0 ############################################### # Global settings -> RTMP server @@ -538,28 +538,12 @@ pathDefaults: # * npt: duration such as "300ms", "1.5m" or "2h45m", valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h" # * smpte: duration such as "300ms", "1.5m" or "2h45m", valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h" rtspRangeStart: - # Size of the UDP buffer of the RTSP client. - # This can be increased to mitigate packet losses. - # It defaults to the default value of the operating system. - rtspUDPReadBufferSize: 0 - - ############################################### - # Default path settings -> MPEG-TS source (when source is MPEG-TS) - - # Size of the UDP buffer of the MPEG-TS client. - # This can be increased to mitigate packet losses. - # It defaults to the default value of the operating system. - mpegtsUDPReadBufferSize: 0 ############################################### # Default path settings -> RTP source (when source is RTP) # session description protocol (SDP) of the RTP stream. rtpSDP: - # Size of the UDP buffer of the RTP client. - # This can be increased to mitigate packet losses. - # It defaults to the default value of the operating system. - rtpUDPReadBufferSize: 0 ############################################### # Default path settings -> Redirect source (when source is "redirect")