1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.ipc;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.junit.Assert.assertFalse;
22 import static org.junit.Assert.assertNotEquals;
23 import static org.junit.Assert.assertTrue;
24 import static org.mockito.Matchers.any;
25 import static org.mockito.Matchers.anyObject;
26 import static org.mockito.Matchers.eq;
27 import static org.mockito.Mockito.doAnswer;
28 import static org.mockito.Mockito.mock;
29 import static org.mockito.Mockito.timeout;
30 import static org.mockito.Mockito.verify;
31 import static org.mockito.Mockito.when;
32
33 import java.io.IOException;
34 import java.net.InetSocketAddress;
35 import java.util.ArrayList;
36 import java.util.HashSet;
37 import java.util.List;
38 import java.util.Map;
39 import java.util.Set;
40 import java.util.concurrent.BlockingQueue;
41 import java.util.concurrent.CountDownLatch;
42 import java.util.concurrent.LinkedBlockingQueue;
43 import java.util.concurrent.ThreadLocalRandom;
44
45 import org.apache.commons.logging.Log;
46 import org.apache.commons.logging.LogFactory;
47 import org.apache.hadoop.conf.Configuration;
48 import org.apache.hadoop.hbase.HBaseConfiguration;
49 import org.apache.hadoop.hbase.HConstants;
50 import org.apache.hadoop.hbase.client.Put;
51 import org.apache.hadoop.hbase.ipc.RpcServer.Call;
52 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl;
53 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
54 import org.apache.hadoop.hbase.protobuf.RequestConverter;
55 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos;
56 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
57 import org.apache.hadoop.hbase.security.User;
58 import org.apache.hadoop.hbase.testclassification.SmallTests;
59 import org.apache.hadoop.hbase.util.Bytes;
60 import org.apache.hadoop.hbase.util.EnvironmentEdge;
61 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
62 import org.apache.hadoop.hbase.util.Threads;
63 import org.junit.Before;
64 import org.junit.Test;
65 import org.junit.experimental.categories.Category;
66 import org.mockito.invocation.InvocationOnMock;
67 import org.mockito.stubbing.Answer;
68
69 import com.google.common.collect.ImmutableList;
70 import com.google.common.collect.ImmutableMap;
71 import com.google.common.collect.ImmutableSet;
72 import com.google.common.collect.Maps;
73 import com.google.protobuf.Message;
74
75 @Category(SmallTests.class)
76 public class TestSimpleRpcScheduler {
77 private static final Log LOG = LogFactory.getLog(TestSimpleRpcScheduler.class);
78
79 private final RpcScheduler.Context CONTEXT = new RpcScheduler.Context() {
80 @Override
81 public InetSocketAddress getListenerAddress() {
82 return InetSocketAddress.createUnresolved("127.0.0.1", 1000);
83 }
84 };
85 private Configuration conf;
86
87 @Before
88 public void setUp() {
89 conf = HBaseConfiguration.create();
90 }
91
92 @Test
93 public void testBasic() throws IOException, InterruptedException {
94 PriorityFunction qosFunction = mock(PriorityFunction.class);
95 RpcScheduler scheduler = new SimpleRpcScheduler(
96 conf, 10, 0, 0, qosFunction, 0);
97 scheduler.init(CONTEXT);
98 scheduler.start();
99 CallRunner task = createMockTask();
100 task.setStatus(new MonitoredRPCHandlerImpl());
101 scheduler.dispatch(task);
102 verify(task, timeout(1000)).run();
103 scheduler.stop();
104 }
105
106 @Test
107 public void testHandlerIsolation() throws IOException, InterruptedException {
108 CallRunner generalTask = createMockTask();
109 CallRunner priorityTask = createMockTask();
110 CallRunner replicationTask = createMockTask();
111 List<CallRunner> tasks = ImmutableList.of(
112 generalTask,
113 priorityTask,
114 replicationTask);
115 Map<CallRunner, Integer> qos = ImmutableMap.of(
116 generalTask, 0,
117 priorityTask, HConstants.HIGH_QOS + 1,
118 replicationTask, HConstants.REPLICATION_QOS);
119 PriorityFunction qosFunction = mock(PriorityFunction.class);
120 final Map<CallRunner, Thread> handlerThreads = Maps.newHashMap();
121 final CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
122 Answer<Void> answerToRun = new Answer<Void>() {
123 @Override
124 public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
125 synchronized (handlerThreads) {
126 handlerThreads.put(
127 (CallRunner) invocationOnMock.getMock(),
128 Thread.currentThread());
129 }
130 countDownLatch.countDown();
131 return null;
132 }
133 };
134 for (CallRunner task : tasks) {
135 task.setStatus(new MonitoredRPCHandlerImpl());
136 doAnswer(answerToRun).when(task).run();
137 }
138
139 RpcScheduler scheduler = new SimpleRpcScheduler(
140 conf, 1, 1 ,1, qosFunction, HConstants.HIGH_QOS);
141 scheduler.init(CONTEXT);
142 scheduler.start();
143 for (CallRunner task : tasks) {
144 when(qosFunction.getPriority((RPCProtos.RequestHeader) anyObject(),
145 (Message) anyObject(), (User) anyObject()))
146 .thenReturn(qos.get(task));
147 scheduler.dispatch(task);
148 }
149 for (CallRunner task : tasks) {
150 verify(task, timeout(1000)).run();
151 }
152 scheduler.stop();
153
154
155 countDownLatch.await();
156 assertEquals(3, ImmutableSet.copyOf(handlerThreads.values()).size());
157 }
158
159 private CallRunner createMockTask() {
160 Call call = mock(Call.class);
161 CallRunner task = mock(CallRunner.class);
162 when(task.getCall()).thenReturn(call);
163 return task;
164 }
165
166 @Test
167 public void testRpcScheduler() throws Exception {
168 testRpcScheduler(SimpleRpcScheduler.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
169 testRpcScheduler(SimpleRpcScheduler.CALL_QUEUE_TYPE_FIFO_CONF_VALUE);
170 }
171
172 private void testRpcScheduler(final String queueType) throws Exception {
173 Configuration schedConf = HBaseConfiguration.create();
174 schedConf.set(SimpleRpcScheduler.CALL_QUEUE_TYPE_CONF_KEY, queueType);
175
176 PriorityFunction priority = mock(PriorityFunction.class);
177 when(priority.getPriority(any(RequestHeader.class),
178 any(Message.class), any(User.class)))
179 .thenReturn(HConstants.NORMAL_QOS);
180
181 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 1, 1, 1, priority,
182 HConstants.QOS_THRESHOLD);
183 try {
184 scheduler.start();
185
186 CallRunner smallCallTask = mock(CallRunner.class);
187 RpcServer.Call smallCall = mock(RpcServer.Call.class);
188 RequestHeader smallHead = RequestHeader.newBuilder().setCallId(1).build();
189 when(smallCallTask.getCall()).thenReturn(smallCall);
190 when(smallCall.getHeader()).thenReturn(smallHead);
191
192 CallRunner largeCallTask = mock(CallRunner.class);
193 RpcServer.Call largeCall = mock(RpcServer.Call.class);
194 RequestHeader largeHead = RequestHeader.newBuilder().setCallId(50).build();
195 when(largeCallTask.getCall()).thenReturn(largeCall);
196 when(largeCall.getHeader()).thenReturn(largeHead);
197
198 CallRunner hugeCallTask = mock(CallRunner.class);
199 RpcServer.Call hugeCall = mock(RpcServer.Call.class);
200 RequestHeader hugeHead = RequestHeader.newBuilder().setCallId(100).build();
201 when(hugeCallTask.getCall()).thenReturn(hugeCall);
202 when(hugeCall.getHeader()).thenReturn(hugeHead);
203
204 when(priority.getDeadline(eq(smallHead), any(Message.class))).thenReturn(0L);
205 when(priority.getDeadline(eq(largeHead), any(Message.class))).thenReturn(50L);
206 when(priority.getDeadline(eq(hugeHead), any(Message.class))).thenReturn(100L);
207
208 final ArrayList<Integer> work = new ArrayList<Integer>();
209 doAnswerTaskExecution(smallCallTask, work, 10, 250);
210 doAnswerTaskExecution(largeCallTask, work, 50, 250);
211 doAnswerTaskExecution(hugeCallTask, work, 100, 250);
212
213 scheduler.dispatch(smallCallTask);
214 scheduler.dispatch(smallCallTask);
215 scheduler.dispatch(smallCallTask);
216 scheduler.dispatch(hugeCallTask);
217 scheduler.dispatch(smallCallTask);
218 scheduler.dispatch(largeCallTask);
219 scheduler.dispatch(smallCallTask);
220 scheduler.dispatch(smallCallTask);
221
222 while (work.size() < 8) {
223 Thread.sleep(100);
224 }
225
226 int seqSum = 0;
227 int totalTime = 0;
228 for (int i = 0; i < work.size(); ++i) {
229 LOG.debug("Request i=" + i + " value=" + work.get(i));
230 seqSum += work.get(i);
231 totalTime += seqSum;
232 }
233 LOG.debug("Total Time: " + totalTime);
234
235
236
237
238 if (queueType.equals(SimpleRpcScheduler.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE)) {
239 assertEquals(530, totalTime);
240 } else if (queueType.equals(SimpleRpcScheduler.CALL_QUEUE_TYPE_FIFO_CONF_VALUE)) {
241 assertEquals(930, totalTime);
242 }
243 } finally {
244 scheduler.stop();
245 }
246 }
247
248 @Test
249 public void testScanQueueWithZeroScanRatio() throws Exception {
250 Configuration schedConf = HBaseConfiguration.create();
251 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
252 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.5f);
253 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0f);
254
255 PriorityFunction priority = mock(PriorityFunction.class);
256 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
257 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
258
259 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 2, 1, 1, priority,
260 HConstants.QOS_THRESHOLD);
261 assertNotEquals(scheduler, null);
262 }
263
264 @Test
265 public void testScanQueues() throws Exception {
266 Configuration schedConf = HBaseConfiguration.create();
267 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
268 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.7f);
269 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5f);
270
271 PriorityFunction priority = mock(PriorityFunction.class);
272 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
273 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
274
275 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 3, 1, 1, priority,
276 HConstants.QOS_THRESHOLD);
277 try {
278 scheduler.start();
279
280 CallRunner putCallTask = mock(CallRunner.class);
281 RpcServer.Call putCall = mock(RpcServer.Call.class);
282 putCall.param = RequestConverter.buildMutateRequest(
283 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
284 RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
285 when(putCallTask.getCall()).thenReturn(putCall);
286 when(putCall.getHeader()).thenReturn(putHead);
287
288 CallRunner getCallTask = mock(CallRunner.class);
289 RpcServer.Call getCall = mock(RpcServer.Call.class);
290 RequestHeader getHead = RequestHeader.newBuilder().setMethodName("get").build();
291 when(getCallTask.getCall()).thenReturn(getCall);
292 when(getCall.getHeader()).thenReturn(getHead);
293
294 CallRunner scanCallTask = mock(CallRunner.class);
295 RpcServer.Call scanCall = mock(RpcServer.Call.class);
296 scanCall.param = ScanRequest.newBuilder().setScannerId(1).build();
297 RequestHeader scanHead = RequestHeader.newBuilder().setMethodName("scan").build();
298 when(scanCallTask.getCall()).thenReturn(scanCall);
299 when(scanCall.getHeader()).thenReturn(scanHead);
300
301 ArrayList<Integer> work = new ArrayList<Integer>();
302 doAnswerTaskExecution(putCallTask, work, 1, 1000);
303 doAnswerTaskExecution(getCallTask, work, 2, 1000);
304 doAnswerTaskExecution(scanCallTask, work, 3, 1000);
305
306
307
308 scheduler.dispatch(putCallTask);
309 scheduler.dispatch(putCallTask);
310 scheduler.dispatch(putCallTask);
311 scheduler.dispatch(getCallTask);
312 scheduler.dispatch(getCallTask);
313 scheduler.dispatch(getCallTask);
314 scheduler.dispatch(scanCallTask);
315 scheduler.dispatch(scanCallTask);
316 scheduler.dispatch(scanCallTask);
317
318 while (work.size() < 6) {
319 Thread.sleep(100);
320 }
321
322 for (int i = 0; i < work.size() - 2; i += 3) {
323 assertNotEquals(work.get(i + 0), work.get(i + 1));
324 assertNotEquals(work.get(i + 0), work.get(i + 2));
325 assertNotEquals(work.get(i + 1), work.get(i + 2));
326 }
327 } finally {
328 scheduler.stop();
329 }
330 }
331
332 private void doAnswerTaskExecution(final CallRunner callTask,
333 final ArrayList<Integer> results, final int value, final int sleepInterval) {
334 callTask.setStatus(new MonitoredRPCHandlerImpl());
335 doAnswer(new Answer<Object>() {
336 @Override
337 public Object answer(InvocationOnMock invocation) {
338 synchronized (results) {
339 results.add(value);
340 }
341 Threads.sleepWithoutInterrupt(sleepInterval);
342 return null;
343 }
344 }).when(callTask).run();
345 }
346
347 private static void waitUntilQueueEmpty(SimpleRpcScheduler scheduler)
348 throws InterruptedException {
349 while (scheduler.getGeneralQueueLength() > 0) {
350 Thread.sleep(100);
351 }
352 }
353
354 @Test
355 public void testSoftAndHardQueueLimits() throws Exception {
356 Configuration schedConf = HBaseConfiguration.create();
357
358 schedConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 0);
359 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 5);
360
361 PriorityFunction priority = mock(PriorityFunction.class);
362 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
363 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
364 SimpleRpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 0, 0, 0, priority,
365 HConstants.QOS_THRESHOLD);
366 try {
367 scheduler.start();
368
369 CallRunner putCallTask = mock(CallRunner.class);
370 RpcServer.Call putCall = mock(RpcServer.Call.class);
371 putCall.param = RequestConverter.buildMutateRequest(
372 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
373 RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
374 when(putCallTask.getCall()).thenReturn(putCall);
375 when(putCall.getHeader()).thenReturn(putHead);
376
377 assertTrue(scheduler.dispatch(putCallTask));
378
379 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 0);
380 scheduler.onConfigurationChange(schedConf);
381 assertFalse(scheduler.dispatch(putCallTask));
382 waitUntilQueueEmpty(scheduler);
383 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 1);
384 scheduler.onConfigurationChange(schedConf);
385 assertTrue(scheduler.dispatch(putCallTask));
386 } finally {
387 scheduler.stop();
388 }
389 }
390
391 private static final class CoDelEnvironmentEdge implements EnvironmentEdge {
392
393 private final BlockingQueue<Long> timeQ = new LinkedBlockingQueue<>();
394
395 private long offset;
396
397 private final Set<String> threadNamePrefixs = new HashSet<>();
398
399 @Override
400 public long currentTime() {
401 for (String threadNamePrefix : threadNamePrefixs) {
402 String threadName = Thread.currentThread().getName();
403 if (threadName.startsWith(threadNamePrefix)) {
404 return timeQ.poll().longValue() + offset;
405 }
406 }
407 return System.currentTimeMillis();
408 }
409 }
410
411 @Test
412 public void testCoDelScheduling() throws Exception {
413 CoDelEnvironmentEdge envEdge = new CoDelEnvironmentEdge();
414 envEdge.threadNamePrefixs.add("RpcServer.CodelFPBQ.default.handler");
415 envEdge.threadNamePrefixs.add("RpcServer.CodelRWQ.default.handler");
416 Configuration schedConf = HBaseConfiguration.create();
417 schedConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 250);
418
419 schedConf.set(SimpleRpcScheduler.CALL_QUEUE_TYPE_CONF_KEY,
420 SimpleRpcScheduler.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
421
422 PriorityFunction priority = mock(PriorityFunction.class);
423 when(priority.getPriority(any(RPCProtos.RequestHeader.class), any(Message.class),
424 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
425 SimpleRpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 1, 1, 1, priority,
426 HConstants.QOS_THRESHOLD);
427 try {
428 scheduler.start();
429 EnvironmentEdgeManager.injectEdge(envEdge);
430 envEdge.offset = 5;
431
432 for (int i = 0; i < 100; i++) {
433 long time = System.currentTimeMillis();
434 envEdge.timeQ.put(time);
435 CallRunner cr = getMockedCallRunner(time, 2);
436 scheduler.dispatch(cr);
437 }
438
439 waitUntilQueueEmpty(scheduler);
440 Thread.sleep(100);
441 assertEquals("None of these calls should have been discarded", 0,
442 scheduler.getNumGeneralCallsDropped());
443
444 envEdge.offset = 151;
445
446 for (int i = 0; i < 20; i++) {
447 long time = System.currentTimeMillis();
448 envEdge.timeQ.put(time);
449 CallRunner cr = getMockedCallRunner(time, 2);
450 scheduler.dispatch(cr);
451 }
452
453
454 waitUntilQueueEmpty(scheduler);
455 Thread.sleep(100);
456 assertEquals("None of these calls should have been discarded", 0,
457 scheduler.getNumGeneralCallsDropped());
458
459 envEdge.offset = 2000;
460
461 for (int i = 0; i < 60; i++) {
462 long time = System.currentTimeMillis();
463 envEdge.timeQ.put(time);
464 CallRunner cr = getMockedCallRunner(time, 100);
465 scheduler.dispatch(cr);
466 }
467
468
469 waitUntilQueueEmpty(scheduler);
470 Thread.sleep(100);
471 assertTrue(
472 "There should have been at least 12 calls dropped however there were "
473 + scheduler.getNumGeneralCallsDropped(),
474 scheduler.getNumGeneralCallsDropped() > 12);
475 } finally {
476 scheduler.stop();
477 }
478 }
479
480
481
482 private CallRunner getMockedCallRunner(long timestamp, final long sleepTime) throws IOException {
483 final RpcServer.Call putCall = mock(RpcServer.Call.class);
484
485 putCall.timestamp = timestamp;
486 putCall.param = RequestConverter.buildMutateRequest(
487 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
488
489 RPCProtos.RequestHeader putHead = RPCProtos.RequestHeader.newBuilder()
490 .setMethodName("mutate")
491 .build();
492 when(putCall.getSize()).thenReturn(9L);
493 when(putCall.getHeader()).thenReturn(putHead);
494
495 CallRunner cr = new CallRunner(null, putCall) {
496 public void run() {
497 try {
498 LOG.warn("Sleeping for " + sleepTime);
499 Thread.sleep(sleepTime);
500 LOG.warn("Done Sleeping for " + sleepTime);
501 } catch (InterruptedException e) {
502 }
503 }
504 public Call getCall() {
505 return putCall;
506 }
507
508 public void drop() {}
509 };
510
511 return cr;
512 }
513 }