Revert "Refactor repeat timer"

common/repeat_timer.go

@@ -1,6 +1,7 @@
 package common
 
 import (
+	"sync"
 	"time"
 )
 
@@ -10,93 +11,76 @@ It's good for keeping connections alive.
 A RepeatTimer must be Stop()'d or it will keep a goroutine alive.
 */
 type RepeatTimer struct {
-	Name   string
-	Ch     <-chan time.Time
-	output chan<- time.Time
-	input  chan repeatCommand
+	Ch chan time.Time
 
-	dur     time.Duration
-	ticker  *time.Ticker
-	stopped bool
+	mtx    sync.Mutex
+	name   string
+	ticker *time.Ticker
+	quit   chan struct{}
+	wg     *sync.WaitGroup
+	dur    time.Duration
 }
 
-type repeatCommand int8
-
-const (
-	Reset repeatCommand = iota
-	RQuit
-)
-
 func NewRepeatTimer(name string, dur time.Duration) *RepeatTimer {
-	c := make(chan time.Time)
 	var t = &RepeatTimer{
-		Name:   name,
-		Ch:     c,
-		output: c,
-		input:  make(chan repeatCommand),
-
-		dur:    dur,
+		Ch:     make(chan time.Time),
 		ticker: time.NewTicker(dur),
+		quit:   make(chan struct{}),
+		wg:     new(sync.WaitGroup),
+		name:   name,
+		dur:    dur,
 	}
-	go t.run()
+	t.wg.Add(1)
+	go t.fireRoutine(t.ticker)
 	return t
 }
 
+func (t *RepeatTimer) fireRoutine(ticker *time.Ticker) {
+	for {
+		select {
+		case t_ := <-ticker.C:
+			t.Ch <- t_
+		case <-t.quit:
+			// needed so we know when we can reset t.quit
+			t.wg.Done()
+			return
+		}
+	}
+}
+
 // Wait the duration again before firing.
 func (t *RepeatTimer) Reset() {
-	t.input <- Reset
+	t.Stop()
+
+	t.mtx.Lock() // Lock
+	defer t.mtx.Unlock()
+
+	t.ticker = time.NewTicker(t.dur)
+	t.quit = make(chan struct{})
+	t.wg.Add(1)
+	go t.fireRoutine(t.ticker)
 }
 
 // For ease of .Stop()'ing services before .Start()'ing them,
 // we ignore .Stop()'s on nil RepeatTimers.
 func (t *RepeatTimer) Stop() bool {
-	// use t.stopped to gracefully handle many Stop() without blocking
-	if t == nil || t.stopped {
+	if t == nil {
 		return false
 	}
-	t.input <- RQuit
-	t.stopped = true
-	return true
-}
+	t.mtx.Lock() // Lock
+	defer t.mtx.Unlock()
 
-func (t *RepeatTimer) run() {
-	done := false
-	for !done {
+	exists := t.ticker != nil
+	if exists {
+		t.ticker.Stop() // does not close the channel
 		select {
-		case cmd := <-t.input:
-			// stop goroutine if the input says so
-			// don't close channels, as closed channels mess up select reads
-			done = t.processInput(cmd)
-		case tick := <-t.ticker.C:
-			t.send(tick)
+		case <-t.Ch:
+			// read off channel if there's anything there
+		default:
 		}
+		close(t.quit)
+		t.wg.Wait() // must wait for quit to close else we race Reset
+		t.ticker = nil
 	}
-}
-
-// send performs blocking send on t.Ch
-func (t *RepeatTimer) send(tick time.Time) {
-	// XXX: possibly it is better to not block:
-	// https://golang.org/src/time/sleep.go#L132
-	// select {
-	// case t.output <- tick:
-	// default:
-	// }
-	t.output <- tick
-}
-
-// all modifications of the internal state of ThrottleTimer
-// happen in this method. It is only called from the run goroutine
-// so we avoid any race conditions
-func (t *RepeatTimer) processInput(cmd repeatCommand) (shutdown bool) {
-	switch cmd {
-	case Reset:
-		t.ticker.Stop()
-		t.ticker = time.NewTicker(t.dur)
-	case RQuit:
-		t.ticker.Stop()
-		shutdown = true
-	default:
-		panic("unknown command!")
-	}
-	return shutdown
+	return exists
 }