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}