import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
/**
 * Returns a 'priority' based on the request type.
 *
 * Currently the returned priority is used for queue selection.
 * See the SimpleRpcScheduler as example. It maintains a queue per 'priory type'
 * HIGH_QOS (meta requests), REPLICATION_QOS (replication requests),
 * NORMAL_QOS (user requests).
 */
@Override
public int getPriority(RequestHeader header, Message param, User user) {
  int priorityByAnnotation = getAnnotatedPriority(header);

  if (priorityByAnnotation >= 0) {
    return priorityByAnnotation;
  }

  // all requests executed by super users have high QoS
  try {
    if (Superusers.isSuperUser(user)) {
      return HConstants.ADMIN_QOS;
    }
  } catch (IllegalStateException ex) {
    // Not good throwing an exception out of here, a runtime anyways.  Let the query go into the
    // server and have it throw the exception if still an issue.  Just mark it normal priority.
    if (LOG.isTraceEnabled()) LOG.trace("Marking normal priority after getting exception=" + ex);
    return HConstants.NORMAL_QOS;
  }

  return getBasePriority(header, param);
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
/**
 * Based on the request content, returns the deadline of the request.
 *
 * @param header
 * @param param
 * @return Deadline of this request. 0 now, otherwise msec of 'delay'
 */
@Override
public long getDeadline(RequestHeader header, Message param) {
  if (param instanceof ScanRequest) {
    ScanRequest request = (ScanRequest)param;
    if (!request.hasScannerId()) {
      return 0;
    }

    // get the 'virtual time' of the scanner, and applies sqrt() to get a
    // nice curve for the delay. More a scanner is used the less priority it gets.
    // The weight is used to have more control on the delay.
    long vtime = rpcServices.getScannerVirtualTime(request.getScannerId());
    return Math.round(Math.sqrt(vtime * scanVirtualTimeWeight));
  }
  return 0;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NP_NULL_ON_SOME_PATH",
    justification="Can't figure why this complaint is happening... see below")
Call(int id, final BlockingService service, final MethodDescriptor md, RequestHeader header,
     Message param, CellScanner cellScanner, Connection connection, Responder responder,
     long size, TraceInfo tinfo, final InetAddress remoteAddress) {
  this.id = id;
  this.service = service;
  this.md = md;
  this.header = header;
  this.param = param;
  this.cellScanner = cellScanner;
  this.connection = connection;
  this.timestamp = System.currentTimeMillis();
  this.response = null;
  this.responder = responder;
  this.isError = false;
  this.size = size;
  this.tinfo = tinfo;
  this.user = connection == null? null: connection.user; // FindBugs: NP_NULL_ON_SOME_PATH
  this.remoteAddress = remoteAddress;
  this.retryImmediatelySupported =
      connection == null? null: connection.retryImmediatelySupported;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForRequestCalledBySuperUser() throws Exception {
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  headerBuilder.setMethodName("foo");
  RequestHeader header = headerBuilder.build();
  PriorityFunction qosFunc = regionServer.rpcServices.getPriority();

  //test superusers
  regionServer.conf.set(Superusers.SUPERUSER_CONF_KEY, "samplesuperuser");
  Superusers.initialize(regionServer.conf);
  assertEquals(HConstants.ADMIN_QOS, qosFunc.getPriority(header, null,
    User.createUserForTesting(regionServer.conf, "samplesuperuser",
      new String[]{"somegroup"})));

  //test supergroups
  regionServer.conf.set(Superusers.SUPERUSER_CONF_KEY, "@samplesupergroup");
  Superusers.initialize(regionServer.conf);
  assertEquals(HConstants.ADMIN_QOS, qosFunc.getPriority(header, null,
    User.createUserForTesting(regionServer.conf, "regularuser",
      new String[]{"samplesupergroup"})));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
/**
 * Based on the request content, returns the deadline of the request.
 *
 * @param header
 * @param param
 * @return Deadline of this request. 0 now, otherwise msec of 'delay'
 */
@Override
public long getDeadline(RequestHeader header, Message param) {
  String methodName = header.getMethodName();
  if (methodName.equalsIgnoreCase("scan")) {
    ScanRequest request = (ScanRequest)param;
    if (!request.hasScannerId()) {
      return 0;
    }

    // get the 'virtual time' of the scanner, and applies sqrt() to get a
    // nice curve for the delay. More a scanner is used the less priority it gets.
    // The weight is used to have more control on the delay.
    long vtime = rpcServices.getScannerVirtualTime(request.getScannerId());
    return Math.round(Math.sqrt(vtime * scanVirtualTimeWeight));
  }
  return 0;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
Call(int id, final BlockingService service, final MethodDescriptor md, RequestHeader header,
     Message param, CellScanner cellScanner, Connection connection, Responder responder,
     long size, TraceInfo tinfo) {
  this.id = id;
  this.service = service;
  this.md = md;
  this.header = header;
  this.param = param;
  this.cellScanner = cellScanner;
  this.connection = connection;
  this.timestamp = System.currentTimeMillis();
  this.response = null;
  this.delayResponse = false;
  this.responder = responder;
  this.isError = false;
  this.size = size;
  this.tinfo = tinfo;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
private boolean isWriteRequest(final RequestHeader header, final Message param) {
  // TODO: Is there a better way to do this?
  String methodName = header.getMethodName();
  if (methodName.equalsIgnoreCase("multi") && param instanceof MultiRequest) {
    MultiRequest multi = (MultiRequest)param;
    for (RegionAction regionAction : multi.getRegionActionList()) {
      for (Action action: regionAction.getActionList()) {
        if (action.hasMutation()) {
          return true;
        }
      }
    }
  }
  if (methodName.equalsIgnoreCase("mutate")) {
    return true;
  }
  return false;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
/**
 * See if the method has an annotation.
 * @param header
 * @return Return the priority from the annotation. If there isn't
 * an annotation, this returns something below zero.
 */
protected int getAnnotatedPriority(RequestHeader header) {
  String methodName = header.getMethodName();
  Integer priorityByAnnotation = annotatedQos.get(methodName);
  if (priorityByAnnotation != null) {
    return priorityByAnnotation;
  }
  return -1;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
private boolean isWriteRequest(final RequestHeader header, final Message param) {
  // TODO: Is there a better way to do this?
  if (param instanceof MultiRequest) {
    MultiRequest multi = (MultiRequest)param;
    for (RegionAction regionAction : multi.getRegionActionList()) {
      for (Action action: regionAction.getActionList()) {
        if (action.hasMutation()) {
          return true;
        }
      }
    }
  }
  if (param instanceof MutateRequest) {
    return true;
  }
  // Below here are methods for master. It's a pretty brittle version of this.
  // Not sure that master actually needs a read/write queue since 90% of requests to
  // master are writing to status or changing the meta table.
  // All other read requests are admin generated and can be processed whenever.
  // However changing that would require a pretty drastic change and should be done for
  // the next major release and not as a fix for HBASE-14239
  if (param instanceof RegionServerStatusProtos.ReportRegionStateTransitionRequest) {
    return true;
  }
  if (param instanceof RegionServerStatusProtos.RegionServerStartupRequest) {
    return true;
  }
  if (param instanceof RegionServerStatusProtos.RegionServerReportRequest) {
    return true;
  }
  return false;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
private boolean isScanRequest(final RequestHeader header, final Message param) {
  if (param instanceof ScanRequest) {
    // The first scan request will be executed as a "short read"
    ScanRequest request = (ScanRequest)param;
    return request.hasScannerId();
  }
  return false;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForMeta() throws IOException {
  priority = regionServer.rpcServices.getPriority();
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  //create a rpc request that has references to hbase:meta region and also
  //uses one of the known argument classes (known argument classes are
  //listed in HRegionServer.QosFunctionImpl.knownArgumentClasses)
  headerBuilder.setMethodName("foo");

  GetRequest.Builder getRequestBuilder = GetRequest.newBuilder();
  RegionSpecifier.Builder regionSpecifierBuilder = RegionSpecifier.newBuilder();
  regionSpecifierBuilder.setType(RegionSpecifierType.REGION_NAME);
  ByteString name = ByteStringer.wrap(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
  regionSpecifierBuilder.setValue(name);
  RegionSpecifier regionSpecifier = regionSpecifierBuilder.build();
  getRequestBuilder.setRegion(regionSpecifier);
  Get.Builder getBuilder = Get.newBuilder();
  getBuilder.setRow(ByteStringer.wrap("somerow".getBytes()));
  getRequestBuilder.setGet(getBuilder.build());
  GetRequest getRequest = getRequestBuilder.build();
  RequestHeader header = headerBuilder.build();
  HRegion mockRegion = Mockito.mock(HRegion.class);
  HRegionServer mockRS = Mockito.mock(HRegionServer.class);
  RSRpcServices mockRpc = Mockito.mock(RSRpcServices.class);
  Mockito.when(mockRS.getRSRpcServices()).thenReturn(mockRpc);
  HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class);
  Mockito.when(mockRpc.getRegion((RegionSpecifier) Mockito.any())).thenReturn(mockRegion);
  Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo);
  Mockito.when(mockRegionInfo.isSystemTable()).thenReturn(true);
  // Presume type.
  ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS);
  assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, getRequest,
    User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"})));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionWithoutKnownArgument() throws IOException {
  //The request is not using any of the
  //known argument classes (it uses one random request class)
  //(known argument classes are listed in
  //HRegionServer.QosFunctionImpl.knownArgumentClasses)
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  headerBuilder.setMethodName("foo");
  RequestHeader header = headerBuilder.build();
  PriorityFunction qosFunc = regionServer.rpcServices.getPriority();
  assertEquals(HConstants.NORMAL_QOS, qosFunc.getPriority(header, null,
    User.createUserForTesting(regionServer.conf, "someuser", new String[]{"somegroup"})));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
private boolean isScanRequest(final RequestHeader header, final Message param) {
  String methodName = header.getMethodName();
  if (methodName.equalsIgnoreCase("scan")) {
    // The first scan request will be executed as a "short read"
    ScanRequest request = (ScanRequest)param;
    return request.hasScannerId();
  }
  return false;
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForMeta() throws IOException {
  priority = regionServer.rpcServices.getPriority();
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  //create a rpc request that has references to hbase:meta region and also
  //uses one of the known argument classes (known argument classes are
  //listed in HRegionServer.QosFunctionImpl.knownArgumentClasses)
  headerBuilder.setMethodName("foo");

  GetRequest.Builder getRequestBuilder = GetRequest.newBuilder();
  RegionSpecifier.Builder regionSpecifierBuilder = RegionSpecifier.newBuilder();
  regionSpecifierBuilder.setType(RegionSpecifierType.REGION_NAME);
  ByteString name = ByteStringer.wrap(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
  regionSpecifierBuilder.setValue(name);
  RegionSpecifier regionSpecifier = regionSpecifierBuilder.build();
  getRequestBuilder.setRegion(regionSpecifier);
  Get.Builder getBuilder = Get.newBuilder();
  getBuilder.setRow(ByteStringer.wrap("somerow".getBytes()));
  getRequestBuilder.setGet(getBuilder.build());
  GetRequest getRequest = getRequestBuilder.build();
  RequestHeader header = headerBuilder.build();
  HRegion mockRegion = Mockito.mock(HRegion.class);
  HRegionServer mockRS = Mockito.mock(HRegionServer.class);
  RSRpcServices mockRpc = Mockito.mock(RSRpcServices.class);
  Mockito.when(mockRS.getRSRpcServices()).thenReturn(mockRpc);
  HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class);
  Mockito.when(mockRpc.getRegion((RegionSpecifier)Mockito.any())).thenReturn(mockRegion);
  Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo);
  Mockito.when(mockRegionInfo.isSystemTable()).thenReturn(true);
  // Presume type.
  ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS);
  assertEquals(HConstants.SYSTEMTABLE_QOS, priority.getPriority(header, getRequest));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionWithoutKnownArgument() throws IOException {
  //The request is not using any of the
  //known argument classes (it uses one random request class)
  //(known argument classes are listed in
  //HRegionServer.QosFunctionImpl.knownArgumentClasses)
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  headerBuilder.setMethodName("foo");
  RequestHeader header = headerBuilder.build();
  PriorityFunction qosFunc = regionServer.rpcServices.getPriority();
  assertEquals(HConstants.NORMAL_QOS, qosFunc.getPriority(header, null));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForMeta() throws IOException {
  priority = regionServer.getPriority();
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  //create a rpc request that has references to hbase:meta region and also
  //uses one of the known argument classes (known argument classes are
  //listed in HRegionServer.QosFunctionImpl.knownArgumentClasses)
  headerBuilder.setMethodName("foo");

  GetRequest.Builder getRequestBuilder = GetRequest.newBuilder();
  RegionSpecifier.Builder regionSpecifierBuilder = RegionSpecifier.newBuilder();
  regionSpecifierBuilder.setType(RegionSpecifierType.REGION_NAME);
  ByteString name = HBaseZeroCopyByteString.wrap(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
  regionSpecifierBuilder.setValue(name);
  RegionSpecifier regionSpecifier = regionSpecifierBuilder.build();
  getRequestBuilder.setRegion(regionSpecifier);
  Get.Builder getBuilder = Get.newBuilder();
  getBuilder.setRow(HBaseZeroCopyByteString.wrap("somerow".getBytes()));
  getRequestBuilder.setGet(getBuilder.build());
  GetRequest getRequest = getRequestBuilder.build();
  RequestHeader header = headerBuilder.build();
  HRegion mockRegion = Mockito.mock(HRegion.class);
  HRegionServer mockRS = Mockito.mock(HRegionServer.class);
  HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class);
  Mockito.when(mockRS.getRegion((RegionSpecifier)Mockito.any())).thenReturn(mockRegion);
  Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo);
  Mockito.when(mockRegionInfo.isMetaTable()).thenReturn(true);
  // Presume type.
  ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS);
  assertEquals(HConstants.HIGH_QOS, priority.getPriority(header, getRequest));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionWithoutKnownArgument() throws IOException {
  //The request is not using any of the
  //known argument classes (it uses one random request class)
  //(known argument classes are listed in
  //HRegionServer.QosFunctionImpl.knownArgumentClasses)
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  headerBuilder.setMethodName("foo");
  RequestHeader header = headerBuilder.build();
  PriorityFunction qosFunc = regionServer.getPriority();
  assertEquals(HConstants.NORMAL_QOS, qosFunc.getPriority(header, null));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
/**
 * Initiates a call by sending the parameter to the remote server.
 * Note: this is not called from the Connection thread, but by other
 * threads.
 * @param call
 * @param priority
 * @see #readResponse()
 */
protected void writeRequest(Call call, final int priority) {
  if (shouldCloseConnection.get()) return;
  try {
    RequestHeader.Builder builder = RequestHeader.newBuilder();
    builder.setCallId(call.id);
    if (Trace.isTracing()) {
      Span s = Trace.currentSpan();
      builder.setTraceInfo(RPCTInfo.newBuilder().
        setParentId(s.getSpanId()).setTraceId(s.getTraceId()));
    }
    builder.setMethodName(call.md.getName());
    builder.setRequestParam(call.param != null);
    ByteBuffer cellBlock = ipcUtil.buildCellBlock(this.codec, this.compressor, call.cells);
    if (cellBlock != null) {
      CellBlockMeta.Builder cellBlockBuilder = CellBlockMeta.newBuilder();
      cellBlockBuilder.setLength(cellBlock.limit());
      builder.setCellBlockMeta(cellBlockBuilder.build());
    }
    // Only pass priority if there one.  Let zero be same as no priority.
    if (priority != 0) builder.setPriority(priority);
    //noinspection SynchronizeOnNonFinalField
    RequestHeader header = builder.build();
    synchronized (this.out) { // FindBugs IS2_INCONSISTENT_SYNC
      IPCUtil.write(this.out, header, call.param, cellBlock);
    }
    if (LOG.isDebugEnabled()) {
      LOG.debug(getName() + ": wrote request header " + TextFormat.shortDebugString(header));
    }
  } catch(IOException e) {
    markClosed(e);
  }
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForMeta() throws IOException {
  priority = regionServer.rpcServices.getPriority();
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  //create a rpc request that has references to hbase:meta region and also
  //uses one of the known argument classes (known argument classes are
  //listed in HRegionServer.QosFunctionImpl.knownArgumentClasses)
  headerBuilder.setMethodName("foo");

  GetRequest.Builder getRequestBuilder = GetRequest.newBuilder();
  RegionSpecifier.Builder regionSpecifierBuilder = RegionSpecifier.newBuilder();
  regionSpecifierBuilder.setType(RegionSpecifierType.REGION_NAME);
  ByteString name = HBaseZeroCopyByteString.wrap(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
  regionSpecifierBuilder.setValue(name);
  RegionSpecifier regionSpecifier = regionSpecifierBuilder.build();
  getRequestBuilder.setRegion(regionSpecifier);
  Get.Builder getBuilder = Get.newBuilder();
  getBuilder.setRow(HBaseZeroCopyByteString.wrap("somerow".getBytes()));
  getRequestBuilder.setGet(getBuilder.build());
  GetRequest getRequest = getRequestBuilder.build();
  RequestHeader header = headerBuilder.build();
  HRegion mockRegion = Mockito.mock(HRegion.class);
  HRegionServer mockRS = Mockito.mock(HRegionServer.class);
  RSRpcServices mockRpc = Mockito.mock(RSRpcServices.class);
  Mockito.when(mockRS.getRSRpcServices()).thenReturn(mockRpc);
  HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class);
  Mockito.when(mockRpc.getRegion((RegionSpecifier)Mockito.any())).thenReturn(mockRegion);
  Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo);
  Mockito.when(mockRegionInfo.isMetaTable()).thenReturn(true);
  // Presume type.
  ((AnnotationReadingPriorityFunction)priority).setRegionServer(mockRS);
  assertEquals(HConstants.HIGH_QOS, priority.getPriority(header, getRequest));
}
 

import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; //导入依赖的package包/类
@Test
public void testQosFunctionForMeta() throws IOException {
  qosFunction = regionServer.getQosFunction();
  RequestHeader.Builder headerBuilder = RequestHeader.newBuilder();
  //create a rpc request that has references to hbase:meta region and also
  //uses one of the known argument classes (known argument classes are
  //listed in HRegionServer.QosFunction.knownArgumentClasses)
  headerBuilder.setMethodName("foo");

  GetRequest.Builder getRequestBuilder = GetRequest.newBuilder();
  RegionSpecifier.Builder regionSpecifierBuilder = RegionSpecifier.newBuilder();
  regionSpecifierBuilder.setType(RegionSpecifierType.REGION_NAME);
  ByteString name = ZeroCopyLiteralByteString.wrap(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
  regionSpecifierBuilder.setValue(name);
  RegionSpecifier regionSpecifier = regionSpecifierBuilder.build();
  getRequestBuilder.setRegion(regionSpecifier);
  Get.Builder getBuilder = Get.newBuilder();
  getBuilder.setRow(ZeroCopyLiteralByteString.wrap("somerow".getBytes()));
  getRequestBuilder.setGet(getBuilder.build());
  GetRequest getRequest = getRequestBuilder.build();
  RequestHeader header = headerBuilder.build();
  HRegion mockRegion = Mockito.mock(HRegion.class);
  HRegionServer mockRS = Mockito.mock(HRegionServer.class);
  HRegionInfo mockRegionInfo = Mockito.mock(HRegionInfo.class);
  Mockito.when(mockRS.getRegion((RegionSpecifier)Mockito.any())).thenReturn(mockRegion);
  Mockito.when(mockRegion.getRegionInfo()).thenReturn(mockRegionInfo);
  Mockito.when(mockRegionInfo.isMetaTable()).thenReturn(true);
  qosFunction.setRegionServer(mockRS);
  assertTrue (qosFunction.apply(new Pair<RequestHeader, Message>(header, getRequest)) ==
    HConstants.HIGH_QOS);
}