YARN-4888. Changes in scheduler to identify resource-requests explicitly by allocation-id. (Subru Krishnan via wangda)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/3f100d76 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/3f100d76 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/3f100d76 Branch: refs/heads/YARN-2915 Commit: 3f100d76ff5df020dbb8ecd4f5b4f9736a0a8270 Parents: e0d131f Author: Wangda Tan <wan...@apache.org> Authored: Fri Aug 5 10:43:35 2016 -0700 Committer: Wangda Tan <wan...@apache.org> Committed: Fri Aug 5 10:43:35 2016 -0700 ---------------------------------------------------------------------- .../src/main/proto/yarn_protos.proto | 2 +- .../hadoop/yarn/server/utils/BuilderUtils.java | 14 +- .../OpportunisticContainerAllocator.java | 3 +- .../scheduler/SchedulerRequestKey.java | 41 ++- .../allocator/RegularContainerAllocator.java | 15 +- .../scheduler/fair/FSAppAttempt.java | 11 +- .../scheduler/fifo/FifoScheduler.java | 7 +- .../yarn/server/resourcemanager/MockAM.java | 25 +- .../scheduler/TestAppSchedulingInfo.java | 37 ++- .../TestSchedulingWithAllocationRequestId.java | 274 +++++++++++++++++++ .../scheduler/capacity/TestUtils.java | 8 +- 11 files changed, 399 insertions(+), 38 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto index 6c337cf..d9ce11f 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto @@ -303,7 +303,7 @@ message ResourceRequestProto { optional bool relax_locality = 5 [default = true]; optional string node_label_expression = 6; optional ExecutionTypeRequestProto execution_type_request = 7; - optional int64 allocation_request_id = 8 [default = -1]; + optional int64 allocation_request_id = 8 [default = 0]; } message ExecutionTypeRequestProto { http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java index 5aa1c41..8ecbea7 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java @@ -236,7 +236,8 @@ public class BuilderUtils { public static Container newContainer(ContainerId containerId, NodeId nodeId, String nodeHttpAddress, Resource resource, Priority priority, - Token containerToken, ExecutionType executionType) { + Token containerToken, ExecutionType executionType, + long allocationRequestId) { Container container = recordFactory.newRecordInstance(Container.class); container.setId(containerId); container.setNodeId(nodeId); @@ -245,6 +246,7 @@ public class BuilderUtils { container.setPriority(priority); container.setContainerToken(containerToken); container.setExecutionType(executionType); + container.setAllocationRequestId(allocationRequestId); return container; } @@ -252,7 +254,15 @@ public class BuilderUtils { String nodeHttpAddress, Resource resource, Priority priority, Token containerToken) { return newContainer(containerId, nodeId, nodeHttpAddress, resource, - priority, containerToken, ExecutionType.GUARANTEED); + priority, containerToken, ExecutionType.GUARANTEED, 0); + } + + public static Container newContainer(ContainerId containerId, NodeId nodeId, + String nodeHttpAddress, Resource resource, Priority priority, + Token containerToken, long allocationRequestId) { + return newContainer(containerId, nodeId, nodeHttpAddress, resource, + priority, containerToken, ExecutionType.GUARANTEED, + allocationRequestId); } public static <T extends Token> T newToken(Class<T> tokenClass, http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/OpportunisticContainerAllocator.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/OpportunisticContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/OpportunisticContainerAllocator.java index ce5bda0..4723233 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/OpportunisticContainerAllocator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/OpportunisticContainerAllocator.java @@ -163,7 +163,8 @@ public class OpportunisticContainerAllocator { Container container = BuilderUtils.newContainer( cId, nodeId, nodeId.getHost() + ":" + webpagePort, capability, rr.getPriority(), containerToken, - containerTokenIdentifier.getExecutionType()); + containerTokenIdentifier.getExecutionType(), + rr.getAllocationRequestId()); return container; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerRequestKey.java ---------------------------------------------------------------------- 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/SchedulerRequestKey.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerRequestKey.java index b4988be..4b640ae 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerRequestKey.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerRequestKey.java @@ -30,9 +30,7 @@ public final class SchedulerRequestKey implements Comparable<SchedulerRequestKey> { private final Priority priority; - - public static final SchedulerRequestKey UNDEFINED = - new SchedulerRequestKey(Priority.UNDEFINED); + private final long allocationRequestId; /** * Factory method to generate a SchedulerRequestKey from a ResourceRequest. @@ -40,7 +38,8 @@ public final class SchedulerRequestKey implements * @return SchedulerRequestKey */ public static SchedulerRequestKey create(ResourceRequest req) { - return new SchedulerRequestKey(req.getPriority()); + return new SchedulerRequestKey(req.getPriority(), + req.getAllocationRequestId()); } /** @@ -50,11 +49,13 @@ public final class SchedulerRequestKey implements * @return SchedulerRequestKey */ public static SchedulerRequestKey extractFrom(Container container) { - return new SchedulerRequestKey(container.getPriority()); + return new SchedulerRequestKey(container.getPriority(), + container.getAllocationRequestId()); } - private SchedulerRequestKey(Priority priority) { + private SchedulerRequestKey(Priority priority, long allocationRequestId) { this.priority = priority; + this.allocationRequestId = allocationRequestId; } /** @@ -66,6 +67,15 @@ public final class SchedulerRequestKey implements return priority; } + /** + * Get the Id of the associated {@link ResourceRequest}. + * + * @return the Id of the associated {@link ResourceRequest} + */ + public long getAllocationRequestId() { + return allocationRequestId; + } + @Override public int compareTo(SchedulerRequestKey o) { if (o == null) { @@ -75,7 +85,12 @@ public final class SchedulerRequestKey implements return 1; } } - return o.getPriority().compareTo(priority); + int priorityCompare = o.getPriority().compareTo(priority); + // we first sort by priority and then by allocationRequestId + if (priorityCompare != 0) { + return priorityCompare; + } + return Long.compare(allocationRequestId, o.getAllocationRequestId()); } @Override @@ -88,12 +103,20 @@ public final class SchedulerRequestKey implements } SchedulerRequestKey that = (SchedulerRequestKey) o; - return getPriority().equals(that.getPriority()); + if (getAllocationRequestId() != that.getAllocationRequestId()) { + return false; + } + return getPriority() != null ? + getPriority().equals(that.getPriority()) : + that.getPriority() == null; } @Override public int hashCode() { - return getPriority().hashCode(); + int result = getPriority() != null ? getPriority().hashCode() : 0; + result = 31 * result + (int) (getAllocationRequestId() ^ ( + getAllocationRequestId() >>> 32)); + return result; } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java ---------------------------------------------------------------------- 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/capacity/allocator/RegularContainerAllocator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java index 21114f7..8d4042c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java @@ -18,6 +18,9 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.allocator; +import java.util.ArrayList; +import java.util.List; + import org.apache.commons.lang.StringUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -34,10 +37,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils; - import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey; - - import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant; @@ -54,9 +54,6 @@ import org.apache.hadoop.yarn.server.utils.BuilderUtils; import org.apache.hadoop.yarn.util.resource.ResourceCalculator; import org.apache.hadoop.yarn.util.resource.Resources; -import java.util.ArrayList; -import java.util.List; - /** * Allocate normal (new) containers, considers locality/label, etc. Using * delayed scheduling mechanism to get better locality allocation. @@ -681,8 +678,10 @@ public class RegularContainerAllocator extends AbstractContainerAllocator { application.getNewContainerId()); // Create the container - return BuilderUtils.newContainer(containerId, nodeId, node.getRMNode() - .getHttpAddress(), capability, schedulerKey.getPriority(), null); + return BuilderUtils.newContainer(containerId, nodeId, + node.getRMNode().getHttpAddress(), capability, + schedulerKey.getPriority(), null, + schedulerKey.getAllocationRequestId()); } private ContainerAllocation handleNewContainerAllocation( http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java ---------------------------------------------------------------------- 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/fair/FSAppAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java index 8f074cd..9e5a807 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java @@ -20,8 +20,8 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair; import java.io.Serializable; import java.text.DecimalFormat; -import java.util.Arrays; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Comparator; import java.util.HashMap; @@ -55,8 +55,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManage import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt; -import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey; import org.apache.hadoop.yarn.server.utils.BuilderUtils; import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator; import org.apache.hadoop.yarn.util.resource.Resources; @@ -495,9 +495,10 @@ public class FSAppAttempt extends SchedulerApplicationAttempt getApplicationAttemptId(), getNewContainerId()); // Create the container - Container container = - BuilderUtils.newContainer(containerId, nodeId, node.getRMNode() - .getHttpAddress(), capability, schedulerKey.getPriority(), null); + Container container = BuilderUtils.newContainer(containerId, nodeId, + node.getRMNode().getHttpAddress(), capability, + schedulerKey.getPriority(), null, + schedulerKey.getAllocationRequestId()); return container; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java ---------------------------------------------------------------------- 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/fifo/FifoScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java index fe8d0af..2863a97 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java @@ -710,9 +710,10 @@ public class FifoScheduler extends .getApplicationAttemptId(), application.getNewContainerId()); // Create the container - Container container = - BuilderUtils.newContainer(containerId, nodeId, node.getRMNode() - .getHttpAddress(), capability, schedulerKey.getPriority(), null); + Container container = BuilderUtils.newContainer(containerId, nodeId, + node.getRMNode().getHttpAddress(), capability, + schedulerKey.getPriority(), null, + schedulerKey.getAllocationRequestId()); // Allocate! http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java index 8f6a6c1..1b11472 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java @@ -128,7 +128,13 @@ public class MockAM { public void addRequests(String[] hosts, int memory, int priority, int containers) throws Exception { - requests.addAll(createReq(hosts, memory, priority, containers)); + addRequests(hosts, memory, priority, containers, 0L); + } + + public void addRequests(String[] hosts, int memory, int priority, + int containers, long allocationRequestId) throws Exception { + requests.addAll( + createReq(hosts, memory, priority, containers, allocationRequestId)); } public AllocateResponse schedule() throws Exception { @@ -159,17 +165,19 @@ public class MockAM { List<ContainerId> releases, String labelExpression) throws Exception { List<ResourceRequest> reqs = createReq(new String[] { host }, memory, priority, numContainers, - labelExpression); + labelExpression, 0L); return allocate(reqs, releases); } - public List<ResourceRequest> createReq(String[] hosts, int memory, int priority, - int containers) throws Exception { - return createReq(hosts, memory, priority, containers, null); + public List<ResourceRequest> createReq(String[] hosts, int memory, + int priority, int containers, long allocationRequestId) throws Exception { + return createReq(hosts, memory, priority, containers, null, + allocationRequestId); } - public List<ResourceRequest> createReq(String[] hosts, int memory, int priority, - int containers, String labelExpression) throws Exception { + public List<ResourceRequest> createReq(String[] hosts, int memory, + int priority, int containers, String labelExpression, + long allocationRequestId) throws Exception { List<ResourceRequest> reqs = new ArrayList<ResourceRequest>(); if (hosts != null) { for (String host : hosts) { @@ -178,10 +186,12 @@ public class MockAM { ResourceRequest hostReq = createResourceReq(host, memory, priority, containers, labelExpression); + hostReq.setAllocationRequestId(allocationRequestId); reqs.add(hostReq); ResourceRequest rackReq = createResourceReq("/default-rack", memory, priority, containers, labelExpression); + rackReq.setAllocationRequestId(allocationRequestId); reqs.add(rackReq); } } @@ -189,6 +199,7 @@ public class MockAM { ResourceRequest offRackReq = createResourceReq(ResourceRequest.ANY, memory, priority, containers, labelExpression); + offRackReq.setAllocationRequestId(allocationRequestId); reqs.add(offRackReq); return reqs; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java ---------------------------------------------------------------------- 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/TestAppSchedulingInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java index a1c6294..503ea34 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java @@ -18,12 +18,17 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; + import java.util.ArrayList; +import java.util.Iterator; +import java.util.TreeSet; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.Priority; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue; import org.junit.Assert; import org.junit.Test; @@ -70,4 +75,34 @@ public class TestAppSchedulingInfo { blacklistRemovals); Assert.assertFalse(appSchedulingInfo.getAndResetBlacklistChanged()); } + + @Test + public void testSchedulerRequestKeyOrdering() { + TreeSet<SchedulerRequestKey> ts = new TreeSet<>(); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(1), 1)); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(1), 2)); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(0), 4)); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(0), 3)); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(2), 5)); + ts.add(TestUtils.toSchedulerKey(Priority.newInstance(2), 6)); + Iterator<SchedulerRequestKey> iter = ts.iterator(); + SchedulerRequestKey sk = iter.next(); + Assert.assertEquals(0, sk.getPriority().getPriority()); + Assert.assertEquals(3, sk.getAllocationRequestId()); + sk = iter.next(); + Assert.assertEquals(0, sk.getPriority().getPriority()); + Assert.assertEquals(4, sk.getAllocationRequestId()); + sk = iter.next(); + Assert.assertEquals(1, sk.getPriority().getPriority()); + Assert.assertEquals(1, sk.getAllocationRequestId()); + sk = iter.next(); + Assert.assertEquals(1, sk.getPriority().getPriority()); + Assert.assertEquals(2, sk.getAllocationRequestId()); + sk = iter.next(); + Assert.assertEquals(2, sk.getPriority().getPriority()); + Assert.assertEquals(5, sk.getAllocationRequestId()); + sk = iter.next(); + Assert.assertEquals(2, sk.getPriority().getPriority()); + Assert.assertEquals(6, sk.getAllocationRequestId()); + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestSchedulingWithAllocationRequestId.java ---------------------------------------------------------------------- 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/TestSchedulingWithAllocationRequestId.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestSchedulingWithAllocationRequestId.java new file mode 100644 index 0000000..e60fd6f --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestSchedulingWithAllocationRequestId.java @@ -0,0 +1,274 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.server.resourcemanager.scheduler; + +import java.util.List; + +import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse; +import org.apache.hadoop.yarn.api.records.Container; +import org.apache.hadoop.yarn.api.records.NodeId; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.server.resourcemanager.MockAM; +import org.apache.hadoop.yarn.server.resourcemanager.MockNM; +import org.apache.hadoop.yarn.server.resourcemanager.MockRM; +import org.apache.hadoop.yarn.server.resourcemanager.ParameterizedSchedulerTestBase; +import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp; +import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Tests for checking Scheduling with allocationRequestId, i.e. mapping of + * allocated containers to the original client {@code ResourceRequest}. + */ +public class TestSchedulingWithAllocationRequestId + extends ParameterizedSchedulerTestBase { + + private static final Logger LOG = + LoggerFactory.getLogger(TestSchedulingWithAllocationRequestId.class); + private static final int GB = 1024; + + @Test + public void testMultipleAllocationRequestIds() throws Exception { + configureScheduler(); + YarnConfiguration conf = getConf(); + MockRM rm = new MockRM(conf); + try { + rm.start(); + + MockNM nm1 = rm.registerNode("127.0.0.1:1234", 4 * GB); + MockNM nm2 = rm.registerNode("127.0.0.2:5678", 4 * GB); + RMApp app1 = rm.submitApp(2048); + // kick the scheduling + nm1.nodeHeartbeat(true); + RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); + MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); + am1.registerAppAttempt(); + + // add request for containers with id 10 & 20 + am1.addRequests(new String[] {"127.0.0.1" }, 2 * GB, 1, 1, 10L); + AllocateResponse allocResponse = am1.schedule(); // send the request + am1.addRequests(new String[] {"127.0.0.2" }, 2 * GB, 1, 2, 20L); + allocResponse = am1.schedule(); // send the request + + // check if request id 10 is satisfied + nm1.nodeHeartbeat(true); + allocResponse = am1.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + List<Container> allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 2 * GB, nm1.getNodeId(), 10); + + // check now if request id 20 is satisfied + nm2.nodeHeartbeat(true); + while (allocResponse.getAllocatedContainers().size() < 2) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + + allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(2, allocated.size()); + for (Container container : allocated) { + checkAllocatedContainer(container, 2 * GB, nm2.getNodeId(), 20); + } + } finally { + if (rm != null) { + rm.stop(); + } + } + } + + @Test + public void testMultipleAllocationRequestDiffPriority() throws Exception { + configureScheduler(); + YarnConfiguration conf = getConf(); + MockRM rm = new MockRM(conf); + try { + rm.start(); + + MockNM nm1 = rm.registerNode("127.0.0.1:1234", 4 * GB); + MockNM nm2 = rm.registerNode("127.0.0.2:5678", 4 * GB); + RMApp app1 = rm.submitApp(2048); + // kick the scheduling + nm1.nodeHeartbeat(true); + RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); + MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); + am1.registerAppAttempt(); + + // add request for containers with id 10 & 20 + am1.addRequests(new String[] {"127.0.0.1" }, 2 * GB, 2, 1, 10L); + AllocateResponse allocResponse = am1.schedule(); // send the request + am1.addRequests(new String[] {"127.0.0.2" }, 2 * GB, 1, 2, 20L); + allocResponse = am1.schedule(); // send the request + + // check if request id 20 is satisfied first + nm2.nodeHeartbeat(true); + while (allocResponse.getAllocatedContainers().size() < 2) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + + List<Container> allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(2, allocated.size()); + for (Container container : allocated) { + checkAllocatedContainer(container, 2 * GB, nm2.getNodeId(), 20); + } + + // check now if request id 10 is satisfied + nm1.nodeHeartbeat(true); + allocResponse = am1.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 2 * GB, nm1.getNodeId(), 10); + } finally { + if (rm != null) { + rm.stop(); + } + } + } + + private void checkAllocatedContainer(Container allocated, int memory, + NodeId nodeId, long allocationRequestId) { + Assert.assertEquals(memory, allocated.getResource().getMemorySize()); + Assert.assertEquals(nodeId, allocated.getNodeId()); + Assert.assertEquals(allocationRequestId, + allocated.getAllocationRequestId()); + } + + @Test + public void testMultipleAppsWithAllocationReqId() throws Exception { + configureScheduler(); + YarnConfiguration conf = getConf(); + MockRM rm = new MockRM(conf); + try { + rm.start(); + + // Register node1 + String host0 = "host_0"; + String host1 = "host_1"; + MockNM nm1 = + new MockNM(host0 + ":1234", 8 * GB, rm.getResourceTrackerService()); + nm1.registerNode(); + + // Register node2 + MockNM nm2 = + new MockNM(host1 + ":2351", 8 * GB, rm.getResourceTrackerService()); + nm2.registerNode(); + + // submit 1st app + RMApp app1 = rm.submitApp(1 * GB, "user_0", "a1"); + MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm1); + + // Submit app1 RR with allocationReqId = 5 + int numContainers = 1; + am1.addRequests(new String[] {host0, host1 }, 1 * GB, 1, numContainers, + 5L); + AllocateResponse allocResponse = am1.schedule(); + + // wait for containers to be allocated. + nm1.nodeHeartbeat(true); + allocResponse = am1.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + + List<Container> allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 1 * GB, nm1.getNodeId(), 5L); + + // Submit another application + RMApp app2 = rm.submitApp(1 * GB, "user_1", "a2"); + MockAM am2 = MockRM.launchAndRegisterAM(app2, rm, nm2); + + // Submit app2 RR with allocationReqId = 5 + am2.addRequests(new String[] {host0, host1 }, 2 * GB, 1, numContainers, + 5L); + am2.schedule(); + + // wait for containers to be allocated. + nm2.nodeHeartbeat(true); + allocResponse = am2.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am2.schedule(); + } + + allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 2 * GB, nm2.getNodeId(), 5L); + + // Now submit app2 RR with allocationReqId = 10 + am2.addRequests(new String[] {host0, host1 }, 3 * GB, 1, numContainers, + 10L); + am2.schedule(); + + // wait for containers to be allocated. + nm1.nodeHeartbeat(true); + allocResponse = am2.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am2.schedule(); + } + + allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 3 * GB, nm1.getNodeId(), 10L); + + // Now submit app1 RR with allocationReqId = 10 + am1.addRequests(new String[] {host0, host1 }, 4 * GB, 1, numContainers, + 10L); + am1.schedule(); + + // wait for containers to be allocated. + nm2.nodeHeartbeat(true); + allocResponse = am1.schedule(); // send the request + while (allocResponse.getAllocatedContainers().size() < 1) { + LOG.info("Waiting for containers to be created for app 1..."); + Thread.sleep(100); + allocResponse = am1.schedule(); + } + + allocated = allocResponse.getAllocatedContainers(); + Assert.assertEquals(1, allocated.size()); + checkAllocatedContainer(allocated.get(0), 4 * GB, nm2.getNodeId(), 10L); + } finally { + if (rm != null) { + rm.stop(); + } + } + } + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/3f100d76/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java ---------------------------------------------------------------------- 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/capacity/TestUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java index c808b5a..66e833f 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java @@ -51,7 +51,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsMana import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer; import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler; - import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode; @@ -412,4 +411,11 @@ public class TestUtils { return SchedulerRequestKey.create(ResourceRequest.newInstance( Priority.newInstance(pri), null, null, 0)); } + + public static SchedulerRequestKey toSchedulerKey(Priority pri, + long allocationRequestId) { + ResourceRequest req = ResourceRequest.newInstance(pri, null, null, 0); + req.setAllocationRequestId(allocationRequestId); + return SchedulerRequestKey.create(req); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
