diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 98388583..7b679dbd 100644 --- a/internal/stream/stream.go +++ b/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. diff --git a/internal/stream/stream_test.go b/internal/stream/stream_test.go index 9b2a39a3..67bdf91f 100644 --- a/internal/stream/stream_test.go +++ b/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() +} diff --git a/internal/stream/sub_stream.go b/internal/stream/sub_stream.go index 20ad64de..8fa7cba4 100644 --- a/internal/stream/sub_stream.go +++ b/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 {