| 41 | } |
| 42 | |
| 43 | func (pm *processorManager) stop() { |
| 44 | // Notify the remote query-frontend or query-scheduler we're shutting down. |
| 45 | // We use a new context to make sure it's not cancelled. |
| 46 | notifyCtx, cancel := context.WithTimeout(context.Background(), notifyShutdownTimeout) |
| 47 | defer cancel() |
| 48 | pm.p.notifyShutdown(notifyCtx, pm.conn, pm.address) |
| 49 | |
| 50 | // Stop all goroutines. |
| 51 | pm.concurrency(0) |
| 52 | |
| 53 | // Wait until they finish. |
| 54 | pm.wg.Wait() |
| 55 | |
| 56 | _ = pm.conn.Close() |
| 57 | } |
| 58 | |
| 59 | func (pm *processorManager) concurrency(n int) { |
| 60 | pm.cancelsMu.Lock() |