Package org.apache.hadoop.hbase.ipc

Class RpcExecutor

java.lang.Object

org.apache.hadoop.hbase.ipc.RpcExecutor

Direct Known Subclasses:

BalancedQueueRpcExecutor, RWQueueRpcExecutor

@LimitedPrivate({"Coprocesssor","Phoenix"})
@Evolving
public abstract class RpcExecutor
extends Object

Runs the CallRunners passed here via dispatch(CallRunner). Subclass and add particular scheduling behavior.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private static class	RpcExecutor.CallPriorityComparator	Comparator used by the "normal callQueue" if DEADLINE_CALL_QUEUE_CONF_KEY is set to true.

Field Summary

Fields

Modifier and Type	Field	Description
private final Abortable	abortable
private final AtomicInteger	activeHandlerCount
static final int	CALL_QUEUE_CODEL_DEFAULT_INTERVAL
static final double	CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD
static final int	CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY
static final String	CALL_QUEUE_CODEL_INTERVAL
static final String	CALL_QUEUE_CODEL_LIFO_THRESHOLD
static final String	CALL_QUEUE_CODEL_TARGET_DELAY
static final String	CALL_QUEUE_HANDLER_FACTOR_CONF_KEY
static final String	CALL_QUEUE_QUEUE_BALANCER_CLASS
static final Class<?>	CALL_QUEUE_QUEUE_BALANCER_CLASS_DEFAULT
static final String	CALL_QUEUE_TYPE_CODEL_CONF_VALUE	The default, 'fifo', has the least friction but is dumb.
static final String	CALL_QUEUE_TYPE_CONF_DEFAULT
static final String	CALL_QUEUE_TYPE_CONF_KEY
static final String	CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE
static final String	CALL_QUEUE_TYPE_FIFO_CONF_VALUE
static final String	CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE
private final org.apache.hadoop.conf.Configuration	conf
protected int	currentQueueLimit
protected static final int	DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT
private final AtomicInteger	failedHandlerCount
private final int	handlerCount
private final List<RpcHandler>	handlers
private static final org.slf4j.Logger	LOG
private String	name
protected final int	numCallQueues
private final LongAdder	numGeneralCallsDropped
private final LongAdder	numLifoModeSwitches
private static final QueueBalancer	ONE_QUEUE	All requests go to the first queue, at index 0
static final String	PLUGGABLE_CALL_QUEUE_CLASS_NAME
static final String	PLUGGABLE_CALL_QUEUE_WITH_FAST_PATH_ENABLED
static final String	QUEUE_MAX_CALL_DELAY_CONF_KEY	max delay in msec used to bound the de-prioritized requests
private final Class<? extends BlockingQueue>	queueClass
private final Object[]	queueInitArgs
protected final List<BlockingQueue<CallRunner>>	queues

Constructor Summary

Constructors

Constructor	Description
RpcExecutor(String name, int handlerCount, int maxQueueLength, PriorityFunction priority, org.apache.hadoop.conf.Configuration conf, Abortable abortable)
RpcExecutor(String name, int handlerCount, String callQueueType, int maxQueueLength, PriorityFunction priority, org.apache.hadoop.conf.Configuration conf, Abortable abortable)

Method Summary

Modifier and Type	Method	Description
protected int	computeNumCallQueues(int handlerCount, float callQueuesHandlersFactor)
abstract boolean	dispatch(CallRunner callTask)	Add the request to the executor queue
int	getActiveHandlerCount()
int	getActiveReadHandlerCount()
int	getActiveScanHandlerCount()
int	getActiveWriteHandlerCount()
static QueueBalancer	getBalancer(String executorName, org.apache.hadoop.conf.Configuration conf, List<BlockingQueue<CallRunner>> queues)
Map<String,Long>	getCallQueueCountsSummary()
protected float	getCallQueueHandlerFactor(org.apache.hadoop.conf.Configuration conf)
Map<String,Long>	getCallQueueSizeSummary()
protected RpcHandler	getHandler(String name, double handlerFailureThreshhold, int handlerCount, BlockingQueue<CallRunner> q, AtomicInteger activeHandlerCount, AtomicInteger failedHandlerCount, Abortable abortable)	Override if providing alternate Handler implementation.
private static String	getMethodName(CallRunner callRunner)	Return the Descriptors.MethodDescriptor.getName() from callRunner or "Unknown".
String	getName()
long	getNumGeneralCallsDropped()
long	getNumLifoModeSwitches()
private Optional<Class<? extends BlockingQueue<CallRunner>>>	getPluggableQueueClass()
int	getQueueLength()	Returns the length of the pending queue
protected List<BlockingQueue<CallRunner>>	getQueues()	Returns the list of request queues
int	getReadQueueLength()
private static long	getRpcCallSize(CallRunner callRunner)	Return the RpcCall.getSize() from callRunner or 0L.
int	getScanQueueLength()
int	getWriteQueueLength()
protected void	initializeQueues(int numQueues)
static boolean	isCodelQueueType(String callQueueType)
static boolean	isDeadlineQueueType(String callQueueType)
static boolean	isFifoQueueType(String callQueueType)
static boolean	isPluggableQueueType(String callQueueType)
static boolean	isPluggableQueueWithFastPath(String callQueueType, org.apache.hadoop.conf.Configuration conf)
void	onConfigurationChange(org.apache.hadoop.conf.Configuration conf)
void	resizeQueues(org.apache.hadoop.conf.Configuration conf)	Update current soft limit for executor's call queues
void	start(int port)
protected void	startHandlers(int port)
protected void	startHandlers(String nameSuffix, int numHandlers, List<BlockingQueue<CallRunner>> callQueues, int qindex, int qsize, int port, AtomicInteger activeHandlerCount)	Start up our handlers.
void	stop()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

