Skip to content

HBASE-28359 Improve quota RateLimiter synchronization #5683

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbeaudreault merged 4 commits into from
Mar 7, 2024
Merged

HBASE-28359 Improve quota RateLimiter synchronization #5683

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
334567e
fix RateLimiter canExecute and waitInterval synchronization
Feb 15, 2024
11cd56c
cleanup
Feb 15, 2024
abe1a32
fix underrepresented read availability
Feb 16, 2024
6d4abbf
PR feedback: new method, no private sync, naming
Feb 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/DefaultOperationQuota.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,7 +131,8 @@ public void close() {

@Override
public long getReadAvailable() {
return readAvailable;
// at this point we've grabbed some quota, so we should use at least that
return Math.max(readAvailable, readConsumed);

@rmdmattingly rmdmattingly Feb 16, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Without this change we've experienced erroneous MultiActionResultTooLarge when riding close to quotas. As readAvailable can be negative, this can even affect otherwise irrelevant operation types. For example, riding close to a read quota and entering negative read availability would cause multiPuts, which do no reading, to throw MultiActionResultTooLarge. The logic for throwing a MultiActionResultTooLarge can be found here. Worse, this exception is retried immediately, so this can snowball into a DDOS of your RPC layer.

@Apache9 Apache9 Feb 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we move this Math.max to caller? Theoretically readAvailable could be negative, the caller should decide how to deal with this.

@rmdmattingly rmdmattingly Feb 18, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I wonder whether it's worth having a separate method, OperationQuota#getMaxResultSize, which encapsulates this logic and is not hiding business logic in what appears to be a simple getter

@bbeaudreault bbeaudreault Feb 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should keep this here. I have three reasons:

  1. There are only 2 callers to this method, and both would need the Math.max() added
  2. We don't currently expose readConsumed
  3. Logically, I think it makes sense to consider readAvailable in this way for all cases. I could even see it being Math.max(readAvailable + readConsumed, readConsumed). This is because of the nuanced quota logic in checkQuota. Details on this one below:

In checkQuota, it iterates all limiters twice. Once to check for available (or throw throttle exception). Once it gets through all limiters, it loops them again to grab the quota. This loop can over consume the quota. At this point we've grabbed the quota and there's no easy way to return it. So we might as well consider this as "available" for us.

The quotas are complicated because there can be many limiters and checkQuota is not synchronized. It'd be more accurate to synchronize the entire checkQuota, but I think it's better for performance to let it be.

@Apache9 Apache9 Feb 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Then better rename the method, or at least add more comments to describe the logic.

@rmdmattingly rmdmattingly Feb 18, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah sorry Bryan, I didn't see your comment until I pushed a change. I think the only difference though would be to remove the readAvailable and readConsumed getters, and move the getMaxResultSize logic out of the interface. Do you think that's worth doing, or do you like the changeset in its current form?

@bbeaudreault bbeaudreault Feb 18, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Personally I prefer the original name, which has the javadoc:

/** Returns the number of bytes available to read to avoid exceeding the quota *

To me "read available" makes sense to be "how much did we grab, plus how much is left". But I can see that might be subjective. So I don't have a strong opinion.

@rmdmattingly rmdmattingly Feb 18, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see both sides. I think it's ok for us to be more specific with the naming here because, like you said, the current usage scope is small enough for us to be specific

}

@Override
Expand Down
19 changes: 16 additions & 3 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/RateLimiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,18 +135,31 @@ protected synchronized long getTimeUnitInMillis() {

/**
* Is there at least one resource available to allow execution?
* @return true if there is at least one resource available, otherwise false
* @return the waitInterval to backoff, or 0 if execution is allowed
*/
public boolean canExecute() {
public long canExecute() {
Comment thread
rmdmattingly marked this conversation as resolved.
Outdated
return canExecute(1);
}

/**
* Are there enough available resources to allow execution?
* @param amount the number of required resources, a non-negative number
* @return the waitInterval to backoff, or 0 if execution is allowed
*/
public synchronized long canExecute(final long amount) {
Comment thread
rmdmattingly marked this conversation as resolved.
Outdated
assert amount >= 0;
if (!isAvailable(amount)) {
return waitInterval(amount);
}
return 0;
}

/**
* Are there enough available resources to allow execution?
* @param amount the number of required resources, a non-negative number
* @return true if there are enough available resources, otherwise false
*/
public synchronized boolean canExecute(final long amount) {
private synchronized boolean isAvailable(final long amount) {
Comment thread
rmdmattingly marked this conversation as resolved.
Outdated
if (isBypass()) {
return true;
}
Expand Down
52 changes: 28 additions & 24 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/TimeBasedLimiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,43 +141,47 @@ private static void setFromTimedQuota(final RateLimiter limiter, final TimedQuot
public void checkQuota(long writeReqs, long estimateWriteSize, long readReqs,
long estimateReadSize, long estimateWriteCapacityUnit, long estimateReadCapacityUnit)
throws RpcThrottlingException {
if (!reqsLimiter.canExecute(writeReqs + readReqs)) {
RpcThrottlingException.throwNumRequestsExceeded(reqsLimiter.waitInterval());
long waitInterval = reqsLimiter.canExecute(writeReqs + readReqs);
if (waitInterval > 0) {
RpcThrottlingException.throwNumRequestsExceeded(waitInterval);
}
if (!reqSizeLimiter.canExecute(estimateWriteSize + estimateReadSize)) {
RpcThrottlingException.throwRequestSizeExceeded(
reqSizeLimiter.waitInterval(estimateWriteSize + estimateReadSize));
waitInterval = reqSizeLimiter.canExecute(estimateWriteSize + estimateReadSize);
if (waitInterval > 0) {
RpcThrottlingException.throwRequestSizeExceeded(waitInterval);
}
if (!reqCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit + estimateReadCapacityUnit)) {
RpcThrottlingException.throwRequestCapacityUnitExceeded(
reqCapacityUnitLimiter.waitInterval(estimateWriteCapacityUnit + estimateReadCapacityUnit));
waitInterval =
reqCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit + estimateReadCapacityUnit);
if (waitInterval > 0) {
RpcThrottlingException.throwRequestCapacityUnitExceeded(waitInterval);
}

if (estimateWriteSize > 0) {
if (!writeReqsLimiter.canExecute(writeReqs)) {
RpcThrottlingException.throwNumWriteRequestsExceeded(writeReqsLimiter.waitInterval());
waitInterval = writeReqsLimiter.canExecute(writeReqs);
if (waitInterval > 0) {
RpcThrottlingException.throwNumWriteRequestsExceeded(waitInterval);
}
if (!writeSizeLimiter.canExecute(estimateWriteSize)) {
RpcThrottlingException
.throwWriteSizeExceeded(writeSizeLimiter.waitInterval(estimateWriteSize));
waitInterval = writeSizeLimiter.canExecute(estimateWriteSize);
if (waitInterval > 0) {
RpcThrottlingException.throwWriteSizeExceeded(waitInterval);
}
if (!writeCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit)) {
RpcThrottlingException.throwWriteCapacityUnitExceeded(
writeCapacityUnitLimiter.waitInterval(estimateWriteCapacityUnit));
waitInterval = writeCapacityUnitLimiter.canExecute(estimateWriteCapacityUnit);
if (waitInterval > 0) {
RpcThrottlingException.throwWriteCapacityUnitExceeded(waitInterval);
}
}

if (estimateReadSize > 0) {
if (!readReqsLimiter.canExecute(readReqs)) {
RpcThrottlingException.throwNumReadRequestsExceeded(readReqsLimiter.waitInterval());
waitInterval = readReqsLimiter.canExecute(readReqs);
if (waitInterval > 0) {
RpcThrottlingException.throwNumReadRequestsExceeded(waitInterval);
}
if (!readSizeLimiter.canExecute(estimateReadSize)) {
RpcThrottlingException
.throwReadSizeExceeded(readSizeLimiter.waitInterval(estimateReadSize));
waitInterval = readSizeLimiter.canExecute(estimateReadSize);
if (waitInterval > 0) {
RpcThrottlingException.throwReadSizeExceeded(waitInterval);
}
if (!readCapacityUnitLimiter.canExecute(estimateReadCapacityUnit)) {
RpcThrottlingException.throwReadCapacityUnitExceeded(
readCapacityUnitLimiter.waitInterval(estimateReadCapacityUnit));
waitInterval = readCapacityUnitLimiter.canExecute(estimateReadCapacityUnit);
if (waitInterval > 0) {
RpcThrottlingException.throwReadCapacityUnitExceeded(waitInterval);
}
}
}
Expand Down
22 changes: 11 additions & 11 deletions hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestBlockBytesScannedQuota.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,9 +103,9 @@ public void testBBSGet() throws Exception {
doPuts(10_000, FAMILY, QUALIFIER, table);
TEST_UTIL.flush(TABLE_NAME);

// Add ~10 block/min limit
// Add ~10 block/sec limit
admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.READ_SIZE,
Math.round(10.1 * blockSize), TimeUnit.MINUTES));
Math.round(10.1 * blockSize), TimeUnit.SECONDS));
triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);

// should execute at max 10 requests
Expand All @@ -132,10 +132,10 @@ public void testBBSScan() throws Exception {
doPuts(10_000, FAMILY, QUALIFIER, table);
TEST_UTIL.flush(TABLE_NAME);

// Add 1 block/min limit.
// Add 1 block/sec limit.
// This should only allow 1 scan per minute, because we estimate 1 block per scan
admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE, blockSize,
TimeUnit.MINUTES));
TimeUnit.SECONDS));
triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);
waitMinuteQuota();

Expand All @@ -148,9 +148,9 @@ public void testBBSScan() throws Exception {
testTraffic(() -> doScans(100, table), 100, 0);
testTraffic(() -> doScans(100, table), 100, 0);

// Add ~3 block/min limit. This should support >1 scans
// Add ~3 block/sec limit. This should support >1 scans
admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE,
Math.round(3.1 * blockSize), TimeUnit.MINUTES));
Math.round(3.1 * blockSize), TimeUnit.SECONDS));
triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);

