From b5f2f83278ef919bcc2ec5f706540708a5afc48a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 9 Sep 2023 17:33:26 +0530 Subject: [PATCH] Fix time stamp adjustment when starting with dummy packets. (#2053) * Fix time stamp adjustment when starting with dmummy packets. - Populated extended values in ExtPacket on dummy packet. - Have to pass reference time stamp offset to first packet time adjustment. * display participant version info --- pkg/rtc/participant.go | 9 +++++++ pkg/rtc/participant_signal.go | 2 +- pkg/sfu/buffer/rtpstats.go | 6 ++--- pkg/sfu/downtrack.go | 13 +++++++--- pkg/sfu/forwarder.go | 46 ++++++++++++++++++++++------------- 5 files changed, 51 insertions(+), 25 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 6a7ccf919..d84ed7cb5 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -16,6 +16,7 @@ package rtc import ( "context" + "fmt" "io" "os" "strconv" @@ -69,12 +70,20 @@ type downTrackState struct { downTrack sfu.DownTrackState } +// --------------------------------------------------------------- + type participantUpdateInfo struct { version uint32 state livekit.ParticipantInfo_State updatedAt time.Time } +func (p participantUpdateInfo) String() string { + return fmt.Sprintf("version: %d, state: %s, updatedAt: %s", p.version, p.state.String(), p.updatedAt.String()) +} + +// --------------------------------------------------------------- + type ParticipantParams struct { Identity livekit.ParticipantIdentity Name livekit.ParticipantName diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 0f0d3c6e7..5214df47e 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -95,7 +95,7 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit. // this is a message delivered out of order, a more recent version of the message had already been // sent. if pi.Version < lastVersion.version { - p.params.Logger.Debugw("skipping outdated participant update", "version", pi.Version, "lastVersion", lastVersion) + p.params.Logger.Debugw("skipping outdated participant update", "otherParticipant", pi.Identity, "otherPID", pi.Sid, "version", pi.Version, "lastVersion", lastVersion) isValid = false } } diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 407e5b341..9fb57c6c5 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -863,13 +863,11 @@ func (r *RTPStats) GetRtt() uint32 { return r.rtt } -func (r *RTPStats) MaybeAdjustFirstPacketTime(srData *RTCPSenderReportData) { +func (r *RTPStats) MaybeAdjustFirstPacketTime(ets uint64) { r.lock.Lock() defer r.lock.Unlock() - if srData != nil { - r.maybeAdjustFirstPacketTime(srData.RTPTimestampExt) - } + r.maybeAdjustFirstPacketTime(ets) } func (r *RTPStats) maybeAdjustFirstPacketTime(ets uint64) { diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 39c961420..6f575500c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1741,7 +1741,6 @@ func (d *DownTrack) sendPaddingOnMute() { // let uptrack have chance to send packet before we send padding time.Sleep(waitBeforeSendPaddingOnMute) - d.params.Logger.Debugw("sending padding on mute") if d.kind == webrtc.RTPCodecTypeVideo { d.sendPaddingOnMuteForVideo() } else if d.mime == "audio/opus" { @@ -1756,6 +1755,9 @@ func (d *DownTrack) sendPaddingOnMuteForVideo() { if d.rtpStats.IsActive() || d.IsClosed() { return } + if i == 0 { + d.params.Logger.Debugw("sending padding on mute") + } d.WritePaddingRTP(20, true, true) time.Sleep(paddingOnMuteInterval) } @@ -1765,10 +1767,15 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { frameRate := uint32(50) frameDuration := time.Duration(1000/frameRate) * time.Millisecond numFrames := frameRate * uint32(maxPaddingOnMuteDuration/time.Second) + first := true for { if d.rtpStats.IsActive() || d.IsClosed() || numFrames <= 0 { return } + if first { + first = false + d.params.Logger.Debugw("sending padding on mute") + } snts, _, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1) if err != nil { d.params.Logger.Warnw("could not get SN/TS for blank frame", err) @@ -1812,8 +1819,8 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { } func (d *DownTrack) HandleRTCPSenderReportData(_payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error { - if layer == d.forwarder.GetReferenceLayerSpatial() { - d.rtpStats.MaybeAdjustFirstPacketTime(srData) + if layer == d.forwarder.GetReferenceLayerSpatial() && srData != nil { + d.rtpStats.MaybeAdjustFirstPacketTime(srData.RTPTimestampExt + d.forwarder.GetReferenceTimestampOffset()) } return nil } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 49c56bc2f..8fc170ab2 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -532,6 +532,13 @@ func (f *Forwarder) GetReferenceLayerSpatial() int32 { return f.referenceLayerSpatial } +func (f *Forwarder) GetReferenceTimestampOffset() uint64 { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.refTSOffset +} + func (f *Forwarder) isDeficientLocked() bool { return f.lastAllocation.IsDeficient } @@ -1516,22 +1523,23 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e if err == nil { extExpectedTS = tsExt } else { - rtpDiff := uint64(0) - if !f.preStartTime.IsZero() && f.refTSOffset == 0 { + if !f.preStartTime.IsZero() { timeSinceFirst := time.Since(f.preStartTime) - rtpDiff = uint64(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9) - f.refTSOffset = f.extFirstTS + rtpDiff - extRefTS - f.logger.Infow( - "calculating refTSOffset", - "preStartTime", f.preStartTime.String(), - "extFirstTS", f.extFirstTS, - "timeSinceFirst", timeSinceFirst, - "rtpDiff", rtpDiff, - "extRefTS", extRefTS, - "refTSOffset", f.refTSOffset, - ) + rtpDiff := uint64(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9) + extExpectedTS = f.extFirstTS + rtpDiff + if f.refTSOffset == 0 { + f.refTSOffset = extExpectedTS - extRefTS + f.logger.Infow( + "calculating refTSOffset", + "preStartTime", f.preStartTime.String(), + "extFirstTS", f.extFirstTS, + "timeSinceFirst", timeSinceFirst, + "rtpDiff", rtpDiff, + "extRefTS", extRefTS, + "refTSOffset", f.refTSOffset, + ) + } } - extExpectedTS += rtpDiff } } extRefTS += f.refTSOffset @@ -1746,17 +1754,21 @@ func (f *Forwarder) maybeStart() { f.started = true f.preStartTime = time.Now() + sequenceNumber := uint16(rand.Intn(1<<14)) + uint16(1<<15) // a random number in third quartile of sequence number space + timestamp := uint32(rand.Intn(1<<30)) + uint32(1<<31) // a random number in third quartile of timestamp space extPkt := &buffer.ExtPacket{ Packet: &rtp.Packet{ Header: rtp.Header{ - SequenceNumber: uint16(rand.Intn(1<<14)) + uint16(1<<15), // a random number in third quartile of sequence number space - Timestamp: uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of timestamp space + SequenceNumber: sequenceNumber, + Timestamp: timestamp, }, }, + ExtSequenceNumber: uint64(sequenceNumber), + ExtTimestamp: uint64(timestamp), } f.rtpMunger.SetLastSnTs(extPkt) - f.extFirstTS = uint64(extPkt.Packet.Timestamp) + f.extFirstTS = uint64(timestamp) f.logger.Debugw( "starting with dummy forwarding", "sequenceNumber", extPkt.Packet.SequenceNumber,