doneChan() returns a channel that can be used to watch for context cancellation across the entire batch. it only closes the returned channel if all contexts in the batch are done. the consequence of this is that if the batch is broken across upstream http queries then the queriers may continue worki
(stop <-chan struct{})
| 78 | // they don't need to. this should be rare. nearly always all jobs in one batch |
| 79 | // will belong to the same upstream http query. |
| 80 | func (b *requestBatch) doneChan(stop <-chan struct{}) <-chan struct{} { |
| 81 | if len(b.pipelineRequests) == 1 { |
| 82 | return b.pipelineRequests[0].OriginalContext().Done() |
| 83 | } |
| 84 | |
| 85 | done := make(chan struct{}) |
| 86 | go func() { |
| 87 | defer close(done) |
| 88 | // tests each request context and only closes done if all are done. |
| 89 | // technically it is only testing one a time, but the loop will only complete |
| 90 | // if all are done. |
| 91 | for _, r := range b.pipelineRequests { |
| 92 | select { |
| 93 | case <-r.OriginalContext().Done(): |
| 94 | case <-stop: |
| 95 | return |
| 96 | } |
| 97 | } |
| 98 | }() |
| 99 | |
| 100 | return done |
| 101 | } |
| 102 | |
| 103 | // reportErrorToPipeline sends errors back up the query frontend http pipeline |
| 104 | func (b *requestBatch) reportErrorToPipeline(err error) { |