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
44 import org.apache.commons.logging.Log;
45 import org.apache.commons.logging.LogFactory;
46 import org.apache.hadoop.conf.Configuration;
47 import org.apache.hadoop.hbase.HBaseConfiguration;
48 import org.apache.hadoop.hbase.HConstants;
49 import org.apache.hadoop.hbase.client.Put;
50 import org.apache.hadoop.hbase.ipc.RpcServer.Call;
51 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl;
52 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
53 import org.apache.hadoop.hbase.protobuf.RequestConverter;
54 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos;
55 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
56 import org.apache.hadoop.hbase.security.User;
57 import org.apache.hadoop.hbase.testclassification.SmallTests;
58 import org.apache.hadoop.hbase.util.Bytes;
59 import org.apache.hadoop.hbase.util.EnvironmentEdge;
60 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
61 import org.apache.hadoop.hbase.util.Threads;
62 import org.junit.Before;
63 import org.junit.Test;
64 import org.junit.experimental.categories.Category;
65 import org.mockito.invocation.InvocationOnMock;
66 import org.mockito.stubbing.Answer;
67
68 import com.google.common.collect.ImmutableList;
69 import com.google.common.collect.ImmutableMap;
70 import com.google.common.collect.ImmutableSet;
71 import com.google.common.collect.Maps;
72 import com.google.protobuf.Message;
73
74 @Category(SmallTests.class)
75 public class TestSimpleRpcScheduler {
76 private static final Log LOG = LogFactory.getLog(TestSimpleRpcScheduler.class);
77
78 private final RpcScheduler.Context CONTEXT = new RpcScheduler.Context() {
79 @Override
80 public InetSocketAddress getListenerAddress() {
81 return InetSocketAddress.createUnresolved("127.0.0.1", 1000);
82 }
83 };
84 private Configuration conf;
85
86 @Before
87 public void setUp() {
88 conf = HBaseConfiguration.create();
89 }
90
91 @Test
92 public void testBasic() throws IOException, InterruptedException {
93 PriorityFunction qosFunction = mock(PriorityFunction.class);
94 RpcScheduler scheduler = new SimpleRpcScheduler(
95 conf, 10, 0, 0, qosFunction, 0);
96 scheduler.init(CONTEXT);
97 scheduler.start();
98 CallRunner task = createMockTask();
99 task.setStatus(new MonitoredRPCHandlerImpl());
100 scheduler.dispatch(task);
101 verify(task, timeout(1000)).run();
102 scheduler.stop();
103 }
104
105 @Test
106 public void testHandlerIsolation() throws IOException, InterruptedException {
107 CallRunner generalTask = createMockTask();
108 CallRunner priorityTask = createMockTask();
109 CallRunner replicationTask = createMockTask();
110 List<CallRunner> tasks = ImmutableList.of(
111 generalTask,
112 priorityTask,
113 replicationTask);
114 Map<CallRunner, Integer> qos = ImmutableMap.of(
115 generalTask, 0,
116 priorityTask, HConstants.HIGH_QOS + 1,
117 replicationTask, HConstants.REPLICATION_QOS);
118 PriorityFunction qosFunction = mock(PriorityFunction.class);
119 final Map<CallRunner, Thread> handlerThreads = Maps.newHashMap();
120 final CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
121 Answer<Void> answerToRun = new Answer<Void>() {
122 @Override
123 public Void answer(InvocationOnMock invocationOnMock) throws Throwable {
124 synchronized (handlerThreads) {
125 handlerThreads.put(
126 (CallRunner) invocationOnMock.getMock(),
127 Thread.currentThread());
128 }
129 countDownLatch.countDown();
130 return null;
131 }
132 };
133 for (CallRunner task : tasks) {
134 task.setStatus(new MonitoredRPCHandlerImpl());
135 doAnswer(answerToRun).when(task).run();
136 }
137
138 RpcScheduler scheduler = new SimpleRpcScheduler(
139 conf, 1, 1 ,1, qosFunction, HConstants.HIGH_QOS);
140 scheduler.init(CONTEXT);
141 scheduler.start();
142 for (CallRunner task : tasks) {
143 when(qosFunction.getPriority((RPCProtos.RequestHeader) anyObject(),
144 (Message) anyObject(), (User) anyObject()))
145 .thenReturn(qos.get(task));
146 scheduler.dispatch(task);
147 }
148 for (CallRunner task : tasks) {
149 verify(task, timeout(1000)).run();
150 }
151 scheduler.stop();
152
153
154 countDownLatch.await();
155 assertEquals(3, ImmutableSet.copyOf(handlerThreads.values()).size());
156 }
157
158 private CallRunner createMockTask() {
159 Call call = mock(Call.class);
160 CallRunner task = mock(CallRunner.class);
161 when(task.getCall()).thenReturn(call);
162 return task;
163 }
164
165 @Test
166 public void testRpcScheduler() throws Exception {
167 testRpcScheduler(SimpleRpcScheduler.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
168 testRpcScheduler(SimpleRpcScheduler.CALL_QUEUE_TYPE_FIFO_CONF_VALUE);
169 }
170
171 private void testRpcScheduler(final String queueType) throws Exception {
172 Configuration schedConf = HBaseConfiguration.create();
173 schedConf.set(SimpleRpcScheduler.CALL_QUEUE_TYPE_CONF_KEY, queueType);
174
175 PriorityFunction priority = mock(PriorityFunction.class);
176 when(priority.getPriority(any(RequestHeader.class),
177 any(Message.class), any(User.class)))
178 .thenReturn(HConstants.NORMAL_QOS);
179
180 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 1, 1, 1, priority,
181 HConstants.QOS_THRESHOLD);
182 try {
183 scheduler.start();
184
185 CallRunner smallCallTask = mock(CallRunner.class);
186 RpcServer.Call smallCall = mock(RpcServer.Call.class);
187 RequestHeader smallHead = RequestHeader.newBuilder().setCallId(1).build();
188 when(smallCallTask.getCall()).thenReturn(smallCall);
189 when(smallCall.getHeader()).thenReturn(smallHead);
190
191 CallRunner largeCallTask = mock(CallRunner.class);
192 RpcServer.Call largeCall = mock(RpcServer.Call.class);
193 RequestHeader largeHead = RequestHeader.newBuilder().setCallId(50).build();
194 when(largeCallTask.getCall()).thenReturn(largeCall);
195 when(largeCall.getHeader()).thenReturn(largeHead);
196
197 CallRunner hugeCallTask = mock(CallRunner.class);
198 RpcServer.Call hugeCall = mock(RpcServer.Call.class);
199 RequestHeader hugeHead = RequestHeader.newBuilder().setCallId(100).build();
200 when(hugeCallTask.getCall()).thenReturn(hugeCall);
201 when(hugeCall.getHeader()).thenReturn(hugeHead);
202
203 when(priority.getDeadline(eq(smallHead), any(Message.class))).thenReturn(0L);
204 when(priority.getDeadline(eq(largeHead), any(Message.class))).thenReturn(50L);
205 when(priority.getDeadline(eq(hugeHead), any(Message.class))).thenReturn(100L);
206
207 final ArrayList<Integer> work = new ArrayList<Integer>();
208 doAnswerTaskExecution(smallCallTask, work, 10, 250);
209 doAnswerTaskExecution(largeCallTask, work, 50, 250);
210 doAnswerTaskExecution(hugeCallTask, work, 100, 250);
211
212 scheduler.dispatch(smallCallTask);
213 scheduler.dispatch(smallCallTask);
214 scheduler.dispatch(smallCallTask);
215 scheduler.dispatch(hugeCallTask);
216 scheduler.dispatch(smallCallTask);
217 scheduler.dispatch(largeCallTask);
218 scheduler.dispatch(smallCallTask);
219 scheduler.dispatch(smallCallTask);
220
221 while (work.size() < 8) {
222 Thread.sleep(100);
223 }
224
225 int seqSum = 0;
226 int totalTime = 0;
227 for (int i = 0; i < work.size(); ++i) {
228 LOG.debug("Request i=" + i + " value=" + work.get(i));
229 seqSum += work.get(i);
230 totalTime += seqSum;
231 }
232 LOG.debug("Total Time: " + totalTime);
233
234
235
236
237 if (queueType.equals(SimpleRpcScheduler.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE)) {
238 assertEquals(530, totalTime);
239 } else if (queueType.equals(SimpleRpcScheduler.CALL_QUEUE_TYPE_FIFO_CONF_VALUE)) {
240 assertEquals(930, totalTime);
241 }
242 } finally {
243 scheduler.stop();
244 }
245 }
246
247 @Test
248 public void testScanQueueWithZeroScanRatio() throws Exception {
249 Configuration schedConf = HBaseConfiguration.create();
250 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
251 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.5f);
252 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0f);
253
254 PriorityFunction priority = mock(PriorityFunction.class);
255 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
256 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
257
258 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 2, 1, 1, priority,
259 HConstants.QOS_THRESHOLD);
260 assertNotEquals(scheduler, null);
261 }
262
263 @Test
264 public void testScanQueues() throws Exception {
265 Configuration schedConf = HBaseConfiguration.create();
266 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
267 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.7f);
268 schedConf.setFloat(SimpleRpcScheduler.CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0.5f);
269
270 PriorityFunction priority = mock(PriorityFunction.class);
271 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
272 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
273
274 RpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 3, 1, 1, priority,
275 HConstants.QOS_THRESHOLD);
276 try {
277 scheduler.start();
278
279 CallRunner putCallTask = mock(CallRunner.class);
280 RpcServer.Call putCall = mock(RpcServer.Call.class);
281 putCall.param = RequestConverter.buildMutateRequest(
282 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
283 RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
284 when(putCallTask.getCall()).thenReturn(putCall);
285 when(putCall.getHeader()).thenReturn(putHead);
286
287 CallRunner getCallTask = mock(CallRunner.class);
288 RpcServer.Call getCall = mock(RpcServer.Call.class);
289 RequestHeader getHead = RequestHeader.newBuilder().setMethodName("get").build();
290 when(getCallTask.getCall()).thenReturn(getCall);
291 when(getCall.getHeader()).thenReturn(getHead);
292
293 CallRunner scanCallTask = mock(CallRunner.class);
294 RpcServer.Call scanCall = mock(RpcServer.Call.class);
295 scanCall.param = ScanRequest.newBuilder().setScannerId(1).build();
296 RequestHeader scanHead = RequestHeader.newBuilder().setMethodName("scan").build();
297 when(scanCallTask.getCall()).thenReturn(scanCall);
298 when(scanCall.getHeader()).thenReturn(scanHead);
299
300 ArrayList<Integer> work = new ArrayList<Integer>();
301 doAnswerTaskExecution(putCallTask, work, 1, 1000);
302 doAnswerTaskExecution(getCallTask, work, 2, 1000);
303 doAnswerTaskExecution(scanCallTask, work, 3, 1000);
304
305
306
307 scheduler.dispatch(putCallTask);
308 scheduler.dispatch(putCallTask);
309 scheduler.dispatch(putCallTask);
310 scheduler.dispatch(getCallTask);
311 scheduler.dispatch(getCallTask);
312 scheduler.dispatch(getCallTask);
313 scheduler.dispatch(scanCallTask);
314 scheduler.dispatch(scanCallTask);
315 scheduler.dispatch(scanCallTask);
316
317 while (work.size() < 6) {
318 Thread.sleep(100);
319 }
320
321 for (int i = 0; i < work.size() - 2; i += 3) {
322 assertNotEquals(work.get(i + 0), work.get(i + 1));
323 assertNotEquals(work.get(i + 0), work.get(i + 2));
324 assertNotEquals(work.get(i + 1), work.get(i + 2));
325 }
326 } finally {
327 scheduler.stop();
328 }
329 }
330
331 private void doAnswerTaskExecution(final CallRunner callTask,
332 final ArrayList<Integer> results, final int value, final int sleepInterval) {
333 callTask.setStatus(new MonitoredRPCHandlerImpl());
334 doAnswer(new Answer<Object>() {
335 @Override
336 public Object answer(InvocationOnMock invocation) {
337 synchronized (results) {
338 results.add(value);
339 }
340 Threads.sleepWithoutInterrupt(sleepInterval);
341 return null;
342 }
343 }).when(callTask).run();
344 }
345
346 private static void waitUntilQueueEmpty(SimpleRpcScheduler scheduler)
347 throws InterruptedException {
348 while (scheduler.getGeneralQueueLength() > 0) {
349 Thread.sleep(100);
350 }
351 }
352
353 @Test
354 public void testSoftAndHardQueueLimits() throws Exception {
355 Configuration schedConf = HBaseConfiguration.create();
356
357 schedConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 0);
358 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 5);
359
360 PriorityFunction priority = mock(PriorityFunction.class);
361 when(priority.getPriority(any(RequestHeader.class), any(Message.class),
362 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
363 SimpleRpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 0, 0, 0, priority,
364 HConstants.QOS_THRESHOLD);
365 try {
366 scheduler.start();
367
368 CallRunner putCallTask = mock(CallRunner.class);
369 RpcServer.Call putCall = mock(RpcServer.Call.class);
370 putCall.param = RequestConverter.buildMutateRequest(
371 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
372 RequestHeader putHead = RequestHeader.newBuilder().setMethodName("mutate").build();
373 when(putCallTask.getCall()).thenReturn(putCall);
374 when(putCall.getHeader()).thenReturn(putHead);
375
376 assertTrue(scheduler.dispatch(putCallTask));
377 waitUntilQueueEmpty(scheduler);
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 }