Function benchmarkWorkerPool

pkg/workerpool/workerpool_test.go:222–275

(b *testing.B, processingDelay time.Duration, publishingDelay time.Duration)

}

func benchmarkWorkerPool(b *testing.B, processingDelay time.Duration, publishingDelay time.Duration) {
	_, ctx := test.New(b)

	var totalQueueDelayMS int64
	var totalHandled int64
	var published, dropped int64

	for r := 0; r < b.N; r++ {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()

		handler := func(ctx context.Context, tm time.Time) {
			delay := time.Now().Sub(tm).Milliseconds()
			atomic.AddInt64(&totalQueueDelayMS, delay)
			atomic.AddInt64(&totalHandled, 1)

			time.Sleep(random.Jitter(processingDelay, 0.15))
		}

		wp := workerpool.NewWorkerPool(workerpool.Config[time.Time]{
			Component: &mockComponent{},
			Context:   ctx,
			Handler:   handler,
		})

		var wg sync.WaitGroup
		publisher := func() {
			defer wg.Done()
			for range 1_000 {
				if err := wp.Publish(ctx, time.Now()); err != nil {
					atomic.AddInt64(&dropped, 1)
				} else {
					atomic.AddInt64(&published, 1)
				}
				time.Sleep(random.Jitter(publishingDelay, 0.15))
			}
		}

		for i := 0; i < workerpool.DefaultMaxWorkers; i++ {
			wg.Add(1)
			go publisher()
		}

		wg.Wait()

		time.Sleep(testTimeout)
		cancel()
		wp.Wait()
	}

	b.ReportMetric(float64(totalQueueDelayMS)/float64(totalHandled), "queueDelayMS")
	b.ReportMetric(float64(published), "published")
	b.ReportMetric(float64(dropped), "dropped")
}

func BenchmarkWorkerPool(b *testing.B) {
	delays := []time.Duration{5 * time.Millisecond, 10 * time.Millisecond, 50 * time.Millisecond}