fix memory leak when reloading the configuration (#4855)

When a path has a MPEG-TS, RTP or WebRTC source and the path
configuration is reloaded, a routine was left open because the reload
channel was not handled. This fixes the issue.

internal/staticsources/hls/source_test.go

@@ -102,14 +102,20 @@ func TestSource(t *testing.T) {
 	ctx, ctxCancel := context.WithCancel(context.Background())
 	defer ctxCancel()
 
+	reloadConf := make(chan *conf.Path)
+
 	go func() {
 		so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 			Context:        ctx,
 			ResolvedSource: "http://localhost:5780/stream.m3u8",
 			Conf:           &conf.Path{},
+			ReloadConf:     reloadConf,
 		})
 		close(done)
 	}()
 
 	<-p.Unit
+
+	// the source must be listening on ReloadConf
+	reloadConf <- nil
 }

internal/staticsources/mpegts/source.go

@@ -66,17 +66,21 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 		readerErr <- s.runReader(nc)
 	}()
 
+	for {
 		select {
 		case err = <-readerErr:
 			nc.Close()
 			return err
 
+		case <-params.ReloadConf:
+
 		case <-params.Context.Done():
 			nc.Close()
 			<-readerErr
 			return fmt.Errorf("terminated")
 		}
 	}
+}
 
 func (s *Source) runReader(nc net.Conn) error {
 	nc.SetReadDeadline(time.Now().Add(time.Duration(s.ReadTimeout)))

internal/staticsources/mpegts/source_test.go

@@ -69,11 +69,14 @@ func TestSourceUDP(t *testing.T) {
 			ctx, ctxCancel := context.WithCancel(context.Background())
 			defer ctxCancel()
 
+			reloadConf := make(chan *conf.Path)
+
 			go func() {
 				so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 					Context:        ctx,
 					ResolvedSource: src,
 					Conf:           &conf.Path{},
+					ReloadConf:     reloadConf,
 				})
 				close(done)
 			}()
@@ -128,6 +131,9 @@ func TestSourceUDP(t *testing.T) {
 			require.NoError(t, err)
 
 			<-p.Unit
+
+			// the source must be listening on ReloadConf
+			reloadConf <- nil
 		})
 	}
 }

internal/staticsources/rtmp/source_test.go

@@ -51,7 +51,50 @@ func TestSource(t *testing.T) {
 
 				defer ln.Close()
 
+				var source string
+
+				if encryption == "plain" {
+					source = "rtmp://"
+				} else {
+					source = "rtmps://"
+				}
+
+				if auth == "auth" {
+					source += "myuser:mypass@"
+				}
+
+				source += "localhost/teststream"
+
+				p := &test.StaticSourceParent{}
+				p.Initialize()
+				defer p.Close()
+
+				so := &Source{
+					ReadTimeout:  conf.Duration(10 * time.Second),
+					WriteTimeout: conf.Duration(10 * time.Second),
+					Parent:       p,
+				}
+
+				done := make(chan struct{})
+				defer func() { <-done }()
+
+				ctx, ctxCancel := context.WithCancel(context.Background())
+				defer ctxCancel()
+
+				reloadConf := make(chan *conf.Path)
+
 				go func() {
+					so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
+						Context:        ctx,
+						ResolvedSource: source,
+						Conf: &conf.Path{
+							SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739",
+						},
+						ReloadConf: reloadConf,
+					})
+					close(done)
+				}()
+
 				for {
 					nconn, err := ln.Accept()
 					require.NoError(t, err)
@@ -89,50 +132,11 @@ func TestSource(t *testing.T) {
 
 					break
 				}
-				}()
-
-				var source string
-
-				if encryption == "plain" {
-					source = "rtmp://"
-				} else {
-					source = "rtmps://"
-				}
-
-				if auth == "auth" {
-					source += "myuser:mypass@"
-				}
-
-				source += "localhost/teststream"
-
-				p := &test.StaticSourceParent{}
-				p.Initialize()
-				defer p.Close()
-
-				so := &Source{
-					ReadTimeout:  conf.Duration(10 * time.Second),
-					WriteTimeout: conf.Duration(10 * time.Second),
-					Parent:       p,
-				}
-
-				done := make(chan struct{})
-				defer func() { <-done }()
-
-				ctx, ctxCancel := context.WithCancel(context.Background())
-				defer ctxCancel()
-
-				go func() {
-					so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
-						Context:        ctx,
-						ResolvedSource: source,
-						Conf: &conf.Path{
-							SourceFingerprint: "33949E05FFFB5FF3E8AA16F8213A6251B4D9363804BA53233C4DA9A46D6F2739",
-						},
-					})
-					close(done)
-				}()
 
 				<-p.Unit
