diff --git a/apidocs/openapi.yaml b/apidocs/openapi.yaml index 12e7afec..b8e50117 100644 --- a/apidocs/openapi.yaml +++ b/apidocs/openapi.yaml @@ -33,6 +33,8 @@ components: type: string readBufferCount: type: integer + udpMaxPayloadSize: + type: integer externalAuthenticationURL: type: string api: diff --git a/internal/conf/conf.go b/internal/conf/conf.go index 0fdaf480..f847b63c 100644 --- a/internal/conf/conf.go +++ b/internal/conf/conf.go @@ -175,6 +175,7 @@ type Conf struct { ReadTimeout StringDuration `json:"readTimeout"` WriteTimeout StringDuration `json:"writeTimeout"` ReadBufferCount int `json:"readBufferCount"` + UDPMaxPayloadSize int `json:"udpMaxPayloadSize"` ExternalAuthenticationURL string `json:"externalAuthenticationURL"` API bool `json:"api"` APIAddress string `json:"apiAddress"` @@ -285,7 +286,13 @@ func (conf *Conf) CheckAndFillMissing() error { conf.ReadBufferCount = 512 } if (conf.ReadBufferCount & (conf.ReadBufferCount - 1)) != 0 { - return fmt.Errorf("'ReadBufferCount' must be a power of two") + return fmt.Errorf("'readBufferCount' must be a power of two") + } + if conf.UDPMaxPayloadSize == 0 { + conf.UDPMaxPayloadSize = 1500 + } + if conf.UDPMaxPayloadSize > 1500 { + return fmt.Errorf("'udpMaxPayloadSize' must be less than 1500") } if conf.ExternalAuthenticationURL != "" { if !strings.HasPrefix(conf.ExternalAuthenticationURL, "http://") && diff --git a/internal/core/core.go b/internal/core/core.go index af3c9e3c..8790f586 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -244,6 +244,7 @@ func (p *Core) createResources(initial bool) error { p.conf.ReadTimeout, p.conf.WriteTimeout, p.conf.ReadBufferCount, + p.conf.UDPMaxPayloadSize, p.conf.Paths, p.externalCmdPool, p.metrics, @@ -486,6 +487,7 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) { newConf.ReadTimeout != p.conf.ReadTimeout || newConf.WriteTimeout != p.conf.WriteTimeout || newConf.ReadBufferCount != p.conf.ReadBufferCount || + newConf.UDPMaxPayloadSize != p.conf.UDPMaxPayloadSize || closeMetrics if !closePathManager && !reflect.DeepEqual(newConf.Paths, p.conf.Paths) { p.pathManager.confReload(newConf.Paths) diff --git a/internal/core/path.go b/internal/core/path.go index 8745c947..793fd60b 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -187,17 +187,18 @@ type pathAPIPathsListSubReq struct { } type path struct { - rtspAddress string - readTimeout conf.StringDuration - writeTimeout conf.StringDuration - readBufferCount int - confName string - conf *conf.PathConf - name string - matches []string - wg *sync.WaitGroup - externalCmdPool *externalcmd.Pool - parent pathParent + rtspAddress string + readTimeout conf.StringDuration + writeTimeout conf.StringDuration + readBufferCount int + udpMaxPayloadSize int + confName string + conf *conf.PathConf + name string + matches []string + wg *sync.WaitGroup + externalCmdPool *externalcmd.Pool + parent pathParent ctx context.Context ctxCancel func() @@ -240,6 +241,7 @@ func newPath( readTimeout conf.StringDuration, writeTimeout conf.StringDuration, readBufferCount int, + udpMaxPayloadSize int, confName string, cnf *conf.PathConf, name string, @@ -255,6 +257,7 @@ func newPath( readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, + udpMaxPayloadSize: udpMaxPayloadSize, confName: confName, conf: cnf, name: name, @@ -632,7 +635,12 @@ func (pa *path) onDemandPublisherStop() { } func (pa *path) sourceSetReady(medias media.Medias, allocateEncoder bool) error { - stream, err := newStream(medias, allocateEncoder, pa.bytesReceived) + stream, err := newStream( + pa.udpMaxPayloadSize, + medias, + allocateEncoder, + pa.bytesReceived, + ) if err != nil { return err } diff --git a/internal/core/path_manager.go b/internal/core/path_manager.go index 0d702024..b2cff291 100644 --- a/internal/core/path_manager.go +++ b/internal/core/path_manager.go @@ -40,14 +40,15 @@ type pathManagerParent interface { } type pathManager struct { - rtspAddress string - readTimeout conf.StringDuration - writeTimeout conf.StringDuration - readBufferCount int - pathConfs map[string]*conf.PathConf - externalCmdPool *externalcmd.Pool - metrics *metrics - parent pathManagerParent + rtspAddress string + readTimeout conf.StringDuration + writeTimeout conf.StringDuration + readBufferCount int + udpMaxPayloadSize int + pathConfs map[string]*conf.PathConf + externalCmdPool *externalcmd.Pool + metrics *metrics + parent pathManagerParent ctx context.Context ctxCancel func() @@ -74,6 +75,7 @@ func newPathManager( readTimeout conf.StringDuration, writeTimeout conf.StringDuration, readBufferCount int, + udpMaxPayloadSize int, pathConfs map[string]*conf.PathConf, externalCmdPool *externalcmd.Pool, metrics *metrics, @@ -86,6 +88,7 @@ func newPathManager( readTimeout: readTimeout, writeTimeout: writeTimeout, readBufferCount: readBufferCount, + udpMaxPayloadSize: udpMaxPayloadSize, pathConfs: pathConfs, externalCmdPool: externalCmdPool, metrics: metrics, @@ -303,6 +306,7 @@ func (pm *pathManager) createPath( pm.readTimeout, pm.writeTimeout, pm.readBufferCount, + pm.udpMaxPayloadSize, pathConfName, pathConf, name, diff --git a/internal/core/stream.go b/internal/core/stream.go index 54dfea7a..05a4d81c 100644 --- a/internal/core/stream.go +++ b/internal/core/stream.go @@ -15,6 +15,7 @@ type stream struct { } func newStream( + udpMaxPayloadSize int, medias media.Medias, generateRTPPackets bool, bytesReceived *uint64, @@ -28,7 +29,7 @@ func newStream( for _, media := range s.rtspStream.Medias() { var err error - s.smedias[media], err = newStreamMedia(media, generateRTPPackets) + s.smedias[media], err = newStreamMedia(udpMaxPayloadSize, media, generateRTPPackets) if err != nil { return nil, err } diff --git a/internal/core/stream_format.go b/internal/core/stream_format.go index 1b5a66d8..94de578e 100644 --- a/internal/core/stream_format.go +++ b/internal/core/stream_format.go @@ -16,8 +16,12 @@ type streamFormat struct { nonRTSPReaders map[reader]func(formatprocessor.Unit) } -func newStreamFormat(forma format.Format, generateRTPPackets bool) (*streamFormat, error) { - proc, err := formatprocessor.New(forma, generateRTPPackets) +func newStreamFormat( + udpMaxPayloadSize int, + forma format.Format, + generateRTPPackets bool, +) (*streamFormat, error) { + proc, err := formatprocessor.New(udpMaxPayloadSize, forma, generateRTPPackets) if err != nil { return nil, err } diff --git a/internal/core/stream_media.go b/internal/core/stream_media.go index b46f8edb..d4146b98 100644 --- a/internal/core/stream_media.go +++ b/internal/core/stream_media.go @@ -9,14 +9,17 @@ type streamMedia struct { formats map[format.Format]*streamFormat } -func newStreamMedia(medi *media.Media, generateRTPPackets bool) (*streamMedia, error) { +func newStreamMedia(udpMaxPayloadSize int, + medi *media.Media, + generateRTPPackets bool, +) (*streamMedia, error) { sm := &streamMedia{ formats: make(map[format.Format]*streamFormat), } for _, forma := range medi.Formats { var err error - sm.formats[forma], err = newStreamFormat(forma, generateRTPPackets) + sm.formats[forma], err = newStreamFormat(udpMaxPayloadSize, forma, generateRTPPackets) if err != nil { return nil, err } diff --git a/internal/core/udp_source.go b/internal/core/udp_source.go index 147de175..7c8241b7 100644 --- a/internal/core/udp_source.go +++ b/internal/core/udp_source.go @@ -21,6 +21,7 @@ import ( const ( multicastTTL = 16 + udpMTU = 1472 ) var opusDurations = [32]int{ @@ -127,7 +128,7 @@ func (s *udpSource) run(ctx context.Context, cnf *conf.PathConf, reloadConf chan } } - midbuffer := make([]byte, 0, 1472) // UDP MTU + midbuffer := make([]byte, 0, udpMTU) midbufferPos := 0 readPacket := func(buf []byte) (int, error) { diff --git a/internal/formatprocessor/generic.go b/internal/formatprocessor/generic.go index 142175e0..8aeb95d7 100644 --- a/internal/formatprocessor/generic.go +++ b/internal/formatprocessor/generic.go @@ -8,11 +8,6 @@ import ( "github.com/pion/rtp" ) -const ( - // 1500 (UDP MTU) - 20 (IP header) - 8 (UDP header) - maxPacketSize = 1472 -) - // UnitGeneric is a generic data unit. type UnitGeneric struct { RTPPackets []*rtp.Packet @@ -29,14 +24,22 @@ func (d *UnitGeneric) GetNTP() time.Time { return d.NTP } -type formatProcessorGeneric struct{} +type formatProcessorGeneric struct { + udpMaxPayloadSize int +} -func newGeneric(forma format.Format, generateRTPPackets bool) (*formatProcessorGeneric, error) { +func newGeneric( + udpMaxPayloadSize int, + forma format.Format, + generateRTPPackets bool, +) (*formatProcessorGeneric, error) { if generateRTPPackets { return nil, fmt.Errorf("we don't know how to generate RTP packets of format %+v", forma) } - return &formatProcessorGeneric{}, nil + return &formatProcessorGeneric{ + udpMaxPayloadSize: udpMaxPayloadSize, + }, nil } func (t *formatProcessorGeneric) Process(unit Unit, hasNonRTSPReaders bool) error { @@ -48,9 +51,9 @@ func (t *formatProcessorGeneric) Process(unit Unit, hasNonRTSPReaders bool) erro pkt.Header.Padding = false pkt.PaddingSize = 0 - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { return fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)", - pkt.MarshalSize(), maxPacketSize) + pkt.MarshalSize(), t.udpMaxPayloadSize) } return nil diff --git a/internal/formatprocessor/generic_test.go b/internal/formatprocessor/generic_test.go index 6f3ac1fb..8e817b73 100644 --- a/internal/formatprocessor/generic_test.go +++ b/internal/formatprocessor/generic_test.go @@ -15,7 +15,7 @@ func TestGenericRemovePadding(t *testing.T) { } forma.Init() - p, err := New(forma, false) + p, err := New(1472, forma, false) require.NoError(t, err) pkt := &rtp.Packet{ diff --git a/internal/formatprocessor/h264.go b/internal/formatprocessor/h264.go index 2a023868..22d06244 100644 --- a/internal/formatprocessor/h264.go +++ b/internal/formatprocessor/h264.go @@ -86,18 +86,21 @@ func (d *UnitH264) GetNTP() time.Time { } type formatProcessorH264 struct { - format *format.H264 + udpMaxPayloadSize int + format *format.H264 encoder *rtph264.Encoder decoder *rtph264.Decoder } func newH264( + udpMaxPayloadSize int, forma *format.H264, allocateEncoder bool, ) (*formatProcessorH264, error) { t := &formatProcessorH264{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -211,11 +214,12 @@ func (t *formatProcessorH264) Process(unit Unit, hasNonRTSPReaders bool) error { pkt.PaddingSize = 0 // RTP packets exceed maximum size: start re-encoding them - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { v1 := pkt.SSRC v2 := pkt.SequenceNumber v3 := pkt.Timestamp t.encoder = &rtph264.Encoder{ + PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadType: pkt.PayloadType, SSRC: &v1, InitialSequenceNumber: &v2, diff --git a/internal/formatprocessor/h264_test.go b/internal/formatprocessor/h264_test.go index 3d2cab08..4aa2f70e 100644 --- a/internal/formatprocessor/h264_test.go +++ b/internal/formatprocessor/h264_test.go @@ -16,7 +16,7 @@ func TestH264DynamicParams(t *testing.T) { PacketizationMode: 1, } - p, err := New(forma, false) + p, err := New(1472, forma, false) require.NoError(t, err) enc := forma.CreateEncoder() @@ -61,7 +61,7 @@ func TestH264OversizedPackets(t *testing.T) { PacketizationMode: 1, } - p, err := New(forma, false) + p, err := New(1472, forma, false) require.NoError(t, err) var out []*rtp.Packet @@ -158,7 +158,7 @@ func TestH264EmptyPacket(t *testing.T) { PacketizationMode: 1, } - p, err := New(forma, true) + p, err := New(1472, forma, true) require.NoError(t, err) unit := &UnitH264{ diff --git a/internal/formatprocessor/h265.go b/internal/formatprocessor/h265.go index 1df453ab..3c90c627 100644 --- a/internal/formatprocessor/h265.go +++ b/internal/formatprocessor/h265.go @@ -93,18 +93,21 @@ func (d *UnitH265) GetNTP() time.Time { } type formatProcessorH265 struct { - format *format.H265 + udpMaxPayloadSize int + format *format.H265 encoder *rtph265.Encoder decoder *rtph265.Decoder } func newH265( + udpMaxPayloadSize int, forma *format.H265, allocateEncoder bool, ) (*formatProcessorH265, error) { t := &formatProcessorH265{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -232,11 +235,12 @@ func (t *formatProcessorH265) Process(unit Unit, hasNonRTSPReaders bool) error { pkt.PaddingSize = 0 // RTP packets exceed maximum size: start re-encoding them - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { v1 := pkt.SSRC v2 := pkt.SequenceNumber v3 := pkt.Timestamp t.encoder = &rtph265.Encoder{ + PayloadMaxSize: t.udpMaxPayloadSize - 12, PayloadType: pkt.PayloadType, SSRC: &v1, InitialSequenceNumber: &v2, diff --git a/internal/formatprocessor/h265_test.go b/internal/formatprocessor/h265_test.go index 2b9687ec..78a19bdf 100644 --- a/internal/formatprocessor/h265_test.go +++ b/internal/formatprocessor/h265_test.go @@ -15,7 +15,7 @@ func TestH265DynamicParams(t *testing.T) { PayloadTyp: 96, } - p, err := New(forma, false) + p, err := New(1472, forma, false) require.NoError(t, err) enc := forma.CreateEncoder() @@ -66,7 +66,7 @@ func TestH265OversizedPackets(t *testing.T) { PPS: []byte{byte(h265.NALUType_PPS_NUT) << 1, 16, 17, 18}, } - p, err := New(forma, false) + p, err := New(1472, forma, false) require.NoError(t, err) var out []*rtp.Packet @@ -150,7 +150,7 @@ func TestH265EmptyPacket(t *testing.T) { PayloadTyp: 96, } - p, err := New(forma, true) + p, err := New(1472, forma, true) require.NoError(t, err) unit := &UnitH265{ diff --git a/internal/formatprocessor/mpeg4audio.go b/internal/formatprocessor/mpeg4audio.go index c5488a97..ac2f0f59 100644 --- a/internal/formatprocessor/mpeg4audio.go +++ b/internal/formatprocessor/mpeg4audio.go @@ -28,17 +28,20 @@ func (d *UnitMPEG4Audio) GetNTP() time.Time { } type formatProcessorMPEG4Audio struct { - format *format.MPEG4Audio - encoder *rtpmpeg4audio.Encoder - decoder *rtpmpeg4audio.Decoder + udpMaxPayloadSize int + format *format.MPEG4Audio + encoder *rtpmpeg4audio.Encoder + decoder *rtpmpeg4audio.Decoder } func newMPEG4Audio( + udpMaxPayloadSize int, forma *format.MPEG4Audio, allocateEncoder bool, ) (*formatProcessorMPEG4Audio, error) { t := &formatProcessorMPEG4Audio{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -58,9 +61,9 @@ func (t *formatProcessorMPEG4Audio) Process(unit Unit, hasNonRTSPReaders bool) e pkt.Header.Padding = false pkt.PaddingSize = 0 - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { return fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)", - pkt.MarshalSize(), maxPacketSize) + pkt.MarshalSize(), t.udpMaxPayloadSize) } // decode from RTP diff --git a/internal/formatprocessor/opus.go b/internal/formatprocessor/opus.go index babc25f5..18790f11 100644 --- a/internal/formatprocessor/opus.go +++ b/internal/formatprocessor/opus.go @@ -28,17 +28,20 @@ func (d *UnitOpus) GetNTP() time.Time { } type formatProcessorOpus struct { - format *format.Opus - encoder *rtpsimpleaudio.Encoder - decoder *rtpsimpleaudio.Decoder + udpMaxPayloadSize int + format *format.Opus + encoder *rtpsimpleaudio.Encoder + decoder *rtpsimpleaudio.Decoder } func newOpus( + udpMaxPayloadSize int, forma *format.Opus, allocateEncoder bool, ) (*formatProcessorOpus, error) { t := &formatProcessorOpus{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -58,9 +61,9 @@ func (t *formatProcessorOpus) Process(unit Unit, hasNonRTSPReaders bool) error { pkt.Header.Padding = false pkt.PaddingSize = 0 - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { return fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)", - pkt.MarshalSize(), maxPacketSize) + pkt.MarshalSize(), t.udpMaxPayloadSize) } // decode from RTP diff --git a/internal/formatprocessor/processor.go b/internal/formatprocessor/processor.go index 5bad09c2..430d9010 100644 --- a/internal/formatprocessor/processor.go +++ b/internal/formatprocessor/processor.go @@ -12,27 +12,31 @@ type Processor interface { } // New allocates a Processor. -func New(forma format.Format, generateRTPPackets bool) (Processor, error) { +func New( + udpMaxPayloadSize int, + forma format.Format, + generateRTPPackets bool, +) (Processor, error) { switch forma := forma.(type) { case *format.H264: - return newH264(forma, generateRTPPackets) + return newH264(udpMaxPayloadSize, forma, generateRTPPackets) case *format.H265: - return newH265(forma, generateRTPPackets) + return newH265(udpMaxPayloadSize, forma, generateRTPPackets) case *format.VP8: - return newVP8(forma, generateRTPPackets) + return newVP8(udpMaxPayloadSize, forma, generateRTPPackets) case *format.VP9: - return newVP9(forma, generateRTPPackets) + return newVP9(udpMaxPayloadSize, forma, generateRTPPackets) case *format.MPEG4Audio: - return newMPEG4Audio(forma, generateRTPPackets) + return newMPEG4Audio(udpMaxPayloadSize, forma, generateRTPPackets) case *format.Opus: - return newOpus(forma, generateRTPPackets) + return newOpus(udpMaxPayloadSize, forma, generateRTPPackets) default: - return newGeneric(forma, generateRTPPackets) + return newGeneric(udpMaxPayloadSize, forma, generateRTPPackets) } } diff --git a/internal/formatprocessor/vp8.go b/internal/formatprocessor/vp8.go index 5e0b0bd2..e2f20f62 100644 --- a/internal/formatprocessor/vp8.go +++ b/internal/formatprocessor/vp8.go @@ -28,17 +28,20 @@ func (d *UnitVP8) GetNTP() time.Time { } type formatProcessorVP8 struct { - format *format.VP8 - encoder *rtpvp8.Encoder - decoder *rtpvp8.Decoder + udpMaxPayloadSize int + format *format.VP8 + encoder *rtpvp8.Encoder + decoder *rtpvp8.Decoder } func newVP8( + udpMaxPayloadSize int, forma *format.VP8, allocateEncoder bool, ) (*formatProcessorVP8, error) { t := &formatProcessorVP8{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -58,9 +61,9 @@ func (t *formatProcessorVP8) Process(unit Unit, hasNonRTSPReaders bool) error { pkt.Header.Padding = false pkt.PaddingSize = 0 - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { return fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)", - pkt.MarshalSize(), maxPacketSize) + pkt.MarshalSize(), t.udpMaxPayloadSize) } // decode from RTP diff --git a/internal/formatprocessor/vp9.go b/internal/formatprocessor/vp9.go index 58ccfd48..9250935d 100644 --- a/internal/formatprocessor/vp9.go +++ b/internal/formatprocessor/vp9.go @@ -28,17 +28,20 @@ func (d *UnitVP9) GetNTP() time.Time { } type formatProcessorVP9 struct { - format *format.VP9 - encoder *rtpvp9.Encoder - decoder *rtpvp9.Decoder + udpMaxPayloadSize int + format *format.VP9 + encoder *rtpvp9.Encoder + decoder *rtpvp9.Decoder } func newVP9( + udpMaxPayloadSize int, forma *format.VP9, allocateEncoder bool, ) (*formatProcessorVP9, error) { t := &formatProcessorVP9{ - format: forma, + udpMaxPayloadSize: udpMaxPayloadSize, + format: forma, } if allocateEncoder { @@ -58,9 +61,9 @@ func (t *formatProcessorVP9) Process(unit Unit, hasNonRTSPReaders bool) error { pkt.Header.Padding = false pkt.PaddingSize = 0 - if pkt.MarshalSize() > maxPacketSize { + if pkt.MarshalSize() > t.udpMaxPayloadSize { return fmt.Errorf("payload size (%d) is greater than maximum allowed (%d)", - pkt.MarshalSize(), maxPacketSize) + pkt.MarshalSize(), t.udpMaxPayloadSize) } // decode from RTP diff --git a/rtsp-simple-server.yml b/rtsp-simple-server.yml index ac491e33..ebdb09ab 100644 --- a/rtsp-simple-server.yml +++ b/rtsp-simple-server.yml @@ -16,6 +16,9 @@ writeTimeout: 10s # Number of read buffers. # A higher value allows a wider throughput, a lower value allows to save RAM. readBufferCount: 512 +# Maximum size of payload of outgoing UDP packets. +# This can be decreased to avoid fragmentation on networks with a low UDP MTU. +udpMaxPayloadSize: 1472 # HTTP URL to perform external authentication. # Every time a user wants to authenticate, the server calls this URL