Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-27093 AsyncNonMetaRegionLocator：put Complete CompletableFuture outside lock block (#4496) 

Signed-off-by: Duo Zhang <zhangduo@apache.org>
(cherry picked from commit 176c43c5ad)
This commit is contained in:
wangzhi 2022-06-07 12:17:52 +08:00 committed by Duo Zhang
parent 046e51085f
commit 5cc614f23f
1 changed files with 44 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncNonMetaRegionLocator.java
View File

@ -35,10 +35,12 @@ import static org.apache.hadoop.hbase.util.Bytes.BYTES_COMPARATOR;
import static org.apache.hadoop.hbase.util.ConcurrentMapUtils.computeIfAbsent; import static org.apache.hadoop.hbase.util.ConcurrentMapUtils.computeIfAbsent;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.HashSet; import java.util.HashSet;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
@ -122,6 +124,26 @@ class AsyncNonMetaRegionLocator {
} }
} }
private static final class RegionLocationsFutureResult {
private final CompletableFuture<RegionLocations> future;
private final RegionLocations result;
private final Throwable e;
public RegionLocationsFutureResult(CompletableFuture<RegionLocations> future,
RegionLocations result, Throwable e) {
this.future = future;
this.result = result;
this.e = e;
}
public void complete() {
if (e != null) {
future.completeExceptionally(e);
}
future.complete(result);
}
}
private static final class TableCache { private static final class TableCache {
private final ConcurrentNavigableMap<byte[], RegionLocations> cache = private final ConcurrentNavigableMap<byte[], RegionLocations> cache =
@ -148,18 +170,20 @@ class AsyncNonMetaRegionLocator {
return allRequests.keySet().stream().filter(r -> !isPending(r)).findFirst(); return allRequests.keySet().stream().filter(r -> !isPending(r)).findFirst();
} }
public void clearCompletedRequests(RegionLocations locations) { public List<RegionLocationsFutureResult> clearCompletedRequests(RegionLocations locations) {
List<RegionLocationsFutureResult> futureResultList = new ArrayList<>();
for (Iterator<Map.Entry<LocateRequest, CompletableFuture<RegionLocations>>> iter = for (Iterator<Map.Entry<LocateRequest, CompletableFuture<RegionLocations>>> iter =
allRequests.entrySet().iterator(); iter.hasNext();) { allRequests.entrySet().iterator(); iter.hasNext();) {
Map.Entry<LocateRequest, CompletableFuture<RegionLocations>> entry = iter.next(); Map.Entry<LocateRequest, CompletableFuture<RegionLocations>> entry = iter.next();
if (tryComplete(entry.getKey(), entry.getValue(), locations)) { if (tryComplete(entry.getKey(), entry.getValue(), locations, futureResultList)) {
iter.remove(); iter.remove();
} }
} }
return futureResultList;
} }
private boolean tryComplete(LocateRequest req, CompletableFuture<RegionLocations> future, private boolean tryComplete(LocateRequest req, CompletableFuture<RegionLocations> future,
RegionLocations locations) { RegionLocations locations, List<RegionLocationsFutureResult> futureResultList) {
if (future.isDone()) { if (future.isDone()) {
return true; return true;
} }
@ -185,7 +209,7 @@ class AsyncNonMetaRegionLocator {
completed = loc.getRegion().containsRow(req.row); completed = loc.getRegion().containsRow(req.row);
} }
if (completed) { if (completed) {
future.complete(locations); futureResultList.add(new RegionLocationsFutureResult(future, locations, null));
return true; return true;
} else { } else {
return false; return false;
@ -320,32 +344,36 @@ class AsyncNonMetaRegionLocator {
TableCache tableCache = getTableCache(tableName); TableCache tableCache = getTableCache(tableName);
if (locs != null) { if (locs != null) {
RegionLocations addedLocs = addToCache(tableCache, locs); RegionLocations addedLocs = addToCache(tableCache, locs);
List<RegionLocationsFutureResult> futureResultList = new ArrayList<>();
synchronized (tableCache) { synchronized (tableCache) {
tableCache.pendingRequests.remove(req); tableCache.pendingRequests.remove(req);
tableCache.clearCompletedRequests(addedLocs); futureResultList.addAll(tableCache.clearCompletedRequests(addedLocs));
// Remove a complete locate request in a synchronized block, so the table cache must have // Remove a complete locate request in a synchronized block, so the table cache must have
// quota to send a candidate request. // quota to send a candidate request.
toSend = tableCache.getCandidate(); toSend = tableCache.getCandidate();
toSend.ifPresent(r -> tableCache.send(r)); toSend.ifPresent(r -> tableCache.send(r));
} }
futureResultList.forEach(RegionLocationsFutureResult::complete);
toSend.ifPresent(r -> locateInMeta(tableName, r)); toSend.ifPresent(r -> locateInMeta(tableName, r));
} else { } else {
// we meet an error // we meet an error
assert error != null; assert error != null;
List<RegionLocationsFutureResult> futureResultList = new ArrayList<>();
synchronized (tableCache) { synchronized (tableCache) {
tableCache.pendingRequests.remove(req); tableCache.pendingRequests.remove(req);
// fail the request itself, no matter whether it is a DoNotRetryIOException, as we have // fail the request itself, no matter whether it is a DoNotRetryIOException, as we have
// already retried several times // already retried several times
CompletableFuture<?> future = tableCache.allRequests.remove(req); CompletableFuture<RegionLocations> future = tableCache.allRequests.remove(req);
if (future != null) { if (future != null) {
future.completeExceptionally(error); futureResultList.add(new RegionLocationsFutureResult(future, null, error));
} }
tableCache.clearCompletedRequests(null); futureResultList.addAll(tableCache.clearCompletedRequests(null));
// Remove a complete locate request in a synchronized block, so the table cache must have // Remove a complete locate request in a synchronized block, so the table cache must have
// quota to send a candidate request. // quota to send a candidate request.
toSend = tableCache.getCandidate(); toSend = tableCache.getCandidate();
toSend.ifPresent(r -> tableCache.send(r)); toSend.ifPresent(r -> tableCache.send(r));
} }
futureResultList.forEach(RegionLocationsFutureResult::complete);
toSend.ifPresent(r -> locateInMeta(tableName, r)); toSend.ifPresent(r -> locateInMeta(tableName, r));
} }
} }
@ -543,9 +571,11 @@ class AsyncNonMetaRegionLocator {
continue; continue;
} }
RegionLocations addedLocs = addToCache(tableCache, locs); RegionLocations addedLocs = addToCache(tableCache, locs);
List<RegionLocationsFutureResult> futureResultList = new ArrayList<>();
synchronized (tableCache) { synchronized (tableCache) {
tableCache.clearCompletedRequests(addedLocs); futureResultList.addAll(tableCache.clearCompletedRequests(addedLocs));
} }
futureResultList.forEach(RegionLocationsFutureResult::complete);
} }
} }
} }
@ -677,12 +707,16 @@ class AsyncNonMetaRegionLocator {
if (tableCache == null) { if (tableCache == null) {
return; return;
} }
List<RegionLocationsFutureResult> futureResultList = new ArrayList<>();
synchronized (tableCache) { synchronized (tableCache) {
if (!tableCache.allRequests.isEmpty()) { if (!tableCache.allRequests.isEmpty()) {
IOException error = new IOException("Cache cleared"); IOException error = new IOException("Cache cleared");
tableCache.allRequests.values().forEach(f -> f.completeExceptionally(error)); tableCache.allRequests.values().forEach(f -> {
futureResultList.add(new RegionLocationsFutureResult(f, null, error));
});
} }
} }
futureResultList.forEach(RegionLocationsFutureResult::complete);
conn.getConnectionMetrics() conn.getConnectionMetrics()
.ifPresent(metrics -> metrics.incrMetaCacheNumClearRegion(tableCache.cache.size())); .ifPresent(metrics -> metrics.incrMetaCacheNumClearRegion(tableCache.cache.size()));
} }