LOG

private static final org.slf4j.Logger LOG

DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT

protected static final int DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT

See Also:

• Constant Field Values

CALL_QUEUE_HANDLER_FACTOR_CONF_KEY

public static final String CALL_QUEUE_HANDLER_FACTOR_CONF_KEY

See Also:

• Constant Field Values

QUEUE_MAX_CALL_DELAY_CONF_KEY

public static final String QUEUE_MAX_CALL_DELAY_CONF_KEY

max delay in msec used to bound the de-prioritized requests

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_CODEL_CONF_VALUE

public static final String CALL_QUEUE_TYPE_CODEL_CONF_VALUE

The default, 'fifo', has the least friction but is dumb. If set to 'deadline', uses a priority queue and de-prioritizes long-running scans. Sorting by priority comes at a cost, reduced throughput.

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE

public static final String CALL_QUEUE_TYPE_DEADLINE_CONF_VALUE

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_FIFO_CONF_VALUE

public static final String CALL_QUEUE_TYPE_FIFO_CONF_VALUE

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE

public static final String CALL_QUEUE_TYPE_PLUGGABLE_CONF_VALUE

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_CONF_KEY

public static final String CALL_QUEUE_TYPE_CONF_KEY

See Also:

• Constant Field Values

CALL_QUEUE_TYPE_CONF_DEFAULT

public static final String CALL_QUEUE_TYPE_CONF_DEFAULT

See Also:

• Constant Field Values

CALL_QUEUE_QUEUE_BALANCER_CLASS

public static final String CALL_QUEUE_QUEUE_BALANCER_CLASS

See Also:

• Constant Field Values

CALL_QUEUE_QUEUE_BALANCER_CLASS_DEFAULT

public static final Class<?> CALL_QUEUE_QUEUE_BALANCER_CLASS_DEFAULT

CALL_QUEUE_CODEL_TARGET_DELAY

public static final String CALL_QUEUE_CODEL_TARGET_DELAY

See Also:

• Constant Field Values

CALL_QUEUE_CODEL_INTERVAL

public static final String CALL_QUEUE_CODEL_INTERVAL

See Also:

• Constant Field Values

CALL_QUEUE_CODEL_LIFO_THRESHOLD

public static final String CALL_QUEUE_CODEL_LIFO_THRESHOLD

See Also:

• Constant Field Values

CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY

public static final int CALL_QUEUE_CODEL_DEFAULT_TARGET_DELAY

See Also:

• Constant Field Values

CALL_QUEUE_CODEL_DEFAULT_INTERVAL

public static final int CALL_QUEUE_CODEL_DEFAULT_INTERVAL

See Also:

• Constant Field Values

CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD

public static final double CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD

See Also:

• Constant Field Values

PLUGGABLE_CALL_QUEUE_CLASS_NAME

public static final String PLUGGABLE_CALL_QUEUE_CLASS_NAME

See Also:

• Constant Field Values

PLUGGABLE_CALL_QUEUE_WITH_FAST_PATH_ENABLED

public static final String PLUGGABLE_CALL_QUEUE_WITH_FAST_PATH_ENABLED

See Also:

• Constant Field Values

numGeneralCallsDropped

private final LongAdder numGeneralCallsDropped

numLifoModeSwitches

private final LongAdder numLifoModeSwitches

numCallQueues

protected final int numCallQueues

queues

