View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * Reads special method annotations and table names to figure a priority for use by QoS facility in
50   * ipc; e.g: rpcs to hbase:meta get priority.
51   */
52  // TODO: Remove.  This is doing way too much work just to figure a priority.  Do as Elliott
53  // suggests and just have the client specify a priority.
54  
55  //The logic for figuring out high priority RPCs is as follows:
56  //1. if the method is annotated with a QosPriority of QOS_HIGH,
57  //   that is honored
58  //2. parse out the protobuf message and see if the request is for meta
59  //   region, and if so, treat it as a high priority RPC
60  //Some optimizations for (2) are done here -
61  //Clients send the argument classname as part of making the RPC. The server
62  //decides whether to deserialize the proto argument message based on the
63  //pre-established set of argument classes (knownArgumentClasses below).
64  //This prevents the server from having to deserialize all proto argument
65  //messages prematurely.
66  //All the argument classes declare a 'getRegion' method that returns a
67  //RegionSpecifier object. Methods can be invoked on the returned object
68  //to figure out whether it is a meta region or not.
69  @InterfaceAudience.Private
70  public class AnnotationReadingPriorityFunction implements PriorityFunction {
71    private static final Log LOG =
72      LogFactory.getLog(AnnotationReadingPriorityFunction.class.getName());
73  
74    /** Used to control the scan delay, currently sqrt(numNextCall * weight) */
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    //We need to mock the regionserver instance for some unit tests (set via
79    //setRegionServer method.
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    // Some caches for helping performance
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    * Calls {@link #AnnotationReadingPriorityFunction(RSRpcServices, Class)} using the result of
104    * {@code rpcServices#getClass()}
105    *
106    * @param rpcServices
107    *          The RPC server implementation
108    */
109   public AnnotationReadingPriorityFunction(final RSRpcServices rpcServices) {
110     this(rpcServices, rpcServices.getClass());
111   }
112 
113   /**
114    * Constructs the priority function given the RPC server implementation and the annotations on the
115    * methods in the provided {@code clz}.
116    *
117    * @param rpcServices
118    *          The RPC server implementation
119    * @param clz
120    *          The concrete RPC server implementation's class
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         // Since we protobuf'd, and then subsequently, when we went with pb style, method names
129         // are capitalized.  This meant that this brittle compare of method names gotten by
130         // reflection no longer matched the method names coming in over pb.  TODO: Get rid of this
131         // check.  For now, workaround is to capitalize the names we got from reflection so they
132         // have chance of matching the pb ones.
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    * Returns a 'priority' based on the request type.
165    *
166    * Currently the returned priority is used for queue selection.
167    * See the SimpleRpcScheduler as example. It maintains a queue per 'priory type'
168    * HIGH_QOS (meta requests), REPLICATION_QOS (replication requests),
169    * NORMAL_QOS (user requests).
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    * See if the method has an annotation.
183    * @param header
184    * @return Return the priority from the annotation. If there isn't
185    * an annotation, this returns something below zero.
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    * Get the priority for a given request from the header and the param
198    * This doesn't consider which user is sending the request at all.
199    * This doesn't consider annotations
200    */
201   protected int getBasePriority(RequestHeader header, Message param) {
202     if (param == null) {
203       return HConstants.NORMAL_QOS;
204     }
205 
206     // Trust the client-set priorities if set
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     //check whether the request has reference to meta region or now.
215     try {
216       // Check if the param has a region specifier; the pb methods are hasRegion and getRegion if
217       // hasRegion returns true.  Not all listed methods have region specifier each time.  For
218       // example, the ScanRequest has it on setup but thereafter relies on the scannerid rather than
219       // send the region over every time.
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       // Not good throwing an exception out of here, a runtime anyways.  Let the query go into the
235       // server and have it throw the exception if still an issue.  Just mark it normal priority.
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) { // scanner methods...
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           // Scanner requests are small in size so TextFormat version should not overwhelm log.
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    * Based on the request content, returns the deadline of the request.
260    *
261    * @param header
262    * @param param
263    * @return Deadline of this request. 0 now, otherwise msec of 'delay'
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       // get the 'virtual time' of the scanner, and applies sqrt() to get a
274       // nice curve for the delay. More a scanner is used the less priority it gets.
275       // The weight is used to have more control on the delay.
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 }