Use packet receive time to calculate PTS. (#755)

The pipeline clock is anchored to when the first track of the
synchronizer group is innitialised. And all packets should have PTS with
reference to that. That clock is set to when the first packet of any
track in the synchronizer group is read off the wire (non-padding
packet). So, (packet_receive_time - pipeline_start_time) should be used
to get the correct PTS.

A burst of packets is an interesting case.
- smooth flow of packets
- 3 seconds of gap
- a burst of three seconds packets

In this case, the following will happen
- old packet drop threshold will kick in and drop some of those packets.
- estimated PTS will be higher by upto (3 seconds -
  old_packet_treshold) for the burst. But, they should still be
  acceptable and pushed with proper PTS, i. e. PTS calculated using RTP
  timestamp diff should be correct.

Have to ensure that different thresholds do not render the pipeline drop
tracks.

Author: boks1971

pkg/synchronizer/track.go

@@ -65,9 +65,9 @@ type TrackSynchronizer struct {
 	oldPacketThreshold                time.Duration
 
 	// timing info
-	initTime         time.Time
-	startTime        time.Time // time first packet was pushed
-	startRTP         uint32    // RTP timestamp of PTS 0
+	startTime        time.Time // time at initialization --> this should be when first packet is received
+	startRTP         uint32    // RTP timestamp of first packet
+	firstTime        time.Time // time at which first packet was pushed
 	lastTS           uint32    // previous RTP timestamp
 	lastTime         time.Time
 	lastPTS          time.Duration // previous presentation timestamp
@@ -141,7 +141,7 @@ func (t *TrackSynchronizer) Initialize(pkt *rtp.Packet) {
 	t.desiredPTSOffset = t.currentPTSOffset
 	t.basePTSOffset = t.desiredPTSOffset
 
-	t.initTime = now
+	t.startTime = now
 	t.startRTP = pkt.Timestamp
 	t.lastPTS = 0
 	t.lastPTSAdjusted = t.currentPTSOffset
@@ -174,17 +174,19 @@ func (t *TrackSynchronizer) getPTSWithoutRebase(pkt jitter.ExtPacket) (time.Dura
 	defer t.Unlock()
 
 	now := mono.Now()
-	if t.startTime.IsZero() {
-		if t.preJitterBufferReceiveTimeEnabled {
-			t.startTime = pkt.ReceivedAt
-		} else {
-			t.startTime = now
-		}
+	var pktReceiveTime time.Time
+	if t.preJitterBufferReceiveTimeEnabled {
+		pktReceiveTime = pkt.ReceivedAt
+	} else {
+		pktReceiveTime = now
+	}
+	if t.firstTime.IsZero() {
+		t.firstTime = now
 		t.logger.Infow(
 			"starting track synchronizer",
 			"state", t,
 			"pktReceiveTime", pkt.ReceivedAt,
-			"startDelay", t.startTime.Sub(pkt.ReceivedAt),
+			"startDelay", t.firstTime.Sub(pkt.ReceivedAt),
 		)
 	}
 
@@ -215,7 +217,7 @@ func (t *TrackSynchronizer) getPTSWithoutRebase(pkt jitter.ExtPacket) (time.Dura
 		}
 	}
 
-	estimatedPTS := now.Sub(t.startTime)
+	estimatedPTS := pktReceiveTime.Sub(t.startTime)
 
 	var pts time.Duration
 	if t.lastPTS == 0 {
@@ -292,11 +294,14 @@ func (t *TrackSynchronizer) getPTSWithRebase(pkt jitter.ExtPacket) (time.Duratio
 	t.Lock()
 	defer t.Unlock()
 
-	if t.startTime.IsZero() {
-		t.startTime = pkt.ReceivedAt
+	now := mono.Now()
+	if t.firstTime.IsZero() {
+		t.firstTime = now
 		t.logger.Infow(
 			"starting track synchronizer",
 			"state", t,
+			"pktReceiveTime", pkt.ReceivedAt,
+			"startDelay", t.firstTime.Sub(pkt.ReceivedAt),
 		)
 	}
 
@@ -329,15 +334,14 @@ func (t *TrackSynchronizer) getPTSWithRebase(pkt jitter.ExtPacket) (time.Duratio
 			"dropping old packet",
 			"currentTS", ts,
 			"receivedAt", pkt.ReceivedAt,
-			"now", mono.Now(),
-			"age", mono.Now().Sub(pkt.ReceivedAt),
+			"now", now,
+			"age", now.Sub(pkt.ReceivedAt),
 			"state", t,
 		)
 		return 0, ErrPacketTooOld
 	}
 
-	now := mono.Now()
-	estimatedPTS := max(time.Nanosecond, now.Sub(t.startTime))
+	estimatedPTS := pkt.ReceivedAt.Sub(t.startTime)
 
 	var pts time.Duration
 	if t.lastPTS == 0 {
@@ -716,9 +720,9 @@ func (t *TrackSynchronizer) MarshalLogObject(e zapcore.ObjectEncoder) error {
 		return nil
 	}
 
-	e.AddTime("initTime", t.initTime)
 	e.AddTime("startTime", t.startTime)
 	e.AddUint32("startRTP", t.startRTP)
+	e.AddTime("firstTime", t.firstTime)
 	e.AddUint32("lastTS", t.lastTS)
 	e.AddTime("lastTime", t.lastTime)
 	e.AddDuration("lastPTS", t.lastPTS)

pkg/synchronizer/track_test.go

@@ -53,7 +53,6 @@ func newTSForTests(tc *testing.T, clockRate uint32, kind webrtc.RTPCodecType) *T
 	}
 	// set a stable startTime well in the past to make time.Since(startTime) > 0
 	t.startTime = time.Now().Add(-150 * time.Millisecond)
-	t.initTime = t.startTime
 	// pick an arbitrary RTP base
 	t.startRTP = 1000
 	return t
@@ -144,7 +143,7 @@ func TestGetPTSWithRebase_PropelsForward(t *testing.T) {
 	ts1 := ts.startRTP + rtp500ms
 	adj1, err := ts.getPTSWithRebase(jitter.ExtPacket{
 		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts1}},
-		ReceivedAt: time.Time{}, // not used for estimatedPTS here
+		ReceivedAt: mono.Now(),
 	})
 	require.NoError(t, err)
 	require.InDelta(t, 500*time.Millisecond, adj1, float64(20*time.Millisecond))
@@ -159,7 +158,7 @@ func TestGetPTSWithRebase_PropelsForward(t *testing.T) {
 
 	adj2, err := ts.getPTSWithRebase(jitter.ExtPacket{
 		Packet:     &rtp.Packet{Header: rtp.Header{Timestamp: ts2}},
-		ReceivedAt: time.Time{},
+		ReceivedAt: mono.Now(),
 	})
 	require.NoError(t, err)
 	require.Equal(t, want, adj2)