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() + } +}