Fix send-under-lock deadlock in MockTimer and MockTicker

advanceable.go

@@ -73,8 +73,20 @@ func (a *advanceable) setCurrent(now time.Time) {
 	a.cond.Broadcast()
 }
 
-// register marks a subscriber to be updated when the current time changes.
+// register marks a subscriber to be updated when the current time changes. It
+// acquires the shared mutex, so callers must NOT already hold it; callers that
+// hold the lock (e.g. MockTimer.Reset) must use registerLocked instead.
 func (a *advanceable) register(subscriber subscriber) {
+	a.m.Lock()
+	defer a.m.Unlock()
+
+	a.registerLocked(subscriber)
+}
+
+// registerLocked is register for callers that already hold the shared mutex. It
+// mutates the subscriber list (also touched by setCurrent under the same lock),
+// so the lock must be held.
+func (a *advanceable) registerLocked(subscriber subscriber) {
 	a.subscribers = append(a.subscribers, subscriber)
 	a.cond.Broadcast()
 }

mock_clock.go

@@ -54,7 +54,9 @@ func (c *MockClock) After(duration time.Duration) <-chan time.Time {
 
 	ch := make(chan time.Time, 1)
 	deadline := c.now.Add(duration)
-	c.register(&afterSubscriber{ch: ch, deadline: deadline})
+	// Already holding the mutex here; use the locked variant to avoid a
+	// self-deadlock on the non-reentrant mutex.
+	c.registerLocked(&afterSubscriber{ch: ch, deadline: deadline})
 	return ch
 }
 

mock_ticker.go

@@ -13,6 +13,10 @@ type MockTicker struct {
 	deadline time.Time
 	ch       chan time.Time
 	stopped  bool
+	// firing is set while the process goroutine has released the lock to deliver
+	// a tick on ch. It exists so a concurrent Stop/BlockingAdvance can observe
+	// that a delivery is in flight even though the lock is momentarily free.
+	firing bool
 }
 
 var _ Ticker = &MockTicker{}
@@ -22,10 +26,13 @@ var _ Advanceable = &MockTicker{}
 // at intervals similar to time.NewTicker().  It will also skip or drop ticks
 // for slow readers similar to time.NewTicker() as well.
 func (c *MockClock) NewTicker(duration time.Duration) Ticker {
+	// Record the ticker args under the lock, then release it before
+	// constructing the ticker: newMockTickerAt -> register acquires the same
+	// mutex (c.m and the advanceable mutex are one and the same), and the mutex
+	// is not reentrant.
 	c.m.Lock()
-	defer c.m.Unlock()
-
 	c.tickerArgs = append(c.tickerArgs, duration)
+	c.m.Unlock()
 
 	return newMockTickerAt(c.advanceable, duration)
 }
@@ -73,17 +80,49 @@ func (t *MockTicker) Stop() {
 }
 
 // BlockingAdvance will bump the ticker's internal time by the given duration. If
