| 201 | } |
| 202 | |
| 203 | func (q *RequestQueue) stopping(_ error) error { |
| 204 | q.mtx.Lock() |
| 205 | defer q.mtx.Unlock() |
| 206 | |
| 207 | for q.queues.len() > 0 && q.connectedQuerierWorkers.Load() > 0 { |
| 208 | q.cond.Wait() |
| 209 | } |
| 210 | |
| 211 | // Only stop after dispatching enqueued requests. |
| 212 | q.stopped = true |
| 213 | |
| 214 | // If there are still goroutines in GetNextRequestForQuerier method, they get notified. |
| 215 | q.cond.Broadcast() |
| 216 | |
| 217 | return nil |
| 218 | } |
| 219 | |
| 220 | func (q *RequestQueue) RegisterQuerierConnection(querier string) { |
| 221 | q.connectedQuerierWorkers.Inc() |