-
Notifications
You must be signed in to change notification settings - Fork 0
/
worker.go
121 lines (97 loc) · 2.36 KB
/
worker.go
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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
package internal
import (
"context"
"fmt"
"runtime"
"github.com/symphony09/ograph/ogcore"
"golang.org/x/sync/errgroup"
)
type Worker struct {
graph *Graph[ogcore.Node]
}
type WorkParams struct {
GorLimit int
ActionsBeforeRun map[string]ogcore.Action
ActionsAfterRun map[string]ogcore.Action
}
func (worker *Worker) Work(ctx context.Context, state ogcore.State, params *WorkParams) error {
todoCh, doneCh := worker.graph.Scheduling()
defer close(doneCh)
doWorks := func(works []*GraphVertex[ogcore.Node]) (err error) {
var currentWorkName string
defer func() {
if info := recover(); info != nil {
err = fmt.Errorf("worker panic on %s, info: %v", currentWorkName, info)
}
doneCh <- works
}()
for _, work := range works {
if ctx.Err() != nil {
return ctx.Err()
}
currentWorkName = work.Name
node := work.Elem
if params.ActionsBeforeRun != nil {
if action := params.ActionsBeforeRun[work.Name]; action != nil {
if err := action(ctx, state); err != nil {
return fmt.Errorf("%s failed, error: %w", work.Name, err)
}
}
}
if node != nil {
if err := node.Run(ctx, state); err != nil {
return fmt.Errorf("%s failed, error: %w", work.Name, err)
}
}
if params.ActionsAfterRun != nil {
if action := params.ActionsAfterRun[work.Name]; action != nil {
if err := action(ctx, state); err != nil {
return fmt.Errorf("%s failed, error: %w", work.Name, err)
}
}
}
}
return nil
}
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(params.GorLimit)
pn := runtime.GOMAXPROCS(0)
if params.GorLimit > 0 && params.GorLimit <= pn {
for i := 0; i < params.GorLimit; i++ {
g.Go(func() (err error) {
for works := range todoCh {
if err := doWorks(works); err != nil {
return err
}
}
return nil
})
}
} else {
if worker.graph.ScheduleNum > pn {
for i := 0; i < pn; i++ {
g.Go(func() (err error) {
for works := range todoCh {
if err := doWorks(works); err != nil {
return err
}
}
return nil
})
}
}
for works := range todoCh {
works := works // https://golang.org/doc/faq#closures_and_goroutines
g.Go(func() (err error) {
return doWorks(works)
})
}
}
err := g.Wait()
return err
}
func NewWorker(graph *Graph[ogcore.Node]) *Worker {
return &Worker{
graph: graph,
}
}