| /** |
| * 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.yarn.util.resource; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.classification.InterfaceStability.Unstable; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.ResourceInformation; |
| import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; |
| import org.apache.hadoop.yarn.util.UnitsConversionUtil; |
| |
| import java.io.PrintWriter; |
| import java.io.StringWriter; |
| import java.util.Arrays; |
| |
| /** |
| * A {@link ResourceCalculator} which uses the concept of |
| * <em>dominant resource</em> to compare multi-dimensional resources. |
| * |
| * Essentially the idea is that the in a multi-resource environment, |
| * the resource allocation should be determined by the dominant share |
| * of an entity (user or queue), which is the maximum share that the |
| * entity has been allocated of any resource. |
| * |
| * In a nutshell, it seeks to maximize the minimum dominant share across |
| * all entities. |
| * |
| * For example, if user A runs CPU-heavy tasks and user B runs |
| * memory-heavy tasks, it attempts to equalize CPU share of user A |
| * with Memory-share of user B. |
| * |
| * In the single resource case, it reduces to max-min fairness for that resource. |
| * |
| * See the Dominant Resource Fairness paper for more details: |
| * www.cs.berkeley.edu/~matei/papers/2011/nsdi_drf.pdf |
| */ |
| @Private |
| @Unstable |
| public class DominantResourceCalculator extends ResourceCalculator { |
| static final Log LOG = LogFactory.getLog(DominantResourceCalculator.class); |
| |
| public DominantResourceCalculator() { |
| } |
| |
| /** |
| * Compare two resources - if the value for every resource type for the lhs |
| * is greater than that of the rhs, return 1. If the value for every resource |
| * type in the lhs is less than the rhs, return -1. Otherwise, return 0 |
| * |
| * @param lhs resource to be compared |
| * @param rhs resource to be compared |
| * @return 0, 1, or -1 |
| */ |
| private int compare(Resource lhs, Resource rhs) { |
| boolean lhsGreater = false; |
| boolean rhsGreater = false; |
| int ret = 0; |
| |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation lhsResourceInformation = lhs |
| .getResourceInformation(i); |
| ResourceInformation rhsResourceInformation = rhs |
| .getResourceInformation(i); |
| int diff = lhsResourceInformation.compareTo(rhsResourceInformation); |
| if (diff >= 1) { |
| lhsGreater = true; |
| } else if (diff <= -1) { |
| rhsGreater = true; |
| } |
| } |
| if (lhsGreater && rhsGreater) { |
| ret = 0; |
| } else if (lhsGreater) { |
| ret = 1; |
| } else if (rhsGreater) { |
| ret = -1; |
| } |
| return ret; |
| } |
| |
| @Override |
| public int compare(Resource clusterResource, Resource lhs, Resource rhs, |
| boolean singleType) { |
| if (lhs.equals(rhs)) { |
| return 0; |
| } |
| |
| if (isInvalidDivisor(clusterResource)) { |
| return this.compare(lhs, rhs); |
| } |
| |
| // We have to calculate the shares for all resource types for both |
| // resources and then look for which resource has the biggest |
| // share overall. |
| ResourceInformation[] clusterRes = clusterResource.getResources(); |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| |
| // If array creation shows up as a time sink, these arrays could be cached |
| // because they're always the same length. |
| double[] lhsShares = new double[maxLength]; |
| double[] rhsShares = new double[maxLength]; |
| double diff; |
| |
| try { |
| if (singleType) { |
| double[] max = new double[2]; |
| |
| calculateShares(clusterRes, lhs, rhs, lhsShares, rhsShares, max); |
| |
| diff = max[0] - max[1]; |
| } else if (maxLength == 2) { |
| // Special case to handle the common scenario of only CPU and memory |
| // so that we can optimize for performance |
| diff = calculateSharesForTwoMandatoryResources(clusterRes, lhs, rhs, |
| lhsShares, rhsShares); |
| } else { |
| calculateShares(clusterRes, lhs, rhs, lhsShares, rhsShares); |
| |
| Arrays.sort(lhsShares); |
| Arrays.sort(rhsShares); |
| |
| diff = compareShares(lhsShares, rhsShares); |
| } |
| } catch (ArrayIndexOutOfBoundsException ex) { |
| StringWriter out = new StringWriter(); // No need to close a StringWriter |
| ex.printStackTrace(new PrintWriter(out)); |
| |
| LOG.error("A problem was encountered while calculating resource " |
| + "availability that should not occur under normal circumstances. " |
| + "Please report this error to the Hadoop community by opening a " |
| + "JIRA ticket at http://issues.apache.org/jira and including the " |
| + "following information:\n* Exception encountered: " + out + "* " |
| + "Cluster resources: " + Arrays.toString(clusterRes) + "\n* " |
| + "LHS resource: " + Arrays.toString(lhs.getResources()) + "\n* " |
| + "RHS resource: " + Arrays.toString(rhs.getResources())); |
| LOG.error("The resource manager is in an inconsistent state. It is safe " |
| + "for the resource manager to be restarted as the error encountered " |
| + "should be transitive. If high availability is enabled, failing " |
| + "over to a standby resource manager is also safe."); |
| throw new YarnRuntimeException("A problem was encountered while " |
| + "calculating resource availability that should not occur under " |
| + "normal circumstances. Please see the log for more information.", |
| ex); |
| } |
| |
| return (int) Math.signum(diff); |
| } |
| |
| /** |
| * Calculate the shares for {@code first} and {@code second} according to |
| * {@code clusterRes}, and store the results in {@code firstShares} and |
| * {@code secondShares}, respectively. All parameters must be non-null. |
| * @param clusterRes the array of ResourceInformation instances that |
| * represents the cluster's maximum resources |
| * @param first the first resource to compare |
| * @param second the second resource to compare |
| * @param firstShares an array to store the shares for the first resource |
| * @param secondShares an array to store the shares for the second resource |
| * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first |
| * resource is less than, equal to, or greater than the max share of the |
| * second resource, respectively |
| * @throws NullPointerException if any parameter is null |
| */ |
| private void calculateShares(ResourceInformation[] clusterRes, Resource first, |
| Resource second, double[] firstShares, double[] secondShares) { |
| ResourceInformation[] firstRes = first.getResources(); |
| ResourceInformation[] secondRes = second.getResources(); |
| |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| firstShares[i] = calculateShare(clusterRes[i], firstRes[i]); |
| secondShares[i] = calculateShare(clusterRes[i], secondRes[i]); |
| } |
| } |
| |
| /** |
| * Calculate the shares for {@code first} and {@code second} according to |
| * {@code clusterRes}, and store the results in {@code firstShares} and |
| * {@code secondShares}, respectively. All parameters must be non-null. |
| * This method assumes that the length of {@code clusterRes} is exactly 2 and |
| * makes performance optimizations based on that assumption. |
| * @param clusterRes the array of ResourceInformation instances that |
| * represents the cluster's maximum resources |
| * @param first the first resource to compare |
| * @param second the second resource to compare |
| * @param firstShares an array to store the shares for the first resource |
| * @param secondShares an array to store the shares for the second resource |
| * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first |
| * resource is less than, equal to, or greater than the max share of the |
| * second resource, respectively |
| * @throws NullPointerException if any parameter is null |
| */ |
| private int calculateSharesForTwoMandatoryResources( |
| ResourceInformation[] clusterRes, Resource first, Resource second, |
| double[] firstShares, double[] secondShares) { |
| ResourceInformation[] firstRes = first.getResources(); |
| ResourceInformation[] secondRes = second.getResources(); |
| firstShares[0] = calculateShare(clusterRes[0], firstRes[0]); |
| secondShares[0] = calculateShare(clusterRes[0], secondRes[0]); |
| firstShares[1] = calculateShare(clusterRes[1], firstRes[1]); |
| secondShares[1] = calculateShare(clusterRes[1], secondRes[1]); |
| |
| int firstDom = 0; |
| int firstSub = 1; |
| if (firstShares[1] > firstShares[0]) { |
| firstDom = 1; |
| firstSub = 0; |
| } |
| int secondDom = 0; |
| int secondSub = 1; |
| if (secondShares[1] > secondShares[0]) { |
| secondDom = 1; |
| secondSub = 0; |
| } |
| |
| if (firstShares[firstDom] > secondShares[secondDom]) { |
| return 1; |
| } else if (firstShares[firstDom] < secondShares[secondDom]) { |
| return -1; |
| } else if (firstShares[firstSub] > secondShares[secondSub]) { |
| return 1; |
| } else if (firstShares[firstSub] < secondShares[secondSub]) { |
| return -1; |
| } else { |
| return 0; |
| } |
| } |
| |
| /** |
| * Calculate the shares for {@code first} and {@code second} according to |
| * {@code clusterRes}, and store the results in {@code firstShares} and |
| * {@code secondShares}, respectively. {@code max} will be populated with |
| * the max shares from {@code firstShare} and {@code secondShare} in the |
| * first and second indices, respectively. All parameters must be non-null, |
| * and {@code max} must have a length of at least 2. |
| * @param clusterRes the array of ResourceInformation instances that |
| * represents the cluster's maximum resources |
| * @param first the first resource to compare |
| * @param second the second resource to compare |
| * @param firstShares an array to store the shares for the first resource |
| * @param secondShares an array to store the shares for the second resource |
| * @param max an array to store the max shares of the first and second |
| * resources |
| * @return -1.0, 0.0, or 1.0, depending on whether the max share of the first |
| * resource is less than, equal to, or greater than the max share of the |
| * second resource, respectively |
| * @throws NullPointerException if any parameter is null |
| * @throws ArrayIndexOutOfBoundsException if the length of {@code max} is |
| * less than 2 |
| */ |
| private void calculateShares(ResourceInformation[] clusterRes, Resource first, |
| Resource second, double[] firstShares, double[] secondShares, |
| double[] max) { |
| ResourceInformation[] firstRes = first.getResources(); |
| ResourceInformation[] secondRes = second.getResources(); |
| |
| max[0] = 0.0; |
| max[1] = 0.0; |
| |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| firstShares[i] = calculateShare(clusterRes[i], firstRes[i]); |
| secondShares[i] = calculateShare(clusterRes[i], secondRes[i]); |
| |
| if (firstShares[i] > max[0]) { |
| max[0] = firstShares[i]; |
| } |
| |
| if (secondShares[i] > max[1]) { |
| max[1] = secondShares[i]; |
| } |
| } |
| } |
| |
| /** |
| * Calculate the share for a resource type. |
| * @param clusterRes the resource type for the cluster maximum |
| * @param res the resource type for which to calculate the share |
| * @return the share |
| */ |
| private double calculateShare(ResourceInformation clusterRes, |
| ResourceInformation res) { |
| // Convert the resources' units into the cluster resource's units |
| long value = UnitsConversionUtil.convert(res.getUnits(), |
| clusterRes.getUnits(), res.getValue()); |
| |
| return (double) value / clusterRes.getValue(); |
| } |
| |
| /** |
| * Compare the two shares arrays by comparing the largest elements, then the |
| * next largest if the previous were equal, etc. The share arrays must be |
| * sorted in ascending order. |
| * @param lhsShares the first share array to compare |
| * @param rhsShares the second share array to compare |
| * @return a number that is less than 0 if the first array is less than the |
| * second, equal to 0 if the arrays are equal, and greater than 0 if the |
| * first array is greater than the second |
| */ |
| private double compareShares(double[] lhsShares, double[] rhsShares) { |
| double diff = 0.0; |
| |
| // lhsShares and rhsShares must necessarily have the same length, because |
| // everyone uses the same master resource list. |
| for (int i = lhsShares.length - 1; i >= 0; i--) { |
| diff = lhsShares[i] - rhsShares[i]; |
| |
| if (diff != 0.0) { |
| break; |
| } |
| } |
| |
| return diff; |
| } |
| |
| @Override |
| public long computeAvailableContainers(Resource available, |
| Resource required) { |
| long min = Long.MAX_VALUE; |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation availableResource = available |
| .getResourceInformation(i); |
| ResourceInformation requiredResource = required.getResourceInformation(i); |
| long requiredResourceValue = UnitsConversionUtil.convert( |
| requiredResource.getUnits(), availableResource.getUnits(), |
| requiredResource.getValue()); |
| if (requiredResourceValue != 0) { |
| long tmp = availableResource.getValue() / requiredResourceValue; |
| min = min < tmp ? min : tmp; |
| } |
| } |
| return min > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) min; |
| } |
| |
| @Override |
| public float divide(Resource clusterResource, |
| Resource numerator, Resource denominator) { |
| int nKnownResourceTypes = ResourceUtils.getNumberOfKnownResourceTypes(); |
| ResourceInformation[] clusterRes = clusterResource.getResources(); |
| // We have to provide the calculateShares() method with somewhere to store |
| // the shares. We don't actually need these shares afterwards. |
| double[] numeratorShares = new double[nKnownResourceTypes]; |
| double[] denominatorShares = new double[nKnownResourceTypes]; |
| // We also have to provide a place for calculateShares() to store the max |
| // shares so that we can use them. |
| double[] max = new double[2]; |
| |
| calculateShares(clusterRes, numerator, denominator, numeratorShares, |
| denominatorShares, max); |
| |
| return (float) (max[0] / max[1]); |
| } |
| |
| @Override |
| public boolean isInvalidDivisor(Resource r) { |
| for (ResourceInformation res : r.getResources()) { |
| if (res.getValue() == 0L) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| @Override |
| public float ratio(Resource a, Resource b) { |
| float ratio = 0.0f; |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation aResourceInformation = a.getResourceInformation(i); |
| ResourceInformation bResourceInformation = b.getResourceInformation(i); |
| long bResourceValue = UnitsConversionUtil.convert( |
| bResourceInformation.getUnits(), aResourceInformation.getUnits(), |
| bResourceInformation.getValue()); |
| float tmp = (float) aResourceInformation.getValue() |
| / (float) bResourceValue; |
| ratio = ratio > tmp ? ratio : tmp; |
| } |
| return ratio; |
| } |
| |
| @Override |
| public Resource divideAndCeil(Resource numerator, int denominator) { |
| return divideAndCeil(numerator, (long) denominator); |
| } |
| |
| public Resource divideAndCeil(Resource numerator, long denominator) { |
| Resource ret = Resource.newInstance(numerator); |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation resourceInformation = ret.getResourceInformation(i); |
| resourceInformation |
| .setValue(divideAndCeil(resourceInformation.getValue(), denominator)); |
| } |
| return ret; |
| } |
| |
| @Override |
| public Resource divideAndCeil(Resource numerator, float denominator) { |
| return Resources.createResource( |
| divideAndCeil(numerator.getMemorySize(), denominator), |
| divideAndCeil(numerator.getVirtualCores(), denominator) |
| ); |
| } |
| |
| @Override |
| public Resource normalize(Resource r, Resource minimumResource, |
| Resource maximumResource, Resource stepFactor) { |
| Resource ret = Resource.newInstance(r); |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation rResourceInformation = r.getResourceInformation(i); |
| ResourceInformation minimumResourceInformation = minimumResource |
| .getResourceInformation(i); |
| ResourceInformation maximumResourceInformation = maximumResource |
| .getResourceInformation(i); |
| ResourceInformation stepFactorResourceInformation = stepFactor |
| .getResourceInformation(i); |
| ResourceInformation tmp = ret.getResourceInformation(i); |
| |
| long rValue = rResourceInformation.getValue(); |
| long minimumValue = UnitsConversionUtil.convert( |
| minimumResourceInformation.getUnits(), |
| rResourceInformation.getUnits(), |
| minimumResourceInformation.getValue()); |
| long maximumValue = UnitsConversionUtil.convert( |
| maximumResourceInformation.getUnits(), |
| rResourceInformation.getUnits(), |
| maximumResourceInformation.getValue()); |
| long stepFactorValue = UnitsConversionUtil.convert( |
| stepFactorResourceInformation.getUnits(), |
| rResourceInformation.getUnits(), |
| stepFactorResourceInformation.getValue()); |
| long value = Math.max(rValue, minimumValue); |
| if (stepFactorValue != 0) { |
| value = roundUp(value, stepFactorValue); |
| } |
| tmp.setValue(Math.min(value, maximumValue)); |
| ret.setResourceInformation(i, tmp); |
| } |
| return ret; |
| } |
| |
| @Override |
| public Resource roundUp(Resource r, Resource stepFactor) { |
| return this.rounding(r, stepFactor, true); |
| } |
| |
| @Override |
| public Resource roundDown(Resource r, Resource stepFactor) { |
| return this.rounding(r, stepFactor, false); |
| } |
| |
| private Resource rounding(Resource r, Resource stepFactor, boolean roundUp) { |
| Resource ret = Resource.newInstance(r); |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation rResourceInformation = r.getResourceInformation(i); |
| ResourceInformation stepFactorResourceInformation = stepFactor |
| .getResourceInformation(i); |
| |
| long rValue = rResourceInformation.getValue(); |
| long stepFactorValue = UnitsConversionUtil.convert( |
| stepFactorResourceInformation.getUnits(), |
| rResourceInformation.getUnits(), |
| stepFactorResourceInformation.getValue()); |
| long value = rValue; |
| if (stepFactorValue != 0) { |
| value = roundUp |
| ? roundUp(rValue, stepFactorValue) |
| : roundDown(rValue, stepFactorValue); |
| } |
| ResourceInformation.copy(rResourceInformation, |
| ret.getResourceInformation(i)); |
| ret.getResourceInformation(i).setValue(value); |
| } |
| return ret; |
| } |
| |
| @Override |
| public Resource multiplyAndNormalizeUp(Resource r, double by, |
| Resource stepFactor) { |
| return this.multiplyAndNormalize(r, by, stepFactor, true); |
| } |
| |
| @Override |
| public Resource multiplyAndNormalizeDown(Resource r, double by, |
| Resource stepFactor) { |
| return this.multiplyAndNormalize(r, by, stepFactor, false); |
| } |
| |
| private Resource multiplyAndNormalize(Resource r, double by, |
| Resource stepFactor, boolean roundUp) { |
| Resource ret = Resource.newInstance(r); |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation rResourceInformation = r.getResourceInformation(i); |
| ResourceInformation stepFactorResourceInformation = stepFactor |
| .getResourceInformation(i); |
| ResourceInformation tmp = ret.getResourceInformation(i); |
| |
| long rValue = rResourceInformation.getValue(); |
| long stepFactorValue = UnitsConversionUtil.convert( |
| stepFactorResourceInformation.getUnits(), |
| rResourceInformation.getUnits(), |
| stepFactorResourceInformation.getValue()); |
| long value; |
| if (stepFactorValue != 0) { |
| value = roundUp |
| ? roundUp((long) Math.ceil((float) (rValue * by)), stepFactorValue) |
| : roundDown((long) (rValue * by), stepFactorValue); |
| } else { |
| value = roundUp |
| ? (long) Math.ceil((float) (rValue * by)) |
| : (long) (rValue * by); |
| } |
| tmp.setValue(value); |
| } |
| return ret; |
| } |
| |
| @Override |
| public boolean fitsIn(Resource smaller, Resource bigger) { |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation sResourceInformation = smaller |
| .getResourceInformation(i); |
| ResourceInformation bResourceInformation = bigger |
| .getResourceInformation(i); |
| long sResourceValue = UnitsConversionUtil.convert( |
| sResourceInformation.getUnits(), bResourceInformation.getUnits(), |
| sResourceInformation.getValue()); |
| if (sResourceValue > bResourceInformation.getValue()) { |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| @Override |
| public boolean isAnyMajorResourceZero(Resource resource) { |
| int maxLength = ResourceUtils.getNumberOfKnownResourceTypes(); |
| for (int i = 0; i < maxLength; i++) { |
| ResourceInformation resourceInformation = resource |
| .getResourceInformation(i); |
| if (resourceInformation.getValue() == 0L) { |
| return true; |
| } |
| } |
| return false; |
| } |
| } |