protected final List<BlockingQueue<CallRunner>> queues

queueClass

private final Class<? extends BlockingQueue> queueClass

queueInitArgs

private final Object[] queueInitArgs

currentQueueLimit

protected volatile int currentQueueLimit

activeHandlerCount

private final AtomicInteger activeHandlerCount

handlers

private final List<RpcHandler> handlers

handlerCount

private final int handlerCount

failedHandlerCount

private final AtomicInteger failedHandlerCount

name

private String name

conf

private final org.apache.hadoop.conf.Configuration conf

abortable

private final Abortable abortable

ONE_QUEUE

private static final QueueBalancer ONE_QUEUE

All requests go to the first queue, at index 0

Constructor Details

RpcExecutor

public RpcExecutor(String name,
 int handlerCount,
 int maxQueueLength,
 PriorityFunction priority,
 org.apache.hadoop.conf.Configuration conf,
 Abortable abortable)

RpcExecutor

public RpcExecutor(String name,
 int handlerCount,
 String callQueueType,
 int maxQueueLength,
 PriorityFunction priority,
 org.apache.hadoop.conf.Configuration conf,
 Abortable abortable)

Method Details

computeNumCallQueues

protected int computeNumCallQueues(int handlerCount,
 float callQueuesHandlersFactor)

getMethodName

private static String getMethodName(CallRunner callRunner)

Return the Descriptors.MethodDescriptor.getName() from callRunner or "Unknown".

getRpcCallSize

private static long getRpcCallSize(CallRunner callRunner)

Return the RpcCall.getSize() from callRunner or 0L.

getCallQueueCountsSummary

public Map<String,Long> getCallQueueCountsSummary()

getCallQueueSizeSummary

public Map<String,Long> getCallQueueSizeSummary()

initializeQueues

protected void initializeQueues(int numQueues)

start

public void start(int port)

stop

public void stop()

dispatch

public abstract boolean dispatch(CallRunner callTask)

Add the request to the executor queue

getQueues

protected List<BlockingQueue<CallRunner>> getQueues()

Returns the list of request queues

startHandlers

protected void startHandlers(int port)

getHandler

protected RpcHandler getHandler(String name,
 double handlerFailureThreshhold,
 int handlerCount,
 BlockingQueue<CallRunner> q,
 AtomicInteger activeHandlerCount,
 AtomicInteger failedHandlerCount,
 Abortable abortable)

Override if providing alternate Handler implementation.

startHandlers

protected void startHandlers(String nameSuffix,
 int numHandlers,
 List<BlockingQueue<CallRunner>> callQueues,
 int qindex,
 int qsize,
 int port,
 AtomicInteger activeHandlerCount)

Start up our handlers.

getBalancer

public static QueueBalancer getBalancer(String executorName,
 org.apache.hadoop.conf.Configuration conf,
 List<BlockingQueue<CallRunner>> queues)

isDeadlineQueueType

public static boolean isDeadlineQueueType(String callQueueType)

isCodelQueueType

public static boolean isCodelQueueType(String callQueueType)

isFifoQueueType

public static boolean isFifoQueueType(String callQueueType)

isPluggableQueueType

public static boolean isPluggableQueueType(String callQueueType)

isPluggableQueueWithFastPath

public static boolean isPluggableQueueWithFastPath(String callQueueType,
 org.apache.hadoop.conf.Configuration conf)

getPluggableQueueClass

private Optional<Class<? extends BlockingQueue<CallRunner>>> getPluggableQueueClass()

getNumGeneralCallsDropped

public long getNumGeneralCallsDropped()

getNumLifoModeSwitches

public long getNumLifoModeSwitches()

getActiveHandlerCount

public int getActiveHandlerCount()

getActiveWriteHandlerCount

public int getActiveWriteHandlerCount()

getActiveReadHandlerCount

public int getActiveReadHandlerCount()

getActiveScanHandlerCount

public int getActiveScanHandlerCount()

getQueueLength

public int getQueueLength()

Returns the length of the pending queue

getReadQueueLength

public int getReadQueueLength()

getScanQueueLength

public int getScanQueueLength()

getWriteQueueLength

public int getWriteQueueLength()

getName

public String getName()

resizeQueues

public void resizeQueues(org.apache.hadoop.conf.Configuration conf)

Update current soft limit for executor's call queues

Parameters:

conf - updated configuration

onConfigurationChange

public void onConfigurationChange(org.apache.hadoop.conf.Configuration conf)

getCallQueueHandlerFactor

protected float getCallQueueHandlerFactor(org.apache.hadoop.conf.Configuration conf)