Merge -c 1370889 from trunk to branch-2 to fix YARN-12. Fix findbugs warnings in FairScheduler. Contributed by Junping Du.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1370890 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Arun Murthy 2012-08-08 18:54:26 +00:00
parent 221e491f5d
commit 9337abd0b0
3 changed files with 176 additions and 51 deletions

View File

@ -32,3 +32,5 @@ Release 2.1.0-alpha - Unreleased
BUG FIXES BUG FIXES
YARN-12. Fix findbugs warnings in FairScheduler. (Junping Du via acmurthy)

View File

@ -129,15 +129,17 @@ class FairSchedulerEventLog {
/** /**
* Flush and close the log. * Flush and close the log.
*/ */
void shutdown() { synchronized void shutdown() {
try { try {
if (appender != null) if (appender != null)
appender.close(); appender.close();
} catch (Exception e) {} } catch (Exception e) {
logDisabled = true; LOG.error("Failed to close fair scheduler event log", e);
logDisabled = true;
}
} }
boolean isEnabled() { synchronized boolean isEnabled() {
return !logDisabled; return !logDisabled;
} }
} }

View File

@ -116,7 +116,37 @@ public class QueueManager {
private long lastReloadAttempt; // Last time we tried to reload the queues file private long lastReloadAttempt; // Last time we tried to reload the queues file
private long lastSuccessfulReload; // Last time we successfully reloaded queues private long lastSuccessfulReload; // Last time we successfully reloaded queues
private boolean lastReloadAttemptFailed = false; private boolean lastReloadAttemptFailed = false;
// Monitor object for minQueueResources
private Object minQueueResourcesMO = new Object();
//Monitor object for maxQueueResources
private Object maxQueueResourcesMO = new Object();
//Monitor object for queueMaxApps
private Object queueMaxAppsMO = new Object();
//Monitor object for userMaxApps
private Object userMaxAppsMO = new Object();
//Monitor object for queueWeights
private Object queueWeightsMO = new Object();
//Monitor object for minSharePreemptionTimeouts
private Object minSharePreemptionTimeoutsMO = new Object();
//Monitor object for queueAcls
private Object queueAclsMO = new Object();
//Monitor object for userMaxAppsDefault
private Object userMaxAppsDefaultMO = new Object();
//Monitor object for queueMaxAppsDefault
private Object queueMaxAppsDefaultMO = new Object();
//Monitor object for defaultSchedulingMode
private Object defaultSchedulingModeMO = new Object();
public QueueManager(FairScheduler scheduler) { public QueueManager(FairScheduler scheduler) {
this.scheduler = scheduler; this.scheduler = scheduler;
} }
@ -145,21 +175,27 @@ public class QueueManager {
/** /**
* Get a queue by name, creating it if necessary * Get a queue by name, creating it if necessary
*/ */
public synchronized FSQueue getQueue(String name) { public FSQueue getQueue(String name) {
FSQueue queue = queues.get(name); synchronized (queues) {
if (queue == null) { FSQueue queue = queues.get(name);
queue = new FSQueue(scheduler, name); if (queue == null) {
queue.setSchedulingMode(defaultSchedulingMode); queue = new FSQueue(scheduler, name);
queues.put(name, queue); synchronized (defaultSchedulingModeMO){
queue.setSchedulingMode(defaultSchedulingMode);
}
queues.put(name, queue);
}
return queue;
} }
return queue;
} }
/** /**
* Return whether a queue exists already. * Return whether a queue exists already.
*/ */
public synchronized boolean exists(String name) { public boolean exists(String name) {
return queues.containsKey(name); synchronized (queues) {
return queues.containsKey(name);
}
} }
/** /**
@ -353,16 +389,16 @@ public class QueueManager {
// Commit the reload; also create any queue defined in the alloc file // Commit the reload; also create any queue defined in the alloc file
// if it does not already exist, so it can be displayed on the web UI. // if it does not already exist, so it can be displayed on the web UI.
synchronized(this) { synchronized(this) {
this.minQueueResources = minQueueResources; setMinResources(minQueueResources);
this.maxQueueResources = maxQueueResources; setMaxResources(maxQueueResources);
this.queueMaxApps = queueMaxApps; setQueueMaxApps(queueMaxApps);
this.userMaxApps = userMaxApps; setUserMaxApps(userMaxApps);
this.queueWeights = queueWeights; setQueueWeights(queueWeights);
this.userMaxAppsDefault = userMaxAppsDefault; setUserMaxAppsDefault(userMaxAppsDefault);
this.queueMaxAppsDefault = queueMaxAppsDefault; setQueueMaxAppsDefault(queueMaxAppsDefault);
this.defaultSchedulingMode = defaultSchedulingMode; setDefaultSchedulingMode(defaultSchedulingMode);
this.minSharePreemptionTimeouts = minSharePreemptionTimeouts; setMinSharePreemptionTimeouts(minSharePreemptionTimeouts);
this.queueAcls = queueAcls; setQueueAcls(queueAcls);
for (String name: queueNamesInAllocFile) { for (String name: queueNamesInAllocFile) {
FSQueue queue = getQueue(name); FSQueue queue = getQueue(name);
if (queueModes.containsKey(name)) { if (queueModes.containsKey(name)) {
@ -392,25 +428,40 @@ public class QueueManager {
* @return the cap set on this queue, or 0 if not set. * @return the cap set on this queue, or 0 if not set.
*/ */
public Resource getMinResources(String queue) { public Resource getMinResources(String queue) {
if (minQueueResources.containsKey(queue)) { synchronized(minQueueResourcesMO) {
return minQueueResources.get(queue); if (minQueueResources.containsKey(queue)) {
} else{ return minQueueResources.get(queue);
return Resources.createResource(0); } else{
return Resources.createResource(0);
}
} }
} }
private void setMinResources(Map<String, Resource> resources) {
synchronized(minQueueResourcesMO) {
minQueueResources = resources;
}
}
/** /**
* Get the maximum resource allocation for the given queue. * Get the maximum resource allocation for the given queue.
* @return the cap set on this queue, or Integer.MAX_VALUE if not set. * @return the cap set on this queue, or Integer.MAX_VALUE if not set.
*/ */
Resource getMaxResources(String queueName) { Resource getMaxResources(String queueName) {
if (maxQueueResources.containsKey(queueName)) { synchronized (maxQueueResourcesMO) {
return maxQueueResources.get(queueName); if (maxQueueResources.containsKey(queueName)) {
} else { return maxQueueResources.get(queueName);
return Resources.createResource(Integer.MAX_VALUE); } else {
return Resources.createResource(Integer.MAX_VALUE);
}
} }
} }
private void setMaxResources(Map<String, Resource> resources) {
synchronized(maxQueueResourcesMO) {
maxQueueResources = resources;
}
}
/** /**
* Add an app in the appropriate queue * Add an app in the appropriate queue
*/ */
@ -428,11 +479,12 @@ public class QueueManager {
/** /**
* Get a collection of all queues * Get a collection of all queues
*/ */
public synchronized Collection<FSQueue> getQueues() { public Collection<FSQueue> getQueues() {
return queues.values(); synchronized (queues) {
return queues.values();
}
} }
/** /**
* Get all queue names that have been seen either in the allocation file or in * Get all queue names that have been seen either in the allocation file or in
* a submitted app. * a submitted app.
@ -447,40 +499,102 @@ public class QueueManager {
} }
public int getUserMaxApps(String user) { public int getUserMaxApps(String user) {
if (userMaxApps.containsKey(user)) { synchronized (userMaxAppsMO) {
return userMaxApps.get(user); if (userMaxApps.containsKey(user)) {
} else { return userMaxApps.get(user);
return userMaxAppsDefault; } else {
return getUserMaxAppsDefault();
}
} }
} }
private void setUserMaxApps(Map<String, Integer> userApps) {
synchronized (userMaxAppsMO) {
userMaxApps = userApps;
}
}
private int getUserMaxAppsDefault() {
synchronized (userMaxAppsDefaultMO){
return userMaxAppsDefault;
}
}
private void setUserMaxAppsDefault(int userMaxApps) {
synchronized (userMaxAppsDefaultMO){
userMaxAppsDefault = userMaxApps;
}
}
public int getQueueMaxApps(String queue) { public int getQueueMaxApps(String queue) {
if (queueMaxApps.containsKey(queue)) { synchronized (queueMaxAppsMO) {
return queueMaxApps.get(queue); if (queueMaxApps.containsKey(queue)) {
} else { return queueMaxApps.get(queue);
} else {
return getQueueMaxAppsDefault();
}
}
}
private void setQueueMaxApps(Map<String, Integer> queueApps) {
synchronized (queueMaxAppsMO) {
queueMaxApps = queueApps;
}
}
private int getQueueMaxAppsDefault(){
synchronized(queueMaxAppsDefaultMO) {
return queueMaxAppsDefault; return queueMaxAppsDefault;
} }
} }
private void setQueueMaxAppsDefault(int queueMaxApps){
synchronized(queueMaxAppsDefaultMO) {
queueMaxAppsDefault = queueMaxApps;
}
}
private void setDefaultSchedulingMode(SchedulingMode schedulingMode){
synchronized(defaultSchedulingModeMO) {
defaultSchedulingMode = schedulingMode;
}
}
public double getQueueWeight(String queue) { public double getQueueWeight(String queue) {
if (queueWeights.containsKey(queue)) { synchronized (queueWeightsMO) {
return queueWeights.get(queue); if (queueWeights.containsKey(queue)) {
} else { return queueWeights.get(queue);
return 1.0; } else {
return 1.0;
}
} }
} }
private void setQueueWeights(Map<String, Double> weights) {
synchronized (queueWeightsMO) {
queueWeights = weights;
}
}
/** /**
* Get a queue's min share preemption timeout, in milliseconds. This is the * Get a queue's min share preemption timeout, in milliseconds. This is the
* time after which jobs in the queue may kill other queues' tasks if they * time after which jobs in the queue may kill other queues' tasks if they
* are below their min share. * are below their min share.
*/ */
public long getMinSharePreemptionTimeout(String queueName) { public long getMinSharePreemptionTimeout(String queueName) {
if (minSharePreemptionTimeouts.containsKey(queueName)) { synchronized (minSharePreemptionTimeoutsMO) {
return minSharePreemptionTimeouts.get(queueName); if (minSharePreemptionTimeouts.containsKey(queueName)) {
return minSharePreemptionTimeouts.get(queueName);
}
} }
return defaultMinSharePreemptionTimeout; return defaultMinSharePreemptionTimeout;
} }
private void setMinSharePreemptionTimeouts(
Map<String, Long> sharePreemptionTimeouts){
synchronized (minSharePreemptionTimeoutsMO) {
minSharePreemptionTimeouts = sharePreemptionTimeouts;
}
}
/** /**
* Get the fair share preemption, in milliseconds. This is the time * Get the fair share preemption, in milliseconds. This is the time
@ -497,9 +611,10 @@ public class QueueManager {
*/ */
public Map<QueueACL, AccessControlList> getQueueAcls(String queue) { public Map<QueueACL, AccessControlList> getQueueAcls(String queue) {
HashMap<QueueACL, AccessControlList> out = new HashMap<QueueACL, AccessControlList>(); HashMap<QueueACL, AccessControlList> out = new HashMap<QueueACL, AccessControlList>();
synchronized (queueAclsMO) {
if (queueAcls.containsKey(queue)) { if (queueAcls.containsKey(queue)) {
out.putAll(queueAcls.get(queue)); out.putAll(queueAcls.get(queue));
}
} }
if (!out.containsKey(QueueACL.ADMINISTER_QUEUE)) { if (!out.containsKey(QueueACL.ADMINISTER_QUEUE)) {
out.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList("*")); out.put(QueueACL.ADMINISTER_QUEUE, new AccessControlList("*"));
@ -509,4 +624,10 @@ public class QueueManager {
} }
return out; return out;
} }
private void setQueueAcls(Map<String, Map<QueueACL, AccessControlList>> queue) {
synchronized (queueAclsMO) {
queueAcls = queue;
}
}
} }