1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.regionserver;
19
20 import java.lang.reflect.Method;
21 import java.util.HashMap;
22 import java.util.Map;
23
24 import org.apache.commons.logging.Log;
25 import org.apache.commons.logging.LogFactory;
26 import org.apache.hadoop.conf.Configuration;
27 import org.apache.hadoop.hbase.HConstants;
28 import org.apache.hadoop.hbase.classification.InterfaceAudience;
29 import org.apache.hadoop.hbase.ipc.PriorityFunction;
30 import org.apache.hadoop.hbase.ipc.QosPriority;
31 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CloseRegionRequest;
32 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest;
33 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionRequest;
34 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetRegionInfoRequest;
35 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.GetStoreFileRequest;
36 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.SplitRegionRequest;
37 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest;
38 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
39 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
40 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier;
41 import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader;
42
43 import com.google.common.annotations.VisibleForTesting;
44 import com.google.protobuf.Message;
45 import com.google.protobuf.TextFormat;
46 import org.apache.hadoop.hbase.security.User;
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69 @InterfaceAudience.Private
70 public class AnnotationReadingPriorityFunction implements PriorityFunction {
71 private static final Log LOG =
72 LogFactory.getLog(AnnotationReadingPriorityFunction.class.getName());
73
74
75 public static final String SCAN_VTIME_WEIGHT_CONF_KEY = "hbase.ipc.server.scan.vtime.weight";
76
77 protected final Map<String, Integer> annotatedQos;
78
79
80 private RSRpcServices rpcServices;
81 @SuppressWarnings("unchecked")
82 private final Class<? extends Message>[] knownArgumentClasses = new Class[]{
83 GetRegionInfoRequest.class,
84 GetStoreFileRequest.class,
85 CloseRegionRequest.class,
86 FlushRegionRequest.class,
87 SplitRegionRequest.class,
88 CompactRegionRequest.class,
89 GetRequest.class,
90 MutateRequest.class,
91 ScanRequest.class
92 };
93
94
95 private final Map<String, Class<? extends Message>> argumentToClassMap =
96 new HashMap<String, Class<? extends Message>>();
97 private final Map<String, Map<Class<? extends Message>, Method>> methodMap =
98 new HashMap<String, Map<Class<? extends Message>, Method>>();
99
100 private final float scanVirtualTimeWeight;
101
102
103
104
105
106
107
108
109 public AnnotationReadingPriorityFunction(final RSRpcServices rpcServices) {
110 this(rpcServices, rpcServices.getClass());
111 }
112
113
114
115
116
117
118
119
120
121
122 public AnnotationReadingPriorityFunction(final RSRpcServices rpcServices,
123 Class<? extends RSRpcServices> clz) {
124 Map<String,Integer> qosMap = new HashMap<String,Integer>();
125 for (Method m : clz.getMethods()) {
126 QosPriority p = m.getAnnotation(QosPriority.class);
127 if (p != null) {
128
129
130
131
132
133 String capitalizedMethodName = capitalize(m.getName());
134 qosMap.put(capitalizedMethodName, p.priority());
135 }
136 }
137 this.rpcServices = rpcServices;
138 this.annotatedQos = qosMap;
139 if (methodMap.get("getRegion") == null) {
140 methodMap.put("hasRegion", new HashMap<Class<? extends Message>, Method>());
141 methodMap.put("getRegion", new HashMap<Class<? extends Message>, Method>());
142 }
143 for (Class<? extends Message> cls : knownArgumentClasses) {
144 argumentToClassMap.put(cls.getName(), cls);
145 try {
146 methodMap.get("hasRegion").put(cls, cls.getDeclaredMethod("hasRegion"));
147 methodMap.get("getRegion").put(cls, cls.getDeclaredMethod("getRegion"));
148 } catch (Exception e) {
149 throw new RuntimeException(e);
150 }
151 }
152
153 Configuration conf = rpcServices.getConfiguration();
154 scanVirtualTimeWeight = conf.getFloat(SCAN_VTIME_WEIGHT_CONF_KEY, 1.0f);
155 }
156
157 private String capitalize(final String s) {
158 StringBuilder strBuilder = new StringBuilder(s);
159 strBuilder.setCharAt(0, Character.toUpperCase(strBuilder.charAt(0)));
160 return strBuilder.toString();
161 }
162
163
164
165
166
167
168
169
170
171 @Override
172 public int getPriority(RequestHeader header, Message param, User user) {
173 int priorityByAnnotation = getAnnotatedPriority(header);
174
175 if (priorityByAnnotation >= 0) {
176 return priorityByAnnotation;
177 }
178 return getBasePriority(header, param);
179 }
180
181
182
183
184
185
186
187 protected int getAnnotatedPriority(RequestHeader header) {
188 String methodName = header.getMethodName();
189 Integer priorityByAnnotation = annotatedQos.get(methodName);
190 if (priorityByAnnotation != null) {
191 return priorityByAnnotation;
192 }
193 return -1;
194 }
195
196
197
198
199
200
201 protected int getBasePriority(RequestHeader header, Message param) {
202 if (param == null) {
203 return HConstants.NORMAL_QOS;
204 }
205
206
207 if (header.hasPriority()) {
208 return header.getPriority();
209 }
210
211 String cls = param.getClass().getName();
212 Class<? extends Message> rpcArgClass = argumentToClassMap.get(cls);
213 RegionSpecifier regionSpecifier = null;
214
215 try {
216
217
218
219
220 Method hasRegion = methodMap.get("hasRegion").get(rpcArgClass);
221 if (hasRegion != null && (Boolean)hasRegion.invoke(param, (Object[])null)) {
222 Method getRegion = methodMap.get("getRegion").get(rpcArgClass);
223 regionSpecifier = (RegionSpecifier)getRegion.invoke(param, (Object[])null);
224 Region region = rpcServices.getRegion(regionSpecifier);
225 if (region.getRegionInfo().isSystemTable()) {
226 if (LOG.isTraceEnabled()) {
227 LOG.trace("High priority because region=" +
228 region.getRegionInfo().getRegionNameAsString());
229 }
230 return HConstants.SYSTEMTABLE_QOS;
231 }
232 }
233 } catch (Exception ex) {
234
235
236 if (LOG.isTraceEnabled()) LOG.trace("Marking normal priority after getting exception=" + ex);
237 return HConstants.NORMAL_QOS;
238 }
239
240 if (param instanceof ScanRequest) {
241 ScanRequest request = (ScanRequest)param;
242 if (!request.hasScannerId()) {
243 return HConstants.NORMAL_QOS;
244 }
245 RegionScanner scanner = rpcServices.getScanner(request.getScannerId());
246 if (scanner != null && scanner.getRegionInfo().isSystemTable()) {
247 if (LOG.isTraceEnabled()) {
248
249 LOG.trace("High priority scanner request " + TextFormat.shortDebugString(request));
250 }
251 return HConstants.SYSTEMTABLE_QOS;
252 }
253 }
254
255 return HConstants.NORMAL_QOS;
256 }
257
258
259
260
261
262
263
264
265 @Override
266 public long getDeadline(RequestHeader header, Message param) {
267 if (param instanceof ScanRequest) {
268 ScanRequest request = (ScanRequest)param;
269 if (!request.hasScannerId()) {
270 return 0;
271 }
272
273
274
275
276 long vtime = rpcServices.getScannerVirtualTime(request.getScannerId());
277 return Math.round(Math.sqrt(vtime * scanVirtualTimeWeight));
278 }
279 return 0;
280 }
281
282 @VisibleForTesting
283 void setRegionServer(final HRegionServer hrs) {
284 this.rpcServices = hrs.getRSRpcServices();
285 }
286 }