Feat/nada interceptor integration

AUTHORS.txt

@@ -15,5 +15,6 @@ boks1971 <[email protected]>
 David Zhao <[email protected]>
 Jonathan Müller <[email protected]>
 Kevin Caffrey <[email protected]>
+Kevin Wang <[email protected]>
 Mathis Engelbart <[email protected]>
 Sean DuBois <[email protected]>

pkg/nada/README.md

@@ -0,0 +1,5 @@
+# [RFC 8698] NADA: A Unified Congestion Control Scheme for Real-Time Media
+
+Notes:
+
+* The receiver in this implementation assumes a monotonically ordered sequence of packets.

pkg/nada/bandwidth_estimator.go

@@ -0,0 +1,121 @@
+package nada
+
+import (
+	"math"
+	"time"
+
+	"github.com/pion/interceptor"
+	"github.com/pion/interceptor/internal/cc"
+	"github.com/pion/interceptor/internal/ntp"
+	"github.com/pion/rtcp"
+	"github.com/pion/rtp"
+)
+
+type NadaBandwidthEstimator struct {
+	*cc.FeedbackAdapter
+	*Sender
+	*Receiver
+
+	onTargetBitrateChangeFunc func(int)
+	lastTargetRate            int
+}
+
+func NewBandwidthEstimator() *NadaBandwidthEstimator {
+	now := time.Now()
+	return &NadaBandwidthEstimator{
+		FeedbackAdapter: cc.NewFeedbackAdapter(),
+		Sender:          NewSender(now, DefaultConfig()),
+		Receiver:        NewReceiver(now, DefaultConfig()),
+	}
+}
+
+func (e *NadaBandwidthEstimator) AddStream(_ *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
+	return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
+		now := time.Now()
+		if err := e.OnSent(now, header, len(payload), attributes); err != nil {
+			return 0, err
+		}
+		return writer.Write(header, payload, attributes)
+	})
+}
+
+func (e *NadaBandwidthEstimator) WriteRTCP(pkts []rtcp.Packet, _ interceptor.Attributes) error {
+	now := time.Now()
+	var acks []cc.Acknowledgment
+	for _, pkt := range pkts {
+		var feedbackSentTime time.Time
+		switch fb := pkt.(type) {
+		case *rtcp.TransportLayerCC:
+			newAcks, err := e.OnTransportCCFeedback(now, fb)
+			if err != nil {
+				return err
+			}
+			for i, ack := range acks {
+				if i == 0 {
+					feedbackSentTime = ack.Arrival
+					continue
+				}
+				if ack.Arrival.After(feedbackSentTime) {
+					feedbackSentTime = ack.Arrival
+				}
+			}
+			acks = append(acks, newAcks...)
+		case *rtcp.CCFeedbackReport:
+			acks = e.OnRFC8888Feedback(now, fb)
+			feedbackSentTime = ntp.ToTime(uint64(fb.ReportTimestamp) << 16)
+		default:
+			continue
+		}
+
+		feedbackMinRTT := time.Duration(math.MaxInt)
+		for _, ack := range acks {
+			if ack.Arrival.IsZero() {
+				continue
+			}
+			pendingTime := feedbackSentTime.Sub(ack.Arrival)
+			rtt := now.Sub(ack.Departure) - pendingTime
+			feedbackMinRTT = time.Duration(minInt(int(rtt), int(feedbackMinRTT)))
+		}
+		if feedbackMinRTT < math.MaxInt {
+			e.UpdateEstimatedRoundTripTime(feedbackMinRTT)
+		}
+	}
+
+	for _, ack := range acks {
+		e.OnReceiveMediaPacket(ack.Arrival, ack.Departure, ack.SequenceNumber, ack.ECN == rtcp.ECNCE, 8*Bits(ack.Size))
+	}
+
+	e.OnReceiveFeedbackReport(now, e.BuildFeedbackReport())
+
+	target := e.GetTargetBitrate()
+	if target != e.lastTargetRate {
+		e.onTargetBitrateChangeFunc(target)
+		e.lastTargetRate = target
+	}
+
+	return nil
+}
+
+func (e *NadaBandwidthEstimator) GetTargetBitrate() int {
+	// TODO: What is the buffer len parameter of GetTargetBitrate?
+	return int(e.GetTargetRate(0))
+}
+
+func (e *NadaBandwidthEstimator) OnTargetBitrateChange(f func(bitrate int)) {
+	e.onTargetBitrateChangeFunc = f
+}
+
+func (e *NadaBandwidthEstimator) GetStats() map[string]interface{} {
+	panic("not implemented")
+}
+
+func (e *NadaBandwidthEstimator) Close() error {
+	panic("not implemented")
+}
+
+func minInt(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}

pkg/nada/config.go

