YARN-6171. ConcurrentModificationException on FSAppAttempt.containersToPreempt. (Miklos Szegedi via kasha)

This commit is contained in:
Karthik Kambatla 2017-02-16 14:54:51 -08:00
parent 5d339c46f5
commit a77f432449
2 changed files with 34 additions and 30 deletions

View File

@ -83,8 +83,10 @@ public class FSAppAttempt extends SchedulerApplicationAttempt
private Resource fairShare = Resources.createResource(0, 0);
// Preemption related variables
private final Object preemptionVariablesLock = new Object();
private final Resource preemptedResources = Resources.clone(Resources.none());
private final Set<RMContainer> containersToPreempt = new HashSet<>();
private Resource fairshareStarvation = Resources.none();
private long lastTimeAtFairShare;
private long nextStarvationCheck;
@ -552,29 +554,29 @@ void resetMinshareStarvation() {
}
void trackContainerForPreemption(RMContainer container) {
if (containersToPreempt.add(container)) {
synchronized (preemptedResources) {
synchronized (preemptionVariablesLock) {
if (containersToPreempt.add(container)) {
Resources.addTo(preemptedResources, container.getAllocatedResource());
}
}
}
private void untrackContainerForPreemption(RMContainer container) {
if (containersToPreempt.remove(container)) {
synchronized (preemptedResources) {
synchronized (preemptionVariablesLock) {
if (containersToPreempt.remove(container)) {
Resources.subtractFrom(preemptedResources,
container.getAllocatedResource());
}
}
}
Set<RMContainer> getPreemptionContainers() {
return containersToPreempt;
}
private Resource getPreemptedResources() {
synchronized (preemptedResources) {
return preemptedResources;
Set<ContainerId> getPreemptionContainerIds() {
synchronized (preemptionVariablesLock) {
Set<ContainerId> preemptionContainerIds = new HashSet<>();
for (RMContainer container : containersToPreempt) {
preemptionContainerIds.add(container.getContainerId());
}
return preemptionContainerIds;
}
}
@ -591,9 +593,11 @@ boolean canContainerBePreempted(RMContainer container) {
return false;
}
if (containersToPreempt.contains(container)) {
// The container is already under consideration for preemption
return false;
synchronized (preemptionVariablesLock) {
if (containersToPreempt.contains(container)) {
// The container is already under consideration for preemption
return false;
}
}
// Check if the app's allocation will be over its fairshare even
@ -969,7 +973,8 @@ private Resource assignContainer(FSSchedulerNode node, boolean reserved) {
if (LOG.isTraceEnabled()) {
LOG.trace("Assign container on " + node.getNodeName()
+ " node, assignType: OFF_SWITCH" + ", allowedLocality: "
+ allowedLocality + ", priority: " + schedulerKey.getPriority()
+ allowedLocality + ", priority: "
+ schedulerKey.getPriority()
+ ", app attempt id: " + this.attemptId);
}
return assignContainer(node, offswitchAsk, NodeType.OFF_SWITCH,
@ -1226,13 +1231,13 @@ public Resource getMaxShare() {
@Override
public Resource getResourceUsage() {
/*
* getResourcesToPreempt() returns zero, except when there are containers
* to preempt. Avoid creating an object in the common case.
*/
return getPreemptedResources().equals(Resources.none())
? getCurrentConsumption()
: Resources.subtract(getCurrentConsumption(), getPreemptedResources());
// Subtract copies the object, so that we have a snapshot,
// in case usage changes, while the caller is using the value
synchronized (preemptionVariablesLock) {
return containersToPreempt.isEmpty()
? getCurrentConsumption()
: Resources.subtract(getCurrentConsumption(), preemptedResources);
}
}
@Override

View File

@ -103,6 +103,7 @@
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/**
* A scheduler that schedules resources between a set of queues. The scheduler
@ -831,8 +832,9 @@ public Allocation allocate(ApplicationAttemptId appAttemptId,
// Release containers
releaseContainers(release, application);
ReentrantReadWriteLock.WriteLock lock = application.getWriteLock();
lock.lock();
try {
application.getWriteLock().lock();
if (!ask.isEmpty()) {
if (LOG.isDebugEnabled()) {
LOG.debug(
@ -847,24 +849,21 @@ public Allocation allocate(ApplicationAttemptId appAttemptId,
application.showRequests();
}
} finally {
application.getWriteLock().unlock();
lock.unlock();
}
Set<ContainerId> preemptionContainerIds =
application.getPreemptionContainerIds();
if (LOG.isDebugEnabled()) {
LOG.debug(
"allocate: post-update" + " applicationAttemptId=" + appAttemptId
+ " #ask=" + ask.size() + " reservation= " + application
.getCurrentReservation());
LOG.debug("Preempting " + application.getPreemptionContainers().size()
LOG.debug("Preempting " + preemptionContainerIds.size()
+ " container(s)");
}
Set<ContainerId> preemptionContainerIds = new HashSet<ContainerId>();
for (RMContainer container : application.getPreemptionContainers()) {
preemptionContainerIds.add(container.getContainerId());
}
application.updateBlacklist(blacklistAdditions, blacklistRemovals);
List<Container> newlyAllocatedContainers =