You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2013/10/31 00:21:20 UTC
git commit: TEZ-590. Vertex in inited state fails when source task
completes (bikas)
Updated Branches:
refs/heads/master 8496fccc1 -> 77549a723
TEZ-590. Vertex in inited state fails when source task completes (bikas)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/77549a72
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/77549a72
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/77549a72
Branch: refs/heads/master
Commit: 77549a7234e9865714ce276c3456f1bc67273a83
Parents: 8496fcc
Author: Bikas Saha <bi...@apache.org>
Authored: Wed Oct 30 16:17:23 2013 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Wed Oct 30 16:17:23 2013 -0700
----------------------------------------------------------------------
.../apache/tez/dag/app/dag/VertexScheduler.java | 2 +-
.../dag/impl/ImmediateStartVertexScheduler.java | 2 +-
.../app/dag/impl/RootInputVertexManager.java | 2 +-
.../dag/app/dag/impl/ShuffleVertexManager.java | 7 +++++-
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 26 ++++++++++++++++----
.../dag/app/dag/impl/TestVertexScheduler.java | 20 +++++++--------
.../split/TezGroupedSplitsInputFormat.java | 2 +-
.../split/TezGroupedSplitsInputFormat.java | 2 +-
.../apache/tez/mapreduce/hadoop/MRHelpers.java | 4 +--
.../hadoop/mapred/split/TestGroupedSplits.java | 4 +--
10 files changed, 46 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexScheduler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexScheduler.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexScheduler.java
index 8bbf446..812e577 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexScheduler.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexScheduler.java
@@ -30,7 +30,7 @@ import org.apache.tez.runtime.api.events.VertexManagerEvent;
// Rename to VertexManager TEZ-364 and move to DAG API. Make abstract class.
public interface VertexScheduler {
void initialize(Configuration conf);
- void onVertexStarted();
+ void onVertexStarted(List<TezTaskAttemptID> completions);
void onSourceTaskCompleted(TezTaskAttemptID attemptId);
void onVertexManagerEventReceived(VertexManagerEvent vmEvent);
void onRootVertexInitialized(String inputName, InputDescriptor inputDescriptor, List<Event> events);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexScheduler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexScheduler.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexScheduler.java
index 6217d91..85017b4 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexScheduler.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexScheduler.java
@@ -39,7 +39,7 @@ public class ImmediateStartVertexScheduler implements VertexScheduler {
}
@Override
- public void onVertexStarted() {
+ public void onVertexStarted(List<TezTaskAttemptID> completions) {
managedVertex.scheduleTasks(managedVertex.getTasks().keySet());
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/RootInputVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/RootInputVertexManager.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/RootInputVertexManager.java
index 4d548d4..0661b7c 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/RootInputVertexManager.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/RootInputVertexManager.java
@@ -74,7 +74,7 @@ public class RootInputVertexManager implements VertexScheduler {
}
@Override
- public void onVertexStarted() {
+ public void onVertexStarted(List<TezTaskAttemptID> completions) {
managedVertex.scheduleTasks(managedVertex.getTasks().keySet());
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
index 63ad8b2..21f0294 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ShuffleVertexManager.java
@@ -176,7 +176,7 @@ public class ShuffleVertexManager implements VertexScheduler {
@Override
- public void onVertexStarted() {
+ public void onVertexStarted(List<TezTaskAttemptID> completions) {
pendingTasks = new ArrayList<TezTaskID>(managedVertex.getTotalTasks());
// track the tasks in this vertex
updatePendingTasks();
@@ -186,6 +186,11 @@ public class ShuffleVertexManager implements VertexScheduler {
" with " + numSourceTasks + " source tasks and " +
totalTasksToSchedule + " pending tasks");
+ if (completions != null) {
+ for (TezTaskAttemptID srcAttemptId : completions) {
+ onSourceTaskCompleted(srcAttemptId);
+ }
+ }
// for the special case when source has 0 tasks or min fraction == 0
schedulePendingTasks();
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
index 9b33557..b9e7ae7 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
@@ -246,7 +246,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
.addTransition(VertexState.INITED, VertexState.INITED,
VertexEventType.V_SOURCE_VERTEX_STARTED,
new SourceVertexStartedTransition())
- .addTransition(VertexState.INITIALIZING, VertexState.INITIALIZING,
+ .addTransition(VertexState.INITED, VertexState.INITED,
VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
.addTransition(VertexState.INITED, VertexState.RUNNING,
@@ -322,6 +322,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
.addTransition(VertexState.SUCCEEDED, VertexState.SUCCEEDED,
EnumSet.of(VertexEventType.V_TERMINATE,
VertexEventType.V_TASK_ATTEMPT_COMPLETED,
+ // after we are done reruns of source tasks should not affect
+ // us. These reruns may be triggered by other consumer vertices.
+ // We should have been in RUNNING state if we had triggered the
+ // reruns.
+ VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
+ // accumulate these in case we get restarted
VertexEventType.V_ROUTE_EVENT,
VertexEventType.V_TASK_COMPLETED))
@@ -338,6 +344,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
VertexEventType.V_TASK_ATTEMPT_COMPLETED,
VertexEventType.V_TASK_COMPLETED,
VertexEventType.V_ROOT_INPUT_INITIALIZED,
+ VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
VertexEventType.V_ROOT_INPUT_FAILED))
// Transitions from KILLED state
@@ -353,6 +360,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
VertexEventType.V_ROUTE_EVENT,
VertexEventType.V_TASK_RESCHEDULED,
VertexEventType.V_TASK_ATTEMPT_COMPLETED,
+ VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
VertexEventType.V_TASK_COMPLETED,
VertexEventType.V_ROOT_INPUT_INITIALIZED,
VertexEventType.V_ROOT_INPUT_FAILED))
@@ -368,6 +376,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
VertexEventType.V_TERMINATE,
VertexEventType.V_TASK_COMPLETED,
VertexEventType.V_TASK_ATTEMPT_COMPLETED,
+ VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
VertexEventType.V_TASK_RESCHEDULED,
VertexEventType.V_INTERNAL_ERROR,
VertexEventType.V_ROOT_INPUT_INITIALIZED,
@@ -415,6 +424,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
private int numInitializedInputs;
private boolean startSignalPending = false;
List<TezEvent> pendingRouteEvents = null;
+ List<TezTaskAttemptID> pendingReportedSrcCompletions = Lists.newLinkedList();
private RootInputInitializerRunner rootInputInitializer;
@@ -1380,7 +1390,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
private void startVertex() {
startedTime = clock.getTime();
- vertexScheduler.onVertexStarted();
+ vertexScheduler.onVertexStarted(pendingReportedSrcCompletions);
+ pendingReportedSrcCompletions.clear();
logJobHistoryVertexStartedEvent();
// TODO: Metrics
@@ -1520,13 +1531,18 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
((VertexEventSourceTaskAttemptCompleted) event).getCompletionEvent();
LOG.info("Source task attempt completed for vertex: " + vertex.getVertexId()
+ " attempt: " + completionEvent.getTaskAttemptId()
- + " with state: " + completionEvent.getTaskAttemptState());
+ + " with state: " + completionEvent.getTaskAttemptState()
+ + " vertexState: " + vertex.getState());
if (TaskAttemptStateInternal.SUCCEEDED.equals(completionEvent
.getTaskAttemptState())) {
vertex.numSuccessSourceAttemptCompletions++;
- vertex.vertexScheduler.onSourceTaskCompleted(completionEvent
- .getTaskAttemptId());
+ if (vertex.getState() == VertexState.RUNNING) {
+ vertex.vertexScheduler.onSourceTaskCompleted(completionEvent
+ .getTaskAttemptId());
+ } else {
+ vertex.pendingReportedSrcCompletions.add(completionEvent.getTaskAttemptId());
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
index 528d03b..f506c99 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexScheduler.java
@@ -142,7 +142,7 @@ public class TestVertexScheduler {
// source vertices have 0 tasks. immediate start of all managed tasks
when(mockSrcVertex1.getTotalTasks()).thenReturn(0);
when(mockSrcVertex2.getTotalTasks()).thenReturn(0);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.isEmpty());
Assert.assertTrue(scheduledTasks.size() == 4); // all tasks scheduled
scheduledTasks.clear();
@@ -164,7 +164,7 @@ public class TestVertexScheduler {
VertexManagerEvent vmEvent = new VertexManagerEvent("Vertex", payload);
// parallelism not change due to large data size
scheduler = createScheduler(conf, mockManagedVertex, 0.1f, 0.1f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 4); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
scheduler.onVertexManagerEventReceived(vmEvent);
@@ -184,7 +184,7 @@ public class TestVertexScheduler {
vmEvent = new VertexManagerEvent("Vertex", payload);
scheduler = createScheduler(conf, mockManagedVertex, 0.5f, 0.5f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertEquals(4, scheduler.pendingTasks.size()); // no tasks scheduled
Assert.assertEquals(4, scheduler.numSourceTasks);
// task completion from non-bipartite stage does nothing
@@ -309,7 +309,7 @@ public class TestVertexScheduler {
// source vertices have 0 tasks. immediate start of all managed tasks
when(mockSrcVertex1.getTotalTasks()).thenReturn(0);
when(mockSrcVertex2.getTotalTasks()).thenReturn(0);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.isEmpty());
Assert.assertTrue(scheduledTasks.size() == 3); // all tasks scheduled
@@ -336,7 +336,7 @@ public class TestVertexScheduler {
// source vertex have some tasks. min, max == 0
scheduler = createScheduler(conf, mockManagedVertex, 0, 0);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.numSourceTasks == 4);
Assert.assertTrue(scheduler.totalTasksToSchedule == 3);
Assert.assertTrue(scheduler.numSourceTasksCompleted == 0);
@@ -356,7 +356,7 @@ public class TestVertexScheduler {
// min, max > 0 and min == max
scheduler = createScheduler(conf, mockManagedVertex, 0.25f, 0.25f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 3); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
// task completion from non-bipartite stage does nothing
@@ -371,7 +371,7 @@ public class TestVertexScheduler {
// min, max > 0 and min == max == absolute max 1.0
scheduler = createScheduler(conf, mockManagedVertex, 1.0f, 1.0f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 3); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
// task completion from non-bipartite stage does nothing
@@ -395,7 +395,7 @@ public class TestVertexScheduler {
// min, max > 0 and min == max
scheduler = createScheduler(conf, mockManagedVertex, 1.0f, 1.0f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 3); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
// task completion from non-bipartite stage does nothing
@@ -419,7 +419,7 @@ public class TestVertexScheduler {
// min, max > and min < max
scheduler = createScheduler(conf, mockManagedVertex, 0.25f, 0.75f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 3); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
scheduler.onSourceTaskCompleted(mockSrcAttemptId11);
@@ -444,7 +444,7 @@ public class TestVertexScheduler {
// min, max > and min < max
scheduler = createScheduler(conf, mockManagedVertex, 0.25f, 1.0f);
- scheduler.onVertexStarted();
+ scheduler.onVertexStarted(null);
Assert.assertTrue(scheduler.pendingTasks.size() == 3); // no tasks scheduled
Assert.assertTrue(scheduler.numSourceTasks == 4);
scheduler.onSourceTaskCompleted(mockSrcAttemptId11);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java b/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java
index 210ddea..604a2b7 100644
--- a/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java
+++ b/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplitsInputFormat.java
@@ -62,7 +62,7 @@ public class TezGroupedSplitsInputFormat<K, V>
}
}
- public void setDesiredNumberOfSPlits(int num) {
+ public void setDesiredNumberOfSplits(int num) {
Preconditions.checkArgument(num > 0);
this.desiredNumSplits = num;
if (LOG.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplitsInputFormat.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplitsInputFormat.java b/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplitsInputFormat.java
index 265bd93..f616744 100644
--- a/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplitsInputFormat.java
+++ b/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplitsInputFormat.java
@@ -61,7 +61,7 @@ public class TezGroupedSplitsInputFormat<K, V> extends InputFormat<K, V>
}
}
- public void setDesiredNumberOfSPlits(int num) {
+ public void setDesiredNumberOfSplits(int num) {
this.desiredNumSplits = num;
if (LOG.isDebugEnabled()) {
LOG.debug("desiredNumSplits: " + desiredNumSplits);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRHelpers.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRHelpers.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRHelpers.java
index cfad9fa..87b4032 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRHelpers.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRHelpers.java
@@ -173,7 +173,7 @@ public class MRHelpers {
new org.apache.hadoop.mapreduce.split.TezGroupedSplitsInputFormat();
groupedFormat.setConf(conf);
groupedFormat.setInputFormat(inputFormat);
- groupedFormat.setDesiredNumberOfSPlits(numTasks);
+ groupedFormat.setDesiredNumberOfSplits(numTasks);
finalInputFormat = groupedFormat;
}
@@ -251,7 +251,7 @@ public class MRHelpers {
new org.apache.hadoop.mapred.split.TezGroupedSplitsInputFormat();
groupedFormat.setConf(jobConf);
groupedFormat.setInputFormat(inputFormat);
- groupedFormat.setDesiredNumberOfSPlits(numTasks);
+ groupedFormat.setDesiredNumberOfSplits(numTasks);
finalInputFormat = groupedFormat;
}
org.apache.hadoop.mapred.InputSplit[] splits = finalInputFormat
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/77549a72/tez-mapreduce/src/test/java/org/apache/hadoop/mapred/split/TestGroupedSplits.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/test/java/org/apache/hadoop/mapred/split/TestGroupedSplits.java b/tez-mapreduce/src/test/java/org/apache/hadoop/mapred/split/TestGroupedSplits.java
index f913b13..583f4a2 100644
--- a/tez-mapreduce/src/test/java/org/apache/hadoop/mapred/split/TestGroupedSplits.java
+++ b/tez-mapreduce/src/test/java/org/apache/hadoop/mapred/split/TestGroupedSplits.java
@@ -93,7 +93,7 @@ public class TestGroupedSplits {
TezGroupedSplitsInputFormat<LongWritable , Text> format =
new TezGroupedSplitsInputFormat<LongWritable, Text>();
format.setConf(job);
- format.setDesiredNumberOfSPlits(1);
+ format.setDesiredNumberOfSplits(1);
format.setInputFormat(wrappedFormat);
LongWritable key = new LongWritable();
Text value = new Text();
@@ -237,7 +237,7 @@ public class TestGroupedSplits {
// TextInputFormat will produce 3 splits
for (int j=1; j<=3; ++j) {
- format.setDesiredNumberOfSPlits(j);
+ format.setDesiredNumberOfSplits(j);
InputSplit[] splits = format.getSplits(job, 100);
if (j==1 || j==3) {
// j==1 covers single split corner case