Repository: pion/interceptor
Pull request #64 (2021-09-11 02:37:08)

Use microseconds for TWCC API
We would expect Milliseconds, but then convert to Microseconds
internally.

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
> Merging [#64](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (6fb03e8) into [master](https://codecov.io/gh/pion/interceptor/commit/aade676d442c4a5b6182dd2041227b25d6444927?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) (aade676) will **decrease** coverage by `0.22%`.
> The diff coverage is `85.71%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/interceptor/pull/64/graphs/tree.svg?width=650&height=150&src=pr&token=FNH9xUG6AX&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)

```diff
@@            Coverage Diff             @@
##           master      #64      +/-   ##
==========================================
- Coverage   81.23%   81.01%   -0.23%     
==========================================
  Files          24       24              
  Lines         810      811       +1     
==========================================
- Hits          658      657       -1     
- Misses        101      102       +1     
- Partials       51       52       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `81.01% <85.71%> (-0.23%)` | :arrow_down: |
| wasm | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [pkg/twcc/twcc.go](https://codecov.io/gh/pion/interceptor/pull/64/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3R3Y2MvdHdjYy5nbw==) | `92.56% <75.00%> (ø)` | |
| [pkg/twcc/sender\_interceptor.go](https://codecov.io/gh/pion/interceptor/pull/64/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL3R3Y2Mvc2VuZGVyX2ludGVyY2VwdG9yLmdv) | `80.00% <100.00%> (+0.28%)` | :arrow_up: |
| [pkg/nack/send\_buffer.go](https://codecov.io/gh/pion/interceptor/pull/64/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-cGtnL25hY2svc2VuZF9idWZmZXIuZ28=) | `85.71% <0.00%> (-5.72%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Last update [aade676...6fb03e8](https://codecov.io/gh/pion/interceptor/pull/64?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).


Patch:
diff --git a/pkg/twcc/sender_interceptor.go b/pkg/twcc/sender_interceptor.go
index d74bc63..cfd68af 100644
--- a/pkg/twcc/sender_interceptor.go
+++ b/pkg/twcc/sender_interceptor.go
@@ -21,7 +21,8 @@ type SenderInterceptor struct {
 	wg    sync.WaitGroup
 	close chan struct{}
 
-	interval time.Duration
+	interval  time.Duration
+	startTime time.Time
 
 	recorder   *Recorder
 	packetChan chan packet
@@ -46,6 +47,7 @@ func NewSenderInterceptor(opts ...Option) (*SenderInterceptor, error) {
 		packetChan: make(chan packet),
 		close:      make(chan struct{}),
 		interval:   100 * time.Millisecond,
+		startTime:  time.Now(),
 	}
 
 	for _, opt := range opts {
@@ -117,7 +119,7 @@ func (s *SenderInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reade
 			s.packetChan <- packet{
 				hdr:            &p.Header,
 				sequenceNumber: tccExt.TransportSequence,
-				arrivalTime:    time.Now().UnixNano(),
+				arrivalTime:    time.Since(s.startTime).Microseconds(),
 				ssrc:           info.SSRC,
 			}
 		}
@@ -152,22 +154,19 @@ func (s *SenderInterceptor) loop(w interceptor.RTCPWriter) {
 	defer s.wg.Done()
 
 	ticker := time.NewTicker(s.interval)
-
 	for {
 		select {
 		case <-s.close:
+			ticker.Stop()
 			return
 		case p := <-s.packetChan:
-			s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime/1e6) // ns -> ms: divide by 1e6
+			s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime)
 
 		case <-ticker.C:
 			// build and send twcc
-			if s.recorder != nil {
-				pkts := s.recorder.BuildFeedbackPacket()
-				_, err := w.Write(pkts, nil)
-				if err != nil {
-					s.log.Error(err.Error())
-				}
+			pkts := s.recorder.BuildFeedbackPacket()
+			if _, err := w.Write(pkts, nil); err != nil {
+				s.log.Error(err.Error())
 			}
 		}
 	}
diff --git a/pkg/twcc/twcc.go b/pkg/twcc/twcc.go
index 175e410..fa89859 100644
--- a/pkg/twcc/twcc.go
+++ b/pkg/twcc/twcc.go
@@ -9,7 +9,7 @@ import (
 
 type pktInfo struct {
 	sequenceNumber uint32
-	timestamp      int64
+	arrivalTime    int64
 }
 
 // Recorder records incoming RTP packets and their delays and creates
@@ -36,14 +36,14 @@ func NewRecorder(senderSSRC uint32) *Recorder {
 }
 
 // Record marks a packet with mediaSSRC and a transport wide sequence number sequenceNumber as received at arrivalTime.
-func (r *Recorder) Record(mediaSSRC uint32, sequenceNumber uint16, arrivalTimeMS int64) {
+func (r *Recorder) Record(mediaSSRC uint32, sequenceNumber uint16, arrivalTime int64) {
 	r.mediaSSRC = mediaSSRC
 	if sequenceNumber < 0x0fff && (r.lastSequenceNumber&0xffff) > 0xf000 {
 		r.cycles += 1 << 16
 	}
 	r.receivedPackets = append(r.receivedPackets, pktInfo{
 		sequenceNumber: r.cycles | uint32(sequenceNumber),
-		timestamp:      arrivalTimeMS,
+		arrivalTime:    arrivalTime,
 	})
 	r.lastSequenceNumber = sequenceNumber
 }
@@ -51,23 +51,23 @@ func (r *Recorder) Record(mediaSSRC uint32, sequenceNumber uint16, arrivalTimeMS
 // BuildFeedbackPacket creates a new RTCP packet containing a TWCC feedback report.
 func (r *Recorder) BuildFeedbackPacket() []rtcp.Packet {
 	tlcc := newFeedback(r.senderSSRC, r.mediaSSRC, r.fbPktCnt)
-	if len(r.receivedPackets) == 0 {
+	if len(r.receivedPackets) < 2 {
 		return []rtcp.Packet{tlcc.getRTCP()}
 	}
 
 	sort.Slice(r.receivedPackets, func(i, j int) bool {
 		return r.receivedPackets[i].sequenceNumber < r.receivedPackets[j].sequenceNumber
 	})
-	tlcc.setBase(uint16(r.receivedPackets[0].sequenceNumber&0xffff), r.receivedPackets[0].timestamp*1000)
+	tlcc.setBase(uint16(r.receivedPackets[0].sequenceNumber&0xffff), r.receivedPackets[0].arrivalTime)
 
 	var pkts []rtcp.Packet
 	for _, pkt := range r.receivedPackets {
-		built := tlcc.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.timestamp*1000)
+		built := tlcc.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.arrivalTime)
 		if !built {
 			pkts = append(pkts, tlcc.getRTCP())
 			r.fbPktCnt++
 			tlcc = newFeedback(r.senderSSRC, r.mediaSSRC, r.fbPktCnt)
-			tlcc.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.timestamp*1000)
+			tlcc.addReceived(uint16(pkt.sequenceNumber&0xffff), pkt.arrivalTime)
 		}
 	}
 	r.receivedPackets = []pktInfo{}


Test patch:
diff --git a/pkg/twcc/twcc_test.go b/pkg/twcc/twcc_test.go
index 41f66ec..73d76a6 100644
--- a/pkg/twcc/twcc_test.go
+++ b/pkg/twcc/twcc_test.go
@@ -337,3 +337,144 @@ func Test_feedback(t *testing.T) {
 		}
 	})
 }
+
+func addRun(t *testing.T, r *Recorder, sequenceNumbers []uint16, arrivalTimes []int64) {
+	assert.Equal(t, len(sequenceNumbers), len(arrivalTimes))
+
+	for i := range sequenceNumbers {
+		r.Record(5000, sequenceNumbers[i], arrivalTimes[i])
+	}
+}
+
+const (
+	scaleFactorReferenceTime = 64000
+)
+
+func increaseTime(arrivalTime *int64, increaseAmount int64) int64 {
+	*arrivalTime += increaseAmount
+	return *arrivalTime
+}
+
+func marshalAll(t *testing.T, pkts []rtcp.Packet) {
+	for _, pkt := range pkts {
+		_, err := pkt.Marshal()
+		assert.NoError(t, err)
+	}
+}
+
+func TestBuildFeedbackPacket(t *testing.T) {
+	r := NewRecorder(5000)
+
+	arrivalTime := int64(scaleFactorReferenceTime)
+	addRun(t, r, []uint16{0, 1, 2, 3, 4, 5, 6, 7}, []int64{
+		scaleFactorReferenceTime,
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor*256),
+	})
+
+	rtcpPackets := r.BuildFeedbackPacket()
+	assert.Equal(t, 1, len(rtcpPackets))
+
+	assert.Equal(t, &rtcp.TransportLayerCC{
+		Header: rtcp.Header{
+			Count:   rtcp.FormatTCC,
+			Type:    rtcp.TypeTransportSpecificFeedback,
+			Padding: true,
+			Length:  8,
+		},
+		SenderSSRC:         5000,
+		MediaSSRC:          5000,
+		BaseSequenceNumber: 0,
+		ReferenceTime:      1,
+		FbPktCount:         0,
+		PacketStatusCount:  8,
+		PacketChunks: []rtcp.PacketStatusChunk{
+			&rtcp.RunLengthChunk{
+				Type:               rtcp.TypeTCCRunLengthChunk,
+				PacketStatusSymbol: rtcp.TypeTCCPacketReceivedSmallDelta,
+				RunLength:          7,
+			},
+			&rtcp.RunLengthChunk{
+				Type:               rtcp.TypeTCCRunLengthChunk,
+				PacketStatusSymbol: rtcp.TypeTCCPacketReceivedLargeDelta,
+				RunLength:          1,
+			},
+		},
+		RecvDeltas: []*rtcp.RecvDelta{
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedLargeDelta, Delta: 256},
+		},
+	}, rtcpPackets[0].(*rtcp.TransportLayerCC))
+	marshalAll(t, rtcpPackets)
+}
+
+func TestBuildFeedbackPacket_Rolling(t *testing.T) {
+	r := NewRecorder(5000)
+
+	arrivalTime := int64(scaleFactorReferenceTime)
+	addRun(t, r, []uint16{0}, []int64{
+		arrivalTime,
+	})
+
+	rtcpPackets := r.BuildFeedbackPacket()
+	assert.Equal(t, 1, len(rtcpPackets)) // Empty TWCC
+
+	addRun(t, r, []uint16{4, 5, 6, 7}, []int64{
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+		increaseTime(&arrivalTime, rtcp.TypeTCCDeltaScaleFactor),
+	})
+
+	rtcpPackets = r.BuildFeedbackPacket()
+	assert.Equal(t, 1, len(rtcpPackets))
+
+	assert.Equal(t, &rtcp.TransportLayerCC{
+		Header: rtcp.Header{
+			Count:   rtcp.FormatTCC,
+			Type:    rtcp.TypeTransportSpecificFeedback,
+			Padding: true,
+			Length:  6,
+		},
+		SenderSSRC:         5000,
+		MediaSSRC:          5000,
+		BaseSequenceNumber: 0,
+		ReferenceTime:      1,
+		FbPktCount:         0,
+		PacketStatusCount:  8,
+		PacketChunks: []rtcp.PacketStatusChunk{
+			&rtcp.StatusVectorChunk{
+				Type:       rtcp.TypeTCCRunLengthChunk,
+				SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit,
+				SymbolList: []uint16{
+					rtcp.TypeTCCPacketReceivedSmallDelta,
+					rtcp.TypeTCCPacketNotReceived,
+					rtcp.TypeTCCPacketNotReceived,
+					rtcp.TypeTCCPacketNotReceived,
+					rtcp.TypeTCCPacketReceivedSmallDelta,
+					rtcp.TypeTCCPacketReceivedSmallDelta,
+					rtcp.TypeTCCPacketReceivedSmallDelta,
+				},
+			},
+		},
+		RecvDeltas: []*rtcp.RecvDelta{
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 0},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+			{Type: rtcp.TypeTCCPacketReceivedSmallDelta, Delta: 1},
+		},
+	}, rtcpPackets[0].(*rtcp.TransportLayerCC))
+	marshalAll(t, rtcpPackets)
+}