You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by hs...@apache.org on 2017/02/02 02:15:57 UTC
twill git commit: TWILL-211,
use ALLOCATE_ONE_INSTANCE_AT_A_TIME for retries to prevent poll
starvation
Repository: twill
Updated Branches:
refs/heads/master 3fed2c09e -> 3bdc5dab4
TWILL-211, use ALLOCATE_ONE_INSTANCE_AT_A_TIME for retries to prevent poll starvation
This closes #29 from Github
Signed-off-by: Henry Saputra <hs...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/twill/commit/3bdc5dab
Tree: http://git-wip-us.apache.org/repos/asf/twill/tree/3bdc5dab
Diff: http://git-wip-us.apache.org/repos/asf/twill/diff/3bdc5dab
Branch: refs/heads/master
Commit: 3bdc5dab4d88ff9437f4bce0b360bc64420d922d
Parents: 3fed2c0
Author: martin <ma...@attivio.com>
Authored: Sun Jan 29 15:28:18 2017 -0500
Committer: Henry Saputra <hs...@apache.org>
Committed: Wed Feb 1 17:55:26 2017 -0800
----------------------------------------------------------------------
.../appmaster/ApplicationMasterService.java | 20 ++++++++++++++++++--
1 file changed, 18 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/twill/blob/3bdc5dab/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
index 4aa3800..985fad1 100644
--- a/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
+++ b/twill-yarn/src/main/java/org/apache/twill/internal/appmaster/ApplicationMasterService.java
@@ -38,6 +38,7 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
+
import org.apache.hadoop.io.Text;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
@@ -103,6 +104,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+
import javax.annotation.Nullable;
/**
@@ -464,7 +466,8 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
//Check the placement policy
TwillSpecification.PlacementPolicy placementPolicy =
placementPolicyManager.getPlacementPolicy(currentAllocationSpecification.getRunnableName());
- if (placementPolicy.getType().equals(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
+ if (placementPolicy != null
+ && placementPolicy.getType().equals(TwillSpecification.PlacementPolicy.Type.DISTRIBUTED)) {
//Update blacklist with hosts which are running DISTRIBUTED runnables
for (String runnable : placementPolicyManager.getFellowRunnables(request.getKey().getRunnableName())) {
@@ -692,6 +695,11 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
amClient.completeContainerRequest(provisionRequest.getRequestId());
}
+ /*
+ * The provisionRequest will either contain a single container (ALLOCATE_ONE_INSTANCE_AT_A_TIME), or all the
+ * containers to satisfy the expectedContainers count. In the later case, the provision request is complete once
+ * all the containers have run at which point we poll() to remove the provisioning request.
+ */
if (expectedContainers.getExpected(runnableName) == runningContainers.count(runnableName) ||
provisioning.peek().getType().equals(AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME)) {
provisioning.poll();
@@ -844,7 +852,15 @@ public final class ApplicationMasterService extends AbstractYarnTwillService imp
addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
}
} else {
- AllocationSpecification allocationSpecification = new AllocationSpecification(capability);
+ AllocationSpecification allocationSpecification;
+ if (numberOfInstances > 1) {
+ allocationSpecification = new AllocationSpecification(capability);
+ } else {
+ // for a single instance, we always insert ALLOCATE_ONE_INSTANCE_AT_A_TIME. for multi-instance
+ // runnables, this case occurs during retries.
+ allocationSpecification = new AllocationSpecification(capability,
+ AllocationSpecification.Type.ALLOCATE_ONE_INSTANCE_AT_A_TIME, runnableName, 0);
+ }
addAllocationSpecification(allocationSpecification, requestsMap, runtimeSpec);
}
return new RunnableContainerRequest(order.getType(), requestsMap, isProvisioned);