| /** |
| * 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.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentSkipListMap; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.commons.collections.IteratorUtils; |
| 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.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.Container; |
| import org.apache.hadoop.yarn.api.records.ContainerId; |
| import org.apache.hadoop.yarn.api.records.NodeId; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.ResourceRequest; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.yarn.server.resourcemanager.RMServerUtils; |
| import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; |
| import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; |
| import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.PlacementSet; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.ResourceRequestUpdateResult; |
| import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet; |
| import org.apache.hadoop.yarn.util.resource.Resources; |
| |
| /** |
| * This class keeps track of all the consumption of an application. This also |
| * keeps track of current running/completed containers for the application. |
| */ |
| @Private |
| @Unstable |
| public class AppSchedulingInfo { |
| |
| private static final Log LOG = LogFactory.getLog(AppSchedulingInfo.class); |
| |
| private final ApplicationId applicationId; |
| private final ApplicationAttemptId applicationAttemptId; |
| private final AtomicLong containerIdCounter; |
| private final String user; |
| |
| private Queue queue; |
| private ActiveUsersManager activeUsersManager; |
| // whether accepted/allocated by scheduler |
| private volatile boolean pending = true; |
| private ResourceUsage appResourceUsage; |
| |
| private AtomicBoolean userBlacklistChanged = new AtomicBoolean(false); |
| // Set of places (nodes / racks) blacklisted by the system. Today, this only |
| // has places blacklisted for AM containers. |
| private final Set<String> placesBlacklistedBySystem = new HashSet<>(); |
| private Set<String> placesBlacklistedByApp = new HashSet<>(); |
| |
| private Set<String> requestedPartitions = new HashSet<>(); |
| |
| private final ConcurrentSkipListMap<SchedulerRequestKey, Integer> |
| schedulerKeys = new ConcurrentSkipListMap<>(); |
| final Map<SchedulerRequestKey, Map<String, ResourceRequest>> |
| resourceRequestMap = new ConcurrentHashMap<>(); |
| final Map<NodeId, Map<SchedulerRequestKey, Map<ContainerId, |
| SchedContainerChangeRequest>>> containerIncreaseRequestMap = |
| new ConcurrentHashMap<>(); |
| |
| private final ReentrantReadWriteLock.ReadLock readLock; |
| private final ReentrantReadWriteLock.WriteLock writeLock; |
| |
| public AppSchedulingInfo(ApplicationAttemptId appAttemptId, |
| String user, Queue queue, ActiveUsersManager activeUsersManager, |
| long epoch, ResourceUsage appResourceUsage) { |
| this.applicationAttemptId = appAttemptId; |
| this.applicationId = appAttemptId.getApplicationId(); |
| this.queue = queue; |
| this.user = user; |
| this.activeUsersManager = activeUsersManager; |
| this.containerIdCounter = new AtomicLong( |
| epoch << ResourceManager.EPOCH_BIT_SHIFT); |
| this.appResourceUsage = appResourceUsage; |
| |
| ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); |
| readLock = lock.readLock(); |
| writeLock = lock.writeLock(); |
| } |
| |
| public ApplicationId getApplicationId() { |
| return applicationId; |
| } |
| |
| public ApplicationAttemptId getApplicationAttemptId() { |
| return applicationAttemptId; |
| } |
| |
| public String getUser() { |
| return user; |
| } |
| |
| public long getNewContainerId() { |
| return this.containerIdCounter.incrementAndGet(); |
| } |
| |
| public String getQueueName() { |
| try { |
| this.readLock.lock(); |
| return queue.getQueueName(); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| public boolean isPending() { |
| return pending; |
| } |
| |
| public Set<String> getRequestedPartitions() { |
| return requestedPartitions; |
| } |
| |
| /** |
| * Clear any pending requests from this application. |
| */ |
| private void clearRequests() { |
| schedulerKeys.clear(); |
| resourceRequestMap.clear(); |
| LOG.info("Application " + applicationId + " requests cleared"); |
| } |
| |
| public boolean hasIncreaseRequest(NodeId nodeId) { |
| try { |
| this.readLock.lock(); |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| return requestsOnNode == null ? false : requestsOnNode.size() > 0; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| public Map<ContainerId, SchedContainerChangeRequest> |
| getIncreaseRequests(NodeId nodeId, SchedulerRequestKey schedulerKey) { |
| try { |
| this.readLock.lock(); |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| return requestsOnNode == null ? null : requestsOnNode.get( |
| schedulerKey); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| /** |
| * return true if any of the existing increase requests are updated, |
| * false if none of them are updated |
| */ |
| public boolean updateIncreaseRequests( |
| List<SchedContainerChangeRequest> increaseRequests) { |
| boolean resourceUpdated = false; |
| |
| try { |
| this.writeLock.lock(); |
| for (SchedContainerChangeRequest r : increaseRequests) { |
| if (r.getRMContainer().getState() != RMContainerState.RUNNING) { |
| LOG.warn("rmContainer's state is not RUNNING, for increase request" |
| + " with container-id=" + r.getContainerId()); |
| continue; |
| } |
| try { |
| RMServerUtils.checkSchedContainerChangeRequest(r, true); |
| } catch (YarnException e) { |
| LOG.warn("Error happens when checking increase request, Ignoring.." |
| + " exception=", e); |
| continue; |
| } |
| NodeId nodeId = r.getRMContainer().getAllocatedNode(); |
| |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| if (null == requestsOnNode) { |
| requestsOnNode = new TreeMap<>(); |
| containerIncreaseRequestMap.put(nodeId, requestsOnNode); |
| } |
| |
| SchedContainerChangeRequest prevChangeRequest = |
| getIncreaseRequest(nodeId, |
| r.getRMContainer().getAllocatedSchedulerKey(), |
| r.getContainerId()); |
| if (null != prevChangeRequest) { |
| if (Resources.equals(prevChangeRequest.getTargetCapacity(), |
| r.getTargetCapacity())) { |
| // increase request hasn't changed |
| continue; |
| } |
| |
| // remove the old one, as we will use the new one going forward |
| removeIncreaseRequest(nodeId, |
| prevChangeRequest.getRMContainer().getAllocatedSchedulerKey(), |
| prevChangeRequest.getContainerId()); |
| } |
| |
| if (Resources.equals(r.getTargetCapacity(), |
| r.getRMContainer().getAllocatedResource())) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Trying to increase container " + r.getContainerId() |
| + ", target capacity = previous capacity = " + prevChangeRequest |
| + ". Will ignore this increase request."); |
| } |
| continue; |
| } |
| |
| // add the new one |
| resourceUpdated = true; |
| insertIncreaseRequest(r); |
| } |
| return resourceUpdated; |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * Insert increase request, adding any missing items in the data-structure |
| * hierarchy. |
| */ |
| private void insertIncreaseRequest(SchedContainerChangeRequest request) { |
| NodeId nodeId = request.getNodeId(); |
| SchedulerRequestKey schedulerKey = |
| request.getRMContainer().getAllocatedSchedulerKey(); |
| ContainerId containerId = request.getContainerId(); |
| |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| if (null == requestsOnNode) { |
| requestsOnNode = new HashMap<>(); |
| containerIncreaseRequestMap.put(nodeId, requestsOnNode); |
| } |
| |
| Map<ContainerId, SchedContainerChangeRequest> requestsOnNodeWithPriority = |
| requestsOnNode.get(schedulerKey); |
| if (null == requestsOnNodeWithPriority) { |
| requestsOnNodeWithPriority = new TreeMap<>(); |
| requestsOnNode.put(schedulerKey, requestsOnNodeWithPriority); |
| incrementSchedulerKeyReference(schedulerKey); |
| } |
| |
| requestsOnNodeWithPriority.put(containerId, request); |
| |
| // update resources |
| String partition = request.getRMContainer().getNodeLabelExpression(); |
| Resource delta = request.getDeltaCapacity(); |
| appResourceUsage.incPending(partition, delta); |
| queue.incPendingResource(partition, delta); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Added increase request:" + request.getContainerId() |
| + " delta=" + delta); |
| } |
| } |
| |
| private void incrementSchedulerKeyReference( |
| SchedulerRequestKey schedulerKey) { |
| Integer schedulerKeyCount = schedulerKeys.get(schedulerKey); |
| if (schedulerKeyCount == null) { |
| schedulerKeys.put(schedulerKey, 1); |
| } else { |
| schedulerKeys.put(schedulerKey, schedulerKeyCount + 1); |
| } |
| } |
| |
| private void decrementSchedulerKeyReference( |
| SchedulerRequestKey schedulerKey) { |
| Integer schedulerKeyCount = schedulerKeys.get(schedulerKey); |
| if (schedulerKeyCount != null) { |
| if (schedulerKeyCount > 1) { |
| schedulerKeys.put(schedulerKey, schedulerKeyCount - 1); |
| } else { |
| schedulerKeys.remove(schedulerKey); |
| } |
| } |
| } |
| |
| public boolean removeIncreaseRequest(NodeId nodeId, |
| SchedulerRequestKey schedulerKey, ContainerId containerId) { |
| try { |
| this.writeLock.lock(); |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| if (null == requestsOnNode) { |
| return false; |
| } |
| |
| Map<ContainerId, SchedContainerChangeRequest> requestsOnNodeWithPriority = |
| requestsOnNode.get(schedulerKey); |
| if (null == requestsOnNodeWithPriority) { |
| return false; |
| } |
| |
| SchedContainerChangeRequest request = |
| requestsOnNodeWithPriority.remove(containerId); |
| |
| // remove hierarchies if it becomes empty |
| if (requestsOnNodeWithPriority.isEmpty()) { |
| requestsOnNode.remove(schedulerKey); |
| decrementSchedulerKeyReference(schedulerKey); |
| } |
| if (requestsOnNode.isEmpty()) { |
| containerIncreaseRequestMap.remove(nodeId); |
| } |
| |
| if (request == null) { |
| return false; |
| } |
| |
| // update queue's pending resource if request exists |
| String partition = request.getRMContainer().getNodeLabelExpression(); |
| Resource delta = request.getDeltaCapacity(); |
| appResourceUsage.decPending(partition, delta); |
| queue.decPendingResource(partition, delta); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("remove increase request:" + request); |
| } |
| |
| return true; |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public SchedContainerChangeRequest getIncreaseRequest(NodeId nodeId, |
| SchedulerRequestKey schedulerKey, ContainerId containerId) { |
| try { |
| this.readLock.lock(); |
| Map<SchedulerRequestKey, Map<ContainerId, SchedContainerChangeRequest>> |
| requestsOnNode = containerIncreaseRequestMap.get(nodeId); |
| if (null == requestsOnNode) { |
| return null; |
| } |
| |
| Map<ContainerId, SchedContainerChangeRequest> requestsOnNodeWithPriority = |
| requestsOnNode.get(schedulerKey); |
| return requestsOnNodeWithPriority == null ? null |
| : requestsOnNodeWithPriority.get(containerId); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| /** |
| * The ApplicationMaster is updating resource requirements for the |
| * application, by asking for more resources and releasing resources acquired |
| * by the application. |
| * |
| * @param requests |
| * resources to be acquired |
| * @param recoverPreemptedRequestForAContainer |
| * recover ResourceRequest on preemption |
| * @return true if any resource was updated, false otherwise |
| */ |
| public boolean updateResourceRequests(List<ResourceRequest> requests, |
| boolean recoverPreemptedRequestForAContainer) { |
| // Flag to track if any incoming requests update "ANY" requests |
| boolean anyResourcesUpdated = false; |
| |
| try { |
| this.writeLock.lock(); |
| // Update resource requests |
| for (ResourceRequest request : requests) { |
| SchedulerRequestKey schedulerKey = SchedulerRequestKey.create(request); |
| String resourceName = request.getResourceName(); |
| |
| // Update node labels if required |
| updateNodeLabels(request); |
| |
| Map<String, ResourceRequest> asks = |
| this.resourceRequestMap.get(schedulerKey); |
| if (asks == null) { |
| asks = new ConcurrentHashMap<>(); |
| this.resourceRequestMap.put(schedulerKey, asks); |
| } |
| |
| // Increment number of containers if recovering preempted resources |
| ResourceRequest lastRequest = asks.get(resourceName); |
| if (recoverPreemptedRequestForAContainer && lastRequest != null) { |
| request.setNumContainers(lastRequest.getNumContainers() + 1); |
| } |
| |
| // Update asks |
| asks.put(resourceName, request); |
| |
| if (resourceName.equals(ResourceRequest.ANY)) { |
| //update the applications requested labels set |
| requestedPartitions.add(request.getNodeLabelExpression() == null |
| ? RMNodeLabelsManager.NO_LABEL : |
| request.getNodeLabelExpression()); |
| |
| anyResourcesUpdated = true; |
| |
| // Update pendingResources |
| updatePendingResources(lastRequest, request, schedulerKey, |
| queue.getMetrics()); |
| } |
| } |
| return anyResourcesUpdated; |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| private void updatePendingResources(ResourceRequest lastRequest, |
| ResourceRequest request, SchedulerRequestKey schedulerKey, |
| QueueMetrics metrics) { |
| int lastRequestContainers = |
| (lastRequest != null) ? lastRequest.getNumContainers() : 0; |
| if (request.getNumContainers() <= 0) { |
| if (lastRequestContainers >= 0) { |
| decrementSchedulerKeyReference(schedulerKey); |
| } |
| LOG.info("checking for deactivate of application :" |
| + this.applicationId); |
| checkForDeactivation(); |
| } else { |
| // Activate application. Metrics activation is done here. |
| if (lastRequestContainers <= 0) { |
| incrementSchedulerKeyReference(schedulerKey); |
| activeUsersManager.activateApplication(user, applicationId); |
| } |
| } |
| |
| Resource lastRequestCapability = |
| lastRequest != null ? lastRequest.getCapability() : Resources.none(); |
| metrics.incrPendingResources(user, |
| request.getNumContainers(), request.getCapability()); |
| metrics.decrPendingResources(user, |
| lastRequestContainers, lastRequestCapability); |
| |
| // update queue: |
| Resource increasedResource = |
| Resources.multiply(request.getCapability(), request.getNumContainers()); |
| queue.incPendingResource(request.getNodeLabelExpression(), |
| increasedResource); |
| appResourceUsage.incPending(request.getNodeLabelExpression(), |
| increasedResource); |
| if (lastRequest != null) { |
| Resource decreasedResource = |
| Resources.multiply(lastRequestCapability, lastRequestContainers); |
| queue.decPendingResource(lastRequest.getNodeLabelExpression(), |
| decreasedResource); |
| appResourceUsage.decPending(lastRequest.getNodeLabelExpression(), |
| decreasedResource); |
| } |
| } |
| |
| private void updateNodeLabels(ResourceRequest request) { |
| SchedulerRequestKey schedulerKey = SchedulerRequestKey.create(request); |
| String resourceName = request.getResourceName(); |
| if (resourceName.equals(ResourceRequest.ANY)) { |
| ResourceRequest previousAnyRequest = |
| getResourceRequest(schedulerKey, resourceName); |
| |
| // When there is change in ANY request label expression, we should |
| // update label for all resource requests already added of same |
| // priority as ANY resource request. |
| if ((null == previousAnyRequest) |
| || hasRequestLabelChanged(previousAnyRequest, request)) { |
| Map<String, ResourceRequest> resourceRequest = |
| getResourceRequests(schedulerKey); |
| if (resourceRequest != null) { |
| for (ResourceRequest r : resourceRequest.values()) { |
| if (!r.getResourceName().equals(ResourceRequest.ANY)) { |
| r.setNodeLabelExpression(request.getNodeLabelExpression()); |
| } |
| } |
| } |
| } |
| } else { |
| ResourceRequest anyRequest = |
| getResourceRequest(schedulerKey, ResourceRequest.ANY); |
| if (anyRequest != null) { |
| request.setNodeLabelExpression(anyRequest.getNodeLabelExpression()); |
| } |
| } |
| } |
| |
| private boolean hasRequestLabelChanged(ResourceRequest requestOne, |
| ResourceRequest requestTwo) { |
| String requestOneLabelExp = requestOne.getNodeLabelExpression(); |
| String requestTwoLabelExp = requestTwo.getNodeLabelExpression(); |
| // First request label expression can be null and second request |
| // is not null then we have to consider it as changed. |
| if ((null == requestOneLabelExp) && (null != requestTwoLabelExp)) { |
| return true; |
| } |
| // If the label is not matching between both request when |
| // requestOneLabelExp is not null. |
| return ((null != requestOneLabelExp) && !(requestOneLabelExp |
| .equals(requestTwoLabelExp))); |
| } |
| |
| /** |
| * The ApplicationMaster is updating the placesBlacklistedByApp used for |
| * containers other than AMs. |
| * |
| * @param blacklistAdditions |
| * resources to be added to the userBlacklist |
| * @param blacklistRemovals |
| * resources to be removed from the userBlacklist |
| */ |
| public void updatePlacesBlacklistedByApp( |
| List<String> blacklistAdditions, List<String> blacklistRemovals) { |
| if (updateBlacklistedPlaces(placesBlacklistedByApp, blacklistAdditions, |
| blacklistRemovals)) { |
| userBlacklistChanged.set(true); |
| } |
| } |
| |
| /** |
| * Update the list of places that are blacklisted by the system. Today the |
| * system only blacklists places when it sees that AMs failed there |
| * |
| * @param blacklistAdditions |
| * resources to be added to placesBlacklistedBySystem |
| * @param blacklistRemovals |
| * resources to be removed from placesBlacklistedBySystem |
| */ |
| public void updatePlacesBlacklistedBySystem( |
| List<String> blacklistAdditions, List<String> blacklistRemovals) { |
| updateBlacklistedPlaces(placesBlacklistedBySystem, blacklistAdditions, |
| blacklistRemovals); |
| } |
| |
| private static boolean updateBlacklistedPlaces(Set<String> blacklist, |
| List<String> blacklistAdditions, List<String> blacklistRemovals) { |
| boolean changed = false; |
| synchronized (blacklist) { |
| if (blacklistAdditions != null) { |
| changed = blacklist.addAll(blacklistAdditions); |
| } |
| |
| if (blacklistRemovals != null) { |
| changed = blacklist.removeAll(blacklistRemovals) || changed; |
| } |
| } |
| return changed; |
| } |
| |
| public boolean getAndResetBlacklistChanged() { |
| return userBlacklistChanged.getAndSet(false); |
| } |
| |
| public Collection<SchedulerRequestKey> getSchedulerKeys() { |
| return schedulerKeys.keySet(); |
| } |
| |
| public Map<String, ResourceRequest> getResourceRequests( |
| SchedulerRequestKey schedulerKey) { |
| return resourceRequestMap.get(schedulerKey); |
| } |
| |
| public List<ResourceRequest> getAllResourceRequests() { |
| List<ResourceRequest> ret = new ArrayList<>(); |
| try { |
| this.readLock.lock(); |
| for (Map<String, ResourceRequest> r : resourceRequestMap.values()) { |
| ret.addAll(r.values()); |
| } |
| } finally { |
| this.readLock.unlock(); |
| } |
| return ret; |
| } |
| |
| public ResourceRequest getResourceRequest(SchedulerRequestKey schedulerKey, |
| String resourceName) { |
| try { |
| this.readLock.lock(); |
| Map<String, ResourceRequest> nodeRequests = |
| resourceRequestMap.get(schedulerKey); |
| return (nodeRequests == null) ? null : nodeRequests.get(resourceName); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| public Resource getResource(SchedulerRequestKey schedulerKey) { |
| try { |
| this.readLock.lock(); |
| ResourceRequest request = |
| getResourceRequest(schedulerKey, ResourceRequest.ANY); |
| return (request == null) ? null : request.getCapability(); |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Method to return the next resource request to be serviced. |
| * |
| * In the initial implementation, we just pick any {@link ResourceRequest} |
| * corresponding to the highest priority. |
| * |
| * @return next {@link ResourceRequest} to allocate resources for. |
| */ |
| @Unstable |
| public synchronized ResourceRequest getNextResourceRequest() { |
| for (ResourceRequest rr: |
| resourceRequestMap.get(schedulerKeys.firstKey()).values()) { |
| return rr; |
| } |
| return null; |
| } |
| |
| /** |
| * Returns if the place (node/rack today) is either blacklisted by the |
| * application (user) or the system |
| * |
| * @param resourceName |
| * the resourcename |
| * @param blacklistedBySystem |
| * true if it should check amBlacklist |
| * @return true if its blacklisted |
| */ |
| public boolean isPlaceBlacklisted(String resourceName, |
| boolean blacklistedBySystem) { |
| if (blacklistedBySystem){ |
| synchronized (placesBlacklistedBySystem) { |
| return placesBlacklistedBySystem.contains(resourceName); |
| } |
| } else { |
| synchronized (placesBlacklistedByApp) { |
| return placesBlacklistedByApp.contains(resourceName); |
| } |
| } |
| } |
| |
| public void increaseContainer(SchedContainerChangeRequest increaseRequest) { |
| NodeId nodeId = increaseRequest.getNodeId(); |
| SchedulerRequestKey schedulerKey = |
| increaseRequest.getRMContainer().getAllocatedSchedulerKey(); |
| ContainerId containerId = increaseRequest.getContainerId(); |
| Resource deltaCapacity = increaseRequest.getDeltaCapacity(); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("allocated increase request : applicationId=" + applicationId |
| + " container=" + containerId + " host=" |
| + increaseRequest.getNodeId() + " user=" + user + " resource=" |
| + deltaCapacity); |
| } |
| try { |
| this.writeLock.lock(); |
| // Set queue metrics |
| queue.getMetrics().allocateResources(user, deltaCapacity); |
| // remove the increase request from pending increase request map |
| removeIncreaseRequest(nodeId, schedulerKey, containerId); |
| // update usage |
| appResourceUsage.incUsed(increaseRequest.getNodePartition(), |
| deltaCapacity); |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public void decreaseContainer(SchedContainerChangeRequest decreaseRequest) { |
| // Delta is negative when it's a decrease request |
| Resource absDelta = Resources.negate(decreaseRequest.getDeltaCapacity()); |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Decrease container : applicationId=" + applicationId |
| + " container=" + decreaseRequest.getContainerId() + " host=" |
| + decreaseRequest.getNodeId() + " user=" + user + " resource=" |
| + absDelta); |
| } |
| |
| try { |
| this.writeLock.lock(); |
| // Set queue metrics |
| queue.getMetrics().releaseResources(user, absDelta); |
| |
| // update usage |
| appResourceUsage.decUsed(decreaseRequest.getNodePartition(), absDelta); |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public List<ResourceRequest> allocate(NodeType type, |
| SchedulerNode node, SchedulerRequestKey schedulerKey, |
| Container containerAllocated) { |
| try { |
| writeLock.lock(); |
| ResourceRequest request; |
| if (type == NodeType.NODE_LOCAL) { |
| request = resourceRequestMap.get(schedulerKey).get(node.getNodeName()); |
| } else if (type == NodeType.RACK_LOCAL) { |
| request = resourceRequestMap.get(schedulerKey).get(node.getRackName()); |
| } else{ |
| request = resourceRequestMap.get(schedulerKey).get(ResourceRequest.ANY); |
| } |
| return allocate(type, node, schedulerKey, request, containerAllocated); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * Resources have been allocated to this application by the resource |
| * scheduler. Track them. |
| * @param type Node Type |
| * @param node SchedulerNode |
| * @param schedulerKey SchedulerRequestKey |
| * @param request ResourceRequest |
| * @param containerAllocated Container Allocated |
| * @return List of ResourceRequests |
| */ |
| public List<ResourceRequest> allocate(NodeType type, |
| SchedulerNode node, SchedulerRequestKey schedulerKey, |
| ResourceRequest request, Container containerAllocated) { |
| try { |
| writeLock.lock(); |
| List<ResourceRequest> resourceRequests = new ArrayList<>(); |
| if (type == NodeType.NODE_LOCAL) { |
| allocateNodeLocal(node, schedulerKey, request, resourceRequests); |
| } else if (type == NodeType.RACK_LOCAL) { |
| allocateRackLocal(node, schedulerKey, request, resourceRequests); |
| } else{ |
| allocateOffSwitch(request, resourceRequests, schedulerKey); |
| } |
| |
| if (null != containerAllocated) { |
| updateMetricsForAllocatedContainer(request, type, containerAllocated); |
| } |
| return resourceRequests; |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * The {@link ResourceScheduler} is allocating data-local resources to the |
| * application. |
| */ |
| private void allocateNodeLocal(SchedulerNode node, |
| SchedulerRequestKey schedulerKey, ResourceRequest nodeLocalRequest, |
| List<ResourceRequest> resourceRequests) { |
| // Update future requirements |
| decResourceRequest(node.getNodeName(), schedulerKey, nodeLocalRequest); |
| |
| ResourceRequest rackLocalRequest = resourceRequestMap.get(schedulerKey).get( |
| node.getRackName()); |
| decResourceRequest(node.getRackName(), schedulerKey, rackLocalRequest); |
| |
| ResourceRequest offRackRequest = resourceRequestMap.get(schedulerKey).get( |
| ResourceRequest.ANY); |
| decrementOutstanding(offRackRequest, schedulerKey); |
| |
| // Update cloned NodeLocal, RackLocal and OffRack requests for recovery |
| resourceRequests.add(cloneResourceRequest(nodeLocalRequest)); |
| resourceRequests.add(cloneResourceRequest(rackLocalRequest)); |
| resourceRequests.add(cloneResourceRequest(offRackRequest)); |
| } |
| |
| private void decResourceRequest(String resourceName, |
| SchedulerRequestKey schedulerKey, ResourceRequest request) { |
| request.setNumContainers(request.getNumContainers() - 1); |
| if (request.getNumContainers() == 0) { |
| resourceRequestMap.get(schedulerKey).remove(resourceName); |
| } |
| } |
| |
| /** |
| * The {@link ResourceScheduler} is allocating data-local resources to the |
| * application. |
| */ |
| private void allocateRackLocal(SchedulerNode node, |
| SchedulerRequestKey schedulerKey, ResourceRequest rackLocalRequest, |
| List<ResourceRequest> resourceRequests) { |
| // Update future requirements |
| decResourceRequest(node.getRackName(), schedulerKey, rackLocalRequest); |
| |
| ResourceRequest offRackRequest = resourceRequestMap.get(schedulerKey).get( |
| ResourceRequest.ANY); |
| decrementOutstanding(offRackRequest, schedulerKey); |
| |
| // Update cloned RackLocal and OffRack requests for recovery |
| resourceRequests.add(cloneResourceRequest(rackLocalRequest)); |
| resourceRequests.add(cloneResourceRequest(offRackRequest)); |
| } |
| |
| /** |
| * The {@link ResourceScheduler} is allocating data-local resources to the |
| * application. |
| */ |
| private void allocateOffSwitch(ResourceRequest offSwitchRequest, |
| List<ResourceRequest> resourceRequests, |
| SchedulerRequestKey schedulerKey) { |
| // Update future requirements |
| decrementOutstanding(offSwitchRequest, schedulerKey); |
| // Update cloned OffRack requests for recovery |
| resourceRequests.add(cloneResourceRequest(offSwitchRequest)); |
| } |
| |
| private void decrementOutstanding(ResourceRequest offSwitchRequest, |
| SchedulerRequestKey schedulerKey) { |
| int numOffSwitchContainers = offSwitchRequest.getNumContainers() - 1; |
| |
| // Do not remove ANY |
| offSwitchRequest.setNumContainers(numOffSwitchContainers); |
| |
| // Do we have any outstanding requests? |
| // If there is nothing, we need to deactivate this application |
| if (numOffSwitchContainers == 0) { |
| decrementSchedulerKeyReference(schedulerKey); |
| checkForDeactivation(); |
| } |
| |
| appResourceUsage.decPending(offSwitchRequest.getNodeLabelExpression(), |
| offSwitchRequest.getCapability()); |
| queue.decPendingResource(offSwitchRequest.getNodeLabelExpression(), |
| offSwitchRequest.getCapability()); |
| } |
| |
| private void checkForDeactivation() { |
| if (schedulerKeys.isEmpty()) { |
| activeUsersManager.deactivateApplication(user, applicationId); |
| } |
| } |
| |
| public void move(Queue newQueue) { |
| try { |
| this.writeLock.lock(); |
| QueueMetrics oldMetrics = queue.getMetrics(); |
| QueueMetrics newMetrics = newQueue.getMetrics(); |
| for (Map<String, ResourceRequest> asks : resourceRequestMap.values()) { |
| ResourceRequest request = asks.get(ResourceRequest.ANY); |
| if (request != null) { |
| oldMetrics.decrPendingResources(user, request.getNumContainers(), |
| request.getCapability()); |
| newMetrics.incrPendingResources(user, request.getNumContainers(), |
| request.getCapability()); |
| |
| Resource delta = Resources.multiply(request.getCapability(), |
| request.getNumContainers()); |
| // Update Queue |
| queue.decPendingResource(request.getNodeLabelExpression(), delta); |
| newQueue.incPendingResource(request.getNodeLabelExpression(), delta); |
| } |
| } |
| oldMetrics.moveAppFrom(this); |
| newMetrics.moveAppTo(this); |
| activeUsersManager.deactivateApplication(user, applicationId); |
| activeUsersManager = newQueue.getActiveUsersManager(); |
| activeUsersManager.activateApplication(user, applicationId); |
| this.queue = newQueue; |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public void stop() { |
| // clear pending resources metrics for the application |
| try { |
| this.writeLock.lock(); |
| QueueMetrics metrics = queue.getMetrics(); |
| for (Map<String, ResourceRequest> asks : resourceRequestMap.values()) { |
| ResourceRequest request = asks.get(ResourceRequest.ANY); |
| if (request != null) { |
| metrics.decrPendingResources(user, request.getNumContainers(), |
| request.getCapability()); |
| |
| // Update Queue |
| queue.decPendingResource( |
| request.getNodeLabelExpression(), |
| Resources.multiply(request.getCapability(), |
| request.getNumContainers())); |
| } |
| } |
| metrics.finishAppAttempt(applicationId, pending, user); |
| |
| // Clear requests themselves |
| clearRequests(); |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public void setQueue(Queue queue) { |
| try { |
| this.writeLock.lock(); |
| this.queue = queue; |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| private Set<String> getBlackList() { |
| return this.placesBlacklistedByApp; |
| } |
| |
| public Set<String> getBlackListCopy() { |
| synchronized (placesBlacklistedByApp) { |
| return new HashSet<>(this.placesBlacklistedByApp); |
| } |
| } |
| |
| public void transferStateFromPreviousAppSchedulingInfo( |
| AppSchedulingInfo appInfo) { |
| // This should not require locking the placesBlacklistedByApp since it will |
| // not be used by this instance until after setCurrentAppAttempt. |
| this.placesBlacklistedByApp = appInfo.getBlackList(); |
| } |
| |
| public void recoverContainer(RMContainer rmContainer) { |
| try { |
| this.writeLock.lock(); |
| QueueMetrics metrics = queue.getMetrics(); |
| if (pending) { |
| // If there was any container to recover, the application was |
| // running from scheduler's POV. |
| pending = false; |
| metrics.runAppAttempt(applicationId, user); |
| } |
| |
| // Container is completed. Skip recovering resources. |
| if (rmContainer.getState().equals(RMContainerState.COMPLETED)) { |
| return; |
| } |
| |
| metrics.allocateResources(user, 1, rmContainer.getAllocatedResource(), |
| false); |
| } finally { |
| this.writeLock.unlock(); |
| } |
| } |
| |
| public ResourceRequest cloneResourceRequest(ResourceRequest request) { |
| ResourceRequest newRequest = ResourceRequest.newBuilder() |
| .priority(request.getPriority()) |
| .resourceName(request.getResourceName()) |
| .capability(request.getCapability()) |
| .numContainers(1) |
| .relaxLocality(request.getRelaxLocality()) |
| .nodeLabelExpression(request.getNodeLabelExpression()).build(); |
| return newRequest; |
| } |
| |
| /* |
| * In async environment, pending resource request could be updated during |
| * scheduling, this method checks pending request before allocating |
| */ |
| public boolean checkAllocation(NodeType type, SchedulerNode node, |
| SchedulerRequestKey schedulerKey) { |
| try { |
| readLock.lock(); |
| ResourceRequest r = resourceRequestMap.get(schedulerKey).get( |
| ResourceRequest.ANY); |
| if (r == null || r.getNumContainers() <= 0) { |
| return false; |
| } |
| if (type == NodeType.RACK_LOCAL || type == NodeType.NODE_LOCAL) { |
| r = resourceRequestMap.get(schedulerKey).get(node.getRackName()); |
| if (r == null || r.getNumContainers() <= 0) { |
| return false; |
| } |
| if (type == NodeType.NODE_LOCAL) { |
| r = resourceRequestMap.get(schedulerKey).get(node.getNodeName()); |
| if (r == null || r.getNumContainers() <= 0) { |
| return false; |
| } |
| } |
| } |
| |
| return true; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| public void updateMetricsForAllocatedContainer( |
| ResourceRequest request, NodeType type, Container containerAllocated) { |
| try { |
| writeLock.lock(); |
| QueueMetrics metrics = queue.getMetrics(); |
| if (pending) { |
| // once an allocation is done we assume the application is |
| // running from scheduler's POV. |
| pending = false; |
| metrics.runAppAttempt(applicationId, user); |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("allocate: applicationId=" + applicationId + " container=" |
| + containerAllocated.getId() + " host=" + containerAllocated |
| .getNodeId().toString() + " user=" + user + " resource=" + request |
| .getCapability() + " type=" + type); |
| } |
| metrics.allocateResources(user, 1, request.getCapability(), true); |
| metrics.incrNodeTypeAggregations(user, type); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| // Get placement-set by specified schedulerKey |
| // Now simply return all node of the input clusterPlacementSet |
| // TODO, need update this when we support global scheduling |
| public <N extends SchedulerNode> SchedulingPlacementSet<N> getSchedulingPlacementSet( |
| SchedulerRequestKey schedulerkey) { |
| return new SchedulingPlacementSet<N>() { |
| @Override |
| @SuppressWarnings("unchecked") |
| public Iterator<N> getPreferredNodeIterator( |
| PlacementSet<N> clusterPlacementSet) { |
| return IteratorUtils.singletonIterator( |
| clusterPlacementSet.getAllNodes().values().iterator().next()); |
| } |
| |
| @Override |
| public ResourceRequestUpdateResult updateResourceRequests( |
| List<ResourceRequest> requests, |
| boolean recoverPreemptedRequestForAContainer) { |
| return null; |
| } |
| |
| @Override |
| public Map<String, ResourceRequest> getResourceRequests() { |
| return null; |
| } |
| |
| @Override |
| public ResourceRequest getResourceRequest(String resourceName, |
| SchedulerRequestKey requestKey) { |
| return null; |
| } |
| |
| @Override |
| public List<ResourceRequest> allocate(NodeType type, SchedulerNode node, |
| ResourceRequest request) { |
| return null; |
| } |
| |
| @Override |
| public Map<NodeId, N> getAllNodes() { |
| return null; |
| } |
| |
| @Override |
| public long getVersion() { |
| return 0; |
| } |
| |
| @Override |
| public String getPartition() { |
| return null; |
| } |
| }; |
| } |
| } |
