Add RFC8888 feedback to GCC

Enables GCC to work with RFC8888 instead of TWCC.

Author: mengelbart

internal/cc/acknowledgment.go

@@ -3,21 +3,25 @@ package cc
 import (
 	"fmt"
 	"time"
+
+	"github.com/pion/rtcp"
 )
 
 // Acknowledgment holds information about a packet and if/when it has been
 // sent/received.
 type Acknowledgment struct {
-	TLCC      uint16
-	Size      int
-	Departure time.Time
-	Arrival   time.Time
-	RTT       time.Duration
+	SequenceNumber uint16 // Either RTP SequenceNumber or TWCC
+	SSRC           uint32
+	Size           int
+	Departure      time.Time
+	Arrival        time.Time
+	RTT            time.Duration
+	ECN            rtcp.ECN
 }
 
 func (a Acknowledgment) String() string {
 	s := "ACK:\n"
-	s += fmt.Sprintf("\tTLCC:\t%v\n", a.TLCC)
+	s += fmt.Sprintf("\tTLCC:\t%v\n", a.SequenceNumber)
 	s += fmt.Sprintf("\tSIZE:\t%v\n", a.Size)
 	s += fmt.Sprintf("\tDEPARTURE:\t%v\n", int64(float64(a.Departure.UnixNano())/1e+6))
 	s += fmt.Sprintf("\tARRIVAL:\t%v\n", int64(float64(a.Arrival.UnixNano())/1e+6))

internal/cc/feedback_adapter.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/ntp"
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
 )
