Ensure workers terminated fully before reading unknownRefs
@@ -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")
}
