Function TestHead_HighConcurrencyReadAndWrite

tsdb/head_test.go:599–797

TestHead_HighConcurrencyReadAndWrite generates 1000 series with a step of 15s and fills a whole block with samples, this means in total it generates 4000 chunks because with a step of 15s there are 4 chunks per block per series. While appending the samples to the head it concurrently queries them fr

(t *testing.T)

// While appending the samples to the head it concurrently queries them from multiple go routines and verifies that the
// returned results are correct.
func TestHead_HighConcurrencyReadAndWrite(t *testing.T) {
	for _, appV2 := range []bool{false, true} {
		t.Run(fmt.Sprintf("appV2=%v", appV2), func(t *testing.T) {
			head, _ := newTestHead(t, DefaultBlockDuration, compression.None, false)

			seriesCnt := 1000
			readConcurrency := 2
			writeConcurrency := 10
			startTs := uint64(DefaultBlockDuration) // Start at the second block relative to the unix epoch.
			qryRange := uint64(5 * time.Minute.Milliseconds())
			step := uint64(15 * time.Second / time.Millisecond)
			endTs := startTs + uint64(DefaultBlockDuration)

			labelSets := make([]labels.Labels, seriesCnt)
			for i := range seriesCnt {
				labelSets[i] = labels.FromStrings("seriesId", strconv.Itoa(i))
			}
			require.NoError(t, head.Init(0))

			g, ctx := errgroup.WithContext(t.Context())
			whileNotCanceled := func(f func() (bool, error)) error {
				for ctx.Err() == nil {
					cont, err := f()
					if err != nil {
						return err
					}
					if !cont {
						return nil
					}
				}
				return nil
			}

			// Create one channel for each write worker, the channels will be used by the coordinator
			// go routine to coordinate which timestamps each write worker has to write.
			writerTsCh := make([]chan uint64, writeConcurrency)
			for writerTsChIdx := range writerTsCh {
				writerTsCh[writerTsChIdx] = make(chan uint64)
			}

			// workerReadyWg is used to synchronize the start of the test,
			// we only start the test once all workers signal that they're ready.
			var workerReadyWg sync.WaitGroup
			workerReadyWg.Add(writeConcurrency + readConcurrency)

			// Start the write workers.
			for wid := range writeConcurrency {
				// Create copy of workerID to be used by worker routine.
				workerID := wid

				g.Go(func() error {
					// The label sets which this worker will write.
					workerLabelSets := labelSets[(seriesCnt/writeConcurrency)*workerID : (seriesCnt/writeConcurrency)*(workerID+1)]

					// Signal that this worker is ready.
					workerReadyWg.Done()

					return whileNotCanceled(func() (bool, error) {