Revert "YARN-7411. Inter-Queue preemption's computeFixpointAllocation need to handle absolute resources while computing normalizedGuarantee. (Sunil G via wangda)"
Pulling this back into branch-2 prevented nodemanagers from coming up if extended resources are present.
This reverts commit ab83765a6c.
This commit is contained in:
Eric E Payne
parent
6973f78a73
commit
a74411ebaa
|
|
@ -26,6 +26,7 @@ import org.apache.hadoop.yarn.api.protocolrecords.ResourceTypes;
|
|||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceInformation;
|
||||
import org.apache.hadoop.yarn.exceptions.ResourceNotFoundException;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceInformationProto;
|
||||
|
|
@ -151,6 +152,17 @@ public class ResourcePBImpl extends Resource {
|
|||
.newInstance(ResourceInformation.VCORES);
|
||||
this.setMemorySize(p.getMemory());
|
||||
this.setVirtualCores(p.getVirtualCores());
|
||||
|
||||
// Update missing resource information on respective index.
|
||||
updateResourceInformationMap(types);
|
||||
}
|
||||
|
||||
private void updateResourceInformationMap(ResourceInformation[] types) {
|
||||
for (int i = 0; i < types.length; i++) {
|
||||
if (resources[i] == null) {
|
||||
resources[i] = ResourceInformation.newInstance(types[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static ResourceInformation newDefaultInformation(
|
||||
|
|
|
|||
|
|
@ -111,14 +111,6 @@ public class DefaultResourceCalculator extends ResourceCalculator {
|
|||
stepFactor.getMemorySize()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource multiplyAndNormalizeUp(Resource r, double[] by,
|
||||
Resource stepFactor) {
|
||||
return Resources.createResource(
|
||||
roundUp((long) (r.getMemorySize() * by[0] + 0.5),
|
||||
stepFactor.getMemorySize()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource multiplyAndNormalizeDown(Resource r, double by,
|
||||
Resource stepFactor) {
|
||||
|
|
|
|||
|
|
@ -495,27 +495,6 @@ public class DominantResourceCalculator extends ResourceCalculator {
|
|||
return ret;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource multiplyAndNormalizeUp(Resource r, double[] by,
|
||||
Resource stepFactor) {
|
||||
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());
|
||||
ret.setResourceValue(i, ResourceCalculator
|
||||
.roundUp((long) Math.ceil(rValue * by[i]), stepFactorValue));
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource multiplyAndNormalizeUp(Resource r, double by,
|
||||
Resource stepFactor) {
|
||||
|
|
|
|||
|
|
@ -126,18 +126,6 @@ public abstract class ResourceCalculator {
|
|||
public abstract Resource multiplyAndNormalizeUp(
|
||||
Resource r, double by, Resource stepFactor);
|
||||
|
||||
/**
|
||||
* Multiply resource <code>r</code> by factor <code>by</code>
|
||||
* and normalize up using step-factor <code>stepFactor</code>.
|
||||
*
|
||||
* @param r resource to be multiplied
|
||||
* @param by multiplier array for all resource types
|
||||
* @param stepFactor factor by which to normalize up
|
||||
* @return resulting normalized resource
|
||||
*/
|
||||
public abstract Resource multiplyAndNormalizeUp(
|
||||
Resource r, double[] by, Resource stepFactor);
|
||||
|
||||
/**
|
||||
* Multiply resource <code>r</code> by factor <code>by</code>
|
||||
* and normalize down using step-factor <code>stepFactor</code>.
|
||||
|
|
|
|||
|
|
@ -354,11 +354,6 @@ public class Resources {
|
|||
return lhs;
|
||||
}
|
||||
|
||||
public static Resource multiplyAndNormalizeUp(ResourceCalculator calculator,
|
||||
Resource lhs, double[] by, Resource factor) {
|
||||
return calculator.multiplyAndNormalizeUp(lhs, by, factor);
|
||||
}
|
||||
|
||||
public static Resource multiplyAndNormalizeUp(
|
||||
ResourceCalculator calculator,Resource lhs, double by, Resource factor) {
|
||||
return calculator.multiplyAndNormalizeUp(lhs, by, factor);
|
||||
|
|
|
|||
|
|
@ -19,11 +19,8 @@
|
|||
package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceInformation;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.policy.PriorityUtilizationQueueOrderingPolicy;
|
||||
import org.apache.hadoop.yarn.util.UnitsConversionUtil;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceUtils;
|
||||
import org.apache.hadoop.yarn.util.resource.Resources;
|
||||
|
||||
import java.util.ArrayList;
|
||||
|
|
@ -201,33 +198,18 @@ public class AbstractPreemptableResourceCalculator {
|
|||
private void resetCapacity(Resource clusterResource,
|
||||
Collection<TempQueuePerPartition> queues, boolean ignoreGuar) {
|
||||
Resource activeCap = Resource.newInstance(0, 0);
|
||||
int maxLength = ResourceUtils.getNumberOfKnownResourceTypes();
|
||||
|
||||
if (ignoreGuar) {
|
||||
for (TempQueuePerPartition q : queues) {
|
||||
for (int i = 0; i < maxLength; i++) {
|
||||
q.normalizedGuarantee[i] = 1.0f / queues.size();
|
||||
}
|
||||
q.normalizedGuarantee = 1.0f / queues.size();
|
||||
}
|
||||
} else {
|
||||
for (TempQueuePerPartition q : queues) {
|
||||
Resources.addTo(activeCap, q.getGuaranteed());
|
||||
}
|
||||
for (TempQueuePerPartition q : queues) {
|
||||
for (int i = 0; i < maxLength; i++) {
|
||||
ResourceInformation nResourceInformation = q.getGuaranteed()
|
||||
.getResourceInformation(i);
|
||||
ResourceInformation dResourceInformation = activeCap
|
||||
.getResourceInformation(i);
|
||||
|
||||
long nValue = nResourceInformation.getValue();
|
||||
long dValue = UnitsConversionUtil.convert(
|
||||
dResourceInformation.getUnits(), nResourceInformation.getUnits(),
|
||||
dResourceInformation.getValue());
|
||||
if (dValue != 0) {
|
||||
q.normalizedGuarantee[i] = (float) nValue / dValue;
|
||||
}
|
||||
}
|
||||
q.normalizedGuarantee = Resources.divide(rc, clusterResource,
|
||||
q.getGuaranteed(), activeCap);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,11 +22,9 @@ import org.apache.hadoop.yarn.api.records.Resource;
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceUtils;
|
||||
import org.apache.hadoop.yarn.util.resource.Resources;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
|
|
@ -48,7 +46,7 @@ public class TempQueuePerPartition extends AbstractPreemptionEntity {
|
|||
Resource untouchableExtra;
|
||||
Resource preemptableExtra;
|
||||
|
||||
double[] normalizedGuarantee;
|
||||
double normalizedGuarantee;
|
||||
|
||||
final ArrayList<TempQueuePerPartition> children;
|
||||
private Collection<TempAppPerPartition> apps;
|
||||
|
|
@ -86,8 +84,7 @@ public class TempQueuePerPartition extends AbstractPreemptionEntity {
|
|||
pendingDeductReserved = Resources.createResource(0);
|
||||
}
|
||||
|
||||
this.normalizedGuarantee = new double[ResourceUtils
|
||||
.getNumberOfKnownResourceTypes()];
|
||||
this.normalizedGuarantee = Float.NaN;
|
||||
this.children = new ArrayList<>();
|
||||
this.apps = new ArrayList<>();
|
||||
this.untouchableExtra = Resource.newInstance(0, 0);
|
||||
|
|
@ -229,9 +226,8 @@ public class TempQueuePerPartition extends AbstractPreemptionEntity {
|
|||
sb.append(" NAME: " + queueName).append(" CUR: ").append(current)
|
||||
.append(" PEN: ").append(pending).append(" RESERVED: ").append(reserved)
|
||||
.append(" GAR: ").append(getGuaranteed()).append(" NORM: ")
|
||||
.append(Arrays.toString(normalizedGuarantee))
|
||||
.append(" IDEAL_ASSIGNED: ").append(idealAssigned)
|
||||
.append(" IDEAL_PREEMPT: ").append(toBePreempted)
|
||||
.append(normalizedGuarantee).append(" IDEAL_ASSIGNED: ")
|
||||
.append(idealAssigned).append(" IDEAL_PREEMPT: ").append(toBePreempted)
|
||||
.append(" ACTUAL_PREEMPT: ").append(getActuallyToBePreempted())
|
||||
.append(" UNTOUCHABLE: ").append(untouchableExtra)
|
||||
.append(" PREEMPTABLE: ").append(preemptableExtra).append("\n");
|
||||
|
|
|
|||
|
|
@ -28,7 +28,6 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
|
|||
import org.apache.hadoop.yarn.api.records.NodeId;
|
||||
import org.apache.hadoop.yarn.api.records.Priority;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceInformation;
|
||||
import org.apache.hadoop.yarn.event.Dispatcher;
|
||||
import org.apache.hadoop.yarn.event.EventHandler;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
|
||||
|
|
@ -58,7 +57,6 @@ import org.apache.hadoop.yarn.util.Clock;
|
|||
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
|
||||
import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceUtils;
|
||||
import org.apache.hadoop.yarn.util.resource.Resources;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
|
|
@ -544,18 +542,6 @@ public class ProportionalCapacityPreemptionPolicyMockFramework {
|
|||
} else {
|
||||
res = Resources.createResource(Integer.valueOf(resource[0]),
|
||||
Integer.valueOf(resource[1]));
|
||||
if (resource.length > 2) {
|
||||
// Using the same order of resources from ResourceUtils, set resource
|
||||
// informations.
|
||||
ResourceInformation[] storedResourceInfo = ResourceUtils
|
||||
.getResourceTypesArray();
|
||||
for (int i = 2; i < resource.length; i++) {
|
||||
res.setResourceInformation(storedResourceInfo[i].getName(),
|
||||
ResourceInformation.newInstance(storedResourceInfo[i].getName(),
|
||||
storedResourceInfo[i].getUnits(),
|
||||
Integer.valueOf(resource[i])));
|
||||
}
|
||||
}
|
||||
}
|
||||
return res;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,17 +18,11 @@
|
|||
|
||||
package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
|
||||
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.ResourceTypes;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceInformation;
|
||||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor;
|
||||
import org.apache.hadoop.yarn.util.resource.ResourceUtils;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.mockito.Matchers.argThat;
|
||||
import static org.mockito.Mockito.never;
|
||||
|
|
@ -619,74 +613,4 @@ public class TestProportionalCapacityPreemptionPolicyForNodePartitions
|
|||
verify(mDisp, never()).handle(
|
||||
argThat(new IsPreemptionRequestFor(getAppAttemptId(3))));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNormalizeGuaranteeWithMultipleResource() throws IOException {
|
||||
// Initialize resource map
|
||||
Map<String, ResourceInformation> riMap = new HashMap<>();
|
||||
String RESOURCE_1 = "res1";
|
||||
|
||||
// Initialize mandatory resources
|
||||
ResourceInformation memory = ResourceInformation.newInstance(
|
||||
ResourceInformation.MEMORY_MB.getName(),
|
||||
ResourceInformation.MEMORY_MB.getUnits(),
|
||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
|
||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
|
||||
ResourceInformation vcores = ResourceInformation.newInstance(
|
||||
ResourceInformation.VCORES.getName(),
|
||||
ResourceInformation.VCORES.getUnits(),
|
||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES,
|
||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
|
||||
riMap.put(ResourceInformation.MEMORY_URI, memory);
|
||||
riMap.put(ResourceInformation.VCORES_URI, vcores);
|
||||
riMap.put(RESOURCE_1, ResourceInformation.newInstance(RESOURCE_1, "", 0,
|
||||
ResourceTypes.COUNTABLE, 0, Integer.MAX_VALUE));
|
||||
|
||||
ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
|
||||
|
||||
/**
|
||||
* Queue structure is:
|
||||
*
|
||||
* <pre>
|
||||
* root
|
||||
* / \
|
||||
* a b
|
||||
* / \ / \
|
||||
* a1 a2 b1 b2
|
||||
* </pre>
|
||||
*
|
||||
* a1 and b2 are using most of resources.
|
||||
* a2 and b1 needs more resources. Both are under served.
|
||||
* hence demand will consider both queue's need while trying to
|
||||
* do preemption.
|
||||
*/
|
||||
String labelsConfig =
|
||||
"=100,true;";
|
||||
String nodesConfig =
|
||||
"n1=;"; // n1 is default partition
|
||||
String queuesConfig =
|
||||
// guaranteed,max,used,pending
|
||||
"root(=[100:100:10 100:100:10 100:100:10 100:100:10]);" + //root
|
||||
"-a(=[50:80:4 100:100:10 80:90:10 30:20:4]);" + // a
|
||||
"--a1(=[25:30:2 100:50:10 80:90:10 0]);" + // a1
|
||||
"--a2(=[25:50:2 100:50:10 0 30:20:4]);" + // a2
|
||||
"-b(=[50:20:6 100:100:10 20:10 40:50:8]);" + // b
|
||||
"--b1(=[25:5:4 100:20:10 0 20:10:4]);" + // b1
|
||||
"--b2(=[25:15:2 100:20:10 20:10 20:10:4])"; // b2
|
||||
String appsConfig=
|
||||
//queueName\t(priority,resource,host,expression,#repeat,reserved)
|
||||
"a1\t" // app1 in a1
|
||||
+ "(1,8:9:1,n1,,10,false);" +
|
||||
"b2\t" // app2 in b2
|
||||
+ "(1,2:1,n1,,10,false)"; // 80 of y
|
||||
|
||||
buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
|
||||
policy.editSchedule();
|
||||
|
||||
verify(mDisp, times(7)).handle(
|
||||
argThat(new IsPreemptionRequestFor(getAppAttemptId(1))));
|
||||
|
||||
riMap.remove(RESOURCE_1);
|
||||
ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue