forked from marianogappa/parseq
-
Notifications
You must be signed in to change notification settings - Fork 0
/
parseq_test.go
105 lines (91 loc) · 2.14 KB
/
parseq_test.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
package parseq
import (
"sync"
"testing"
"time"
"math/rand"
)
func TestOutperformsSequential(t *testing.T) {
p, err := New(5, processAfter(50*time.Millisecond))
if err != nil {
panic(err)
}
go p.Start()
go func() {
p.Input <- 666
time.Sleep(10 * time.Millisecond)
p.Input <- 667
time.Sleep(10 * time.Millisecond)
p.Input <- 668
time.Sleep(10 * time.Millisecond)
p.Input <- 669
time.Sleep(10 * time.Millisecond)
p.Input <- 670
}()
start := time.Now()
<-p.Output // min(elapsed)=50ms
<-p.Output // min(elapsed)=60ms
<-p.Output // min(elapsed)=70ms
<-p.Output // min(elapsed)=80ms
<-p.Output // min(elapsed)=90ms
elapsed := time.Now().Sub(start)
if elapsed > 150*time.Millisecond { // 150ms for
t.Error("test took too long; parallel strategy is ineffective!")
}
t.Log("min(elapsed) was 90ms; took", elapsed, "against sequential time of 250ms")
p.Close()
}
func TestOrderedOutput(t *testing.T) {
r := rand.New(rand.NewSource(99))
p, err := New(5, processAfterRandom(r))
if err != nil {
panic(err)
}
go p.Start()
go func() {
p.Input <- 666
time.Sleep(10 * time.Millisecond)
p.Input <- 667
time.Sleep(10 * time.Millisecond)
p.Input <- 668
time.Sleep(10 * time.Millisecond)
p.Input <- 669
time.Sleep(10 * time.Millisecond)
p.Input <- 670
p.Close()
}()
a := <-p.Output
b := <-p.Output
c := <-p.Output
d := <-p.Output
time.Sleep(10 * time.Millisecond)
e := <-p.Output
if a.(int) != 666 ||
b.(int) != 667 ||
c.(int) != 668 ||
d.(int) != 669 ||
e.(int) != 670 {
t.Error("output came out out of order: ", a, b, c, d, e)
}
}
func processAfter(d time.Duration) processFuncGenerator {
return processGenerator(func(v interface{}) interface{} {
time.Sleep(d)
return v
})
}
func processAfterRandom(r *rand.Rand) processFuncGenerator {
var mu sync.Mutex
return processGenerator(func(v interface{}) interface{} {
mu.Lock()
rnd := r.Intn(41)
mu.Unlock()
time.Sleep(time.Duration(rnd+10) * time.Millisecond) //sleep between 10ms and 50ms
return v
})
}
func processGenerator(f func(interface{}) interface{}) processFuncGenerator {
return func(i int) (ProcessFunc, error) {
return f, nil
}
}