Update ack_timer to RFC 9260

Author: philipch07

ack_timer.go

@@ -4,18 +4,18 @@
 package sctp
 
 import (
-	"math"
 	"sync"
 	"time"
 )
 
 const (
+	// RFC 9260 section 6.2.1: the delayed SACK timer SHOULD be 200ms or less. Max 500ms.
 	ackInterval time.Duration = 200 * time.Millisecond
 )
 
-// ackTimerObserver is the inteface to an ack timer observer.
+// ackTimerObserver is the interface to an ack timer observer.
 type ackTimerObserver interface {
-	onAckTimeout()
+	onAckTimeout() // called to send a SACK now (either timer fired or 2nd packet rule)
 }
 
 type ackTimerState uint8
@@ -26,74 +26,119 @@ const (
 	ackTimerClosed
 )
 
-// ackTimer provides the retnransmission timer conforms with RFC 4960 Sec 6.3.1.
+// ackTimer provides the delayed-ACK behavior per RFC 9260 section 6.2.1.
 type ackTimer struct {
 	timer    *time.Timer
 	observer ackTimerObserver
-	mutex    sync.Mutex
-	state    ackTimerState
-	pending  uint8
+
+	mutex   sync.Mutex
+	state   ackTimerState
+	packets uint8 // number of packets observed since last SACK trigger (0 or 1 while running)
 }
 
 // newAckTimer creates a new acknowledgement timer used to enable delayed ack.
 func newAckTimer(observer ackTimerObserver) *ackTimer {
 	t := &ackTimer{observer: observer}
-	t.timer = time.AfterFunc(math.MaxInt64, t.timeout)
-	t.timer.Stop()
+	t.timer = time.NewTimer(time.Hour)
+
+	if !t.timer.Stop() {
+		<-t.timer.C
+	}
 
 	return t
 }
 
 func (t *ackTimer) timeout() {
 	t.mutex.Lock()
-	if t.pending--; t.pending == 0 && t.state == ackTimerStarted {
-		t.state = ackTimerStopped
-		defer t.observer.onAckTimeout()
+
+	if t.state != ackTimerStarted {
+		t.mutex.Unlock()
+
+		return
 	}
+
+	t.state = ackTimerStopped
+	t.packets = 0
+	obs := t.observer
+
 	t.mutex.Unlock()
+
+	// Trigger SACK on timer expiry.
+	obs.onAckTimeout()
 }
 
-// start starts the timer.
+// start notes an arriving packet eligible for delayed SACK logic.
+// 1st call while stopped: arm timer.
+// 2nd call while timer running: send ACK immediately and stop timer.
 func (t *ackTimer) start() bool {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
-	// this timer is already closed or already running
-	if t.state != ackTimerStopped {
+	if t.state == ackTimerClosed {
 		return false
 	}
 
-	t.state = ackTimerStarted
-	t.pending++
-	t.timer.Reset(ackInterval)
+	switch t.state {
+	case ackTimerStopped:
+		t.state = ackTimerStarted
+		t.packets = 1
+
+		// Arm timer to ackInterval and hook timeout.
+		t.timer.Reset(ackInterval)
+
+		go func() { // isolate timer callback
+			<-t.timer.C
+			t.timeout()
+		}()
+
+		return true
 
-	return true
+	case ackTimerStarted:
+		// Second packet rule: send SACK immediately.
+		t.packets++
+
+		if t.packets >= 2 {
+			t.timer.Stop()
+
+			t.state = ackTimerStopped
+			t.packets = 0
+			obs := t.observer
+
+			// Fire outside the lock to avoid re-entrancy issues.
+			go obs.onAckTimeout()
+		}
+
+		return true
+
+	default: // closed
+		return false
+	}
 }
 
-// stops the timer. this is similar to stop() but subsequent start() call
-// will fail (the timer is no longer usable).
+// stop cancels a running delayed-ACK cycle (no immediate ACK).
 func (t *ackTimer) stop() {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
 	if t.state == ackTimerStarted {
-		if t.timer.Stop() {
-			t.pending--
-		}
+		t.timer.Stop()
+
 		t.state = ackTimerStopped
+		t.packets = 0
 	}
 }
 
-// closes the timer. this is similar to stop() but subsequent start() call
-// will fail (the timer is no longer usable).
+// close permanently disables the timer; subsequent start() calls are ignored.
 func (t *ackTimer) close() {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
-	if t.state == ackTimerStarted && t.timer.Stop() {
-		t.pending--
+	if t.state == ackTimerStarted {
+		t.timer.Stop()
 	}
+
 	t.state = ackTimerClosed
+	t.packets = 0
 }
 
 // isRunning tests if the timer is running.

ack_timer_test.go

@@ -22,50 +22,48 @@ func (o *testAckTimerObserver) onAckTimeout() {
 }
 
 func TestAckTimer(t *testing.T) {
-	t.Run("start and close", func(t *testing.T) {
+	t.Run("start and close (second packet rule)", func(t *testing.T) {
 		var nCbs uint32
 		rt := newAckTimer(&testAckTimerObserver{
 			onAckTO: func() {
-				t.Log("ack timed out")
 				atomic.AddUint32(&nCbs, 1)
 			},
 		})
 
 		for i := 0; i < 2; i++ {
-			// should start ok
+			// First eligible packet -> arm delayed SACK timer
 			ok := rt.start()
-			assert.True(t, ok, "start() should succeed")
-			assert.True(t, rt.isRunning(), "should be running")
+			assert.True(t, ok, "first start() should succeed")
+			assert.True(t, rt.isRunning(), "timer should be running after first start")
 
-			// subsequent start is a noop
+			// Second eligible packet -> immediate SACK, timer stops (RFC 9260 section 6.2.1)
 			ok = rt.start()
-			assert.False(t, ok, "start() should NOT succeed once closed")
-			assert.True(t, rt.isRunning(), "should be running")
+			assert.True(t, ok, "second start() should be accepted and trigger immediate ACK")
 
-			// Sleep more than 2 * 200msec interval to test if it times out only once
-			time.Sleep(ackInterval*2 + 50*time.Millisecond)
+			// Give the async callback time to fire.
+			time.Sleep(20 * time.Millisecond)
 
 			assert.Equalf(t, uint32(1), atomic.LoadUint32(&nCbs),
-				"should be called once (actual: %d)", atomic.LoadUint32(&nCbs))
+				"immediate ACK should have fired exactly once (iter %d)", i)
+			assert.False(t, rt.isRunning(), "timer should have stopped after immediate ACK")
 
+			// Reset for next iteration
 			atomic.StoreUint32(&nCbs, 0)
 		}
 
-		// should close ok
+		// Close disables future starts
 		rt.close()
-		assert.False(t, rt.isRunning(), "should not be running")
+		assert.False(t, rt.isRunning(), "timer should not be running after close")
 
-		// once closed, it cannot start
 		ok := rt.start()
 		assert.False(t, ok, "start() should NOT succeed once closed")
-		assert.False(t, rt.isRunning(), "should not be running")
+		assert.False(t, rt.isRunning(), "still not running after refused start on closed timer")
 	})
 
-	t.Run("start and stop", func(t *testing.T) {
+	t.Run("start and stop (no timeout)", func(t *testing.T) {
 		var nCbs uint32
 		rt := newAckTimer(&testAckTimerObserver{
 			onAckTO: func() {
-				t.Log("ack timed out")
 				atomic.AddUint32(&nCbs, 1)
 			},
 		})
@@ -74,26 +72,25 @@ func TestAckTimer(t *testing.T) {
 			// should start ok
 			ok := rt.start()
 			assert.True(t, ok, "start() should succeed")
-			assert.True(t, rt.isRunning(), "should be running")
+			assert.True(t, rt.isRunning(), "timer should be running")
 
 			// stop immedidately
 			rt.stop()
-			assert.False(t, rt.isRunning(), "should not be running")
+			assert.False(t, rt.isRunning(), "timer should not be running after stop()")
 		}
 
-		// Sleep more than 200msec of interval to test if it never times out
+		// Wait > interval to ensure no stray timeout happened.
 		time.Sleep(ackInterval + 50*time.Millisecond)
-
 		assert.Equalf(t, uint32(0), atomic.LoadUint32(&nCbs),
-			"should not be timed out (actual: %d)", atomic.LoadUint32(&nCbs))
+			"no ACK should have fired after stop() (got %d)", atomic.LoadUint32(&nCbs))
 
 		// can start again
 		ok := rt.start()
 		assert.True(t, ok, "start() should succeed again")
-		assert.True(t, rt.isRunning(), "should be running")
+		assert.True(t, rt.isRunning(), "timer should be running")
 
-		// should close ok
+		// Close disables and stops
 		rt.close()
-		assert.False(t, rt.isRunning(), "should not be running")
+		assert.False(t, rt.isRunning(), "timer should not be running after close")
 	})
 }