blob: b1b1f1fc89ad175cd73994c8394ed1c30c68f6fd (
plain)
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
|
// Copyright 2015 syzkaller project authors. All rights reserved.
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
package ipc
import (
"sync"
)
// Gate limits concurrency level and window to the given value.
// Limitation of concurrency window means that if a very old activity is still
// running it will not let new activities to start even if concurrency level is low.
type Gate struct {
cv *sync.Cond
busy []bool
pos int
running int
stop bool
f func()
}
// If f is not nil, it will be called after each batch of c activities.
func NewGate(c int, f func()) *Gate {
return &Gate{
cv: sync.NewCond(new(sync.Mutex)),
busy: make([]bool, c),
f: f,
}
}
func (g *Gate) Enter() int {
g.cv.L.Lock()
for g.busy[g.pos] || g.stop {
g.cv.Wait()
}
idx := g.pos
g.pos++
if g.pos >= len(g.busy) {
g.pos = 0
}
g.busy[idx] = true
g.running++
if g.running > len(g.busy) {
panic("broken gate")
}
g.cv.L.Unlock()
return idx
}
func (g *Gate) Leave(idx int) {
g.cv.L.Lock()
if !g.busy[idx] {
panic("broken gate")
}
g.busy[idx] = false
g.running--
if g.running < 0 {
panic("broken gate")
}
if idx == 0 && g.f != nil {
if g.stop {
panic("broken gate")
}
g.stop = true
for g.running != 0 {
g.cv.Wait()
}
g.stop = false
g.f()
g.cv.Broadcast()
}
if idx == g.pos && !g.stop || g.running == 0 && g.stop {
g.cv.Broadcast()
}
g.cv.L.Unlock()
}
|
