Correctly handle select on multiple channels in Queues (#22146)
There are a few places in FlushQueueWithContext which make an incorrect assumption about how `select` on multiple channels works. The problem is best expressed by looking at the following example: ```go package main import "fmt" func main() { closedChan := make(chan struct{}) close(closedChan) toClose := make(chan struct{}) count := 0 for { select { case <-closedChan: count++ fmt.Println(count) if count == 2 { close(toClose) } case <-toClose: return } } } ``` This PR double-checks that the contexts are closed outside of checking if there is data in the dataChan. It also rationalises the WorkerPool FlushWithContext because the previous implementation failed to handle pausing correctly. This will probably fix the underlying problem in #22145 Fix #22145 Signed-off-by: Andrew Thornton <art27@cantab.net> Signed-off-by: Andrew Thornton <art27@cantab.net>
This commit is contained in:
parent
47efba78ec
commit
a609cae9fb
3 changed files with 43 additions and 57 deletions
|
@ -109,32 +109,6 @@ func (q *ChannelQueue) Flush(timeout time.Duration) error {
|
|||
return q.FlushWithContext(ctx)
|
||||
}
|
||||
|
||||
// FlushWithContext is very similar to CleanUp but it will return as soon as the dataChan is empty
|
||||
func (q *ChannelQueue) FlushWithContext(ctx context.Context) error {
|
||||
log.Trace("ChannelQueue: %d Flush", q.qid)
|
||||
paused, _ := q.IsPausedIsResumed()
|
||||
for {
|
||||
select {
|
||||
case <-paused:
|
||||
return nil
|
||||
case data, ok := <-q.dataChan:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if unhandled := q.handle(data); unhandled != nil {
|
||||
log.Error("Unhandled Data whilst flushing queue %d", q.qid)
|
||||
}
|
||||
atomic.AddInt64(&q.numInQueue, -1)
|
||||
case <-q.baseCtx.Done():
|
||||
return q.baseCtx.Err()
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown processing from this queue
|
||||
func (q *ChannelQueue) Shutdown() {
|
||||
q.lock.Lock()
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue