YARN-10916. Investigate and simplify GuaranteedOrZeroCapacityOverTimePolicy#computeQueueManagementChanges. Contributed by Szilard Nemeth
This commit is contained in:
parent
35b8441fd9
commit
20aeb5ecc3
@ -0,0 +1,107 @@
|
|||||||
|
/*
|
||||||
|
* 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.queuemanagement;
|
||||||
|
|
||||||
|
import org.apache.hadoop.classification.VisibleForTesting;
|
||||||
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacities;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.Set;
|
||||||
|
|
||||||
|
import static org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueueUtils.EPSILON;
|
||||||
|
|
||||||
|
public class DeactivatedLeafQueuesByLabel {
|
||||||
|
private String parentQueuePath;
|
||||||
|
private String nodeLabel;
|
||||||
|
private Map<String, QueueCapacities> deactivatedLeafQueues;
|
||||||
|
private float sumOfChildQueueActivatedCapacity;
|
||||||
|
private float parentAbsoluteCapacity;
|
||||||
|
private float leafQueueTemplateAbsoluteCapacity;
|
||||||
|
private float availableCapacity;
|
||||||
|
private float totalDeactivatedCapacity;
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
public DeactivatedLeafQueuesByLabel() {}
|
||||||
|
|
||||||
|
public DeactivatedLeafQueuesByLabel(
|
||||||
|
Map<String, QueueCapacities> deactivatedLeafQueues,
|
||||||
|
String parentQueuePath,
|
||||||
|
String nodeLabel,
|
||||||
|
float sumOfChildQueueActivatedCapacity,
|
||||||
|
float parentAbsoluteCapacity,
|
||||||
|
float leafQueueTemplateAbsoluteCapacity) {
|
||||||
|
this.parentQueuePath = parentQueuePath;
|
||||||
|
this.nodeLabel = nodeLabel;
|
||||||
|
this.deactivatedLeafQueues = deactivatedLeafQueues;
|
||||||
|
this.sumOfChildQueueActivatedCapacity = sumOfChildQueueActivatedCapacity;
|
||||||
|
this.parentAbsoluteCapacity = parentAbsoluteCapacity;
|
||||||
|
this.leafQueueTemplateAbsoluteCapacity = leafQueueTemplateAbsoluteCapacity;
|
||||||
|
|
||||||
|
this.totalDeactivatedCapacity = getTotalDeactivatedCapacity();
|
||||||
|
this.availableCapacity = parentAbsoluteCapacity - sumOfChildQueueActivatedCapacity +
|
||||||
|
this.totalDeactivatedCapacity + EPSILON;
|
||||||
|
}
|
||||||
|
|
||||||
|
float getTotalDeactivatedCapacity() {
|
||||||
|
float deactivatedCapacity = 0;
|
||||||
|
for (Map.Entry<String, QueueCapacities> deactivatedQueueCapacity :
|
||||||
|
deactivatedLeafQueues.entrySet()) {
|
||||||
|
deactivatedCapacity += deactivatedQueueCapacity.getValue().getAbsoluteCapacity(nodeLabel);
|
||||||
|
}
|
||||||
|
return deactivatedCapacity;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Set<String> getQueues() {
|
||||||
|
return deactivatedLeafQueues.keySet();
|
||||||
|
}
|
||||||
|
|
||||||
|
public void printToDebug(Logger logger) {
|
||||||
|
if (logger.isDebugEnabled()) {
|
||||||
|
logger.debug("Parent queue = {}, nodeLabel = {}, absCapacity = {}, " +
|
||||||
|
"leafQueueAbsoluteCapacity = {}, deactivatedCapacity = {}, " +
|
||||||
|
"absChildActivatedCapacity = {}, availableCapacity = {}",
|
||||||
|
parentQueuePath, nodeLabel, parentAbsoluteCapacity,
|
||||||
|
leafQueueTemplateAbsoluteCapacity, getTotalDeactivatedCapacity(),
|
||||||
|
sumOfChildQueueActivatedCapacity, availableCapacity);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
public int getMaxLeavesToBeActivated(int numPendingApps) {
|
||||||
|
float childQueueAbsoluteCapacity = leafQueueTemplateAbsoluteCapacity;
|
||||||
|
if (childQueueAbsoluteCapacity > 0) {
|
||||||
|
int numLeafQueuesNeeded = (int) Math.floor(availableCapacity / childQueueAbsoluteCapacity);
|
||||||
|
return Math.min(numLeafQueuesNeeded, numPendingApps);
|
||||||
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
public boolean canActivateLeafQueues() {
|
||||||
|
return availableCapacity >= leafQueueTemplateAbsoluteCapacity;
|
||||||
|
}
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
public void setAvailableCapacity(float availableCapacity) {
|
||||||
|
this.availableCapacity = availableCapacity;
|
||||||
|
}
|
||||||
|
|
||||||
|
@VisibleForTesting
|
||||||
|
public void setLeafQueueTemplateAbsoluteCapacity(float leafQueueTemplateAbsoluteCapacity) {
|
||||||
|
this.leafQueueTemplateAbsoluteCapacity = leafQueueTemplateAbsoluteCapacity;
|
||||||
|
}
|
||||||
|
}
|
@ -21,12 +21,9 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
|
|||||||
import org.apache.hadoop.classification.VisibleForTesting;
|
import org.apache.hadoop.classification.VisibleForTesting;
|
||||||
import org.apache.hadoop.yarn.api.records.Resource;
|
import org.apache.hadoop.yarn.api.records.Resource;
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueueUtils;
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueueUtils;
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
|
|
||||||
.QueueManagementDynamicEditPolicy;
|
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
|
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler
|
||||||
.SchedulerDynamicEditException;
|
.SchedulerDynamicEditException;
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
|
||||||
@ -84,6 +81,7 @@ import static org.apache.hadoop.yarn.server.resourcemanager.scheduler
|
|||||||
public class GuaranteedOrZeroCapacityOverTimePolicy
|
public class GuaranteedOrZeroCapacityOverTimePolicy
|
||||||
implements AutoCreatedQueueManagementPolicy {
|
implements AutoCreatedQueueManagementPolicy {
|
||||||
|
|
||||||
|
private static final int DEFAULT_QUEUE_PRINT_SIZE_LIMIT = 25;
|
||||||
private CapacitySchedulerContext scheduler;
|
private CapacitySchedulerContext scheduler;
|
||||||
private ManagedParentQueue managedParentQueue;
|
private ManagedParentQueue managedParentQueue;
|
||||||
|
|
||||||
@ -345,12 +343,11 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compute/Adjust child queue capacities
|
* Computes / adjusts child queue capacities for auto created leaf queues.
|
||||||
* for auto created leaf queues
|
* This method computes queue entitlements but does not update LeafQueueState or
|
||||||
* This computes queue entitlements but does not update LeafQueueState or
|
* queue capacities.
|
||||||
* queue capacities. Scheduler calls commitQueueManagemetChanges after
|
* Scheduler calls commitQueueManagementChanges after validation after applying queue changes
|
||||||
* validation after applying queue changes and commits to LeafQueueState
|
* and commits to LeafQueueState are done in commitQueueManagementChanges.
|
||||||
* are done in commitQueueManagementChanges.
|
|
||||||
*
|
*
|
||||||
* @return List of Queue Management change suggestions which could potentially
|
* @return List of Queue Management change suggestions which could potentially
|
||||||
* be committed/rejected by the scheduler due to validation failures
|
* be committed/rejected by the scheduler due to validation failures
|
||||||
@ -367,117 +364,98 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
managedParentQueue.getAutoCreatedQueueManagementPolicy());
|
managedParentQueue.getAutoCreatedQueueManagementPolicy());
|
||||||
|
|
||||||
//TODO : Add support for node labels on leaf queue template configurations
|
//TODO : Add support for node labels on leaf queue template configurations
|
||||||
//synch/add missing leaf queue(s) if any to state
|
//sync / add missing leaf queue(s) if any TO state
|
||||||
updateLeafQueueState();
|
updateLeafQueueState();
|
||||||
|
|
||||||
readLock.lock();
|
readLock.lock();
|
||||||
try {
|
try {
|
||||||
List<QueueManagementChange> queueManagementChanges = new ArrayList<>();
|
LeafQueueEntitlements leafQueueEntitlements = new LeafQueueEntitlements();
|
||||||
List<FiCaSchedulerApp> pendingApps = getSortedPendingApplications();
|
|
||||||
|
|
||||||
//Map of LeafQueue->QueueCapacities - keep adding the computed
|
|
||||||
// entitlements to this map and finally
|
|
||||||
// build the leaf queue configuration Template for all identified leaf
|
|
||||||
// queues
|
|
||||||
Map<String, QueueCapacities> leafQueueEntitlements = new HashMap<>();
|
|
||||||
for (String nodeLabel : leafQueueTemplateNodeLabels) {
|
for (String nodeLabel : leafQueueTemplateNodeLabels) {
|
||||||
// check if any leaf queues need to be deactivated based on pending
|
DeactivatedLeafQueuesByLabel deactivatedLeafQueues =
|
||||||
// applications
|
deactivateLeafQueues(nodeLabel, leafQueueEntitlements);
|
||||||
float parentAbsoluteCapacity =
|
deactivatedLeafQueues.printToDebug(LOG);
|
||||||
managedParentQueue.getQueueCapacities().getAbsoluteCapacity(
|
|
||||||
nodeLabel);
|
|
||||||
float leafQueueTemplateAbsoluteCapacity =
|
|
||||||
leafQueueTemplateCapacities.getAbsoluteCapacity(nodeLabel);
|
|
||||||
Map<String, QueueCapacities> deactivatedLeafQueues =
|
|
||||||
deactivateLeafQueuesIfInActive(managedParentQueue, nodeLabel,
|
|
||||||
leafQueueEntitlements);
|
|
||||||
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
if ( deactivatedLeafQueues.size() > 0) {
|
|
||||||
LOG.debug("Parent queue = {}, " +
|
|
||||||
", nodeLabel = {}, deactivated leaf queues = [{}] ",
|
|
||||||
managedParentQueue.getQueuePath(), nodeLabel,
|
|
||||||
deactivatedLeafQueues.size() > 25 ? deactivatedLeafQueues
|
|
||||||
.size() : deactivatedLeafQueues);
|
|
||||||
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
float deactivatedCapacity = getTotalDeactivatedCapacity(
|
|
||||||
deactivatedLeafQueues, nodeLabel);
|
|
||||||
|
|
||||||
float sumOfChildQueueActivatedCapacity = parentQueueState.
|
|
||||||
getAbsoluteActivatedChildQueueCapacity(nodeLabel);
|
|
||||||
|
|
||||||
//Check if we need to activate anything at all?
|
//Check if we need to activate anything at all?
|
||||||
float availableCapacity =
|
if (deactivatedLeafQueues.canActivateLeafQueues()) {
|
||||||
parentAbsoluteCapacity - sumOfChildQueueActivatedCapacity
|
activateLeafQueues(leafQueueEntitlements, nodeLabel, deactivatedLeafQueues);
|
||||||
+ deactivatedCapacity + EPSILON;
|
|
||||||
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
LOG.debug("Parent queue = " + managedParentQueue.getQueuePath()
|
|
||||||
+ ", nodeLabel = " + nodeLabel + ", absCapacity = "
|
|
||||||
+ parentAbsoluteCapacity + ", leafQueueAbsoluteCapacity = "
|
|
||||||
+ leafQueueTemplateAbsoluteCapacity + ", deactivatedCapacity = "
|
|
||||||
+ deactivatedCapacity + " , absChildActivatedCapacity = "
|
|
||||||
+ sumOfChildQueueActivatedCapacity + ", availableCapacity = "
|
|
||||||
+ availableCapacity);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (availableCapacity >= leafQueueTemplateAbsoluteCapacity) {
|
|
||||||
//sort applications across leaf queues by submit time
|
|
||||||
if (pendingApps.size() > 0) {
|
|
||||||
int maxLeafQueuesTobeActivated = getMaxLeavesToBeActivated(
|
|
||||||
availableCapacity, leafQueueTemplateAbsoluteCapacity,
|
|
||||||
pendingApps.size());
|
|
||||||
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
LOG.debug("Parent queue = " + managedParentQueue.getQueuePath()
|
|
||||||
+ " : Found " + maxLeafQueuesTobeActivated + " leaf queues"
|
|
||||||
+ " to be activated with " + pendingApps.size() + " apps ");
|
|
||||||
}
|
|
||||||
|
|
||||||
LinkedHashSet<String> leafQueuesToBeActivated = getSortedLeafQueues(
|
|
||||||
nodeLabel, pendingApps, maxLeafQueuesTobeActivated,
|
|
||||||
deactivatedLeafQueues.keySet());
|
|
||||||
|
|
||||||
//Compute entitlement changes for the identified leaf queues
|
|
||||||
// which is appended to the List of computedEntitlements
|
|
||||||
updateLeafQueueCapacitiesByLabel(nodeLabel, leafQueuesToBeActivated,
|
|
||||||
leafQueueEntitlements);
|
|
||||||
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
if (leafQueuesToBeActivated.size() > 0) {
|
|
||||||
LOG.debug("Activated leaf queues : [{}]",
|
|
||||||
leafQueuesToBeActivated.size() < 25 ?
|
|
||||||
leafQueuesToBeActivated : leafQueuesToBeActivated.size());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
//Populate new entitlements
|
//Populate new entitlements
|
||||||
|
return leafQueueEntitlements.mapToQueueManagementChanges((leafQueueName, capacities) -> {
|
||||||
for (final Iterator<Map.Entry<String, QueueCapacities>> iterator =
|
|
||||||
leafQueueEntitlements.entrySet().iterator(); iterator.hasNext(); ) {
|
|
||||||
Map.Entry<String, QueueCapacities> queueCapacities = iterator.next();
|
|
||||||
String leafQueueName = queueCapacities.getKey();
|
|
||||||
AutoCreatedLeafQueue leafQueue =
|
AutoCreatedLeafQueue leafQueue =
|
||||||
(AutoCreatedLeafQueue) scheduler.getCapacitySchedulerQueueManager()
|
(AutoCreatedLeafQueue) scheduler.getCapacitySchedulerQueueManager()
|
||||||
.getQueue(leafQueueName);
|
.getQueue(leafQueueName);
|
||||||
AutoCreatedLeafQueueConfig newTemplate = buildTemplate(
|
AutoCreatedLeafQueueConfig newTemplate = buildTemplate(capacities);
|
||||||
queueCapacities.getValue());
|
return new QueueManagementChange.UpdateQueue(leafQueue, newTemplate);
|
||||||
queueManagementChanges.add(
|
});
|
||||||
new QueueManagementChange.UpdateQueue(leafQueue, newTemplate));
|
|
||||||
|
|
||||||
}
|
|
||||||
return queueManagementChanges;
|
|
||||||
} finally {
|
} finally {
|
||||||
readLock.unlock();
|
readLock.unlock();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void activateLeafQueues(LeafQueueEntitlements leafQueueEntitlements, String nodeLabel,
|
||||||
|
DeactivatedLeafQueuesByLabel deactivatedLeafQueues) throws SchedulerDynamicEditException {
|
||||||
|
//sort applications across leaf queues by submit time
|
||||||
|
List<FiCaSchedulerApp> pendingApps = getSortedPendingApplications();
|
||||||
|
if (pendingApps.size() > 0) {
|
||||||
|
int maxLeafQueuesTobeActivated = deactivatedLeafQueues.
|
||||||
|
getMaxLeavesToBeActivated(pendingApps.size());
|
||||||
|
|
||||||
|
if (LOG.isDebugEnabled()) {
|
||||||
|
LOG.debug("Parent queue = {}, Found {} leaf queues to be activated with {} aps",
|
||||||
|
managedParentQueue.getQueuePath(), maxLeafQueuesTobeActivated, pendingApps.size());
|
||||||
|
}
|
||||||
|
|
||||||
|
Set<String> leafQueuesToBeActivated = getSortedLeafQueues(
|
||||||
|
nodeLabel, pendingApps, maxLeafQueuesTobeActivated,
|
||||||
|
deactivatedLeafQueues.getQueues());
|
||||||
|
|
||||||
|
// Compute entitlement changes for the identified leaf queues
|
||||||
|
// which is appended to the List of computedEntitlements
|
||||||
|
updateLeafQueueCapacitiesByLabel(nodeLabel, leafQueuesToBeActivated, leafQueueEntitlements);
|
||||||
|
|
||||||
|
if (LOG.isDebugEnabled() && leafQueuesToBeActivated.size() > 0) {
|
||||||
|
LOG.debug("Activated leaf queues : [{}]",
|
||||||
|
getListContentsUpToLimit(leafQueuesToBeActivated));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private Object getListContentsUpToLimit(Set<String> leafQueuesToBeActivated) {
|
||||||
|
return leafQueuesToBeActivated.size() < DEFAULT_QUEUE_PRINT_SIZE_LIMIT ?
|
||||||
|
leafQueuesToBeActivated : leafQueuesToBeActivated.size();
|
||||||
|
}
|
||||||
|
|
||||||
|
private Object getMapUpToLimit(Map<String, QueueCapacities> deactivatedLeafQueues) {
|
||||||
|
return deactivatedLeafQueues.size() > DEFAULT_QUEUE_PRINT_SIZE_LIMIT ?
|
||||||
|
deactivatedLeafQueues.size() : deactivatedLeafQueues;
|
||||||
|
}
|
||||||
|
|
||||||
|
private DeactivatedLeafQueuesByLabel deactivateLeafQueues(String nodeLabel,
|
||||||
|
LeafQueueEntitlements leafQueueEntitlements) throws SchedulerDynamicEditException {
|
||||||
|
// check if any leaf queues need to be deactivated based on pending applications
|
||||||
|
float parentAbsoluteCapacity =
|
||||||
|
managedParentQueue.getQueueCapacities().getAbsoluteCapacity(nodeLabel);
|
||||||
|
float leafQueueTemplateAbsoluteCapacity =
|
||||||
|
leafQueueTemplateCapacities.getAbsoluteCapacity(nodeLabel);
|
||||||
|
Map<String, QueueCapacities> deactivatedLeafQueues =
|
||||||
|
deactivateLeafQueuesIfInActive(managedParentQueue, nodeLabel, leafQueueEntitlements);
|
||||||
|
|
||||||
|
if (LOG.isDebugEnabled() && deactivatedLeafQueues.size() > 0) {
|
||||||
|
LOG.debug("Parent queue = {}, nodeLabel = {}, deactivated leaf queues = [{}] ",
|
||||||
|
managedParentQueue.getQueuePath(), nodeLabel,
|
||||||
|
getMapUpToLimit(deactivatedLeafQueues));
|
||||||
|
}
|
||||||
|
|
||||||
|
return new DeactivatedLeafQueuesByLabel(deactivatedLeafQueues,
|
||||||
|
managedParentQueue.getQueuePath(),
|
||||||
|
nodeLabel,
|
||||||
|
parentQueueState.getAbsoluteActivatedChildQueueCapacity(nodeLabel),
|
||||||
|
parentAbsoluteCapacity,
|
||||||
|
leafQueueTemplateAbsoluteCapacity);
|
||||||
|
}
|
||||||
|
|
||||||
private void updateTemplateAbsoluteCapacities(QueueCapacities parentQueueCapacities,
|
private void updateTemplateAbsoluteCapacities(QueueCapacities parentQueueCapacities,
|
||||||
GuaranteedOrZeroCapacityOverTimePolicy policy) {
|
GuaranteedOrZeroCapacityOverTimePolicy policy) {
|
||||||
writeLock.lock();
|
writeLock.lock();
|
||||||
@ -496,19 +474,6 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
updateTemplateAbsoluteCapacities(queueCapacities, this);
|
updateTemplateAbsoluteCapacities(queueCapacities, this);
|
||||||
}
|
}
|
||||||
|
|
||||||
private float getTotalDeactivatedCapacity(
|
|
||||||
Map<String, QueueCapacities> deactivatedLeafQueues, String nodeLabel) {
|
|
||||||
float deactivatedCapacity = 0;
|
|
||||||
for (Iterator<Map.Entry<String, QueueCapacities>> iterator =
|
|
||||||
deactivatedLeafQueues.entrySet().iterator(); iterator.hasNext(); ) {
|
|
||||||
Map.Entry<String, QueueCapacities> deactivatedQueueCapacity =
|
|
||||||
iterator.next();
|
|
||||||
deactivatedCapacity +=
|
|
||||||
deactivatedQueueCapacity.getValue().getAbsoluteCapacity(nodeLabel);
|
|
||||||
}
|
|
||||||
return deactivatedCapacity;
|
|
||||||
}
|
|
||||||
|
|
||||||
@VisibleForTesting
|
@VisibleForTesting
|
||||||
void updateLeafQueueState() {
|
void updateLeafQueueState() {
|
||||||
writeLock.lock();
|
writeLock.lock();
|
||||||
@ -608,9 +573,15 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Map of LeafQueue -> QueueCapacities - keep adding the computed
|
||||||
|
* entitlements to this map and finally
|
||||||
|
* build the leaf queue configuration Template for all identified leaf
|
||||||
|
* queues
|
||||||
|
*/
|
||||||
private Map<String, QueueCapacities> deactivateLeafQueuesIfInActive(
|
private Map<String, QueueCapacities> deactivateLeafQueuesIfInActive(
|
||||||
ParentQueue parentQueue, String nodeLabel,
|
ParentQueue parentQueue, String nodeLabel,
|
||||||
Map<String, QueueCapacities> leafQueueEntitlements)
|
LeafQueueEntitlements leafQueueEntitlements)
|
||||||
throws SchedulerDynamicEditException {
|
throws SchedulerDynamicEditException {
|
||||||
Map<String, QueueCapacities> deactivatedQueues = new HashMap<>();
|
Map<String, QueueCapacities> deactivatedQueues = new HashMap<>();
|
||||||
|
|
||||||
@ -618,16 +589,9 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
AutoCreatedLeafQueue leafQueue = (AutoCreatedLeafQueue) childQueue;
|
AutoCreatedLeafQueue leafQueue = (AutoCreatedLeafQueue) childQueue;
|
||||||
if (leafQueue != null) {
|
if (leafQueue != null) {
|
||||||
if (isActive(leafQueue, nodeLabel) && !hasPendingApps(leafQueue)) {
|
if (isActive(leafQueue, nodeLabel) && !hasPendingApps(leafQueue)) {
|
||||||
if (!leafQueueEntitlements.containsKey(leafQueue.getQueuePath())) {
|
QueueCapacities capacities = leafQueueEntitlements.getCapacityOfQueue(leafQueue);
|
||||||
leafQueueEntitlements.put(leafQueue.getQueuePath(),
|
|
||||||
new QueueCapacities(false));
|
|
||||||
}
|
|
||||||
|
|
||||||
QueueCapacities capacities = leafQueueEntitlements.get(
|
|
||||||
leafQueue.getQueuePath());
|
|
||||||
updateToZeroCapacity(capacities, nodeLabel, (LeafQueue)childQueue);
|
updateToZeroCapacity(capacities, nodeLabel, (LeafQueue)childQueue);
|
||||||
deactivatedQueues.put(leafQueue.getQueuePath(),
|
deactivatedQueues.put(leafQueue.getQueuePath(), leafQueueTemplateCapacities);
|
||||||
leafQueueTemplateCapacities);
|
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
LOG.warn("Could not find queue in scheduler while trying" + " to "
|
LOG.warn("Could not find queue in scheduler while trying" + " to "
|
||||||
@ -640,35 +604,16 @@ public class GuaranteedOrZeroCapacityOverTimePolicy
|
|||||||
|
|
||||||
private void updateLeafQueueCapacitiesByLabel(String nodeLabel,
|
private void updateLeafQueueCapacitiesByLabel(String nodeLabel,
|
||||||
Set<String> leafQueuesToBeActivated,
|
Set<String> leafQueuesToBeActivated,
|
||||||
Map<String, QueueCapacities> leafQueueEntitlements) {
|
LeafQueueEntitlements leafQueueEntitlements) {
|
||||||
for (String curLeafQueue : leafQueuesToBeActivated) {
|
for (String leafQueue : leafQueuesToBeActivated) {
|
||||||
if (!leafQueueEntitlements.containsKey(curLeafQueue)) {
|
QueueCapacities capacities = leafQueueEntitlements.getCapacityOfQueueByPath(leafQueue);
|
||||||
leafQueueEntitlements.put(curLeafQueue, new QueueCapacities(false));
|
|
||||||
// Activate queues if capacity is available
|
|
||||||
}
|
|
||||||
|
|
||||||
QueueCapacities capacities = leafQueueEntitlements.get(curLeafQueue);
|
|
||||||
updateCapacityFromTemplate(capacities, nodeLabel);
|
updateCapacityFromTemplate(capacities, nodeLabel);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@VisibleForTesting
|
|
||||||
public int getMaxLeavesToBeActivated(float availableCapacity,
|
|
||||||
float childQueueAbsoluteCapacity, int numPendingApps)
|
|
||||||
throws SchedulerDynamicEditException {
|
|
||||||
|
|
||||||
if (childQueueAbsoluteCapacity > 0) {
|
|
||||||
int numLeafQueuesNeeded = (int) Math.floor(
|
|
||||||
availableCapacity / childQueueAbsoluteCapacity);
|
|
||||||
|
|
||||||
return Math.min(numLeafQueuesNeeded, numPendingApps);
|
|
||||||
}
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Commit queue management changes - which involves updating required state
|
* Commit queue management changes - which involves updating required state
|
||||||
* on parent/underlying leaf queues
|
* on parent/underlying leaf queues.
|
||||||
*
|
*
|
||||||
* @param queueManagementChanges Queue Management changes to commit
|
* @param queueManagementChanges Queue Management changes to commit
|
||||||
* @throws SchedulerDynamicEditException when validation fails
|
* @throws SchedulerDynamicEditException when validation fails
|
||||||
|
@ -0,0 +1,52 @@
|
|||||||
|
/*
|
||||||
|
* 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.queuemanagement;
|
||||||
|
|
||||||
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.AutoCreatedLeafQueue;
|
||||||
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacities;
|
||||||
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueManagementChange;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.function.BiFunction;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
|
public class LeafQueueEntitlements {
|
||||||
|
private final Map<String, QueueCapacities> entitlements = new HashMap<>();
|
||||||
|
|
||||||
|
public QueueCapacities getCapacityOfQueue(AutoCreatedLeafQueue leafQueue) {
|
||||||
|
return getCapacityOfQueueByPath(leafQueue.getQueuePath());
|
||||||
|
}
|
||||||
|
|
||||||
|
public QueueCapacities getCapacityOfQueueByPath(String leafQueuePath) {
|
||||||
|
if (!entitlements.containsKey(leafQueuePath)) {
|
||||||
|
entitlements.put(leafQueuePath, new QueueCapacities(false));
|
||||||
|
}
|
||||||
|
return entitlements.get(leafQueuePath);
|
||||||
|
}
|
||||||
|
|
||||||
|
public Map<String, QueueCapacities> getEntitlements() {
|
||||||
|
return entitlements;
|
||||||
|
}
|
||||||
|
|
||||||
|
public List<QueueManagementChange> mapToQueueManagementChanges(
|
||||||
|
BiFunction<String, QueueCapacities, QueueManagementChange> func) {
|
||||||
|
return entitlements.entrySet().stream().map(e -> func.apply(e.getKey(), e.getValue()))
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
}
|
||||||
|
}
|
@ -1,4 +1,4 @@
|
|||||||
/**
|
/*
|
||||||
* Licensed to the Apache Software Foundation (ASF) under one
|
* Licensed to the Apache Software Foundation (ASF) under one
|
||||||
* or more contributor license agreements. See the NOTICE file
|
* or more contributor license agreements. See the NOTICE file
|
||||||
* distributed with this work for additional information
|
* distributed with this work for additional information
|
||||||
@ -15,26 +15,9 @@
|
|||||||
* See the License for the specific language governing permissions and
|
* See the License for the specific language governing permissions and
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
|
|
||||||
|
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerDynamicEditException;
|
/**
|
||||||
|
* Contains classes that are related to the newer
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
|
* version of CS placement engine.
|
||||||
.queuemanagement.GuaranteedOrZeroCapacityOverTimePolicy;
|
*/
|
||||||
import org.junit.Test;
|
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.queuemanagement;
|
||||||
|
|
||||||
import static org.junit.Assert.assertEquals;
|
|
||||||
|
|
||||||
public class TestGuaranteedOrZeroCapacityOverTimePolicy {
|
|
||||||
|
|
||||||
@Test
|
|
||||||
public void testGetMaxLeavesToBeActivated()
|
|
||||||
throws SchedulerDynamicEditException {
|
|
||||||
GuaranteedOrZeroCapacityOverTimePolicy policy =
|
|
||||||
new GuaranteedOrZeroCapacityOverTimePolicy();
|
|
||||||
|
|
||||||
assertEquals(1, policy.getMaxLeavesToBeActivated(0.17f, 0.03f, 1));
|
|
||||||
assertEquals(5, policy.getMaxLeavesToBeActivated(0.17f, 0.03f, 7));
|
|
||||||
assertEquals(0, policy.getMaxLeavesToBeActivated(0, 0.03f, 10));
|
|
||||||
}
|
|
||||||
}
|
|
@ -0,0 +1,42 @@
|
|||||||
|
/*
|
||||||
|
* 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.queuemanagement;
|
||||||
|
|
||||||
|
import org.junit.Test;
|
||||||
|
|
||||||
|
import static org.junit.Assert.*;
|
||||||
|
import static org.mockito.Mockito.spy;
|
||||||
|
|
||||||
|
public class TestDeactivatedLeafQueuesByLabel {
|
||||||
|
@Test
|
||||||
|
public void testGetMaxLeavesToBeActivated() {
|
||||||
|
DeactivatedLeafQueuesByLabel d1 = spy(DeactivatedLeafQueuesByLabel.class);
|
||||||
|
d1.setAvailableCapacity(0.17f);
|
||||||
|
d1.setLeafQueueTemplateAbsoluteCapacity(0.03f);
|
||||||
|
assertEquals(1, d1.getMaxLeavesToBeActivated(1));
|
||||||
|
|
||||||
|
DeactivatedLeafQueuesByLabel d2 = spy(DeactivatedLeafQueuesByLabel.class);
|
||||||
|
d2.setAvailableCapacity(0.17f);
|
||||||
|
d2.setLeafQueueTemplateAbsoluteCapacity(0.03f);
|
||||||
|
assertEquals(5, d2.getMaxLeavesToBeActivated(7));
|
||||||
|
|
||||||
|
DeactivatedLeafQueuesByLabel d3 = spy(DeactivatedLeafQueuesByLabel.class);
|
||||||
|
d3.setAvailableCapacity(0f);
|
||||||
|
d3.setLeafQueueTemplateAbsoluteCapacity(0.03f);
|
||||||
|
assertEquals(0, d3.getMaxLeavesToBeActivated(10));
|
||||||
|
}
|
||||||
|
}
|
Loading…
x
Reference in New Issue
Block a user