From 4974cacb940d5e47720fe5dc34afca99ba69c3ab Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Sun, 22 Feb 2026 18:25:32 +0100 Subject: [PATCH] improve video/audio sync of alwaysAvailable (#5443) (#5508) store elapsed time once for the entire stream and start PTS of sub streams from there. --- internal/stream/stream.go | 29 ++++++++++++++++++++++++---- internal/stream/stream_format.go | 24 ++++++++++------------- internal/stream/stream_media.go | 10 ++++++---- internal/stream/sub_stream.go | 2 +- internal/stream/sub_stream_format.go | 23 +++++++++++----------- 5 files changed, 53 insertions(+), 35 deletions(-) diff --git a/internal/stream/stream.go b/internal/stream/stream.go index d50cb48c..305aefcd 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -239,6 +239,11 @@ type Stream struct { readers map[*Reader]struct{} processingErrors *errordumper.Dumper + timeMutex sync.Mutex + firstTimeReceived bool + lastPTS time.Duration + lastSystemTime time.Time + hasReaders chan struct{} } @@ -286,14 +291,17 @@ func (s *Stream) Initialize() error { } s.processingErrors.Start() + s.lastSystemTime = time.Now() + for _, media := range s.Desc.Medias { sm := &streamMedia{ media: media, alwaysAvailable: s.AlwaysAvailable, rtpMaxPayloadSize: s.RTPMaxPayloadSize, replaceNTP: s.ReplaceNTP, - onBytesReceived: s.onBytesReceived, - onBytesSent: s.onBytesSent, + addBytesReceived: s.addBytesReceived, + addBytesSent: s.addBytesSent, + updateLastTime: s.updateLastTime, writeRTSP: s.writeRTSP, processingErrors: s.processingErrors, parent: s.Parent, @@ -466,14 +474,27 @@ func (s *Stream) WaitForReaders() { <-s.hasReaders } -func (s *Stream) onBytesReceived(v uint64) { +func (s *Stream) addBytesReceived(v uint64) { atomic.AddUint64(s.bytesReceived, v) } -func (s *Stream) onBytesSent(v uint64) { +func (s *Stream) addBytesSent(v uint64) { atomic.AddUint64(s.bytesSent, v) } +func (s *Stream) updateLastTime(pts time.Duration) { + s.timeMutex.Lock() + defer s.timeMutex.Unlock() + + s.firstTimeReceived = true + + if pts > s.lastPTS { + s.lastPTS = pts + } + + s.lastSystemTime = time.Now() +} + func (s *Stream) writeRTSP(medi *description.Media, pkts []*rtp.Packet, ntp time.Time) { if s.rtspStream != nil { for _, pkt := range pkts { diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index cc46a0ed..8f76ed44 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -44,26 +44,22 @@ type streamFormat struct { rtpMaxPayloadSize int replaceNTP bool processingErrors *errordumper.Dumper - onBytesReceived func(uint64) - onBytesSent func(uint64) + addBytesReceived func(uint64) + addBytesSent func(uint64) + updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) parent logger.Writer - firstReceived bool - lastPTS int64 - lastSystemTime time.Time - ptsOffset int64 - formatUpdater formatUpdater - unitRemuxer unitRemuxer - rtpEncoder rtpEncoder - rtpTimeOffset uint32 - ntpEstimator *ntpestimator.Estimator - onDatas map[*Reader]OnDataFunc + ptsOffset int64 + formatUpdater formatUpdater + unitRemuxer unitRemuxer + rtpEncoder rtpEncoder + rtpTimeOffset uint32 + ntpEstimator *ntpestimator.Estimator + onDatas map[*Reader]OnDataFunc } func (sf *streamFormat) initialize() error { - sf.lastSystemTime = time.Now() - sf.formatUpdater = newFormatUpdater(sf.format) sf.unitRemuxer = newUnitRemuxer(sf.format) diff --git a/internal/stream/stream_media.go b/internal/stream/stream_media.go index b92d401c..90e314c4 100644 --- a/internal/stream/stream_media.go +++ b/internal/stream/stream_media.go @@ -15,8 +15,9 @@ type streamMedia struct { alwaysAvailable bool rtpMaxPayloadSize int replaceNTP bool - onBytesReceived func(uint64) - onBytesSent func(uint64) + addBytesReceived func(uint64) + addBytesSent func(uint64) + updateLastTime func(time.Duration) writeRTSP func(*description.Media, []*rtp.Packet, time.Time) processingErrors *errordumper.Dumper parent logger.Writer @@ -35,8 +36,9 @@ func (sm *streamMedia) initialize() error { rtpMaxPayloadSize: sm.rtpMaxPayloadSize, replaceNTP: sm.replaceNTP, processingErrors: sm.processingErrors, - onBytesReceived: sm.onBytesReceived, - onBytesSent: sm.onBytesSent, + addBytesReceived: sm.addBytesReceived, + addBytesSent: sm.addBytesSent, + updateLastTime: sm.updateLastTime, writeRTSP: sm.writeRTSP, parent: sm.parent, } diff --git a/internal/stream/sub_stream.go b/internal/stream/sub_stream.go index 2e49552f..079189d6 100644 --- a/internal/stream/sub_stream.go +++ b/internal/stream/sub_stream.go @@ -187,7 +187,7 @@ func (ss *SubStream) Initialize() error { for _, ssm := range ss.medias { for _, ssf := range ssm.formats { - ssf.initialize2() + ssf.initialize2(ss.Stream.firstTimeReceived, ss.Stream.lastPTS, ss.Stream.lastSystemTime) } } diff --git a/internal/stream/sub_stream_format.go b/internal/stream/sub_stream_format.go index 42fbaa82..848f6a8d 100644 --- a/internal/stream/sub_stream_format.go +++ b/internal/stream/sub_stream_format.go @@ -45,7 +45,7 @@ func (ssf *subStreamFormat) initialize() error { return nil } -func (ssf *subStreamFormat) initialize2() { +func (ssf *subStreamFormat) initialize2(firstTimeReceived bool, lastPTS time.Duration, lastSystemTime time.Time) { if ssf.tempRTPEncoder != nil { if ssf.streamFormat.rtpEncoder == nil { ssf.streamFormat.rtpEncoder = ssf.tempRTPEncoder @@ -57,10 +57,10 @@ func (ssf *subStreamFormat) initialize2() { } if ssf.streamFormat.alwaysAvailable { - if ssf.streamFormat.firstReceived { - deltaT := max(1, multiplyAndDivide( - int64(time.Since(ssf.streamFormat.lastSystemTime)), int64(ssf.streamFormat.format.ClockRate()), int64(time.Second))) - ssf.streamFormat.ptsOffset = ssf.streamFormat.lastPTS + deltaT + if firstTimeReceived { + ptsOffsetGo := lastPTS + time.Since(lastSystemTime) + ssf.streamFormat.ptsOffset = multiplyAndDivide(int64(ptsOffsetGo), + int64(ssf.streamFormat.format.ClockRate()), int64(time.Second)) } switch curFormat := ssf.curFormat.(type) { @@ -101,12 +101,11 @@ func (ssf *subStreamFormat) writeUnit(u *unit.Unit) { func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { if ssf.streamFormat.alwaysAvailable { - ssf.streamFormat.firstReceived = true u.PTS += ssf.streamFormat.ptsOffset - if u.PTS > ssf.streamFormat.lastPTS { - ssf.streamFormat.lastPTS = u.PTS - } - ssf.streamFormat.lastSystemTime = time.Now() + + ssf.streamFormat.updateLastTime( + multiplyAndDivide2(time.Duration(u.PTS), + time.Second, time.Duration(ssf.streamFormat.format.ClockRate()))) } if ssf.streamFormat.replaceNTP { @@ -171,7 +170,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { } size := unitSize(u) - ssf.streamFormat.onBytesReceived(size) + ssf.streamFormat.addBytesReceived(size) ssf.streamFormat.writeRTSP(ssf.streamFormat.media, u.RTPPackets, u.NTP) @@ -180,7 +179,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { cOnData := onData sr.push(func() error { if !csr.SkipBytesSent { - ssf.streamFormat.onBytesSent(size) + ssf.streamFormat.addBytesSent(size) } return cOnData(u) })