@@ -0,0 +1,107 @@
+package nada
+
+import "time"
+
+// Bits represents a unit of one bit.
+type Bits uint32
+
+// BitsPerSecond represents a unit of one bit per second.
+type BitsPerSecond float64
+
+const (
+	// Kbps represents 1 kbps.
+	Kbps = BitsPerSecond(1_000)
+	// Mbps represents 1 Mbps.
+	Mbps = BitsPerSecond(1_000_000)
+)
+
+// Config represents the configuration of a NADA bandwidth estimator.
+type Config struct {
+	// Weight of priority of the flow
+	Priority float64
+	// Minimum rate of the application supported by the media encoder
+	MinimumRate BitsPerSecond // RMIN
+	// Maximum rate of the application supported by media encoder
+	MaximumRate BitsPerSecond // RMAX
+	// Reference congestion level
+	ReferenceCongestionLevel time.Duration // XREF
+	// Scaling parameter for gradual rate update calculation
+	Kappa float64
+	// Scaling parameter for gradual rate update calculation
+	Eta float64
+	// Upper bound of RTT in gradual rate update calculation
+	Tau time.Duration
+	// Target feedback interval
+	Delta time.Duration
+
+	// Observation window in time for calculating packet summary statistics at receiver
+	LogWindow time.Duration // LOGWIN
+	// Threshold for determining queuing delay build up at receiver
+	QueueingDelayThreshold time.Duration
+	// Bound on filtering delay
+	FilteringDelay time.Duration // DFILT
+	// Upper bound on rate increase ratio for accelerated ramp-up
+	GammaMax float64
+	// Upper bound on self-inflicted queueing delay during ramp up
+	QueueBound time.Duration // QBOUND
+
+	// Multiplier for self-scaling the expiration threshold of the last observed loss
+	// (loss_exp) based on measured average loss interval (loss_int)
+	LossMultiplier float64 // MULTILOSS
+	// Delay threshold for invoking non-linear warping
+	DelayThreshold time.Duration // QTH
+	// Scaling parameter in the exponent of non-linear warping
+	Lambda float64
+
+	// Reference packet loss ratio
+	ReferencePacketLossRatio float64 // PLRREF
+	// Reference packet marking ratio
+	ReferencePacketMarkingRatio float64 // PMRREF
+	// Reference delay penalty for loss when lacket loss ratio is at least PLRREF
+	ReferenceDelayLoss time.Duration // DLOSS
+	// Reference delay penalty for ECN marking when packet marking is at PMRREF
+	ReferenceDelayMarking time.Duration // DMARK
+
+	// Frame rate of incoming video
+	FrameRate float64 // FRAMERATE
+	// Scaling parameter for modulating outgoing sending rate
+	BetaSending float64
+	// Scaling parameter for modulating video encoder target rate
+	BetaVideoEncoder float64
+	// Smoothing factor in exponential smoothing of packet loss and marking rate
+	Alpha float64
+}
+
+// DefaultConfig returns the default configuration recommended by the specification.
+func DefaultConfig() Config {
+	return Config{
+		Priority:                 1.0,
+		MinimumRate:              150 * Kbps,
+		MaximumRate:              1500 * Kbps,
+		ReferenceCongestionLevel: 10 * time.Millisecond,
+		Kappa:                    0.5,
+		Eta:                      2.0,
+		Tau:                      500 * time.Millisecond,
+		Delta:                    100 * time.Millisecond,
+
+		LogWindow:              500 * time.Millisecond,
+		QueueingDelayThreshold: 10 * time.Millisecond,
+		FilteringDelay:         120 * time.Millisecond,
+		GammaMax:               0.5,
+		QueueBound:             50 * time.Millisecond,
+
+		LossMultiplier: 7.0,
+		DelayThreshold: 50 * time.Millisecond,
+		Lambda:         0.5,
+
+		ReferencePacketLossRatio:    0.01,
+		ReferencePacketMarkingRatio: 0.01,
+		ReferenceDelayLoss:          10 * time.Millisecond,
+		ReferenceDelayMarking:       2 * time.Millisecond,
+
+		FrameRate:        30.0,
+		BetaSending:      0.1,
+		BetaVideoEncoder: 0.1,
+		Alpha:            0.1,
+	}
+}

pkg/nada/ecn/ecn.go

@@ -0,0 +1,23 @@
+// Package ecn provides ExplicitCongestionNotification (ECN) support.
+package ecn
+
+import (
+	"errors"
+	"syscall"
+)
+
+var errNoECN = errors.New("no ECN control message")
+
+// CheckExplicitCongestionNotification checks if the given oob data includes an ECN bit set.
+func CheckExplicitCongestionNotification(oob []byte) (uint8, error) {
+	ctrlMsgs, err := syscall.ParseSocketControlMessage(oob)
+	if err != nil {
+		return 0, err
+	}
+	for _, ctrlMsg := range ctrlMsgs {
+		if ctrlMsg.Header.Type == syscall.IP_TOS {
+			return (ctrlMsg.Data[0] & 0x3), nil
+		}
+	}
+	return 0, errNoECN
+}

pkg/nada/ecn/ecn_darwin.go

@@ -0,0 +1,11 @@
+package ecn
+
+import (
+	"net"
+)
+
+// EnableExplicitCongestionNotification enables ECN on the given connection.
+func EnableExplicitCongestionNotification(conn *net.UDPConn) error {
+	// noop.
+	return nil
+}

pkg/nada/ecn/ecn_linux.go

@@ -0,0 +1,17 @@
+package ecn
+
+import (
+	"net"
+	"reflect"
+	"syscall"
+)
+
+// EnableExplicitCongestionNotification enables ECN on the given connection.
+func EnableExplicitCongestionNotification(conn *net.UDPConn) error {
+	ptrVal := reflect.ValueOf(*conn)
+	fdmember := reflect.Indirect(ptrVal).FieldByName("fd")
+	pfdmember := reflect.Indirect(fdmember).FieldByName("pfd")
+	netfdmember := reflect.Indirect(pfdmember).FieldByName("Sysfd")
+	fd := int(netfdmember.Int())
+	return syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_RECVTOS, 1)
+}

pkg/nada/nada.go

@@ -0,0 +1,2 @@
+// Package nada provides an implementation of the NADA congestion control algorithm.
+package nada