1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.ipc;
20
21 import java.util.ArrayList;
22 import java.util.Comparator;
23 import java.util.List;
24 import java.util.Locale;
25 import java.util.concurrent.BlockingQueue;
26 import java.util.concurrent.LinkedBlockingQueue;
27 import java.util.concurrent.ThreadLocalRandom;
28 import java.util.concurrent.atomic.AtomicInteger;
29 import java.util.concurrent.atomic.AtomicLong;
30
31 import org.apache.commons.logging.Log;
32 import org.apache.commons.logging.LogFactory;
33 import org.apache.hadoop.conf.Configuration;
34 import org.apache.hadoop.hbase.Abortable;
35 import org.apache.hadoop.hbase.HConstants;
36 import org.apache.hadoop.hbase.classification.InterfaceAudience;
37 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler;
38 import org.apache.hadoop.hbase.util.BoundedPriorityBlockingQueue;
39 import org.apache.hadoop.hbase.util.ReflectionUtils;
40 import org.apache.hadoop.util.StringUtils;
41
42 import com.google.common.base.Preconditions;
43 import com.google.common.base.Strings;
44
45
46
47
48
49 @InterfaceAudience.Private
50 public abstract class RpcExecutor {
51 private static final Log LOG = LogFactory.getLog(RpcExecutor.class);
52
53 protected static final int DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT = 250;
54 public static final String CALL_QUEUE_HANDLER_FACTOR_CONF_KEY = "hbase.ipc.server.callqueue.handler.factor";
55
56
57 public static final String QUEUE_MAX_CALL_DELAY_CONF_KEY = "hbase.ipc.server.queue.max.call.delay";
58
59
60
61
62
63
64 public static final String CALL_QUEUE_TYPE_CODEL_CONF_VALUE = "codel";
65 public static final String CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE = "deadline";
66 public static final String CALL_QUEUE_TYPE_FIFO_CONF_VALUE = "fifo";
67 public static final String CALL_QUEUE_TYPE_CONF_KEY = "hbase.ipc.server.callqueue.type";
68 public static final String CALL_QUEUE_TYPE_CONF_DEFAULT = CALL_QUEUE_TYPE_FIFO_CONF_VALUE;
69
70
71 public static final String CALL_QUEUE_CODEL_TARGET_DELAY = "hbase.ipc.server.callqueue.codel.target.delay";
72 public static final String CALL_QUEUE_CODEL_INTERVAL = "hbase.ipc.server.callqueue.codel.interval";
73 public static final String CALL_QUEUE_CODEL_LIFO_THRESHOLD = "hbase.ipc.server.callqueue.codel.lifo.threshold";
74
75 public static final int CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY = 100;
76 public static final int CALL_QUEUE_CODEL_DEFAULT_INTERVAL = 100;
77 public static final double CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD = 0.8;
78
79 private AtomicLong numGeneralCallsDropped = new AtomicLong();
80 private AtomicLong numLifoModeSwitches = new AtomicLong();
81
82 protected final int numCallQueues;
83 protected final List<BlockingQueue<CallRunner>> queues;
84 private final Class<? extends BlockingQueue> queueClass;
85 private final Object[] queueInitArgs;
86
87 private final PriorityFunction priority;
88
89 protected volatile int currentQueueLimit;
90
91 private final AtomicInteger activeHandlerCount = new AtomicInteger(0);
92 private final List<Handler> handlers;
93 private final int handlerCount;
94 private final AtomicInteger failedHandlerCount = new AtomicInteger(0);
95
96 private String name;
97 private boolean running;
98
99 private Configuration conf = null;
100 private Abortable abortable = null;
101
102 @Deprecated
103 public RpcExecutor(final String name, final int handlerCount, final int numCallQueues) {
104 this.name = Strings.nullToEmpty(name);
105 this.handlers = new ArrayList<Handler>(handlerCount);
106 this.handlerCount = handlerCount;
107 this.numCallQueues = numCallQueues;
108 this.queues = new ArrayList<>(this.numCallQueues);
109 this.queueClass = null;
110 this.queueInitArgs = new Object[0];
111 this.priority = null;
112 }
113
114 @Deprecated
115 public RpcExecutor(final String name, final int handlerCount, final int numCallQueues,
116 final Configuration conf, final Abortable abortable) {
117 this(name, handlerCount, numCallQueues);
118 this.conf = conf;
119 this.abortable = abortable;
120 }
121
122 public RpcExecutor(final String name, final int handlerCount, final int maxQueueLength,
123 final PriorityFunction priority, final Configuration conf, final Abortable abortable) {
124 this(name, handlerCount, conf.get(CALL_QUEUE_TYPE_CONF_KEY,
125 CALL_QUEUE_TYPE_CONF_DEFAULT), maxQueueLength, priority, conf, abortable);
126 }
127
128 public RpcExecutor(final String name, final int handlerCount, final String callQueueType,
129 final int maxQueueLength, final PriorityFunction priority, final Configuration conf,
130 final Abortable abortable) {
131 this.name = Strings.nullToEmpty(name);
132 this.conf = conf;
133 this.abortable = abortable;
134
135 float callQueuesHandlersFactor = this.conf.getFloat(CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 0);
136 this.numCallQueues = computeNumCallQueues(handlerCount, callQueuesHandlersFactor);
137 this.queues = new ArrayList<>(this.numCallQueues);
138
139 this.handlerCount = Math.max(handlerCount, this.numCallQueues);
140 this.handlers = new ArrayList<>(this.handlerCount);
141
142 this.priority = priority;
143
144 if (isDeadlineQueueType(callQueueType)) {
145 this.name += ".Deadline";
146 this.queueInitArgs = new Object[] { maxQueueLength,
147 new CallPriorityComparator(conf, this.priority) };
148 this.queueClass = BoundedPriorityBlockingQueue.class;
149 } else if (isCodelQueueType(callQueueType)) {
150 this.name += ".Codel";
151 int codelTargetDelay = conf.getInt(CALL_QUEUE_CODEL_TARGET_DELAY,
152 CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY);
153 int codelInterval = conf.getInt(CALL_QUEUE_CODEL_INTERVAL, CALL_QUEUE_CODEL_DEFAULT_INTERVAL);
154 double codelLifoThreshold = conf.getDouble(CALL_QUEUE_CODEL_LIFO_THRESHOLD,
155 CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD);
156 queueInitArgs = new Object[] { maxQueueLength, codelTargetDelay, codelInterval,
157 codelLifoThreshold, numGeneralCallsDropped, numLifoModeSwitches };
158 queueClass = AdaptiveLifoCoDelCallQueue.class;
159 } else {
160 this.name += ".Fifo";
161 queueInitArgs = new Object[] { maxQueueLength };
162 queueClass = LinkedBlockingQueue.class;
163 }
164
165 LOG.info("RpcExecutor " + " name " + " using " + callQueueType
166 + " as call queue; numCallQueues=" + numCallQueues + "; maxQueueLength=" + maxQueueLength
167 + "; handlerCount=" + handlerCount);
168 }
169
170 protected int computeNumCallQueues(final int handlerCount, final float callQueuesHandlersFactor) {
171 return Math.max(1, (int) Math.round(handlerCount * callQueuesHandlersFactor));
172 }
173
174 protected void initializeQueues(final int numQueues) {
175 if (queueInitArgs.length > 0) {
176 currentQueueLimit = (int) queueInitArgs[0];
177 queueInitArgs[0] = Math.max((int) queueInitArgs[0], DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT);
178 }
179 for (int i = 0; i < numQueues; ++i) {
180 queues
181 .add((BlockingQueue<CallRunner>) ReflectionUtils.newInstance(queueClass, queueInitArgs));
182 }
183 }
184
185 public void start(final int port) {
186 running = true;
187 startHandlers(port);
188 }
189
190 public void stop() {
191 running = false;
192 for (Thread handler : handlers) {
193 handler.interrupt();
194 }
195 }
196
197
198 public abstract boolean dispatch(final CallRunner callTask) throws InterruptedException;
199
200
201 public List<BlockingQueue<CallRunner>> getQueues() {
202 return queues;
203 }
204
205 protected void startHandlers(final int port) {
206 List<BlockingQueue<CallRunner>> callQueues = getQueues();
207 startHandlers(null, handlerCount, callQueues, 0, callQueues.size(), port, activeHandlerCount);
208 }
209
210
211
212
213 protected Handler getHandler(final String name, final double handlerFailureThreshhold,
214 final BlockingQueue<CallRunner> q, final AtomicInteger activeHandlerCount) {
215 return new Handler(name, handlerFailureThreshhold, q, activeHandlerCount);
216 }
217
218
219
220
221 protected void startHandlers(final String nameSuffix, final int numHandlers,
222 final List<BlockingQueue<CallRunner>> callQueues, final int qindex, final int qsize,
223 final int port, final AtomicInteger activeHandlerCount) {
224 final String threadPrefix = name + Strings.nullToEmpty(nameSuffix);
225 double handlerFailureThreshhold = conf == null ? 1.0 : conf.getDouble(
226 HConstants.REGION_SERVER_HANDLER_ABORT_ON_ERROR_PERCENT,
227 HConstants.DEFAULT_REGION_SERVER_HANDLER_ABORT_ON_ERROR_PERCENT);
228 for (int i = 0; i < numHandlers; i++) {
229 final int index = qindex + (i % qsize);
230 String name = "RpcServer." + threadPrefix + ".handler=" + handlers.size() + ",queue=" + index
231 + ",port=" + port;
232 Handler handler = getHandler(name, handlerFailureThreshhold, callQueues.get(index),
233 activeHandlerCount);
234 handler.start();
235 LOG.debug("Started " + name);
236 handlers.add(handler);
237 }
238 }
239
240
241
242
243 protected class Handler extends Thread {
244
245
246
247 final BlockingQueue<CallRunner> q;
248
249 final double handlerFailureThreshhold;
250
251
252 final AtomicInteger activeHandlerCount;
253
254 Handler(final String name, final double handlerFailureThreshhold,
255 final BlockingQueue<CallRunner> q, final AtomicInteger activeHandlerCount) {
256 super(name);
257 setDaemon(true);
258 this.q = q;
259 this.handlerFailureThreshhold = handlerFailureThreshhold;
260 this.activeHandlerCount = activeHandlerCount;
261 }
262
263
264
265
266
267 protected CallRunner getCallRunner() throws InterruptedException {
268 return this.q.take();
269 }
270
271 @Override
272 public void run() {
273 boolean interrupted = false;
274 try {
275 while (running) {
276 try {
277 run(getCallRunner());
278 } catch (InterruptedException e) {
279 interrupted = true;
280 }
281 }
282 } catch (Exception e) {
283 LOG.warn(e);
284 throw e;
285 } finally {
286 if (interrupted) {
287 Thread.currentThread().interrupt();
288 }
289 }
290 }
291
292 private void run(CallRunner cr) {
293 MonitoredRPCHandler status = RpcServer.getStatus();
294 cr.setStatus(status);
295 try {
296 this.activeHandlerCount.incrementAndGet();
297 cr.run();
298 } catch (Throwable e) {
299 if (e instanceof Error) {
300 int failedCount = failedHandlerCount.incrementAndGet();
301 if (this.handlerFailureThreshhold >= 0
302 && failedCount > handlerCount * this.handlerFailureThreshhold) {
303 String message = "Number of failed RpcServer handler runs exceeded threshhold "
304 + this.handlerFailureThreshhold + "; reason: " + StringUtils.stringifyException(e);
305 if (abortable != null) {
306 abortable.abort(message, e);
307 } else {
308 LOG.error("Error but can't abort because abortable is null: "
309 + StringUtils.stringifyException(e));
310 throw e;
311 }
312 } else {
313 LOG.warn("Handler errors " + StringUtils.stringifyException(e));
314 }
315 } else {
316 LOG.warn("Handler exception " + StringUtils.stringifyException(e));
317 }
318 } finally {
319 this.activeHandlerCount.decrementAndGet();
320 }
321 }
322 }
323
324 public static abstract class QueueBalancer {
325
326
327
328 public abstract int getNextQueue();
329 }
330
331 public static QueueBalancer getBalancer(int queueSize) {
332 Preconditions.checkArgument(queueSize > 0, "Queue size is <= 0, must be at least 1");
333 if (queueSize == 1) {
334 return ONE_QUEUE;
335 } else {
336 return new RandomQueueBalancer(queueSize);
337 }
338 }
339
340
341
342
343 private static QueueBalancer ONE_QUEUE = new QueueBalancer() {
344 @Override
345 public int getNextQueue() {
346 return 0;
347 }
348 };
349
350
351
352
353 private static class RandomQueueBalancer extends QueueBalancer {
354 private final int queueSize;
355
356 public RandomQueueBalancer(int queueSize) {
357 this.queueSize = queueSize;
358 }
359
360 public int getNextQueue() {
361 return ThreadLocalRandom.current().nextInt(queueSize);
362 }
363 }
364
365
366
367
368
369
370
371 private static class CallPriorityComparator implements Comparator<CallRunner> {
372 private final static int DEFAULT_MAX_CALL_DELAY = 5000;
373
374 private final PriorityFunction priority;
375 private final int maxDelay;
376
377 public CallPriorityComparator(final Configuration conf, final PriorityFunction priority) {
378 this.priority = priority;
379 this.maxDelay = conf.getInt(QUEUE_MAX_CALL_DELAY_CONF_KEY, DEFAULT_MAX_CALL_DELAY);
380 }
381
382 @Override
383 public int compare(CallRunner a, CallRunner b) {
384 RpcServer.Call callA = a.getCall();
385 RpcServer.Call callB = b.getCall();
386 long deadlineA = priority.getDeadline(callA.getHeader(), callA.param);
387 long deadlineB = priority.getDeadline(callB.getHeader(), callB.param);
388 deadlineA = callA.timestamp + Math.min(deadlineA, maxDelay);
389 deadlineB = callB.timestamp + Math.min(deadlineB, maxDelay);
390 return Long.compare(deadlineA, deadlineB);
391 }
392 }
393
394 public static boolean isDeadlineQueueType(final String callQueueType) {
395 return callQueueType.equals(CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
396 }
397
398 public static boolean isCodelQueueType(final String callQueueType) {
399 return callQueueType.equals(CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
400 }
401
402 public static boolean isFifoQueueType(final String callQueueType) {
403 return callQueueType.equals(CALL_QUEUE_TYPE_FIFO_CONF_VALUE);
404 }
405
406 public long getNumGeneralCallsDropped() {
407 return numGeneralCallsDropped.get();
408 }
409
410 public long getNumLifoModeSwitches() {
411 return numLifoModeSwitches.get();
412 }
413
414 public int getActiveHandlerCount() {
415 return activeHandlerCount.get();
416 }
417
418 public int getActiveWriteHandlerCount() {
419 return 0;
420 }
421
422 public int getActiveReadHandlerCount() {
423 return 0;
424 }
425
426 public int getActiveScanHandlerCount() {
427 return 0;
428 }
429
430
431 public int getQueueLength() {
432 int length = 0;
433 for (final BlockingQueue<CallRunner> queue: queues) {
434 length += queue.size();
435 }
436 return length;
437 }
438
439 public int getReadQueueLength() {
440 return 0;
441 }
442
443 public int getScanQueueLength() {
444 return 0;
445 }
446
447 public int getWriteQueueLength() {
448 return 0;
449 }
450
451 public String getName() {
452 return this.name;
453 }
454
455
456
457
458
459 public void resizeQueues(Configuration conf) {
460 String configKey = RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH;
461 if (name != null && name.toLowerCase(Locale.ROOT).contains("priority")) {
462 configKey = RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH;
463 }
464 currentQueueLimit = conf.getInt(configKey, currentQueueLimit);
465 }
466
467 public void onConfigurationChange(Configuration conf) {
468
469 int codelTargetDelay = conf.getInt(CALL_QUEUE_CODEL_TARGET_DELAY,
470 CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY);
471 int codelInterval = conf.getInt(CALL_QUEUE_CODEL_INTERVAL, CALL_QUEUE_CODEL_DEFAULT_INTERVAL);
472 double codelLifoThreshold = conf.getDouble(CALL_QUEUE_CODEL_LIFO_THRESHOLD,
473 CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD);
474
475 for (BlockingQueue<CallRunner> queue : queues) {
476 if (queue instanceof AdaptiveLifoCoDelCallQueue) {
477 ((AdaptiveLifoCoDelCallQueue) queue).updateTunables(codelTargetDelay, codelInterval,
478 codelLifoThreshold);
479 }
480 }
481 }
482 }