-
Notifications
You must be signed in to change notification settings - Fork 336
/
throttle.go
74 lines (67 loc) · 1.65 KB
/
throttle.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
package nutsdb
import (
"sync"
)
// Throttle allows a limited number of workers to run at a time. It also
// provides a mechanism to check for errors encountered by workers and wait for
// them to finish.
type Throttle struct {
once sync.Once
wg sync.WaitGroup
ch chan struct{}
errCh chan error
finishErr error
}
// NewThrottle creates a new throttle with a max number of workers.
func NewThrottle(max int) *Throttle {
return &Throttle{
ch: make(chan struct{}, max),
errCh: make(chan error, max),
}
}
// Do should be called by workers before they start working. It blocks if there
// are already maximum number of workers working. If it detects an error from
// previously Done workers, it would return it.
func (t *Throttle) Do() error {
for {
select {
case t.ch <- struct{}{}:
t.wg.Add(1)
return nil
case err := <-t.errCh:
if err != nil {
return err
}
}
}
}
// Done should be called by workers when they finish working. They can also
// pass the error status of work done.
func (t *Throttle) Done(err error) {
if err != nil {
t.errCh <- err
}
select {
case <-t.ch:
default:
panic("Throttle Do Done mismatch")
}
t.wg.Done()
}
// Finish waits until all workers have finished working. It would return any error passed by Done.
// If Finish is called multiple time, it will wait for workers to finish only once(first time).
// From next calls, it will return same error as found on first call.
func (t *Throttle) Finish() error {
t.once.Do(func() {
t.wg.Wait()
close(t.ch)
close(t.errCh)
for err := range t.errCh {
if err != nil {
t.finishErr = err
return
}
}
})
return t.finishErr
}