This is an automated email from the ASF dual-hosted git repository.
apurtell pushed a commit to branch branch-2.5
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2.5 by this push:
new 7ebb557178d HBASE-29912 Codel lifoThreshold should be applied on soft
queue limit instead of hard limit (#8346)
7ebb557178d is described below
commit 7ebb557178d4eab7001906063c1a476a991c45cb
Author: Umesh <[email protected]>
AuthorDate: Fri Jun 19 04:33:29 2026 +0530
HBASE-29912 Codel lifoThreshold should be applied on soft queue limit
instead of hard limit (#8346)
Signed-off-by: Andrew Purtell <[email protected]>
---
.../hbase/ipc/AdaptiveLifoCoDelCallQueue.java | 17 +-
.../org/apache/hadoop/hbase/ipc/RpcExecutor.java | 43 ++--
.../hadoop/hbase/ipc/TestSimpleRpcScheduler.java | 252 +++++++++++++++++----
3 files changed, 246 insertions(+), 66 deletions(-)
diff --git
a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/AdaptiveLifoCoDelCallQueue.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/AdaptiveLifoCoDelCallQueue.java
index dee517c2e0a..89fde2e1c14 100644
---
a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/AdaptiveLifoCoDelCallQueue.java
+++
b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/AdaptiveLifoCoDelCallQueue.java
@@ -44,7 +44,7 @@ public class AdaptiveLifoCoDelCallQueue implements
BlockingQueue<CallRunner> {
private LinkedBlockingDeque<CallRunner> queue;
// so we can calculate actual threshold to switch to LIFO under load
- private int maxCapacity;
+ private volatile int softLimit;
// metrics (shared across all queues)
private LongAdder numGeneralCallsDropped;
@@ -71,14 +71,15 @@ public class AdaptiveLifoCoDelCallQueue implements
BlockingQueue<CallRunner> {
private AtomicBoolean isOverloaded = new AtomicBoolean(false);
public AdaptiveLifoCoDelCallQueue(int capacity, int targetDelay, int
interval,
- double lifoThreshold, LongAdder numGeneralCallsDropped, LongAdder
numLifoModeSwitches) {
- this.maxCapacity = capacity;
+ double lifoThreshold, LongAdder numGeneralCallsDropped, LongAdder
numLifoModeSwitches,
+ int currentQueueLimit) {
this.queue = new LinkedBlockingDeque<>(capacity);
this.codelTargetDelay = targetDelay;
this.codelInterval = interval;
this.lifoThreshold = lifoThreshold;
this.numGeneralCallsDropped = numGeneralCallsDropped;
this.numLifoModeSwitches = numLifoModeSwitches;
+ this.softLimit = currentQueueLimit;
}
/**
@@ -86,12 +87,14 @@ public class AdaptiveLifoCoDelCallQueue implements
BlockingQueue<CallRunner> {
* @param newCodelTargetDelay new CoDel target delay
* @param newCodelInterval new CoDel interval
* @param newLifoThreshold new Adaptive Lifo threshold
+ * @param currentQueueLimit new soft limit of queue
*/
- public void updateTunables(int newCodelTargetDelay, int newCodelInterval,
- double newLifoThreshold) {
+ public void updateTunables(int newCodelTargetDelay, int newCodelInterval,
double newLifoThreshold,
+ int currentQueueLimit) {
this.codelTargetDelay = newCodelTargetDelay;
this.codelInterval = newCodelInterval;
this.lifoThreshold = newLifoThreshold;
+ this.softLimit = currentQueueLimit;
}
/**
@@ -104,7 +107,7 @@ public class AdaptiveLifoCoDelCallQueue implements
BlockingQueue<CallRunner> {
public CallRunner take() throws InterruptedException {
CallRunner cr;
while (true) {
- if (((double) queue.size() / this.maxCapacity) > lifoThreshold) {
+ if (((double) queue.size() / this.softLimit) > lifoThreshold) {
numLifoModeSwitches.increment();
cr = queue.takeLast();
} else {
@@ -124,7 +127,7 @@ public class AdaptiveLifoCoDelCallQueue implements
BlockingQueue<CallRunner> {
CallRunner cr;
boolean switched = false;
while (true) {
- if (((double) queue.size() / this.maxCapacity) > lifoThreshold) {
+ if (((double) queue.size() / this.softLimit) > lifoThreshold) {
// Only count once per switch.
if (!switched) {
switched = true;
diff --git
a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java
index c4abb048f75..ff99f253b8e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java
@@ -107,7 +107,11 @@ public abstract class RpcExecutor {
private final Class<? extends BlockingQueue> queueClass;
private final Object[] queueInitArgs;
+ // this is soft limit of the queue, not size/capacity.
protected volatile int currentQueueLimit;
+ // While initializing we will use hard limit as the capacity of queue, it
will let us dynamically
+ // change the queue limit
+ protected final int queueHardLimit;
private final AtomicInteger activeHandlerCount = new AtomicInteger(0);
private final List<RpcHandler> handlers;
@@ -161,11 +165,13 @@ public abstract class RpcExecutor {
int handlerCountPerQueue = this.handlerCount / this.numCallQueues;
maxQueueLength = handlerCountPerQueue *
RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER;
}
+ currentQueueLimit = maxQueueLength;
+ queueHardLimit = Math.max(maxQueueLength,
DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT);
if (isDeadlineQueueType(callQueueType)) {
this.name += ".Deadline";
this.queueInitArgs =
- new Object[] { maxQueueLength, new CallPriorityComparator(conf,
priority) };
+ new Object[] { queueHardLimit, new CallPriorityComparator(conf,
priority) };
this.queueClass = BoundedPriorityBlockingQueue.class;
} else if (isCodelQueueType(callQueueType)) {
this.name += ".Codel";
@@ -174,8 +180,8 @@ public abstract class RpcExecutor {
int codelInterval = conf.getInt(CALL_QUEUE_CODEL_INTERVAL,
CALL_QUEUE_CODEL_DEFAULT_INTERVAL);
double codelLifoThreshold =
conf.getDouble(CALL_QUEUE_CODEL_LIFO_THRESHOLD,
CALL_QUEUE_CODEL_DEFAULT_LIFO_THRESHOLD);
- this.queueInitArgs = new Object[] { maxQueueLength, codelTargetDelay,
codelInterval,
- codelLifoThreshold, numGeneralCallsDropped, numLifoModeSwitches };
+ this.queueInitArgs = new Object[] { queueHardLimit, codelTargetDelay,
codelInterval,
+ codelLifoThreshold, numGeneralCallsDropped, numLifoModeSwitches,
currentQueueLimit };
this.queueClass = AdaptiveLifoCoDelCallQueue.class;
} else if (isPluggableQueueType(callQueueType)) {
Optional<Class<? extends BlockingQueue<CallRunner>>> pluggableQueueClass
=
@@ -185,12 +191,12 @@ public abstract class RpcExecutor {
throw new PluggableRpcQueueNotFound(
"Pluggable call queue failed to load and selected call" + " queue
type required");
} else {
- this.queueInitArgs = new Object[] { maxQueueLength, priority, conf };
+ this.queueInitArgs = new Object[] { queueHardLimit, priority, conf };
this.queueClass = pluggableQueueClass.get();
}
} else {
this.name += ".Fifo";
- this.queueInitArgs = new Object[] { maxQueueLength };
+ this.queueInitArgs = new Object[] { queueHardLimit };
this.queueClass = LinkedBlockingQueue.class;
}
@@ -231,20 +237,7 @@ public abstract class RpcExecutor {
.collect(Collectors.groupingBy(Pair::getFirst,
Collectors.summingLong(Pair::getSecond)));
}
- // This method can only be called ONCE per executor instance.
- // Before calling: queueInitArgs[0] contains the soft limit (desired queue
capacity)
- // After calling: queueInitArgs[0] is set to hard limit and
currentQueueLimit stores the original
- // soft limit.
- // Multiple calls would incorrectly use the hard limit as the soft limit.
- // As all the queues has same initArgs and queueClass, there should be no
need to call this again.
protected void initializeQueues(final int numQueues) {
- if (!queues.isEmpty()) {
- throw new RuntimeException("Queues are already initialized");
- }
- if (queueInitArgs.length > 0) {
- currentQueueLimit = (int) queueInitArgs[0];
- queueInitArgs[0] = Math.max((int) queueInitArgs[0],
DEFAULT_CALL_QUEUE_SIZE_HARD_LIMIT);
- }
for (int i = 0; i < numQueues; ++i) {
queues.add(ReflectionUtils.newInstance(queueClass, queueInitArgs));
}
@@ -464,7 +457,15 @@ public abstract class RpcExecutor {
}
}
final int queueLimit = currentQueueLimit;
- currentQueueLimit = conf.getInt(configKey, queueLimit);
+ int newQueueLimit = conf.getInt(configKey, queueLimit);
+ if (newQueueLimit > queueHardLimit) {
+ LOG.warn(
+ "Requested soft limit {} exceeds queue hard limit/capacity {}. "
+ + "A region server restart is required to grow the underlying
queue.",
+ newQueueLimit, queueHardLimit);
+ newQueueLimit = currentQueueLimit;
+ }
+ currentQueueLimit = newQueueLimit;
}
public void onConfigurationChange(Configuration conf) {
@@ -477,8 +478,10 @@ public abstract class RpcExecutor {
for (BlockingQueue<CallRunner> queue : queues) {
if (queue instanceof AdaptiveLifoCoDelCallQueue) {
+ // current queue Limit for executor is already updated as part of
resizeQueues, we need to
+ // let codel queue also make aware of it
((AdaptiveLifoCoDelCallQueue) queue).updateTunables(codelTargetDelay,
codelInterval,
- codelLifoThreshold);
+ codelLifoThreshold, currentQueueLimit);
} else if (queue instanceof ConfigurationObserver) {
((ConfigurationObserver) queue).onConfigurationChange(conf);
}
diff --git
a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java
index 74d4eb2b864..767351497b5 100644
---
a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java
+++
b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hbase.ipc;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -34,13 +35,14 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Abortable;
import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -87,6 +89,7 @@ public class TestSimpleRpcScheduler {
};
private Configuration conf;
private String testMethodName;
+ private final AtomicInteger requestProcessed = new AtomicInteger(0);
@BeforeEach
public void setUp(TestInfo testInfo) {
@@ -226,6 +229,7 @@ public class TestSimpleRpcScheduler {
RequestHeader header =
RequestHeader.newBuilder().setPriority(priority).build();
when(task.getRpcCall()).thenReturn(call);
when(call.getHeader()).thenReturn(header);
+
when(call.getReceiveTime()).thenReturn(EnvironmentEdgeManager.currentTime());
return task;
}
@@ -567,66 +571,84 @@ public class TestSimpleRpcScheduler {
}
}
- private static final class CoDelEnvironmentEdge implements EnvironmentEdge {
+ private final class CoDelEnvironmentEdge implements EnvironmentEdge {
- private final BlockingQueue<Long> timeQ = new LinkedBlockingQueue<>();
+ private long perRequestOffset;
+ private long gapInRequests;
+ private long testStartTime = 0;
- private long offset;
+ private int requestCount = 0;
- private final Set<String> threadNamePrefixs = new HashSet<>();
+ private final Set<String> threadNamePrefixes = new HashSet<>();
+ private final String testThread = Thread.currentThread().getName();
@Override
public long currentTime() {
- for (String threadNamePrefix : threadNamePrefixs) {
- String threadName = Thread.currentThread().getName();
+ String threadName = Thread.currentThread().getName();
+ if (threadName.equals(testThread)) {
+ // test thread will create a request and add that to scheduler
+ // Replicating a constant rate of request arrival.
+ long requestStartTime = testStartTime + requestCount * gapInRequests;
+ requestCount++;
+ return requestStartTime;
+ }
+ // handler thread will pick the request from queue, this time will
depend on processing time
+ // of handler
+ for (String threadNamePrefix : threadNamePrefixes) {
if (threadName.startsWith(threadNamePrefix)) {
- if (timeQ != null) {
- Long qTime = timeQ.poll();
- if (qTime != null) {
- return qTime.longValue() + offset;
- }
+ long requestPickTime;
+ if (gapInRequests >= perRequestOffset) {
+ // it means handler can complete the processing of last request
and will be free when it
+ // will come to serve next request. We don't need it in fast path
but just in case for
+ // other tests
+ requestPickTime = testStartTime + requestProcessed.get() *
gapInRequests;
+ } else {
+ // this means handler will still be busy processing the last
requests when new request
+ // will come in queue
+ requestPickTime = testStartTime + requestProcessed.get() *
perRequestOffset;
}
+ return requestPickTime;
}
}
return System.currentTimeMillis();
}
+
+ public void startTestFor(int arrivalRatePerSecond, long
requestProcessingTime) {
+ this.perRequestOffset = requestProcessingTime;
+ this.gapInRequests = 1000L / arrivalRatePerSecond;
+ this.requestCount = 0;
+ requestProcessed.set(0);
+ this.testStartTime = System.currentTimeMillis();
+ }
}
- // FIX. I don't get this test (St.Ack). When I time this test, the minDelay
is > 2 * codel delay
- // from the get go. So we are always overloaded. The test below would seem
to complete the
- // queuing of all the CallRunners inside the codel check interval. I don't
think we are skipping
- // codel checking. Second, I think this test has been broken since
HBASE-16089 Add on FastPath for
- // CoDel went in. The thread name we were looking for was the name BEFORE we
updated: i.e.
- // "RpcServer.CodelBQ.default.handler". But same patch changed the name of
the codel fastpath
- // thread to: new FastPathBalancedQueueRpcExecutor("CodelFPBQ.default",
handlerCount,
- // numCallQueues... Codel is hard to test. This test is going to be flakey
given it all
- // timer-based. Disabling for now till chat with authors.
+ // This test is fixing the processing time and arrival rate of the request
and checking if CoDel
+ // can utilize the maximum capacity of system
@Test
public void testCoDelScheduling() throws Exception {
CoDelEnvironmentEdge envEdge = new CoDelEnvironmentEdge();
- envEdge.threadNamePrefixs.add("RpcServer.default.FPBQ.Codel.handler");
+ envEdge.threadNamePrefixes.add("RpcServer.default.FPBQ.Codel.handler");
Configuration schedConf = HBaseConfiguration.create();
schedConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 250);
schedConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
+ schedConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 5);
PriorityFunction priority = mock(PriorityFunction.class);
when(priority.getPriority(any(), any(),
any())).thenReturn(HConstants.NORMAL_QOS);
SimpleRpcScheduler scheduler =
new SimpleRpcScheduler(schedConf, 1, 1, 1, priority,
HConstants.QOS_THRESHOLD);
+
try {
- // Loading mocked call runner can take a good amount of time the first
time through
- // (haven't looked why). Load it for first time here outside of the
timed loop.
- getMockedCallRunner(EnvironmentEdgeManager.currentTime(), 2);
scheduler.start();
EnvironmentEdgeManager.injectEdge(envEdge);
- envEdge.offset = 5;
- // Calls faster than min delay
- // LOG.info("Start");
+
+ // incoming traffic < capacity
+ envEdge.startTestFor(20, 40);
for (int i = 0; i < 100; i++) {
long time = EnvironmentEdgeManager.currentTime();
- envEdge.timeQ.put(time);
CallRunner cr = getMockedCallRunner(time, 2);
scheduler.dispatch(cr);
+ Thread.sleep(50);
}
// LOG.info("Loop done");
// make sure fast calls are handled
@@ -635,13 +657,15 @@ public class TestSimpleRpcScheduler {
assertEquals(0, scheduler.getNumGeneralCallsDropped(),
"None of these calls should have been discarded");
- envEdge.offset = 151;
- // calls slower than min delay, but not individually slow enough to be
dropped
+ // incoming traffic = capacity
+ envEdge.startTestFor(20, 50);
for (int i = 0; i < 20; i++) {
long time = EnvironmentEdgeManager.currentTime();
- envEdge.timeQ.put(time);
CallRunner cr = getMockedCallRunner(time, 2);
scheduler.dispatch(cr);
+ // We have mocked the arrival time and the time request get picked
from queue but we need to
+ // also make sure that queue fill at the arrival rate
+ Thread.sleep(50);
}
// make sure somewhat slow calls are handled
@@ -650,21 +674,28 @@ public class TestSimpleRpcScheduler {
assertEquals(0, scheduler.getNumGeneralCallsDropped(),
"None of these calls should have been discarded");
- envEdge.offset = 2000;
+ // incoming traffic > capacity
+ envEdge.startTestFor(20, 60);
// now slow calls and the ones to be dropped
for (int i = 0; i < 60; i++) {
long time = EnvironmentEdgeManager.currentTime();
- envEdge.timeQ.put(time);
CallRunner cr = getMockedCallRunner(time, 100);
scheduler.dispatch(cr);
+ Thread.sleep(50);
}
// make sure somewhat slow calls are handled
waitUntilQueueEmpty(scheduler);
Thread.sleep(100);
- assertTrue(scheduler.getNumGeneralCallsDropped() > 12,
- "There should have been at least 12 calls dropped however there were "
+ // 60 calls will take 3 second with 20 rps. And in 3 second with 60
processing time we can
+ // only serve 50 request
+ assertTrue(requestProcessed.get() >= 48,
+ "Number of processed requests should be greater then 90% of capacity"
+ + requestProcessed.get());
+ assertTrue(scheduler.getNumGeneralCallsDropped() >= 9,
+ "There should have been at least 9 calls dropped however there were "
+ scheduler.getNumGeneralCallsDropped());
+
} finally {
scheduler.stop();
}
@@ -779,9 +810,8 @@ public class TestSimpleRpcScheduler {
return;
}
try {
- LOG.warn("Sleeping for " + sleepTime);
Thread.sleep(sleepTime);
- LOG.warn("Done Sleeping for " + sleepTime);
+ requestProcessed.incrementAndGet();
} catch (InterruptedException e) {
}
}
@@ -796,4 +826,148 @@ public class TestSimpleRpcScheduler {
}
};
}
+
+ /**
+ * Test LIFO switching behavior through actual RPC calls. This test verifies
that when the queue
+ * fills beyond the LIFO threshold, newer calls are processed before older
calls (LIFO mode).
+ */
+ @Test
+ public void testCoDelLifoWithRpcCalls() throws Exception {
+ Configuration testConf = HBaseConfiguration.create();
+ testConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
+ RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
+ int maxCallQueueLength = 50;
+ double coDelLifoThreshold = 0.8;
+ testConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH,
maxCallQueueLength);
+ testConf.setDouble(RpcExecutor.CALL_QUEUE_CODEL_LIFO_THRESHOLD,
coDelLifoThreshold);
+ testConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 100);
+ testConf.setInt(RpcExecutor.CALL_QUEUE_CODEL_INTERVAL, 100);
+ testConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 1); // Single
handler to control
+ // processing
+
+ PriorityFunction priority = mock(PriorityFunction.class);
+ when(priority.getPriority(any(), any(),
any())).thenReturn(HConstants.NORMAL_QOS);
+ SimpleRpcScheduler scheduler =
+ new SimpleRpcScheduler(testConf, 1, 0, 0, priority,
HConstants.QOS_THRESHOLD);
+
+ try {
+ scheduler.init(CONTEXT);
+ scheduler.start();
+
+ // Track completion order
+ final List<Integer> completedCalls = Collections.synchronizedList(new
ArrayList<>());
+
+ // Dispatch many slow calls rapidly to fill the queue beyond 80%
threshold
+ // With queue limit of 50, we need > 40 calls to cross 80%
+ int numCalls = 48;
+ for (int i = 0; i < numCalls; i++) {
+ final int callId = i;
+ CallRunner call = createMockTask(HConstants.NORMAL_QOS);
+ call.setStatus(new MonitoredRPCHandlerImpl("test"));
+ doAnswer(invocation -> {
+ completedCalls.add(callId);
+ Thread.sleep(100); // Slow processing to allow queue to build up
+ return null;
+ }).when(call).run();
+ scheduler.dispatch(call);
+ // No delay between dispatches - rapidly fill the queue
+ }
+
+ // Wait for some calls to complete
+ await().atMost(2, TimeUnit.SECONDS).until(() -> completedCalls.size() >=
3);
+
+ // Check that we had LIFO switches
+ long lifoSwitches = scheduler.getNumLifoModeSwitches();
+ assertTrue(lifoSwitches > 0,
+ "Should have switched to LIFO mode at least once, but got: " +
lifoSwitches);
+
+ // Verify LIFO behavior: Among first completed calls, we should see
higher call IDs
+ // (indicating later dispatched calls completed first)
+ int maxCallIdCompleted = -1;
+ for (int i = 0; i < completedCalls.size(); i++) {
+ maxCallIdCompleted = Math.max(maxCallIdCompleted,
completedCalls.get(i));
+ }
+ // At least one of the early completed calls should have a high ID (>20)
+ // indicating LIFO processing
+ assertTrue(maxCallIdCompleted > maxCallQueueLength * coDelLifoThreshold,
+ "Expected LIFO behavior: early completed calls should include call
arrived after threshold "
+ + "maxCallIdCompleted: " + maxCallIdCompleted);
+
+ } finally {
+ scheduler.stop();
+ }
+ }
+
+ /**
+ * Test that CoDel queue returns to FIFO mode after draining below threshold.
+ */
+ @Test
+ public void testCoDelQueueDrainAndFifoReturn() throws Exception {
+ Configuration testConf = HBaseConfiguration.create();
+ testConf.set(RpcExecutor.CALL_QUEUE_TYPE_CONF_KEY,
+ RpcExecutor.CALL_QUEUE_TYPE_CODEL_CONF_VALUE);
+ testConf.setLong(RpcExecutor.CALL_QUEUE_CODEL_TARGET_DELAY, 100);
+ testConf.setInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, 50);
+ testConf.setDouble(RpcExecutor.CALL_QUEUE_CODEL_LIFO_THRESHOLD, 0.8);
+ testConf.setInt(HConstants.REGION_SERVER_HANDLER_COUNT, 2);
+
+ PriorityFunction priority = mock(PriorityFunction.class);
+ when(priority.getPriority(any(), any(),
any())).thenReturn(HConstants.NORMAL_QOS);
+ SimpleRpcScheduler scheduler =
+ new SimpleRpcScheduler(testConf, 2, 0, 0, priority,
HConstants.QOS_THRESHOLD);
+
+ try {
+ scheduler.init(CONTEXT);
+ scheduler.start();
+
+ final List<Integer> completedCalls = Collections.synchronizedList(new
ArrayList<>());
+
+ // Fill queue rapidly to trigger LIFO (>40 calls for 80% of 50)
+ for (int i = 0; i < 48; i++) {
+ final int callId = i;
+ CallRunner call = createMockTask(HConstants.NORMAL_QOS);
+ call.setStatus(new MonitoredRPCHandlerImpl("test"));
+ doAnswer(invocation -> {
+ completedCalls.add(callId);
+ Thread.sleep(80);
+ return null;
+ }).when(call).run();
+ scheduler.dispatch(call);
+ }
+
+ // Wait for calls to complete
+ await().atMost(1, TimeUnit.SECONDS).until(() -> completedCalls.size() >=
3);
+ assertTrue(scheduler.getNumLifoModeSwitches() > 0, "Should have entered
LIFO mode");
+
+ await().atMost(2, TimeUnit.SECONDS).until(
+ () -> scheduler.getGeneralQueueLength() == 0 &&
scheduler.getActiveRpcHandlerCount() == 0);
+
+ long pastNumLifoModeSwitches = scheduler.getNumLifoModeSwitches();
+ // Send new calls - should process in FIFO order
+ completedCalls.clear();
+ for (int i = 100; i < 105; i++) {
+ final int callId = i;
+ CallRunner call = createMockTask(HConstants.NORMAL_QOS);
+ call.setStatus(new MonitoredRPCHandlerImpl("test"));
+ doAnswer(invocation -> {
+ completedCalls.add(callId);
+ Thread.sleep(50);
+ return null;
+ }).when(call).run();
+ scheduler.dispatch(call);
+ }
+
+ // Wait for these calls to complete
+ await().atMost(2, TimeUnit.SECONDS).until(() -> completedCalls.size() >=
2);
+
+ long newLifoSwitch = scheduler.getNumLifoModeSwitches() -
pastNumLifoModeSwitches;
+ // Allow at most 1 violation due to concurrent execution by 2 handlers
+ assertEquals(0, newLifoSwitch,
+ "Queue should not switch to LIFO last 5 calls but number of LIFO
switch are : "
+ + newLifoSwitch);
+ } finally {
+ scheduler.stop();
+ }
+ }
+
}
