You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2014/08/26 09:26:12 UTC
git commit: TEZ-1486. Event routing should not throw an exception if
the EdgePlugin does not generate a routing table in cases where the
destination vertex has a parallelism of 0. (sseth) (cherry picked from commit
fc86dbf0a61fa746551a1b4f358eab6f414a77f
Repository: tez
Updated Branches:
refs/heads/branch-0.5 4178b887f -> 0a62c5edc
TEZ-1486. Event routing should not throw an exception if the EdgePlugin
does not generate a routing table in cases where the destination
vertex has a parallelism of 0. (sseth)
(cherry picked from commit fc86dbf0a61fa746551a1b4f358eab6f414a77f7)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/0a62c5ed
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/0a62c5ed
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/0a62c5ed
Branch: refs/heads/branch-0.5
Commit: 0a62c5edccf1b19a8915a818fc7183c598caf5f5
Parents: 4178b88
Author: Siddharth Seth <ss...@apache.org>
Authored: Tue Aug 26 00:24:40 2014 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Tue Aug 26 00:25:54 2014 -0700
----------------------------------------------------------------------
.../apache/tez/dag/api/EntityDescriptor.java | 7 +
.../apache/tez/dag/api/RootInputLeafOutput.java | 6 +
.../apache/tez/runtime/api/InputSpecUpdate.java | 5 +
.../org/apache/tez/dag/app/dag/impl/Edge.java | 51 +++++--
.../tez/dag/app/dag/impl/TestVertexImpl.java | 150 +++++++++++++++++++
5 files changed, 206 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/0a62c5ed/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java b/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
index 1ad004f..2caf819 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/EntityDescriptor.java
@@ -123,4 +123,11 @@ public abstract class EntityDescriptor<T extends EntityDescriptor<T>> implements
this.userPayload = DagTypeConverters.convertToTezUserPayload(ByteBuffer.wrap(bb), version);
}
}
+
+ @Override
+ public String toString() {
+ boolean hasPayload =
+ userPayload == null ? false : userPayload.getPayload() == null ? false : true;
+ return "ClassName=" + className + ", hasPayload=" + hasPayload;
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a62c5ed/tez-api/src/main/java/org/apache/tez/dag/api/RootInputLeafOutput.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/RootInputLeafOutput.java b/tez-api/src/main/java/org/apache/tez/dag/api/RootInputLeafOutput.java
index 4449ca3..953a78d 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/RootInputLeafOutput.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/RootInputLeafOutput.java
@@ -45,4 +45,10 @@ public class RootInputLeafOutput <T extends EntityDescriptor<T>, S extends Entit
return this.controllerDescriptor;
}
+ @Override
+ public String toString() {
+ return "{InputName=" + name + "}, {Descriptor=" + ioDescriptor + "}, {ControllerDescriptor=" +
+ controllerDescriptor + "}";
+ }
+
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a62c5ed/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java b/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
index b81f15c..c6bbba7 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/InputSpecUpdate.java
@@ -102,4 +102,9 @@ public class InputSpecUpdate {
public List<Integer> getAllNumPhysicalInputs() {
return numPhysicalInputs;
}
+
+ @Override
+ public String toString() {
+ return "forAllWorkUnits=" + forAllWorkUnits + ", update=" + numPhysicalInputs.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a62c5ed/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
index b6a3bc1..47937bc 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
@@ -23,6 +23,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.tez.common.ReflectionUtils;
import org.apache.tez.dag.api.EdgeManagerPlugin;
@@ -54,6 +56,8 @@ import com.google.common.collect.Maps;
public class Edge {
+ private static final Log LOG = LogFactory.getLog(Edge.class);
+
class EdgeManagerPluginContextImpl implements EdgeManagerPluginContext {
private final UserPayload userPayload;
@@ -356,25 +360,38 @@ public class Edge {
TezTaskAttemptID srcAttemptId = tezEvent.getSourceInfo()
.getTaskAttemptID();
int srcTaskIndex = srcAttemptId.getTaskID().getId();
- if (isDataMovementEvent) {
- DataMovementEvent dmEvent = (DataMovementEvent)tezEvent.getEvent();
- edgeManager.routeDataMovementEventToDestination(dmEvent,
+
+ boolean routingRequired = true;
+ if (edgeManagerContext.getDestinationVertexNumTasks() == 0) {
+ routingRequired = false;
+ LOG.info("Not routing events since destination vertex has 0 tasks" +
+ generateCommonDebugString(srcTaskIndex, tezEvent));
+ } else if (edgeManagerContext.getDestinationVertexNumTasks() < 0) {
+ throw new TezUncheckedException(
+ "Internal error. Not expected to route events to a destination until parallelism is determined" +
+ generateCommonDebugString(srcTaskIndex, tezEvent));
+ }
+
+ if (routingRequired) {
+ if (isDataMovementEvent) {
+ DataMovementEvent dmEvent = (DataMovementEvent) tezEvent.getEvent();
+ edgeManager.routeDataMovementEventToDestination(dmEvent,
srcTaskIndex, dmEvent.getSourceIndex(),
destTaskAndInputIndices);
- } else {
- edgeManager.routeInputSourceTaskFailedEventToDestination(srcTaskIndex,
- destTaskAndInputIndices);
+ } else {
+ edgeManager.routeInputSourceTaskFailedEventToDestination(srcTaskIndex,
+ destTaskAndInputIndices);
+ }
}
+
if (!destTaskAndInputIndices.isEmpty()) {
- sendDmEventOrIfEventToTasks(tezEvent, srcTaskIndex, isDataMovementEvent, destTaskAndInputIndices);
- } else {
+ sendDmEventOrIfEventToTasks(tezEvent, srcTaskIndex, isDataMovementEvent,
+ destTaskAndInputIndices);
+ } else if (routingRequired) {
throw new TezUncheckedException("Event must be routed." +
- " sourceVertex=" + sourceVertex.getVertexId() +
- " srcIndex = " + srcTaskIndex +
- " destAttemptId=" + destinationVertex.getVertexId() +
- " edgeManager=" + edgeManager.getClass().getName() +
- " Event type=" + tezEvent.getEventType());
+ generateCommonDebugString(srcTaskIndex, tezEvent));
}
+
break;
default:
throw new TezUncheckedException("Unhandled tez event type: "
@@ -402,4 +419,12 @@ public class Edge {
return this.destinationVertex.getName();
}
+ private String generateCommonDebugString(int srcTaskIndex, TezEvent tezEvent) {
+ return new StringBuilder()
+ .append(" sourceVertex=").append(sourceVertex.getVertexId())
+ .append(" srcIndex = ").append(srcTaskIndex)
+ .append(" destAttemptId=").append(destinationVertex.getVertexId())
+ .append(" edgeManager=").append(edgeManager.getClass().getName())
+ .append(" Event type=").append(tezEvent.getEventType()).toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a62c5ed/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
index 5e74c84..d4243ba 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
@@ -105,6 +105,7 @@ import org.apache.tez.dag.app.dag.DAG;
import org.apache.tez.dag.app.dag.RootInputInitializerManager;
import org.apache.tez.dag.app.dag.Task;
import org.apache.tez.dag.app.dag.TaskAttemptStateInternal;
+import org.apache.tez.dag.app.dag.TaskStateInternal;
import org.apache.tez.dag.app.dag.Vertex;
import org.apache.tez.dag.app.dag.VertexState;
import org.apache.tez.dag.app.dag.VertexTerminationCause;
@@ -426,6 +427,70 @@ public class TestVertexImpl {
return dag.createDag(conf);
}
+ private DAGPlan createDAGPlanWithInitializer0Tasks(String initializerClassName) {
+ LOG.info("Setting up dag plan with input initializer and 0 tasks");
+ DAGPlan dag = DAGPlan.newBuilder()
+ .setName("initializerWith0Tasks")
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex1")
+ .setType(PlanVertexType.NORMAL)
+ .addInputs(
+ RootInputLeafOutputProto.newBuilder()
+ .setControllerDescriptor(
+ TezEntityDescriptorProto.newBuilder().setClassName(
+ initializerClassName))
+ .setName("input1")
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("InputClazz")
+ .build()
+ ).build()
+ )
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(-1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x1.y1")
+ .build()
+ )
+ .addInEdgeId("e1")
+ .build()
+ )
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex2")
+ .setType(PlanVertexType.NORMAL)
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x2.y2")
+ .build()
+ )
+ .addOutEdgeId("e1")
+ .build()
+ )
+ .addEdge(
+ EdgePlan.newBuilder()
+ .setEdgeDestination(TezEntityDescriptorProto.newBuilder().setClassName("v1_v2"))
+ .setInputVertexName("vertex2")
+ .setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
+ .setOutputVertexName("vertex1")
+ .setDataMovementType(PlanEdgeDataMovementType.BROADCAST)
+ .setId("e1")
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
+ .build()
+ )
+ .build();
+ return dag;
+ }
+
private DAGPlan createDAGPlanWithInputInitializer(String initializerClassName) {
LOG.info("Setting up dag plan with input initializer");
DAGPlan dag = DAGPlan.newBuilder()
@@ -2818,6 +2883,50 @@ public class TestVertexImpl {
Assert.assertEquals(true, initializerManager2.hasShutDown);
}
+ @Test(timeout = 10000)
+ public void testVertexWithInitializerParallelismSetTo0() throws InterruptedException {
+ useCustomInitializer = true;
+ customInitializer = new RootInitializerSettingParallelismTo0(null);
+ RootInitializerSettingParallelismTo0 initializer =
+ (RootInitializerSettingParallelismTo0) customInitializer;
+ setupPreDagCreation();
+ dagPlan =
+ createDAGPlanWithInitializer0Tasks(RootInitializerSettingParallelismTo0.class.getName());
+ setupPostDagCreation();
+
+ VertexImpl v1 = vertices.get("vertex1");
+ VertexImpl v2 = vertices.get("vertex2");
+
+ initVertex(v2);
+
+ TezTaskID v2t1 = TezTaskID.getInstance(v2.getVertexId(), 0);
+
+ TezTaskAttemptID ta1V2T1 = TezTaskAttemptID.getInstance(v2t1, 0);
+
+ TezEvent tezEvent = new TezEvent(DataMovementEvent.create(null),
+ new EventMetaData(EventProducerConsumerType.OUTPUT, "vertex2", "vertex1", ta1V2T1));
+ List<TezEvent> events = new LinkedList<TezEvent>();
+ events.add(tezEvent);
+ v1.handle(new VertexEventRouteEvent(v1.getVertexId(), events));
+
+ startVertex(v2);
+ dispatcher.await();
+ v2.handle(new VertexEventTaskAttemptCompleted(ta1V2T1, TaskAttemptStateInternal.SUCCEEDED));
+ v2.handle(new VertexEventTaskCompleted(v2t1, TaskState.SUCCEEDED));
+ dispatcher.await();
+ Assert.assertEquals(VertexState.SUCCEEDED, v2.getState());
+
+ initializer.go();
+ while (v1.getState() == VertexState.INITIALIZING || v1.getState() == VertexState.INITED) {
+ Thread.sleep(10);
+ }
+
+ while (v1.getState() != VertexState.SUCCEEDED) {
+ Thread.sleep(10);
+ }
+ Assert.assertEquals(VertexState.SUCCEEDED, v1.getState());
+ }
+
@SuppressWarnings("unchecked")
@Test(timeout = 10000)
public void testRootInputInitializerEvent() throws Exception {
@@ -3441,6 +3550,47 @@ public class TestVertexImpl {
}
@InterfaceAudience.Private
+ public static class RootInitializerSettingParallelismTo0 extends InputInitializer {
+
+ private final ReentrantLock lock = new ReentrantLock();
+ private final Condition condition = lock.newCondition();
+
+ public RootInitializerSettingParallelismTo0(InputInitializerContext initializerContext) {
+ super(initializerContext);
+ }
+
+ @Override
+ public List<Event> initialize() throws Exception {
+ InputConfigureVertexTasksEvent event = InputConfigureVertexTasksEvent.create(0, null,
+ InputSpecUpdate.getDefaultSinglePhysicalInputSpecUpdate());
+ List<Event> events = new LinkedList<Event>();
+ events.add(event);
+ lock.lock();
+ try {
+ condition.await();
+ } finally {
+ lock.unlock();
+ }
+ LOG.info("Received signal to proceed. Returning event to set parallelism to 0");
+ return events;
+ }
+
+ public void go() {
+ lock.lock();
+ try {
+ LOG.info("Signallying initializer to proceed");
+ condition.signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void handleInputInitializerEvent(List<InputInitializerEvent> events) throws Exception {
+ }
+ }
+
+ @InterfaceAudience.Private
public static class EventHandlingRootInputInitializer extends InputInitializer {
final AtomicBoolean initStarted = new AtomicBoolean(false);