Browse Source

rename readBufferCount into writeQueueSize (#2248)

pull/2249/head
Alessandro Ros 2 years ago committed by GitHub
parent
commit
bf8e69ea89
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 4
      README.md
  2. 2
      apidocs/openapi.yaml
  3. 12
      internal/conf/conf.go
  4. 6
      internal/conf/conf_test.go
  5. 32
      internal/core/core.go
  6. 8
      internal/core/hls_manager.go
  7. 8
      internal/core/hls_muxer.go
  8. 8
      internal/core/path.go
  9. 8
      internal/core/path_manager.go
  10. 8
      internal/core/rtmp_conn.go
  11. 8
      internal/core/rtmp_server.go
  12. 6
      internal/core/rtsp_server.go
  13. 31
      internal/core/rtsp_source.go
  14. 4
      internal/core/source_static.go
  15. 8
      internal/core/srt_conn.go
  16. 8
      internal/core/srt_server.go
  17. 20
      internal/core/webrtc_manager.go
  18. 18
      internal/core/webrtc_session.go
  19. 4
      mediamtx.yml

4
README.md

@ -1276,10 +1276,10 @@ rtsps://localhost:8322/mystream @@ -1276,10 +1276,10 @@ rtsps://localhost:8322/mystream
In some scenarios, when publishing or reading from the server with RTSP, frames can get corrupted. This can be caused by multiple reasons:
* the packet buffer of the server is too small and can't keep up with the stream throughput. A solution consists in increasing its size:
* the write queue of the server is too small and can't keep up with the stream throughput. A solution consists in increasing its size:
```yml
readBufferCount: 1024
writeQueueSize: 1024
```
* The stream throughput is too big and the stream can't be transmitted correctly with the UDP transport protocol. UDP is more performant, faster and more efficient than TCP, but doesn't have a retransmission mechanism, that is needed in case of streams that need a large bandwidth. A solution consists in switching to TCP:

2
apidocs/openapi.yaml

@ -31,7 +31,7 @@ components: @@ -31,7 +31,7 @@ components:
type: string
writeTimeout:
type: string
readBufferCount:
writeQueueSize:
type: integer
udpMaxPayloadSize:
type: integer

12
internal/conf/conf.go

@ -95,7 +95,8 @@ type Conf struct { @@ -95,7 +95,8 @@ type Conf struct {
LogFile string `json:"logFile"`
ReadTimeout StringDuration `json:"readTimeout"`
WriteTimeout StringDuration `json:"writeTimeout"`
ReadBufferCount int `json:"readBufferCount"`
ReadBufferCount int `json:"readBufferCount"` // deprecated
WriteQueueSize int `json:"writeQueueSize"`
UDPMaxPayloadSize int `json:"udpMaxPayloadSize"`
ExternalAuthenticationURL string `json:"externalAuthenticationURL"`
API bool `json:"api"`
@ -218,8 +219,11 @@ func (conf Conf) Clone() *Conf { @@ -218,8 +219,11 @@ func (conf Conf) Clone() *Conf {
// Check checks the configuration for errors.
func (conf *Conf) Check() error {
// general
if (conf.ReadBufferCount & (conf.ReadBufferCount - 1)) != 0 {
return fmt.Errorf("'readBufferCount' must be a power of two")
if conf.ReadBufferCount != 0 {
conf.WriteQueueSize = conf.ReadBufferCount
}
if (conf.WriteQueueSize & (conf.WriteQueueSize - 1)) != 0 {
return fmt.Errorf("'writeQueueSize' must be a power of two")
}
if conf.UDPMaxPayloadSize > 1472 {
return fmt.Errorf("'udpMaxPayloadSize' must be less than 1472")
@ -317,7 +321,7 @@ func (conf *Conf) UnmarshalJSON(b []byte) error { @@ -317,7 +321,7 @@ func (conf *Conf) UnmarshalJSON(b []byte) error {
conf.LogFile = "mediamtx.log"
conf.ReadTimeout = 10 * StringDuration(time.Second)
conf.WriteTimeout = 10 * StringDuration(time.Second)
conf.ReadBufferCount = 512
conf.WriteQueueSize = 512
conf.UDPMaxPayloadSize = 1472
conf.APIAddress = "127.0.0.1:9997"
conf.MetricsAddress = "127.0.0.1:9998"

6
internal/conf/conf_test.go

@ -217,9 +217,9 @@ func TestConfErrors(t *testing.T) { @@ -217,9 +217,9 @@ func TestConfErrors(t *testing.T) {
"json: unknown field \"invalid\"",
},
{
"invalid readBufferCount",
"readBufferCount: 1001\n",
"'readBufferCount' must be a power of two",
"invalid writeQueueSize",
"writeQueueSize: 1001\n",
"'writeQueueSize' must be a power of two",
},
{
"invalid udpMaxPayloadSize",

32
internal/core/core.go

@ -246,7 +246,7 @@ func (p *Core) createResources(initial bool) error { @@ -246,7 +246,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.AuthMethods,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
p.conf.UDPMaxPayloadSize,
p.conf.Paths,
p.externalCmdPool,
@ -266,7 +266,7 @@ func (p *Core) createResources(initial bool) error { @@ -266,7 +266,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.AuthMethods,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
useUDP,
useMulticast,
p.conf.RTPAddress,
@ -301,7 +301,7 @@ func (p *Core) createResources(initial bool) error { @@ -301,7 +301,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.AuthMethods,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
false,
false,
"",
@ -335,7 +335,7 @@ func (p *Core) createResources(initial bool) error { @@ -335,7 +335,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.RTMPAddress,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
false,
"",
"",
@ -361,7 +361,7 @@ func (p *Core) createResources(initial bool) error { @@ -361,7 +361,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.RTMPSAddress,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
true,
p.conf.RTMPServerCert,
p.conf.RTMPServerKey,
@ -397,7 +397,7 @@ func (p *Core) createResources(initial bool) error { @@ -397,7 +397,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.HLSTrustedProxies,
p.conf.HLSDirectory,
p.conf.ReadTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
p.pathManager,
p.metrics,
p,
@ -419,7 +419,7 @@ func (p *Core) createResources(initial bool) error { @@ -419,7 +419,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.WebRTCTrustedProxies,
p.conf.WebRTCICEServers2,
p.conf.ReadTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
p.conf.WebRTCICEHostNAT1To1IPs,
p.conf.WebRTCICEUDPMuxAddress,
p.conf.WebRTCICETCPMuxAddress,
@ -439,7 +439,7 @@ func (p *Core) createResources(initial bool) error { @@ -439,7 +439,7 @@ func (p *Core) createResources(initial bool) error {
p.conf.SRTAddress,
p.conf.ReadTimeout,
p.conf.WriteTimeout,
p.conf.ReadBufferCount,
p.conf.WriteQueueSize,
p.conf.UDPMaxPayloadSize,
p.externalCmdPool,
p.pathManager,
@ -504,7 +504,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -504,7 +504,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
!reflect.DeepEqual(newConf.AuthMethods, p.conf.AuthMethods) ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.UDPMaxPayloadSize != p.conf.UDPMaxPayloadSize ||
closeMetrics
if !closePathManager && !reflect.DeepEqual(newConf.Paths, p.conf.Paths) {
@ -518,7 +518,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -518,7 +518,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
!reflect.DeepEqual(newConf.AuthMethods, p.conf.AuthMethods) ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
!reflect.DeepEqual(newConf.Protocols, p.conf.Protocols) ||
newConf.RTPAddress != p.conf.RTPAddress ||
newConf.RTCPAddress != p.conf.RTCPAddress ||
@ -539,7 +539,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -539,7 +539,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
!reflect.DeepEqual(newConf.AuthMethods, p.conf.AuthMethods) ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.ServerCert != p.conf.ServerCert ||
newConf.ServerKey != p.conf.ServerKey ||
newConf.RTSPAddress != p.conf.RTSPAddress ||
@ -555,7 +555,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -555,7 +555,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.RTMPAddress != p.conf.RTMPAddress ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.RTSPAddress != p.conf.RTSPAddress ||
newConf.RunOnConnect != p.conf.RunOnConnect ||
newConf.RunOnConnectRestart != p.conf.RunOnConnectRestart ||
@ -568,7 +568,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -568,7 +568,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.RTMPSAddress != p.conf.RTMPSAddress ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.RTMPServerCert != p.conf.RTMPServerCert ||
newConf.RTMPServerKey != p.conf.RTMPServerKey ||
newConf.RTSPAddress != p.conf.RTSPAddress ||
@ -594,7 +594,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -594,7 +594,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
!reflect.DeepEqual(newConf.HLSTrustedProxies, p.conf.HLSTrustedProxies) ||
newConf.HLSDirectory != p.conf.HLSDirectory ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
closePathManager ||
closeMetrics
@ -608,7 +608,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -608,7 +608,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
!reflect.DeepEqual(newConf.WebRTCTrustedProxies, p.conf.WebRTCTrustedProxies) ||
!reflect.DeepEqual(newConf.WebRTCICEServers2, p.conf.WebRTCICEServers2) ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
!reflect.DeepEqual(newConf.WebRTCICEHostNAT1To1IPs, p.conf.WebRTCICEHostNAT1To1IPs) ||
newConf.WebRTCICEUDPMuxAddress != p.conf.WebRTCICEUDPMuxAddress ||
newConf.WebRTCICETCPMuxAddress != p.conf.WebRTCICETCPMuxAddress ||
@ -620,7 +620,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { @@ -620,7 +620,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
newConf.SRTAddress != p.conf.SRTAddress ||
newConf.ReadTimeout != p.conf.ReadTimeout ||
newConf.WriteTimeout != p.conf.WriteTimeout ||
newConf.ReadBufferCount != p.conf.ReadBufferCount ||
newConf.WriteQueueSize != p.conf.WriteQueueSize ||
newConf.UDPMaxPayloadSize != p.conf.UDPMaxPayloadSize ||
closePathManager

8
internal/core/hls_manager.go

@ -42,7 +42,7 @@ type hlsManager struct { @@ -42,7 +42,7 @@ type hlsManager struct {
partDuration conf.StringDuration
segmentMaxSize conf.StringSize
directory string
readBufferCount int
writeQueueSize int
pathManager *pathManager
metrics *metrics
parent hlsManagerParent
@ -78,7 +78,7 @@ func newHLSManager( @@ -78,7 +78,7 @@ func newHLSManager(
trustedProxies conf.IPsOrCIDRs,
directory string,
readTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
pathManager *pathManager,
metrics *metrics,
parent hlsManagerParent,
@ -94,7 +94,7 @@ func newHLSManager( @@ -94,7 +94,7 @@ func newHLSManager(
partDuration: partDuration,
segmentMaxSize: segmentMaxSize,
directory: directory,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
pathManager: pathManager,
parent: parent,
metrics: metrics,
@ -241,7 +241,7 @@ func (m *hlsManager) createMuxer(pathName string, remoteAddr string) *hlsMuxer { @@ -241,7 +241,7 @@ func (m *hlsManager) createMuxer(pathName string, remoteAddr string) *hlsMuxer {
m.partDuration,
m.segmentMaxSize,
m.directory,
m.readBufferCount,
m.writeQueueSize,
&m.wg,
pathName,
m.pathManager,

8
internal/core/hls_muxer.go

@ -66,7 +66,7 @@ type hlsMuxer struct { @@ -66,7 +66,7 @@ type hlsMuxer struct {
partDuration conf.StringDuration
segmentMaxSize conf.StringSize
directory string
readBufferCount int
writeQueueSize int
wg *sync.WaitGroup
pathName string
pathManager *pathManager
@ -96,7 +96,7 @@ func newHLSMuxer( @@ -96,7 +96,7 @@ func newHLSMuxer(
partDuration conf.StringDuration,
segmentMaxSize conf.StringSize,
directory string,
readBufferCount int,
writeQueueSize int,
wg *sync.WaitGroup,
pathName string,
pathManager *pathManager,
@ -113,7 +113,7 @@ func newHLSMuxer( @@ -113,7 +113,7 @@ func newHLSMuxer(
partDuration: partDuration,
segmentMaxSize: segmentMaxSize,
directory: directory,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
wg: wg,
pathName: pathName,
pathManager: pathManager,
@ -254,7 +254,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{}) @@ -254,7 +254,7 @@ func (m *hlsMuxer) runInner(innerCtx context.Context, innerReady chan struct{})
defer m.path.removeReader(pathRemoveReaderReq{author: m})
m.ringBuffer, _ = ringbuffer.New(uint64(m.readBufferCount))
m.ringBuffer, _ = ringbuffer.New(uint64(m.writeQueueSize))
var medias media.Medias

8
internal/core/path.go

@ -174,7 +174,7 @@ type path struct { @@ -174,7 +174,7 @@ type path struct {
rtspAddress string
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
udpMaxPayloadSize int
confName string
conf *conf.PathConf
@ -225,7 +225,7 @@ func newPath( @@ -225,7 +225,7 @@ func newPath(
rtspAddress string,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
udpMaxPayloadSize int,
confName string,
cnf *conf.PathConf,
@ -241,7 +241,7 @@ func newPath( @@ -241,7 +241,7 @@ func newPath(
rtspAddress: rtspAddress,
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
confName: confName,
conf: cnf,
@ -310,7 +310,7 @@ func (pa *path) run() { @@ -310,7 +310,7 @@ func (pa *path) run() {
pa.conf,
pa.readTimeout,
pa.writeTimeout,
pa.readBufferCount,
pa.writeQueueSize,
pa)
if !pa.conf.SourceOnDemand {

8
internal/core/path_manager.go

@ -69,7 +69,7 @@ type pathManager struct { @@ -69,7 +69,7 @@ type pathManager struct {
authMethods conf.AuthMethods
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
udpMaxPayloadSize int
pathConfs map[string]*conf.PathConf
externalCmdPool *externalcmd.Pool
@ -103,7 +103,7 @@ func newPathManager( @@ -103,7 +103,7 @@ func newPathManager(
authMethods conf.AuthMethods,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
udpMaxPayloadSize int,
pathConfs map[string]*conf.PathConf,
externalCmdPool *externalcmd.Pool,
@ -118,7 +118,7 @@ func newPathManager( @@ -118,7 +118,7 @@ func newPathManager(
authMethods: authMethods,
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
pathConfs: pathConfs,
externalCmdPool: externalCmdPool,
@ -352,7 +352,7 @@ func (pm *pathManager) createPath( @@ -352,7 +352,7 @@ func (pm *pathManager) createPath(
pm.rtspAddress,
pm.readTimeout,
pm.writeTimeout,
pm.readBufferCount,
pm.writeQueueSize,
pm.udpMaxPayloadSize,
pathConfName,
pathConf,

8
internal/core/rtmp_conn.go

@ -60,7 +60,7 @@ type rtmpConn struct { @@ -60,7 +60,7 @@ type rtmpConn struct {
rtspAddress string
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
runOnConnect string
runOnConnectRestart bool
wg *sync.WaitGroup
@ -85,7 +85,7 @@ func newRTMPConn( @@ -85,7 +85,7 @@ func newRTMPConn(
rtspAddress string,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
runOnConnect string,
runOnConnectRestart bool,
wg *sync.WaitGroup,
@ -101,7 +101,7 @@ func newRTMPConn( @@ -101,7 +101,7 @@ func newRTMPConn(
rtspAddress: rtspAddress,
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
runOnConnect: runOnConnect,
runOnConnectRestart: runOnConnectRestart,
wg: wg,
@ -241,7 +241,7 @@ func (c *rtmpConn) runRead(conn *rtmp.Conn, u *url.URL) error { @@ -241,7 +241,7 @@ func (c *rtmpConn) runRead(conn *rtmp.Conn, u *url.URL) error {
c.pathName = pathName
c.mutex.Unlock()
ringBuffer, _ := ringbuffer.New(uint64(c.readBufferCount))
ringBuffer, _ := ringbuffer.New(uint64(c.writeQueueSize))
go func() {
<-c.ctx.Done()
ringBuffer.Close()

8
internal/core/rtmp_server.go

@ -50,7 +50,7 @@ type rtmpServerParent interface { @@ -50,7 +50,7 @@ type rtmpServerParent interface {
type rtmpServer struct {
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
isTLS bool
rtspAddress string
runOnConnect string
@ -79,7 +79,7 @@ func newRTMPServer( @@ -79,7 +79,7 @@ func newRTMPServer(
address string,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
isTLS bool,
serverCert string,
serverKey string,
@ -113,7 +113,7 @@ func newRTMPServer( @@ -113,7 +113,7 @@ func newRTMPServer(
s := &rtmpServer{
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
rtspAddress: rtspAddress,
runOnConnect: runOnConnect,
runOnConnectRestart: runOnConnectRestart,
@ -185,7 +185,7 @@ outer: @@ -185,7 +185,7 @@ outer:
s.rtspAddress,
s.readTimeout,
s.writeTimeout,
s.readBufferCount,
s.writeQueueSize,
s.runOnConnect,
s.runOnConnectRestart,
&s.wg,

6
internal/core/rtsp_server.go

@ -67,7 +67,7 @@ func newRTSPServer( @@ -67,7 +67,7 @@ func newRTSPServer(
authMethods []headers.AuthMethod,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
useUDP bool,
useMulticast bool,
rtpAddress string,
@ -111,8 +111,8 @@ func newRTSPServer( @@ -111,8 +111,8 @@ func newRTSPServer(
Handler: s,
ReadTimeout: time.Duration(readTimeout),
WriteTimeout: time.Duration(writeTimeout),
ReadBufferCount: readBufferCount,
WriteBufferCount: readBufferCount,
ReadBufferCount: writeQueueSize,
WriteBufferCount: writeQueueSize,
RTSPAddress: address,
}

31
internal/core/rtsp_source.go

@ -66,23 +66,23 @@ type rtspSourceParent interface { @@ -66,23 +66,23 @@ type rtspSourceParent interface {
}
type rtspSource struct {
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
parent rtspSourceParent
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
writeQueueSize int
parent rtspSourceParent
}
func newRTSPSource(
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
parent rtspSourceParent,
) *rtspSource {
return &rtspSource{
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
parent: parent,
readTimeout: readTimeout,
writeTimeout: writeTimeout,
writeQueueSize: writeQueueSize,
parent: parent,
}
}
@ -95,12 +95,13 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha @@ -95,12 +95,13 @@ func (s *rtspSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf cha
s.Log(logger.Debug, "connecting")
c := &gortsplib.Client{
Transport: cnf.SourceProtocol.Transport,
TLSConfig: tlsConfigForFingerprint(cnf.SourceFingerprint),
ReadTimeout: time.Duration(s.readTimeout),
WriteTimeout: time.Duration(s.writeTimeout),
ReadBufferCount: s.readBufferCount,
AnyPortEnable: cnf.SourceAnyPortEnable,
Transport: cnf.SourceProtocol.Transport,
TLSConfig: tlsConfigForFingerprint(cnf.SourceFingerprint),
ReadTimeout: time.Duration(s.readTimeout),
WriteTimeout: time.Duration(s.writeTimeout),
ReadBufferCount: s.writeQueueSize,
WriteBufferCount: s.writeQueueSize,
AnyPortEnable: cnf.SourceAnyPortEnable,
OnRequest: func(req *base.Request) {
s.Log(logger.Debug, "c->s %v", req)
},

4
internal/core/source_static.go

@ -49,7 +49,7 @@ func newSourceStatic( @@ -49,7 +49,7 @@ func newSourceStatic(
cnf *conf.PathConf,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
parent sourceStaticParent,
) *sourceStatic {
s := &sourceStatic{
@ -66,7 +66,7 @@ func newSourceStatic( @@ -66,7 +66,7 @@ func newSourceStatic(
s.impl = newRTSPSource(
readTimeout,
writeTimeout,
readBufferCount,
writeQueueSize,
s)
case strings.HasPrefix(cnf.Source, "rtmp://") ||

8
internal/core/srt_conn.go

@ -50,7 +50,7 @@ type srtConnParent interface { @@ -50,7 +50,7 @@ type srtConnParent interface {
type srtConn struct {
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
udpMaxPayloadSize int
connReq srt.ConnRequest
wg *sync.WaitGroup
@ -75,7 +75,7 @@ func newSRTConn( @@ -75,7 +75,7 @@ func newSRTConn(
parentCtx context.Context,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
udpMaxPayloadSize int,
connReq srt.ConnRequest,
wg *sync.WaitGroup,
@ -88,7 +88,7 @@ func newSRTConn( @@ -88,7 +88,7 @@ func newSRTConn(
c := &srtConn{
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
connReq: connReq,
wg: wg,
@ -416,7 +416,7 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass @@ -416,7 +416,7 @@ func (c *srtConn) runRead(req srtNewConnReq, pathName string, user string, pass
c.conn = sconn
c.mutex.Unlock()
ringBuffer, _ := ringbuffer.New(uint64(c.readBufferCount))
ringBuffer, _ := ringbuffer.New(uint64(c.writeQueueSize))
go func() {
<-c.ctx.Done()
ringBuffer.Close()

8
internal/core/srt_server.go

@ -59,7 +59,7 @@ type srtServerParent interface { @@ -59,7 +59,7 @@ type srtServerParent interface {
type srtServer struct {
readTimeout conf.StringDuration
writeTimeout conf.StringDuration
readBufferCount int
writeQueueSize int
udpMaxPayloadSize int
externalCmdPool *externalcmd.Pool
pathManager *pathManager
@ -84,7 +84,7 @@ func newSRTServer( @@ -84,7 +84,7 @@ func newSRTServer(
address string,
readTimeout conf.StringDuration,
writeTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
udpMaxPayloadSize int,
externalCmdPool *externalcmd.Pool,
pathManager *pathManager,
@ -104,7 +104,7 @@ func newSRTServer( @@ -104,7 +104,7 @@ func newSRTServer(
s := &srtServer{
readTimeout: readTimeout,
writeTimeout: writeTimeout,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
udpMaxPayloadSize: udpMaxPayloadSize,
externalCmdPool: externalCmdPool,
pathManager: pathManager,
@ -161,7 +161,7 @@ outer: @@ -161,7 +161,7 @@ outer:
s.ctx,
s.readTimeout,
s.writeTimeout,
s.readBufferCount,
s.writeQueueSize,
s.udpMaxPayloadSize,
req.connReq,
&s.wg,

20
internal/core/webrtc_manager.go

@ -276,13 +276,13 @@ type webRTCManagerParent interface { @@ -276,13 +276,13 @@ type webRTCManagerParent interface {
}
type webRTCManager struct {
allowOrigin string
trustedProxies conf.IPsOrCIDRs
iceServers []conf.WebRTCICEServer
readBufferCount int
pathManager *pathManager
metrics *metrics
parent webRTCManagerParent
allowOrigin string
trustedProxies conf.IPsOrCIDRs
iceServers []conf.WebRTCICEServer
writeQueueSize int
pathManager *pathManager
metrics *metrics
parent webRTCManagerParent
ctx context.Context
ctxCancel func()
@ -314,7 +314,7 @@ func newWebRTCManager( @@ -314,7 +314,7 @@ func newWebRTCManager(
trustedProxies conf.IPsOrCIDRs,
iceServers []conf.WebRTCICEServer,
readTimeout conf.StringDuration,
readBufferCount int,
writeQueueSize int,
iceHostNAT1To1IPs []string,
iceUDPMuxAddress string,
iceTCPMuxAddress string,
@ -328,7 +328,7 @@ func newWebRTCManager( @@ -328,7 +328,7 @@ func newWebRTCManager(
allowOrigin: allowOrigin,
trustedProxies: trustedProxies,
iceServers: iceServers,
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
pathManager: pathManager,
metrics: metrics,
parent: parent,
@ -436,7 +436,7 @@ outer: @@ -436,7 +436,7 @@ outer:
case req := <-m.chNewSession:
sx := newWebRTCSession(
m.ctx,
m.readBufferCount,
m.writeQueueSize,
m.api,
req,
&wg,

18
internal/core/webrtc_session.go

@ -175,12 +175,12 @@ type webRTCSessionPathManager interface { @@ -175,12 +175,12 @@ type webRTCSessionPathManager interface {
}
type webRTCSession struct {
readBufferCount int
api *webrtc.API
req webRTCNewSessionReq
wg *sync.WaitGroup
pathManager webRTCSessionPathManager
parent *webRTCManager
writeQueueSize int
api *webrtc.API
req webRTCNewSessionReq
wg *sync.WaitGroup
pathManager webRTCSessionPathManager
parent *webRTCManager
ctx context.Context
ctxCancel func()
@ -196,7 +196,7 @@ type webRTCSession struct { @@ -196,7 +196,7 @@ type webRTCSession struct {
func newWebRTCSession(
parentCtx context.Context,
readBufferCount int,
writeQueueSize int,
api *webrtc.API,
req webRTCNewSessionReq,
wg *sync.WaitGroup,
@ -206,7 +206,7 @@ func newWebRTCSession( @@ -206,7 +206,7 @@ func newWebRTCSession(
ctx, ctxCancel := context.WithCancel(parentCtx)
s := &webRTCSession{
readBufferCount: readBufferCount,
writeQueueSize: writeQueueSize,
api: api,
req: req,
wg: wg,
@ -509,7 +509,7 @@ func (s *webRTCSession) runRead() (int, error) { @@ -509,7 +509,7 @@ func (s *webRTCSession) runRead() (int, error) {
s.pc = pc
s.mutex.Unlock()
ringBuffer, _ := ringbuffer.New(uint64(s.readBufferCount))
ringBuffer, _ := ringbuffer.New(uint64(s.writeQueueSize))
defer ringBuffer.Close()
writeError := make(chan error)

4
mediamtx.yml

@ -13,9 +13,9 @@ logFile: mediamtx.log @@ -13,9 +13,9 @@ logFile: mediamtx.log
readTimeout: 10s
# Timeout of write operations.
writeTimeout: 10s
# Number of read buffers.
# Size of the queue of outgoing packets.
# A higher value allows to increase throughput, a lower value allows to save RAM.
readBufferCount: 512
writeQueueSize: 512
# Maximum size of outgoing UDP packets.
# This can be decreased to avoid fragmentation on networks with a low UDP MTU.
udpMaxPayloadSize: 1472

Loading…
Cancel
Save