From 192f75bfd46cb4c436c83193bcbff43c411ae7d2 Mon Sep 17 00:00:00 2001 From: Eric Fritz Date: Thu, 23 Jul 2026 19:55:53 -0500 Subject: [PATCH] Fix send-under-lock deadlock in MockTimer and MockTicker Both MockTimer and MockTicker delivered their tick by sending on an unbuffered channel while holding the internal mutex. A consumer that calls Stop() (or Reset(), or drains the channel) needs that same mutex, so if a tick was mid-send when the consumer reached Stop(), Stop() blocked on the lock, the send never completed, and the lock was never released -- a permanent deadlock. This is the deadlock that #9 reverted #6-#8 to avoid. Fix: deliver the tick with the lock released. - MockTimer/MockTicker now snapshot the value to send under the lock, drop the lock around the channel send (or AfterFunc), then re-acquire it, so a concurrent Stop/Reset/drain can always make progress. - MockTimer keeps a single long-lived process goroutine re-armed via the cond variable instead of spawning a fresh goroutine on every Reset (the old Reset leaked a goroutine and raced multiple deliverers on one channel). - MockTicker advances its deadline relative to the now value captured at fire time, not the live now, so an Advance during the (now lock-free) delivery window no longer skips a tick. - register() now takes the mutex itself (fixing the pre-existing data race on the subscriber list); callers already holding the lock use registerLocked. MockClock.After / MockTimer.Reset use registerLocked, and MockClock.NewTicker releases the lock before constructing the ticker to avoid re-entrant locking. Delivered values are snapshotted under the lock, resolving the -race failure where the released-lock send read now concurrently with Advance. Adds stress regression tests for both types (concurrent Stop during fire) that hang and fail via the test timeout if the deadlock regresses. Verified with `go test -race -count=10` on the full suite. --- advanceable.go | 14 +++++- mock_clock.go | 4 +- mock_ticker.go | 71 ++++++++++++++++++++++---- mock_ticker_test.go | 41 +++++++++++++++ mock_timer.go | 120 ++++++++++++++++++++++++++++++++++---------- mock_timer_test.go | 52 +++++++++++++++++++ 6 files changed, 262 insertions(+), 40 deletions(-) diff --git a/advanceable.go b/advanceable.go index 4db7ce7..4df2f39 100644 --- a/advanceable.go +++ b/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() } diff --git a/mock_clock.go b/mock_clock.go index f8cfea1..70f3b4e 100644 --- a/mock_clock.go +++ b/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 } diff --git a/mock_ticker.go b/mock_ticker.go index 4a55593..9b0611f 100644 --- a/mock_ticker.go +++ b/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 } diff --git a/mock_ticker_test.go b/mock_ticker_test.go index 060d992..3c5020f 100644 --- a/mock_ticker_test.go +++ b/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() + } +} diff --git a/mock_timer.go b/mock_timer.go index bd2ebbb..6286fb9 100644 --- a/mock_timer.go +++ b/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 } diff --git a/mock_timer_test.go b/mock_timer_test.go index ef92556..2e0460a 100644 --- a/mock_timer_test.go +++ b/mock_timer_test.go @@ -1,6 +1,7 @@ package glock import ( + "sync" "sync/atomic" "testing" "time" @@ -332,3 +333,54 @@ func TestMockTimer(t *testing.T) { }) }) } + +// TestMockTimerConcurrentStopDuringFire is a regression test for a deadlock +// where a firing timer sent on its channel while holding the internal mutex, so +// a concurrent Stop() (which needs that mutex) could block forever. The pattern +// mirrors a consumer that calls Stop()/drain/Reset on each iteration while the +// timer is being advanced past its deadline. If the deadlock regresses this test +// hangs and fails via the package test timeout. +func TestMockTimerConcurrentStopDuringFire(t *testing.T) { + t.Parallel() + + const interval = time.Second + for iter := 0; iter < 2000; iter++ { + clock := NewMockClock() + timer := clock.NewTimer(interval) + + var wg sync.WaitGroup + wg.Add(2) + + stop := make(chan struct{}) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + case <-timer.Chan(): + timer.Reset(interval) + default: + if !timer.Stop() { + select { + case <-timer.Chan(): + default: + } + } + timer.Reset(interval) + } + } + }() + + go func() { + defer wg.Done() + for i := 0; i < 50; i++ { + clock.Advance(interval) + } + close(stop) + }() + + wg.Wait() + timer.Stop() + } +}