You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by ka...@apache.org on 2014/08/12 23:43:28 UTC
svn commit: r1617600 [1/2] - in /hadoop/common/trunk/hadoop-yarn-project: ./
hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/
hadoop-yarn/hadoop-yarn-server/ha...
Author: kasha
Date: Tue Aug 12 21:43:27 2014
New Revision: 1617600
URL: http://svn.apache.org/r1617600
Log:
YARN-2399. FairScheduler: Merge AppSchedulable and FSSchedulerApp into FSAppAttempt. (kasha)
Added:
hadoop/common/trunk/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
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java
Modified:
hadoop/common/trunk/hadoop-yarn-project/CHANGES.txt
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FifoAppComparator.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/MaxRunningAppsEnforcer.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/NewAppWeightBooster.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/WeightAdjuster.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSLeafQueue.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java
hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestMaxRunningAppsEnforcer.java
Modified: hadoop/common/trunk/hadoop-yarn-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/CHANGES.txt?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/CHANGES.txt (original)
+++ hadoop/common/trunk/hadoop-yarn-project/CHANGES.txt Tue Aug 12 21:43:27 2014
@@ -122,6 +122,9 @@ Release 2.6.0 - UNRELEASED
YARN-2317. Updated the document about how to write YARN applications. (Li Lu via
zjshen)
+ YARN-2399. FairScheduler: Merge AppSchedulable and FSSchedulerApp into
+ FSAppAttempt. (kasha)
+
OPTIMIZATIONS
BUG FIXES
Added: hadoop/common/trunk/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
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/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?rev=1617600&view=auto
==============================================================================
--- hadoop/common/trunk/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 (added)
+++ hadoop/common/trunk/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 Tue Aug 12 21:43:27 2014
@@ -0,0 +1,768 @@
+/**
+ * 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.fair;
+
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger;
+import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerFinishedEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
+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.utils.BuilderUtils;
+import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.Resources;
+
+/**
+ * Represents an application attempt from the viewpoint of the Fair Scheduler.
+ */
+@Private
+@Unstable
+public class FSAppAttempt extends SchedulerApplicationAttempt
+ implements Schedulable {
+
+ private static final Log LOG = LogFactory.getLog(FSAppAttempt.class);
+ private static final DefaultResourceCalculator RESOURCE_CALCULATOR
+ = new DefaultResourceCalculator();
+
+ private long startTime;
+ private Priority priority;
+ private ResourceWeights resourceWeights;
+ private Resource demand = Resources.createResource(0);
+ private FairScheduler scheduler;
+ private Resource fairShare = Resources.createResource(0, 0);
+ private Resource preemptedResources = Resources.createResource(0);
+ private RMContainerComparator comparator = new RMContainerComparator();
+ private final Map<RMContainer, Long> preemptionMap = new HashMap<RMContainer, Long>();
+
+ /**
+ * Delay scheduling: We often want to prioritize scheduling of node-local
+ * containers over rack-local or off-switch containers. To acheive this
+ * we first only allow node-local assigments for a given prioirty level,
+ * then relax the locality threshold once we've had a long enough period
+ * without succesfully scheduling. We measure both the number of "missed"
+ * scheduling opportunities since the last container was scheduled
+ * at the current allowed level and the time since the last container
+ * was scheduled. Currently we use only the former.
+ */
+ private final Map<Priority, NodeType> allowedLocalityLevel =
+ new HashMap<Priority, NodeType>();
+
+ public FSAppAttempt(FairScheduler scheduler,
+ ApplicationAttemptId applicationAttemptId, String user, FSLeafQueue queue,
+ ActiveUsersManager activeUsersManager, RMContext rmContext) {
+ super(applicationAttemptId, user, queue, activeUsersManager, rmContext);
+
+ this.scheduler = scheduler;
+ this.startTime = scheduler.getClock().getTime();
+ this.priority = Priority.newInstance(1);
+ this.resourceWeights = new ResourceWeights();
+ }
+
+ public ResourceWeights getResourceWeights() {
+ return resourceWeights;
+ }
+
+ /**
+ * Get metrics reference from containing queue.
+ */
+ public QueueMetrics getMetrics() {
+ return queue.getMetrics();
+ }
+
+ synchronized public void containerCompleted(RMContainer rmContainer,
+ ContainerStatus containerStatus, RMContainerEventType event) {
+
+ Container container = rmContainer.getContainer();
+ ContainerId containerId = container.getId();
+
+ // Remove from the list of newly allocated containers if found
+ newlyAllocatedContainers.remove(rmContainer);
+
+ // Inform the container
+ rmContainer.handle(
+ new RMContainerFinishedEvent(
+ containerId,
+ containerStatus,
+ event)
+ );
+ LOG.info("Completed container: " + rmContainer.getContainerId() +
+ " in state: " + rmContainer.getState() + " event:" + event);
+
+ // Remove from the list of containers
+ liveContainers.remove(rmContainer.getContainerId());
+
+ RMAuditLogger.logSuccess(getUser(),
+ AuditConstants.RELEASE_CONTAINER, "SchedulerApp",
+ getApplicationId(), containerId);
+
+ // Update usage metrics
+ Resource containerResource = rmContainer.getContainer().getResource();
+ queue.getMetrics().releaseResources(getUser(), 1, containerResource);
+ Resources.subtractFrom(currentConsumption, containerResource);
+
+ // remove from preemption map if it is completed
+ preemptionMap.remove(rmContainer);
+ }
+
+ private synchronized void unreserveInternal(
+ Priority priority, FSSchedulerNode node) {
+ Map<NodeId, RMContainer> reservedContainers =
+ this.reservedContainers.get(priority);
+ RMContainer reservedContainer = reservedContainers.remove(node.getNodeID());
+ if (reservedContainers.isEmpty()) {
+ this.reservedContainers.remove(priority);
+ }
+
+ // Reset the re-reservation count
+ resetReReservations(priority);
+
+ Resource resource = reservedContainer.getContainer().getResource();
+ Resources.subtractFrom(currentReservation, resource);
+
+ LOG.info("Application " + getApplicationId() + " unreserved " + " on node "
+ + node + ", currently has " + reservedContainers.size() + " at priority "
+ + priority + "; currentReservation " + currentReservation);
+ }
+
+ public synchronized float getLocalityWaitFactor(
+ Priority priority, int clusterNodes) {
+ // Estimate: Required unique resources (i.e. hosts + racks)
+ int requiredResources =
+ Math.max(this.getResourceRequests(priority).size() - 1, 0);
+
+ // waitFactor can't be more than '1'
+ // i.e. no point skipping more than clustersize opportunities
+ return Math.min(((float)requiredResources / clusterNodes), 1.0f);
+ }
+
+ /**
+ * Return the level at which we are allowed to schedule containers, given the
+ * current size of the cluster and thresholds indicating how many nodes to
+ * fail at (as a fraction of cluster size) before relaxing scheduling
+ * constraints.
+ */
+ public synchronized NodeType getAllowedLocalityLevel(Priority priority,
+ int numNodes, double nodeLocalityThreshold, double rackLocalityThreshold) {
+ // upper limit on threshold
+ if (nodeLocalityThreshold > 1.0) { nodeLocalityThreshold = 1.0; }
+ if (rackLocalityThreshold > 1.0) { rackLocalityThreshold = 1.0; }
+
+ // If delay scheduling is not being used, can schedule anywhere
+ if (nodeLocalityThreshold < 0.0 || rackLocalityThreshold < 0.0) {
+ return NodeType.OFF_SWITCH;
+ }
+
+ // Default level is NODE_LOCAL
+ if (!allowedLocalityLevel.containsKey(priority)) {
+ allowedLocalityLevel.put(priority, NodeType.NODE_LOCAL);
+ return NodeType.NODE_LOCAL;
+ }
+
+ NodeType allowed = allowedLocalityLevel.get(priority);
+
+ // If level is already most liberal, we're done
+ if (allowed.equals(NodeType.OFF_SWITCH)) return NodeType.OFF_SWITCH;
+
+ double threshold = allowed.equals(NodeType.NODE_LOCAL) ? nodeLocalityThreshold :
+ rackLocalityThreshold;
+
+ // Relax locality constraints once we've surpassed threshold.
+ if (getSchedulingOpportunities(priority) > (numNodes * threshold)) {
+ if (allowed.equals(NodeType.NODE_LOCAL)) {
+ allowedLocalityLevel.put(priority, NodeType.RACK_LOCAL);
+ resetSchedulingOpportunities(priority);
+ }
+ else if (allowed.equals(NodeType.RACK_LOCAL)) {
+ allowedLocalityLevel.put(priority, NodeType.OFF_SWITCH);
+ resetSchedulingOpportunities(priority);
+ }
+ }
+ return allowedLocalityLevel.get(priority);
+ }
+
+ /**
+ * Return the level at which we are allowed to schedule containers.
+ * Given the thresholds indicating how much time passed before relaxing
+ * scheduling constraints.
+ */
+ public synchronized NodeType getAllowedLocalityLevelByTime(Priority priority,
+ long nodeLocalityDelayMs, long rackLocalityDelayMs,
+ long currentTimeMs) {
+
+ // if not being used, can schedule anywhere
+ if (nodeLocalityDelayMs < 0 || rackLocalityDelayMs < 0) {
+ return NodeType.OFF_SWITCH;
+ }
+
+ // default level is NODE_LOCAL
+ if (! allowedLocalityLevel.containsKey(priority)) {
+ allowedLocalityLevel.put(priority, NodeType.NODE_LOCAL);
+ return NodeType.NODE_LOCAL;
+ }
+
+ NodeType allowed = allowedLocalityLevel.get(priority);
+
+ // if level is already most liberal, we're done
+ if (allowed.equals(NodeType.OFF_SWITCH)) {
+ return NodeType.OFF_SWITCH;
+ }
+
+ // check waiting time
+ long waitTime = currentTimeMs;
+ if (lastScheduledContainer.containsKey(priority)) {
+ waitTime -= lastScheduledContainer.get(priority);
+ } else {
+ waitTime -= getStartTime();
+ }
+
+ long thresholdTime = allowed.equals(NodeType.NODE_LOCAL) ?
+ nodeLocalityDelayMs : rackLocalityDelayMs;
+
+ if (waitTime > thresholdTime) {
+ if (allowed.equals(NodeType.NODE_LOCAL)) {
+ allowedLocalityLevel.put(priority, NodeType.RACK_LOCAL);
+ resetSchedulingOpportunities(priority, currentTimeMs);
+ } else if (allowed.equals(NodeType.RACK_LOCAL)) {
+ allowedLocalityLevel.put(priority, NodeType.OFF_SWITCH);
+ resetSchedulingOpportunities(priority, currentTimeMs);
+ }
+ }
+ return allowedLocalityLevel.get(priority);
+ }
+
+ synchronized public RMContainer allocate(NodeType type, FSSchedulerNode node,
+ Priority priority, ResourceRequest request,
+ Container container) {
+ // Update allowed locality level
+ NodeType allowed = allowedLocalityLevel.get(priority);
+ if (allowed != null) {
+ if (allowed.equals(NodeType.OFF_SWITCH) &&
+ (type.equals(NodeType.NODE_LOCAL) ||
+ type.equals(NodeType.RACK_LOCAL))) {
+ this.resetAllowedLocalityLevel(priority, type);
+ }
+ else if (allowed.equals(NodeType.RACK_LOCAL) &&
+ type.equals(NodeType.NODE_LOCAL)) {
+ this.resetAllowedLocalityLevel(priority, type);
+ }
+ }
+
+ // Required sanity check - AM can call 'allocate' to update resource
+ // request without locking the scheduler, hence we need to check
+ if (getTotalRequiredResources(priority) <= 0) {
+ return null;
+ }
+
+ // Create RMContainer
+ RMContainer rmContainer = new RMContainerImpl(container,
+ getApplicationAttemptId(), node.getNodeID(),
+ appSchedulingInfo.getUser(), rmContext);
+
+ // Add it to allContainers list.
+ newlyAllocatedContainers.add(rmContainer);
+ liveContainers.put(container.getId(), rmContainer);
+
+ // Update consumption and track allocations
+ List<ResourceRequest> resourceRequestList = appSchedulingInfo.allocate(
+ type, node, priority, request, container);
+ Resources.addTo(currentConsumption, container.getResource());
+
+ // Update resource requests related to "request" and store in RMContainer
+ ((RMContainerImpl) rmContainer).setResourceRequests(resourceRequestList);
+
+ // Inform the container
+ rmContainer.handle(
+ new RMContainerEvent(container.getId(), RMContainerEventType.START));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("allocate: applicationAttemptId="
+ + container.getId().getApplicationAttemptId()
+ + " container=" + container.getId() + " host="
+ + container.getNodeId().getHost() + " type=" + type);
+ }
+ RMAuditLogger.logSuccess(getUser(),
+ AuditConstants.ALLOC_CONTAINER, "SchedulerApp",
+ getApplicationId(), container.getId());
+
+ return rmContainer;
+ }
+
+ /**
+ * Should be called when the scheduler assigns a container at a higher
+ * degree of locality than the current threshold. Reset the allowed locality
+ * level to a higher degree of locality.
+ */
+ public synchronized void resetAllowedLocalityLevel(Priority priority,
+ NodeType level) {
+ NodeType old = allowedLocalityLevel.get(priority);
+ LOG.info("Raising locality level from " + old + " to " + level + " at " +
+ " priority " + priority);
+ allowedLocalityLevel.put(priority, level);
+ }
+
+ // related methods
+ public void addPreemption(RMContainer container, long time) {
+ assert preemptionMap.get(container) == null;
+ preemptionMap.put(container, time);
+ Resources.addTo(preemptedResources, container.getAllocatedResource());
+ }
+
+ public Long getContainerPreemptionTime(RMContainer container) {
+ return preemptionMap.get(container);
+ }
+
+ public Set<RMContainer> getPreemptionContainers() {
+ return preemptionMap.keySet();
+ }
+
+ @Override
+ public FSLeafQueue getQueue() {
+ return (FSLeafQueue)super.getQueue();
+ }
+
+ public Resource getPreemptedResources() {
+ return preemptedResources;
+ }
+
+ public void resetPreemptedResources() {
+ preemptedResources = Resources.createResource(0);
+ for (RMContainer container : getPreemptionContainers()) {
+ Resources.addTo(preemptedResources, container.getAllocatedResource());
+ }
+ }
+
+ public void clearPreemptedResources() {
+ preemptedResources.setMemory(0);
+ preemptedResources.setVirtualCores(0);
+ }
+
+ /**
+ * Create and return a container object reflecting an allocation for the
+ * given appliction on the given node with the given capability and
+ * priority.
+ */
+ public Container createContainer(
+ FSSchedulerNode node, Resource capability, Priority priority) {
+
+ NodeId nodeId = node.getRMNode().getNodeID();
+ ContainerId containerId = BuilderUtils.newContainerId(
+ getApplicationAttemptId(), getNewContainerId());
+
+ // Create the container
+ Container container =
+ BuilderUtils.newContainer(containerId, nodeId, node.getRMNode()
+ .getHttpAddress(), capability, priority, null);
+
+ return container;
+ }
+
+ /**
+ * Reserve a spot for {@code container} on this {@code node}. If
+ * the container is {@code alreadyReserved} on the node, simply
+ * update relevant bookeeping. This dispatches ro relevant handlers
+ * in {@link FSSchedulerNode}..
+ */
+ private void reserve(Priority priority, FSSchedulerNode node,
+ Container container, boolean alreadyReserved) {
+ LOG.info("Making reservation: node=" + node.getNodeName() +
+ " app_id=" + getApplicationId());
+
+ if (!alreadyReserved) {
+ getMetrics().reserveResource(getUser(), container.getResource());
+ RMContainer rmContainer =
+ super.reserve(node, priority, null, container);
+ node.reserveResource(this, priority, rmContainer);
+ } else {
+ RMContainer rmContainer = node.getReservedContainer();
+ super.reserve(node, priority, rmContainer, container);
+ node.reserveResource(this, priority, rmContainer);
+ }
+ }
+
+ /**
+ * Remove the reservation on {@code node} at the given {@link Priority}.
+ * This dispatches SchedulerNode handlers as well.
+ */
+ public void unreserve(Priority priority, FSSchedulerNode node) {
+ RMContainer rmContainer = node.getReservedContainer();
+ unreserveInternal(priority, node);
+ node.unreserveResource(this);
+ getMetrics().unreserveResource(
+ getUser(), rmContainer.getContainer().getResource());
+ }
+
+ /**
+ * Assign a container to this node to facilitate {@code request}. If node does
+ * not have enough memory, create a reservation. This is called once we are
+ * sure the particular request should be facilitated by this node.
+ *
+ * @param node
+ * The node to try placing the container on.
+ * @param request
+ * The ResourceRequest we're trying to satisfy.
+ * @param type
+ * The locality of the assignment.
+ * @param reserved
+ * Whether there's already a container reserved for this app on the node.
+ * @return
+ * If an assignment was made, returns the resources allocated to the
+ * container. If a reservation was made, returns
+ * FairScheduler.CONTAINER_RESERVED. If no assignment or reservation was
+ * made, returns an empty resource.
+ */
+ private Resource assignContainer(
+ FSSchedulerNode node, ResourceRequest request, NodeType type,
+ boolean reserved) {
+
+ // How much does this request need?
+ Resource capability = request.getCapability();
+
+ // How much does the node have?
+ Resource available = node.getAvailableResource();
+
+ Container container = null;
+ if (reserved) {
+ container = node.getReservedContainer().getContainer();
+ } else {
+ container = createContainer(node, capability, request.getPriority());
+ }
+
+ // Can we allocate a container on this node?
+ if (Resources.fitsIn(capability, available)) {
+ // Inform the application of the new container for this request
+ RMContainer allocatedContainer =
+ allocate(type, node, request.getPriority(), request, container);
+ if (allocatedContainer == null) {
+ // Did the application need this resource?
+ if (reserved) {
+ unreserve(request.getPriority(), node);
+ }
+ return Resources.none();
+ }
+
+ // If we had previously made a reservation, delete it
+ if (reserved) {
+ unreserve(request.getPriority(), node);
+ }
+
+ // Inform the node
+ node.allocateContainer(allocatedContainer);
+
+ // If this container is used to run AM, update the leaf queue's AM usage
+ if (getLiveContainers().size() == 1 && !getUnmanagedAM()) {
+ getQueue().addAMResourceUsage(container.getResource());
+ setAmRunning(true);
+ }
+
+ return container.getResource();
+ } else {
+ // The desired container won't fit here, so reserve
+ reserve(request.getPriority(), node, container, reserved);
+
+ return FairScheduler.CONTAINER_RESERVED;
+ }
+ }
+
+ private Resource assignContainer(FSSchedulerNode node, boolean reserved) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Node offered to app: " + getName() + " reserved: " + reserved);
+ }
+
+ Collection<Priority> prioritiesToTry = (reserved) ?
+ Arrays.asList(node.getReservedContainer().getReservedPriority()) :
+ getPriorities();
+
+ // For each priority, see if we can schedule a node local, rack local
+ // or off-switch request. Rack of off-switch requests may be delayed
+ // (not scheduled) in order to promote better locality.
+ synchronized (this) {
+ for (Priority priority : prioritiesToTry) {
+ if (getTotalRequiredResources(priority) <= 0 ||
+ !hasContainerForNode(priority, node)) {
+ continue;
+ }
+
+ addSchedulingOpportunity(priority);
+
+ // Check the AM resource usage for the leaf queue
+ if (getLiveContainers().size() == 0 && !getUnmanagedAM()) {
+ if (!getQueue().canRunAppAM(getAMResource())) {
+ return Resources.none();
+ }
+ }
+
+ ResourceRequest rackLocalRequest = getResourceRequest(priority,
+ node.getRackName());
+ ResourceRequest localRequest = getResourceRequest(priority,
+ node.getNodeName());
+
+ if (localRequest != null && !localRequest.getRelaxLocality()) {
+ LOG.warn("Relax locality off is not supported on local request: "
+ + localRequest);
+ }
+
+ NodeType allowedLocality;
+ if (scheduler.isContinuousSchedulingEnabled()) {
+ allowedLocality = getAllowedLocalityLevelByTime(priority,
+ scheduler.getNodeLocalityDelayMs(),
+ scheduler.getRackLocalityDelayMs(),
+ scheduler.getClock().getTime());
+ } else {
+ allowedLocality = getAllowedLocalityLevel(priority,
+ scheduler.getNumClusterNodes(),
+ scheduler.getNodeLocalityThreshold(),
+ scheduler.getRackLocalityThreshold());
+ }
+
+ if (rackLocalRequest != null && rackLocalRequest.getNumContainers() != 0
+ && localRequest != null && localRequest.getNumContainers() != 0) {
+ return assignContainer(node, localRequest,
+ NodeType.NODE_LOCAL, reserved);
+ }
+
+ if (rackLocalRequest != null && !rackLocalRequest.getRelaxLocality()) {
+ continue;
+ }
+
+ if (rackLocalRequest != null && rackLocalRequest.getNumContainers() != 0
+ && (allowedLocality.equals(NodeType.RACK_LOCAL) ||
+ allowedLocality.equals(NodeType.OFF_SWITCH))) {
+ return assignContainer(node, rackLocalRequest,
+ NodeType.RACK_LOCAL, reserved);
+ }
+
+ ResourceRequest offSwitchRequest =
+ getResourceRequest(priority, ResourceRequest.ANY);
+ if (offSwitchRequest != null && !offSwitchRequest.getRelaxLocality()) {
+ continue;
+ }
+
+ if (offSwitchRequest != null && offSwitchRequest.getNumContainers() != 0
+ && allowedLocality.equals(NodeType.OFF_SWITCH)) {
+ return assignContainer(node, offSwitchRequest,
+ NodeType.OFF_SWITCH, reserved);
+ }
+ }
+ }
+ return Resources.none();
+ }
+
+ /**
+ * Called when this application already has an existing reservation on the
+ * given node. Sees whether we can turn the reservation into an allocation.
+ * Also checks whether the application needs the reservation anymore, and
+ * releases it if not.
+ *
+ * @param node
+ * Node that the application has an existing reservation on
+ */
+ public Resource assignReservedContainer(FSSchedulerNode node) {
+ RMContainer rmContainer = node.getReservedContainer();
+ Priority priority = rmContainer.getReservedPriority();
+
+ // Make sure the application still needs requests at this priority
+ if (getTotalRequiredResources(priority) == 0) {
+ unreserve(priority, node);
+ return Resources.none();
+ }
+
+ // Fail early if the reserved container won't fit.
+ // Note that we have an assumption here that there's only one container size
+ // per priority.
+ if (!Resources.fitsIn(node.getReservedContainer().getReservedResource(),
+ node.getAvailableResource())) {
+ return Resources.none();
+ }
+
+ return assignContainer(node, true);
+ }
+
+
+ /**
+ * Whether this app has containers requests that could be satisfied on the
+ * given node, if the node had full space.
+ */
+ public boolean hasContainerForNode(Priority prio, FSSchedulerNode node) {
+ ResourceRequest anyRequest = getResourceRequest(prio, ResourceRequest.ANY);
+ ResourceRequest rackRequest = getResourceRequest(prio, node.getRackName());
+ ResourceRequest nodeRequest = getResourceRequest(prio, node.getNodeName());
+
+ return
+ // There must be outstanding requests at the given priority:
+ anyRequest != null && anyRequest.getNumContainers() > 0 &&
+ // If locality relaxation is turned off at *-level, there must be a
+ // non-zero request for the node's rack:
+ (anyRequest.getRelaxLocality() ||
+ (rackRequest != null && rackRequest.getNumContainers() > 0)) &&
+ // If locality relaxation is turned off at rack-level, there must be a
+ // non-zero request at the node:
+ (rackRequest == null || rackRequest.getRelaxLocality() ||
+ (nodeRequest != null && nodeRequest.getNumContainers() > 0)) &&
+ // The requested container must be able to fit on the node:
+ Resources.lessThanOrEqual(RESOURCE_CALCULATOR, null,
+ anyRequest.getCapability(), node.getRMNode().getTotalCapability());
+ }
+
+
+ static class RMContainerComparator implements Comparator<RMContainer>,
+ Serializable {
+ @Override
+ public int compare(RMContainer c1, RMContainer c2) {
+ int ret = c1.getContainer().getPriority().compareTo(
+ c2.getContainer().getPriority());
+ if (ret == 0) {
+ return c2.getContainerId().compareTo(c1.getContainerId());
+ }
+ return ret;
+ }
+ }
+
+ /* Schedulable methods implementation */
+
+ @Override
+ public String getName() {
+ return getApplicationId().toString();
+ }
+
+ @Override
+ public Resource getDemand() {
+ return demand;
+ }
+
+ @Override
+ public long getStartTime() {
+ return startTime;
+ }
+
+ @Override
+ public Resource getMinShare() {
+ return Resources.none();
+ }
+
+ @Override
+ public Resource getMaxShare() {
+ return Resources.unbounded();
+ }
+
+ @Override
+ public Resource getResourceUsage() {
+ // Here the getPreemptedResources() always return zero, except in
+ // a preemption round
+ return Resources.subtract(getCurrentConsumption(), getPreemptedResources());
+ }
+
+ @Override
+ public ResourceWeights getWeights() {
+ return scheduler.getAppWeight(this);
+ }
+
+ @Override
+ public Priority getPriority() {
+ // Right now per-app priorities are not passed to scheduler,
+ // so everyone has the same priority.
+ return priority;
+ }
+
+ @Override
+ public Resource getFairShare() {
+ return this.fairShare;
+ }
+
+ @Override
+ public void setFairShare(Resource fairShare) {
+ this.fairShare = fairShare;
+ }
+
+ @Override
+ public boolean isActive() {
+ return true;
+ }
+
+
+ @Override
+ public void updateDemand() {
+ demand = Resources.createResource(0);
+ // Demand is current consumption plus outstanding requests
+ Resources.addTo(demand, getCurrentConsumption());
+
+ // Add up outstanding resource requests
+ synchronized (this) {
+ for (Priority p : getPriorities()) {
+ for (ResourceRequest r : getResourceRequests(p).values()) {
+ Resource total = Resources.multiply(r.getCapability(), r.getNumContainers());
+ Resources.addTo(demand, total);
+ }
+ }
+ }
+ }
+
+ @Override
+ public Resource assignContainer(FSSchedulerNode node) {
+ return assignContainer(node, false);
+ }
+
+ /**
+ * Preempt a running container according to the priority
+ */
+ @Override
+ public RMContainer preemptContainer() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("App " + getName() + " is going to preempt a running " +
+ "container");
+ }
+
+ RMContainer toBePreempted = null;
+ for (RMContainer container : getLiveContainers()) {
+ if (!getPreemptionContainers().contains(container) &&
+ (toBePreempted == null ||
+ comparator.compare(toBePreempted, container) > 0)) {
+ toBePreempted = container;
+ }
+ }
+ return toBePreempted;
+ }
+}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java Tue Aug 12 21:43:27 2014
@@ -44,11 +44,11 @@ import org.apache.hadoop.yarn.util.resou
public class FSLeafQueue extends FSQueue {
private static final Log LOG = LogFactory.getLog(
FSLeafQueue.class.getName());
-
- private final List<AppSchedulable> runnableAppScheds = // apps that are runnable
- new ArrayList<AppSchedulable>();
- private final List<AppSchedulable> nonRunnableAppScheds =
- new ArrayList<AppSchedulable>();
+
+ private final List<FSAppAttempt> runnableApps = // apps that are runnable
+ new ArrayList<FSAppAttempt>();
+ private final List<FSAppAttempt> nonRunnableApps =
+ new ArrayList<FSAppAttempt>();
private Resource demand = Resources.createResource(0);
@@ -70,33 +70,31 @@ public class FSLeafQueue extends FSQueue
amResourceUsage = Resource.newInstance(0, 0);
}
- public void addApp(FSSchedulerApp app, boolean runnable) {
- AppSchedulable appSchedulable = new AppSchedulable(scheduler, app, this);
- app.setAppSchedulable(appSchedulable);
+ public void addApp(FSAppAttempt app, boolean runnable) {
if (runnable) {
- runnableAppScheds.add(appSchedulable);
+ runnableApps.add(app);
} else {
- nonRunnableAppScheds.add(appSchedulable);
+ nonRunnableApps.add(app);
}
}
// for testing
- void addAppSchedulable(AppSchedulable appSched) {
- runnableAppScheds.add(appSched);
+ void addAppSchedulable(FSAppAttempt appSched) {
+ runnableApps.add(appSched);
}
/**
* Removes the given app from this queue.
* @return whether or not the app was runnable
*/
- public boolean removeApp(FSSchedulerApp app) {
- if (runnableAppScheds.remove(app.getAppSchedulable())) {
+ public boolean removeApp(FSAppAttempt app) {
+ if (runnableApps.remove(app)) {
// Update AM resource usage
if (app.isAmRunning() && app.getAMResource() != null) {
Resources.subtractFrom(amResourceUsage, app.getAMResource());
}
return true;
- } else if (nonRunnableAppScheds.remove(app.getAppSchedulable())) {
+ } else if (nonRunnableApps.remove(app)) {
return false;
} else {
throw new IllegalStateException("Given app to remove " + app +
@@ -104,22 +102,22 @@ public class FSLeafQueue extends FSQueue
}
}
- public Collection<AppSchedulable> getRunnableAppSchedulables() {
- return runnableAppScheds;
+ public Collection<FSAppAttempt> getRunnableAppSchedulables() {
+ return runnableApps;
}
- public List<AppSchedulable> getNonRunnableAppSchedulables() {
- return nonRunnableAppScheds;
+ public List<FSAppAttempt> getNonRunnableAppSchedulables() {
+ return nonRunnableApps;
}
@Override
public void collectSchedulerApplications(
Collection<ApplicationAttemptId> apps) {
- for (AppSchedulable appSched : runnableAppScheds) {
- apps.add(appSched.getApp().getApplicationAttemptId());
+ for (FSAppAttempt appSched : runnableApps) {
+ apps.add(appSched.getApplicationAttemptId());
}
- for (AppSchedulable appSched : nonRunnableAppScheds) {
- apps.add(appSched.getApp().getApplicationAttemptId());
+ for (FSAppAttempt appSched : nonRunnableApps) {
+ apps.add(appSched.getApplicationAttemptId());
}
}
@@ -145,10 +143,10 @@ public class FSLeafQueue extends FSQueue
@Override
public Resource getResourceUsage() {
Resource usage = Resources.createResource(0);
- for (AppSchedulable app : runnableAppScheds) {
+ for (FSAppAttempt app : runnableApps) {
Resources.addTo(usage, app.getResourceUsage());
}
- for (AppSchedulable app : nonRunnableAppScheds) {
+ for (FSAppAttempt app : nonRunnableApps) {
Resources.addTo(usage, app.getResourceUsage());
}
return usage;
@@ -165,13 +163,13 @@ public class FSLeafQueue extends FSQueue
Resource maxRes = scheduler.getAllocationConfiguration()
.getMaxResources(getName());
demand = Resources.createResource(0);
- for (AppSchedulable sched : runnableAppScheds) {
+ for (FSAppAttempt sched : runnableApps) {
if (Resources.equals(demand, maxRes)) {
break;
}
updateDemandForApp(sched, maxRes);
}
- for (AppSchedulable sched : nonRunnableAppScheds) {
+ for (FSAppAttempt sched : nonRunnableApps) {
if (Resources.equals(demand, maxRes)) {
break;
}
@@ -183,7 +181,7 @@ public class FSLeafQueue extends FSQueue
}
}
- private void updateDemandForApp(AppSchedulable sched, Resource maxRes) {
+ private void updateDemandForApp(FSAppAttempt sched, Resource maxRes) {
sched.updateDemand();
Resource toAdd = sched.getDemand();
if (LOG.isDebugEnabled()) {
@@ -207,9 +205,9 @@ public class FSLeafQueue extends FSQueue
}
Comparator<Schedulable> comparator = policy.getComparator();
- Collections.sort(runnableAppScheds, comparator);
- for (AppSchedulable sched : runnableAppScheds) {
- if (SchedulerAppUtils.isBlacklisted(sched.getApp(), node, LOG)) {
+ Collections.sort(runnableApps, comparator);
+ for (FSAppAttempt sched : runnableApps) {
+ if (SchedulerAppUtils.isBlacklisted(sched, node, LOG)) {
continue;
}
@@ -237,8 +235,8 @@ public class FSLeafQueue extends FSQueue
// Choose the app that is most over fair share
Comparator<Schedulable> comparator = policy.getComparator();
- AppSchedulable candidateSched = null;
- for (AppSchedulable sched : runnableAppScheds) {
+ FSAppAttempt candidateSched = null;
+ for (FSAppAttempt sched : runnableApps) {
if (candidateSched == null ||
comparator.compare(sched, candidateSched) > 0) {
candidateSched = sched;
@@ -291,7 +289,7 @@ public class FSLeafQueue extends FSQueue
@Override
public int getNumRunnableApps() {
- return runnableAppScheds.size();
+ return runnableApps.size();
}
@Override
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java Tue Aug 12 21:43:27 2014
@@ -39,7 +39,8 @@ import org.apache.hadoop.yarn.util.resou
@Private
@Unstable
-public abstract class FSQueue extends Schedulable implements Queue {
+public abstract class FSQueue implements Queue, Schedulable {
+ private Resource fairShare = Resources.createResource(0, 0);
private final String name;
protected final FairScheduler scheduler;
private final FSQueueMetrics metrics;
@@ -139,10 +140,15 @@ public abstract class FSQueue extends Sc
public FSQueueMetrics getMetrics() {
return metrics;
}
-
+
+ /** Get the fair share assigned to this Schedulable. */
+ public Resource getFairShare() {
+ return fairShare;
+ }
+
@Override
public void setFairShare(Resource fairShare) {
- super.setFairShare(fairShare);
+ this.fairShare = fairShare;
metrics.setFairShare(fairShare);
}
@@ -187,4 +193,16 @@ public abstract class FSQueue extends Sc
}
return true;
}
+
+ @Override
+ public boolean isActive() {
+ return getNumRunnableApps() > 0;
+ }
+
+ /** Convenient toString implementation for debugging. */
+ @Override
+ public String toString() {
+ return String.format("[%s, demand=%s, running=%s, share=%s, w=%s]",
+ getName(), getDemand(), getResourceUsage(), fairShare, getWeights());
+ }
}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java Tue Aug 12 21:43:27 2014
@@ -35,7 +35,7 @@ public class FSSchedulerNode extends Sch
private static final Log LOG = LogFactory.getLog(FSSchedulerNode.class);
- private AppSchedulable reservedAppSchedulable;
+ private FSAppAttempt reservedAppSchedulable;
public FSSchedulerNode(RMNode node, boolean usePortForNodeName) {
super(node, usePortForNodeName);
@@ -76,7 +76,7 @@ public class FSSchedulerNode extends Sch
" on node " + this + " for application " + application);
}
setReservedContainer(container);
- this.reservedAppSchedulable = ((FSSchedulerApp) application).getAppSchedulable();
+ this.reservedAppSchedulable = (FSAppAttempt) application;
}
@Override
@@ -98,7 +98,7 @@ public class FSSchedulerNode extends Sch
this.reservedAppSchedulable = null;
}
- public synchronized AppSchedulable getReservedAppSchedulable() {
+ public synchronized FSAppAttempt getReservedAppSchedulable() {
return reservedAppSchedulable;
}
}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java Tue Aug 12 21:43:27 2014
@@ -117,7 +117,7 @@ import com.google.common.base.Preconditi
@Unstable
@SuppressWarnings("unchecked")
public class FairScheduler extends
- AbstractYarnScheduler<FSSchedulerApp, FSSchedulerNode> {
+ AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> {
private FairSchedulerConfiguration conf;
private Resource incrAllocation;
@@ -432,8 +432,8 @@ public class FairScheduler extends
try {
// Reset preemptedResource for each app
for (FSLeafQueue queue : getQueueManager().getLeafQueues()) {
- for (AppSchedulable app : queue.getRunnableAppSchedulables()) {
- app.getApp().resetPreemptedResources();
+ for (FSAppAttempt app : queue.getRunnableAppSchedulables()) {
+ app.resetPreemptedResources();
}
}
@@ -453,8 +453,8 @@ public class FairScheduler extends
} finally {
// Clear preemptedResources for each app
for (FSLeafQueue queue : getQueueManager().getLeafQueues()) {
- for (AppSchedulable app : queue.getRunnableAppSchedulables()) {
- app.getApp().clearPreemptedResources();
+ for (FSAppAttempt app : queue.getRunnableAppSchedulables()) {
+ app.clearPreemptedResources();
}
}
}
@@ -465,7 +465,7 @@ public class FairScheduler extends
protected void warnOrKillContainer(RMContainer container) {
ApplicationAttemptId appAttemptId = container.getApplicationAttemptId();
- FSSchedulerApp app = getSchedulerApp(appAttemptId);
+ FSAppAttempt app = getSchedulerApp(appAttemptId);
FSLeafQueue queue = app.getQueue();
LOG.info("Preempting container (prio=" + container.getContainer().getPriority() +
"res=" + container.getContainer().getResource() +
@@ -490,7 +490,7 @@ public class FairScheduler extends
(getClock().getTime() - time) + "ms)");
}
} else {
- // track the request in the FSSchedulerApp itself
+ // track the request in the FSAppAttempt itself
app.addPreemption(container, getClock().getTime());
}
}
@@ -541,7 +541,7 @@ public class FairScheduler extends
}
// synchronized for sizeBasedWeight
- public synchronized ResourceWeights getAppWeight(AppSchedulable app) {
+ public synchronized ResourceWeights getAppWeight(FSAppAttempt app) {
double weight = 1.0;
if (sizeBasedWeight) {
// Set weight based on current memory demand
@@ -636,8 +636,8 @@ public class FairScheduler extends
return;
}
- SchedulerApplication<FSSchedulerApp> application =
- new SchedulerApplication<FSSchedulerApp>(queue, user);
+ SchedulerApplication<FSAppAttempt> application =
+ new SchedulerApplication<FSAppAttempt>(queue, user);
applications.put(applicationId, application);
queue.getMetrics().submitApp(user);
@@ -661,13 +661,13 @@ public class FairScheduler extends
ApplicationAttemptId applicationAttemptId,
boolean transferStateFromPreviousAttempt,
boolean isAttemptRecovering) {
- SchedulerApplication<FSSchedulerApp> application =
+ SchedulerApplication<FSAppAttempt> application =
applications.get(applicationAttemptId.getApplicationId());
String user = application.getUser();
FSLeafQueue queue = (FSLeafQueue) application.getQueue();
- FSSchedulerApp attempt =
- new FSSchedulerApp(applicationAttemptId, user,
+ FSAppAttempt attempt =
+ new FSAppAttempt(this, applicationAttemptId, user,
queue, new ActiveUsersManager(getRootQueueMetrics()),
rmContext);
if (transferStateFromPreviousAttempt) {
@@ -742,7 +742,7 @@ public class FairScheduler extends
private synchronized void removeApplication(ApplicationId applicationId,
RMAppState finalState) {
- SchedulerApplication<FSSchedulerApp> application =
+ SchedulerApplication<FSAppAttempt> application =
applications.get(applicationId);
if (application == null){
LOG.warn("Couldn't find application " + applicationId);
@@ -757,9 +757,9 @@ public class FairScheduler extends
RMAppAttemptState rmAppAttemptFinalState, boolean keepContainers) {
LOG.info("Application " + applicationAttemptId + " is done." +
" finalState=" + rmAppAttemptFinalState);
- SchedulerApplication<FSSchedulerApp> application =
+ SchedulerApplication<FSAppAttempt> application =
applications.get(applicationAttemptId.getApplicationId());
- FSSchedulerApp attempt = getSchedulerApp(applicationAttemptId);
+ FSAppAttempt attempt = getSchedulerApp(applicationAttemptId);
if (attempt == null || application == null) {
LOG.info("Unknown application " + applicationAttemptId + " has completed!");
@@ -820,7 +820,7 @@ public class FairScheduler extends
Container container = rmContainer.getContainer();
// Get the application for the finished container
- FSSchedulerApp application =
+ FSAppAttempt application =
getCurrentAttemptForContainer(container.getId());
ApplicationId appId =
container.getId().getApplicationAttemptId().getApplicationId();
@@ -835,8 +835,7 @@ public class FairScheduler extends
FSSchedulerNode node = getFSSchedulerNode(container.getNodeId());
if (rmContainer.getState() == RMContainerState.RESERVED) {
- application.unreserve(node, rmContainer.getReservedPriority());
- node.unreserveResource(application);
+ application.unreserve(rmContainer.getReservedPriority(), node);
} else {
application.containerCompleted(rmContainer, containerStatus, event);
node.releaseContainer(container);
@@ -896,7 +895,7 @@ public class FairScheduler extends
List<ResourceRequest> ask, List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals) {
// Make sure this application exists
- FSSchedulerApp application = getSchedulerApp(appAttemptId);
+ FSAppAttempt application = getSchedulerApp(appAttemptId);
if (application == null) {
LOG.info("Calling allocate on removed " +
"or non existant application " + appAttemptId);
@@ -1066,13 +1065,13 @@ public class FairScheduler extends
// 1. Check for reserved applications
// 2. Schedule if there are no reservations
- AppSchedulable reservedAppSchedulable = node.getReservedAppSchedulable();
+ FSAppAttempt reservedAppSchedulable = node.getReservedAppSchedulable();
if (reservedAppSchedulable != null) {
Priority reservedPriority = node.getReservedContainer().getReservedPriority();
if (!reservedAppSchedulable.hasContainerForNode(reservedPriority, node)) {
// Don't hold the reservation if app can no longer use it
LOG.info("Releasing reservation that cannot be satisfied for application "
- + reservedAppSchedulable.getApp().getApplicationAttemptId()
+ + reservedAppSchedulable.getApplicationAttemptId()
+ " on node " + node);
reservedAppSchedulable.unreserve(reservedPriority, node);
reservedAppSchedulable = null;
@@ -1080,7 +1079,7 @@ public class FairScheduler extends
// Reservation exists; try to fulfill the reservation
if (LOG.isDebugEnabled()) {
LOG.debug("Trying to fulfill reservation for application "
- + reservedAppSchedulable.getApp().getApplicationAttemptId()
+ + reservedAppSchedulable.getApplicationAttemptId()
+ " on node: " + node);
}
@@ -1105,8 +1104,8 @@ public class FairScheduler extends
updateRootQueueMetrics();
}
- public FSSchedulerApp getSchedulerApp(ApplicationAttemptId appAttemptId) {
- return (FSSchedulerApp) super.getApplicationAttempt(appAttemptId);
+ public FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId) {
+ return super.getApplicationAttempt(appAttemptId);
}
/**
@@ -1268,8 +1267,8 @@ public class FairScheduler extends
fsOpDurations = FSOpDurations.getInstance(true);
// This stores per-application scheduling information
- this.applications =
- new ConcurrentHashMap<ApplicationId,SchedulerApplication<FSSchedulerApp>>();
+ this.applications = new ConcurrentHashMap<
+ ApplicationId, SchedulerApplication<FSAppAttempt>>();
this.eventLog = new FairSchedulerEventLog();
eventLog.init(this.conf);
@@ -1369,7 +1368,7 @@ public class FairScheduler extends
@Override
public List<QueueUserACLInfo> getQueueUserAclInfo() {
- UserGroupInformation user = null;
+ UserGroupInformation user;
try {
user = UserGroupInformation.getCurrentUser();
} catch (IOException ioe) {
@@ -1431,11 +1430,11 @@ public class FairScheduler extends
@Override
public synchronized String moveApplication(ApplicationId appId,
String queueName) throws YarnException {
- SchedulerApplication<FSSchedulerApp> app = applications.get(appId);
+ SchedulerApplication<FSAppAttempt> app = applications.get(appId);
if (app == null) {
throw new YarnException("App to be moved " + appId + " not found.");
}
- FSSchedulerApp attempt = (FSSchedulerApp) app.getCurrentAppAttempt();
+ FSAppAttempt attempt = (FSAppAttempt) app.getCurrentAppAttempt();
// To serialize with FairScheduler#allocate, synchronize on app attempt
synchronized (attempt) {
FSLeafQueue oldQueue = (FSLeafQueue) app.getQueue();
@@ -1448,8 +1447,7 @@ public class FairScheduler extends
return oldQueue.getQueueName();
}
- if (oldQueue.getRunnableAppSchedulables().contains(
- attempt.getAppSchedulable())) {
+ if (oldQueue.getRunnableAppSchedulables().contains(attempt)) {
verifyMoveDoesNotViolateConstraints(attempt, oldQueue, targetQueue);
}
@@ -1458,7 +1456,7 @@ public class FairScheduler extends
}
}
- private void verifyMoveDoesNotViolateConstraints(FSSchedulerApp app,
+ private void verifyMoveDoesNotViolateConstraints(FSAppAttempt app,
FSLeafQueue oldQueue, FSLeafQueue targetQueue) throws YarnException {
String queueName = targetQueue.getQueueName();
ApplicationAttemptId appAttId = app.getApplicationAttemptId();
@@ -1495,8 +1493,8 @@ public class FairScheduler extends
* Helper for moveApplication, which has appropriate synchronization, so all
* operations will be atomic.
*/
- private void executeMove(SchedulerApplication<FSSchedulerApp> app,
- FSSchedulerApp attempt, FSLeafQueue oldQueue, FSLeafQueue newQueue) {
+ private void executeMove(SchedulerApplication<FSAppAttempt> app,
+ FSAppAttempt attempt, FSLeafQueue oldQueue, FSLeafQueue newQueue) {
boolean wasRunnable = oldQueue.removeApp(attempt);
// if app was not runnable before, it may be runnable now
boolean nowRunnable = maxRunningEnforcer.canAppBeRunnable(newQueue,
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FifoAppComparator.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FifoAppComparator.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FifoAppComparator.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FifoAppComparator.java Tue Aug 12 21:43:27 2014
@@ -25,15 +25,15 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.classification.InterfaceStability.Unstable;
/**
- * Order {@link AppSchedulable} objects by priority and then by submit time, as
+ * Order {@link FSAppAttempt} objects by priority and then by submit time, as
* in the default scheduler in Hadoop.
*/
@Private
@Unstable
-public class FifoAppComparator implements Comparator<AppSchedulable>, Serializable {
+public class FifoAppComparator implements Comparator<FSAppAttempt>, Serializable {
private static final long serialVersionUID = 3428835083489547918L;
- public int compare(AppSchedulable a1, AppSchedulable a2) {
+ public int compare(FSAppAttempt a1, FSAppAttempt a2) {
int res = a1.getPriority().compareTo(a2.getPriority());
if (res == 0) {
if (a1.getStartTime() < a2.getStartTime()) {
@@ -44,7 +44,7 @@ public class FifoAppComparator implement
}
if (res == 0) {
// If there is a tie, break it by app ID to get a deterministic order
- res = a1.getApp().getApplicationId().compareTo(a2.getApp().getApplicationId());
+ res = a1.getApplicationId().compareTo(a2.getApplicationId());
}
return res;
}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/MaxRunningAppsEnforcer.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/MaxRunningAppsEnforcer.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/MaxRunningAppsEnforcer.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/MaxRunningAppsEnforcer.java Tue Aug 12 21:43:27 2014
@@ -43,7 +43,7 @@ public class MaxRunningAppsEnforcer {
// Tracks the number of running applications by user.
private final Map<String, Integer> usersNumRunnableApps;
@VisibleForTesting
- final ListMultimap<String, AppSchedulable> usersNonRunnableApps;
+ final ListMultimap<String, FSAppAttempt> usersNonRunnableApps;
public MaxRunningAppsEnforcer(FairScheduler scheduler) {
this.scheduler = scheduler;
@@ -80,7 +80,7 @@ public class MaxRunningAppsEnforcer {
* Tracks the given new runnable app for purposes of maintaining max running
* app limits.
*/
- public void trackRunnableApp(FSSchedulerApp app) {
+ public void trackRunnableApp(FSAppAttempt app) {
String user = app.getUser();
FSLeafQueue queue = app.getQueue();
// Increment running counts for all parent queues
@@ -99,9 +99,9 @@ public class MaxRunningAppsEnforcer {
* Tracks the given new non runnable app so that it can be made runnable when
* it would not violate max running app limits.
*/
- public void trackNonRunnableApp(FSSchedulerApp app) {
+ public void trackNonRunnableApp(FSAppAttempt app) {
String user = app.getUser();
- usersNonRunnableApps.put(user, app.getAppSchedulable());
+ usersNonRunnableApps.put(user, app);
}
/**
@@ -111,7 +111,7 @@ public class MaxRunningAppsEnforcer {
* Runs in O(n log(n)) where n is the number of queues that are under the
* highest queue that went from having no slack to having slack.
*/
- public void updateRunnabilityOnAppRemoval(FSSchedulerApp app, FSLeafQueue queue) {
+ public void updateRunnabilityOnAppRemoval(FSAppAttempt app, FSLeafQueue queue) {
AllocationConfiguration allocConf = scheduler.getAllocationConfiguration();
// childqueueX might have no pending apps itself, but if a queue higher up
@@ -133,8 +133,8 @@ public class MaxRunningAppsEnforcer {
parent = parent.getParent();
}
- List<List<AppSchedulable>> appsNowMaybeRunnable =
- new ArrayList<List<AppSchedulable>>();
+ List<List<FSAppAttempt>> appsNowMaybeRunnable =
+ new ArrayList<List<FSAppAttempt>>();
// Compile lists of apps which may now be runnable
// We gather lists instead of building a set of all non-runnable apps so
@@ -150,26 +150,26 @@ public class MaxRunningAppsEnforcer {
userNumRunning = 0;
}
if (userNumRunning == allocConf.getUserMaxApps(user) - 1) {
- List<AppSchedulable> userWaitingApps = usersNonRunnableApps.get(user);
+ List<FSAppAttempt> userWaitingApps = usersNonRunnableApps.get(user);
if (userWaitingApps != null) {
appsNowMaybeRunnable.add(userWaitingApps);
}
}
// Scan through and check whether this means that any apps are now runnable
- Iterator<FSSchedulerApp> iter = new MultiListStartTimeIterator(
+ Iterator<FSAppAttempt> iter = new MultiListStartTimeIterator(
appsNowMaybeRunnable);
- FSSchedulerApp prev = null;
- List<AppSchedulable> noLongerPendingApps = new ArrayList<AppSchedulable>();
+ FSAppAttempt prev = null;
+ List<FSAppAttempt> noLongerPendingApps = new ArrayList<FSAppAttempt>();
while (iter.hasNext()) {
- FSSchedulerApp next = iter.next();
+ FSAppAttempt next = iter.next();
if (next == prev) {
continue;
}
if (canAppBeRunnable(next.getQueue(), next.getUser())) {
trackRunnableApp(next);
- AppSchedulable appSched = next.getAppSchedulable();
+ FSAppAttempt appSched = next;
next.getQueue().getRunnableAppSchedulables().add(appSched);
noLongerPendingApps.add(appSched);
@@ -186,14 +186,14 @@ public class MaxRunningAppsEnforcer {
// We remove the apps from their pending lists afterwards so that we don't
// pull them out from under the iterator. If they are not in these lists
// in the first place, there is a bug.
- for (AppSchedulable appSched : noLongerPendingApps) {
- if (!appSched.getApp().getQueue().getNonRunnableAppSchedulables()
+ for (FSAppAttempt appSched : noLongerPendingApps) {
+ if (!appSched.getQueue().getNonRunnableAppSchedulables()
.remove(appSched)) {
LOG.error("Can't make app runnable that does not already exist in queue"
+ " as non-runnable: " + appSched + ". This should never happen.");
}
- if (!usersNonRunnableApps.remove(appSched.getApp().getUser(), appSched)) {
+ if (!usersNonRunnableApps.remove(appSched.getUser(), appSched)) {
LOG.error("Waiting app " + appSched + " expected to be in "
+ "usersNonRunnableApps, but was not. This should never happen.");
}
@@ -204,7 +204,7 @@ public class MaxRunningAppsEnforcer {
* Updates the relevant tracking variables after a runnable app with the given
* queue and user has been removed.
*/
- public void untrackRunnableApp(FSSchedulerApp app) {
+ public void untrackRunnableApp(FSAppAttempt app) {
// Update usersRunnableApps
String user = app.getUser();
int newUserNumRunning = usersNumRunnableApps.get(user) - 1;
@@ -226,8 +226,8 @@ public class MaxRunningAppsEnforcer {
/**
* Stops tracking the given non-runnable app
*/
- public void untrackNonRunnableApp(FSSchedulerApp app) {
- usersNonRunnableApps.remove(app.getUser(), app.getAppSchedulable());
+ public void untrackNonRunnableApp(FSAppAttempt app) {
+ usersNonRunnableApps.remove(app.getUser(), app);
}
/**
@@ -235,7 +235,7 @@ public class MaxRunningAppsEnforcer {
* of non-runnable applications.
*/
private void gatherPossiblyRunnableAppLists(FSQueue queue,
- List<List<AppSchedulable>> appLists) {
+ List<List<FSAppAttempt>> appLists) {
if (queue.getNumRunnableApps() < scheduler.getAllocationConfiguration()
.getQueueMaxApps(queue.getName())) {
if (queue instanceof FSLeafQueue) {
@@ -259,14 +259,14 @@ public class MaxRunningAppsEnforcer {
* of O(num lists) time.
*/
static class MultiListStartTimeIterator implements
- Iterator<FSSchedulerApp> {
+ Iterator<FSAppAttempt> {
- private List<AppSchedulable>[] appLists;
+ private List<FSAppAttempt>[] appLists;
private int[] curPositionsInAppLists;
private PriorityQueue<IndexAndTime> appListsByCurStartTime;
@SuppressWarnings("unchecked")
- public MultiListStartTimeIterator(List<List<AppSchedulable>> appListList) {
+ public MultiListStartTimeIterator(List<List<FSAppAttempt>> appListList) {
appLists = appListList.toArray(new List[appListList.size()]);
curPositionsInAppLists = new int[appLists.length];
appListsByCurStartTime = new PriorityQueue<IndexAndTime>();
@@ -284,10 +284,10 @@ public class MaxRunningAppsEnforcer {
}
@Override
- public FSSchedulerApp next() {
+ public FSAppAttempt next() {
IndexAndTime indexAndTime = appListsByCurStartTime.remove();
int nextListIndex = indexAndTime.index;
- AppSchedulable next = appLists[nextListIndex]
+ FSAppAttempt next = appLists[nextListIndex]
.get(curPositionsInAppLists[nextListIndex]);
curPositionsInAppLists[nextListIndex]++;
@@ -299,7 +299,7 @@ public class MaxRunningAppsEnforcer {
}
appListsByCurStartTime.add(indexAndTime);
- return next.getApp();
+ return next;
}
@Override
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/NewAppWeightBooster.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/NewAppWeightBooster.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/NewAppWeightBooster.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/NewAppWeightBooster.java Tue Aug 12 21:43:27 2014
@@ -48,7 +48,7 @@ public class NewAppWeightBooster extends
super.setConf(conf);
}
- public double adjustWeight(AppSchedulable app, double curWeight) {
+ public double adjustWeight(FSAppAttempt app, double curWeight) {
long start = app.getStartTime();
long now = System.currentTimeMillis();
if (now - start < duration) {
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java Tue Aug 12 21:43:27 2014
@@ -27,20 +27,14 @@ import org.apache.hadoop.yarn.server.res
import org.apache.hadoop.yarn.util.resource.Resources;
/**
- * A Schedulable represents an entity that can launch tasks, such as a job
- * or a queue. It provides a common interface so that algorithms such as fair
- * sharing can be applied both within a queue and across queues. There are
- * currently two types of Schedulables: JobSchedulables, which represent a
- * single job, and QueueSchedulables, which allocate among jobs in their queue.
- *
- * Separate sets of Schedulables are used for maps and reduces. Each queue has
- * both a mapSchedulable and a reduceSchedulable, and so does each job.
+ * A Schedulable represents an entity that can be scheduled such as an
+ * application or a queue. It provides a common interface so that algorithms
+ * such as fair sharing can be applied both within a queue and across queues.
*
* A Schedulable is responsible for three roles:
- * 1) It can launch tasks through assignTask().
- * 2) It provides information about the job/queue to the scheduler, including:
+ * 1) Assign resources through {@link #assignContainer}.
+ * 2) It provides information about the app/queue to the scheduler, including:
* - Demand (maximum number of tasks required)
- * - Number of currently running tasks
* - Minimum share (for queues)
* - Job/queue weight (for fair sharing)
* - Start time and priority (for FIFO)
@@ -57,81 +51,61 @@ import org.apache.hadoop.yarn.util.resou
*/
@Private
@Unstable
-public abstract class Schedulable {
- /** Fair share assigned to this Schedulable */
- private Resource fairShare = Resources.createResource(0);
-
+public interface Schedulable {
/**
* Name of job/queue, used for debugging as well as for breaking ties in
* scheduling order deterministically.
*/
- public abstract String getName();
+ public String getName();
/**
* Maximum number of resources required by this Schedulable. This is defined as
* number of currently utilized resources + number of unlaunched resources (that
* are either not yet launched or need to be speculated).
*/
- public abstract Resource getDemand();
+ public Resource getDemand();
/** Get the aggregate amount of resources consumed by the schedulable. */
- public abstract Resource getResourceUsage();
+ public Resource getResourceUsage();
/** Minimum Resource share assigned to the schedulable. */
- public abstract Resource getMinShare();
+ public Resource getMinShare();
/** Maximum Resource share assigned to the schedulable. */
- public abstract Resource getMaxShare();
+ public Resource getMaxShare();
/** Job/queue weight in fair sharing. */
- public abstract ResourceWeights getWeights();
+ public ResourceWeights getWeights();
/** Start time for jobs in FIFO queues; meaningless for QueueSchedulables.*/
- public abstract long getStartTime();
+ public long getStartTime();
/** Job priority for jobs in FIFO queues; meaningless for QueueSchedulables. */
- public abstract Priority getPriority();
+ public Priority getPriority();
/** Refresh the Schedulable's demand and those of its children if any. */
- public abstract void updateDemand();
+ public void updateDemand();
/**
* Assign a container on this node if possible, and return the amount of
* resources assigned.
*/
- public abstract Resource assignContainer(FSSchedulerNode node);
+ public Resource assignContainer(FSSchedulerNode node);
/**
* Preempt a container from this Schedulable if possible.
*/
- public abstract RMContainer preemptContainer();
-
- /** Assign a fair share to this Schedulable. */
- public void setFairShare(Resource fairShare) {
- this.fairShare = fairShare;
- }
+ public RMContainer preemptContainer();
/** Get the fair share assigned to this Schedulable. */
- public Resource getFairShare() {
- return fairShare;
- }
+ public Resource getFairShare();
+
+ /** Assign a fair share to this Schedulable. */
+ public void setFairShare(Resource fairShare);
/**
* Returns true if queue has atleast one app running. Always returns true for
* AppSchedulables.
*/
- public boolean isActive() {
- if (this instanceof FSQueue) {
- FSQueue queue = (FSQueue) this;
- return queue.getNumRunnableApps() > 0;
- }
- return true;
- }
-
- /** Convenient toString implementation for debugging. */
- @Override
- public String toString() {
- return String.format("[%s, demand=%s, running=%s, share=%s, w=%s]",
- getName(), getDemand(), getResourceUsage(), fairShare, getWeights());
- }
+ public boolean isActive();
}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/WeightAdjuster.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/WeightAdjuster.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/WeightAdjuster.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/WeightAdjuster.java Tue Aug 12 21:43:27 2014
@@ -32,5 +32,5 @@ import org.apache.hadoop.conf.Configurab
@Private
@Unstable
public interface WeightAdjuster {
- public double adjustWeight(AppSchedulable app, double curWeight);
+ public double adjustWeight(FSAppAttempt app, double curWeight);
}
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerInfo.java Tue Aug 12 21:43:27 2014
@@ -46,8 +46,7 @@ public class FairSchedulerInfo extends S
}
public int getAppFairShare(ApplicationAttemptId appAttemptId) {
- return scheduler.getSchedulerApp(appAttemptId).
- getAppSchedulable().getFairShare().getMemory();
+ return scheduler.getSchedulerApp(appAttemptId).getFairShare().getMemory();
}
public FairSchedulerQueueInfo getRootQueueInfo() {
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/dao/FairSchedulerLeafQueueInfo.java Tue Aug 12 21:43:27 2014
@@ -24,7 +24,8 @@ import javax.xml.bind.annotation.XmlAcce
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.AppSchedulable;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair
+ .FSAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
@@ -39,9 +40,9 @@ public class FairSchedulerLeafQueueInfo
public FairSchedulerLeafQueueInfo(FSLeafQueue queue, FairScheduler scheduler) {
super(queue, scheduler);
- Collection<AppSchedulable> apps = queue.getRunnableAppSchedulables();
- for (AppSchedulable app : apps) {
- if (app.getApp().isPending()) {
+ Collection<FSAppAttempt> apps = queue.getRunnableAppSchedulables();
+ for (FSAppAttempt app : apps) {
+ if (app.isPending()) {
numPendingApps++;
} else {
numActiveApps++;
Modified: hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java?rev=1617600&r1=1617599&r2=1617600&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java (original)
+++ hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FakeSchedulable.java Tue Aug 12 21:43:27 2014
@@ -28,10 +28,11 @@ import org.apache.hadoop.yarn.util.resou
/**
* Dummy implementation of Schedulable for unit testing.
*/
-public class FakeSchedulable extends Schedulable {
+public class FakeSchedulable implements Schedulable {
private Resource usage;
private Resource minShare;
private Resource maxShare;
+ private Resource fairShare;
private ResourceWeights weights;
private Priority priority;
private long startTime;
@@ -90,6 +91,21 @@ public class FakeSchedulable extends Sch
}
@Override
+ public Resource getFairShare() {
+ return this.fairShare;
+ }
+
+ @Override
+ public void setFairShare(Resource fairShare) {
+ this.fairShare = fairShare;
+ }
+
+ @Override
+ public boolean isActive() {
+ return true;
+ }
+
+ @Override
public Resource getDemand() {
return null;
}