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
21 import java.util.Comparator;
22 import java.util.concurrent.BlockingQueue;
23 import java.util.concurrent.atomic.AtomicLong;
24
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.hbase.Abortable;
29 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
30 import org.apache.hadoop.hbase.HConstants;
31 import org.apache.hadoop.hbase.classification.InterfaceAudience;
32 import org.apache.hadoop.hbase.classification.InterfaceStability;
33 import org.apache.hadoop.hbase.conf.ConfigurationObserver;
34 import org.apache.hadoop.hbase.util.BoundedPriorityBlockingQueue;
35
36
37
38
39
40 @InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC, HBaseInterfaceAudience.PHOENIX})
41 @InterfaceStability.Evolving
42 public class SimpleRpcScheduler extends RpcScheduler implements ConfigurationObserver {
43 private static final Log LOG = LogFactory.getLog(SimpleRpcScheduler.class);
44
45 public static final String CALL_QUEUE_READ_SHARE_CONF_KEY =
46 "hbase.ipc.server.callqueue.read.ratio";
47 public static final String CALL_QUEUE_SCAN_SHARE_CONF_KEY =
48 "hbase.ipc.server.callqueue.scan.ratio";
49 public static final String CALL_QUEUE_HANDLER_FACTOR_CONF_KEY =
50 "hbase.ipc.server.callqueue.handler.factor";
51
52
53
54
55
56
57 public static final String CALL_QUEUE_TYPE_CODEL_CONF_VALUE = "codel";
58 public static final String CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE = "deadline";
59 public static final String CALL_QUEUE_TYPE_FIFO_CONF_VALUE = "fifo";
60 public static final String CALL_QUEUE_TYPE_CONF_KEY = "hbase.ipc.server.callqueue.type";
61 public static final String CALL_QUEUE_TYPE_CONF_DEFAULT = CALL_QUEUE_TYPE_FIFO_CONF_VALUE;
62
63
64 public static final String QUEUE_MAX_CALL_DELAY_CONF_KEY
65 = "hbase.ipc.server.queue.max.call.delay";
66
67
68 public static final String CALL_QUEUE_CODEL_TARGET_DELAY =
69 "hbase.ipc.server.callqueue.codel.target.delay";
70 public static final String CALL_QUEUE_CODEL_INTERVAL =
71 "hbase.ipc.server.callqueue.codel.interval";
72 public static final String CALL_QUEUE_CODEL_LIFO_THRESHOLD =
73 "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
83
84
85
86 @Override
87 public void onConfigurationChange(Configuration conf) {
88 callExecutor.resizeQueues(conf);
89 if (priorityExecutor != null) {
90 priorityExecutor.resizeQueues(conf);
91 }
92 if (replicationExecutor != null) {
93 replicationExecutor.resizeQueues(conf);
94 }
95
96 String callQueueType = conf.get(CALL_QUEUE_TYPE_CONF_KEY, CALL_QUEUE_TYPE_CONF_DEFAULT);
97 if (isCodelQueueType(callQueueType)) {
98
99 int codelTargetDelay = conf.getInt(CALL_QUEUE_CODEL_TARGET_DELAY,
100 CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY);
101 int codelInterval = conf.getInt(CALL_QUEUE_CODEL_INTERVAL,
102 CALL_QUEUE_CODEL_DEFAULT_INTERVAL);
103 double codelLifoThreshold = conf.getDouble(CALL_QUEUE_CODEL_LIFO_THRESHOLD,
104 CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD);
105
106 for (BlockingQueue<CallRunner> queue : callExecutor.getQueues()) {
107 if (queue instanceof AdaptiveLifoCoDelCallQueue) {
108 ((AdaptiveLifoCoDelCallQueue) queue).updateTunables(codelTargetDelay,
109 codelInterval, codelLifoThreshold);
110 }
111 }
112 }
113 }
114
115
116
117
118
119
120
121
122 private static class CallPriorityComparator implements Comparator<CallRunner> {
123 private final static int DEFAULT_MAX_CALL_DELAY = 5000;
124
125 private final PriorityFunction priority;
126 private final int maxDelay;
127
128 public CallPriorityComparator(final Configuration conf, final PriorityFunction priority) {
129 this.priority = priority;
130 this.maxDelay = conf.getInt(QUEUE_MAX_CALL_DELAY_CONF_KEY, DEFAULT_MAX_CALL_DELAY);
131 }
132
133 @Override
134 public int compare(CallRunner a, CallRunner b) {
135 RpcServer.Call callA = a.getCall();
136 RpcServer.Call callB = b.getCall();
137 long deadlineA = priority.getDeadline(callA.getHeader(), callA.param);
138 long deadlineB = priority.getDeadline(callB.getHeader(), callB.param);
139 deadlineA = callA.timestamp + Math.min(deadlineA, maxDelay);
140 deadlineB = callB.timestamp + Math.min(deadlineB, maxDelay);
141 return Long.compare(deadlineA, deadlineB);
142 }
143 }
144
145 private int port;
146 private final PriorityFunction priority;
147 private final RpcExecutor callExecutor;
148 private final RpcExecutor priorityExecutor;
149 private final RpcExecutor replicationExecutor;
150
151
152 private final int highPriorityLevel;
153
154 private Abortable abortable = null;
155
156
157
158
159
160
161
162
163
164 public SimpleRpcScheduler(
165 Configuration conf,
166 int handlerCount,
167 int priorityHandlerCount,
168 int replicationHandlerCount,
169 PriorityFunction priority,
170 Abortable server,
171 int highPriorityLevel) {
172
173 int maxQueueLength = conf.getInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH,
174 handlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER);
175 int maxPriorityQueueLength =
176 conf.getInt(RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH, maxQueueLength);
177
178 this.priority = priority;
179 this.highPriorityLevel = highPriorityLevel;
180 this.abortable = server;
181
182 String callQueueType = conf.get(CALL_QUEUE_TYPE_CONF_KEY, CALL_QUEUE_TYPE_CONF_DEFAULT);
183 float callqReadShare = conf.getFloat(CALL_QUEUE_READ_SHARE_CONF_KEY, 0);
184 float callqScanShare = conf.getFloat(CALL_QUEUE_SCAN_SHARE_CONF_KEY, 0);
185
186 int codelTargetDelay = conf.getInt(CALL_QUEUE_CODEL_TARGET_DELAY,
187 CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY);
188 int codelInterval = conf.getInt(CALL_QUEUE_CODEL_INTERVAL,
189 CALL_QUEUE_CODEL_DEFAULT_INTERVAL);
190 double codelLifoThreshold = conf.getDouble(CALL_QUEUE_CODEL_LIFO_THRESHOLD,
191 CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD);
192
193 float callQueuesHandlersFactor = conf.getFloat(CALL_QUEUE_HANDLER_FACTOR_CONF_KEY, 0);
194 int numCallQueues = Math.max(1, (int)Math.round(handlerCount * callQueuesHandlersFactor));
195
196 LOG.info("Using " + callQueueType + " as user call queue, count=" + numCallQueues);
197
198 if (numCallQueues > 1 && callqReadShare > 0) {
199
200 if (isDeadlineQueueType(callQueueType)) {
201 CallPriorityComparator callPriority = new CallPriorityComparator(conf, this.priority);
202 callExecutor = new RWQueueRpcExecutor("DeadlineRWQ.default", handlerCount, numCallQueues,
203 callqReadShare, callqScanShare, maxQueueLength, conf, abortable,
204 BoundedPriorityBlockingQueue.class, callPriority);
205 } else if (isCodelQueueType(callQueueType)) {
206 Object[] callQueueInitArgs = {maxQueueLength, codelTargetDelay, codelInterval,
207 codelLifoThreshold, numGeneralCallsDropped, numLifoModeSwitches};
208 callExecutor = new RWQueueRpcExecutor("CodelRWQ.default", handlerCount,
209 numCallQueues, callqReadShare, callqScanShare,
210 AdaptiveLifoCoDelCallQueue.class, callQueueInitArgs,
211 AdaptiveLifoCoDelCallQueue.class, callQueueInitArgs);
212 } else {
213
214 callExecutor = new RWQueueRpcExecutor("FifoRWQ.default", handlerCount, numCallQueues,
215 callqReadShare, callqScanShare, maxQueueLength, conf, abortable);
216 }
217 } else {
218
219 if (isDeadlineQueueType(callQueueType)) {
220 CallPriorityComparator callPriority = new CallPriorityComparator(conf, this.priority);
221 callExecutor =
222 new BalancedQueueRpcExecutor("BQDeadline.default", handlerCount, numCallQueues,
223 conf, abortable, BoundedPriorityBlockingQueue.class, maxQueueLength, callPriority);
224 } else if (isCodelQueueType(callQueueType)) {
225 callExecutor =
226 new FastPathBalancedQueueRpcExecutor("CodelFPBQ.default", handlerCount, numCallQueues,
227 conf, abortable, AdaptiveLifoCoDelCallQueue.class, maxQueueLength,
228 codelTargetDelay, codelInterval, codelLifoThreshold,
229 numGeneralCallsDropped, numLifoModeSwitches);
230 } else {
231
232 callExecutor = new FastPathBalancedQueueRpcExecutor("FifoWFPBQ.default",
233 handlerCount, numCallQueues, maxQueueLength, conf, abortable);
234 }
235 }
236
237
238 this.priorityExecutor = priorityHandlerCount > 0?
239 new FastPathBalancedQueueRpcExecutor("FifoWFPBQ.priority", priorityHandlerCount,
240 2, maxPriorityQueueLength, conf, abortable): null;
241 this.replicationExecutor = replicationHandlerCount > 0?
242 new FastPathBalancedQueueRpcExecutor("FifoWFPBQ.replication",
243 replicationHandlerCount, 1, maxQueueLength, conf, abortable) : null;
244 }
245
246 private static boolean isDeadlineQueueType(final String callQueueType) {
247 return callQueueType.equals(CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE);
248 }
249
250 private static boolean isCodelQueueType(final String callQueueType) {
251 return callQueueType.equals(CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
252 }
253
254 public SimpleRpcScheduler(
255 Configuration conf,
256 int handlerCount,
257 int priorityHandlerCount,
258 int replicationHandlerCount,
259 PriorityFunction priority,
260 int highPriorityLevel) {
261 this(conf, handlerCount, priorityHandlerCount, replicationHandlerCount, priority,
262 null, highPriorityLevel);
263 }
264
265 @Override
266 public void init(Context context) {
267 this.port = context.getListenerAddress().getPort();
268 }
269
270 @Override
271 public void start() {
272 callExecutor.start(port);
273 if (priorityExecutor != null) priorityExecutor.start(port);
274 if (replicationExecutor != null) replicationExecutor.start(port);
275 }
276
277 @Override
278 public void stop() {
279 callExecutor.stop();
280 if (priorityExecutor != null) priorityExecutor.stop();
281 if (replicationExecutor != null) replicationExecutor.stop();
282 }
283
284 @Override
285 public boolean dispatch(CallRunner callTask) throws InterruptedException {
286 RpcServer.Call call = callTask.getCall();
287 int level = priority.getPriority(call.getHeader(), call.param, call.getRequestUser());
288 if (priorityExecutor != null && level > highPriorityLevel) {
289 return priorityExecutor.dispatch(callTask);
290 } else if (replicationExecutor != null && level == HConstants.REPLICATION_QOS) {
291 return replicationExecutor.dispatch(callTask);
292 } else {
293 return callExecutor.dispatch(callTask);
294 }
295 }
296
297 @Override
298 public int getGeneralQueueLength() {
299 return callExecutor.getQueueLength();
300 }
301
302 @Override
303 public int getPriorityQueueLength() {
304 return priorityExecutor == null ? 0 : priorityExecutor.getQueueLength();
305 }
306
307 @Override
308 public int getReplicationQueueLength() {
309 return replicationExecutor == null ? 0 : replicationExecutor.getQueueLength();
310 }
311
312 @Override
313 public int getActiveRpcHandlerCount() {
314 return callExecutor.getActiveHandlerCount() +
315 (priorityExecutor == null ? 0 : priorityExecutor.getActiveHandlerCount()) +
316 (replicationExecutor == null ? 0 : replicationExecutor.getActiveHandlerCount());
317 }
318
319 @Override
320 public long getNumGeneralCallsDropped() {
321 return numGeneralCallsDropped.get();
322 }
323
324 @Override
325 public long getNumLifoModeSwitches() {
326 return numLifoModeSwitches.get();
327 }
328 }
329