From d4e172fffcbffd7c550c261a786cf53bbf48e22f Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Wed, 24 Nov 2021 13:11:50 -0500 Subject: [PATCH 01/10] YARN-11005. Implement the core QUEUE_LENGTH_THEN_RESOURCES OContainer allocation policy * Implement the `QUEUE_LENGTH_THEN_RESOURCES` `LoadComparator` * Add `Resource` requested to select node methods when allocating `OPPORTUNISTIC` containers * Add node capacity and node allocated resources to `ClusterNode` * Extend `DominantResourceCalculator` to be able to compute min-share * Add tests and extend existing ones to consider 3 resource dimensions --- .../resource/DominantResourceCalculator.java | 37 ++- ...ortunisticContainerAllocatorAMService.java | 4 +- ...alizedOpportunisticContainerAllocator.java | 21 +- .../scheduler/distributed/ClusterNode.java | 34 +++ .../distributed/NodeQueueLoadMonitor.java | 214 +++++++++++--- .../distributed/TestNodeQueueLoadMonitor.java | 269 +++++++++++++++++- 6 files changed, 519 insertions(+), 60 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java index 135f3486cebae4..6707711d4c16a0 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java @@ -404,18 +404,51 @@ public boolean isAllInvalidDivisor(Resource r) { @Override public float ratio(Resource a, Resource b) { - float ratio = 0.0f; + return ratio(a, b, true); + } + + /** + * Computes the ratio of resource a over resource b, + * where the boolean flag {@literal isDominantShare} allows + * specification of whether the max- or min-share should be computed. + * @param a the numerator resource. + * @param b the denominator resource. + * @param isDominantShare whether the dominant (max) share should be computed, + * computes the min-share if false. + * @return the max- or min-share ratio of the resources. + */ + private float ratio(Resource a, Resource b, boolean isDominantShare) { + float ratio = isDominantShare ? 0.0f : 1.0f; int maxLength = ResourceUtils.getNumberOfCountableResourceTypes(); for (int i = 0; i < maxLength; i++) { ResourceInformation aResourceInformation = a.getResourceInformation(i); ResourceInformation bResourceInformation = b.getResourceInformation(i); final float tmp = divideSafelyAsFloat(aResourceInformation.getValue(), bResourceInformation.getValue()); - ratio = ratio > tmp ? ratio : tmp; + if (isDominantShare) { + ratio = Math.max(ratio, tmp); + } else { + ratio = Math.min(ratio, tmp); + } } return ratio; } + /** + * Computes the ratio of resource a over resource b. + * However, different from {@link this#ratio(Resource, Resource)}, + * this returns the min-share of the resources. + * For example, ratio(Resource(10, 50), Resource(100, 100)) would return 0.5, + * whereas minRatio(Resource(10, 50), Resource(100, 100)) would return 0.1. + * @param a the numerator resource. + * @param b the denominator resource. + * @return the min-share ratio of the resources. + */ + @Unstable + public float minRatio(Resource a, Resource b) { + return ratio(a, b, false); + } + @Override public Resource divideAndCeil(Resource numerator, int denominator) { return divideAndCeil(numerator, (long) denominator); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java index 01c9b93417e6b2..15b5c7f765976c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/OpportunisticContainerAllocatorAMService.java @@ -257,7 +257,9 @@ public OpportunisticContainerAllocatorAMService(RMContext rmContext, int limitMin, limitMax; - if (comparator == NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH) { + if (comparator == NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH || + comparator == + NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES) { limitMin = rmContext.getYarnConfiguration() .getInt(YarnConfiguration.NM_CONTAINER_QUEUING_MIN_QUEUE_LENGTH, YarnConfiguration. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java index 223e1a1bf30cb0..c01256573acdd3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java @@ -251,14 +251,15 @@ private List allocateNodeLocal( String userName, Map> allocations) throws YarnException { List allocatedContainers = new ArrayList<>(); + final ResourceRequest resourceRequest = enrichedAsk.getRequest(); while (toAllocate > 0) { RMNode node = nodeQueueLoadMonitor.selectLocalNode(nodeLocation, - blacklist); + blacklist, resourceRequest.getCapability()); if (node != null) { toAllocate--; Container container = createContainer(rmIdentifier, appParams, idCounter, id, userName, allocations, nodeLocation, - enrichedAsk.getRequest(), convertToRemoteNode(node)); + resourceRequest, convertToRemoteNode(node)); allocatedContainers.add(container); LOG.info("Allocated [{}] as opportunistic at location [{}]", container.getId(), nodeLocation); @@ -280,14 +281,15 @@ private List allocateRackLocal(EnrichedResourceRequest enrichedAsk, String userName, Map> allocations) throws YarnException { List allocatedContainers = new ArrayList<>(); + final ResourceRequest resourceRequest = enrichedAsk.getRequest(); while (toAllocate > 0) { RMNode node = nodeQueueLoadMonitor.selectRackLocalNode(rackLocation, - blacklist); + blacklist, resourceRequest.getCapability()); if (node != null) { toAllocate--; Container container = createContainer(rmIdentifier, appParams, idCounter, id, userName, allocations, rackLocation, - enrichedAsk.getRequest(), convertToRemoteNode(node)); + resourceRequest, convertToRemoteNode(node)); allocatedContainers.add(container); metrics.incrRackLocalOppContainers(); LOG.info("Allocated [{}] as opportunistic at location [{}]", @@ -309,17 +311,20 @@ private List allocateAny(EnrichedResourceRequest enrichedAsk, String userName, Map> allocations) throws YarnException { List allocatedContainers = new ArrayList<>(); + final ResourceRequest resourceRequest = enrichedAsk.getRequest(); while (toAllocate > 0) { - RMNode node = nodeQueueLoadMonitor.selectAnyNode(blacklist); + RMNode node = nodeQueueLoadMonitor.selectAnyNode( + blacklist, resourceRequest.getCapability()); if (node != null) { toAllocate--; Container container = createContainer(rmIdentifier, appParams, idCounter, id, userName, allocations, ResourceRequest.ANY, - enrichedAsk.getRequest(), convertToRemoteNode(node)); + resourceRequest, convertToRemoteNode(node)); allocatedContainers.add(container); metrics.incrOffSwitchOppContainers(); - LOG.info("Allocated [{}] as opportunistic at location [{}]", - container.getId(), ResourceRequest.ANY); + LOG.info( + "Allocated [{}] as opportunistic at location [{}] by OFF_SWITCH", + container.getId(), node.getNodeID()); } else { // we couldn't allocate any - break the loop. break; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index f92a92bb17c40a..84ea39314fb6c3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -22,6 +22,8 @@ import java.util.HashSet; import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.yarn.api.records.NodeId; +import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.util.resource.Resources; /** * Represents a node in the cluster from the NodeQueueLoadMonitor's perspective @@ -33,6 +35,8 @@ public class ClusterNode { final NodeId nodeId; private int queueCapacity = 0; private final HashSet labels; + private Resource capability = Resources.none(); + private Resource allocatedResource = Resources.none(); public ClusterNode(NodeId nodeId) { this.nodeId = nodeId; @@ -40,6 +44,36 @@ public ClusterNode(NodeId nodeId) { updateTimestamp(); } + public ClusterNode setCapability(Resource capability) { + if (capability == null) { + this.capability = Resources.none(); + } else { + this.capability = capability; + } + return this; + } + + public ClusterNode setAllocatedResource( + Resource allocatedResource) { + if (allocatedResource == null) { + this.allocatedResource = Resources.none(); + } else { + this.allocatedResource = allocatedResource; + } + return this; + } + + /** + * The latest allocated resource, as reported by node heartbeats. + */ + public Resource getAllocatedResource() { + return this.allocatedResource; + } + + public Resource getCapability() { + return this.capability; + } + public ClusterNode setQueueLength(int qLength) { this.queueLength.set(qLength); return this; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index 1e2b7ef6da4753..8610668f3cbfc4 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -18,8 +18,13 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.distributed; +import org.apache.commons.math3.util.Precision; import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.yarn.api.records.NodeState; +import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator; +import org.apache.hadoop.yarn.util.resource.ResourceCalculator; +import org.apache.hadoop.yarn.util.resource.Resources; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.yarn.api.records.NodeId; @@ -64,40 +69,165 @@ public class NodeQueueLoadMonitor implements ClusterMonitor { * of two Nodes are compared. */ public enum LoadComparator implements Comparator { + /** + * This policy only considers queue length. + * When allocating, increments queue length without looking at resources + * available on the node, and when sorting, also only sorts by queue length. + */ QUEUE_LENGTH, - QUEUE_WAIT_TIME; + /** + * This policy only considers the wait time of containers in the queue. + * Neither looks at resources nor at queue length. + */ + QUEUE_WAIT_TIME, + /** + * This policy considers both queue length and resources. + * When allocating, first decrements resources available on a node. + * If resources are available, does not place OContainers on the node queue. + * When sorting, it first sorts by queue length, + * then by available resources. + */ + QUEUE_LENGTH_THEN_RESOURCES; + + private Resource clusterResource = Resources.none(); + private final DominantResourceCalculator resourceCalculator = + new DominantResourceCalculator(); + + private Resource computeAvailableResource(final ClusterNode clusterNode) { + return Resources.subtractNonNegative( + clusterNode.getCapability(), + clusterNode.getAllocatedResource()); + } @Override public int compare(ClusterNode o1, ClusterNode o2) { - if (getMetric(o1) == getMetric(o2)) { - return (int)(o2.getTimestamp() - o1.getTimestamp()); + int diff = 0; + switch (this) { + case QUEUE_LENGTH_THEN_RESOURCES: + diff = o1.getQueueLength().get() - o2.getQueueLength().get(); + if (diff != 0) { + break; + } + + // NOTE: cluster resource should be + // set always before LoadComparator is used + final Resource availableResource1 = computeAvailableResource(o1); + final Resource availableResource2 = computeAvailableResource(o2); + final boolean isClusterResourceLeq0 = resourceCalculator + .isAnyMajorResourceZeroOrNegative(clusterResource); + if (!isClusterResourceLeq0) { + // Takes the least available resource of the two nodes, + // normalized to the overall cluster resource + final float availableRatio1 = + resourceCalculator.minRatio(availableResource1, clusterResource); + final float availableRatio2 = + resourceCalculator.minRatio(availableResource2, clusterResource); + + // The one with more available resources should be placed first + diff = Precision + .compareTo(availableRatio2, availableRatio1, Precision.EPSILON); + } + + if (diff == 0) { + // Compare absolute value if ratios are the same + diff = availableResource2.getVirtualCores() - availableResource1.getVirtualCores(); + } + + if (diff == 0) { + diff = Long.compare(availableResource2.getMemorySize(), + availableResource1.getMemorySize()); + } + break; + case QUEUE_WAIT_TIME: + case QUEUE_LENGTH: + default: + diff = getMetric(o1) - getMetric(o2); + break; + } + + if (diff == 0) { + return (int) (o2.getTimestamp() - o1.getTimestamp()); } - return getMetric(o1) - getMetric(o2); + return diff; + } + + public void setClusterResource(Resource clusterResource) { + this.clusterResource = clusterResource; + } + + public ResourceCalculator getResourceCalculator() { + return resourceCalculator; } public int getMetric(ClusterNode c) { - return (this == QUEUE_LENGTH) ? - c.getQueueLength().get() : c.getQueueWaitTime().get(); + switch (this) { + case QUEUE_WAIT_TIME: + return c.getQueueWaitTime().get(); + case QUEUE_LENGTH: + case QUEUE_LENGTH_THEN_RESOURCES: + default: + return c.getQueueLength().get(); + } } /** * Increment the metric by a delta if it is below the threshold. * @param c ClusterNode * @param incrementSize increment size + * @param requested the requested resource * @return true if the metric was below threshold and was incremented. */ - public boolean compareAndIncrement(ClusterNode c, int incrementSize) { - if(this == QUEUE_LENGTH) { - int ret = c.getQueueLength().addAndGet(incrementSize); - if (ret <= c.getQueueCapacity()) { + public boolean compareAndIncrement( + ClusterNode c, int incrementSize, Resource requested) { + if (this == QUEUE_LENGTH_THEN_RESOURCES) { + // Assignment and getting value is atomic + // Can be slightly inaccurate here, don't grab lock for performance + final Resource capability = c.getCapability(); + final Resource currAllocated = c.getAllocatedResource(); + final Resource currAvailable = Resources.subtractNonNegative( + capability, currAllocated); + if (resourceCalculator.fitsIn(requested, currAvailable)) { + final Resource newAllocated = Resources.add(currAllocated, requested); + c.setAllocatedResource(newAllocated); + return true; + } + + if (!resourceCalculator.fitsIn(requested, capability)) { + // If does not fit at all, do not allocate + return false; + } + } + if (this == QUEUE_LENGTH || this == QUEUE_LENGTH_THEN_RESOURCES) { + int beforeIncrement = c.getQueueLength().getAndAdd(incrementSize); + if (c.getQueueLength().get() <= c.getQueueCapacity()) { return true; } - c.getQueueLength().addAndGet(-incrementSize); + c.getQueueLength().set(beforeIncrement); return false; } // for queue wait time, we don't have any threshold. return true; } + + /** + * Whether we should be placing OContainers on a node. + * @param cn the clusterNode + * @return whether we should be placing OContainers on a node. + */ + public boolean isNodeAvailable(final ClusterNode cn) { + int queueCapacity = cn.getQueueCapacity(); + int queueLength = cn.getQueueLength().get(); + if (this == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES) { + if (queueCapacity <= 0) { + return queueLength <= 0; + } else { + return queueLength < queueCapacity; + } + } + // In the special case where queueCapacity is 0 for the node, + // the container can be allocated on the node but will be rejected there + return queueCapacity <= 0 || queueLength < queueCapacity; + } } private final ScheduledExecutorService scheduledExecutor; @@ -261,12 +391,15 @@ protected void onNewNodeAdded( if (rmNode.getState() != NodeState.DECOMMISSIONING && (estimatedQueueWaitTime != -1 || - comparator == LoadComparator.QUEUE_LENGTH)) { + comparator == LoadComparator.QUEUE_LENGTH || + comparator == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES)) { this.clusterNodes.put(rmNode.getNodeID(), new ClusterNode(rmNode.getNodeID()) .setQueueWaitTime(estimatedQueueWaitTime) .setQueueLength(waitQueueLength) .setNodeLabels(rmNode.getNodeLabels()) + .setCapability(rmNode.getTotalCapability()) + .setAllocatedResource(rmNode.getAllocatedContainerResource()) .setQueueCapacity(opportQueueCapacity)); LOG.info( "Inserting ClusterNode [{}] with queue wait time [{}] and " @@ -295,11 +428,14 @@ protected void onExistingNodeUpdated( if (rmNode.getState() != NodeState.DECOMMISSIONING && (estimatedQueueWaitTime != -1 || - comparator == LoadComparator.QUEUE_LENGTH)) { + comparator == LoadComparator.QUEUE_LENGTH || + comparator == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES)) { clusterNode .setQueueWaitTime(estimatedQueueWaitTime) .setQueueLength(waitQueueLength) .setNodeLabels(rmNode.getNodeLabels()) + .setCapability(rmNode.getTotalCapability()) + .setAllocatedResource(rmNode.getAllocatedContainerResource()) .updateTimestamp(); LOG.debug("Updating ClusterNode [{}] with queue wait time [{}] and" + " wait queue length [{}]", rmNode.getNodeID(), @@ -345,27 +481,31 @@ public List selectLeastLoadedNodes(int k) { } } - public RMNode selectLocalNode(String hostName, Set blacklist) { + public RMNode selectLocalNode( + String hostName, Set blacklist, Resource request) { if (blacklist.contains(hostName)) { return null; } RMNode node = nodeByHostName.get(hostName); if (node != null) { ClusterNode clusterNode = clusterNodes.get(node.getNodeID()); - if (comparator.compareAndIncrement(clusterNode, 1)) { + if (clusterNode != null && comparator + .compareAndIncrement(clusterNode, 1, request)) { return node; } } return null; } - public RMNode selectRackLocalNode(String rackName, Set blacklist) { + public RMNode selectRackLocalNode( + String rackName, Set blacklist, Resource request) { Set nodesOnRack = nodeIdsByRack.get(rackName); if (nodesOnRack != null) { for (NodeId nodeId : nodesOnRack) { if (!blacklist.contains(nodeId.getHost())) { ClusterNode node = clusterNodes.get(nodeId); - if (node != null && comparator.compareAndIncrement(node, 1)) { + if (node != null && + comparator.compareAndIncrement(node, 1, request)) { return nodeByHostName.get(nodeId.getHost()); } } @@ -374,7 +514,7 @@ public RMNode selectRackLocalNode(String rackName, Set blacklist) { return null; } - public RMNode selectAnyNode(Set blacklist) { + public RMNode selectAnyNode(Set blacklist, Resource request) { List nodeIds = getCandidatesForSelectAnyNode(); int size = nodeIds.size(); if (size <= 0) { @@ -388,7 +528,8 @@ public RMNode selectAnyNode(Set blacklist) { NodeId nodeId = nodeIds.get(index); if (nodeId != null && !blacklist.contains(nodeId.getHost())) { ClusterNode node = clusterNodes.get(nodeId); - if (node != null && comparator.compareAndIncrement(node, 1)) { + if (node != null && comparator.compareAndIncrement( + node, 1, request)) { return nodeByHostName.get(nodeId.getHost()); } } @@ -402,7 +543,10 @@ protected List getCandidatesForSelectAnyNode() { protected void removeFromNodeIdsByRack(RMNode removedNode) { nodeIdsByRack.computeIfPresent(removedNode.getRackName(), - (k, v) -> v).remove(removedNode.getNodeID()); + (k, v) -> { + v.remove(removedNode.getNodeID()); + return v; + }); } protected void addIntoNodeIdsByRack(RMNode addedNode) { @@ -414,21 +558,21 @@ protected List sortNodes(boolean excludeFullNodes) { ReentrantReadWriteLock.ReadLock readLock = clusterNodesLock.readLock(); readLock.lock(); try { - ArrayList aList = new ArrayList<>(this.clusterNodes.values()); - List retList = new ArrayList<>(); - Object[] nodes = aList.toArray(); - // Collections.sort would do something similar by calling Arrays.sort - // internally but would finally iterate through the input list (aList) - // to reset the value of each element. Since we don't really care about - // 'aList', we can use the iteration to create the list of nodeIds which - // is what we ultimately care about. - Arrays.sort(nodes, (Comparator)comparator); - for (int j=0; j < nodes.length; j++) { - ClusterNode cNode = (ClusterNode)nodes[j]; - // Only add node to the result list when either condition is met: - // 1. we don't exclude full nodes - // 2. we do exclude full nodes, but the current node is not full - if (!excludeFullNodes || !cNode.isQueueFull()) { + final ClusterNode[] nodes = new ClusterNode[clusterNodes.size()]; + int nodesIdx = 0; + final Resource clusterResource = Resource.newInstance(Resources.none()); + for (final ClusterNode node : this.clusterNodes.values()) { + Resources.addTo(clusterResource, node.getCapability()); + nodes[nodesIdx] = node; + nodesIdx++; + } + + comparator.setClusterResource(clusterResource); + + final List retList = new ArrayList<>(); + Arrays.sort(nodes, comparator); + for (final ClusterNode cNode : nodes) { + if (!excludeFullNodes || comparator.isNodeAvailable(cNode)) { retList.add(cNode); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java index eedd9de93bae46..1be2d44d08c2e7 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java @@ -18,17 +18,29 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.distributed; +import org.apache.hadoop.net.NodeBase; +import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.yarn.api.protocolrecords.ResourceTypes; import org.apache.hadoop.yarn.api.records.NodeId; import org.apache.hadoop.yarn.api.records.NodeState; +import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.api.records.ResourceInformation; +import org.apache.hadoop.yarn.api.records.ResourceUtilization; import org.apache.hadoop.yarn.server.api.records.ContainerQueuingLimit; import org.apache.hadoop.yarn.server.api.records.OpportunisticContainersStatus; import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode; +import org.apache.hadoop.yarn.util.resource.ResourceUtils; +import org.apache.hadoop.yarn.util.resource.Resources; import org.junit.Assert; +import org.junit.BeforeClass; import org.junit.Test; import org.mockito.Mockito; +import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; /** @@ -36,8 +48,18 @@ */ public class TestNodeQueueLoadMonitor { + // Extra resource type to test that all resource dimensions are considered + private static final String NETWORK_RESOURCE = "network"; private final static int DEFAULT_MAX_QUEUE_LENGTH = 200; + // Note: The following variables are private static resources + // re-initialized on each test because resource dimensions considered + // are initialized in a static method. + // Declaring them as static final will "lock-in" resource dimensions and + // disallow specification of a new resource dimension ("network") in tests. + private static Resource defaultResourceRequested; + private static Resource defaultCapacity; + static class FakeNodeId extends NodeId { final String host; final int port; @@ -70,6 +92,44 @@ public String toString() { } } + private static Resource newResourceInstance(long memory, int vCores) { + return newResourceInstance(memory, vCores, 0L); + } + + private static Resource newResourceInstance( + final long memory, final int vCores, final long network) { + return Resource.newInstance(memory, vCores, + ImmutableMap.of(NETWORK_RESOURCE, network)); + } + + private static long getNetworkResourceValue(final Resource resource) { + return resource.getResourceValue(NETWORK_RESOURCE); + } + + public static void addNewTypesToResources(String... resourceTypes) { + // Initialize resource map + Map riMap = new HashMap<>(); + + // Initialize mandatory resources + riMap.put(ResourceInformation.MEMORY_URI, ResourceInformation.MEMORY_MB); + riMap.put(ResourceInformation.VCORES_URI, ResourceInformation.VCORES); + + for (String newResource : resourceTypes) { + riMap.put(newResource, ResourceInformation + .newInstance(newResource, "", 0, ResourceTypes.COUNTABLE, 0, + Integer.MAX_VALUE)); + } + + ResourceUtils.initializeResourcesFromResourceInformationMap(riMap); + } + + @BeforeClass + public static void classSetUp() { + addNewTypesToResources(NETWORK_RESOURCE); + defaultResourceRequested = newResourceInstance(128, 1, 1); + defaultCapacity = newResourceInstance(1024, 8, 1000); + } + @Test public void testWaitTimeSort() { NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( @@ -79,7 +139,6 @@ public void testWaitTimeSort() { selector.updateNode(createRMNode("h3", 3, 10, 10)); selector.computeTask.run(); List nodeIds = selector.selectNodes(); - System.out.println("1-> " + nodeIds); Assert.assertEquals("h2:2", nodeIds.get(0).toString()); Assert.assertEquals("h3:3", nodeIds.get(1).toString()); Assert.assertEquals("h1:1", nodeIds.get(2).toString()); @@ -88,7 +147,6 @@ public void testWaitTimeSort() { selector.updateNode(createRMNode("h3", 3, 2, 10)); selector.computeTask.run(); nodeIds = selector.selectNodes(); - System.out.println("2-> "+ nodeIds); Assert.assertEquals("h3:3", nodeIds.get(0).toString()); Assert.assertEquals("h2:2", nodeIds.get(1).toString()); Assert.assertEquals("h1:1", nodeIds.get(2).toString()); @@ -97,7 +155,6 @@ public void testWaitTimeSort() { selector.updateNode(createRMNode("h4", 4, -1, 10)); selector.computeTask.run(); nodeIds = selector.selectNodes(); - System.out.println("3-> "+ nodeIds); // No change Assert.assertEquals("h3:3", nodeIds.get(0).toString()); Assert.assertEquals("h2:2", nodeIds.get(1).toString()); @@ -186,6 +243,152 @@ public void testQueueLengthSort() { Assert.assertEquals("h4:4", nodeIds.get(2).toString()); } + @Test + public void testQueueLengthThenResourcesSort() { + NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( + NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES); + + // Node and queue sizes were selected such that we can determine the + // order of these nodes in the selectNodes call deterministically + // h2 -> h1 -> h3 -> h4 + selector.updateNode(createRMNode( + "h1", 1, -1, 0, + Resources.multiply(defaultResourceRequested, 3), defaultCapacity)); + selector.updateNode(createRMNode( + "h2", 2, -1, 0, + Resources.multiply(defaultResourceRequested, 2), defaultCapacity)); + selector.updateNode(createRMNode( + "h3", 3, -1, 5, + Resources.multiply(defaultResourceRequested, 3), defaultCapacity)); + selector.updateNode(createRMNode( + "h4", 4, -1, 10, + Resources.multiply(defaultResourceRequested, 2), defaultCapacity)); + selector.computeTask.run(); + List nodeIds = selector.selectNodes(); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h1:1", nodeIds.get(1).toString()); + Assert.assertEquals("h3:3", nodeIds.get(2).toString()); + Assert.assertEquals("h4:4", nodeIds.get(3).toString()); + + // Now update node3 + // node3 should now rank after node4 since it has the same queue length + // but less resources available + selector.updateNode(createRMNode( + "h3", 3, -1, 10, + Resources.multiply(defaultResourceRequested, 3), defaultCapacity)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h1:1", nodeIds.get(1).toString()); + Assert.assertEquals("h4:4", nodeIds.get(2).toString()); + Assert.assertEquals("h3:3", nodeIds.get(3).toString()); + + // Now update h3 and fill its queue -- it should no longer be available + selector.updateNode(createRMNode("h3", 3, -1, + DEFAULT_MAX_QUEUE_LENGTH)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + // h3 is queued up, so we should only have 3 nodes left + Assert.assertEquals(3, nodeIds.size()); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h1:1", nodeIds.get(1).toString()); + Assert.assertEquals("h4:4", nodeIds.get(2).toString()); + + // Now update h2 to Decommissioning state + selector.updateNode(createRMNode("h2", 2, -1, + 5, NodeState.DECOMMISSIONING)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + // h2 is decommissioned, and h3 is full, so we should only have 2 nodes + Assert.assertEquals(2, nodeIds.size()); + Assert.assertEquals("h1:1", nodeIds.get(0).toString()); + Assert.assertEquals("h4:4", nodeIds.get(1).toString()); + + // Now update h2 back to Running state + selector.updateNode(createRMNode( + "h2", 2, -1, 0, + Resources.multiply(defaultResourceRequested, 2), defaultCapacity)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals(3, nodeIds.size()); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h1:1", nodeIds.get(1).toString()); + Assert.assertEquals("h4:4", nodeIds.get(2).toString()); + + // Now update h2 to have a zero queue capacity. + // Make sure that here it is still in the pool. + selector.updateNode(createRMNode( + "h2", 2, -1, 0, 0, + Resources.multiply(defaultResourceRequested, 2), + defaultCapacity)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals(3, nodeIds.size()); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h1:1", nodeIds.get(1).toString()); + Assert.assertEquals("h4:4", nodeIds.get(2).toString()); + + // Now update h2 to have a positive queue length but a zero queue capacity. + // Make sure that here it is no longer in the pool. + // Need to first remove the node, because node capacity is not updated. + selector.removeNode(createRMNode( + "h2", 2, -1, 0, 0, + Resources.multiply(defaultResourceRequested, 2), + defaultCapacity)); + selector.updateNode(createRMNode( + "h2", 2, -1, 1, 0, + Resources.multiply(defaultResourceRequested, 2), + defaultCapacity)); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals(2, nodeIds.size()); + Assert.assertEquals("h1:1", nodeIds.get(0).toString()); + Assert.assertEquals("h4:4", nodeIds.get(1).toString()); + } + + /** + * Tests that when using QUEUE_LENGTH_THEN_RESOURCES decrements the amount + * of resources on the internal {@link ClusterNode} representation. + */ + @Test + public void testQueueLengthThenResourcesDecrementsAvailable() { + NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( + NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES); + RMNode node = createRMNode("h1", 1, -1, 0); + selector.addNode(null, node); + selector.updateNode(node); + selector.updateSortedNodes(); + + ClusterNode clusterNode = selector.getClusterNodes().get(node.getNodeID()); + Assert.assertEquals(Resources.none(), + clusterNode.getAllocatedResource()); + + // Has enough resources + RMNode selectedNode = selector.selectAnyNode( + Collections.emptySet(), defaultResourceRequested); + Assert.assertNotNull(selectedNode); + Assert.assertEquals(node.getNodeID(), selectedNode.getNodeID()); + + clusterNode = selector.getClusterNodes().get(node.getNodeID()); + Assert.assertEquals(defaultResourceRequested, + clusterNode.getAllocatedResource()); + + // Does not have enough resources, but can queue + selectedNode = selector.selectAnyNode( + Collections.emptySet(), defaultCapacity); + Assert.assertNotNull(selectedNode); + Assert.assertEquals(node.getNodeID(), selectedNode.getNodeID()); + + clusterNode = selector.getClusterNodes().get(node.getNodeID()); + Assert.assertEquals(1, clusterNode.getQueueLength().get()); + + // Does not have enough resources and cannot queue + selectedNode = selector.selectAnyNode( + Collections.emptySet(), + Resources.add(defaultResourceRequested, defaultCapacity)); + Assert.assertNull(selectedNode); + } + @Test public void testContainerQueuingLimit() { NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( @@ -254,18 +457,22 @@ public void testSelectLocalNode() { // basic test for selecting node which has queue length less // than queue capacity. Set blacklist = new HashSet<>(); - RMNode node = selector.selectLocalNode("h1", blacklist); + RMNode node = selector.selectLocalNode( + "h1", blacklist, defaultResourceRequested); Assert.assertEquals("h1", node.getHostName()); // if node has been added to blacklist blacklist.add("h1"); - node = selector.selectLocalNode("h1", blacklist); + node = selector.selectLocalNode( + "h1", blacklist, defaultResourceRequested); Assert.assertNull(node); - node = selector.selectLocalNode("h2", blacklist); + node = selector.selectLocalNode( + "h2", blacklist, defaultResourceRequested); Assert.assertNull(node); - node = selector.selectLocalNode("h3", blacklist); + node = selector.selectLocalNode( + "h3", blacklist, defaultResourceRequested); Assert.assertEquals("h3", node.getHostName()); } @@ -293,19 +500,23 @@ public void testSelectRackLocalNode() { // basic test for selecting node which has queue length less // than queue capacity. Set blacklist = new HashSet<>(); - RMNode node = selector.selectRackLocalNode("rack1", blacklist); + RMNode node = selector.selectRackLocalNode( + "rack1", blacklist, defaultResourceRequested); Assert.assertEquals("h1", node.getHostName()); // if node has been added to blacklist blacklist.add("h1"); - node = selector.selectRackLocalNode("rack1", blacklist); + node = selector.selectRackLocalNode( + "rack1", blacklist, defaultResourceRequested); Assert.assertNull(node); - node = selector.selectRackLocalNode("rack2", blacklist); + node = selector.selectRackLocalNode( + "rack2", blacklist, defaultResourceRequested); Assert.assertEquals("h3", node.getHostName()); blacklist.add("h3"); - node = selector.selectRackLocalNode("rack2", blacklist); + node = selector.selectRackLocalNode( + "rack2", blacklist, defaultResourceRequested); Assert.assertNull(node); } @@ -337,17 +548,17 @@ public void testSelectAnyNode() { // basic test for selecting node which has queue length // less than queue capacity. Set blacklist = new HashSet<>(); - RMNode node = selector.selectAnyNode(blacklist); + RMNode node = selector.selectAnyNode(blacklist, defaultResourceRequested); Assert.assertTrue(node.getHostName().equals("h1") || node.getHostName().equals("h3")); // if node has been added to blacklist blacklist.add("h1"); - node = selector.selectAnyNode(blacklist); + node = selector.selectAnyNode(blacklist, defaultResourceRequested); Assert.assertEquals("h3", node.getHostName()); blacklist.add("h3"); - node = selector.selectAnyNode(blacklist); + node = selector.selectAnyNode(blacklist, defaultResourceRequested); Assert.assertNull(node); } @@ -377,12 +588,42 @@ private RMNode createRMNode(String host, int port, String rack, private RMNode createRMNode(String host, int port, String rack, int waitTime, int queueLength, int queueCapacity, NodeState state) { + return createRMNode(host, port, rack, waitTime, queueLength, queueCapacity, + state, ResourceUtilization.newInstance(0, 0, 0), + Resources.none(), defaultCapacity); + } + + private RMNode createRMNode( + String host, int port, int waitTime, int queueLength, + Resource allocatedResource, Resource nodeResource) { + return createRMNode(host, port, waitTime, queueLength, + DEFAULT_MAX_QUEUE_LENGTH, allocatedResource, nodeResource); + } + + private RMNode createRMNode( + String host, int port, int waitTime, int queueLength, int queueCapacity, + Resource allocatedResource, Resource nodeResource) { + return createRMNode(host, port, "default", waitTime, queueLength, + queueCapacity, NodeState.RUNNING, + ResourceUtilization.newInstance(0, 0, 0), + allocatedResource, nodeResource); + } + + private RMNode createRMNode(String host, int port, String rack, + int waitTime, int queueLength, int queueCapacity, NodeState state, + ResourceUtilization utilization, Resource allocatedResource, + Resource nodeResource) { RMNode node1 = Mockito.mock(RMNode.class); NodeId nID1 = new FakeNodeId(host, port); Mockito.when(node1.getHostName()).thenReturn(host); Mockito.when(node1.getRackName()).thenReturn(rack); + Mockito.when(node1.getNode()).thenReturn(new NodeBase("/" + host)); Mockito.when(node1.getNodeID()).thenReturn(nID1); Mockito.when(node1.getState()).thenReturn(state); + Mockito.when(node1.getTotalCapability()).thenReturn(nodeResource); + Mockito.when(node1.getNodeUtilization()).thenReturn(utilization); + Mockito.when(node1.getAllocatedContainerResource()).thenReturn( + allocatedResource); OpportunisticContainersStatus status1 = Mockito.mock(OpportunisticContainersStatus.class); Mockito.when(status1.getEstimatedQueueWaitTime()) From 98da8de01d912e092c2b482f4f52d4634e1852bc Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Thu, 25 Nov 2021 10:45:39 -0500 Subject: [PATCH 02/10] Remove unneeded comment --- .../resourcemanager/scheduler/distributed/ClusterNode.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 84ea39314fb6c3..93774b92f79806 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -63,9 +63,6 @@ public ClusterNode setAllocatedResource( return this; } - /** - * The latest allocated resource, as reported by node heartbeats. - */ public Resource getAllocatedResource() { return this.allocatedResource; } From b9939ed23f8d32b0c02271efaa15ae7a9b5e6cd5 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Thu, 25 Nov 2021 23:27:03 -0500 Subject: [PATCH 03/10] Fix checkstyle, javadoc, and findbugs errors --- .../util/resource/DominantResourceCalculator.java | 2 +- .../scheduler/distributed/ClusterNode.java | 12 ++++++------ .../distributed/NodeQueueLoadMonitor.java | 2 +- .../distributed/TestNodeQueueLoadMonitor.java | 14 ++++++-------- 4 files changed, 14 insertions(+), 16 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java index 6707711d4c16a0..8fe9c7931a2894 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java @@ -436,7 +436,7 @@ private float ratio(Resource a, Resource b, boolean isDominantShare) { /** * Computes the ratio of resource a over resource b. - * However, different from {@link this#ratio(Resource, Resource)}, + * However, different from ratio(Resource, Resource), * this returns the min-share of the resources. * For example, ratio(Resource(10, 50), Resource(100, 100)) would return 0.5, * whereas minRatio(Resource(10, 50), Resource(100, 100)) would return 0.1. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 93774b92f79806..5bef4637aeda78 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -44,21 +44,21 @@ public ClusterNode(NodeId nodeId) { updateTimestamp(); } - public ClusterNode setCapability(Resource capability) { - if (capability == null) { + public ClusterNode setCapability(Resource nodeCapability) { + if (nodeCapability == null) { this.capability = Resources.none(); } else { - this.capability = capability; + this.capability = nodeCapability; } return this; } public ClusterNode setAllocatedResource( - Resource allocatedResource) { - if (allocatedResource == null) { + Resource allocResource) { + if (allocResource == null) { this.allocatedResource = Resources.none(); } else { - this.allocatedResource = allocatedResource; + this.allocatedResource = allocResource; } return this; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index 8610668f3cbfc4..116c88dff014e1 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -151,7 +151,7 @@ public int compare(ClusterNode o1, ClusterNode o2) { return diff; } - public void setClusterResource(Resource clusterResource) { + private void setClusterResource(Resource clusterResource) { this.clusterResource = clusterResource; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java index 1be2d44d08c2e7..d06d12ce41f898 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java @@ -589,8 +589,7 @@ private RMNode createRMNode(String host, int port, String rack, private RMNode createRMNode(String host, int port, String rack, int waitTime, int queueLength, int queueCapacity, NodeState state) { return createRMNode(host, port, rack, waitTime, queueLength, queueCapacity, - state, ResourceUtilization.newInstance(0, 0, 0), - Resources.none(), defaultCapacity); + state, Resources.none(), defaultCapacity); } private RMNode createRMNode( @@ -604,15 +603,13 @@ private RMNode createRMNode( String host, int port, int waitTime, int queueLength, int queueCapacity, Resource allocatedResource, Resource nodeResource) { return createRMNode(host, port, "default", waitTime, queueLength, - queueCapacity, NodeState.RUNNING, - ResourceUtilization.newInstance(0, 0, 0), - allocatedResource, nodeResource); + queueCapacity, NodeState.RUNNING, allocatedResource, nodeResource); } + @SuppressWarnings("parameternumber") private RMNode createRMNode(String host, int port, String rack, int waitTime, int queueLength, int queueCapacity, NodeState state, - ResourceUtilization utilization, Resource allocatedResource, - Resource nodeResource) { + Resource allocatedResource, Resource nodeResource) { RMNode node1 = Mockito.mock(RMNode.class); NodeId nID1 = new FakeNodeId(host, port); Mockito.when(node1.getHostName()).thenReturn(host); @@ -621,7 +618,8 @@ private RMNode createRMNode(String host, int port, String rack, Mockito.when(node1.getNodeID()).thenReturn(nID1); Mockito.when(node1.getState()).thenReturn(state); Mockito.when(node1.getTotalCapability()).thenReturn(nodeResource); - Mockito.when(node1.getNodeUtilization()).thenReturn(utilization); + Mockito.when(node1.getNodeUtilization()).thenReturn( + ResourceUtilization.newInstance(0, 0, 0)); Mockito.when(node1.getAllocatedContainerResource()).thenReturn( allocatedResource); OpportunisticContainersStatus status1 = From fb36f73b50e128fbc49cdb096bdd07858a29f300 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Mon, 29 Nov 2021 09:39:35 -0500 Subject: [PATCH 04/10] Revert to old comment --- .../CentralizedOpportunisticContainerAllocator.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java index c01256573acdd3..482612de250ab5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/CentralizedOpportunisticContainerAllocator.java @@ -322,9 +322,8 @@ private List allocateAny(EnrichedResourceRequest enrichedAsk, resourceRequest, convertToRemoteNode(node)); allocatedContainers.add(container); metrics.incrOffSwitchOppContainers(); - LOG.info( - "Allocated [{}] as opportunistic at location [{}] by OFF_SWITCH", - container.getId(), node.getNodeID()); + LOG.info("Allocated [{}] as opportunistic at location [{}]", + container.getId(), ResourceRequest.ANY); } else { // we couldn't allocate any - break the loop. break; From d422e60e036ee26685459be3662853a2f5ddb894 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Thu, 2 Dec 2021 09:59:11 -0500 Subject: [PATCH 05/10] Add test for node resource capability changes --- .../distributed/TestNodeQueueLoadMonitor.java | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java index d06d12ce41f898..aeb51c4d8d5fc5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java @@ -389,6 +389,62 @@ public void testQueueLengthThenResourcesDecrementsAvailable() { Assert.assertNull(selectedNode); } + @Test + public void testQueueLengthThenResourcesCapabilityChange() { + NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( + NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES); + + // Node sizes were selected such that we can determine the + // order of these nodes in the selectNodes call deterministically + // h1 -> h2 -> h3 -> h4 + selector.updateNode(createRMNode( + "h1", 1, -1, 0, + Resources.multiply(defaultResourceRequested, 1), defaultCapacity)); + selector.updateNode(createRMNode( + "h2", 2, -1, 0, + Resources.multiply(defaultResourceRequested, 2), defaultCapacity)); + selector.updateNode(createRMNode( + "h3", 3, -1, 0, + Resources.multiply(defaultResourceRequested, 3), defaultCapacity)); + selector.updateNode(createRMNode( + "h4", 4, -1, 0, + Resources.multiply(defaultResourceRequested, 4), defaultCapacity)); + selector.computeTask.run(); + List nodeIds = selector.selectNodes(); + Assert.assertEquals("h1:1", nodeIds.get(0).toString()); + Assert.assertEquals("h2:2", nodeIds.get(1).toString()); + Assert.assertEquals("h3:3", nodeIds.get(2).toString()); + Assert.assertEquals("h4:4", nodeIds.get(3).toString()); + + // Now update node1 to have only defaultResourceRequested available + // by changing its capability to 2x defaultResourceReqeusted + // node1 should now rank last + selector.updateNode(createRMNode( + "h1", 1, -1, 0, + Resources.multiply(defaultResourceRequested, 1), + Resources.multiply(defaultResourceRequested, 2))); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals("h2:2", nodeIds.get(0).toString()); + Assert.assertEquals("h3:3", nodeIds.get(1).toString()); + Assert.assertEquals("h4:4", nodeIds.get(2).toString()); + Assert.assertEquals("h1:1", nodeIds.get(3).toString()); + + // Now update node2 to have no resources available + // by changing its capability to 1x defaultResourceReqeusted + // node2 should now rank last + selector.updateNode(createRMNode( + "h2", 2, -1, 0, + Resources.multiply(defaultResourceRequested, 1), + Resources.multiply(defaultResourceRequested, 1))); + selector.computeTask.run(); + nodeIds = selector.selectNodes(); + Assert.assertEquals("h3:3", nodeIds.get(0).toString()); + Assert.assertEquals("h4:4", nodeIds.get(1).toString()); + Assert.assertEquals("h1:1", nodeIds.get(2).toString()); + Assert.assertEquals("h2:2", nodeIds.get(3).toString()); + } + @Test public void testContainerQueuingLimit() { NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor( From e9e069e073d90310c7dbdd81aa0051ae1b532a29 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Fri, 3 Dec 2021 09:53:18 -0500 Subject: [PATCH 06/10] Use read and write locks to ensure correctness in ClusterNode --- .../scheduler/distributed/ClusterNode.java | 204 +++++++++++++++--- .../distributed/NodeQueueLoadMonitor.java | 55 ++--- .../distributed/TestNodeQueueLoadMonitor.java | 2 +- 3 files changed, 186 insertions(+), 75 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 5bef4637aeda78..247543d7c6a332 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -20,105 +20,239 @@ import java.util.Collection; import java.util.HashSet; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; + import org.apache.hadoop.yarn.api.records.NodeId; import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.util.resource.ResourceCalculator; import org.apache.hadoop.yarn.util.resource.Resources; /** * Represents a node in the cluster from the NodeQueueLoadMonitor's perspective */ public class ClusterNode { - private final AtomicInteger queueLength = new AtomicInteger(0); - private final AtomicInteger queueWaitTime = new AtomicInteger(-1); + private int queueLength = 0; + private int queueWaitTime = -1; private long timestamp; final NodeId nodeId; private int queueCapacity = 0; private final HashSet labels; private Resource capability = Resources.none(); private Resource allocatedResource = Resources.none(); + private final ReentrantReadWriteLock.WriteLock writeLock; + private final ReentrantReadWriteLock.ReadLock readLock; public ClusterNode(NodeId nodeId) { this.nodeId = nodeId; this.labels = new HashSet<>(); + final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + writeLock = lock.writeLock(); + readLock = lock.readLock(); updateTimestamp(); } public ClusterNode setCapability(Resource nodeCapability) { - if (nodeCapability == null) { - this.capability = Resources.none(); - } else { - this.capability = nodeCapability; + writeLock.lock(); + try { + if (nodeCapability == null) { + this.capability = Resources.none(); + } else { + this.capability = nodeCapability; + } + return this; + } + finally { + writeLock.unlock(); } - return this; } public ClusterNode setAllocatedResource( Resource allocResource) { - if (allocResource == null) { - this.allocatedResource = Resources.none(); - } else { - this.allocatedResource = allocResource; + writeLock.lock(); + try { + if (allocResource == null) { + this.allocatedResource = Resources.none(); + } else { + this.allocatedResource = allocResource; + } + return this; + } + finally { + writeLock.unlock(); } - return this; } public Resource getAllocatedResource() { - return this.allocatedResource; + readLock.lock(); + try { + return this.allocatedResource; + } + finally { + readLock.unlock(); + } + } + + public Resource getAvailableResource() { + readLock.lock(); + try { + return Resources.subtractNonNegative(capability, allocatedResource); + } finally { + readLock.unlock(); + } } public Resource getCapability() { - return this.capability; + readLock.lock(); + try { + return this.capability; + } + finally { + readLock.unlock(); + } } public ClusterNode setQueueLength(int qLength) { - this.queueLength.set(qLength); - return this; + writeLock.lock(); + try { + this.queueLength = qLength; + return this; + } finally { + writeLock.unlock(); + } } public ClusterNode setQueueWaitTime(int wTime) { - this.queueWaitTime.set(wTime); - return this; + writeLock.lock(); + try { + this.queueWaitTime = wTime; + return this; + } finally { + writeLock.unlock(); + } } public ClusterNode updateTimestamp() { - this.timestamp = System.currentTimeMillis(); - return this; + writeLock.lock(); + try { + this.timestamp = System.currentTimeMillis(); + return this; + } finally { + writeLock.unlock(); + } } public ClusterNode setQueueCapacity(int capacity) { - this.queueCapacity = capacity; - return this; + writeLock.lock(); + try { + this.queueCapacity = capacity; + return this; + } finally { + writeLock.unlock(); + } } public ClusterNode setNodeLabels(Collection labelsToAdd) { - labels.clear(); - labels.addAll(labelsToAdd); - return this; + writeLock.lock(); + try { + labels.clear(); + labels.addAll(labelsToAdd); + return this; + } finally { + writeLock.unlock(); + } } public boolean hasLabel(String label) { - return this.labels.contains(label); + readLock.lock(); + try { + return this.labels.contains(label); + } finally { + readLock.unlock(); + } } public long getTimestamp() { - return this.timestamp; + readLock.lock(); + try { + return this.timestamp; + } finally { + readLock.unlock(); + } } - public AtomicInteger getQueueLength() { - return this.queueLength; + public int getQueueLength() { + readLock.lock(); + try { + return this.queueLength; + } finally { + readLock.unlock(); + } } - public AtomicInteger getQueueWaitTime() { - return this.queueWaitTime; + public int getQueueWaitTime() { + readLock.lock(); + try { + return this.queueWaitTime; + } finally { + readLock.unlock(); + } } public int getQueueCapacity() { - return this.queueCapacity; + readLock.lock(); + try { + return this.queueCapacity; + } finally { + readLock.unlock(); + } + } + + public boolean compareAndIncrementAllocation( + final int incrementQLen, + final ResourceCalculator resourceCalculator, + final Resource requested) { + writeLock.lock(); + try { + final Resource currAvailable = Resources.subtractNonNegative( + capability, allocatedResource); + if (resourceCalculator.fitsIn(requested, currAvailable)) { + allocatedResource = Resources.add(allocatedResource, requested); + return true; + } + + if (!resourceCalculator.fitsIn(requested, capability)) { + // If does not fit at all, do not allocate + return false; + } + + return compareAndIncrementAllocation(incrementQLen); + } finally { + writeLock.unlock(); + } + } + + public boolean compareAndIncrementAllocation( + final int incrementQLen) { + writeLock.lock(); + try { + final int added = queueLength + incrementQLen; + if (added <= queueCapacity) { + queueLength = added; + return true; + } + return false; + } finally { + writeLock.unlock(); + } } public boolean isQueueFull() { - return this.queueCapacity > 0 && - this.queueLength.get() >= this.queueCapacity; + readLock.lock(); + try { + return this.queueCapacity > 0 && + this.queueLength >= this.queueCapacity; + } finally { + readLock.unlock(); + } } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index 116c88dff014e1..aaed3babaf7f43 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -93,26 +93,20 @@ public enum LoadComparator implements Comparator { private final DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); - private Resource computeAvailableResource(final ClusterNode clusterNode) { - return Resources.subtractNonNegative( - clusterNode.getCapability(), - clusterNode.getAllocatedResource()); - } - @Override public int compare(ClusterNode o1, ClusterNode o2) { int diff = 0; switch (this) { case QUEUE_LENGTH_THEN_RESOURCES: - diff = o1.getQueueLength().get() - o2.getQueueLength().get(); + diff = o1.getQueueLength() - o2.getQueueLength(); if (diff != 0) { break; } // NOTE: cluster resource should be // set always before LoadComparator is used - final Resource availableResource1 = computeAvailableResource(o1); - final Resource availableResource2 = computeAvailableResource(o2); + final Resource availableResource1 = o1.getAvailableResource(); + final Resource availableResource2 = o2.getAvailableResource(); final boolean isClusterResourceLeq0 = resourceCalculator .isAnyMajorResourceZeroOrNegative(clusterResource); if (!isClusterResourceLeq0) { @@ -162,11 +156,11 @@ public ResourceCalculator getResourceCalculator() { public int getMetric(ClusterNode c) { switch (this) { case QUEUE_WAIT_TIME: - return c.getQueueWaitTime().get(); + return c.getQueueWaitTime(); case QUEUE_LENGTH: case QUEUE_LENGTH_THEN_RESOURCES: default: - return c.getQueueLength().get(); + return c.getQueueLength(); } } @@ -179,34 +173,17 @@ public int getMetric(ClusterNode c) { */ public boolean compareAndIncrement( ClusterNode c, int incrementSize, Resource requested) { - if (this == QUEUE_LENGTH_THEN_RESOURCES) { - // Assignment and getting value is atomic - // Can be slightly inaccurate here, don't grab lock for performance - final Resource capability = c.getCapability(); - final Resource currAllocated = c.getAllocatedResource(); - final Resource currAvailable = Resources.subtractNonNegative( - capability, currAllocated); - if (resourceCalculator.fitsIn(requested, currAvailable)) { - final Resource newAllocated = Resources.add(currAllocated, requested); - c.setAllocatedResource(newAllocated); - return true; - } - - if (!resourceCalculator.fitsIn(requested, capability)) { - // If does not fit at all, do not allocate - return false; - } - } - if (this == QUEUE_LENGTH || this == QUEUE_LENGTH_THEN_RESOURCES) { - int beforeIncrement = c.getQueueLength().getAndAdd(incrementSize); - if (c.getQueueLength().get() <= c.getQueueCapacity()) { - return true; - } - c.getQueueLength().set(beforeIncrement); - return false; + switch (this) { + case QUEUE_LENGTH_THEN_RESOURCES: + return c.compareAndIncrementAllocation( + incrementSize, resourceCalculator, requested); + case QUEUE_WAIT_TIME: + // for queue wait time, we don't have any threshold. + return true; + case QUEUE_LENGTH: + default: + return c.compareAndIncrementAllocation(incrementSize); } - // for queue wait time, we don't have any threshold. - return true; } /** @@ -216,7 +193,7 @@ public boolean compareAndIncrement( */ public boolean isNodeAvailable(final ClusterNode cn) { int queueCapacity = cn.getQueueCapacity(); - int queueLength = cn.getQueueLength().get(); + int queueLength = cn.getQueueLength(); if (this == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES) { if (queueCapacity <= 0) { return queueLength <= 0; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java index aeb51c4d8d5fc5..83c0226870d8cd 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java @@ -380,7 +380,7 @@ public void testQueueLengthThenResourcesDecrementsAvailable() { Assert.assertEquals(node.getNodeID(), selectedNode.getNodeID()); clusterNode = selector.getClusterNodes().get(node.getNodeID()); - Assert.assertEquals(1, clusterNode.getQueueLength().get()); + Assert.assertEquals(1, clusterNode.getQueueLength()); // Does not have enough resources and cannot queue selectedNode = selector.selectAnyNode( From 332066afbe42d0887749a504dfa519e5397482cb Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Fri, 3 Dec 2021 14:03:43 -0500 Subject: [PATCH 07/10] Extract compare function, set condition for using min ratio computation, add javadocs, formatting issues --- .../scheduler/distributed/ClusterNode.java | 12 +- .../distributed/NodeQueueLoadMonitor.java | 114 ++++++++++++------ 2 files changed, 84 insertions(+), 42 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 247543d7c6a332..076885ff16dcf5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -60,8 +60,7 @@ public ClusterNode setCapability(Resource nodeCapability) { this.capability = nodeCapability; } return this; - } - finally { + } finally { writeLock.unlock(); } } @@ -76,8 +75,7 @@ public ClusterNode setAllocatedResource( this.allocatedResource = allocResource; } return this; - } - finally { + } finally { writeLock.unlock(); } } @@ -86,8 +84,7 @@ public Resource getAllocatedResource() { readLock.lock(); try { return this.allocatedResource; - } - finally { + } finally { readLock.unlock(); } } @@ -105,8 +102,7 @@ public Resource getCapability() { readLock.lock(); try { return this.capability; - } - finally { + } finally { readLock.unlock(); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index aaed3babaf7f43..843fb4bd88f504 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -93,44 +93,67 @@ public enum LoadComparator implements Comparator { private final DominantResourceCalculator resourceCalculator = new DominantResourceCalculator(); + private boolean shouldPerformMinRatioComputation() { + if (clusterResource == null) { + return false; + } + + return !resourceCalculator.isAnyMajorResourceZeroOrNegative( + clusterResource); + } + + /** + * Compares queue length of nodes first (shortest first), + * then compares available resources normalized + * over cluster resources (most available resources first). + * @param o1 the first ClusterNode + * @param o2 the second ClusterNode + * @return the difference the two ClusterNodes for sorting + */ + private int compareQueueLengthThenResources( + final ClusterNode o1, final ClusterNode o2) { + int diff = o1.getQueueLength() - o2.getQueueLength(); + if (diff != 0) { + return diff; + } + + final Resource availableResource1 = o1.getAvailableResource(); + final Resource availableResource2 = o2.getAvailableResource(); + + // Cluster resource should be valid before performing min-ratio logic + // Use raw available resource comparison otherwise + if (shouldPerformMinRatioComputation()) { + // Takes the least available resource of the two nodes, + // normalized to the overall cluster resource + final float availableRatio1 = + resourceCalculator.minRatio(availableResource1, clusterResource); + final float availableRatio2 = + resourceCalculator.minRatio(availableResource2, clusterResource); + + // The one with more available resources should be placed first + diff = Precision.compareTo( + availableRatio2, availableRatio1, Precision.EPSILON); + } + + if (diff == 0) { + // Compare absolute value if ratios are the same + diff = availableResource2.getVirtualCores() - availableResource1.getVirtualCores(); + } + + if (diff == 0) { + diff = Long.compare(availableResource2.getMemorySize(), + availableResource1.getMemorySize()); + } + + return diff; + } + @Override public int compare(ClusterNode o1, ClusterNode o2) { - int diff = 0; + int diff; switch (this) { case QUEUE_LENGTH_THEN_RESOURCES: - diff = o1.getQueueLength() - o2.getQueueLength(); - if (diff != 0) { - break; - } - - // NOTE: cluster resource should be - // set always before LoadComparator is used - final Resource availableResource1 = o1.getAvailableResource(); - final Resource availableResource2 = o2.getAvailableResource(); - final boolean isClusterResourceLeq0 = resourceCalculator - .isAnyMajorResourceZeroOrNegative(clusterResource); - if (!isClusterResourceLeq0) { - // Takes the least available resource of the two nodes, - // normalized to the overall cluster resource - final float availableRatio1 = - resourceCalculator.minRatio(availableResource1, clusterResource); - final float availableRatio2 = - resourceCalculator.minRatio(availableResource2, clusterResource); - - // The one with more available resources should be placed first - diff = Precision - .compareTo(availableRatio2, availableRatio1, Precision.EPSILON); - } - - if (diff == 0) { - // Compare absolute value if ratios are the same - diff = availableResource2.getVirtualCores() - availableResource1.getVirtualCores(); - } - - if (diff == 0) { - diff = Long.compare(availableResource2.getMemorySize(), - availableResource1.getMemorySize()); - } + diff = compareQueueLengthThenResources(o1, o2); break; case QUEUE_WAIT_TIME: case QUEUE_LENGTH: @@ -458,6 +481,14 @@ public List selectLeastLoadedNodes(int k) { } } + /** + * Selects the node as specified by hostName for resource allocation, + * unless the node has been blacklisted. + * @param hostName the hostname of the node for local resource allocation + * @param blacklist the blacklisted nodes + * @param request the requested resource + * @return the selected node, null if the node is full or is blacklisted + */ public RMNode selectLocalNode( String hostName, Set blacklist, Resource request) { if (blacklist.contains(hostName)) { @@ -474,6 +505,14 @@ public RMNode selectLocalNode( return null; } + /** + * Selects a node from the rack as specified by rackName + * for resource allocation, excluding blacklisted nodes + * @param rackName the rack name for rack-local resource allocation + * @param blacklist the blacklisted nodes + * @param request the requested resource + * @return the selected node, null if no suitable nodes + */ public RMNode selectRackLocalNode( String rackName, Set blacklist, Resource request) { Set nodesOnRack = nodeIdsByRack.get(rackName); @@ -491,6 +530,13 @@ public RMNode selectRackLocalNode( return null; } + /** + * Selects a node from all ClusterNodes for resource allocation, + * excluding blacklisted nodes. + * @param blacklist the blacklisted nodes + * @param request the requested resource + * @return the selected node, null if no suitable nodes + */ public RMNode selectAnyNode(Set blacklist, Resource request) { List nodeIds = getCandidatesForSelectAnyNode(); int size = nodeIds.size(); From e345582149162d9399af5c170130d3a24bee09b1 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Mon, 6 Dec 2021 09:31:41 -0500 Subject: [PATCH 08/10] Fit compareAndIncrement to one single line --- .../resourcemanager/scheduler/distributed/ClusterNode.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 076885ff16dcf5..754c7407e61b94 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -227,8 +227,7 @@ public boolean compareAndIncrementAllocation( } } - public boolean compareAndIncrementAllocation( - final int incrementQLen) { + public boolean compareAndIncrementAllocation(final int incrementQLen) { writeLock.lock(); try { final int added = queueLength + incrementQLen; From 63f11d695c16c064d268f8ad8de25952ba1fd3d6 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Tue, 7 Dec 2021 10:37:55 -0500 Subject: [PATCH 09/10] Set properties in one shot --- .../scheduler/distributed/ClusterNode.java | 145 ++++++++++-------- .../distributed/NodeQueueLoadMonitor.java | 29 ++-- 2 files changed, 96 insertions(+), 78 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java index 754c7407e61b94..065f97c4ce23e2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/ClusterNode.java @@ -31,6 +31,65 @@ * Represents a node in the cluster from the NodeQueueLoadMonitor's perspective */ public class ClusterNode { + /** + * Properties class used to initialize/change fields in ClusterNode. + */ + public static final class Properties { + private int queueLength = 0; + private int queueWaitTime = -1; + private long timestamp; + private int queueCapacity = 0; + private boolean queueCapacityIsSet = false; + private final HashSet labels; + private Resource capability = null; + private Resource allocatedResource = null; + + public static Properties newInstance() { + return new Properties(); + } + + Properties setQueueLength(int qLength) { + this.queueLength = qLength; + return this; + } + + Properties setQueueWaitTime(int wTime) { + this.queueWaitTime = wTime; + return this; + } + + Properties updateTimestamp() { + this.timestamp = System.currentTimeMillis(); + return this; + } + + Properties setQueueCapacity(int capacity) { + this.queueCapacity = capacity; + this.queueCapacityIsSet = true; + return this; + } + + Properties setNodeLabels(Collection labelsToAdd) { + labels.clear(); + labels.addAll(labelsToAdd); + return this; + } + + Properties setCapability(Resource nodeCapability) { + this.capability = nodeCapability; + return this; + } + + Properties setAllocatedResource(Resource allocResource) { + this.allocatedResource = allocResource; + return this; + } + + private Properties() { + labels = new HashSet<>(); + } + } + private int queueLength = 0; private int queueWaitTime = -1; private long timestamp; @@ -46,34 +105,35 @@ public ClusterNode(NodeId nodeId) { this.nodeId = nodeId; this.labels = new HashSet<>(); final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); - writeLock = lock.writeLock(); - readLock = lock.readLock(); - updateTimestamp(); + this.writeLock = lock.writeLock(); + this.readLock = lock.readLock(); + this.timestamp = System.currentTimeMillis(); } - public ClusterNode setCapability(Resource nodeCapability) { + public ClusterNode setProperties(final Properties properties) { writeLock.lock(); try { - if (nodeCapability == null) { + if (properties.capability == null) { this.capability = Resources.none(); } else { - this.capability = nodeCapability; + this.capability = properties.capability; } - return this; - } finally { - writeLock.unlock(); - } - } - public ClusterNode setAllocatedResource( - Resource allocResource) { - writeLock.lock(); - try { - if (allocResource == null) { + if (properties.allocatedResource == null) { this.allocatedResource = Resources.none(); } else { - this.allocatedResource = allocResource; + this.allocatedResource = properties.allocatedResource; + } + + this.queueLength = properties.queueLength; + this.queueWaitTime = properties.queueWaitTime; + this.timestamp = properties.timestamp; + if (properties.queueCapacityIsSet) { + // queue capacity is only set on node add, not on node updates + this.queueCapacity = properties.queueCapacity; } + this.labels.clear(); + this.labels.addAll(properties.labels); return this; } finally { writeLock.unlock(); @@ -107,57 +167,6 @@ public Resource getCapability() { } } - public ClusterNode setQueueLength(int qLength) { - writeLock.lock(); - try { - this.queueLength = qLength; - return this; - } finally { - writeLock.unlock(); - } - } - - public ClusterNode setQueueWaitTime(int wTime) { - writeLock.lock(); - try { - this.queueWaitTime = wTime; - return this; - } finally { - writeLock.unlock(); - } - } - - public ClusterNode updateTimestamp() { - writeLock.lock(); - try { - this.timestamp = System.currentTimeMillis(); - return this; - } finally { - writeLock.unlock(); - } - } - - public ClusterNode setQueueCapacity(int capacity) { - writeLock.lock(); - try { - this.queueCapacity = capacity; - return this; - } finally { - writeLock.unlock(); - } - } - - public ClusterNode setNodeLabels(Collection labelsToAdd) { - writeLock.lock(); - try { - labels.clear(); - labels.addAll(labelsToAdd); - return this; - } finally { - writeLock.unlock(); - } - } - public boolean hasLabel(String label) { readLock.lock(); try { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index 843fb4bd88f504..b6b21f158c1b10 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -393,14 +393,19 @@ protected void onNewNodeAdded( (estimatedQueueWaitTime != -1 || comparator == LoadComparator.QUEUE_LENGTH || comparator == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES)) { - this.clusterNodes.put(rmNode.getNodeID(), - new ClusterNode(rmNode.getNodeID()) + final ClusterNode.Properties properties = + ClusterNode.Properties.newInstance() .setQueueWaitTime(estimatedQueueWaitTime) .setQueueLength(waitQueueLength) .setNodeLabels(rmNode.getNodeLabels()) .setCapability(rmNode.getTotalCapability()) .setAllocatedResource(rmNode.getAllocatedContainerResource()) - .setQueueCapacity(opportQueueCapacity)); + .setQueueCapacity(opportQueueCapacity) + .updateTimestamp(); + + this.clusterNodes.put(rmNode.getNodeID(), + new ClusterNode(rmNode.getNodeID()).setProperties(properties)); + LOG.info( "Inserting ClusterNode [{}] with queue wait time [{}] and " + "wait queue length [{}]", @@ -430,13 +435,17 @@ protected void onExistingNodeUpdated( (estimatedQueueWaitTime != -1 || comparator == LoadComparator.QUEUE_LENGTH || comparator == LoadComparator.QUEUE_LENGTH_THEN_RESOURCES)) { - clusterNode - .setQueueWaitTime(estimatedQueueWaitTime) - .setQueueLength(waitQueueLength) - .setNodeLabels(rmNode.getNodeLabels()) - .setCapability(rmNode.getTotalCapability()) - .setAllocatedResource(rmNode.getAllocatedContainerResource()) - .updateTimestamp(); + final ClusterNode.Properties properties = + ClusterNode.Properties.newInstance() + .setQueueWaitTime(estimatedQueueWaitTime) + .setQueueLength(waitQueueLength) + .setNodeLabels(rmNode.getNodeLabels()) + .setCapability(rmNode.getTotalCapability()) + .setAllocatedResource(rmNode.getAllocatedContainerResource()) + .updateTimestamp(); + + clusterNode.setProperties(properties); + LOG.debug("Updating ClusterNode [{}] with queue wait time [{}] and" + " wait queue length [{}]", rmNode.getNodeID(), estimatedQueueWaitTime, waitQueueLength); From 3b3c0a935c67df871aae2b7eeab1432f2d90d471 Mon Sep 17 00:00:00 2001 From: Andrew Chung Date: Tue, 7 Dec 2021 11:27:04 -0500 Subject: [PATCH 10/10] Add a unit test for comparator --- .../distributed/NodeQueueLoadMonitor.java | 3 +- .../distributed/TestNodeQueueLoadMonitor.java | 197 ++++++++++++++++++ 2 files changed, 199 insertions(+), 1 deletion(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java index b6b21f158c1b10..74dc2ce7f9fe95 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java @@ -168,7 +168,8 @@ public int compare(ClusterNode o1, ClusterNode o2) { return diff; } - private void setClusterResource(Resource clusterResource) { + @VisibleForTesting + void setClusterResource(Resource clusterResource) { this.clusterResource = clusterResource; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java index 83c0226870d8cd..c8ebe6f8f20672 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/TestNodeQueueLoadMonitor.java @@ -618,6 +618,203 @@ public void testSelectAnyNode() { Assert.assertNull(node); } + @Test + public void testQueueLengthThenResourcesComparator() { + NodeQueueLoadMonitor.LoadComparator comparator = + NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH_THEN_RESOURCES; + + NodeId n1 = new FakeNodeId("n1", 5000); + NodeId n2 = new FakeNodeId("n2", 5000); + + // Case 1: larger available cores should be ranked first + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(6, 6)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn1, cn2) < 0); + } + + // Case 2: Shorter queue should be ranked first before comparing resources + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(5); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(3, 3)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn1, cn2) < 0); + } + + // Case 3: No capability vs with capability, + // with capability should come first + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(Resources.none()) + .setCapability(newResourceInstance(1, 1, 1000)) + .setQueueLength(5); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(Resources.none()) + .setCapability(Resources.none()) + .setQueueLength(5); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn1, cn2) < 0); + } + + // Case 4: Compare same values + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertEquals(0, comparator.compare(cn1, cn2)); + } + + // Case 5: If ratio is the same, compare raw values + // by VCores first, then memory + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(6, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 6)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + // Both are 60% allocated, but CN1 has 5 avail VCores, CN2 only has 4 + Assert.assertTrue(comparator.compare(cn1, cn2) < 0); + } + + // Case 6: by VCores absolute value + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 6)) + .setCapability(newResourceInstance(10, 12, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn2, cn1) < 0); + } + + // Case 7: by memory absolute value + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 5)) + .setCapability(newResourceInstance(10, 10, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(6, 5)) + .setCapability(newResourceInstance(12, 10, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn2, cn1) < 0); + } + + // Case 8: Memory should be more constraining in the overall cluster, + // so rank the node with less allocated memory first + { + ClusterNode.Properties cn1Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(5, 11)) + .setCapability(newResourceInstance(10, 100, 1000)) + .setQueueLength(10); + ClusterNode cn1 = new ClusterNode(n1); + cn1.setProperties(cn1Props); + + ClusterNode.Properties cn2Props = + ClusterNode.Properties.newInstance() + .setAllocatedResource(newResourceInstance(6, 10)) + .setCapability(newResourceInstance(10, 100, 1000)) + .setQueueLength(10); + ClusterNode cn2 = new ClusterNode(n2); + cn2.setProperties(cn2Props); + + comparator.setClusterResource( + Resources.add(cn1.getCapability(), cn2.getCapability())); + Assert.assertTrue(comparator.compare(cn1, cn2) < 0); + } + } + private RMNode createRMNode(String host, int port, int waitTime, int queueLength) { return createRMNode(host, port, waitTime, queueLength,