don't leave WaitForState goroutine running

Make sure that we can cancel the WaitForState refresh loop when reaching
a timeout, otherwise it may run indefinitely. There's no need to try and
store and read the Result concurrently, just pass the value over a
channel.
This commit is contained in:
James Bardin 2017-04-19 10:10:07 -04:00
parent 4a782583b6
commit af5e22cf94
1 changed files with 61 additions and 30 deletions

View File

@ -2,7 +2,6 @@ package resource
import ( import (
"log" "log"
"sync/atomic"
"time" "time"
) )
@ -62,33 +61,45 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
conf.ContinuousTargetOccurence = 1 conf.ContinuousTargetOccurence = 1
} }
// We can't safely read the result values if we timeout, so store them in
// an atomic.Value
type Result struct { type Result struct {
Result interface{} Result interface{}
State string State string
Error error Error error
Done bool
} }
var lastResult atomic.Value
lastResult.Store(Result{})
doneCh := make(chan struct{}) // read ever result from the refresh loop, waiting for a positive result.Done
resCh := make(chan Result, 1)
// cancellation channel for the refresh loop
cancelCh := make(chan struct{})
go func() { go func() {
defer close(doneCh) defer close(resCh)
// Wait for the delay
time.Sleep(conf.Delay) time.Sleep(conf.Delay)
wait := 100 * time.Millisecond // start with 0 delay for the first loop
var wait time.Duration
for { for {
// wait and watch for cancellation
select {
case <-cancelCh:
return
case <-time.After(wait):
// first round had no wait
if wait == 0 {
wait = 100 * time.Millisecond
}
}
res, currentState, err := conf.Refresh() res, currentState, err := conf.Refresh()
result := Result{ result := Result{
Result: res, Result: res,
State: currentState, State: currentState,
Error: err, Error: err,
} }
lastResult.Store(result) resCh <- result
if err != nil { if err != nil {
return return
@ -98,6 +109,8 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
if res == nil && len(conf.Target) == 0 { if res == nil && len(conf.Target) == 0 {
targetOccurence += 1 targetOccurence += 1
if conf.ContinuousTargetOccurence == targetOccurence { if conf.ContinuousTargetOccurence == targetOccurence {
result.Done = true
resCh <- result
return return
} else { } else {
continue continue
@ -113,7 +126,7 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
LastError: err, LastError: err,
Retries: notfoundTick, Retries: notfoundTick,
} }
lastResult.Store(result) resCh <- result
return return
} }
} else { } else {
@ -126,6 +139,8 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
found = true found = true
targetOccurence += 1 targetOccurence += 1
if conf.ContinuousTargetOccurence == targetOccurence { if conf.ContinuousTargetOccurence == targetOccurence {
result.Done = true
resCh <- result
return return
} else { } else {
continue continue
@ -147,7 +162,7 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
State: result.State, State: result.State,
ExpectedState: conf.Target, ExpectedState: conf.Target,
} }
lastResult.Store(result) resCh <- result
return return
} }
} }
@ -162,30 +177,46 @@ func (conf *StateChangeConf) WaitForState() (interface{}, error) {
} else if wait > 10*time.Second { } else if wait > 10*time.Second {
wait = 10 * time.Second wait = 10 * time.Second
} }
// Wait between refreshes using exponential backoff, except when
// waiting for the target state to reoccur.
if targetOccurence == 0 {
wait *= 2
}
} }
log.Printf("[TRACE] Waiting %s before next try", wait) log.Printf("[TRACE] Waiting %s before next try", wait)
time.Sleep(wait)
// Wait between refreshes using exponential backoff, except when
// waiting for the target state to reoccur.
if targetOccurence == 0 {
wait *= 2
}
} }
}() }()
select { // store the last value result from the refresh loop
case <-doneCh: lastResult := Result{}
r := lastResult.Load().(Result)
return r.Result, r.Error timeout := time.After(conf.Timeout)
case <-time.After(conf.Timeout): for {
r := lastResult.Load().(Result) select {
return nil, &TimeoutError{ case r, ok := <-resCh:
LastError: r.Error, // channel closed, so return the last result
LastState: r.State, if !ok {
Timeout: conf.Timeout, return lastResult.Result, lastResult.Error
ExpectedState: conf.Target, }
// we reached the intended state
if r.Done {
return r.Result, r.Error
}
// still waiting, store the last result
lastResult = r
case <-timeout:
close(cancelCh)
return nil, &TimeoutError{
LastError: lastResult.Error,
LastState: lastResult.State,
Timeout: conf.Timeout,
ExpectedState: conf.Target,
}
} }
} }
} }