+
+				// the source must be listening on ReloadConf
+				reloadConf <- nil
 			})
 		}
 	}

internal/staticsources/rtp/source.go

@@ -80,17 +80,21 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 		readerErr <- s.runReader(&desc, nc)
 	}()
 
+	for {
 		select {
 		case err = <-readerErr:
 			nc.Close()
 			return err
 
+		case <-params.ReloadConf:
+
 		case <-params.Context.Done():
 			nc.Close()
 			<-readerErr
 			return fmt.Errorf("terminated")
 		}
 	}
+}
 
 func (s *Source) runReader(desc *description.Session, nc net.Conn) error {
 	decodeErrors := &counterdumper.CounterDumper{

internal/staticsources/rtp/source_test.go

@@ -69,6 +69,8 @@ func TestSourceUDP(t *testing.T) {
 			ctx, ctxCancel := context.WithCancel(context.Background())
 			defer ctxCancel()
 
+			reloadConf := make(chan *conf.Path)
+
 			go func() {
 				so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 					Context:        ctx,
@@ -83,6 +85,7 @@ func TestSourceUDP(t *testing.T) {
 							"a=rtpmap:96 H264/90000\n" +
 							"a=fmtp:96 profile-level-id=42e01e;packetization-mode=1\n",
 					},
+					ReloadConf: reloadConf,
 				})
 				close(done)
 			}()
@@ -139,6 +142,9 @@ func TestSourceUDP(t *testing.T) {
 			}
 
 			<-p.Unit
+
+			// the source must be listening on ReloadConf
+			reloadConf <- nil
 		})
 	}
 }

internal/staticsources/rtsp/source_test.go

@@ -169,16 +169,22 @@ func TestSource(t *testing.T) {
 			ctx, ctxCancel := context.WithCancel(context.Background())
 			defer ctxCancel()
 
+			reloadConf := make(chan *conf.Path)
+
 			go func() {
 				so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 					Context:        ctx,
 					ResolvedSource: ur,
 					Conf:           cnf,
+					ReloadConf:     reloadConf,
 				})
 				close(done)
 			}()
 
 			<-p.Unit
+
+			// the source must be listening on ReloadConf
+			reloadConf <- nil
 		})
 	}
 }

internal/staticsources/srt/source_test.go

