blob: 1331e65f98a81f0c02a8f8cce191338b638a41ec (
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
77
78
79
|
// Copyright 2020 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package par
import (
"sync"
"testing"
)
func TestQueueIdle(t *testing.T) {
q := NewQueue(1)
select {
case <-q.Idle():
default:
t.Errorf("NewQueue(1) is not initially idle.")
}
started := make(chan struct{})
unblock := make(chan struct{})
q.Add(func() {
close(started)
<-unblock
})
<-started
idle := q.Idle()
select {
case <-idle:
t.Errorf("NewQueue(1) is marked idle while processing work.")
default:
}
close(unblock)
<-idle // Should be closed as soon as the Add callback returns.
}
func TestQueueBacklog(t *testing.T) {
const (
maxActive = 2
totalWork = 3 * maxActive
)
q := NewQueue(maxActive)
t.Logf("q = NewQueue(%d)", maxActive)
var wg sync.WaitGroup
wg.Add(totalWork)
started := make([]chan struct{}, totalWork)
unblock := make(chan struct{})
for i := range started {
started[i] = make(chan struct{})
i := i
q.Add(func() {
close(started[i])
<-unblock
wg.Done()
})
}
for i, c := range started {
if i < maxActive {
<-c // Work item i should be started immediately.
} else {
select {
case <-c:
t.Errorf("Work item %d started before previous items finished.", i)
default:
}
}
}
close(unblock)
for _, c := range started[maxActive:] {
<-c
}
wg.Wait()
}
|
