queue/observability_benchmark_test.go at main · goforj/queue · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
package queue

import (
	"context"
	"testing"
	"time"
)

func BenchmarkStatsCollectorObserve(b *testing.B) {
	collector := NewStatsCollector()
	b.ReportAllocs()
	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		now := time.Now()
		jobKey := "bench-job"
		collector.Observe(context.Background(), Event{
			Kind:   EventEnqueueAccepted,
			Driver: DriverSync,
			Queue:  "default",
			JobKey: jobKey,
			Time:   now,
		})
		collector.Observe(context.Background(), Event{
			Kind:   EventProcessStarted,
			Driver: DriverSync,
			Queue:  "default",
			JobKey: jobKey,
			Time:   now.Add(1 * time.Millisecond),
		})
		collector.Observe(context.Background(), Event{
			Kind:     EventProcessSucceeded,
			Driver:   DriverSync,
			Queue:    "default",
			JobKey:   jobKey,
			Duration: 2 * time.Millisecond,
			Time:     now.Add(3 * time.Millisecond),
		})
	}
}

func BenchmarkEnqueueSync_NoObserver(b *testing.B) {
	q, err := newRuntime(Config{Driver: DriverSync})
	if err != nil {
		b.Fatalf("new queue failed: %v", err)
	}
	q.Register("job:bench:no-observer", func(context.Context, Job) error { return nil })
	if err := q.Workers(1).StartWorkers(context.Background()); err != nil {
		b.Fatalf("start workers failed: %v", err)
	}
	defer q.Shutdown(context.Background())
	job := NewJob("job:bench:no-observer").OnQueue("default")

	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		if err := q.Dispatch(job); err != nil {
			b.Fatalf("dispatch failed: %v", err)
		}
	}
}

func BenchmarkEnqueueSync_WithObserver(b *testing.B) {
	collector := NewStatsCollector()
	q, err := newRuntime(Config{
		Driver:   DriverSync,
		Observer: collector,
	})
	if err != nil {
		b.Fatalf("new queue failed: %v", err)
	}
	q.Register("job:bench:with-observer", func(context.Context, Job) error { return nil })
	if err := q.Workers(1).StartWorkers(context.Background()); err != nil {
		b.Fatalf("start workers failed: %v", err)
	}
	defer q.Shutdown(context.Background())
	job := NewJob("job:bench:with-observer").OnQueue("default")

	b.ReportAllocs()
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		if err := q.Dispatch(job); err != nil {
			b.Fatalf("dispatch failed: %v", err)
		}
	}
}