Add NADA implementation

This change adds the NADA congestion control implementation.
The binding to cc is left TODO.

Author: kevmo314

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/config.go

@@ -0,0 +1,97 @@
+package rfc8698
+
+import "time"
+
+type Bits uint32
+
+type BitsPerSecond float64
+
+const Kbps = BitsPerSecond(1_000)
+const Mbps = BitsPerSecond(1_000_000)
+
+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
+	κ float64
+	// Scaling parameter for gradual rate update calculation
+	η float64
+	// Upper bound of RTT in gradual rate update calculation
+	τ time.Duration
+	// Target feedback interval
+	δ 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
+	γ_max 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
+	λ 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
+	β_s float64
+	// Scaling parameter for modulating video encoder target rate
+	β_v float64
+	// Smoothing factor in exponential smoothing of packet loss and marking rate
+	α float64
+}
+
+var DefaultConfig = Config{
+	Priority:                 1.0,
+	MinimumRate:              150 * Kbps,
+	MaximumRate:              1500 * Kbps,
+	ReferenceCongestionLevel: 10 * time.Millisecond,
+	κ:                        0.5,
+	η:                        2.0,
+	τ:                        500 * time.Millisecond,
+	δ:                        100 * time.Millisecond,
+
+	LogWindow:              500 * time.Millisecond,
+	QueueingDelayThreshold: 10 * time.Millisecond,
+	FilteringDelay:         120 * time.Millisecond,
+	γ_max:                  0.5,
+	QueueBound:             50 * time.Millisecond,
+
+	LossMultiplier: 7.0,
+	DelayThreshold: 50 * time.Millisecond,
+	λ:              0.5,
+
+	ReferencePacketLossRatio:    0.01,
+	ReferencePacketMarkingRatio: 0.01,
+	ReferenceDelayLoss:          10 * time.Millisecond,
+	ReferenceDelayMarking:       2 * time.Millisecond,
+
+	FrameRate: 30.0,
+	β_s:       0.1,
+	β_v:       0.1,
+	α:         0.1,
+}

pkg/nada/ecn/ecn.go

@@ -0,0 +1,20 @@
+package ecn
+
+import (
+	"errors"
+	"syscall"
+)
+
+// 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, errors.New("no ECN control message")
+}

pkg/nada/ecn/ecn_darwin.go

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

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) {
+	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())
+	syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_RECVTOS, 1)
+}

pkg/nada/packet_stream.go

@@ -0,0 +1,123 @@
+package rfc8698
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+)
+
+type packet struct {
+	ts            time.Time
+	seq           uint16
+	ecn           bool
+	size          Bits
+	queueingDelay bool
+}
+
+// String returns a string representation of the packet.
+func (p *packet) String() string {
+	return fmt.Sprintf("%v@%v", p.seq, p.ts.Nanosecond()%1000)
+}
+
+type packetStream struct {
+	sync.Mutex
+
+	window             time.Duration
+	packets            []*packet
+	markCount          uint16
+	totalSize          Bits
+	queueingDelayCount uint16
+
+	tail uint16
+}
+
+func newPacketStream(window time.Duration) *packetStream {
+	return &packetStream{
+		window: window,
+	}
+}
+
+var errTimeOrder = errors.New("invalid packet timestamp ordering")
+
+// add writes a packet to the underlying stream.
+func (ps *packetStream) add(ts time.Time, seq uint16, ecn bool, size Bits, queueingDelay bool) error {
+	ps.Lock()
+	defer ps.Unlock()
+
+	if len(ps.packets) > 0 && ps.packets[len(ps.packets)-1].ts.After(ts) {
+		return errTimeOrder
+	}
+	// check if the packet seq already exists.
+	for _, p := range ps.packets {
+		if p.seq == seq {
+			return errTimeOrder
+		}
+	}
+	ps.packets = append(ps.packets, &packet{
+		ts:            ts,
+		seq:           seq,
+		ecn:           ecn,
+		size:          size,
+		queueingDelay: queueingDelay,
+	})
+	if ecn {
+		ps.markCount++
+	}
+	ps.totalSize += size
+	if queueingDelay {
+		ps.queueingDelayCount++
+	}
+	return nil
+}
+
+// prune removes packets that are older than the window and returns the loss and marking rate.
+func (ps *packetStream) prune(now time.Time) (loss float64, marking float64, receivingRate BitsPerSecond, hasQueueingDelay bool) {
+	ps.Lock()
+	defer ps.Unlock()
+
+	startTs := now.Add(-ps.window)
+	start := 0
+	for ; start < len(ps.packets) && ps.packets[start].ts.Before(startTs); start++ {
+		// decrement mark count if ecn.
+		if ps.packets[start].ecn {
+			ps.markCount--
+		}
+		ps.totalSize -= ps.packets[start].size
+		if ps.packets[start].queueingDelay {
+			ps.queueingDelayCount--
+		}
+	}
+	if start > 0 {
+		ps.packets = ps.packets[start:]
+	}
+	seqs := make([]uint16, len(ps.packets))
+	for i, p := range ps.packets {
+		seqs[i] = p.seq
+	}
+	begin, end := GetSeqRange(seqs)
+	loss = 1 - float64(len(ps.packets))/float64(end-begin+1)
+	marking = float64(ps.markCount) / float64(end-begin+1)
+	return loss, marking, BitsPerSecond(float64(ps.totalSize) / ps.window.Seconds()), ps.queueingDelayCount > 0
+}
+
+
+func GetSeqRange(seqs []uint16) (uint16, uint16) {
+	minDelta := 0
+	maxDelta := 0
+	seq0 := seqs[0]
+	for _, seq := range seqs {
+		delta := int(seq - seq0)
+		if seq-seq0 >= 16384 {
+			delta -= (1 << 16)
+			if delta < minDelta {
+				minDelta = delta
+			}
+		} else {
+			if delta > maxDelta {
+				maxDelta = delta
+			}
+		}
+	}
+	return seq0 + uint16(minDelta), seq0 + uint16(maxDelta)
+}