1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.regionserver;
21
22 import org.apache.commons.logging.Log;
23 import org.apache.commons.logging.LogFactory;
24 import org.junit.After;
25 import org.junit.Assert;
26 import org.junit.Before;
27 import org.junit.Test;
28
29
30
31
32 public class TestPriorityCompactionQueue {
33 static final Log LOG = LogFactory.getLog(TestPriorityCompactionQueue.class);
34
35 @Before
36 public void setUp() {
37 }
38
39 @After
40 public void tearDown() {
41
42 }
43
44 class DummyHRegion extends HRegion {
45 String name;
46
47 DummyHRegion(String name) {
48 super();
49 this.name = name;
50 }
51
52 public int hashCode() {
53 return name.hashCode();
54 }
55
56 public boolean equals(DummyHRegion r) {
57 return name.equals(r.name);
58 }
59
60 public String toString() {
61 return "[DummyHRegion " + name + "]";
62 }
63
64 public String getRegionNameAsString() {
65 return name;
66 }
67 }
68
69 protected void getAndCheckRegion(PriorityCompactionQueue pq,
70 HRegion checkRegion) {
71 HRegion r = pq.remove();
72 if (r != checkRegion) {
73 Assert.assertTrue("Didn't get expected " + checkRegion + " got " + r, r
74 .equals(checkRegion));
75 }
76 }
77
78 protected void addRegion(PriorityCompactionQueue pq, HRegion r, int p) {
79 pq.add(r, p);
80 try {
81
82
83
84
85 Thread.sleep(1);
86 } catch (InterruptedException ex) {
87
88 }
89 }
90
91
92
93
94
95
96 @Test public void testPriorityQueue() throws InterruptedException {
97 PriorityCompactionQueue pq = new PriorityCompactionQueue();
98
99 HRegion r1 = new DummyHRegion("r1");
100 HRegion r2 = new DummyHRegion("r2");
101 HRegion r3 = new DummyHRegion("r3");
102 HRegion r4 = new DummyHRegion("r4");
103 HRegion r5 = new DummyHRegion("r5");
104
105
106
107 addRegion(pq, r1, 0);
108 addRegion(pq, r2, 0);
109 addRegion(pq, r3, 0);
110 addRegion(pq, r4, 0);
111 addRegion(pq, r5, 0);
112
113 getAndCheckRegion(pq, r1);
114 getAndCheckRegion(pq, r2);
115 getAndCheckRegion(pq, r3);
116 getAndCheckRegion(pq, r4);
117 getAndCheckRegion(pq, r5);
118
119
120
121 addRegion(pq, r1, 0);
122 addRegion(pq, r2, CompactSplitThread.PRIORITY_USER);
123 addRegion(pq, r3, 0);
124 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
125 addRegion(pq, r5, 0);
126
127 getAndCheckRegion(pq, r1);
128 getAndCheckRegion(pq, r3);
129 getAndCheckRegion(pq, r5);
130 getAndCheckRegion(pq, r2);
131 getAndCheckRegion(pq, r4);
132
133
134
135 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
136 addRegion(pq, r2, CompactSplitThread.PRIORITY_USER);
137 addRegion(pq, r3, CompactSplitThread.PRIORITY_USER);
138 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
139 addRegion(pq, r5, 0);
140
141 getAndCheckRegion(pq, r5);
142 getAndCheckRegion(pq, r1);
143 getAndCheckRegion(pq, r2);
144 getAndCheckRegion(pq, r3);
145 getAndCheckRegion(pq, r4);
146
147
148
149 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
150 addRegion(pq, r2, 0);
151 addRegion(pq, r3, CompactSplitThread.PRIORITY_USER);
152 Thread.sleep(1000);
153 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
154 addRegion(pq, r5, 0);
155
156 getAndCheckRegion(pq, r2);
157 getAndCheckRegion(pq, r5);
158 getAndCheckRegion(pq, r1);
159 getAndCheckRegion(pq, r3);
160 getAndCheckRegion(pq, r4);
161
162
163 pq = new PriorityCompactionQueue();
164
165
166
167
168 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
169 addRegion(pq, r2, CompactSplitThread.PRIORITY_USER);
170 addRegion(pq, r3, CompactSplitThread.PRIORITY_USER);
171 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
172 addRegion(pq, r5, CompactSplitThread.PRIORITY_USER);
173 addRegion(pq, r3, 0);
174
175 getAndCheckRegion(pq, r3);
176 getAndCheckRegion(pq, r1);
177 getAndCheckRegion(pq, r2);
178 getAndCheckRegion(pq, r4);
179 getAndCheckRegion(pq, r5);
180
181 Assert.assertTrue("Queue should be empty.", pq.size() == 0);
182
183
184
185 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
186 addRegion(pq, r2, CompactSplitThread.PRIORITY_USER);
187 addRegion(pq, r3, CompactSplitThread.PRIORITY_USER);
188 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
189 addRegion(pq, r5, CompactSplitThread.PRIORITY_USER);
190 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
191 addRegion(pq, r2, CompactSplitThread.PRIORITY_USER);
192 addRegion(pq, r3, CompactSplitThread.PRIORITY_USER);
193 addRegion(pq, r4, CompactSplitThread.PRIORITY_USER);
194 addRegion(pq, r5, CompactSplitThread.PRIORITY_USER);
195
196 getAndCheckRegion(pq, r1);
197 getAndCheckRegion(pq, r2);
198 getAndCheckRegion(pq, r3);
199 getAndCheckRegion(pq, r4);
200 getAndCheckRegion(pq, r5);
201
202 Assert.assertTrue("Queue should be empty.", pq.size() == 0);
203
204
205
206 addRegion(pq, r1, CompactSplitThread.PRIORITY_USER);
207 addRegion(pq, r2, -1);
208 addRegion(pq, r3, 0);
209 addRegion(pq, r4, -2);
210
211 getAndCheckRegion(pq, r4);
212 getAndCheckRegion(pq, r2);
213 getAndCheckRegion(pq, r3);
214 getAndCheckRegion(pq, r1);
215
216 Assert.assertTrue("Queue should be empty.", pq.size() == 0);
217 }
218 }