func()

in tsdb/wal.go [866:999]
111 lines of code
27 McCabe index (conditional complexity)

func (r *walReader) Read(
	seriesf func([]record.RefSeries),
	samplesf func([]record.RefSample),
	deletesf func([]tombstones.Stone),
) error {
	// Concurrency for replaying the WAL is very limited. We at least split out decoding and
	// processing into separate threads.
	// Historically, the processing is the bottleneck with reading and decoding using only
	// 15% of the CPU.
	var (
		seriesPool sync.Pool
		samplePool sync.Pool
		deletePool sync.Pool
	)
	donec := make(chan struct{})
	datac := make(chan interface{}, 100)

	go func() {
		defer close(donec)

		for x := range datac {
			switch v := x.(type) {
			case []record.RefSeries:
				if seriesf != nil {
					seriesf(v)
				}
				//lint:ignore SA6002 safe to ignore and actually fixing it has some performance penalty.
				seriesPool.Put(v[:0])
			case []record.RefSample:
				if samplesf != nil {
					samplesf(v)
				}
				//lint:ignore SA6002 safe to ignore and actually fixing it has some performance penalty.
				samplePool.Put(v[:0])
			case []tombstones.Stone:
				if deletesf != nil {
					deletesf(v)
				}
				//lint:ignore SA6002 safe to ignore and actually fixing it has some performance penalty.
				deletePool.Put(v[:0])
			default:
				level.Error(r.logger).Log("msg", "unexpected data type")
			}
		}
	}()

	var err error

	for r.next() {
		et, flag, b := r.at()

		// In decoding below we never return a walCorruptionErr for now.
		// Those should generally be caught by entry decoding before.
		switch et {
		case WALEntrySeries:
			var series []record.RefSeries
			if v := seriesPool.Get(); v == nil {
				series = make([]record.RefSeries, 0, 512)
			} else {
				series = v.([]record.RefSeries)
			}

			err = r.decodeSeries(flag, b, &series)
			if err != nil {
				err = errors.Wrap(err, "decode series entry")
				break
			}
			datac <- series

			cf := r.current()
			for _, s := range series {
				if cf.minSeries > s.Ref {
					cf.minSeries = s.Ref
				}
			}
		case WALEntrySamples:
			var samples []record.RefSample
			if v := samplePool.Get(); v == nil {
				samples = make([]record.RefSample, 0, 512)
			} else {
				samples = v.([]record.RefSample)
			}

			err = r.decodeSamples(flag, b, &samples)
			if err != nil {
				err = errors.Wrap(err, "decode samples entry")
				break
			}
			datac <- samples

			// Update the times for the WAL segment file.
			cf := r.current()
			for _, s := range samples {
				if cf.maxTime < s.T {
					cf.maxTime = s.T
				}
			}
		case WALEntryDeletes:
			var deletes []tombstones.Stone
			if v := deletePool.Get(); v == nil {
				deletes = make([]tombstones.Stone, 0, 512)
			} else {
				deletes = v.([]tombstones.Stone)
			}

			err = r.decodeDeletes(flag, b, &deletes)
			if err != nil {
				err = errors.Wrap(err, "decode delete entry")
				break
			}
			datac <- deletes

			// Update the times for the WAL segment file.
			cf := r.current()
			for _, s := range deletes {
				for _, iv := range s.Intervals {
					if cf.maxTime < iv.Maxt {
						cf.maxTime = iv.Maxt
					}
				}
			}
		}
	}
	close(datac)
	<-donec

	if err != nil {
		return err
	}
	if r.Err() != nil {
		return errors.Wrap(r.Err(), "read entry")
	}
	return nil
}