ndimiduk commented on a change in pull request #2454: URL: https://github.com/apache/hbase/pull/2454#discussion_r499870148
########## File path: hbase-server/src/main/java/org/apache/hadoop/hbase/master/normalizer/RegionNormalizerWorker.java ########## @@ -0,0 +1,254 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.master.normalizer; + +import java.io.IOException; +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.client.TableDescriptor; +import org.apache.hadoop.hbase.master.HMaster; +import org.apache.hadoop.hbase.master.MasterServices; +import org.apache.yetus.audience.InterfaceAudience; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter; +import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils; + +/** + * Consumes normalization request targets ({@link TableName}s) off the + * {@link RegionNormalizerWorkQueue}, dispatches them to the {@link RegionNormalizer}, + * and executes the resulting {@link NormalizationPlan}s. + */ +@InterfaceAudience.Private +class RegionNormalizerWorker implements Runnable { + private static final Logger LOG = LoggerFactory.getLogger(RegionNormalizerWorker.class); + private static final String RATE_LIMIT_BYTES_PER_SEC_KEY = + "hbase.normalizer.throughput.max_bytes_per_sec"; + private static final long RATE_UNLIMITED_BYTES = 1_000_000_000_000L; // 1TB/sec + + private final MasterServices masterServices; + private final RegionNormalizer regionNormalizer; + private final RegionNormalizerWorkQueue<TableName> workQueue; + private final RateLimiter rateLimiter; + + private final long[] skippedCount; + private long splitPlanCount; + private long mergePlanCount; + + public RegionNormalizerWorker( + final Configuration configuration, + final MasterServices masterServices, + final RegionNormalizer regionNormalizer, + final RegionNormalizerWorkQueue<TableName> workQueue + ) { + this.masterServices = masterServices; + this.regionNormalizer = regionNormalizer; + this.workQueue = workQueue; + this.skippedCount = new long[NormalizationPlan.PlanType.values().length]; + this.splitPlanCount = 0; + this.mergePlanCount = 0; + this.rateLimiter = loadRateLimiter(configuration); + } + + private static RateLimiter loadRateLimiter(final Configuration configuration) { + long rateLimitBytes = + configuration.getLongBytes(RATE_LIMIT_BYTES_PER_SEC_KEY, RATE_UNLIMITED_BYTES); + long rateLimitMbs = rateLimitBytes / 1_000_000L; + if (rateLimitMbs <= 0) { + LOG.warn("Configured value {}={} is <= 1MB. Falling back to default.", + RATE_LIMIT_BYTES_PER_SEC_KEY, rateLimitBytes); + rateLimitBytes = RATE_UNLIMITED_BYTES; + rateLimitMbs = RATE_UNLIMITED_BYTES / 1_000_000L; + } + LOG.info("Normalizer rate limit set to {}", + rateLimitBytes == RATE_UNLIMITED_BYTES ? "unlimited" : rateLimitMbs + " MB/sec"); + return RateLimiter.create(rateLimitMbs); + } + + /** + * @see RegionNormalizerManager#planSkipped(NormalizationPlan.PlanType) + */ + void planSkipped(NormalizationPlan.PlanType type) { + synchronized (skippedCount) { + // updates come here via procedure threads, so synchronize access to this counter. + skippedCount[type.ordinal()]++; + } + } + + /** + * @see RegionNormalizerManager#getSkippedCount(NormalizationPlan.PlanType) + */ + long getSkippedCount(NormalizationPlan.PlanType type) { + return skippedCount[type.ordinal()]; + } + + /** + * @see HMaster#getSplitPlanCount() + */ + long getSplitPlanCount() { + return splitPlanCount; + } + + /** + * @see HMaster#getMergePlanCount() + */ + long getMergePlanCount() { + return mergePlanCount; + } + + @Override + public void run() { + while (true) { + if (Thread.interrupted()) { + LOG.debug("interrupt detected. terminating."); + break; + } + final TableName tableName; + try { + tableName = workQueue.take(); + } catch (InterruptedException e) { + LOG.debug("interrupt detected. terminating."); + break; + } + + final List<NormalizationPlan> plans = calculatePlans(tableName); + submitPlans(plans); + } + } + + private List<NormalizationPlan> calculatePlans(final TableName tableName) { + try { + final TableDescriptor tblDesc = masterServices.getTableDescriptors().get(tableName); + if (tblDesc != null && !tblDesc.isNormalizationEnabled()) { + LOG.debug("Skipping table {} because normalization is disabled in its table properties.", + tableName); + return Collections.emptyList(); + } + } catch (IOException e) { + LOG.debug("Skipping table {} because unable to access its table descriptor.", tableName, e); + return Collections.emptyList(); + } + + if (masterServices.skipRegionManagementAction("region normalizer")) { + return Collections.emptyList(); + } + + final List<NormalizationPlan> plans = regionNormalizer.computePlansForTable(tableName); + if (CollectionUtils.isEmpty(plans)) { + LOG.debug("No normalization required for table {}.", tableName); + return Collections.emptyList(); + } + return plans; + } + + private void submitPlans(final List<NormalizationPlan> plans) { + // as of this writing, `plan.submit()` is non-blocking and uses Async Admin APIs to submit + // task, so there's no artificial rate-limiting of merge/split requests due to this serial loop. + for (NormalizationPlan plan : plans) { + switch (plan.getType()) { + case MERGE: { + submitMergePlan((MergeNormalizationPlan) plan); + break; + } + case SPLIT: { + submitSplitPlan((SplitNormalizationPlan) plan); + break; + } + case NONE: + LOG.debug("Nothing to do for {} with PlanType=NONE. Ignoring.", plan); + planSkipped(plan.getType()); + break; + default: + LOG.warn("Plan {} is of an unrecognized PlanType. Ignoring.", plan); + planSkipped(plan.getType()); + break; + } + } + } + + /** + * Interacts with {@link MasterServices} in order to execute a plan. + */ + private void submitMergePlan(final MergeNormalizationPlan plan) { + final int totalSizeMb; + try { + final long totalSizeMbLong = plan.getNormalizationTargets() + .stream() + .mapToLong(NormalizationTarget::getRegionSizeMb) + .reduce(0, Math::addExact); + totalSizeMb = Math.toIntExact(totalSizeMbLong); + } catch (ArithmeticException e) { + LOG.debug("Sum of merge request size overflows rate limiter data type. {}", plan); + planSkipped(plan.getType()); + return; + } + + final RegionInfo[] infos = plan.getNormalizationTargets() + .stream() + .map(NormalizationTarget::getRegionInfo) + .toArray(RegionInfo[]::new); + final long rateLimitedSecs = Math.round(rateLimiter.acquire(Math.max(1, totalSizeMb))); + LOG.debug("Rate limiting delayed this operation by {}", Duration.ofSeconds(rateLimitedSecs)); Review comment: So you're suggesting that the thread's blocking could match more closely the way resources are consumed. By checking on the rate limit after submitting the procedure, the thread is held until after the submitted work as been conceptually processed. I think that makes sense. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
