istio

1
// Copyright Istio Authors
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14

15
package queue
16

17
import (
18
	"errors"
19
	"sync"
20
	"testing"
21
	"time"
22

23
	"go.uber.org/atomic"
24

25
	"istio.io/istio/pkg/test/util/assert"
26
	"istio.io/istio/pkg/test/util/retry"
27
)
28

29
func TestOrdering(t *testing.T) {
30
	numValues := 1000
31

32
	q := NewQueue(1 * time.Microsecond)
33
	stop := make(chan struct{})
34
	defer close(stop)
35

36
	wg := sync.WaitGroup{}
37
	wg.Add(numValues)
38
	mu := sync.Mutex{}
39
	out := make([]int, 0)
40
	for i := 0; i < numValues; i++ {
41
		i := i
42

43
		q.Push(func() error {
44
			mu.Lock()
45
			out = append(out, i)
46
			defer mu.Unlock()
47
			wg.Done()
48
			return nil
49
		})
50

51
		// Start the queue at the halfway point.
52
		if i == numValues/2 {
53
			go q.Run(stop)
54
		}
55
	}
56

57
	// wait for all task processed
58
	wg.Wait()
59

60
	if len(out) != numValues {
61
		t.Fatalf("expected output array length %d to equal %d", len(out), numValues)
62
	}
63

64
	for i := 0; i < numValues; i++ {
65
		if i != out[i] {
66
			t.Fatalf("expected out[%d] %v to equal %v", i, out[i], i)
67
		}
68
	}
69
}
70

71
func TestRetry(t *testing.T) {
72
	q := NewQueue(1 * time.Microsecond)
73
	stop := make(chan struct{})
74
	defer close(stop)
75

76
	// Push a task that fails the first time and retries.
77
	wg := sync.WaitGroup{}
78
	wg.Add(2)
79
	failed := false
80
	q.Push(func() error {
81
		defer wg.Done()
82
		if failed {
83
			return nil
84
		}
85
		failed = true
86
		return errors.New("fake error")
87
	})
88

89
	go q.Run(stop)
90

91
	// wait for the task to run twice.
92
	wg.Wait()
93
}
94

95
func TestResourceFree(t *testing.T) {
96
	q := NewQueue(1 * time.Microsecond)
97
	stop := make(chan struct{})
98
	signal := make(chan struct{})
99
	go func() {
100
		q.Run(stop)
101
		signal <- struct{}{}
102
	}()
103

104
	q.Push(func() error {
105
		t.Log("mock exec")
106
		return nil
107
	})
108

109
	// mock queue block wait cond signal
110
	time.AfterFunc(10*time.Millisecond, func() {
111
		close(stop)
112
	})
113

114
	select {
115
	case <-time.After(200 * time.Millisecond):
116
		t.Error("close stop, method exit timeout.")
117
	case <-signal:
118
		t.Log("queue return.")
119
	}
120
}
121

122
func TestClosed(t *testing.T) {
123
	t.Run("immediate close", func(t *testing.T) {
124
		stop := make(chan struct{})
125
		q := NewQueue(0)
126
		go q.Run(stop)
127
		close(stop)
128
		if err := WaitForClose(q, 10*time.Second); err != nil {
129
			t.Error(err)
130
		}
131
	})
132
	t.Run("no tasks after close", func(t *testing.T) {
133
		stop := make(chan struct{})
134
		q := NewQueue(0)
135
		taskComplete := atomic.NewBool(false)
136
		q.Push(func() error {
137
			close(stop)
138
			return nil
139
		})
140
		go q.Run(stop)
141
		if err := WaitForClose(q, 10*time.Second); err != nil {
142
			t.Error()
143
		}
144
		q.Push(func() error {
145
			taskComplete.Store(true)
146
			return nil
147
		})
148
		if taskComplete.Load() {
149
			t.Error("task ran on closed queue")
150
		}
151
	})
152
}
153

154
func TestSync(t *testing.T) {
155
	handles := atomic.NewInt32(0)
156
	task := func() error {
157
		handles.Inc()
158
		return nil
159
	}
160
	q := NewQueue(0)
161
	q.Push(task)
162
	stop := make(chan struct{})
163
	go q.Run(stop)
164
	retry.UntilOrFail(t, q.HasSynced, retry.Delay(time.Microsecond))
165
	// Must always be 1 since we are synced
166
	assert.Equal(t, handles.Load(), 1)
167
	close(stop)
168
	assert.NoError(t, WaitForClose(q, time.Second))
169
}
170