Clean up locks and add comments

mengelbart · mengelbart · commit a8f8a9beefcc · 2026-02-04T14:08:20.000+01:00
diff --git a/pkg/rtpfb/history.go b/pkg/rtpfb/history.go
@@ -4,6 +4,7 @@
 package rtpfb
 
 import (
+	"sync"
 	"time"
 
 	"github.com/pion/rtcp"
@@ -22,6 +23,7 @@ type ssrcSequenceNumber struct {
 // report. buildReport can be used to create a new report including all packets
 // from nextReport to highestAcked.
 type history struct {
+	lock               sync.RWMutex
 	counter            uint64
 	twccToCounter      map[uint16]uint64
 	ssrcSeqNrToCounter map[ssrcSequenceNumber]uint64
@@ -35,12 +37,14 @@ type history struct {
 
 func newHistory() *history {
 	return &history{
+		lock:               sync.RWMutex{},
 		counter:            0,
 		twccToCounter:      map[uint16]uint64{},
 		ssrcSeqNrToCounter: map[ssrcSequenceNumber]uint64{},
 		packets:            make(map[uint64]*PacketReport),
 		highestAcked:       0,
 		nextReport:         0,
+		cleanUntil:         0,
 	}
 }
 
@@ -52,21 +56,20 @@ func (h *history) addOutgoing(
 	size int,
 	departure time.Time,
 ) {
-	count := h.counter
-	h.counter++
+	h.lock.Lock()
+	defer h.lock.Unlock()
 
 	if isTWCC {
-		h.twccToCounter[twccSequenceNumber] = count
+		h.twccToCounter[twccSequenceNumber] = h.counter
 	} else {
 		h.ssrcSeqNrToCounter[ssrcSequenceNumber{
 			ssrc:           ssrc,
 			sequenceNumber: rtpSequenceNumber,
-		}] = count
+		}] = h.counter
 	}
-
-	h.packets[count] = &PacketReport{
+	h.packets[h.counter] = &PacketReport{
 		SSRC:               ssrc,
-		SequenceNumber:     count,
+		SequenceNumber:     h.counter,
 		RTPSequenceNumber:  rtpSequenceNumber,
 		TWCCSequenceNumber: twccSequenceNumber,
 		Size:               size,
@@ -75,15 +78,22 @@ func (h *history) addOutgoing(
 		Arrival:            time.Time{},
 		ECN:                rtcp.ECNNonECT,
 	}
+	h.counter++
 }
 
+// onFeedback maps an incoming ack for counter to the PacketReport stored when
+// the packet was sent. If the packet cannot be found, the ack is ignored.
+//
+// onFeedback must be called while holding the lock for reading.
+// onFeedback returns the time between ts and the time the packet was sent.
 func (h *history) onFeedback(ts time.Time, counter uint64, ack acknowledgement) (time.Duration, bool) {
 	p, ok := h.packets[counter]
 	if !ok {
+		// ignore ack for unknown packet
 		return 0, false
 	}
 	p.Arrived = ack.arrived
-	if p.Arrived {
+	if p.Arrived && h.highestAcked < p.SequenceNumber {
 		h.highestAcked = p.SequenceNumber
 	}
 	p.Arrival = ack.arrival
@@ -92,7 +102,12 @@ func (h *history) onFeedback(ts time.Time, counter uint64, ack acknowledgement)
 	return ts.Sub(p.Departure), true
 }
 
+// onTWCCFeedback maps an acknowledgement to the counter by TWCC sequence number
+// and then calls onFeedback.
 func (h *history) onTWCCFeedback(ts time.Time, ack acknowledgement) (time.Duration, bool) {
+	h.lock.RLock()
+	defer h.lock.RUnlock()
+
 	counter, ok := h.twccToCounter[ack.sequenceNumber]
 	if !ok {
 		// ignore ack for unknown packet
@@ -102,7 +117,12 @@ func (h *history) onTWCCFeedback(ts time.Time, ack acknowledgement) (time.Durati
 	return h.onFeedback(ts, counter, ack)
 }
 
+// onCCFBFeedback maps an acknowledgement to the counter by ssrc and sequence
+// number and then calls onFeedback.
 func (h *history) onCCFBFeedback(ts time.Time, ssrc uint32, ack acknowledgement) (time.Duration, bool) {
+	h.lock.RLock()
+	defer h.lock.RUnlock()
+
 	counter, ok := h.ssrcSeqNrToCounter[ssrcSequenceNumber{
 		ssrc:           ssrc,
 		sequenceNumber: ack.sequenceNumber,
@@ -124,6 +144,9 @@ func (h *history) onCCFBFeedback(ts time.Time, ssrc uint32, ack acknowledgement)
 //
 //nolint:godox
 func (h *history) buildReport() []PacketReport {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
 	if h.nextReport > h.highestAcked {
 		return nil
 	}
@@ -145,6 +168,8 @@ func (h *history) buildReport() []PacketReport {
 	return res
 }
 
+// delete removes p from the history. It must be called while holding the lock
+// for writing.
 func (h *history) delete(p *PacketReport) {
 	if p.IsTWCC {
 		delete(h.twccToCounter, p.TWCCSequenceNumber)
@@ -155,6 +180,9 @@ func (h *history) delete(p *PacketReport) {
 	})
 }
 
+// cleanBefore removes all entries in the interval [h.cleanBefore, counter).
+// cleanBefore must be called while holding the lock for writing, because it
+// calls out to delete.
 func (h *history) cleanBefore(counter uint64) {
 	for i := h.cleanUntil; i < counter; i++ {
 		if p, ok := h.packets[i]; ok {
diff --git a/pkg/rtpfb/interceptor.go b/pkg/rtpfb/interceptor.go
@@ -6,7 +6,6 @@ package rtpfb
 
 import (
 	"math"
-	"sync"
 	"time"
 
 	"github.com/pion/interceptor"
@@ -72,7 +71,6 @@ func NewInterceptor(opts ...Option) (*InterceptorFactory, error) {
 func (f *InterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) {
 	in := &Interceptor{
 		NoOp:      interceptor.NoOp{},
-		lock:      sync.RWMutex{},
 		log:       logging.NewDefaultLoggerFactory().NewLogger("ccfb_interceptor"),
 		timestamp: time.Now,
 		history:   newHistory(),
@@ -94,7 +92,6 @@ func (f *InterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor,
 // report, a PacketReport will be added to the ccfb.Report.
 type Interceptor struct {
 	interceptor.NoOp
-	lock      sync.RWMutex
 	log       logging.LeveledLogger
 	timestamp func() time.Time
 
@@ -109,9 +106,6 @@ func (i *Interceptor) bindTWCCStream(twccHdrExtID uint8, writer interceptor.RTPW
 	) (int, error) {
 		ts := i.timestamp()
 
-		i.lock.Lock()
-		defer i.lock.Unlock()
-
 		var twccHdrExt rtp.TransportCCExtension
 		if err := twccHdrExt.Unmarshal(header.GetExtension(twccHdrExtID)); err != nil {
 			i.log.Warnf(
@@ -145,9 +139,6 @@ func (i *Interceptor) bindCCFBStream(writer interceptor.RTPWriter) interceptor.R
 	) (int, error) {
 		ts := i.timestamp()
 
-		i.lock.Lock()
-		defer i.lock.Unlock()
-
 		i.history.addOutgoing(
 			header.SSRC,
 			header.SequenceNumber,
@@ -217,9 +208,6 @@ func (i *Interceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.
 
 //nolint:cyclop
 func (i *Interceptor) processFeedback(ts time.Time, pkts []rtcp.Packet) (time.Duration, []PacketReport) {
-	i.lock.Lock()
-	defer i.lock.Unlock()
-
 	shortestRTT := time.Duration(math.MaxInt64)
 	var ackDelay time.Duration
 