// should execute some requests, but not all
Expand All @@ -174,10 +174,10 @@ public void testBBSMultiGet() throws Exception {
doPuts(rowCount, FAMILY, QUALIFIER, table);
TEST_UTIL.flush(TABLE_NAME);

// Add 1 block/min limit.
// Add 1 block/sec limit.
// This should only allow 1 multiget per minute, because we estimate 1 block per multiget
admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE, blockSize,
TimeUnit.MINUTES));
TimeUnit.SECONDS));
triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);
waitMinuteQuota();

Expand All @@ -190,9 +190,9 @@ public void testBBSMultiGet() throws Exception {
testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);
testTraffic(() -> doMultiGets(100, 10, rowCount, FAMILY, QUALIFIER, table), 100, 0);

// Add ~100 block/min limit
// Add ~100 block/sec limit
admin.setQuota(QuotaSettingsFactory.throttleUser(userName, ThrottleType.REQUEST_SIZE,
Math.round(100.1 * blockSize), TimeUnit.MINUTES));
Math.round(100.1 * blockSize), TimeUnit.SECONDS));
triggerUserCacheRefresh(TEST_UTIL, false, TABLE_NAME);

// should execute approximately 10 batches of 10 requests
Expand All @@ -211,7 +211,7 @@ public void testBBSMultiGet() throws Exception {

private void testTraffic(Callable<Long> trafficCallable, long expectedSuccess, long marginOfError)
throws Exception {
TEST_UTIL.waitFor(90_000, () -> {
TEST_UTIL.waitFor(5_000, () -> {
long actualSuccess;
try {
actualSuccess = trafficCallable.call();
Expand Down
Loading