<commit_before>/**
 * 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.capacity;

import java.io.IOException;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.util.Sets;
import org.apache.hadoop.util.Time;
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.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.QueueState;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.security.AccessType;
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.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.*;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.AMState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.UsersManager.User;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ContainerAllocationProposal;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.SchedulerContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.CandidateNodeSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.CandidateNodeSetUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FifoOrderingPolicyForPendingApps;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.IteratorSelector;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPolicy;
import org.apache.hadoop.yarn.server.utils.Lock;
import org.apache.hadoop.yarn.server.utils.Lock.NoLock;
import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.hadoop.yarn.util.resource.Resources;

import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Private
@Unstable
public class LeafQueue extends AbstractCSQueue {
  private static final Logger LOG =
      LoggerFactory.getLogger(LeafQueue.class);

  private float absoluteUsedCapacity = 0.0f;

  // TODO the max applications should consider label
  protected int maxApplications;
  protected volatile int maxApplicationsPerUser;
  
  private float maxAMResourcePerQueuePercent;

  private volatile int nodeLocalityDelay;
  private volatile int rackLocalityAdditionalDelay;
  private volatile boolean rackLocalityFullReset;

  Map<ApplicationAttemptId, FiCaSchedulerApp> applicationAttemptMap =
      new ConcurrentHashMap<>();

  private Priority defaultAppPriorityPerQueue;

  private final OrderingPolicy<FiCaSchedulerApp> pendingOrderingPolicy;

  private volatile float minimumAllocationFactor;

  private final RecordFactory recordFactory = 
    RecordFactoryProvider.getRecordFactory(null);

  private final UsersManager usersManager;

  // cache last cluster resource to compute actual capacity
  private Resource lastClusterResource = Resources.none();

  private final QueueResourceLimitsInfo queueResourceLimitsInfo =
      new QueueResourceLimitsInfo();

  private volatile ResourceLimits cachedResourceLimitsForHeadroom = null;

  private volatile OrderingPolicy<FiCaSchedulerApp> orderingPolicy = null;

  // Map<Partition, Map<SchedulingMode, Map<User, CachedUserLimit>>>
  // Not thread safe: only the last level is a ConcurrentMap
  @VisibleForTesting
  Map<String, Map<SchedulingMode, ConcurrentMap<String, CachedUserLimit>>>
      userLimitsCache = new HashMap<>();

  // Not thread safe
  @VisibleForTesting
  long currentUserLimitCacheVersion = 0;

  // record all ignore partition exclusivityRMContainer, this will be used to do
  // preemption, key is the partition of the RMContainer allocated on
  private Map<String, TreeSet<RMContainer>> ignorePartitionExclusivityRMContainers =
      new ConcurrentHashMap<>();

  List<AppPriorityACLGroup> priorityAcls =
      new ArrayList<AppPriorityACLGroup>();

  private final List<FiCaSchedulerApp> runnableApps = new ArrayList<>();
  private final List<FiCaSchedulerApp> nonRunnableApps = new ArrayList<>();

  @SuppressWarnings({ "unchecked", "rawtypes" })
  public LeafQueue(CapacitySchedulerContext cs,
      String queueName, CSQueue parent, CSQueue old) throws IOException {
    this(cs, cs.getConfiguration(), queueName, parent, old, false);
  }

  public LeafQueue(CapacitySchedulerContext cs,
      CapacitySchedulerConfiguration configuration,
      String queueName, CSQueue parent, CSQueue old) throws IOException {
    this(cs, configuration, queueName, parent, old, false);
  }

  public LeafQueue(CapacitySchedulerContext cs,
      CapacitySchedulerConfiguration configuration,
      String queueName, CSQueue parent, CSQueue old, boolean isDynamic) throws
      IOException {
    super(cs, configuration, queueName, parent, old);
    setDynamicQueue(isDynamic);

    this.usersManager = new UsersManager(metrics, this, labelManager, csContext,
        resourceCalculator);

    // One time initialization is enough since it is static ordering policy
    this.pendingOrderingPolicy = new FifoOrderingPolicyForPendingApps();

    LOG.debug("LeafQueue: name={}, fullname={}", queueName, getQueuePath());

    setupQueueConfigs(cs.getClusterResource(), configuration);

  }

  @SuppressWarnings("checkstyle:nowhitespaceafter")
  protected void setupQueueConfigs(Resource clusterResource,
      CapacitySchedulerConfiguration conf) throws
      IOException {
    writeLock.lock();
    try {
      CapacitySchedulerConfiguration schedConf = csContext.getConfiguration();
      super.setupQueueConfigs(clusterResource, conf);

      this.lastClusterResource = clusterResource;

      this.cachedResourceLimitsForHeadroom = new ResourceLimits(
          clusterResource);

      // Initialize headroom info, also used for calculating application
      // master resource limits.  Since this happens during queue initialization
      // and all queues may not be realized yet, we'll use (optimistic)
      // absoluteMaxCapacity (it will be replaced with the more accurate
      // absoluteMaxAvailCapacity during headroom/userlimit/allocation events)
      setQueueResourceLimitsInfo(clusterResource);

      setOrderingPolicy(
          conf.<FiCaSchedulerApp>getAppOrderingPolicy(getQueuePath()));

      usersManager.setUserLimit(conf.getUserLimit(getQueuePath()));
      usersManager.setUserLimitFactor(conf.getUserLimitFactor(getQueuePath()));

      maxAMResourcePerQueuePercent =
          conf.getMaximumApplicationMasterResourcePerQueuePercent(
              getQueuePath());

      maxApplications = conf.getMaximumApplicationsPerQueue(getQueuePath());
      if (maxApplications < 0) {
        int maxGlobalPerQueueApps =
            csContext.getConfiguration().getGlobalMaximumApplicationsPerQueue();
        if (maxGlobalPerQueueApps > 0) {
          maxApplications = maxGlobalPerQueueApps;
        }
      }

      priorityAcls = conf.getPriorityAcls(getQueuePath(),
          csContext.getMaxClusterLevelAppPriority());

      if (!SchedulerUtils.checkQueueLabelExpression(this.accessibleLabels,
          this.defaultLabelExpression, null)) {
        throw new IOException(
            "Invalid default label expression of " + " queue=" + getQueuePath()
                + " doesn't have permission to access all labels "
                + "in default label expression. labelExpression of resource request="
                + (this.defaultLabelExpression == null ?
                "" :
                this.defaultLabelExpression) + ". Queue labels=" + (
                getAccessibleNodeLabels() == null ?
                    "" :
                    StringUtils
                        .join(getAccessibleNodeLabels().iterator(), ',')));
      }

      nodeLocalityDelay = schedConf.getNodeLocalityDelay();
      rackLocalityAdditionalDelay = schedConf
          .getRackLocalityAdditionalDelay();
      rackLocalityFullReset = schedConf
          .getRackLocalityFullReset();

      // re-init this since max allocation could have changed
      this.minimumAllocationFactor = Resources.ratio(resourceCalculator,
          Resources.subtract(maximumAllocation, minimumAllocation),
          maximumAllocation);

      StringBuilder aclsString = new StringBuilder();
      for (Map.Entry<AccessType, AccessControlList> e : acls.entrySet()) {
        aclsString.append(e.getKey() + ":" + e.getValue().getAclString());
      }

      StringBuilder labelStrBuilder = new StringBuilder();
      if (accessibleLabels != null) {
        for (String s : accessibleLabels) {
          labelStrBuilder.append(s)
              .append(",");
        }
      }

      defaultAppPriorityPerQueue = Priority.newInstance(
          conf.getDefaultApplicationPriorityConfPerQueue(getQueuePath()));

      // Validate leaf queue's user's weights.
      float queueUserLimit = Math.min(100.0f, conf.getUserLimit(getQueuePath()));
      getUserWeights().validateForLeafQueue(queueUserLimit, getQueuePath());
      usersManager.updateUserWeights();

      LOG.info(
          "Initializing " + getQueuePath() + "\n" +
              getExtendedCapacityOrWeightString() + "\n"
              + "absoluteCapacity = " + queueCapacities.getAbsoluteCapacity()
              + " [= parentAbsoluteCapacity * capacity ]" + "\n"
              + "maxCapacity = " + queueCapacities.getMaximumCapacity()
              + " [= configuredMaxCapacity ]" + "\n" + "absoluteMaxCapacity = "
              + queueCapacities.getAbsoluteMaximumCapacity()
              + " [= 1.0 maximumCapacity undefined, "
              + "(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]"
              + "\n" + "effectiveMinResource=" +
              getEffectiveCapacity(CommonNodeLabelsManager.NO_LABEL) + "\n"
              + " , effectiveMaxResource=" +
              getEffectiveMaxCapacity(CommonNodeLabelsManager.NO_LABEL)
              + "\n" + "userLimit = " + usersManager.getUserLimit()
              + " [= configuredUserLimit ]" + "\n" + "userLimitFactor = "
              + usersManager.getUserLimitFactor()
              + " [= configuredUserLimitFactor ]" + "\n" + "maxApplications = "
              + maxApplications
              + " [= configuredMaximumSystemApplicationsPerQueue or"
              + " (int)(configuredMaximumSystemApplications * absoluteCapacity)]"
              + "\n" + "maxApplicationsPerUser = " + maxApplicationsPerUser
              + " [= (int)(maxApplications * (userLimit / 100.0f) * "
              + "userLimitFactor) ]" + "\n"
              + "maxParallelApps = " + getMaxParallelApps() + "\n"
              + "usedCapacity = " +
              + queueCapacities.getUsedCapacity() + " [= usedResourcesMemory / "
              + "(clusterResourceMemory * absoluteCapacity)]" + "\n"
              + "absoluteUsedCapacity = " + absoluteUsedCapacity
              + " [= usedResourcesMemory / clusterResourceMemory]" + "\n"
              + "maxAMResourcePerQueuePercent = " + maxAMResourcePerQueuePercent
              + " [= configuredMaximumAMResourcePercent ]" + "\n"
              + "minimumAllocationFactor = " + minimumAllocationFactor
              + " [= (float)(maximumAllocationMemory - minimumAllocationMemory) / "
              + "maximumAllocationMemory ]" + "\n" + "maximumAllocation = "
              + maximumAllocation + " [= configuredMaxAllocation ]" + "\n"
              + "numContainers = " + numContainers
              + " [= currentNumContainers ]" + "\n" + "state = " + getState()
              + " [= configuredState ]" + "\n" + "acls = " + aclsString
              + " [= configuredAcls ]" + "\n"
              + "nodeLocalityDelay = " + nodeLocalityDelay + "\n"
              + "rackLocalityAdditionalDelay = "
              + rackLocalityAdditionalDelay + "\n"
              + "labels=" + labelStrBuilder.toString() + "\n"
              + "reservationsContinueLooking = "
              + reservationsContinueLooking + "\n" + "preemptionDisabled = "
              + getPreemptionDisabled() + "\n" + "defaultAppPriorityPerQueue = "
              + defaultAppPriorityPerQueue + "\npriority = " + priority
              + "\nmaxLifetime = " + getMaximumApplicationLifetime()
              + " seconds" + "\ndefaultLifetime = "
              + getDefaultApplicationLifetime() + " seconds");
    } finally {
      writeLock.unlock();
    }
  }

  /**
   * Used only by tests.
   */
  @Private
  public float getMinimumAllocationFactor() {
    return minimumAllocationFactor;
  }
  
  /**
   * Used only by tests.
   */
  @Private
  public float getMaxAMResourcePerQueuePercent() {
    return maxAMResourcePerQueuePercent;
  }

  public int getMaxApplications() {
    return maxApplications;
  }

  public int getMaxApplicationsPerUser() {
    return maxApplicationsPerUser;
  }

  /**
   *
   * @return UsersManager instance.
   */
  public UsersManager getUsersManager() {
    return usersManager;
  }

  @Override
  public AbstractUsersManager getAbstractUsersManager() {
    return usersManager;
  }

  @Override
  public List<CSQueue> getChildQueues() {
    return null;
  }
  
  /**
   * Set user limit.
   * @param userLimit new user limit
   */
  @VisibleForTesting
  void setUserLimit(float userLimit) {
    usersManager.setUserLimit(userLimit);
    usersManager.userLimitNeedsRecompute();
  }

  /**
   * Set user limit factor.
   * @param userLimitFactor new user limit factor
   */
  @VisibleForTesting
  void setUserLimitFactor(float userLimitFactor) {
    usersManager.setUserLimitFactor(userLimitFactor);
    usersManager.userLimitNeedsRecompute();
  }

  @Override
  public int getNumApplications() {
    readLock.lock();
    try {
      return getNumPendingApplications() + getNumActiveApplications() +
          getNumNonRunnableApps();
    } finally {
      readLock.unlock();
    }
  }

  public int getNumPendingApplications() {
    readLock.lock();
    try {
      return pendingOrderingPolicy.getNumSchedulableEntities();
    } finally {
      readLock.unlock();
    }
  }

  public int getNumActiveApplications() {
    readLock.lock();
    try {
      return orderingPolicy.getNumSchedulableEntities();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public int getNumPendingApplications(String user) {
    readLock.lock();
    try {
      User u = getUser(user);
      if (null == u) {
        return 0;
      }
      return u.getPendingApplications();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public int getNumActiveApplications(String user) {
    readLock.lock();
    try {
      User u = getUser(user);
      if (null == u) {
        return 0;
      }
      return u.getActiveApplications();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public float getUserLimit() {
    return usersManager.getUserLimit();
  }

  @Private
  public float getUserLimitFactor() {
    return usersManager.getUserLimitFactor();
  }

  @Override
  public QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive) {
    QueueInfo queueInfo = getQueueInfo();
    return queueInfo;
  }

  @Override
  public List<QueueUserACLInfo>
  getQueueUserAclInfo(UserGroupInformation user) {
    readLock.lock();
    try {
      QueueUserACLInfo userAclInfo = recordFactory.newRecordInstance(
          QueueUserACLInfo.class);
      List<QueueACL> operations = new ArrayList<>();
      for (QueueACL operation : QueueACL.values()) {
        if (hasAccess(operation, user)) {
          operations.add(operation);
        }
      }

      userAclInfo.setQueueName(getQueuePath());
      userAclInfo.setUserAcls(operations);
      return Collections.singletonList(userAclInfo);
    } finally {
      readLock.unlock();
    }

  }

  public String toString() {
    readLock.lock();
    try {
      return getQueuePath() + ": " + getCapacityOrWeightString()
          + ", " + "absoluteCapacity=" + queueCapacities.getAbsoluteCapacity()
          + ", " + "usedResources=" + queueUsage.getUsed() + ", "
          + "usedCapacity=" + getUsedCapacity() + ", " + "absoluteUsedCapacity="
          + getAbsoluteUsedCapacity() + ", " + "numApps=" + getNumApplications()
          + ", " + "numContainers=" + getNumContainers() + ", "
          + "effectiveMinResource=" +
          getEffectiveCapacity(CommonNodeLabelsManager.NO_LABEL) +
          " , effectiveMaxResource=" +
          getEffectiveMaxCapacity(CommonNodeLabelsManager.NO_LABEL);
    } finally {
      readLock.unlock();
    }
  }

  protected String getExtendedCapacityOrWeightString() {
    if (queueCapacities.getWeight() != -1) {
      return "weight = " + queueCapacities.getWeight()
          + " [= (float) configuredCapacity (with w suffix)] " + "\n"
          + "normalizedWeight = " + queueCapacities.getNormalizedWeight()
          + " [= (float) configuredCapacity / sum(configuredCapacity of " +
          "all queues under the parent)]";
    } else {
      return "capacity = " + queueCapacities.getCapacity()
          + " [= (float) configuredCapacity / 100 ]";
    }
  }

  @VisibleForTesting
  public User getUser(String userName) {
    return usersManager.getUser(userName);
  }

  @Private
  public List<AppPriorityACLGroup> getPriorityACLs() {
    readLock.lock();
    try {
      return new ArrayList<>(priorityAcls);
    } finally {
      readLock.unlock();
    }
  }

  protected void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource,
      CapacitySchedulerConfiguration configuration) throws
      IOException {

    writeLock.lock();
    try {
      // We skip reinitialize for dynamic queues, when this is called, and
      // new queue is different from this queue, we will make this queue to be
      // static queue.
      if (newlyParsedQueue != this) {
        this.setDynamicQueue(false);
      }

      // Sanity check
      if (!(newlyParsedQueue instanceof LeafQueue) || !newlyParsedQueue
          .getQueuePath().equals(getQueuePath())) {
        throw new IOException(
            "Trying to reinitialize " + getQueuePath() + " from "
                + newlyParsedQueue.getQueuePath());
      }

      LeafQueue newlyParsedLeafQueue = (LeafQueue) newlyParsedQueue;

      // don't allow the maximum allocation to be decreased in size
      // since we have already told running AM's the size
      Resource oldMax = getMaximumAllocation();
      Resource newMax = newlyParsedLeafQueue.getMaximumAllocation();

      if (!Resources.fitsIn(oldMax, newMax)) {
        throw new IOException("Trying to reinitialize " + getQueuePath()
            + " the maximum allocation size can not be decreased!"
            + " Current setting: " + oldMax + ", trying to set it to: "
            + newMax);
      }

      setupQueueConfigs(clusterResource, configuration);
    } finally {
      writeLock.unlock();
    }
  }

  @Override
  public void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource)
  throws IOException {
    reinitialize(newlyParsedQueue, clusterResource,
        csContext.getConfiguration());
  }

  @Override
  public void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName) {
    submitApplicationAttempt(application, userName, false);
  }

  @Override
  public void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName, boolean isMoveApp) {
    // Careful! Locking order is important!
    writeLock.lock();
    try {
      // TODO, should use getUser, use this method just to avoid UT failure
      // which is caused by wrong invoking order, will fix UT separately
      User user = usersManager.getUserAndAddIfAbsent(userName);

      // Add the attempt to our data-structures
      addApplicationAttempt(application, user);
    } finally {
      writeLock.unlock();
    }

    // We don't want to update metrics for move app
    if (!isMoveApp) {
      boolean unmanagedAM = application.getAppSchedulingInfo() != null &&
          application.getAppSchedulingInfo().isUnmanagedAM();
      metrics.submitAppAttempt(userName, unmanagedAM);
    }

    getParent().submitApplicationAttempt(application, userName);
  }

  @Override
  public void submitApplication(ApplicationId applicationId, String userName,
      String queue)  throws AccessControlException {
    // Careful! Locking order is important!
    validateSubmitApplication(applicationId, userName, queue);

    // Signal for expired auto deletion.
    updateLastSubmittedTimeStamp();

    // Inform the parent queue
    try {
      getParent().submitApplication(applicationId, userName, queue);
    } catch (AccessControlException ace) {
      LOG.info("Failed to submit application to parent-queue: " +
          getParent().getQueuePath(), ace);
      throw ace;
    }

  }

  public void validateSubmitApplication(ApplicationId applicationId,
      String userName, String queue) throws AccessControlException {
    writeLock.lock();
    try {
      // Check if the queue is accepting jobs
      if (getState() != QueueState.RUNNING) {
        String msg = "Queue " + getQueuePath()
            + " is STOPPED. Cannot accept submission of application: "
            + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }

      // Check submission limits for queues
      //TODO recalculate max applications because they can depend on capacity
      if (getNumApplications() >= getMaxApplications() && !(this instanceof AutoCreatedLeafQueue)) {
        String msg =
            "Queue " + getQueuePath() + " already has " + getNumApplications()
                + " applications,"
                + " cannot accept submission of application: " + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }

      // Check submission limits for the user on this queue
      User user = usersManager.getUserAndAddIfAbsent(userName);
      //TODO recalculate max applications because they can depend on capacity
      if (user.getTotalApplications() >= getMaxApplicationsPerUser() &&  !(this instanceof AutoCreatedLeafQueue)) {
        String msg = "Queue " + getQueuePath() + " already has " + user
            .getTotalApplications() + " applications from user " + userName
            + " cannot accept submission of application: " + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }
    } finally {
      writeLock.unlock();
    }

    try {
      getParent().validateSubmitApplication(applicationId, userName, queue);
    } catch (AccessControlException ace) {
      LOG.info("Failed to submit application to parent-queue: " + 
          getParent().getQueuePath(), ace);
      throw ace;
    }
  }
  
  public Resource getAMResourceLimit() {
    return queueUsage.getAMLimit();
  }

  public Resource getAMResourceLimitPerPartition(String nodePartition) {
    return queueUsage.getAMLimit(nodePartition);
  }

  @VisibleForTesting
  public Resource calculateAndGetAMResourceLimit() {
    return calculateAndGetAMResourceLimitPerPartition(
        RMNodeLabelsManager.NO_LABEL);
  }

  @VisibleForTesting
  public Resource getUserAMResourceLimit() {
    return getUserAMResourceLimitPerPartition(RMNodeLabelsManager.NO_LABEL,
         null);
  }

  public Resource getUserAMResourceLimitPerPartition(
      String nodePartition, String userName) {
    float userWeight = 1.0f;
    if (userName != null && getUser(userName) != null) {
      userWeight = getUser(userName).getWeight();
    }

    readLock.lock();
    try {
      /*
       * The user am resource limit is based on the same approach as the user
       * limit (as it should represent a subset of that). This means that it uses
       * the absolute queue capacity (per partition) instead of the max and is
       * modified by the userlimit and the userlimit factor as is the userlimit
       */
      float effectiveUserLimit = Math.max(usersManager.getUserLimit() / 100.0f,
          1.0f / Math.max(getAbstractUsersManager().getNumActiveUsers(), 1));
      float preWeightedUserLimit = effectiveUserLimit;
      effectiveUserLimit = Math.min(effectiveUserLimit * userWeight, 1.0f);

      Resource queuePartitionResource = getEffectiveCapacity(nodePartition);

      Resource userAMLimit = Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionResource,
          queueCapacities.getMaxAMResourcePercentage(nodePartition)
              * effectiveUserLimit * usersManager.getUserLimitFactor(),
          minimumAllocation);

      if (getUserLimitFactor() == -1) {
        userAMLimit = Resources.multiplyAndNormalizeUp(
            resourceCalculator, queuePartitionResource,
            queueCapacities.getMaxAMResourcePercentage(nodePartition),
            minimumAllocation);
      }

      userAMLimit =
          Resources.min(resourceCalculator, lastClusterResource,
              userAMLimit,
              Resources.clone(getAMResourceLimitPerPartition(nodePartition)));

      Resource preWeighteduserAMLimit =
          Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionResource,
          queueCapacities.getMaxAMResourcePercentage(nodePartition)
              * preWeightedUserLimit * usersManager.getUserLimitFactor(),
          minimumAllocation);

      if (getUserLimitFactor() == -1) {
        preWeighteduserAMLimit = Resources.multiplyAndNormalizeUp(
            resourceCalculator, queuePartitionResource,
            queueCapacities.getMaxAMResourcePercentage(nodePartition),
            minimumAllocation);
      }

      preWeighteduserAMLimit =
          Resources.min(resourceCalculator, lastClusterResource,
              preWeighteduserAMLimit,
              Resources.clone(getAMResourceLimitPerPartition(nodePartition)));
      queueUsage.setUserAMLimit(nodePartition, preWeighteduserAMLimit);

      LOG.debug("Effective user AM limit for \"{}\":{}. Effective weighted"
          + " user AM limit: {}. User weight: {}", userName,
          preWeighteduserAMLimit, userAMLimit, userWeight);
      return userAMLimit;
    } finally {
      readLock.unlock();
    }

  }

  public Resource calculateAndGetAMResourceLimitPerPartition(
      String nodePartition) {
    writeLock.lock();
    try {
      /*
       * For non-labeled partition, get the max value from resources currently
       * available to the queue and the absolute resources guaranteed for the
       * partition in the queue. For labeled partition, consider only the absolute
       * resources guaranteed. Multiply this value (based on labeled/
       * non-labeled), * with per-partition am-resource-percent to get the max am
       * resource limit for this queue and partition.
       */
      Resource queuePartitionResource = getEffectiveCapacity(nodePartition);

      Resource queueCurrentLimit = Resources.none();
      // For non-labeled partition, we need to consider the current queue
      // usage limit.
      if (nodePartition.equals(RMNodeLabelsManager.NO_LABEL)) {
        synchronized (queueResourceLimitsInfo){
          queueCurrentLimit = queueResourceLimitsInfo.getQueueCurrentLimit();
        }
      }

      float amResourcePercent = queueCapacities.getMaxAMResourcePercentage(
          nodePartition);

      // Current usable resource for this queue and partition is the max of
      // queueCurrentLimit and queuePartitionResource.
      // If any of the resources available to this queue are less than queue's
      // guarantee, use the guarantee as the queuePartitionUsableResource
      // because nothing less than the queue's guarantee should be used when
      // calculating the AM limit.
      Resource queuePartitionUsableResource = (Resources.fitsIn(
          resourceCalculator, queuePartitionResource, queueCurrentLimit)) ?
              queueCurrentLimit : queuePartitionResource;

      Resource amResouceLimit = Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionUsableResource, amResourcePercent,
          minimumAllocation);

      metrics.setAMResouceLimit(nodePartition, amResouceLimit);
      queueUsage.setAMLimit(nodePartition, amResouceLimit);
      LOG.debug("Queue: {}, node label : {}, queue partition resource : {},"
          + " queue current limit : {}, queue partition usable resource : {},"
          + " amResourceLimit : {}", getQueuePath(), nodePartition,
          queuePartitionResource, queueCurrentLimit,
          queuePartitionUsableResource, amResouceLimit);
      return amResouceLimit;
    } finally {
      writeLock.unlock();
    }
  }

  protected void activateApplications() {
    writeLock.lock();
    try {
      // limit of allowed resource usage for application masters
      Map<String, Resource> userAmPartitionLimit =
          new HashMap<String, Resource>();

      // AM Resource Limit for accessible labels can be pre-calculated.
      // This will help in updating AMResourceLimit for all labels when queue
      // is initialized for the first time (when no applications are present).
      for (String nodePartition : getNodeLabelsForQueue()) {
        calculateAndGetAMResourceLimitPerPartition(nodePartition);
      }

      for (Iterator<FiCaSchedulerApp> fsApp =
           getPendingAppsOrderingPolicy()
               .getAssignmentIterator(IteratorSelector.EMPTY_ITERATOR_SELECTOR);
           fsApp.hasNext(); ) {
        FiCaSchedulerApp application = fsApp.next();
        ApplicationId applicationId = application.getApplicationId();

        // Get the am-node-partition associated with each application
        // and calculate max-am resource limit for this partition.
        String partitionName = application.getAppAMNodePartitionName();

        Resource amLimit = getAMResourceLimitPerPartition(partitionName);
        // Verify whether we already calculated am-limit for this label.
        if (amLimit == null) {
          amLimit = calculateAndGetAMResourceLimitPerPartition(partitionName);
        }
        // Check am resource limit.
        Resource amIfStarted = Resources.add(
            application.getAMResource(partitionName),
            queueUsage.getAMUsed(partitionName));

        if (LOG.isDebugEnabled()) {
          LOG.debug("application " + application.getId() + " AMResource "
              + application.getAMResource(partitionName)
              + " maxAMResourcePerQueuePercent " + maxAMResourcePerQueuePercent
              + " amLimit " + amLimit + " lastClusterResource "
              + lastClusterResource + " amIfStarted " + amIfStarted
              + " AM node-partition name " + partitionName);
        }

        if (!resourceCalculator.fitsIn(amIfStarted, amLimit)) {
          if (getNumActiveApplications() < 1 || (Resources.lessThanOrEqual(
              resourceCalculator, lastClusterResource,
              queueUsage.getAMUsed(partitionName), Resources.none()))) {
            LOG.warn("maximum-am-resource-percent is insufficient to start a"
                + " single application in queue, it is likely set too low."
                + " skipping enforcement to allow at least one application"
                + " to start");
          } else{
            application.updateAMContainerDiagnostics(AMState.INACTIVATED,
                CSAMContainerLaunchDiagnosticsConstants.QUEUE_AM_RESOURCE_LIMIT_EXCEED);
            LOG.debug("Not activating application {} as  amIfStarted: {}"
                + " exceeds amLimit: {}", applicationId, amIfStarted, amLimit);
            continue;
          }
        }

        // Check user am resource limit
        User user = getUser(application.getUser());
        Resource userAMLimit = userAmPartitionLimit.get(partitionName);

        // Verify whether we already calculated user-am-limit for this label.
        if (userAMLimit == null) {
          userAMLimit = getUserAMResourceLimitPerPartition(partitionName,
              application.getUser());
          userAmPartitionLimit.put(partitionName, userAMLimit);
        }

        Resource userAmIfStarted = Resources.add(
            application.getAMResource(partitionName),
            user.getConsumedAMResources(partitionName));
<commit_msg>YARN-10934. Fix LeafQueue#activateApplication NPE when the user of the pending application is missing from usersManager. Contributed by <NAME> <commit_after>/**
 * 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.capacity;

import java.io.IOException;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.authorize.AccessControlList;
import org.apache.hadoop.util.Sets;
import org.apache.hadoop.util.Time;
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.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueInfo;
import org.apache.hadoop.yarn.api.records.QueueState;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.security.AccessType;
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.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.*;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.AMState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.UsersManager.User;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ContainerAllocationProposal;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.SchedulerContainer;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.CandidateNodeSet;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.CandidateNodeSetUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.FifoOrderingPolicyForPendingApps;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.IteratorSelector;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPolicy;
import org.apache.hadoop.yarn.server.utils.Lock;
import org.apache.hadoop.yarn.server.utils.Lock.NoLock;
import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.hadoop.yarn.util.resource.Resources;

import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Private
@Unstable
public class LeafQueue extends AbstractCSQueue {
  private static final Logger LOG =
      LoggerFactory.getLogger(LeafQueue.class);

  private float absoluteUsedCapacity = 0.0f;

  // TODO the max applications should consider label
  protected int maxApplications;
  protected volatile int maxApplicationsPerUser;
  
  private float maxAMResourcePerQueuePercent;

  private volatile int nodeLocalityDelay;
  private volatile int rackLocalityAdditionalDelay;
  private volatile boolean rackLocalityFullReset;

  Map<ApplicationAttemptId, FiCaSchedulerApp> applicationAttemptMap =
      new ConcurrentHashMap<>();

  private Priority defaultAppPriorityPerQueue;

  private final OrderingPolicy<FiCaSchedulerApp> pendingOrderingPolicy;

  private volatile float minimumAllocationFactor;

  private final RecordFactory recordFactory = 
    RecordFactoryProvider.getRecordFactory(null);

  private final UsersManager usersManager;

  // cache last cluster resource to compute actual capacity
  private Resource lastClusterResource = Resources.none();

  private final QueueResourceLimitsInfo queueResourceLimitsInfo =
      new QueueResourceLimitsInfo();

  private volatile ResourceLimits cachedResourceLimitsForHeadroom = null;

  private volatile OrderingPolicy<FiCaSchedulerApp> orderingPolicy = null;

  // Map<Partition, Map<SchedulingMode, Map<User, CachedUserLimit>>>
  // Not thread safe: only the last level is a ConcurrentMap
  @VisibleForTesting
  Map<String, Map<SchedulingMode, ConcurrentMap<String, CachedUserLimit>>>
      userLimitsCache = new HashMap<>();

  // Not thread safe
  @VisibleForTesting
  long currentUserLimitCacheVersion = 0;

  // record all ignore partition exclusivityRMContainer, this will be used to do
  // preemption, key is the partition of the RMContainer allocated on
  private Map<String, TreeSet<RMContainer>> ignorePartitionExclusivityRMContainers =
      new ConcurrentHashMap<>();

  List<AppPriorityACLGroup> priorityAcls =
      new ArrayList<AppPriorityACLGroup>();

  private final List<FiCaSchedulerApp> runnableApps = new ArrayList<>();
  private final List<FiCaSchedulerApp> nonRunnableApps = new ArrayList<>();

  @SuppressWarnings({ "unchecked", "rawtypes" })
  public LeafQueue(CapacitySchedulerContext cs,
      String queueName, CSQueue parent, CSQueue old) throws IOException {
    this(cs, cs.getConfiguration(), queueName, parent, old, false);
  }

  public LeafQueue(CapacitySchedulerContext cs,
      CapacitySchedulerConfiguration configuration,
      String queueName, CSQueue parent, CSQueue old) throws IOException {
    this(cs, configuration, queueName, parent, old, false);
  }

  public LeafQueue(CapacitySchedulerContext cs,
      CapacitySchedulerConfiguration configuration,
      String queueName, CSQueue parent, CSQueue old, boolean isDynamic) throws
      IOException {
    super(cs, configuration, queueName, parent, old);
    setDynamicQueue(isDynamic);

    this.usersManager = new UsersManager(metrics, this, labelManager, csContext,
        resourceCalculator);

    // One time initialization is enough since it is static ordering policy
    this.pendingOrderingPolicy = new FifoOrderingPolicyForPendingApps();

    LOG.debug("LeafQueue: name={}, fullname={}", queueName, getQueuePath());

    setupQueueConfigs(cs.getClusterResource(), configuration);

  }

  @SuppressWarnings("checkstyle:nowhitespaceafter")
  protected void setupQueueConfigs(Resource clusterResource,
      CapacitySchedulerConfiguration conf) throws
      IOException {
    writeLock.lock();
    try {
      CapacitySchedulerConfiguration schedConf = csContext.getConfiguration();
      super.setupQueueConfigs(clusterResource, conf);

      this.lastClusterResource = clusterResource;

      this.cachedResourceLimitsForHeadroom = new ResourceLimits(
          clusterResource);

      // Initialize headroom info, also used for calculating application
      // master resource limits.  Since this happens during queue initialization
      // and all queues may not be realized yet, we'll use (optimistic)
      // absoluteMaxCapacity (it will be replaced with the more accurate
      // absoluteMaxAvailCapacity during headroom/userlimit/allocation events)
      setQueueResourceLimitsInfo(clusterResource);

      setOrderingPolicy(
          conf.<FiCaSchedulerApp>getAppOrderingPolicy(getQueuePath()));

      usersManager.setUserLimit(conf.getUserLimit(getQueuePath()));
      usersManager.setUserLimitFactor(conf.getUserLimitFactor(getQueuePath()));

      maxAMResourcePerQueuePercent =
          conf.getMaximumApplicationMasterResourcePerQueuePercent(
              getQueuePath());

      maxApplications = conf.getMaximumApplicationsPerQueue(getQueuePath());
      if (maxApplications < 0) {
        int maxGlobalPerQueueApps =
            csContext.getConfiguration().getGlobalMaximumApplicationsPerQueue();
        if (maxGlobalPerQueueApps > 0) {
          maxApplications = maxGlobalPerQueueApps;
        }
      }

      priorityAcls = conf.getPriorityAcls(getQueuePath(),
          csContext.getMaxClusterLevelAppPriority());

      if (!SchedulerUtils.checkQueueLabelExpression(this.accessibleLabels,
          this.defaultLabelExpression, null)) {
        throw new IOException(
            "Invalid default label expression of " + " queue=" + getQueuePath()
                + " doesn't have permission to access all labels "
                + "in default label expression. labelExpression of resource request="
                + (this.defaultLabelExpression == null ?
                "" :
                this.defaultLabelExpression) + ". Queue labels=" + (
                getAccessibleNodeLabels() == null ?
                    "" :
                    StringUtils
                        .join(getAccessibleNodeLabels().iterator(), ',')));
      }

      nodeLocalityDelay = schedConf.getNodeLocalityDelay();
      rackLocalityAdditionalDelay = schedConf
          .getRackLocalityAdditionalDelay();
      rackLocalityFullReset = schedConf
          .getRackLocalityFullReset();

      // re-init this since max allocation could have changed
      this.minimumAllocationFactor = Resources.ratio(resourceCalculator,
          Resources.subtract(maximumAllocation, minimumAllocation),
          maximumAllocation);

      StringBuilder aclsString = new StringBuilder();
      for (Map.Entry<AccessType, AccessControlList> e : acls.entrySet()) {
        aclsString.append(e.getKey() + ":" + e.getValue().getAclString());
      }

      StringBuilder labelStrBuilder = new StringBuilder();
      if (accessibleLabels != null) {
        for (String s : accessibleLabels) {
          labelStrBuilder.append(s)
              .append(",");
        }
      }

      defaultAppPriorityPerQueue = Priority.newInstance(
          conf.getDefaultApplicationPriorityConfPerQueue(getQueuePath()));

      // Validate leaf queue's user's weights.
      float queueUserLimit = Math.min(100.0f, conf.getUserLimit(getQueuePath()));
      getUserWeights().validateForLeafQueue(queueUserLimit, getQueuePath());
      usersManager.updateUserWeights();

      LOG.info(
          "Initializing " + getQueuePath() + "\n" +
              getExtendedCapacityOrWeightString() + "\n"
              + "absoluteCapacity = " + queueCapacities.getAbsoluteCapacity()
              + " [= parentAbsoluteCapacity * capacity ]" + "\n"
              + "maxCapacity = " + queueCapacities.getMaximumCapacity()
              + " [= configuredMaxCapacity ]" + "\n" + "absoluteMaxCapacity = "
              + queueCapacities.getAbsoluteMaximumCapacity()
              + " [= 1.0 maximumCapacity undefined, "
              + "(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]"
              + "\n" + "effectiveMinResource=" +
              getEffectiveCapacity(CommonNodeLabelsManager.NO_LABEL) + "\n"
              + " , effectiveMaxResource=" +
              getEffectiveMaxCapacity(CommonNodeLabelsManager.NO_LABEL)
              + "\n" + "userLimit = " + usersManager.getUserLimit()
              + " [= configuredUserLimit ]" + "\n" + "userLimitFactor = "
              + usersManager.getUserLimitFactor()
              + " [= configuredUserLimitFactor ]" + "\n" + "maxApplications = "
              + maxApplications
              + " [= configuredMaximumSystemApplicationsPerQueue or"
              + " (int)(configuredMaximumSystemApplications * absoluteCapacity)]"
              + "\n" + "maxApplicationsPerUser = " + maxApplicationsPerUser
              + " [= (int)(maxApplications * (userLimit / 100.0f) * "
              + "userLimitFactor) ]" + "\n"
              + "maxParallelApps = " + getMaxParallelApps() + "\n"
              + "usedCapacity = " +
              + queueCapacities.getUsedCapacity() + " [= usedResourcesMemory / "
              + "(clusterResourceMemory * absoluteCapacity)]" + "\n"
              + "absoluteUsedCapacity = " + absoluteUsedCapacity
              + " [= usedResourcesMemory / clusterResourceMemory]" + "\n"
              + "maxAMResourcePerQueuePercent = " + maxAMResourcePerQueuePercent
              + " [= configuredMaximumAMResourcePercent ]" + "\n"
              + "minimumAllocationFactor = " + minimumAllocationFactor
              + " [= (float)(maximumAllocationMemory - minimumAllocationMemory) / "
              + "maximumAllocationMemory ]" + "\n" + "maximumAllocation = "
              + maximumAllocation + " [= configuredMaxAllocation ]" + "\n"
              + "numContainers = " + numContainers
              + " [= currentNumContainers ]" + "\n" + "state = " + getState()
              + " [= configuredState ]" + "\n" + "acls = " + aclsString
              + " [= configuredAcls ]" + "\n"
              + "nodeLocalityDelay = " + nodeLocalityDelay + "\n"
              + "rackLocalityAdditionalDelay = "
              + rackLocalityAdditionalDelay + "\n"
              + "labels=" + labelStrBuilder.toString() + "\n"
              + "reservationsContinueLooking = "
              + reservationsContinueLooking + "\n" + "preemptionDisabled = "
              + getPreemptionDisabled() + "\n" + "defaultAppPriorityPerQueue = "
              + defaultAppPriorityPerQueue + "\npriority = " + priority
              + "\nmaxLifetime = " + getMaximumApplicationLifetime()
              + " seconds" + "\ndefaultLifetime = "
              + getDefaultApplicationLifetime() + " seconds");
    } finally {
      writeLock.unlock();
    }
  }

  /**
   * Used only by tests.
   */
  @Private
  public float getMinimumAllocationFactor() {
    return minimumAllocationFactor;
  }
  
  /**
   * Used only by tests.
   */
  @Private
  public float getMaxAMResourcePerQueuePercent() {
    return maxAMResourcePerQueuePercent;
  }

  public int getMaxApplications() {
    return maxApplications;
  }

  public int getMaxApplicationsPerUser() {
    return maxApplicationsPerUser;
  }

  /**
   *
   * @return UsersManager instance.
   */
  public UsersManager getUsersManager() {
    return usersManager;
  }

  @Override
  public AbstractUsersManager getAbstractUsersManager() {
    return usersManager;
  }

  @Override
  public List<CSQueue> getChildQueues() {
    return null;
  }
  
  /**
   * Set user limit.
   * @param userLimit new user limit
   */
  @VisibleForTesting
  void setUserLimit(float userLimit) {
    usersManager.setUserLimit(userLimit);
    usersManager.userLimitNeedsRecompute();
  }

  /**
   * Set user limit factor.
   * @param userLimitFactor new user limit factor
   */
  @VisibleForTesting
  void setUserLimitFactor(float userLimitFactor) {
    usersManager.setUserLimitFactor(userLimitFactor);
    usersManager.userLimitNeedsRecompute();
  }

  @Override
  public int getNumApplications() {
    readLock.lock();
    try {
      return getNumPendingApplications() + getNumActiveApplications() +
          getNumNonRunnableApps();
    } finally {
      readLock.unlock();
    }
  }

  public int getNumPendingApplications() {
    readLock.lock();
    try {
      return pendingOrderingPolicy.getNumSchedulableEntities();
    } finally {
      readLock.unlock();
    }
  }

  public int getNumActiveApplications() {
    readLock.lock();
    try {
      return orderingPolicy.getNumSchedulableEntities();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public int getNumPendingApplications(String user) {
    readLock.lock();
    try {
      User u = getUser(user);
      if (null == u) {
        return 0;
      }
      return u.getPendingApplications();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public int getNumActiveApplications(String user) {
    readLock.lock();
    try {
      User u = getUser(user);
      if (null == u) {
        return 0;
      }
      return u.getActiveApplications();
    } finally {
      readLock.unlock();
    }
  }

  @Private
  public float getUserLimit() {
    return usersManager.getUserLimit();
  }

  @Private
  public float getUserLimitFactor() {
    return usersManager.getUserLimitFactor();
  }

  @Override
  public QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive) {
    QueueInfo queueInfo = getQueueInfo();
    return queueInfo;
  }

  @Override
  public List<QueueUserACLInfo>
  getQueueUserAclInfo(UserGroupInformation user) {
    readLock.lock();
    try {
      QueueUserACLInfo userAclInfo = recordFactory.newRecordInstance(
          QueueUserACLInfo.class);
      List<QueueACL> operations = new ArrayList<>();
      for (QueueACL operation : QueueACL.values()) {
        if (hasAccess(operation, user)) {
          operations.add(operation);
        }
      }

      userAclInfo.setQueueName(getQueuePath());
      userAclInfo.setUserAcls(operations);
      return Collections.singletonList(userAclInfo);
    } finally {
      readLock.unlock();
    }

  }

  public String toString() {
    readLock.lock();
    try {
      return getQueuePath() + ": " + getCapacityOrWeightString()
          + ", " + "absoluteCapacity=" + queueCapacities.getAbsoluteCapacity()
          + ", " + "usedResources=" + queueUsage.getUsed() + ", "
          + "usedCapacity=" + getUsedCapacity() + ", " + "absoluteUsedCapacity="
          + getAbsoluteUsedCapacity() + ", " + "numApps=" + getNumApplications()
          + ", " + "numContainers=" + getNumContainers() + ", "
          + "effectiveMinResource=" +
          getEffectiveCapacity(CommonNodeLabelsManager.NO_LABEL) +
          " , effectiveMaxResource=" +
          getEffectiveMaxCapacity(CommonNodeLabelsManager.NO_LABEL);
    } finally {
      readLock.unlock();
    }
  }

  protected String getExtendedCapacityOrWeightString() {
    if (queueCapacities.getWeight() != -1) {
      return "weight = " + queueCapacities.getWeight()
          + " [= (float) configuredCapacity (with w suffix)] " + "\n"
          + "normalizedWeight = " + queueCapacities.getNormalizedWeight()
          + " [= (float) configuredCapacity / sum(configuredCapacity of " +
          "all queues under the parent)]";
    } else {
      return "capacity = " + queueCapacities.getCapacity()
          + " [= (float) configuredCapacity / 100 ]";
    }
  }

  @VisibleForTesting
  public User getUser(String userName) {
    return usersManager.getUser(userName);
  }

  @Private
  public List<AppPriorityACLGroup> getPriorityACLs() {
    readLock.lock();
    try {
      return new ArrayList<>(priorityAcls);
    } finally {
      readLock.unlock();
    }
  }

  protected void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource,
      CapacitySchedulerConfiguration configuration) throws
      IOException {

    writeLock.lock();
    try {
      // We skip reinitialize for dynamic queues, when this is called, and
      // new queue is different from this queue, we will make this queue to be
      // static queue.
      if (newlyParsedQueue != this) {
        this.setDynamicQueue(false);
      }

      // Sanity check
      if (!(newlyParsedQueue instanceof LeafQueue) || !newlyParsedQueue
          .getQueuePath().equals(getQueuePath())) {
        throw new IOException(
            "Trying to reinitialize " + getQueuePath() + " from "
                + newlyParsedQueue.getQueuePath());
      }

      LeafQueue newlyParsedLeafQueue = (LeafQueue) newlyParsedQueue;

      // don't allow the maximum allocation to be decreased in size
      // since we have already told running AM's the size
      Resource oldMax = getMaximumAllocation();
      Resource newMax = newlyParsedLeafQueue.getMaximumAllocation();

      if (!Resources.fitsIn(oldMax, newMax)) {
        throw new IOException("Trying to reinitialize " + getQueuePath()
            + " the maximum allocation size can not be decreased!"
            + " Current setting: " + oldMax + ", trying to set it to: "
            + newMax);
      }

      setupQueueConfigs(clusterResource, configuration);
    } finally {
      writeLock.unlock();
    }
  }

  @Override
  public void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource)
  throws IOException {
    reinitialize(newlyParsedQueue, clusterResource,
        csContext.getConfiguration());
  }

  @Override
  public void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName) {
    submitApplicationAttempt(application, userName, false);
  }

  @Override
  public void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName, boolean isMoveApp) {
    // Careful! Locking order is important!
    writeLock.lock();
    try {
      // TODO, should use getUser, use this method just to avoid UT failure
      // which is caused by wrong invoking order, will fix UT separately
      User user = usersManager.getUserAndAddIfAbsent(userName);

      // Add the attempt to our data-structures
      addApplicationAttempt(application, user);
    } finally {
      writeLock.unlock();
    }

    // We don't want to update metrics for move app
    if (!isMoveApp) {
      boolean unmanagedAM = application.getAppSchedulingInfo() != null &&
          application.getAppSchedulingInfo().isUnmanagedAM();
      metrics.submitAppAttempt(userName, unmanagedAM);
    }

    getParent().submitApplicationAttempt(application, userName);
  }

  @Override
  public void submitApplication(ApplicationId applicationId, String userName,
      String queue)  throws AccessControlException {
    // Careful! Locking order is important!
    validateSubmitApplication(applicationId, userName, queue);

    // Signal for expired auto deletion.
    updateLastSubmittedTimeStamp();

    // Inform the parent queue
    try {
      getParent().submitApplication(applicationId, userName, queue);
    } catch (AccessControlException ace) {
      LOG.info("Failed to submit application to parent-queue: " +
          getParent().getQueuePath(), ace);
      throw ace;
    }

  }

  public void validateSubmitApplication(ApplicationId applicationId,
      String userName, String queue) throws AccessControlException {
    writeLock.lock();
    try {
      // Check if the queue is accepting jobs
      if (getState() != QueueState.RUNNING) {
        String msg = "Queue " + getQueuePath()
            + " is STOPPED. Cannot accept submission of application: "
            + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }

      // Check submission limits for queues
      //TODO recalculate max applications because they can depend on capacity
      if (getNumApplications() >= getMaxApplications() && !(this instanceof AutoCreatedLeafQueue)) {
        String msg =
            "Queue " + getQueuePath() + " already has " + getNumApplications()
                + " applications,"
                + " cannot accept submission of application: " + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }

      // Check submission limits for the user on this queue
      User user = usersManager.getUserAndAddIfAbsent(userName);
      //TODO recalculate max applications because they can depend on capacity
      if (user.getTotalApplications() >= getMaxApplicationsPerUser() &&  !(this instanceof AutoCreatedLeafQueue)) {
        String msg = "Queue " + getQueuePath() + " already has " + user
            .getTotalApplications() + " applications from user " + userName
            + " cannot accept submission of application: " + applicationId;
        LOG.info(msg);
        throw new AccessControlException(msg);
      }
    } finally {
      writeLock.unlock();
    }

    try {
      getParent().validateSubmitApplication(applicationId, userName, queue);
    } catch (AccessControlException ace) {
      LOG.info("Failed to submit application to parent-queue: " + 
          getParent().getQueuePath(), ace);
      throw ace;
    }
  }
  
  public Resource getAMResourceLimit() {
    return queueUsage.getAMLimit();
  }

  public Resource getAMResourceLimitPerPartition(String nodePartition) {
    return queueUsage.getAMLimit(nodePartition);
  }

  @VisibleForTesting
  public Resource calculateAndGetAMResourceLimit() {
    return calculateAndGetAMResourceLimitPerPartition(
        RMNodeLabelsManager.NO_LABEL);
  }

  @VisibleForTesting
  public Resource getUserAMResourceLimit() {
    return getUserAMResourceLimitPerPartition(RMNodeLabelsManager.NO_LABEL,
         null);
  }

  public Resource getUserAMResourceLimitPerPartition(
      String nodePartition, String userName) {
    float userWeight = 1.0f;
    if (userName != null && getUser(userName) != null) {
      userWeight = getUser(userName).getWeight();
    }

    readLock.lock();
    try {
      /*
       * The user am resource limit is based on the same approach as the user
       * limit (as it should represent a subset of that). This means that it uses
       * the absolute queue capacity (per partition) instead of the max and is
       * modified by the userlimit and the userlimit factor as is the userlimit
       */
      float effectiveUserLimit = Math.max(usersManager.getUserLimit() / 100.0f,
          1.0f / Math.max(getAbstractUsersManager().getNumActiveUsers(), 1));
      float preWeightedUserLimit = effectiveUserLimit;
      effectiveUserLimit = Math.min(effectiveUserLimit * userWeight, 1.0f);

      Resource queuePartitionResource = getEffectiveCapacity(nodePartition);

      Resource userAMLimit = Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionResource,
          queueCapacities.getMaxAMResourcePercentage(nodePartition)
              * effectiveUserLimit * usersManager.getUserLimitFactor(),
          minimumAllocation);

      if (getUserLimitFactor() == -1) {
        userAMLimit = Resources.multiplyAndNormalizeUp(
            resourceCalculator, queuePartitionResource,
            queueCapacities.getMaxAMResourcePercentage(nodePartition),
            minimumAllocation);
      }

      userAMLimit =
          Resources.min(resourceCalculator, lastClusterResource,
              userAMLimit,
              Resources.clone(getAMResourceLimitPerPartition(nodePartition)));

      Resource preWeighteduserAMLimit =
          Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionResource,
          queueCapacities.getMaxAMResourcePercentage(nodePartition)
              * preWeightedUserLimit * usersManager.getUserLimitFactor(),
          minimumAllocation);

      if (getUserLimitFactor() == -1) {
        preWeighteduserAMLimit = Resources.multiplyAndNormalizeUp(
            resourceCalculator, queuePartitionResource,
            queueCapacities.getMaxAMResourcePercentage(nodePartition),
            minimumAllocation);
      }

      preWeighteduserAMLimit =
          Resources.min(resourceCalculator, lastClusterResource,
              preWeighteduserAMLimit,
              Resources.clone(getAMResourceLimitPerPartition(nodePartition)));
      queueUsage.setUserAMLimit(nodePartition, preWeighteduserAMLimit);

      LOG.debug("Effective user AM limit for \"{}\":{}. Effective weighted"
          + " user AM limit: {}. User weight: {}", userName,
          preWeighteduserAMLimit, userAMLimit, userWeight);
      return userAMLimit;
    } finally {
      readLock.unlock();
    }

  }

  public Resource calculateAndGetAMResourceLimitPerPartition(
      String nodePartition) {
    writeLock.lock();
    try {
      /*
       * For non-labeled partition, get the max value from resources currently
       * available to the queue and the absolute resources guaranteed for the
       * partition in the queue. For labeled partition, consider only the absolute
       * resources guaranteed. Multiply this value (based on labeled/
       * non-labeled), * with per-partition am-resource-percent to get the max am
       * resource limit for this queue and partition.
       */
      Resource queuePartitionResource = getEffectiveCapacity(nodePartition);

      Resource queueCurrentLimit = Resources.none();
      // For non-labeled partition, we need to consider the current queue
      // usage limit.
      if (nodePartition.equals(RMNodeLabelsManager.NO_LABEL)) {
        synchronized (queueResourceLimitsInfo){
          queueCurrentLimit = queueResourceLimitsInfo.getQueueCurrentLimit();
        }
      }

      float amResourcePercent = queueCapacities.getMaxAMResourcePercentage(
          nodePartition);

      // Current usable resource for this queue and partition is the max of
      // queueCurrentLimit and queuePartitionResource.
      // If any of the resources available to this queue are less than queue's
      // guarantee, use the guarantee as the queuePartitionUsableResource
      // because nothing less than the queue's guarantee should be used when
      // calculating the AM limit.
      Resource queuePartitionUsableResource = (Resources.fitsIn(
          resourceCalculator, queuePartitionResource, queueCurrentLimit)) ?
              queueCurrentLimit : queuePartitionResource;

      Resource amResouceLimit = Resources.multiplyAndNormalizeUp(
          resourceCalculator, queuePartitionUsableResource, amResourcePercent,
          minimumAllocation);

      metrics.setAMResouceLimit(nodePartition, amResouceLimit);
      queueUsage.setAMLimit(nodePartition, amResouceLimit);
      LOG.debug("Queue: {}, node label : {}, queue partition resource : {},"
          + " queue current limit : {}, queue partition usable resource : {},"
          + " amResourceLimit : {}", getQueuePath(), nodePartition,
          queuePartitionResource, queueCurrentLimit,
          queuePartitionUsableResource, amResouceLimit);
      return amResouceLimit;
    } finally {
      writeLock.unlock();
    }
  }

  protected void activateApplications() {
    writeLock.lock();
    try {
      // limit of allowed resource usage for application masters
      Map<String, Resource> userAmPartitionLimit =
          new HashMap<String, Resource>();

      // AM Resource Limit for accessible labels can be pre-calculated.
      // This will help in updating AMResourceLimit for all labels when queue
      // is initialized for the first time (when no applications are present).
      for (String nodePartition : getNodeLabelsForQueue()) {
        calculateAndGetAMResourceLimitPerPartition(nodePartition);
      }

      for (Iterator<FiCaSchedulerApp> fsApp =
           getPendingAppsOrderingPolicy()
               .getAssignmentIterator(IteratorSelector.EMPTY_ITERATOR_SELECTOR);
           fsApp.hasNext(); ) {
        FiCaSchedulerApp application = fsApp.next();
        ApplicationId applicationId = application.getApplicationId();

        // Get the am-node-partition associated with each application
        // and calculate max-am resource limit for this partition.
        String partitionName = application.getAppAMNodePartitionName();

        Resource amLimit = getAMResourceLimitPerPartition(partitionName);
        // Verify whether we already calculated am-limit for this label.
        if (amLimit == null) {
          amLimit = calculateAndGetAMResourceLimitPerPartition(partitionName);
        }
        // Check am resource limit.
        Resource amIfStarted = Resources.add(
            application.getAMResource(partitionName),
            queueUsage.getAMUsed(partitionName));

        if (LOG.isDebugEnabled()) {
          LOG.debug("application " + application.getId() + " AMResource "
              + application.getAMResource(partitionName)
              + " maxAMResourcePerQueuePercent " + maxAMResourcePerQueuePercent
              + " amLimit " + amLimit + " lastClusterResource "
              + lastClusterResource + " amIfStarted " + amIfStarted
              + " AM node-partition name " + partitionName);
        }

        if (!resourceCalculator.fitsIn(amIfStarted, amLimit)) {
          if (getNumActiveApplications() < 1 || (Resources.lessThanOrEqual(
              resourceCalculator, lastClusterResource,
              queueUsage.getAMUsed(partitionName), Resources.none()))) {
            LOG.warn("maximum-am-resource-percent is insufficient to start a"
                + " single application in queue, it is likely set too low."
                + " skipping enforcement to allow at least one application"
                + " to start");
          } else{
            application.updateAMContainerDiagnostics(AMState.INACTIVATED,
                CSAMContainerLaunchDiagnosticsConstants.QUEUE_AM_RESOURCE_LIMIT_EXCEED);
            LOG.debug("Not activating application {} as  amIfStarted: {}"
                + " exceeds amLimit: {}", applicationId, amIfStarted, amLimit);
            continue;
          }
        }

        // Check user am resource limit
        User user = usersManager.getUserAndAddIfAbsent(application.getUser());
        Resource userAMLimit = userAmPartitionLimit.get(partitionName);

        // Verify whether we already calculated user-am-limit for this label.
        if (userAMLimit == null) {
          userAMLimit = getUserAMResourceLimitPerPartition(partitionName,
              application.getUser());
          userAmPartitionLimit.put(partitionName, userAMLimit);
        }

        Resource userAmIfStarted = Resources.add(
            application.getAMResource(partitionName),
            user.getConsumedAMResources(partitionName));

{'fraction_non_alphanumeric': 0.09193942075738328, 'fraction_numerical': 0.0013845868364331902, 'mean_word_length': 3.1496179588935616, 'pattern_counts': {'":': 6, '<': 70, '<?xml version=': 0, '>': 72, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2411677', 'n_tokens_mistral': 18582, 'n_tokens_neox': 17972, 'n_words': 4142}