Skip to content

Commit d4aa90d

Browse files
wchevreuilwchevreuil
committed
HBASE-30300 CacheAwareLoadBalancer disproportionally favouring cache ratio over skewness (#8502)
Co-authored-by: Claude Code Opus 4.6 <noreply@anthropic.com> Signed-off-by: Tak Lon (Stephen) Wu <taklwu@apache.org> Reviewed-by: Kevin Geiszler <kevin.j.geiszler@gmail.com> Change-Id: I263b26675495effa06a5231144b89a6f4d1634e7
1 parent 2940b74 commit d4aa90d

3 files changed

Lines changed: 261 additions & 352 deletions

File tree

‎hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/BalancerClusterState.java‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -391,8 +391,8 @@ protected BalancerClusterState(Map<ServerName, List<RegionInfo>> clusterState,
391391
serversPerRack);
392392
}
393393

394-
this.serverBlockCacheFreeSize = new long[numServers];
395394
if (serverBlockCacheFreeByServer != null) {
395+
this.serverBlockCacheFreeSize = new long[numServers];
396396
for (int i = 0; i < numServers; i++) {
397397
ServerName sn = servers[i];
398398
this.serverBlockCacheFreeSize[i] =
@@ -900,6 +900,20 @@ void regionMoved(int region, int oldServer, int newServer) {
900900
}
901901
numRegionsPerServerPerTable[tableIndex][newServer]++;
902902

903+
// Update server block cache free size to reflect the region move. The destination server
904+
// will need to cache this region (reducing its free space), and the source server frees
905+
// space when blocks are evicted on region close.
906+
if (serverBlockCacheFreeSize != null) {
907+
long regionSizeBytes = (long) getRegionSizeMinusColdDataMB(region) * 1024L * 1024L;
908+
if (regionSizeBytes > 0) {
909+
serverBlockCacheFreeSize[newServer] =
910+
Math.max(0, serverBlockCacheFreeSize[newServer] - regionSizeBytes);
911+
if (oldServer >= 0) {
912+
serverBlockCacheFreeSize[oldServer] += regionSizeBytes;
913+
}
914+
}
915+
}
916+
903917
// update for servers
904918
int primary = regionIndexToPrimaryIndex[region];
905919
if (oldServer >= 0) {

‎hbase-server/src/main/java/org/apache/hadoop/hbase/master/balancer/CacheAwareLoadBalancer.java‎

Lines changed: 104 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
*/
2828

2929
import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_PERSISTENT_PATH_KEY;
30+
import static org.apache.hadoop.hbase.HConstants.BUCKET_CACHE_SIZE_KEY;
3031

3132
import java.math.BigDecimal;
3233
import java.text.DecimalFormat;
@@ -111,6 +112,8 @@ public enum GeneratorFunctionType {
111112
private float lowCacheRatioThreshold;
112113
private float potentialCacheRatioAfterMove;
113114
private float minFreeCacheSpaceFactor;
115+
private long cachePrefetchOverheadBytes;
116+
private boolean cacheSpaceTrackingEnabled;
114117

115118
private BigDecimal simulatedRatio = BigDecimal.ZERO;
116119

@@ -141,6 +144,16 @@ public void updateConfiguration(Configuration configuration) {
141144
POTENTIAL_CACHE_RATIO_AFTER_MOVE_DEFAULT);
142145
minFreeCacheSpaceFactor =
143146
configuration.getFloat(MIN_FREE_CACHE_SPACE_FACTOR_KEY, MIN_FREE_CACHE_SPACE_FACTOR_DEFAULT);
147+
float bucketCacheSizeMB = configuration.getFloat(BUCKET_CACHE_SIZE_KEY, 0F);
148+
float acceptableFactor = configuration.getFloat("hbase.bucketcache.acceptfactor", 0.95f);
149+
cachePrefetchOverheadBytes =
150+
(long) (bucketCacheSizeMB * 1024L * 1024L * (1 - acceptableFactor));
151+
cacheSpaceTrackingEnabled = bucketCacheSizeMB > 0;
152+
if (!cacheSpaceTrackingEnabled) {
153+
LOG.warn("{} is not configured on the master. The free-space relocation heuristic cannot "
154+
+ "account for the prefetch threshold and may move regions to servers where prefetch "
155+
+ "will be blocked.", BUCKET_CACHE_SIZE_KEY);
156+
}
144157
}
145158

146159
/**
@@ -207,11 +220,14 @@ public void updateClusterMetrics(ClusterMetrics clusterMetrics) {
207220
}
208221

209222
protected Map<ServerName, Long> getServerBlockCacheFreeBytes() {
210-
if (clusterStatus == null) {
223+
if (clusterStatus == null || !cacheSpaceTrackingEnabled) {
211224
return null;
212225
}
213226
Map<ServerName, Long> map = new HashMap<>();
214-
clusterStatus.getLiveServerMetrics().forEach((sn, sm) -> map.put(sn, sm.getCacheFreeSize()));
227+
clusterStatus.getLiveServerMetrics().forEach((sn, sm) -> {
228+
long effectiveFree = Math.max(0, sm.getCacheFreeSize() - cachePrefetchOverheadBytes);
229+
map.put(sn, effectiveFree);
230+
});
215231
return map;
216232
}
217233

@@ -291,36 +307,71 @@ private RegionInfo getRegionInfoByEncodedName(BalancerClusterState cluster, Stri
291307
return null;
292308
}
293309

310+
private boolean serverHasCacheSpaceForRegion(BalancerClusterState cluster, int region,
311+
int server) {
312+
if (cluster.serverBlockCacheFreeSize == null) {
313+
return true;
314+
}
315+
int regionSizeMb = cluster.getRegionSizeMinusColdDataMB(region);
316+
if (regionSizeMb <= 0) {
317+
return true;
318+
}
319+
long bytesNeeded = (long) regionSizeMb * 1024L * 1024L;
320+
return cluster.serverBlockCacheFreeSize[server] >= bytesNeeded;
321+
}
322+
294323
@Override
295324
public void throttle(RegionPlan plan) {
325+
synchronized (this) {
326+
try {
327+
// Release the monitor while waiting to avoid blocking other threads.
328+
wait(getThrottleDurationMs(plan));
329+
} catch (InterruptedException e) {
330+
throw new RuntimeException(e);
331+
}
332+
}
333+
}
334+
335+
public long getThrottleDurationMs(RegionPlan plan) {
296336
Pair<ServerName, Float> rsRatio = this.regionCacheRatioOnOldServerMap.get(plan.getRegionName());
297337
if (
298338
rsRatio != null && plan.getDestination().equals(rsRatio.getFirst())
299339
&& rsRatio.getSecond() >= ratioThreshold
300340
) {
301341
LOG.debug("Moving region {} to server {} with cache ratio {}. No throttling needed.",
302342
plan.getRegionInfo().getEncodedName(), plan.getDestination(), rsRatio.getSecond());
343+
return 0L;
344+
}
345+
// Skip throttling for regions with low cache ratio on their source server — there is
346+
// negligible cached data to lose, so no warm-up delay is needed on the destination.
347+
float cacheRatioOnSource = getRegionCacheRatioOnSource(plan);
348+
if (cacheRatioOnSource < lowCacheRatioThreshold) {
349+
LOG.debug(
350+
"Moving region {} to server {} with low cache ratio {} on source. No throttling needed.",
351+
plan.getRegionInfo().getEncodedName(), plan.getDestination(), cacheRatioOnSource);
352+
return 0L;
353+
}
354+
if (rsRatio != null) {
355+
LOG.debug("Moving region {} to server {} with cache ratio: {}. Throttling move for {}ms.",
356+
plan.getRegionInfo().getEncodedName(), plan.getDestination(),
357+
plan.getDestination().equals(rsRatio.getFirst()) ? rsRatio.getSecond() : "unknown",
358+
sleepTime);
303359
} else {
304-
if (rsRatio != null) {
305-
LOG.debug("Moving region {} to server {} with cache ratio: {}. Throttling move for {}ms.",
306-
plan.getRegionInfo().getEncodedName(), plan.getDestination(),
307-
plan.getDestination().equals(rsRatio.getFirst()) ? rsRatio.getSecond() : "unknown",
308-
sleepTime);
309-
} else {
310-
LOG.debug(
311-
"Moving region {} to server {} with no cache ratio info for the region. "
312-
+ "Throttling move for {}ms.",
313-
plan.getRegionInfo().getEncodedName(), plan.getDestination(), sleepTime);
314-
}
315-
synchronized (this) {
316-
try {
317-
// Release the monitor while waiting to avoid blocking other threads.
318-
wait(sleepTime);
319-
} catch (InterruptedException e) {
320-
throw new RuntimeException(e);
321-
}
322-
}
360+
LOG.debug(
361+
"Moving region {} to server {} with no cache ratio info for the region. "
362+
+ "Throttling move for {}ms.",
363+
plan.getRegionInfo().getEncodedName(), plan.getDestination(), sleepTime);
323364
}
365+
return sleepTime;
366+
}
367+
368+
private float getRegionCacheRatioOnSource(RegionPlan plan) {
369+
Deque<BalancerRegionLoad> regionLoad = loads.get(plan.getRegionName());
370+
if (regionLoad != null && !regionLoad.isEmpty()) {
371+
return regionLoad.getFirst().getCurrentRegionCacheRatio();
372+
}
373+
// Unknown cache ratio — assume it may be cached and require throttling
374+
return 1.0f;
324375
}
325376

326377
@Override
@@ -452,6 +503,28 @@ private boolean moveRegionToOldServer(BalancerClusterState cluster, int regionIn
452503
return false;
453504
}
454505

506+
// If the region is already well-cached on its current server, don't disrupt it.
507+
// The old server's historical cache data may be stale, and moving a hot region
508+
// causes unnecessary cache churn.
509+
if (cacheRatioOnCurrentServer >= ratioThreshold) {
510+
if (LOG.isDebugEnabled()) {
511+
LOG.debug(
512+
"Region {} not moved from {} to {} as it is already well-cached ({}) on current server",
513+
cluster.regions[regionIndex].getEncodedName(), cluster.servers[currentServerIndex],
514+
cluster.servers[oldServerIndex], cacheRatioOnCurrentServer);
515+
}
516+
return false;
517+
}
518+
519+
if (!serverHasCacheSpaceForRegion(cluster, regionIndex, oldServerIndex)) {
520+
if (LOG.isDebugEnabled()) {
521+
LOG.debug("Region {} not moved from {} to {} as destination server lacks cache space",
522+
cluster.regions[regionIndex].getEncodedName(), cluster.servers[currentServerIndex],
523+
cluster.servers[oldServerIndex]);
524+
}
525+
return false;
526+
}
527+
455528
DecimalFormat df = new DecimalFormat("#");
456529
df.setMaximumFractionDigits(4);
457530

@@ -512,60 +585,6 @@ private class CacheAwareSkewnessCandidateGenerator extends LoadCandidateGenerato
512585
@Override
513586
BalanceAction pickRandomRegions(BalancerClusterState cluster, int thisServer, int otherServer) {
514587
simulatedRatio = BigDecimal.ZERO;
515-
// First move all the regions which were hosted previously on some other server back to their
516-
// old servers
517-
if (
518-
!regionCacheRatioOnOldServerMap.isEmpty()
519-
&& regionCacheRatioOnOldServerMap.entrySet().iterator().hasNext()
520-
) {
521-
// Get the first region index in the historical cache ratio list
522-
Map.Entry<String, Pair<ServerName, Float>> regionEntry =
523-
regionCacheRatioOnOldServerMap.entrySet().iterator().next();
524-
String regionEncodedName = regionEntry.getKey();
525-
526-
RegionInfo regionInfo = getRegionInfoByEncodedName(cluster, regionEncodedName);
527-
if (regionInfo == null) {
528-
LOG.warn("Region {} does not exist", regionEncodedName);
529-
regionCacheRatioOnOldServerMap.remove(regionEncodedName);
530-
return BalanceAction.NULL_ACTION;
531-
}
532-
if (regionInfo.isMetaRegion() || regionInfo.getTable().isSystemTable()) {
533-
regionCacheRatioOnOldServerMap.remove(regionEncodedName);
534-
return BalanceAction.NULL_ACTION;
535-
}
536-
537-
int regionIndex = cluster.regionsToIndex.get(regionInfo);
538-
539-
// Get the current host name for this region
540-
thisServer = cluster.regionIndexToServerIndex[regionIndex];
541-
542-
// Get the old server index
543-
otherServer = cluster.serversToIndex.get(regionEntry.getValue().getFirst().getAddress());
544-
545-
regionCacheRatioOnOldServerMap.remove(regionEncodedName);
546-
547-
if (otherServer < 0) {
548-
// The old server has been moved to other host and hence, the region cannot be moved back
549-
// to the old server
550-
if (LOG.isDebugEnabled()) {
551-
LOG.debug(
552-
"CacheAwareSkewnessCandidateGenerator: Region {} not moved to the old "
553-
+ "server {} as the server does not exist",
554-
regionEncodedName, regionEntry.getValue().getFirst().getHostname());
555-
}
556-
return BalanceAction.NULL_ACTION;
557-
}
558-
559-
if (LOG.isDebugEnabled()) {
560-
LOG.debug(
561-
"CacheAwareSkewnessCandidateGenerator: Region {} moved from {} to {} as it "
562-
+ "was hosted their earlier",
563-
regionEncodedName, cluster.servers[thisServer].getHostname(),
564-
cluster.servers[otherServer].getHostname());
565-
}
566-
567-
return getAction(thisServer, regionIndex, otherServer, -1);
568-
}
569588

570589
if (thisServer < 0 || otherServer < 0) {
571590
return BalanceAction.NULL_ACTION;
@@ -642,7 +661,7 @@ protected void regionMoved(int region, int oldServer, int newServer) {
642661

643662
@Override
644663
public final void updateWeight(Map<Class<? extends CandidateGenerator>, Double> weights) {
645-
weights.merge(LoadCandidateGenerator.class, cost(), Double::sum);
664+
weights.merge(CacheAwareSkewnessCandidateGenerator.class, cost(), Double::sum);
646665
}
647666
}
648667

@@ -739,6 +758,9 @@ private double[] computeCurrentWeightedContributions(BalancerClusterState cluste
739758
* from low ratios when capacity exists somewhere in the cluster.
740759
*/
741760
private double potentialBestWeightedFromFreeCache(BalancerClusterState cluster, int region) {
761+
if (cluster.serverBlockCacheFreeSize == null) {
762+
return 0.0;
763+
}
742764
float observedRatio = cluster.getSumRegionCacheAndColdDataRatio(region);
743765
if (observedRatio >= lowCacheRatioThreshold) {
744766
return 0.0;
@@ -769,9 +791,11 @@ protected void regionMoved(int region, int oldServer, int newServer) {
769791
if (simulatedRatio.equals(BigDecimal.ZERO)) {
770792
double potentialCachedSizeOnNewServer =
771793
cluster.getRegionSizeMinusColdDataMB(region) * potentialCacheRatioAfterMove;
772-
boolean simulateCacheBasedOnFreeSpace =
773-
cluster.getOrComputeRegionCacheRatio(region, oldServer) < lowCacheRatioThreshold
774-
&& cluster.serverBlockCacheFreeSize[newServer] >= potentialCachedSizeOnNewServer;
794+
long potentialCachedBytesOnNewServer =
795+
(long) (potentialCachedSizeOnNewServer * 1024L * 1024L);
796+
boolean simulateCacheBasedOnFreeSpace = cluster.serverBlockCacheFreeSize != null
797+
&& cluster.getOrComputeRegionCacheRatio(region, oldServer) < lowCacheRatioThreshold
798+
&& cluster.serverBlockCacheFreeSize[newServer] >= potentialCachedBytesOnNewServer;
775799
double regionCacheRatioOnNewServer = simulateCacheBasedOnFreeSpace
776800
? potentialCachedSizeOnNewServer
777801
: cluster.getOrComputeWeightedRegionCacheRatio(region, newServer);
@@ -808,7 +832,7 @@ private int getServerWithBestCacheRatioForRegion(int region) {
808832

809833
@Override
810834
public void updateWeight(Map<Class<? extends CandidateGenerator>, Double> weights) {
811-
weights.merge(LoadCandidateGenerator.class, cost(), Double::sum);
835+
weights.merge(CacheAwareCandidateGenerator.class, cost(), Double::sum);
812836
}
813837
}
814838
}

0 commit comments

Comments
 (0)