Update ack_timer to use goroutines and RFC 9260

Author: philipch07

ack_timer.go

@@ -4,103 +4,204 @@
 package sctp
 
 import (
-	"math"
 	"sync"
 	"time"
 )
 
 const (
+	// RFC 9260 delayed SACK timer recommendation: 200 ms. It must not exceed 500ms.
+	// This timer only provides the max wait bound.
 	ackInterval time.Duration = 200 * time.Millisecond
 )
 
-// ackTimerObserver is the inteface to an ack timer observer.
 type ackTimerObserver interface {
 	onAckTimeout()
 }
 
-type ackTimerState uint8
+type timerCmdKind uint8
 
 const (
-	ackTimerStopped ackTimerState = iota
-	ackTimerStarted
-	ackTimerClosed
+	cmdStart timerCmdKind = iota
+	cmdStop
+	cmdClose
+	cmdIsRunning
 )
 
-// ackTimer provides the retnransmission timer conforms with RFC 4960 Sec 6.3.1.
+type timerCmd struct {
+	kind timerCmdKind
+	resp chan bool // for start/stop/isRunning result, nil for close
+}
+
+// ackTimer: single goroutine + reusable time.Timer, controlled by commands.
 type ackTimer struct {
-	timer    *time.Timer
 	observer ackTimerObserver
-	mutex    sync.Mutex
-	state    ackTimerState
-	pending  uint8
+
+	mutex  sync.Mutex
+	closed bool
+
+	cmdCh chan timerCmd
+	done  chan struct{}
 }
 
-// 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 := &ackTimer{
+		observer: observer,
+		cmdCh:    make(chan timerCmd),
+		done:     make(chan struct{}),
+	}
+
+	go t.run()
 
 	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()
+func (t *ackTimer) run() { //nolint:gocognit,cyclop
+	timer := time.NewTimer(time.Hour)
+
+	// Ensure timer is fully stopped and channel drained.
+	if !timer.Stop() {
+		select {
+		case <-timer.C:
+		default:
+		}
+	}
+
+	var (
+		running bool
+		timerC  <-chan time.Time // nil when not running
+	)
+
+	for {
+		select {
+		case <-timerC:
+			// timer fired, disarm then notify.
+			running = false
+			timerC = nil
+			t.observer.onAckTimeout()
+
+		case cmd := <-t.cmdCh:
+			switch cmd.kind {
+			case cmdStart:
+				if t.closed || running {
+					if cmd.resp != nil {
+						cmd.resp <- false
+					}
+
+					break
+				}
+
+				// Reset reusable timer to ackInterval.
+				if !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+
+				timer.Reset(ackInterval)
+				timerC = timer.C
+				running = true
+
+				if cmd.resp != nil {
+					cmd.resp <- true
+				}
+
+			case cmdStop:
+				if running {
+					// Stop and drain if necessary.
+					if !timer.Stop() {
+						select {
+						case <-timer.C:
+						default:
+						}
+					}
+
+					timerC = nil
+					running = false
+				}
+
+				if cmd.resp != nil {
+					cmd.resp <- true
+				}
+
+			case cmdIsRunning:
+				if cmd.resp != nil {
+					cmd.resp <- running
+				}
+
+			case cmdClose:
+				// Stop and drain once, exit loop.
+				if running && !timer.Stop() {
+					select {
+					case <-timer.C:
+					default:
+					}
+				}
+
+				close(t.done)
+
+				return
+			}
+		}
 	}
-	t.mutex.Unlock()
 }
 
-// start starts the timer.
+// start arms the timer and returns false if already started or closed.
 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.closed {
 		return false
 	}
 
-	t.state = ackTimerStarted
-	t.pending++
-	t.timer.Reset(ackInterval)
+	resp := make(chan bool, 1)
+	t.cmdCh <- timerCmd{kind: cmdStart, resp: resp}
 
-	return true
+	return <-resp
 }
 
