diff --git a/head.go b/head.go index 8f5beb4fb..8ba1a71bc 100644 --- a/head.go +++ b/head.go @@ -233,16 +233,20 @@ func (h *Head) ReadWAL() error { // They are connected through a ring of channels which ensures that all sample batches // read from the WAL are processed in order. var ( + wg sync.WaitGroup n = runtime.GOMAXPROCS(0) firstInput = make(chan []RefSample, 300) input = firstInput ) + wg.Add(n) + for i := 0; i < n; i++ { output := make(chan []RefSample, 300) go func(i int, input <-chan []RefSample, output chan<- []RefSample) { unknown := h.processWALSamples(mint, uint64(i), uint64(n), input, output) atomic.AddUint64(&unknownRefs, unknown) + wg.Done() }(i, input, output) // The output feeds the next worker goroutine. For the last worker, @@ -288,6 +292,8 @@ func (h *Head) ReadWAL() error { close(firstInput) for range input { } + wg.Wait() + if err != nil { return errors.Wrap(err, "consume WAL") }