@@ -16,9 +17,8 @@ import (
 const TwccExtensionAttributesKey = iota
 
 var (
-	errMissingTWCCExtensionID = errors.New("missing transport layer cc header extension id")
-	errMissingTWCCExtension   = errors.New("missing transport layer cc header extension")
-	errInvalidFeedback        = errors.New("invalid feedback")
+	errMissingTWCCExtension = errors.New("missing transport layer cc header extension")
+	errInvalidFeedback      = errors.New("invalid feedback")
 )
 
 // FeedbackAdapter converts incoming RTCP Packets (TWCC and RFC8888) into Acknowledgments.
@@ -33,15 +33,24 @@ func NewFeedbackAdapter() *FeedbackAdapter {
 	return &FeedbackAdapter{history: newFeedbackHistory(250)}
 }
 
-// OnSent records that and when an outgoing packet was sent for later mapping to
-// acknowledgments
-func (f *FeedbackAdapter) OnSent(ts time.Time, header *rtp.Header, size int, attributes interceptor.Attributes) error {
-	hdrExtensionID := attributes.Get(TwccExtensionAttributesKey)
-	id, ok := hdrExtensionID.(uint8)
-	if !ok || hdrExtensionID == 0 {
-		return errMissingTWCCExtensionID
-	}
-	sequenceNumber := header.GetExtension(id)
+func (f *FeedbackAdapter) onSentRFC8888(ts time.Time, header *rtp.Header, size int) error {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.history.add(Acknowledgment{
+		SequenceNumber: header.SequenceNumber,
+		SSRC:           header.SSRC,
+		Size:           size,
+		Departure:      ts,
+		Arrival:        time.Time{},
+		RTT:            0,
+		ECN:            0,
+	})
+	return nil
+}
+
+func (f *FeedbackAdapter) onSentTWCC(ts time.Time, extID uint8, header *rtp.Header, size int) error {
+	sequenceNumber := header.GetExtension(extID)
 	var tccExt rtp.TransportCCExtension
 	err := tccExt.Unmarshal(sequenceNumber)
 	if err != nil {
@@ -51,23 +60,41 @@ func (f *FeedbackAdapter) OnSent(ts time.Time, header *rtp.Header, size int, att
 	f.lock.Lock()
 	defer f.lock.Unlock()
 	f.history.add(Acknowledgment{
-		TLCC:      tccExt.TransportSequence,
-		Size:      header.MarshalSize() + size,
-		Departure: ts,
-		Arrival:   time.Time{},
-		RTT:       0,
+		SequenceNumber: tccExt.TransportSequence,
+		SSRC:           0,
+		Size:           header.MarshalSize() + size,
+		Departure:      ts,
+		Arrival:        time.Time{},
+		RTT:            0,
+		ECN:            0,
 	})
 	return nil
 }
 
+// OnSent records that and when an outgoing packet was sent for later mapping to
+// acknowledgments
+func (f *FeedbackAdapter) OnSent(ts time.Time, header *rtp.Header, size int, attributes interceptor.Attributes) error {
+	hdrExtensionID := attributes.Get(TwccExtensionAttributesKey)
+	id, ok := hdrExtensionID.(uint8)
+	if ok && hdrExtensionID != 0 {
+		return f.onSentTWCC(ts, id, header, size)
+	}
+
+	return f.onSentRFC8888(ts, header, size)
+}
+
 func (f *FeedbackAdapter) unpackRunLengthChunk(ts time.Time, start uint16, refTime time.Time, chunk *rtcp.RunLengthChunk, deltas []*rtcp.RecvDelta) (consumedDeltas int, nextRef time.Time, acks []Acknowledgment, err error) {
 	result := make([]Acknowledgment, chunk.RunLength)
 	deltaIndex := 0
 
 	end := start + chunk.RunLength
 	resultIndex := 0
 	for i := start; i != end; i++ {
-		if ack, ok := f.history.get(i); ok {
+		key := feedbackHistoryKey{
+			ssrc:           0,
+			sequenceNumber: i,
+		}
+		if ack, ok := f.history.get(key); ok {
 			if chunk.PacketStatusSymbol != rtcp.TypeTCCPacketNotReceived {
 				if len(deltas)-1 < deltaIndex {
 					return deltaIndex, refTime, result, errInvalidFeedback
@@ -89,7 +116,11 @@ func (f *FeedbackAdapter) unpackStatusVectorChunk(ts time.Time, start uint16, re
 	deltaIndex := 0
 	resultIndex := 0
 	for i, symbol := range chunk.SymbolList {
-		if ack, ok := f.history.get(start + uint16(i)); ok {
+		key := feedbackHistoryKey{
+			ssrc:           0,
+			sequenceNumber: start + uint16(i),
+		}
+		if ack, ok := f.history.get(key); ok {
 			if symbol != rtcp.TypeTCCPacketNotReceived {
 				if len(deltas)-1 < deltaIndex {
 					return deltaIndex, refTime, result, errInvalidFeedback
@@ -146,21 +177,55 @@ func (f *FeedbackAdapter) OnTransportCCFeedback(ts time.Time, feedback *rtcp.Tra
 	return result, nil
 }
 
+// OnRFC8888Feedback converts incoming Congestion Control Feedback RTCP packet
+// to Acknowledgments.
+func (f *FeedbackAdapter) OnRFC8888Feedback(ts time.Time, feedback *rtcp.CCFeedbackReport) []Acknowledgment {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	result := []Acknowledgment{}
+	referenceTime := ntp.ToTime(uint64(feedback.ReportTimestamp) << 16)
+	for _, rb := range feedback.ReportBlocks {
+		for i, mb := range rb.MetricBlocks {
+			sequenceNumber := rb.BeginSequence + uint16(i)
+			key := feedbackHistoryKey{
+				ssrc:           rb.MediaSSRC,
+				sequenceNumber: sequenceNumber,
+			}
+			if ack, ok := f.history.get(key); ok {
+				if mb.Received {
+					delta := time.Duration((float64(mb.ArrivalTimeOffset) / 1024.0) * float64(time.Second))
+					ack.Arrival = referenceTime.Add(-delta)
+					ack.RTT = ts.Sub(ack.Departure)
+					ack.ECN = mb.ECN
+				}
+				result = append(result, ack)
+			}
+		}
+	}
+	return result
+}
+
+type feedbackHistoryKey struct {
+	ssrc           uint32
+	sequenceNumber uint16
+}
+
 type feedbackHistory struct {
 	size      int
 	evictList *list.List
-	items     map[uint16]*list.Element
+	items     map[feedbackHistoryKey]*list.Element
 }
 
 func newFeedbackHistory(size int) *feedbackHistory {
 	return &feedbackHistory{
 		size:      size,
 		evictList: list.New(),
-		items:     make(map[uint16]*list.Element),
+		items:     make(map[feedbackHistoryKey]*list.Element),
 	}
 }
 
-func (f *feedbackHistory) get(key uint16) (Acknowledgment, bool) {
+func (f *feedbackHistory) get(key feedbackHistoryKey) (Acknowledgment, bool) {
 	ent, ok := f.items[key]
 	if ok {
 		if ack, ok := ent.Value.(Acknowledgment); ok {
@@ -171,15 +236,19 @@ func (f *feedbackHistory) get(key uint16) (Acknowledgment, bool) {
 }
 
 func (f *feedbackHistory) add(ack Acknowledgment) {
+	key := feedbackHistoryKey{
+		ssrc:           ack.SSRC,
+		sequenceNumber: ack.SequenceNumber,
+	}
 	// Check for existing
-	if ent, ok := f.items[ack.TLCC]; ok {
+	if ent, ok := f.items[key]; ok {
 		f.evictList.MoveToFront(ent)
 		ent.Value = ack
 		return
 	}
 	// Add new
 	ent := f.evictList.PushFront(ack)
-	f.items[ack.TLCC] = ent
+	f.items[key] = ent
 	// Evict if necessary
 	if f.evictList.Len() > f.size {
 		f.removeOldest()
@@ -190,7 +259,11 @@ func (f *feedbackHistory) removeOldest() {
 	if ent := f.evictList.Back(); ent != nil {
 		f.evictList.Remove(ent)
 		if ack, ok := ent.Value.(Acknowledgment); ok {
-			delete(f.items, ack.TLCC)
+			key := feedbackHistoryKey{
+				ssrc:           ack.SSRC,
+				sequenceNumber: ack.SequenceNumber,
+			}
+			delete(f.items, key)
 		}
 	}
 }