cubefs

1
// Copyright 2023 The CubeFS 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
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package datanode
16

17
import (
18
	"testing"
19
	"time"
20

21
	"github.com/stretchr/testify/require"
22
)
23

24
func TestLimitIOBase(t *testing.T) {
25
	f := func() {}
26
	for _, flowIO := range [][2]int{
27
		{-1, -1},
28
		{0, -1},
29
		{-1, 0},
30
		{100, -1},
31
		{1 << 20, 4},
32
	} {
33
		l := newIOLimiter(flowIO[0], flowIO[1])
34
		l.ResetFlow(flowIO[0])
35
		l.ResetIO(flowIO[1])
36
		l.Run(0, f)
37
		l.Run(10, f)
38
		require.True(t, l.TryRun(1, f))
39
		l.Close()
40
	}
41

42
	{
43
		l := newIOLimiter(1<<10, 0)
44
		l.Run(10, f)
45
		st := l.Status()
46
		t.Logf("status: %+v", st)
47
		require.Equal(t, 1<<10, st.FlowLimit)
48
		require.True(t, st.FlowUsed > 0)
49
		require.True(t, st.FlowUsed <= 10)
50
		require.True(t, l.TryRun(10, f))
51
		l.Run(1<<20, f)
52
		l.TryRun(1<<20, f)
53
		l.Close()
54
	}
55
	{
56
		done := make(chan struct{})
57
		l := newIOLimiter(-1, 2)
58
		st := l.Status()
59
		t.Logf("before status: %+v", st)
60
		for ii := 0; ii < st.IOConcurrency; ii++ {
61
			go func() {
62
				l.Run(0, func() { <-done })
63
			}()
64
		}
65
		for ii := 0; ii < st.IOQueue*2; ii++ {
66
			go func() {
67
				l.Run(0, func() { <-done })
68
			}()
69
		}
70
		time.Sleep(100 * time.Millisecond)
71
		t.Logf("after status: %+v", l.Status())
72
		require.False(t, l.TryRun(0, f))
73
		close(done)
74
		q := l.getIO()
75
		l.Close()
76
		q.Run(f)
77
		require.True(t, q.TryRun(f))
78
		t.Logf("closed status: %+v", q.Status())
79
	}
80
}
81

82
func TestLimitIOConcurrency(t *testing.T) {
83
	l := newIOLimiter(1<<10, 10)
84
	done := make(chan struct{})
85
	go func() {
86
		for {
87
			select {
88
			case <-done:
89
				return
90
			default:
91
			}
92

93
			time.Sleep(time.Microsecond)
94
			l.ResetFlow(1 << 10)
95
			l.ResetIO(10)
96
		}
97
	}()
98

99
	go func() {
100
		for {
101
			select {
102
			case <-done:
103
				return
104
			default:
105
			}
106

107
			time.Sleep(time.Microsecond)
108
			go func() {
109
				l.Run(1, func() { <-done })
110
			}()
111
		}
112
	}()
113

114
	go func() {
115
		for {
116
			select {
117
			case <-done:
118
				return
119
			default:
120
			}
121

122
			time.Sleep(time.Microsecond)
123
			go func() {
124
				l.TryRun(1, func() { <-done })
125
			}()
126
		}
127
	}()
128

129
	time.Sleep(500 * time.Millisecond)
130
	close(done)
131
	l.Close()
132
}
133