You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2017/02/28 22:43:56 UTC
[12/12] twill git commit: Guard against YARN returning mismatch
container size case.
Guard against YARN returning mismatch container size case.
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/4b60b4fd
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/4b60b4fd
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/4b60b4fd
Branch: refs/heads/feature/twill-186
Commit: 4b60b4fd0696650482062c1c1d15ba5ba52e1088
Parents: b83095a
Author: Terence Yim <ch...@apache.org>
Authored: Tue Feb 28 14:43:37 2017 -0800
Committer: Terence Yim <ch...@apache.org>
Committed: Tue Feb 28 14:43:37 2017 -0800
----------------------------------------------------------------------
.../internal/yarn/AbstractYarnAMClient.java | 45 +++++++++++---------
1 file changed, 26 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/4b60b4fd/twill-yarn/src/main/java/org/apache/twill/internal/yarn/AbstractYarnAMClient.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/yarn/AbstractYarnAMClient.java b/twill-yarn/src/main/java/org/apache/twill/internal/yarn/AbstractYarnAMClient.java
index 3f291b9..ff62c80 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/yarn/AbstractYarnAMClient.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/yarn/AbstractYarnAMClient.java
@@ -20,6 +20,8 @@ package org.apache.twill.internal.yarn;
import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.LinkedHashMultimap;
+import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
@@ -36,6 +38,7 @@ import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.net.URL;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
import javax.annotation.Nullable;
@@ -50,12 +53,11 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
private static final Logger LOG = LoggerFactory.getLogger(AbstractYarnAMClient.class);
// Map from a unique ID to inflight requests
- private final Multimap<String, T> containerRequests;
-
- // List of requests pending to send through allocate call
- private final List<T> requests;
+ private final Multimap<String, T> inflightRequests;
+ // Map from a unique ID to pending requests. It is for recording
+ private final Multimap<String, T> pendingRequests;
// List of requests pending to remove through allocate call
- private final List<T> removes;
+ private final List<T> pendingRemoves;
//List of pending blacklist additions for the next allocate call
private final List<String> blacklistAdditions;
//List of pending blacklist removals for the next allocate call
@@ -81,9 +83,9 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
Preconditions.checkArgument(masterContainerId != null,
"Missing %s from environment", containerIdEnvName);
this.containerId = ConverterUtils.toContainerId(masterContainerId);
- this.containerRequests = ArrayListMultimap.create();
- this.requests = Lists.newLinkedList();
- this.removes = Lists.newLinkedList();
+ this.inflightRequests = ArrayListMultimap.create();
+ this.pendingRequests = ArrayListMultimap.create();
+ this.pendingRemoves = Lists.newLinkedList();
this.blacklistAdditions = Lists.newArrayList();
this.blacklistRemovals = Lists.newArrayList();
this.blacklistedResources = Lists.newArrayList();
@@ -109,16 +111,17 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
// With bug YARN-314, if we mix the allocate call with new container request of the same priority,
// in some cases the RM would not see the new request (based on sorting of resource capability),
// but rather only see the one with size = 0.
- if (removes.isEmpty()) {
- for (T request : requests) {
- addContainerRequest(request);
+ if (inflightRequests.isEmpty()) {
+ for (Map.Entry<String, T> entry : pendingRequests.entries()) {
+ addContainerRequest(entry.getValue());
}
- requests.clear();
+ inflightRequests.putAll(pendingRequests);
+ pendingRequests.clear();
} else {
- for (T request : removes) {
+ for (T request : pendingRemoves) {
removeContainerRequest(request);
}
- removes.clear();
+ pendingRemoves.clear();
}
if (!blacklistAdditions.isEmpty() || !blacklistRemovals.isEmpty()) {
@@ -131,7 +134,12 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
List<RunnableProcessLauncher> launchers = allocateResponse.getLaunchers();
if (!launchers.isEmpty()) {
- handler.acquired(launchers);
+ // Only call handler acquire if there is actually inflight requests.
+ // This is to workaround the YARN behavior that it can return more containers being asked,
+ // such that it causes us to launch process in the pending requests with the wrong container size
+ if (!inflightRequests.isEmpty()) {
+ handler.acquired(launchers);
+ }
// If no process has been launched through the given launcher, return the container.
for (ProcessLauncher<YarnContainerInfo> l : launchers) {
@@ -165,8 +173,7 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
for (int i = 0; i < count; i++) {
T request = createContainerRequest(priority, capability, hosts, racks, relaxLocality);
- containerRequests.put(id, request);
- requests.add(request);
+ pendingRequests.put(id, request);
}
return id;
@@ -203,8 +210,8 @@ public abstract class AbstractYarnAMClient<T> extends AbstractIdleService implem
@Override
public final synchronized void completeContainerRequest(String id) {
- for (T request : containerRequests.removeAll(id)) {
- removes.add(request);
+ for (T request : inflightRequests.removeAll(id)) {
+ pendingRemoves.add(request);
}
}