-// If the new internal time passes the next tick threshold, a signal will be sent.
+// the new internal time passes the next tick threshold, a signal will be sent.
 // This method will not return until the signal is read by a consumer of the
 // ticker.
 func (t *MockTicker) BlockingAdvance(duration time.Duration) {
-	t.m.Lock()
-	defer t.m.Unlock()
+	t.cond.L.Lock()
+	defer t.cond.L.Unlock()
 
 	t.now = t.now.Add(duration)
 
-	if !t.now.Before(t.deadline) {
-		t.ch <- t.deadline
+	if t.stopped || t.now.Before(t.deadline) {
+		return
+	}
+
+	// Deliver the scheduled tick with the lock released so a concurrent Stop can
+	// still make progress, then drop any ticks that elapsed by the fire-time now
+	// (matching time.Ticker's slow-reader behavior). Advancement is relative to
+	// firedNow, not the live t.now, so a concurrent Advance during delivery does
+	// not cause us to skip its tick.
+	firedNow := t.now
+	t.deliver(t.deadline)
+	t.advanceDeadline(firedNow)
+}
+
+// deliver sends the given tick value on the channel with the lock released, then
+// re-acquires it. It must be called with t.cond.L held; on return the lock is
+// held again. Releasing the lock around the send is what prevents a deadlock:
+// the send blocks on an unbuffered channel, and a concurrent Stop or channel
+// consumer must be able to take the lock while that send is pending.
+func (t *MockTicker) deliver(tick time.Time) {
+	t.firing = true
+	t.cond.L.Unlock()
+	t.ch <- tick
+	t.cond.L.Lock()
+	t.firing = false
+}
+
+// advanceDeadline rolls the deadline forward past the given reference time,
+// dropping any intermediate ticks. Must be called with the lock held. The
+// reference time is passed explicitly (rather than read from t.now) so it
+// reflects the instant the just-delivered tick fired, even if t.now has moved
+// on while the delivery briefly released the lock.
+func (t *MockTicker) advanceDeadline(ref time.Time) {
+	for !ref.Before(t.deadline) {
 		t.deadline = t.deadline.Add(t.duration)
 	}
 }
@@ -94,11 +133,18 @@ func (t *MockTicker) process() {
 
 	for !t.stopped {
 		if !t.now.Before(t.deadline) {
-			t.ch <- t.deadline
-
-			for !t.now.Before(t.deadline) {
-				t.deadline = t.deadline.Add(t.duration)
+			// Snapshot the tick value and the fire-time now under the lock,
+			// deliver with the lock released, then advance past any ticks dropped
+			// for a slow reader relative to the fire-time now. A concurrent Stop
+			// taken during delivery is observed on the next loop iteration.
+			tick := t.deadline
+			firedNow := t.now
+			t.deliver(tick)
+			if t.stopped {
+				return
 			}
+			t.advanceDeadline(firedNow)
+			continue
 		}
 
 		t.cond.Wait()
@@ -107,5 +153,8 @@ func (t *MockTicker) process() {
 
 // signal conforms to the subscriber interface.
 func (t *MockTicker) signal(now time.Time) (requeue bool) {
+	// setCurrent (the caller) updates now and broadcasts under the lock, which
+	// wakes the process goroutine to re-check the deadline. We only report
+	// whether we remain interested in future updates.
 	return !t.stopped
 }

mock_ticker_test.go

@@ -1,6 +1,7 @@
 package glock
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -249,3 +250,43 @@ func TestTickerStopped(t *testing.T) {
 	clock.Advance(2 * time.Second)
 	consistently(t, chanDoesNotReceive(ticker.Chan()))
 }
+
+// TestMockTickerConcurrentStopDuringFire is a regression test for a deadlock
+// where a firing ticker sent on its channel while holding the internal mutex, so
+// a concurrent Stop() (which needs that mutex) could block forever. If the
+// deadlock regresses this test hangs and fails via the package test timeout.
+func TestMockTickerConcurrentStopDuringFire(t *testing.T) {
+	t.Parallel()
+
+	const interval = time.Second
+	for iter := 0; iter < 3000; iter++ {
+		clock := NewMockClock()
+		ticker := clock.NewTicker(interval)
+
+		var wg sync.WaitGroup
+		wg.Add(2)
+
+		drained := make(chan struct{})
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-ticker.Chan():
+				case <-drained:
+					return
+				}
+			}
+		}()
+
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 20; i++ {
+				clock.Advance(interval)
+			}
+			ticker.Stop()
+			close(drained)
+		}()
+
+		wg.Wait()
+	}
+}

mock_timer.go

@@ -4,8 +4,8 @@ import (
 	"time"
 )
 
-func sendTime(t *MockTimer) {
-	t.ch <- t.now
+func sendTime(t *MockTimer, now time.Time) {
+	t.ch <- now
 }
 
 // MockTimer is an implementation of Timer that can be moved forward in time
@@ -16,7 +16,17 @@ type MockTimer struct {
 	deadline time.Time
 	ch       chan time.Time
 	stopped  bool
-	f        func(*MockTimer)
+	// fired records whether the current arming has already elapsed and had its
+	// signal delivered. It gates the (deadline-reached) trigger so a single
+	// arming fires exactly once, and lets Stop/Reset report whether the timer
+	// was still running (see the time.Timer semantics they mirror).
+	fired bool
+	// firing is set while the process goroutine has released the lock to
+	// deliver a signal on ch (or invoke the AfterFunc). It exists so a
+	// concurrent Stop/Reset/BlockingAdvance can tell an in-flight delivery is
+	// underway even though the lock is momentarily free.
+	firing bool
+	f      func(*MockTimer, time.Time)
 }
 
 var _ Timer = &MockTimer{}
@@ -31,7 +41,7 @@ func (c *MockClock) NewTimer(duration time.Duration) Timer {
 // AfterFunc creates a new Timer tied to the internal MockClock time that functions
 // similar to time.AfterFunc().
 func (c *MockClock) AfterFunc(duration time.Duration, f func()) Timer {
-	return newMockTimerAt(c.advanceable, duration, func(mt *MockTimer) {
+	return newMockTimerAt(c.advanceable, duration, func(mt *MockTimer, _ time.Time) {
 		go f()
 	})
 }
@@ -49,7 +59,7 @@ func NewMockTimerAt(now time.Time, duration time.Duration) *MockTimer {
 func newMockTimerAt(
 	advanceable *advanceable,
 	duration time.Duration,
-	f func(*MockTimer),
+	f func(*MockTimer, time.Time),
 ) *MockTimer {
 	if duration == 0 {
 		panic("duration cannot be 0")
@@ -62,6 +72,10 @@ func newMockTimerAt(
 		f:           f,
 	}
 
+	// A single, long-lived process goroutine owns delivery for this timer for
+	// its whole lifetime. Reset re-arms it by clearing state and broadcasting
+	// rather than spawning another goroutine (which would leak goroutines and
+	// race multiple deliverers on the same channel).
 	go t.process()
 	advanceable.register(t)
 
@@ -81,16 +95,22 @@ func (t *MockTimer) Reset(duration time.Duration) bool {
 	t.cond.L.Lock()
 	defer t.cond.L.Unlock()
 
-	wasRunning := !t.stopped
+	// The timer was "running" iff it had neither been stopped nor already
+	// fired. An in-flight delivery (firing) still counts as running.
+	wasRunning := !t.stopped && !t.fired
 
 	t.deadline = t.now.Add(duration)
 	t.stopped = false
+	t.fired = false
 
-	if !wasRunning {
-		go t.process()
-		t.advanceable.register(t)
-	}
+	// Re-register with the advanceable so a future Advance re-notifies this
+	// timer. register is idempotent enough for our purposes: a timer only
+	// remains subscribed while signal() reports interest, and an expired timer
+	// drops itself, so a re-armed timer must re-subscribe.
+	t.advanceable.registerLocked(t)
 
+	// Wake the long-lived process goroutine to re-evaluate the (possibly
+	// already-passed) new deadline.
 	t.cond.Broadcast()
 
 	return wasRunning
@@ -101,7 +121,9 @@ func (t *MockTimer) Stop() bool {
 	t.cond.L.Lock()
 	defer t.cond.L.Unlock()
 
-	if t.stopped {
+	// Stop reports false when the timer has already fired or was already
+	// stopped, matching time.Timer.Stop.
+	if t.stopped || t.fired {
 		return false
 	}
 
@@ -116,35 +138,79 @@ func (t *MockTimer) Stop() bool {
 // This method will not return until the signal is read by a consumer of the Timer's
 // channel.
 func (t *MockTimer) BlockingAdvance(duration time.Duration) {
-	t.m.Lock()
-	defer t.m.Unlock()
-
+	t.cond.L.Lock()
 	t.now = t.now.Add(duration)
 
-	t.tryExecute()
-}
+	// If this advance does not cross the deadline (or the timer is not running),
+	// there is nothing to deliver: release and return without blocking.
+	if t.stopped || t.fired || t.now.Before(t.deadline) {
+		t.cond.L.Unlock()
+		return
+	}
 
-func (t *MockTimer) tryExecute() {
-	if !t.now.Before(t.deadline) {
-		t.stopped = true
-		t.cond.Broadcast()
+	// The deadline is crossed. Mark the arming as fired and deliver the signal
+	// directly here (releasing the lock around the delivery) so the caller
+	// blocks until a consumer reads the channel — matching the documented
+	// BlockingAdvance contract. The AfterFunc variant does not write to the
+	// channel, so it does not block the caller.
+	t.fired = true
+	t.cond.Broadcast()
+	t.deliver()
 
-		t.f(t)
-	}
+	t.cond.L.Unlock()
+}
+
+// deliver invokes the timer's trigger function with the lock released, then
+// re-acquires it. It must be called with t.cond.L held; on return the lock is
+// held again. Releasing the lock around t.f is what prevents a deadlock: t.f
+// (sendTime) blocks on an unbuffered channel send, and a concurrent
+// Stop/Reset/Chan-drain must be able to take the lock while that send is
+// pending.
+func (t *MockTimer) deliver() {
+	// Snapshot the value to deliver while the lock is still held; t.f runs with
+	// the lock released, so it must not touch shared fields (t.now is written by
+	// Advance under the same lock).
+	now := t.now
+	t.firing = true
+	t.cond.L.Unlock()
+	t.f(t, now)
+	t.cond.L.Lock()
+	t.firing = false
 }
 
 func (t *MockTimer) process() {
 	t.cond.L.Lock()
 	defer t.cond.L.Unlock()
 
-	for !t.stopped {
-		t.tryExecute()
-		t.cond.Wait()
+	for {
+		switch {
+		case t.stopped:
+			// Park until Reset re-arms us (or the process is abandoned when the
+			// clock and all references are dropped).
+			t.cond.Wait()
+
+		case !t.fired && !t.now.Before(t.deadline):
+			// Deadline reached and not yet delivered for this arming. Mark fired
+			// first so Stop/Reset observe consistent state, then deliver with the
+			// lock released. If a concurrent Reset re-armed us during delivery,
+			// t.fired is false again on return and we loop to re-evaluate.
+			t.fired = true
+			t.cond.Broadcast()
+			t.deliver()
+
+		default:
+			// Either waiting for the deadline, or already fired and waiting for a
+			// Reset. Sleep until state changes.
+			t.cond.Wait()
+		}
 	}
-
 }
 
 // signal conforms to the subscriber interface.
 func (t *MockTimer) signal(now time.Time) bool {
-	return !t.stopped
+	// setCurrent (the caller) updates now and broadcasts under the lock, which
+	// wakes the process goroutine to re-check the deadline. Remain subscribed
+	// only while the timer is still live (not stopped and not already fired for
+	// the current arming); Reset re-subscribes on re-arm.
+	return !t.stopped && !t.fired
 }