@@ -1,7 +1,6 @@
 package srt
 
 import (
-	"bufio"
 	"context"
 	"testing"
 	"time"
@@ -20,42 +19,8 @@ func TestSource(t *testing.T) {
 	require.NoError(t, err)
 	defer ln.Close()
 
-	go func() {
-		req, err2 := ln.Accept2()
-		require.NoError(t, err2)
-
-		require.Equal(t, "sidname", req.StreamId())
-		err2 = req.SetPassphrase("ttest1234567")
-		require.NoError(t, err2)
-
-		conn, err2 := req.Accept()
-		require.NoError(t, err2)
-		defer conn.Close()
-
-		track := &mpegts.Track{
-			Codec: &mpegts.CodecH264{},
-		}
-
-		bw := bufio.NewWriter(conn)
-		w := &mpegts.Writer{W: bw, Tracks: []*mpegts.Track{track}}
-		err2 = w.Initialize()
-		require.NoError(t, err2)
-
-		err2 = w.WriteH264(track, 0, 0, [][]byte{{ // IDR
-			5, 1,
-		}})
-		require.NoError(t, err2)
-
-		err2 = bw.Flush()
-		require.NoError(t, err2)
-
-		// wait for internal SRT queue to be written
-		time.Sleep(500 * time.Millisecond)
-	}()
-
 	p := &test.StaticSourceParent{}
 	p.Initialize()
-	defer p.Close()
 
 	so := &Source{
 		ReadTimeout: conf.Duration(10 * time.Second),
@@ -68,14 +33,50 @@ func TestSource(t *testing.T) {
 	ctx, ctxCancel := context.WithCancel(context.Background())
 	defer ctxCancel()
 
+	reloadConf := make(chan *conf.Path)
+
 	go func() {
 		so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 			Context:        ctx,
 			ResolvedSource: "srt://127.0.0.1:9002?streamid=sidname&passphrase=ttest1234567",
 			Conf:           &conf.Path{},
+			ReloadConf:     reloadConf,
 		})
 		close(done)
 	}()
 
+	req, err2 := ln.Accept2()
+	require.NoError(t, err2)
+
+	require.Equal(t, "sidname", req.StreamId())
+	err2 = req.SetPassphrase("ttest1234567")
+	require.NoError(t, err2)
+
+	conn, err2 := req.Accept()
+	require.NoError(t, err2)
+	defer conn.Close()
+
+	track := &mpegts.Track{Codec: &mpegts.CodecH264{}}
+
+	w := &mpegts.Writer{W: conn, Tracks: []*mpegts.Track{track}}
+	err2 = w.Initialize()
+	require.NoError(t, err2)
+
+	err2 = w.WriteH264(track, 0, 0, [][]byte{{ // IDR
+		5, 1,
+	}})
+	require.NoError(t, err2)
+
+	err = w.WriteH264(track, 0, 0, [][]byte{{ // non-IDR
+		5, 2,
+	}})
+	require.NoError(t, err)
+
 	<-p.Unit
+
+	// the source must be listening on ReloadConf
+	reloadConf <- nil
+
+	// stop test reader before 2nd H264 packet is received to avoid a crash
+	p.Close()
 }

internal/staticsources/webrtc/source.go

@@ -2,6 +2,8 @@
 package webrtc
 
 import (
+	"context"
+	"fmt"
 	"net/http"
 	"net/url"
 	"strings"
@@ -65,12 +67,12 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 	if err != nil {
 		return err
 	}
-	defer client.Close() //nolint:errcheck
 
 	var stream *stream.Stream
 
 	medias, err := webrtc.ToStream(client.PeerConnection(), &stream)
 	if err != nil {
+		client.Close() //nolint:errcheck
 		return err
 	}
 
@@ -79,6 +81,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 		GenerateRTPPackets: true,
 	})
 	if rres.Err != nil {
+		client.Close() //nolint:errcheck
 		return rres.Err
 	}
 
@@ -88,7 +91,26 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 
 	client.StartReading()
 
-	return client.Wait(params.Context)
+	readErr := make(chan error)
+
+	go func() {
+		readErr <- client.Wait(context.Background())
+	}()
+
+	for {
+		select {
+		case err = <-readErr:
+			client.Close() //nolint:errcheck
+			return err
+
+		case <-params.ReloadConf:
+
+		case <-params.Context.Done():
+			client.Close() //nolint:errcheck
+			<-readErr
+			return fmt.Errorf("terminated")
+		}
+	}
 }
 
 // APISourceDescribe implements StaticSource.

internal/staticsources/webrtc/source_test.go

@@ -138,14 +138,20 @@ func TestSource(t *testing.T) {
 	ctx, ctxCancel := context.WithCancel(context.Background())
 	defer ctxCancel()
 
+	reloadConf := make(chan *conf.Path)
+
 	go func() {
 		so.Run(defs.StaticSourceRunParams{ //nolint:errcheck
 			Context:        ctx,
 			ResolvedSource: "whep://localhost:9003/my/resource",
 			Conf:           &conf.Path{},
+			ReloadConf:     reloadConf,
 		})
 		close(done)
 	}()
 
 	<-p.Unit
+
+	// the source must be listening on ReloadConf
+	reloadConf <- nil
 }