Function benchmarkGetNextForQuerier

modules/frontend/queue/queue_test.go:131–164

(b *testing.B, listeners int, messages int)

}

func benchmarkGetNextForQuerier(b *testing.B, listeners int, messages int) {
	const user = "user"

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	stop := make(chan struct{})
	requestsPulled := atomic.NewInt32(0)

	q, start := queueWithListeners(ctx, listeners, 1, func(_ []Request) {
		if requestsPulled.Inc() == int32(messages) {
			stop <- struct{}{}
		}
	})
	close(start)

	req := &mockRequest{}
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		for j := 0; j < messages; j++ {
			err := q.EnqueueRequest(user, req)
			if err != nil {
				panic(err)
			}
		}

		<-stop
		requestsPulled.Sub(int32(messages))
	}

	err := q.stopping(nil)
	if err != nil {
		panic(err)
	}
}

func queueWithListeners(ctx context.Context, listeners int, batchSize int, listenerFn func(r []Request)) (*RequestQueue, chan struct{}) {
	g := prometheus.NewGaugeVec(prometheus.GaugeOpts{