fix race condition during sub-stream creation (#6075) (#6095)

When a stream with always-available turned on switches from offline to
online, or from a publisher to another, the reader mutex was not
acquired during writing of codec parameters. This is now fixed.

internal/stream/stream.go

@@ -492,6 +492,9 @@ func (s *Stream) RTSPSStream(server *gortsplib.Server) (*gortsplib.ServerStream,
 // AddReader adds a reader.
 // Used by all protocols except RTSP.
 func (s *Stream) AddReader(r *Reader) {
+	r.queueSize = s.WriteQueueSize
+	r.start()
+
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -506,9 +509,6 @@ func (s *Stream) AddReader(r *Reader) {
 		}
 	}
 
-	r.queueSize = s.WriteQueueSize
-	r.start()
-
 	select {
 	case <-s.hasReaders:
 	default:
@@ -520,9 +520,6 @@ func (s *Stream) AddReader(r *Reader) {
 // Used by all protocols except RTSP.
 func (s *Stream) RemoveReader(r *Reader) {
 	s.mutex.Lock()
-	defer s.mutex.Unlock()
-
-	r.stop()
 
 	for medi, formats := range r.onDatas {
 		sm := s.medias[medi]
@@ -534,6 +531,10 @@ func (s *Stream) RemoveReader(r *Reader) {
 	}
 
 	delete(s.readers, r)
+
+	s.mutex.Unlock()
+
+	r.stop()
 }
 
 // WaitForReaders waits for the stream to have at least one reader.

internal/stream/stream_test.go

@@ -828,3 +828,79 @@ func TestStreamAlwaysAvailable(t *testing.T) {
 		})
 	}
 }
+
+func TestStreamAlwaysAvailableConcurrentReaders(t *testing.T) {
+	strm := &Stream{
+		AlwaysAvailable: true,
+		AlwaysAvailableTracks: []conf.AlwaysAvailableTrack{
+			{Codec: conf.CodecH264},
+		},
+		WriteQueueSize:    512,
+		RTPMaxPayloadSize: 1450,
+		ReplaceNTP:        true,
+		Parent:            &nilLogger{},
+	}
+	err := strm.Initialize()
+	require.NoError(t, err)
+	defer strm.Close()
+
+	origMedia := strm.OrigDesc.Medias[0]
+	origFormat := origMedia.Formats[0]
+
+	start := make(chan struct{})
+	ready := make(chan struct{}, 1)
+	done := make(chan struct{})
+
+	var readersWG sync.WaitGroup
+	for range 4 {
+		readersWG.Go(func() {
+			<-start
+
+			for {
+				select {
+				case <-done:
+					return
+				default:
+				}
+
+				r := &Reader{Parent: &nilLogger{}}
+				r.OnData(origMedia, origFormat, func(_ *unit.Unit) error {
+					return nil
+				})
+
+				strm.AddReader(r)
+				select {
+				case ready <- struct{}{}:
+				default:
+				}
+				strm.RemoveReader(r)
+			}
+		})
+	}
+
+	close(start)
+	<-ready
+
+	for range 256 {
+		subStream := &SubStream{
+			Stream: strm,
+			InDesc: &description.Session{Medias: []*description.Media{
+				{
+					Type: description.MediaTypeVideo,
+					Formats: []format.Format{&format.H264{
+						PacketizationMode: 1,
+						SPS:               offlineH264SPS,
+						PPS:               offlineH264PPS,
+					}},
+				},
+			}},
+			UseRTPPackets: false,
+		}
+
+		err = subStream.Initialize()
+		require.NoError(t, err)
+	}
+
+	close(done)
+	readersWG.Wait()
+}

internal/stream/sub_stream.go

@@ -150,9 +150,11 @@ func (ss *SubStream) Initialize() error {
 		}
 	}
 
+	// keep mutex open to use writeUnit() inside initialize2()
 	ss.Stream.mutex.Lock()
+	defer ss.Stream.mutex.Unlock()
+
 	ss.Stream.subStream = ss
-	ss.Stream.mutex.Unlock()
 
 	for _, ssm := range ss.medias {
 		for _, ssf := range ssm.formats {