-// stops the timer. this is similar to stop() but subsequent start() call
-// will fail (the timer is no longer usable).
+// stop disarms the timer if running.
 func (t *ackTimer) stop() {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
-	if t.state == ackTimerStarted {
-		if t.timer.Stop() {
-			t.pending--
-		}
-		t.state = ackTimerStopped
+	if t.closed {
+		return
 	}
+
+	resp := make(chan bool, 1)
+	t.cmdCh <- timerCmd{kind: cmdStop, resp: resp}
+	<-resp
 }
 
-// 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 and cannot be started again.
 func (t *ackTimer) close() {
 	t.mutex.Lock()
-	defer t.mutex.Unlock()
+	if t.closed {
+		t.mutex.Unlock()
 
-	if t.state == ackTimerStarted && t.timer.Stop() {
-		t.pending--
+		return
 	}
-	t.state = ackTimerClosed
+
+	t.closed = true
+	t.mutex.Unlock()
+
+	t.cmdCh <- timerCmd{kind: cmdClose}
+	<-t.done
 }
 
-// isRunning tests if the timer is running.
-// Debug purpose only.
+// isRunning is for debugging/tests.
 func (t *ackTimer) isRunning() bool {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
-	return t.state == ackTimerStarted
+	if t.closed {
+		return false
+	}
+
+	resp := make(chan bool, 1)
+	t.cmdCh <- timerCmd{kind: cmdIsRunning, resp: resp}
+
+	return <-resp
 }

ack_timer_test.go

@@ -26,23 +26,21 @@ func TestAckTimer(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
 			ok := rt.start()
 			assert.True(t, ok, "start() should succeed")
 			assert.True(t, rt.isRunning(), "should be running")
 
 			// subsequent start is a noop
 			ok = rt.start()
-			assert.False(t, ok, "start() should NOT succeed once closed")
-			assert.True(t, rt.isRunning(), "should be running")
+			assert.False(t, ok, "start() should NOT succeed while already running")
+			assert.True(t, rt.isRunning(), "should still be running")
 
-			// Sleep more than 2 * 200msec interval to test if it times out only once
+			// Sleep more than 2 * ackInterval to ensure only one timeout occurs.
 			time.Sleep(ackInterval*2 + 50*time.Millisecond)
 
 			assert.Equalf(t, uint32(1), atomic.LoadUint32(&nCbs),
@@ -51,11 +49,9 @@ func TestAckTimer(t *testing.T) {
 			atomic.StoreUint32(&nCbs, 0)
 		}
 
-		// should close ok
+		// close prevents further starts
 		rt.close()
-		assert.False(t, rt.isRunning(), "should not be running")
-
-		// once closed, it cannot start
+		assert.False(t, rt.isRunning(), "should not be running after close")
 		ok := rt.start()
 		assert.False(t, ok, "start() should NOT succeed once closed")
 		assert.False(t, rt.isRunning(), "should not be running")
@@ -65,35 +61,86 @@ func TestAckTimer(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
 			ok := rt.start()
 			assert.True(t, ok, "start() should succeed")
 			assert.True(t, rt.isRunning(), "should be running")
 
-			// stop immedidately
+			// stop immediately
 			rt.stop()
-			assert.False(t, rt.isRunning(), "should not be running")
+			assert.False(t, rt.isRunning(), "should not be running after stop()")
 		}
 
-		// Sleep more than 200msec of interval to test if it never times out
+		// ensure no callbacks after stops
 		time.Sleep(ackInterval + 50*time.Millisecond)
-
 		assert.Equalf(t, uint32(0), atomic.LoadUint32(&nCbs),
-			"should not be timed out (actual: %d)", atomic.LoadUint32(&nCbs))
+			"should not time out (actual: %d)", atomic.LoadUint32(&nCbs))
 
-		// can start again
+		// can start again after stop
 		ok := rt.start()
 		assert.True(t, ok, "start() should succeed again")
 		assert.True(t, rt.isRunning(), "should be running")
 
-		// should close ok
 		rt.close()
-		assert.False(t, rt.isRunning(), "should not be running")
+		assert.False(t, rt.isRunning(), "should not be running after close")
+	})
+
+	t.Run("stop shortly before deadline prevents fire", func(t *testing.T) {
+		var nCbs uint32
+		rt := newAckTimer(&testAckTimerObserver{
+			onAckTO: func() {
+				atomic.AddUint32(&nCbs, 1)
+			},
+		})
+
+		ok := rt.start()
+		assert.True(t, ok)
+		assert.True(t, rt.isRunning())
+
+		// Stop just before the deadline; leave a little margin.
+		time.Sleep(ackInterval/2 + 10*time.Millisecond)
+		rt.stop()
+		assert.False(t, rt.isRunning())
+
+		// Wait past the original deadline; must not fire.
+		time.Sleep(ackInterval)
+		assert.Equal(t, uint32(0), atomic.LoadUint32(&nCbs))
+	})
+
+	t.Run("light start/stop stress (no leaks, no stray fires)", func(t *testing.T) {
+		var nCbs uint32
+		rt := newAckTimer(&testAckTimerObserver{
+			onAckTO: func() {
+				atomic.AddUint32(&nCbs, 1)
+			},
+		})
+
+		// Several quick start/stop cycles; none should fire.
+		for i := 0; i < 10; i++ {
+			ok := rt.start()
+			// It's possible a previous iteration is still running if logic regressed; guard expectations.
+			if ok {
+				assert.True(t, rt.isRunning())
+			}
+
+			// Keep well under the interval.
+			time.Sleep(ackInterval / 10)
+			rt.stop()
+			assert.False(t, rt.isRunning())
+		}
+
+		time.Sleep(ackInterval + 50*time.Millisecond)
+		assert.Equal(t, uint32(0), atomic.LoadUint32(&nCbs), "no callbacks expected in stress loop")
+
+		// Final sanity: a clean start should fire exactly once.
+		atomic.StoreUint32(&nCbs, 0)
+		ok := rt.start()
+		assert.True(t, ok)
+		time.Sleep(ackInterval*2 + 50*time.Millisecond)
+		assert.Equal(t, uint32(1), atomic.LoadUint32(&nCbs))
 	})
 }