Extract minRTT window for RACK

Author: philipch07

association.go

@@ -229,15 +229,11 @@ type Association struct {
 	ackTimer   *ackTimer
 
 	// RACK / TLP state
-	rackEnabled   bool
-	rackReoWnd    time.Duration // dynamic reordering window
-	rackMinRTT    time.Duration // min observed RTT
-	rackMinRTTWnd time.Duration // the window used to determine minRTT, defaults to 30s
-	rackRTTDeque  []struct {    // sliding window of RTT samples within rackMinRTTWnd seconds
-		time time.Time
-		rtt  time.Duration
-	}
-	rackDeliveredTime           time.Time // send time of most recently delivered original chunk
+	rackEnabled                 bool
+	rackReoWnd                  time.Duration // dynamic reordering window
+	rackMinRTT                  time.Duration // min observed RTT
+	rackMinRTTWnd               *windowedMin  // the window used to determine minRTT, defaults to 30s
+	rackDeliveredTime           time.Time     // send time of most recently delivered original chunk
 	rackHighestDeliveredOrigTSN uint32
 	rackReorderingSeen          bool // ever observed reordering for this association
 	rackKeepInflatedRecoveries  int  // keep inflated reoWnd for 16 loss recoveries
@@ -436,11 +432,8 @@ func createAssociation(config Config) *Association {
 			assoc.rackWCDelAck = 200 * time.Millisecond // WCDelAckT, RACK for SCTP section 2C
 		}
 
-		if config.RACKMinRTTWnd != 0 {
-			assoc.rackMinRTTWnd = config.RACKMinRTTWnd
-		} else {
-			assoc.rackMinRTTWnd = 30 * time.Second // default 30s window to determine minRTT
-		}
+		// defaults to 30s window to determine minRTT
+		assoc.rackMinRTTWnd = newWindowedMin(config.RACKMinRTTWnd)
 
 		assoc.timerUpdateCh = make(chan struct{}, 1)
 		go assoc.timerLoop()
@@ -1791,7 +1784,7 @@ func (a *Association) processSelectiveAck(selectiveAckChunk *chunkSelectiveAck)
 					// use a window to determine minRtt instead of a global min
 					// as the RTT can fluctuate, which can cause problems if going from a
 					// high RTT to a low RTT.
-					a.rackPushRTT(now, now.Sub(chunkPayload.since))
+					a.rackMinRTTWnd.Push(now, now.Sub(chunkPayload.since))
 
 					if chunkPayload.since.After(newestDeliveredSendTime) {
 						newestDeliveredSendTime = chunkPayload.since
@@ -1842,7 +1835,7 @@ func (a *Association) processSelectiveAck(selectiveAckChunk *chunkSelectiveAck)
 
 					// RACK
 					if a.rackEnabled {
-						a.rackPushRTT(now, now.Sub(chunkPayload.since))
+						a.rackMinRTTWnd.Push(now, now.Sub(chunkPayload.since))
 
 						if chunkPayload.since.After(newestDeliveredSendTime) {
 							newestDeliveredSendTime = chunkPayload.since
@@ -3065,11 +3058,7 @@ func (a *Association) startRackTimer(dur time.Duration) {
 
 	a.timerMu.Unlock()
 
-	// poke the loop
-	select {
-	case a.timerUpdateCh <- struct{}{}:
-	default:
-	}
+	a.pokeTimerLoop()
 }
 
 func (a *Association) stopRackTimer() {
@@ -3081,10 +3070,7 @@ func (a *Association) stopRackTimer() {
 	a.rackDeadline = time.Time{}
 	a.timerMu.Unlock()
 
-	select {
-	case a.timerUpdateCh <- struct{}{}:
-	default:
-	}
+	a.pokeTimerLoop()
 }
 
 func (a *Association) startPTOTimer(dur time.Duration) {
@@ -3102,11 +3088,7 @@ func (a *Association) startPTOTimer(dur time.Duration) {
 
 	a.timerMu.Unlock()
 
-	// poke the loop
-	select {
-	case a.timerUpdateCh <- struct{}{}:
-	default:
-	}
+	a.pokeTimerLoop()
 }
 
 func (a *Association) stopPTOTimer() {
@@ -3118,10 +3100,7 @@ func (a *Association) stopPTOTimer() {
 	a.ptoDeadline = time.Time{}
 	a.timerMu.Unlock()
 
-	select {
-	case a.timerUpdateCh <- struct{}{}:
-	default:
-	}
+	a.pokeTimerLoop()
 }
 
 // drainTimer safely stops a timer and drains its channel if needed.
@@ -3250,7 +3229,7 @@ func (a *Association) onRackAfterSACK( // nolint:gocognit,cyclop,gocyclo
 	}
 
 	// 2) Maintain ReoWND (RACK for SCTP section 2B)
-	if minRTT := a.rackWindowMin(now); minRTT > 0 {
+	if minRTT := a.rackMinRTTWnd.Min(now); minRTT > 0 {
 		a.rackMinRTT = minRTT
 	}
 
@@ -3442,53 +3421,10 @@ func (a *Association) onPTOTimerLocked() {
 	}
 }
 
-// push a new RTT sample and keep a deque within [now - rackMinRTTWnd, now].
-func (a *Association) rackPushRTT(now time.Time, rtt time.Duration) {
-	cutoff := now.Add(-a.rackMinRTTWnd)
-
-	// remove the expired rtts
-	deque := a.rackRTTDeque
-	i := 0
-	for i < len(deque) && deque[i].time.Before(cutoff) {
-		i++
-	}
-
-	if i > 0 {
-		deque = deque[i:]
-	}
-
-	// drop tails with >= rtt to keep nondecreasing by rtt
-	for len(deque) > 0 && deque[len(deque)-1].rtt >= rtt {
-		deque = deque[:len(deque)-1]
-	}
-
-	deque = append(deque, struct {
-		time time.Time
-		rtt  time.Duration
-	}{time: now, rtt: rtt})
-
-	a.rackRTTDeque = deque
-}
-
-// return current windowed MinRTT (or 0 if no recent samples).
-func (a *Association) rackWindowMin(now time.Time) time.Duration {
-	cutoff := now.Add(-a.rackMinRTTWnd)
-	deque := a.rackRTTDeque
-
-	// prune expired
-	i := 0
-	for i < len(deque) && deque[i].time.Before(cutoff) {
-		i++
-	}
-
-	if i > 0 {
-		deque = deque[i:]
-		a.rackRTTDeque = deque
-	}
-
-	if len(deque) == 0 {
-		return 0
+func (a *Association) pokeTimerLoop() {
+	// enqueue a single wake-up without blocking.
+	select {
+	case a.timerUpdateCh <- struct{}{}:
+	default:
 	}
-
-	return deque[0].rtt
 }

association_test.go

@@ -3964,7 +3964,7 @@ func TestRACK_SuppressReoWndDuringRecovery_NoReorderingSeen(t *testing.T) {
 	assert.Equal(t, time.Duration(0), assoc.rackReoWnd, "reoWnd should stay 0 until a minRTT sample exists")
 
 	now := time.Now()
-	assoc.rackPushRTT(now, 120*time.Millisecond)
+	assoc.rackMinRTTWnd.Push(now, 120*time.Millisecond)
 
 	assoc.onRackAfterSACK(false, time.Time{}, 0, &chunkSelectiveAck{})
 	assert.Equal(

windowedmin.go

@@ -0,0 +1,73 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package sctp
+
+import "time"
+
+// windowedMin maintains a monotonic deque of (time,value) to answer
+// the minimum over a sliding window efficiently.
+// Not thread-safe; caller must synchronize (Association already does).
+type windowedMin struct {
+	rackMinRTTWnd time.Duration
+	deque         []entry
+}
+
+type entry struct {
+	t time.Time
+	v time.Duration
+}
+
+func newWindowedMin(window time.Duration) *windowedMin {
+	if window <= 0 {
+		window = 30 * time.Second
+	}
+
+	return &windowedMin{rackMinRTTWnd: window}
+}
+
+// prune removes elements older than (now - wnd).
+func (window *windowedMin) prune(now time.Time) {
+	cutoff := now.Add(-window.rackMinRTTWnd)
+	i := 0
+	for i < len(window.deque) && window.deque[i].t.Before(cutoff) {
+		i++
+	}
+
+	if i > 0 {
+		window.deque = window.deque[i:]
+	}
+}
+
+// Push inserts a new sample and preserves monotonic non-decreasing values.
+func (window *windowedMin) Push(now time.Time, v time.Duration) {
+	window.prune(now)
+
+	for n := len(window.deque); n > 0 && window.deque[n-1].v >= v; n-- {
+		window.deque = window.deque[:n-1]
+	}
+
+	window.deque = append(
+		window.deque,
+		entry{
+			t: now,
+			v: v,
+		},
+	)
+}
+
+// Min returns the minimum value in the current window or 0 if empty.
+func (window *windowedMin) Min(now time.Time) time.Duration {
+	window.prune(now)
+
+	if len(window.deque) == 0 {
+		return 0
+	}
+
+	return window.deque[0].v
+}
+
+// Len is only for tests/diagnostics.
+func (window *windowedMin) Len() int {
+	return len(window.deque)
+}

windowedmin_test.go

@@ -0,0 +1,65 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+package sctp
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWindowedMin_Basic(t *testing.T) {
+	window := newWindowedMin(100 * time.Millisecond)
+	base := time.Unix(0, 0)
+
+	window.Push(base, 30*time.Millisecond)
+	assert.Equal(t, 30*time.Millisecond, window.Min(base))
+
+	window.Push(base.Add(10*time.Millisecond), 20*time.Millisecond)
+	assert.Equal(t, 20*time.Millisecond, window.Min(base.Add(10*time.Millisecond)))
+
+	// larger value shouldn't change min
+	window.Push(base.Add(20*time.Millisecond), 40*time.Millisecond)
+	assert.Equal(t, 20*time.Millisecond, window.Min(base.Add(20*time.Millisecond)))
+
+	// decreasing again shrinks min
+	window.Push(base.Add(30*time.Millisecond), 10*time.Millisecond)
+	assert.Equal(t, 10*time.Millisecond, window.Min(base.Add(30*time.Millisecond)))
+}
+
+func TestWindowedMin_WindowExpiry(t *testing.T) {
+	window := newWindowedMin(50 * time.Millisecond)
+	base := time.Unix(0, 0)
+
+	window.Push(base, 10*time.Millisecond)                          // t=0
+	window.Push(base.Add(10*time.Millisecond), 20*time.Millisecond) // t=10ms
+
+	// at t=60ms, first sample is expired, m becomes 20ms
+	m := window.Min(base.Add(60 * time.Millisecond))
+	assert.Equal(t, 20*time.Millisecond, m)
+
+	// at t=200ms, all are expired -> 0
+	m = window.Min(base.Add(200 * time.Millisecond))
+	assert.Zero(t, m)
+}
+
+func TestWindowedMin_EqualValues(t *testing.T) {
+	window := newWindowedMin(1 * time.Second)
+	base := time.Unix(0, 0)
+
+	window.Push(base, 15*time.Millisecond)
+	window.Push(base.Add(1*time.Millisecond), 15*time.Millisecond)
+
+	assert.Equal(t, 1, window.Len())
+	assert.Equal(t, 15*time.Millisecond, window.Min(base.Add(2*time.Millisecond)))
+}
+
+func TestWindowedMin_DefaultWindow30s(t *testing.T) {
+	zeroWnd := newWindowedMin(0)
+	negativeWnd := newWindowedMin(-5 * time.Second)
+
+	assert.Equal(t, 30*time.Second, zeroWnd.rackMinRTTWnd)
+	assert.Equal(t, 30*time.Second, negativeWnd.rackMinRTTWnd)
+}