OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

improve video/audio sync of alwaysAvailable (#5443) (#5508)

Browse Source 
store elapsed time once for the entire stream and start PTS of sub
streams from there.
This commit is contained in:
Alessandro Ros
2026-02-22 18:25:32 +01:00
committed by GitHub
parent b9942c7cfa
commit 4974cacb94
5 changed files with 53 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/stream/stream.go
+25 -4
View File
@@ -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 {
internal/stream/stream_format.go
+10 -14
View File
@@ -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)
internal/stream/stream_media.go
+6 -4
View File
@@ -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,
}
internal/stream/sub_stream.go
+1 -1
View File
@@ -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)
}
}
internal/stream/sub_stream_format.go
+11 -12
View File
@@ -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)
})