Skip to content
This repository has been archived by the owner on Jul 22, 2022. It is now read-only.

Commit

Permalink
YARN-8990. Fix fair scheduler race condition in app submit and queue …
Browse files Browse the repository at this point in the history 
…cleanup. (Contributed by Wilfred Spiegelenburg)

(cherry picked from commit 524a752)
  • Loading branch information
@haibchen @sjrand
haibchen authored and sjrand committed Mar 21, 2021
1 parent 1f0af9a commit 5f0c2b0
Show file tree
Hide file tree
Showing 3 changed files with 108 additions and 46 deletions.
14 changes: 14 additions & 0 deletions ...c/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -666,4 +666,18 @@ public void addAssignedApp(ApplicationId applicationId) {
writeLock.unlock();
}
}

/**
* This method is called when an application is removed from this queue
* during the submit process.
* @param applicationId the application's id
*/
public void removeAssignedApp(ApplicationId applicationId) {
writeLock.lock();
try {
assignedApps.remove(applicationId);
} finally {
writeLock.unlock();
}
}
}
27 changes: 18 additions & 9 deletions ...main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -479,7 +479,7 @@ protected void addApplication(ApplicationId applicationId,
RMApp rmApp = rmContext.getRMApps().get(applicationId);
// This will re-create the queue on restore, however this could fail if
// the config was changed.
FSLeafQueue queue = assignToQueue(rmApp, queueName, user);
FSLeafQueue queue = assignToQueue(rmApp, queueName, user, applicationId);
if (queue == null) {
if (!isAppRecovering) {
return;
Expand Down Expand Up @@ -516,6 +516,7 @@ protected void addApplication(ApplicationId applicationId,
applicationId, queue.getName(),
invalidAMResourceRequests, queue.getMaxShare());
rejectApplicationWithMessage(applicationId, msg);
queue.removeAssignedApp(applicationId);
return;
}
}
Expand All @@ -525,14 +526,15 @@ protected void addApplication(ApplicationId applicationId,
if (!isAppRecovering) {
// Enforce ACLs
UserGroupInformation userUgi = UserGroupInformation.createRemoteUser(
user);
user);

if (!queue.hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi) && !queue
.hasAccess(QueueACL.ADMINISTER_QUEUE, userUgi)) {
.hasAccess(QueueACL.ADMINISTER_QUEUE, userUgi)) {
String msg = "User " + userUgi.getUserName()
+ " cannot submit applications to queue " + queue.getName()
+ "(requested queuename is " + queueName + ")";
+ " cannot submit applications to queue " + queue.getName()
+ "(requested queuename is " + queueName + ")";
rejectApplicationWithMessage(applicationId, msg);
queue.removeAssignedApp(applicationId);
return;
}
}
Expand All @@ -541,7 +543,6 @@ protected void addApplication(ApplicationId applicationId,
new SchedulerApplication<FSAppAttempt>(queue, user);
applications.put(applicationId, application);
queue.getMetrics().submitApp(user);
queue.addAssignedApp(applicationId);

LOG.info("Accepted application " + applicationId + " from user: " + user
+ ", in queue: " + queue.getName()
Expand Down Expand Up @@ -618,11 +619,19 @@ protected void addApplicationAttempt(
}

/**
* Helper method that attempts to assign the app to a queue. The method is
* responsible to call the appropriate event-handler if the app is rejected.
* Helper method for the tests to assign the app to a queue.
*/
@VisibleForTesting
FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) {
return assignToQueue(rmApp, queueName, user, null);
}

/**
* Helper method that attempts to assign the app to a queue. The method is
* responsible to call the appropriate event-handler if the app is rejected.
*/
private FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user,
ApplicationId applicationId) {
FSLeafQueue queue = null;
String appRejectMsg = null;

Expand All @@ -632,7 +641,7 @@ FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) {
if (queueName == null) {
appRejectMsg = "Application rejected by queue placement policy";
} else {
queue = queueMgr.getLeafQueue(queueName, true);
queue = queueMgr.getLeafQueue(queueName, true, applicationId);
if (queue == null) {
appRejectMsg = queueName + " is not a leaf queue";
}
Expand Down
113 changes: 76 additions & 37 deletions .../main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
import org.xml.sax.SAXException;
Expand Down Expand Up @@ -71,7 +72,7 @@ private void execute() {
Boolean removed =
removeEmptyIncompatibleQueues(queueToCreate, queueType).orElse(null);
if (Boolean.TRUE.equals(removed)) {
FSQueue queue = getQueue(queueToCreate, true, queueType, false);
FSQueue queue = getQueue(queueToCreate, true, queueType, false, null);
if (queue != null &&
// if queueToCreate is present in the allocation config, set it
// to static
Expand Down Expand Up @@ -124,30 +125,49 @@ public void initialize(Configuration conf) throws IOException,

/**
* Get a leaf queue by name, creating it if the create param is
* true and is necessary.
* If the queue is not or can not be a leaf queue, i.e. it already exists as a
* parent queue, or one of the parents in its name is already a leaf queue,
* null is returned.
* <code>true</code> and the queue does not exist.
* If the queue is not or can not be a leaf queue, i.e. it already exists as
* a parent queue, or one of the parents in its name is already a leaf queue,
* <code>null</code> is returned.
*
* The root part of the name is optional, so a queue underneath the root
* named "queue1" could be referred to as just "queue1", and a queue named
* "queue2" underneath a parent named "parent1" that is underneath the root
* could be referred to as just "parent1.queue2".
* @param name name of the queue
* @param create <code>true</code> if the queue must be created if it does
* not exist, <code>false</code> otherwise
* @return the leaf queue or <code>null</code> if the queue cannot be found
*/
public FSLeafQueue getLeafQueue(String name, boolean create) {
return getLeafQueue(name, create, true);
return getLeafQueue(name, create, null, true);
}

private FSLeafQueue getLeafQueue(
String name,
boolean create,
boolean recomputeSteadyShares) {
FSQueue queue = getQueue(
name,
create,
FSQueueType.LEAF,
recomputeSteadyShares
);
/**
* Get a leaf queue by name, creating it if the create param is
* <code>true</code> and the queue does not exist.
* If the queue is not or can not be a leaf queue, i.e. it already exists as
* a parent queue, or one of the parents in its name is already a leaf queue,
* <code>null</code> is returned.
*
* If the application will be assigned to the queue if the applicationId is
* not <code>null</code>
* @param name name of the queue
* @param create <code>true</code> if the queue must be created if it does
* not exist, <code>false</code> otherwise
* @param applicationId the application ID to assign to the queue
* @return the leaf queue or <code>null</code> if teh queue cannot be found
*/
public FSLeafQueue getLeafQueue(String name, boolean create,
ApplicationId applicationId) {
return getLeafQueue(name, create, applicationId, true);
}

private FSLeafQueue getLeafQueue(String name, boolean create,
ApplicationId applicationId,
boolean recomputeSteadyShares) {
FSQueue queue = getQueue(name, create, FSQueueType.LEAF,
recomputeSteadyShares, applicationId);
if (queue instanceof FSParentQueue) {
return null;
}
Expand All @@ -168,42 +188,55 @@ public boolean removeLeafQueue(String name) {

/**
* Get a parent queue by name, creating it if the create param is
* true and is necessary.
* If the queue is not or can not be a parent queue,
* i.e. it already exists as a
* leaf queue, or one of the parents in its name is already a leaf queue,
* null is returned.
* <code>true</code> and the queue does not exist.
* If the queue is not or can not be a parent queue, i.e. it already exists
* as a leaf queue, or one of the parents in its name is already a leaf
* queue, <code>null</code> is returned.
*
* The root part of the name is optional, so a queue underneath the root
* named "queue1" could be referred to as just "queue1", and a queue named
* "queue2" underneath a parent named "parent1" that is underneath the root
* could be referred to as just "parent1.queue2".
* @param name name of the queue
* @param create <code>true</code> if the queue must be created if it does
* not exist, <code>false</code> otherwise
* @return the parent queue or <code>null</code> if the queue cannot be found
*/
public FSParentQueue getParentQueue(String name, boolean create) {
return getParentQueue(name, create, true);
}

public FSParentQueue getParentQueue(
String name,
boolean create,
/**
* Get a parent queue by name, creating it if the create param is
* <code>true</code> and the queue does not exist.
* If the queue is not or can not be a parent queue, i.e. it already exists
* as a leaf queue, or one of the parents in its name is already a leaf
* queue, <code>null</code> is returned.
*
* The root part of the name is optional, so a queue underneath the root
* named "queue1" could be referred to as just "queue1", and a queue named
* "queue2" underneath a parent named "parent1" that is underneath the root
* could be referred to as just "parent1.queue2".
* @param name name of the queue
* @param create <code>true</code> if the queue must be created if it does
* not exist, <code>false</code> otherwise
* @param recomputeSteadyShares <code>true</code> if the steady fair share
* should be recalculated when a queue is added,
* <code>false</code> otherwise
* @return the parent queue or <code>null</code> if the queue cannot be found
*/
public FSParentQueue getParentQueue(String name, boolean create,
boolean recomputeSteadyShares) {
FSQueue queue = getQueue(
name,
create,
FSQueueType.PARENT,
recomputeSteadyShares
);
FSQueue queue = getQueue(name, create, FSQueueType.PARENT,
recomputeSteadyShares, null);
if (queue instanceof FSLeafQueue) {
return null;
}
return (FSParentQueue) queue;
}

private FSQueue getQueue(
String name,
boolean create,
FSQueueType queueType,
boolean recomputeSteadyShares) {
private FSQueue getQueue(String name, boolean create, FSQueueType queueType,
boolean recomputeSteadyShares, ApplicationId applicationId) {
boolean recompute = recomputeSteadyShares;
name = ensureRootPrefix(name);
FSQueue queue;
Expand All @@ -215,8 +248,14 @@ private FSQueue getQueue(
} else {
recompute = false;
}
// At this point the queue exists and we need to assign the app if to the
// but only to a leaf queue
if (applicationId != null && queue instanceof FSLeafQueue) {
((FSLeafQueue)queue).addAssignedApp(applicationId);
}
}
if (recompute) {
// Don't recompute if it is an existing queue or no change was made
if (recompute && queue != null) {
rootQueue.recomputeSteadyShares();
}
return queue;
Expand Down Expand Up @@ -594,7 +633,7 @@ private void ensureQueueExistsAndIsCompatibleAndIsStatic(
incompatibleQueuesPendingRemoval.add(
new IncompatibleQueueRemovalTask(name, queueType));
} else {
FSQueue queue = getQueue(name, true, queueType, false);
FSQueue queue = getQueue(name, true, queueType, false, null);
if (queue != null) {
queue.setDynamic(false);
}
Expand Down

0 comments on commit 5f0c2b0

Please sign in to comment.