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 import static org.junit.Assert.assertEquals;
20 import static org.junit.Assert.assertFalse;
21 import static org.junit.Assert.assertNotEquals;
22 import static org.junit.Assert.assertTrue;
23 import static org.mockito.Matchers.any;
24 import static org.mockito.Matchers.anyObject;
25 import static org.mockito.Matchers.eq;
26 import static org.mockito.Mockito.doAnswer;
27 import static org.mockito.Mockito.mock;
28 import static org.mockito.Mockito.timeout;
29 import static org.mockito.Mockito.verify;
30 import static org.mockito.Mockito.when;
31
32 import java.io.IOException;
33 import java.net.InetSocketAddress;
34 import java.util.ArrayList;
35 import java.util.HashSet;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Set;
39 import java.util.concurrent.BlockingQueue;
40 import java.util.concurrent.CountDownLatch;
41 import java.util.concurrent.LinkedBlockingQueue;
42 import java.util.concurrent.ThreadLocalRandom;
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.RequestConverter;
53 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
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(RpcExecutor.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
168 testRpcScheduler(RpcExecutor.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(RpcExecutor.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(RpcExecutor.CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE)) {
238 assertEquals(530, totalTime);
239 } else if (queueType.equals(RpcExecutor.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(RpcExecutor.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
251 schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.5f);
252 schedConf.setFloat(RWQueueRpcExecutor.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(RpcExecutor.CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 1.0f);
267 schedConf.setFloat(RWQueueRpcExecutor.CALL_QUEUE_READ_SHARE_CONF_KEY, 0.7f);
268 schedConf.setFloat(RWQueueRpcExecutor.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
378 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 0);
379 scheduler.onConfigurationChange(schedConf);
380 assertFalse(scheduler.dispatch(putCallTask));
381 waitUntilQueueEmpty(scheduler);
382 schedConf.setInt("hbase.ipc.server.max.callqueue.length", 1);
383 scheduler.onConfigurationChange(schedConf);
384 assertTrue(scheduler.dispatch(putCallTask));
385 } finally {
386 scheduler.stop();
387 }
388 }
389
390 private static final class CoDelEnvironmentEdge implements EnvironmentEdge {
391
392 private final BlockingQueue<Long> timeQ = new LinkedBlockingQueue<>();
393
394 private long offset;
395
396 private final Set<String> threadNamePrefixs = new HashSet<>();
397
398 @Override
399 public long currentTime() {
400 for (String threadNamePrefix : threadNamePrefixs) {
401 String threadName = Thread.currentThread().getName();
402 if (threadName.startsWith(threadNamePrefix)) {
403 return timeQ.poll().longValue() + offset;
404 }
405 }
406 return System.currentTimeMillis();
407 }
408 }
409
410 @Test
411 public void testCoDelScheduling() throws Exception {
412 CoDelEnvironmentEdge envEdge = new CoDelEnvironmentEdge();
413 envEdge.threadNamePrefixs.add("RpcServer.default.FPBQ.Codel.handler");
414 Configuration schedConf = HBaseConfiguration.create();
415 schedConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 250);
416
417 schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
418 RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
419
420 PriorityFunction priority = mock(PriorityFunction.class);
421 when(priority.getPriority(any(RPCProtos.RequestHeader.class), any(Message.class),
422 any(User.class))).thenReturn(HConstants.NORMAL_QOS);
423 SimpleRpcScheduler scheduler = new SimpleRpcScheduler(schedConf, 1, 1, 1, priority,
424 HConstants.QOS_THRESHOLD);
425 try {
426 scheduler.start();
427 EnvironmentEdgeManager.injectEdge(envEdge);
428 envEdge.offset = 5;
429
430 for (int i = 0; i < 100; i++) {
431 long time = System.currentTimeMillis();
432 envEdge.timeQ.put(time);
433 CallRunner cr = getMockedCallRunner(time, 2);
434 scheduler.dispatch(cr);
435 }
436
437 waitUntilQueueEmpty(scheduler);
438 Thread.sleep(100);
439 assertEquals("None of these calls should have been discarded", 0,
440 scheduler.getNumGeneralCallsDropped());
441
442 envEdge.offset = 151;
443
444 for (int i = 0; i < 20; i++) {
445 long time = System.currentTimeMillis();
446 envEdge.timeQ.put(time);
447 CallRunner cr = getMockedCallRunner(time, 2);
448 scheduler.dispatch(cr);
449 }
450
451
452 waitUntilQueueEmpty(scheduler);
453 Thread.sleep(100);
454 assertEquals("None of these calls should have been discarded", 0,
455 scheduler.getNumGeneralCallsDropped());
456
457 envEdge.offset = 2000;
458
459 for (int i = 0; i < 60; i++) {
460 long time = System.currentTimeMillis();
461 envEdge.timeQ.put(time);
462 CallRunner cr = getMockedCallRunner(time, 100);
463 scheduler.dispatch(cr);
464 }
465
466
467 waitUntilQueueEmpty(scheduler);
468 Thread.sleep(100);
469 assertTrue(
470 "There should have been at least 12 calls dropped however there were "
471 + scheduler.getNumGeneralCallsDropped(),
472 scheduler.getNumGeneralCallsDropped() > 12);
473 } finally {
474 scheduler.stop();
475 }
476 }
477
478
479
480 private CallRunner getMockedCallRunner(long timestamp, final long sleepTime) throws IOException {
481 final RpcServer.Call putCall = mock(RpcServer.Call.class);
482
483 putCall.timestamp = timestamp;
484 putCall.param = RequestConverter.buildMutateRequest(
485 Bytes.toBytes("abc"), new Put(Bytes.toBytes("row")));
486
487 RPCProtos.RequestHeader putHead = RPCProtos.RequestHeader.newBuilder()
488 .setMethodName("mutate")
489 .build();
490 when(putCall.getSize()).thenReturn(9L);
491 when(putCall.getHeader()).thenReturn(putHead);
492
493 CallRunner cr = new CallRunner(null, putCall) {
494 public void run() {
495 try {
496 LOG.warn("Sleeping for " + sleepTime);
497 Thread.sleep(sleepTime);
498 LOG.warn("Done Sleeping for " + sleepTime);
499 } catch (InterruptedException e) {
500 }
501 }
502 public Call getCall() {
503 return putCall;
504 }
505
506 public void drop() {}
507 };
508
509 return cr;
510 }
511 }