simplify the batcher's timers

This commit is contained in:
Alex Dadgar 2017-07-03 11:51:58 -07:00
parent 7154e4e08f
commit 5fdee898b2

View file

@ -58,40 +58,28 @@ func (b *EvalBatcher) CreateEval(e *structs.Evaluation) *EvalFuture {
// batcher is the long lived batcher goroutine // batcher is the long lived batcher goroutine
func (b *EvalBatcher) batcher() { func (b *EvalBatcher) batcher() {
timer := time.NewTimer(b.batch) var timerCh <-chan time.Time
evals := make(map[string]*structs.Evaluation) evals := make(map[string]*structs.Evaluation)
for { for {
select { select {
case <-b.ctx.Done(): case <-b.ctx.Done():
timer.Stop()
return return
case e := <-b.inCh: case e := <-b.inCh:
if len(evals) == 0 { if timerCh == nil {
if !timer.Stop() { timerCh = time.After(b.batch)
<-timer.C
}
timer.Reset(b.batch)
} }
evals[e.DeploymentID] = e evals[e.DeploymentID] = e
case <-timer.C: case <-timerCh:
if len(evals) == 0 {
// Reset the timer
timer.Reset(b.batch)
continue
}
// Capture the future // Capture the future
b.l.Lock() b.l.Lock()
f := b.f f := b.f
b.f = nil b.f = nil
b.l.Unlock() b.l.Unlock()
// Shouldn't be possible but protect ourselves // Shouldn't be possible
if f == nil { if f == nil {
// Reset the timer panic("no future")
timer.Reset(b.batch)
continue
} }
// Capture the evals // Capture the evals
@ -100,14 +88,13 @@ func (b *EvalBatcher) batcher() {
all = append(all, e) all = append(all, e)
} }
// Upsert the evals // Upsert the evals in a go routine
f.Set(b.raft.UpsertEvals(all)) go f.Set(b.raft.UpsertEvals(all))
// Reset the evals list // Reset the evals list and timer
evals = make(map[string]*structs.Evaluation) evals = make(map[string]*structs.Evaluation)
timerCh = nil
// Reset the timer
timer.Reset(b.batch)
} }
} }
} }