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
|
package flamegraph
import (
"context"
"fmt"
"ior/internal/event"
"runtime"
"sync"
)
type IorDataCollector struct {
flamegraphName string
Ch chan *event.Pair
Done chan error
workers []worker
}
func New(flamegraphName ...string) IorDataCollector {
name := "default"
if len(flamegraphName) > 0 && flamegraphName[0] != "" {
name = flamegraphName[0]
}
f := IorDataCollector{
flamegraphName: name,
Ch: make(chan *event.Pair, 4096),
Done: make(chan error, 1),
}
numWorkers := runtime.NumCPU() / 4
if numWorkers == 0 {
numWorkers = 1
}
for range numWorkers {
f.workers = append(f.workers, newWorker())
}
return f
}
func (f IorDataCollector) Start(ctx context.Context) {
go func() {
defer close(f.Done)
var wg sync.WaitGroup
wg.Add(len(f.workers))
for i, worker := range f.workers {
fmt.Println("Starting flamegraph worker", i)
go worker.run(ctx, &wg, f.Ch)
}
wg.Wait()
iod := f.workers[0].iod
if len(f.workers) > 1 {
for i, w := range f.workers[1:] {
iod = iod.merge(w.iod)
fmt.Println("Worker", i+1, "merged")
}
}
if err := iod.serializeToFile(f.flamegraphName); err != nil {
f.Done <- err
return
}
f.Done <- nil
}()
}
|
