YARN-10949. Simplify AbstractCSQueue#updateMaxAppRelatedField and find a more meaningful name for this method. Contributed by Andras Gyori
This commit is contained in:
parent
414d40155c
commit
35b8441fd9
|
@ -1462,53 +1462,6 @@ public abstract class AbstractCSQueue implements CSQueue {
|
||||||
configuredMaxResource, parentMaxResource));
|
configuredMaxResource, parentMaxResource));
|
||||||
}
|
}
|
||||||
|
|
||||||
void updateMaxAppRelatedField(CapacitySchedulerConfiguration conf,
|
|
||||||
LeafQueue leafQueue) {
|
|
||||||
int maxApplications = conf.getMaximumApplicationsPerQueue(queuePath);
|
|
||||||
int maxGlobalPerQueueApps = conf.getGlobalMaximumApplicationsPerQueue();
|
|
||||||
String maxLabel = RMNodeLabelsManager.NO_LABEL;
|
|
||||||
|
|
||||||
if (maxApplications < 0) {
|
|
||||||
for (String label : configuredNodeLabels) {
|
|
||||||
int maxApplicationsByLabel = 0;
|
|
||||||
if (maxGlobalPerQueueApps > 0) {
|
|
||||||
// In absolute mode, should
|
|
||||||
// shrink when change to corresponding label capacity.
|
|
||||||
maxApplicationsByLabel = this.capacityConfigType
|
|
||||||
!= CapacityConfigType.ABSOLUTE_RESOURCE ?
|
|
||||||
maxGlobalPerQueueApps :
|
|
||||||
(int) (maxGlobalPerQueueApps * queueCapacities
|
|
||||||
.getAbsoluteCapacity(label));
|
|
||||||
} else {
|
|
||||||
maxApplicationsByLabel = (int) (conf.getMaximumSystemApplications()
|
|
||||||
* queueCapacities.getAbsoluteCapacity(label));
|
|
||||||
}
|
|
||||||
if (maxApplicationsByLabel > maxApplications) {
|
|
||||||
maxApplications = maxApplicationsByLabel;
|
|
||||||
maxLabel = label;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
leafQueue.setMaxApplications(maxApplications);
|
|
||||||
|
|
||||||
int maxApplicationsPerUser = Math.min(maxApplications,
|
|
||||||
(int) (maxApplications
|
|
||||||
* (leafQueue.getUsersManager().getUserLimit() / 100.0f)
|
|
||||||
* leafQueue.getUsersManager().getUserLimitFactor()));
|
|
||||||
if (leafQueue.getUsersManager().getUserLimitFactor() == -1) {
|
|
||||||
maxApplicationsPerUser = maxApplications;
|
|
||||||
}
|
|
||||||
|
|
||||||
leafQueue.setMaxApplicationsPerUser(maxApplicationsPerUser);
|
|
||||||
LOG.info("LeafQueue:" + leafQueue.getQueuePath() +
|
|
||||||
"update max app related, maxApplications="
|
|
||||||
+ maxApplications + ", maxApplicationsPerUser="
|
|
||||||
+ maxApplicationsPerUser + ", Abs Cap:" + queueCapacities
|
|
||||||
.getAbsoluteCapacity(maxLabel) + ", Cap: " + queueCapacities
|
|
||||||
.getCapacity(maxLabel) + ", MaxCap : " + queueCapacities
|
|
||||||
.getMaximumCapacity(maxLabel));
|
|
||||||
}
|
|
||||||
|
|
||||||
void deriveCapacityFromAbsoluteConfigurations(String label,
|
void deriveCapacityFromAbsoluteConfigurations(String label,
|
||||||
Resource clusterResource) {
|
Resource clusterResource) {
|
||||||
// Update capacity with a float calculated from the parent's minResources
|
// Update capacity with a float calculated from the parent's minResources
|
||||||
|
|
|
@ -461,6 +461,10 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
|
||||||
return getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label + DOT;
|
return getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label + DOT;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void setMaximumSystemApplications(int numMaxApps) {
|
||||||
|
setInt(MAXIMUM_SYSTEM_APPLICATIONS, numMaxApps);
|
||||||
|
}
|
||||||
|
|
||||||
public int getMaximumSystemApplications() {
|
public int getMaximumSystemApplications() {
|
||||||
int maxApplications =
|
int maxApplications =
|
||||||
getInt(MAXIMUM_SYSTEM_APPLICATIONS, DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS);
|
getInt(MAXIMUM_SYSTEM_APPLICATIONS, DEFAULT_MAXIMUM_SYSTEM_APPLICATIIONS);
|
||||||
|
|
|
@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.*;
|
import java.util.*;
|
||||||
import java.util.Map.Entry;
|
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.ConcurrentMap;
|
import java.util.concurrent.ConcurrentMap;
|
||||||
|
|
||||||
|
@ -1939,8 +1938,9 @@ public class LeafQueue extends AbstractCSQueue {
|
||||||
updateAbsoluteCapacities();
|
updateAbsoluteCapacities();
|
||||||
|
|
||||||
super.updateEffectiveResources(clusterResource);
|
super.updateEffectiveResources(clusterResource);
|
||||||
super.updateMaxAppRelatedField(csContext.getConfiguration(),
|
|
||||||
this);
|
// Update maximum applications for the queue and for users
|
||||||
|
updateMaximumApplications(csContext.getConfiguration());
|
||||||
|
|
||||||
updateCurrentResourceLimits(currentResourceLimits, clusterResource);
|
updateCurrentResourceLimits(currentResourceLimits, clusterResource);
|
||||||
|
|
||||||
|
@ -2326,6 +2326,58 @@ public class LeafQueue extends AbstractCSQueue {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void updateMaximumApplications(CapacitySchedulerConfiguration conf) {
|
||||||
|
int maxAppsForQueue = conf.getMaximumApplicationsPerQueue(getQueuePath());
|
||||||
|
|
||||||
|
int maxDefaultPerQueueApps = conf.getGlobalMaximumApplicationsPerQueue();
|
||||||
|
int maxSystemApps = conf.getMaximumSystemApplications();
|
||||||
|
int baseMaxApplications = maxDefaultPerQueueApps > 0 ?
|
||||||
|
Math.min(maxDefaultPerQueueApps, maxSystemApps)
|
||||||
|
: maxSystemApps;
|
||||||
|
|
||||||
|
String maxLabel = RMNodeLabelsManager.NO_LABEL;
|
||||||
|
if (maxAppsForQueue < 0) {
|
||||||
|
if (maxDefaultPerQueueApps > 0 && this.capacityConfigType
|
||||||
|
!= CapacityConfigType.ABSOLUTE_RESOURCE) {
|
||||||
|
maxAppsForQueue = baseMaxApplications;
|
||||||
|
} else {
|
||||||
|
for (String label : configuredNodeLabels) {
|
||||||
|
int maxApplicationsByLabel = (int) (baseMaxApplications
|
||||||
|
* queueCapacities.getAbsoluteCapacity(label));
|
||||||
|
if (maxApplicationsByLabel > maxAppsForQueue) {
|
||||||
|
maxAppsForQueue = maxApplicationsByLabel;
|
||||||
|
maxLabel = label;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
setMaxApplications(maxAppsForQueue);
|
||||||
|
|
||||||
|
updateMaxAppsPerUser();
|
||||||
|
|
||||||
|
LOG.info("LeafQueue:" + getQueuePath() +
|
||||||
|
"update max app related, maxApplications="
|
||||||
|
+ maxAppsForQueue + ", maxApplicationsPerUser="
|
||||||
|
+ maxApplicationsPerUser + ", Abs Cap:" + queueCapacities
|
||||||
|
.getAbsoluteCapacity(maxLabel) + ", Cap: " + queueCapacities
|
||||||
|
.getCapacity(maxLabel) + ", MaxCap : " + queueCapacities
|
||||||
|
.getMaximumCapacity(maxLabel));
|
||||||
|
}
|
||||||
|
|
||||||
|
private void updateMaxAppsPerUser() {
|
||||||
|
int maxAppsPerUser = maxApplications;
|
||||||
|
if (getUsersManager().getUserLimitFactor() != -1) {
|
||||||
|
int maxApplicationsWithUserLimits = (int) (maxApplications
|
||||||
|
* (getUsersManager().getUserLimit() / 100.0f)
|
||||||
|
* getUsersManager().getUserLimitFactor());
|
||||||
|
maxAppsPerUser = Math.min(maxApplications,
|
||||||
|
maxApplicationsWithUserLimits);
|
||||||
|
}
|
||||||
|
|
||||||
|
setMaxApplicationsPerUser(maxAppsPerUser);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Get all valid users in this queue.
|
* Get all valid users in this queue.
|
||||||
* @return user list
|
* @return user list
|
||||||
|
|
|
@ -870,29 +870,7 @@ public class TestApplicationLimits {
|
||||||
+ "submission of application: " + app3.getApplicationId(),
|
+ "submission of application: " + app3.getApplicationId(),
|
||||||
app3.getDiagnostics().toString());
|
app3.getDiagnostics().toString());
|
||||||
|
|
||||||
// based on Global limit of queue usert application is rejected
|
// based on per user max app settings, app should be rejected instantly
|
||||||
RMApp app11 = MockRMAppSubmitter.submit(rm,
|
|
||||||
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
|
||||||
.withAppName("app")
|
|
||||||
.withUser("user")
|
|
||||||
.withAcls(null)
|
|
||||||
.withQueue("d")
|
|
||||||
.withWaitForAppAcceptedState(false)
|
|
||||||
.build());
|
|
||||||
rm.drainEvents();
|
|
||||||
rm.waitForState(app11.getApplicationId(), RMAppState.ACCEPTED);
|
|
||||||
assertEquals(RMAppState.ACCEPTED, app11.getState());
|
|
||||||
RMApp app12 = MockRMAppSubmitter.submit(rm,
|
|
||||||
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
|
||||||
.withAppName("app")
|
|
||||||
.withUser("user")
|
|
||||||
.withAcls(null)
|
|
||||||
.withQueue("d")
|
|
||||||
.withWaitForAppAcceptedState(false)
|
|
||||||
.build());
|
|
||||||
rm.drainEvents();
|
|
||||||
rm.waitForState(app12.getApplicationId(), RMAppState.ACCEPTED);
|
|
||||||
assertEquals(RMAppState.ACCEPTED, app12.getState());
|
|
||||||
RMApp app13 = MockRMAppSubmitter.submit(rm,
|
RMApp app13 = MockRMAppSubmitter.submit(rm,
|
||||||
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
||||||
.withAppName("app")
|
.withAppName("app")
|
||||||
|
@ -906,10 +884,32 @@ public class TestApplicationLimits {
|
||||||
assertEquals(RMAppState.FAILED, app13.getState());
|
assertEquals(RMAppState.FAILED, app13.getState());
|
||||||
assertEquals(
|
assertEquals(
|
||||||
"org.apache.hadoop.security.AccessControlException: Queue"
|
"org.apache.hadoop.security.AccessControlException: Queue"
|
||||||
+ " root.d already has 2 applications from user user cannot"
|
+ " root.d already has 0 applications from user user cannot"
|
||||||
+ " accept submission of application: " + app13.getApplicationId(),
|
+ " accept submission of application: " + app13.getApplicationId(),
|
||||||
app13.getDiagnostics().toString());
|
app13.getDiagnostics().toString());
|
||||||
|
|
||||||
|
RMApp app11 = MockRMAppSubmitter.submit(rm,
|
||||||
|
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
||||||
|
.withAppName("app")
|
||||||
|
.withUser("user2")
|
||||||
|
.withAcls(null)
|
||||||
|
.withQueue("a2")
|
||||||
|
.withWaitForAppAcceptedState(false)
|
||||||
|
.build());
|
||||||
|
rm.drainEvents();
|
||||||
|
rm.waitForState(app11.getApplicationId(), RMAppState.ACCEPTED);
|
||||||
|
assertEquals(RMAppState.ACCEPTED, app11.getState());
|
||||||
|
RMApp app12 = MockRMAppSubmitter.submit(rm,
|
||||||
|
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
||||||
|
.withAppName("app")
|
||||||
|
.withUser("user2")
|
||||||
|
.withAcls(null)
|
||||||
|
.withQueue("a2")
|
||||||
|
.withWaitForAppAcceptedState(false)
|
||||||
|
.build());
|
||||||
|
rm.drainEvents();
|
||||||
|
rm.waitForState(app12.getApplicationId(), RMAppState.ACCEPTED);
|
||||||
|
assertEquals(RMAppState.ACCEPTED, app12.getState());
|
||||||
// based on system max limit application is rejected
|
// based on system max limit application is rejected
|
||||||
RMApp app14 = MockRMAppSubmitter.submit(rm,
|
RMApp app14 = MockRMAppSubmitter.submit(rm,
|
||||||
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
MockRMAppSubmissionData.Builder.createWithMemory(GB, rm)
|
||||||
|
@ -938,7 +938,6 @@ public class TestApplicationLimits {
|
||||||
app15.getDiagnostics().toString());
|
app15.getDiagnostics().toString());
|
||||||
|
|
||||||
rm.killApp(app2.getApplicationId());
|
rm.killApp(app2.getApplicationId());
|
||||||
rm.killApp(app11.getApplicationId());
|
|
||||||
rm.killApp(app13.getApplicationId());
|
rm.killApp(app13.getApplicationId());
|
||||||
rm.killApp(app14.getApplicationId());
|
rm.killApp(app14.getApplicationId());
|
||||||
rm.stop();
|
rm.stop();
|
||||||
|
|
|
@ -5137,6 +5137,13 @@ public class TestLeafQueue {
|
||||||
assertEquals(1.0, leafQueue.getAbsoluteMaximumCapacity(),
|
assertEquals(1.0, leafQueue.getAbsoluteMaximumCapacity(),
|
||||||
EPSILON);
|
EPSILON);
|
||||||
|
|
||||||
|
// limit maximum apps by max system apps
|
||||||
|
csConf.setMaximumSystemApplications(15);
|
||||||
|
leafQueue.updateClusterResource(Resource.newInstance(0, 0),
|
||||||
|
new ResourceLimits(Resource.newInstance(0, 0)));
|
||||||
|
|
||||||
|
assertEquals(15, leafQueue.getMaxApplications());
|
||||||
|
|
||||||
} finally {
|
} finally {
|
||||||
//revert config changes
|
//revert config changes
|
||||||
csConf.setNodeLocalityDelay(
|
csConf.setNodeLocalityDelay(
|
||||||
|
|
|
@ -1137,7 +1137,7 @@ public class TestParentQueue {
|
||||||
assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());
|
assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());
|
||||||
|
|
||||||
// Set GlobalMaximumApplicationsPerQueue in csConf
|
// Set GlobalMaximumApplicationsPerQueue in csConf
|
||||||
csConf.setGlobalMaximumApplicationsPerQueue(20000);
|
csConf.setGlobalMaximumApplicationsPerQueue(8000);
|
||||||
root.updateClusterResource(clusterResource,
|
root.updateClusterResource(clusterResource,
|
||||||
new ResourceLimits(clusterResource));
|
new ResourceLimits(clusterResource));
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue