001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.ipc;
019
020import java.io.IOException;
021import java.util.HashMap;
022import java.util.concurrent.ArrayBlockingQueue;
023import java.util.concurrent.ThreadPoolExecutor;
024import java.util.concurrent.TimeUnit;
025import java.util.concurrent.atomic.AtomicInteger;
026
027import org.apache.hadoop.conf.Configuration;
028import org.apache.hadoop.hbase.DaemonThreadFactory;
029import org.apache.yetus.audience.InterfaceAudience;
030import org.apache.yetus.audience.InterfaceStability;
031import org.slf4j.Logger;
032import org.slf4j.LoggerFactory;
033
034/**
035 * A special {@code }RpcScheduler} only used for master. This scheduler separates RegionServerReport
036 * requests to independent handlers to avoid these requests block other requests. To use this
037 * scheduler, please set "hbase.master.rpc.scheduler.factory.class" to
038 * "org.apache.hadoop.hbase.ipc.MasterFifoRpcScheduler".
039 */
040@InterfaceAudience.Private
041@InterfaceStability.Evolving
042public class MasterFifoRpcScheduler extends FifoRpcScheduler {
043  private static final Logger LOG = LoggerFactory.getLogger(MasterFifoRpcScheduler.class);
044
045  /**
046   * Set RSReport requests handlers count when masters use MasterFifoRpcScheduler. The default value
047   * is half of "hbase.regionserver.handler.count" value, but at least 1. The other handlers count
048   * is "hbase.regionserver.handler.count" value minus RSReport handlers count, but at least 1 too.
049   */
050  public static final String MASTER_SERVER_REPORT_HANDLER_COUNT =
051      "hbase.master.server.report.handler.count";
052  private static final String REGION_SERVER_REPORT = "RegionServerReport";
053  private final int rsReportHandlerCount;
054  private final int rsRsreportMaxQueueLength;
055  private final AtomicInteger rsReportQueueSize = new AtomicInteger(0);
056  private ThreadPoolExecutor rsReportExecutor;
057
058  public MasterFifoRpcScheduler(Configuration conf, int callHandlerCount,
059      int rsReportHandlerCount) {
060    super(conf, callHandlerCount);
061    this.rsReportHandlerCount = rsReportHandlerCount;
062    this.rsRsreportMaxQueueLength = conf.getInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH,
063      rsReportHandlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER);
064  }
065
066  @Override
067  public void start() {
068    LOG.info(
069      "Using {} as call queue; handlerCount={}; maxQueueLength={}; rsReportHandlerCount={}; "
070          + "rsReportMaxQueueLength={}",
071      this.getClass().getSimpleName(), handlerCount, maxQueueLength, rsReportHandlerCount,
072      rsRsreportMaxQueueLength);
073    this.executor = new ThreadPoolExecutor(handlerCount, handlerCount, 60, TimeUnit.SECONDS,
074        new ArrayBlockingQueue<Runnable>(maxQueueLength),
075        new DaemonThreadFactory("MasterFifoRpcScheduler.call.handler"),
076        new ThreadPoolExecutor.CallerRunsPolicy());
077    this.rsReportExecutor = new ThreadPoolExecutor(rsReportHandlerCount, rsReportHandlerCount, 60,
078        TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(rsRsreportMaxQueueLength),
079        new DaemonThreadFactory("MasterFifoRpcScheduler.RSReport.handler"),
080        new ThreadPoolExecutor.CallerRunsPolicy());
081  }
082
083  @Override
084  public void stop() {
085    this.executor.shutdown();
086    this.rsReportExecutor.shutdown();
087  }
088
089  @Override
090  public boolean dispatch(final CallRunner task) throws IOException, InterruptedException {
091    String method = getCallMethod(task);
092    if (rsReportExecutor != null && method != null && method.equals(REGION_SERVER_REPORT)) {
093      return executeRpcCall(rsReportExecutor, rsReportQueueSize, task);
094    } else {
095      return executeRpcCall(executor, queueSize, task);
096    }
097  }
098
099  @Override
100  public int getGeneralQueueLength() {
101    return executor.getQueue().size() + rsReportExecutor.getQueue().size();
102  }
103
104  @Override
105  public int getActiveRpcHandlerCount() {
106    return executor.getActiveCount() + rsReportExecutor.getActiveCount();
107  }
108
109  @Override
110  public CallQueueInfo getCallQueueInfo() {
111    String queueName = "Master Fifo Queue";
112
113    HashMap<String, Long> methodCount = new HashMap<>();
114    HashMap<String, Long> methodSize = new HashMap<>();
115
116    CallQueueInfo callQueueInfo = new CallQueueInfo();
117    callQueueInfo.setCallMethodCount(queueName, methodCount);
118    callQueueInfo.setCallMethodSize(queueName, methodSize);
119
120    updateMethodCountAndSizeByQueue(executor.getQueue(), methodCount, methodSize);
121    updateMethodCountAndSizeByQueue(rsReportExecutor.getQueue(), methodCount, methodSize);
122
123    return callQueueInfo;
124  }
125}