| /* Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.tez.dag.app.dag.impl; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.EnumSet; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReadWriteLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.classification.InterfaceAudience.Private; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.mapred.MRVertexOutputCommitter; |
| import org.apache.hadoop.security.Credentials; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; |
| import org.apache.hadoop.yarn.api.records.LocalResource; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.event.EventHandler; |
| import org.apache.hadoop.yarn.state.InvalidStateTransitonException; |
| import org.apache.hadoop.yarn.state.MultipleArcTransition; |
| import org.apache.hadoop.yarn.state.SingleArcTransition; |
| import org.apache.hadoop.yarn.state.StateMachine; |
| import org.apache.hadoop.yarn.state.StateMachineFactory; |
| import org.apache.hadoop.yarn.util.Clock; |
| import org.apache.tez.common.counters.TezCounters; |
| import org.apache.tez.dag.api.DagTypeConverters; |
| import org.apache.tez.dag.api.EdgeProperty.DataMovementType; |
| import org.apache.tez.dag.api.ProcessorDescriptor; |
| import org.apache.tez.dag.api.TezUncheckedException; |
| import org.apache.tez.dag.api.VertexLocationHint; |
| import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint; |
| import org.apache.tez.dag.api.client.ProgressBuilder; |
| import org.apache.tez.dag.api.client.VertexStatus; |
| import org.apache.tez.dag.api.client.VertexStatusBuilder; |
| import org.apache.tez.dag.api.committer.NullVertexOutputCommitter; |
| import org.apache.tez.dag.api.committer.VertexContext; |
| import org.apache.tez.dag.api.committer.VertexOutputCommitter; |
| import org.apache.tez.dag.api.oldrecords.TaskState; |
| import org.apache.tez.dag.api.records.DAGProtos.VertexPlan; |
| import org.apache.tez.dag.app.AppContext; |
| import org.apache.tez.dag.app.ContainerContext; |
| import org.apache.tez.dag.app.TaskAttemptListener; |
| import org.apache.tez.dag.app.TaskHeartbeatHandler; |
| import org.apache.tez.dag.app.dag.DAG; |
| import org.apache.tez.dag.app.dag.EdgeManager; |
| import org.apache.tez.dag.app.dag.Task; |
| import org.apache.tez.dag.app.dag.TaskAttemptStateInternal; |
| import org.apache.tez.dag.app.dag.TaskTerminationCause; |
| import org.apache.tez.dag.app.dag.Vertex; |
| import org.apache.tez.dag.app.dag.VertexScheduler; |
| import org.apache.tez.dag.app.dag.VertexState; |
| import org.apache.tez.dag.app.dag.VertexTerminationCause; |
| import org.apache.tez.dag.app.dag.event.DAGEvent; |
| import org.apache.tez.dag.app.dag.event.DAGEventDiagnosticsUpdate; |
| import org.apache.tez.dag.app.dag.event.DAGEventType; |
| import org.apache.tez.dag.app.dag.event.DAGEventVertexCompleted; |
| import org.apache.tez.dag.app.dag.event.DAGEventVertexReRunning; |
| import org.apache.tez.dag.app.dag.event.TaskAttemptEvent; |
| import org.apache.tez.dag.app.dag.event.TaskAttemptEventAttemptFailed; |
| import org.apache.tez.dag.app.dag.event.TaskAttemptEventStatusUpdate; |
| import org.apache.tez.dag.app.dag.event.TaskAttemptEventType; |
| import org.apache.tez.dag.app.dag.event.TaskEvent; |
| import org.apache.tez.dag.app.dag.event.TaskEventTermination; |
| import org.apache.tez.dag.app.dag.event.TaskEventType; |
| import org.apache.tez.dag.app.dag.event.VertexEvent; |
| import org.apache.tez.dag.app.dag.event.VertexEventRouteEvent; |
| import org.apache.tez.dag.app.dag.event.VertexEventSourceTaskAttemptCompleted; |
| import org.apache.tez.dag.app.dag.event.VertexEventSourceVertexStarted; |
| import org.apache.tez.dag.app.dag.event.VertexEventTaskAttemptCompleted; |
| import org.apache.tez.dag.app.dag.event.VertexEventTaskCompleted; |
| import org.apache.tez.dag.app.dag.event.VertexEventTaskReschedule; |
| import org.apache.tez.dag.app.dag.event.VertexEventTermination; |
| import org.apache.tez.dag.app.dag.event.VertexEventType; |
| import org.apache.tez.dag.history.DAGHistoryEvent; |
| import org.apache.tez.dag.history.events.VertexFinishedEvent; |
| import org.apache.tez.dag.history.events.VertexStartedEvent; |
| import org.apache.tez.dag.records.TezDAGID; |
| import org.apache.tez.dag.records.TezTaskAttemptID; |
| import org.apache.tez.dag.records.TezTaskID; |
| import org.apache.tez.dag.records.TezVertexID; |
| import org.apache.tez.runtime.api.events.DataMovementEvent; |
| import org.apache.tez.runtime.api.events.InputFailedEvent; |
| import org.apache.tez.runtime.api.events.TaskAttemptFailedEvent; |
| import org.apache.tez.runtime.api.events.TaskStatusUpdateEvent; |
| import org.apache.tez.runtime.api.impl.EventMetaData; |
| import org.apache.tez.runtime.api.impl.InputSpec; |
| import org.apache.tez.runtime.api.impl.OutputSpec; |
| import org.apache.tez.runtime.api.impl.TezEvent; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.collect.HashMultiset; |
| import com.google.common.collect.Multiset; |
| |
| |
| /** Implementation of Vertex interface. Maintains the state machines of Vertex. |
| * The read and write calls use ReadWriteLock for concurrency. |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, |
| EventHandler<VertexEvent> { |
| |
| private static final String LINE_SEPARATOR = System |
| .getProperty("line.separator"); |
| |
| private static final Log LOG = LogFactory.getLog(VertexImpl.class); |
| |
| //final fields |
| private final Clock clock; |
| |
| |
| private final Lock readLock; |
| private final Lock writeLock; |
| private final TaskAttemptListener taskAttemptListener; |
| private final TaskHeartbeatHandler taskHeartbeatHandler; |
| private final Object tasksSyncHandle = new Object(); |
| |
| private final EventHandler eventHandler; |
| // TODO Metrics |
| //private final MRAppMetrics metrics; |
| private final AppContext appContext; |
| |
| private boolean lazyTasksCopyNeeded = false; |
| // must be a linked map for ordering |
| volatile LinkedHashMap<TezTaskID, Task> tasks = new LinkedHashMap<TezTaskID, Task>(); |
| private Object fullCountersLock = new Object(); |
| private TezCounters fullCounters = null; |
| private Resource taskResource; |
| |
| private Configuration conf; |
| |
| //fields initialized in init |
| |
| private int numStartedSourceVertices = 0; |
| private int distanceFromRoot = 0; |
| |
| private final List<String> diagnostics = new ArrayList<String>(); |
| |
| //task/attempt related datastructures |
| @VisibleForTesting |
| int numSuccessSourceAttemptCompletions = 0; |
| |
| List<InputSpec> inputSpecList; |
| List<OutputSpec> outputSpecList; |
| |
| private static final InternalErrorTransition |
| INTERNAL_ERROR_TRANSITION = new InternalErrorTransition(); |
| private static final RouteEventTransition |
| ROUTE_EVENT_TRANSITION = new RouteEventTransition(); |
| private static final TaskAttemptCompletedEventTransition |
| TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION = |
| new TaskAttemptCompletedEventTransition(); |
| private static final SourceTaskAttemptCompletedEventTransition |
| SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION = |
| new SourceTaskAttemptCompletedEventTransition(); |
| |
| protected static final |
| StateMachineFactory<VertexImpl, VertexState, VertexEventType, VertexEvent> |
| stateMachineFactory |
| = new StateMachineFactory<VertexImpl, VertexState, VertexEventType, VertexEvent> |
| (VertexState.NEW) |
| |
| // Transitions from NEW state |
| .addTransition |
| (VertexState.NEW, |
| EnumSet.of(VertexState.INITED, VertexState.FAILED), |
| VertexEventType.V_INIT, |
| new InitTransition()) |
| .addTransition(VertexState.NEW, VertexState.KILLED, |
| VertexEventType.V_TERMINATE, |
| new TerminateNewVertexTransition()) |
| .addTransition(VertexState.NEW, VertexState.ERROR, |
| VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| |
| // Transitions from INITED state |
| .addTransition(VertexState.INITED, VertexState.INITED, |
| VertexEventType.V_SOURCE_VERTEX_STARTED, |
| new SourceVertexStartedTransition()) |
| .addTransition(VertexState.INITED, VertexState.RUNNING, |
| VertexEventType.V_START, |
| new StartTransition()) |
| |
| .addTransition(VertexState.INITED, VertexState.KILLED, |
| VertexEventType.V_TERMINATE, |
| new TerminateInitedVertexTransition()) |
| .addTransition(VertexState.INITED, VertexState.ERROR, |
| VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| |
| // Transitions from RUNNING state |
| .addTransition(VertexState.RUNNING, VertexState.RUNNING, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION) |
| .addTransition(VertexState.RUNNING, VertexState.RUNNING, |
| VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED, |
| SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION) |
| .addTransition |
| (VertexState.RUNNING, |
| EnumSet.of(VertexState.RUNNING, |
| VertexState.SUCCEEDED, VertexState.TERMINATING, VertexState.FAILED), |
| VertexEventType.V_TASK_COMPLETED, |
| new TaskCompletedTransition()) |
| .addTransition(VertexState.RUNNING, VertexState.TERMINATING, |
| VertexEventType.V_TERMINATE, |
| new VertexKilledTransition()) |
| .addTransition(VertexState.RUNNING, VertexState.RUNNING, |
| VertexEventType.V_TASK_RESCHEDULED, |
| new TaskRescheduledTransition()) |
| .addTransition( |
| VertexState.RUNNING, |
| VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| .addTransition( |
| VertexState.RUNNING, |
| VertexState.RUNNING, VertexEventType.V_ROUTE_EVENT, |
| ROUTE_EVENT_TRANSITION) |
| |
| // Transitions from TERMINATING state. |
| .addTransition |
| (VertexState.TERMINATING, |
| EnumSet.of(VertexState.TERMINATING, VertexState.KILLED, VertexState.FAILED), |
| VertexEventType.V_TASK_COMPLETED, |
| new TaskCompletedTransition()) |
| .addTransition(VertexState.TERMINATING, VertexState.TERMINATING, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION) // TODO shouldnt be done for KILL_WAIT vertex |
| .addTransition(VertexState.TERMINATING, VertexState.TERMINATING, |
| VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED, |
| SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION) |
| .addTransition( |
| VertexState.TERMINATING, |
| VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| // Ignore-able events |
| .addTransition(VertexState.TERMINATING, VertexState.TERMINATING, |
| EnumSet.of(VertexEventType.V_TERMINATE, |
| VertexEventType.V_TASK_RESCHEDULED)) |
| |
| // Transitions from SUCCEEDED state |
| .addTransition( |
| VertexState.SUCCEEDED, |
| VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| .addTransition(VertexState.SUCCEEDED, |
| EnumSet.of(VertexState.RUNNING, VertexState.FAILED), |
| VertexEventType.V_TASK_RESCHEDULED, |
| new TaskRescheduledAfterVertexSuccessTransition()) |
| |
| // Ignore-able events |
| .addTransition(VertexState.SUCCEEDED, VertexState.SUCCEEDED, |
| EnumSet.of(VertexEventType.V_TERMINATE, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| VertexEventType.V_TASK_COMPLETED)) |
| |
| // Transitions from FAILED state |
| .addTransition( |
| VertexState.FAILED, |
| VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| // Ignore-able events |
| .addTransition(VertexState.FAILED, VertexState.FAILED, |
| EnumSet.of(VertexEventType.V_TERMINATE, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| VertexEventType.V_TASK_COMPLETED)) |
| |
| // Transitions from KILLED state |
| .addTransition( |
| VertexState.KILLED, |
| VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, |
| INTERNAL_ERROR_TRANSITION) |
| // Ignore-able events |
| .addTransition(VertexState.KILLED, VertexState.KILLED, |
| EnumSet.of(VertexEventType.V_TERMINATE, |
| VertexEventType.V_START, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| VertexEventType.V_TASK_COMPLETED)) |
| |
| // No transitions from INTERNAL_ERROR state. Ignore all. |
| .addTransition( |
| VertexState.ERROR, |
| VertexState.ERROR, |
| EnumSet.of(VertexEventType.V_INIT, |
| VertexEventType.V_TERMINATE, |
| VertexEventType.V_TASK_COMPLETED, |
| VertexEventType.V_TASK_ATTEMPT_COMPLETED, |
| VertexEventType.V_TASK_RESCHEDULED, |
| VertexEventType.V_DIAGNOSTIC_UPDATE, |
| VertexEventType.V_INTERNAL_ERROR)) |
| // create the topology tables |
| .installTopology(); |
| |
| private final StateMachine<VertexState, VertexEventType, VertexEvent> |
| stateMachine; |
| |
| //changing fields while the vertex is running |
| private int numTasks; |
| private int completedTaskCount = 0; |
| private int succeededTaskCount = 0; |
| private int failedTaskCount = 0; |
| private int killedTaskCount = 0; |
| |
| private long initTime; |
| private long startTime; |
| private long finishTime; |
| private float progress; |
| |
| private Credentials credentials; |
| |
| private final TezVertexID vertexId; //runtime assigned id. |
| private final VertexPlan vertexPlan; |
| |
| private final String vertexName; |
| private final ProcessorDescriptor processorDescriptor; |
| |
| // For committer |
| private final VertexContext vertexContext; |
| |
| @VisibleForTesting |
| Map<Vertex, Edge> sourceVertices; |
| private Map<Vertex, Edge> targetVertices; |
| |
| private VertexScheduler vertexScheduler; |
| |
| private VertexOutputCommitter committer; |
| private AtomicBoolean committed = new AtomicBoolean(false); |
| private VertexLocationHint vertexLocationHint; |
| private Map<String, LocalResource> localResources; |
| private Map<String, String> environment; |
| private final String javaOpts; |
| private final ContainerContext containerContext; |
| private VertexTerminationCause terminationCause; |
| |
| public VertexImpl(TezVertexID vertexId, VertexPlan vertexPlan, |
| String vertexName, Configuration conf, EventHandler eventHandler, |
| TaskAttemptListener taskAttemptListener, |
| Credentials credentials, Clock clock, |
| // TODO: Recovery |
| //Map<TaskId, TaskInfo> completedTasksFromPreviousRun, |
| // TODO Metrics |
| //MRAppMetrics metrics, |
| TaskHeartbeatHandler thh, |
| AppContext appContext, VertexLocationHint vertexLocationHint) { |
| this.vertexId = vertexId; |
| this.vertexPlan = vertexPlan; |
| this.vertexName = vertexName; |
| this.conf = conf; |
| //this.metrics = metrics; |
| this.clock = clock; |
| // TODO: Recovery |
| //this.completedTasksFromPreviousRun = completedTasksFromPreviousRun; |
| this.appContext = appContext; |
| |
| this.taskAttemptListener = taskAttemptListener; |
| this.taskHeartbeatHandler = thh; |
| this.eventHandler = eventHandler; |
| ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); |
| this.readLock = readWriteLock.readLock(); |
| this.writeLock = readWriteLock.writeLock(); |
| |
| this.credentials = credentials; |
| this.committer = new NullVertexOutputCommitter(); |
| this.vertexLocationHint = vertexLocationHint; |
| if (LOG.isDebugEnabled()) { |
| logLocationHints(this.vertexLocationHint); |
| } |
| |
| this.taskResource = DagTypeConverters |
| .createResourceRequestFromTaskConfig(vertexPlan.getTaskConfig()); |
| this.processorDescriptor = DagTypeConverters |
| .convertProcessorDescriptorFromDAGPlan(vertexPlan |
| .getProcessorDescriptor()); |
| this.localResources = DagTypeConverters |
| .createLocalResourceMapFromDAGPlan(vertexPlan.getTaskConfig() |
| .getLocalResourceList()); |
| this.environment = DagTypeConverters |
| .createEnvironmentMapFromDAGPlan(vertexPlan.getTaskConfig() |
| .getEnvironmentSettingList()); |
| this.javaOpts = vertexPlan.getTaskConfig().hasJavaOpts() ? vertexPlan |
| .getTaskConfig().getJavaOpts() : null; |
| |
| this.vertexContext = new VertexContext(getDAGId(), |
| this.processorDescriptor.getUserPayload(), this.vertexId, |
| getApplicationAttemptId()); |
| |
| this.containerContext = new ContainerContext(this.localResources, |
| this.credentials, this.environment, this.javaOpts); |
| // This "this leak" is okay because the retained pointer is in an |
| // instance variable. |
| stateMachine = stateMachineFactory.make(this); |
| } |
| |
| protected StateMachine<VertexState, VertexEventType, VertexEvent> getStateMachine() { |
| return stateMachine; |
| } |
| |
| @Override |
| public TezVertexID getVertexId() { |
| return vertexId; |
| } |
| |
| @Override |
| public VertexPlan getVertexPlan() { |
| return vertexPlan; |
| } |
| |
| @Override |
| public int getDistanceFromRoot() { |
| return distanceFromRoot; |
| } |
| |
| @Override |
| public String getName() { |
| return vertexName; |
| } |
| |
| EventHandler getEventHandler() { |
| return this.eventHandler; |
| } |
| |
| @Override |
| public Task getTask(TezTaskID taskID) { |
| readLock.lock(); |
| try { |
| return tasks.get(taskID); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public Task getTask(int taskIndex) { |
| readLock.lock(); |
| try { |
| // does it matter to create a duplicate list for efficiency |
| // instead of traversing the map |
| // local assign to LinkedHashMap to ensure that sequential traversal |
| // assumption is satisfied |
| LinkedHashMap<TezTaskID, Task> taskList = tasks; |
| int i=0; |
| for(Map.Entry<TezTaskID, Task> entry : taskList.entrySet()) { |
| if(taskIndex == i) { |
| return entry.getValue(); |
| } |
| ++i; |
| } |
| return null; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public int getTotalTasks() { |
| return numTasks; |
| } |
| |
| @Override |
| public int getCompletedTasks() { |
| readLock.lock(); |
| try { |
| return succeededTaskCount + failedTaskCount + killedTaskCount; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public int getSucceededTasks() { |
| readLock.lock(); |
| try { |
| return succeededTaskCount; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public TezCounters getAllCounters() { |
| |
| readLock.lock(); |
| |
| try { |
| VertexState state = getInternalState(); |
| if (state == VertexState.ERROR || state == VertexState.FAILED |
| || state == VertexState.KILLED || state == VertexState.SUCCEEDED) { |
| this.mayBeConstructFinalFullCounters(); |
| return fullCounters; |
| } |
| |
| TezCounters counters = new TezCounters(); |
| return incrTaskCounters(counters, tasks.values()); |
| |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| public static TezCounters incrTaskCounters( |
| TezCounters counters, Collection<Task> tasks) { |
| for (Task task : tasks) { |
| counters.incrAllCounters(task.getCounters()); |
| } |
| return counters; |
| } |
| |
| @Override |
| public List<String> getDiagnostics() { |
| readLock.lock(); |
| try { |
| return diagnostics; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public float getProgress() { |
| this.readLock.lock(); |
| try { |
| computeProgress(); |
| return progress; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public ProgressBuilder getVertexProgress() { |
| this.readLock.lock(); |
| try { |
| ProgressBuilder progress = new ProgressBuilder(); |
| progress.setTotalTaskCount(numTasks); |
| progress.setSucceededTaskCount(succeededTaskCount); |
| progress.setRunningTaskCount(0); // TODO TEZ-130 |
| progress.setFailedTaskCount(failedTaskCount); |
| progress.setKilledTaskCount(killedTaskCount); |
| return progress; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public VertexStatusBuilder getVertexStatus() { |
| this.readLock.lock(); |
| try { |
| VertexStatusBuilder status = new VertexStatusBuilder(); |
| status.setState(getInternalState()); |
| status.setDiagnostics(diagnostics); |
| status.setProgress(getVertexProgress()); |
| return status; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| private void computeProgress() { |
| this.readLock.lock(); |
| try { |
| float progress = 0f; |
| for (Task task : this.tasks.values()) { |
| progress += (task.isFinished() ? 1f : task.getProgress()); |
| } |
| if (this.numTasks != 0) { |
| progress /= this.numTasks; |
| } |
| this.progress = progress; |
| } finally { |
| this.readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public Map<TezTaskID, Task> getTasks() { |
| synchronized (tasksSyncHandle) { |
| lazyTasksCopyNeeded = true; |
| return Collections.unmodifiableMap(tasks); |
| } |
| } |
| |
| @Override |
| public VertexState getState() { |
| readLock.lock(); |
| try { |
| return getStateMachine().getCurrentState(); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Set the terminationCause if it had not yet been set. |
| * |
| * @param trigger The trigger |
| * @return true if setting the value succeeded. |
| */ |
| boolean trySetTerminationCause(VertexTerminationCause trigger) { |
| if(terminationCause == null){ |
| terminationCause = trigger; |
| return true; |
| } |
| return false; |
| } |
| |
| public VertexTerminationCause getTerminationCause(){ |
| readLock.lock(); |
| try { |
| return terminationCause; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| // TODO Create InputReadyVertexManager that schedules when there is something |
| // to read and use that as default instead of ImmediateStart.TEZ-480 |
| @Override |
| public void scheduleTasks(Collection<TezTaskID> taskIDs) { |
| readLock.lock(); |
| try { |
| for (TezTaskID taskID : taskIDs) { |
| eventHandler.handle(new TaskEvent(taskID, |
| TaskEventType.T_SCHEDULE)); |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void setParallelism(int parallelism, |
| Map<Vertex, EdgeManager> sourceEdgeManagers) { |
| writeLock.lock(); |
| try { |
| if (parallelism >= numTasks) { |
| // not that hard to support perhaps. but checking right now since there |
| // is no use case for it and checking may catch other bugs. |
| throw new TezUncheckedException( |
| "Increasing parallelism is not supported"); |
| } |
| if (parallelism == numTasks) { |
| LOG.info("Ingoring setParallelism to current value: " + parallelism); |
| return; |
| } |
| |
| // start buffering incoming events so that we can re-route existing events |
| for (Edge edge : sourceVertices.values()) { |
| edge.startEventBuffering(); |
| } |
| |
| // Use a set since the same event may have been sent to multiple tasks |
| // and we want to avoid duplicates |
| Set<TezEvent> pendingEvents = new HashSet<TezEvent>(); |
| |
| LOG.info("Vertex " + getVertexId() + " parallelism set to " + parallelism); |
| // assign to local variable of LinkedHashMap to make sure that changing |
| // type of task causes compile error. We depend on LinkedHashMap for order |
| LinkedHashMap<TezTaskID, Task> currentTasks = this.tasks; |
| Iterator<Map.Entry<TezTaskID, Task>> iter = currentTasks.entrySet() |
| .iterator(); |
| int i = 0; |
| while (iter.hasNext()) { |
| i++; |
| Map.Entry<TezTaskID, Task> entry = iter.next(); |
| Task task = entry.getValue(); |
| if (task.getState() != TaskState.NEW) { |
| throw new TezUncheckedException( |
| "All tasks must be in initial state when changing parallelism" |
| + " for vertex: " + getVertexId() + " name: " + getName()); |
| } |
| pendingEvents.addAll(task.getAndClearTaskTezEvents()); |
| if (i <= parallelism) { |
| continue; |
| } |
| LOG.info("Removing task: " + entry.getKey()); |
| iter.remove(); |
| } |
| this.numTasks = parallelism; |
| assert tasks.size() == numTasks; |
| |
| // set new edge managers |
| if(sourceEdgeManagers != null) { |
| for(Map.Entry<Vertex, EdgeManager> entry : sourceEdgeManagers.entrySet()) { |
| Vertex sourceVertex = entry.getKey(); |
| EdgeManager edgeManager = entry.getValue(); |
| Edge edge = sourceVertices.get(sourceVertex); |
| LOG.info("Replacing edge manager for source:" |
| + sourceVertex.getVertexId() + " destination: " + getVertexId()); |
| edge.setEdgeManager(edgeManager); |
| } |
| } |
| |
| // Re-route all existing TezEvents according to new routing table |
| // At this point only events attributed to source task attempts can be |
| // re-routed. e.g. DataMovement or InputFailed events. |
| // This assumption is fine for now since these tasks haven't been started. |
| // So they can only get events generated from source task attempts that |
| // have already been started. |
| DAG dag = getDAG(); |
| for(TezEvent event : pendingEvents) { |
| TezVertexID sourceVertexId = event.getSourceInfo().getTaskAttemptID() |
| .getTaskID().getVertexID(); |
| Vertex sourceVertex = dag.getVertex(sourceVertexId); |
| Edge sourceEdge = sourceVertices.get(sourceVertex); |
| sourceEdge.sendTezEventToDestinationTasks(event); |
| } |
| |
| // stop buffering events |
| for (Edge edge : sourceVertices.values()) { |
| edge.stopEventBuffering(); |
| } |
| |
| } finally { |
| writeLock.unlock(); |
| } |
| |
| } |
| |
| @Override |
| /** |
| * The only entry point to change the Vertex. |
| */ |
| public void handle(VertexEvent event) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Processing VertexEvent " + event.getVertexId() |
| + " of type " + event.getType() + " while in state " |
| + getInternalState() + ". Event: " + event); |
| } |
| try { |
| writeLock.lock(); |
| VertexState oldState = getInternalState(); |
| try { |
| getStateMachine().doTransition(event.getType(), event); |
| } catch (InvalidStateTransitonException e) { |
| String message = "Invalid event " + event.getType() + |
| " on vertex " + this.vertexName + |
| " with vertexId " + this.vertexId + |
| " at current state " + oldState; |
| LOG.error("Can't handle " + message, e); |
| addDiagnostic(message); |
| eventHandler.handle(new VertexEvent(this.vertexId, |
| VertexEventType.V_INTERNAL_ERROR)); |
| } |
| |
| if (oldState != getInternalState()) { |
| LOG.info(vertexId + " transitioned from " + oldState + " to " |
| + getInternalState()); |
| } |
| } |
| |
| finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| private VertexState getInternalState() { |
| readLock.lock(); |
| try { |
| return getStateMachine().getCurrentState(); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| //helpful in testing |
| protected void addTask(Task task) { |
| synchronized (tasksSyncHandle) { |
| if (lazyTasksCopyNeeded) { |
| LinkedHashMap<TezTaskID, Task> newTasks = new LinkedHashMap<TezTaskID, Task>(); |
| newTasks.putAll(tasks); |
| tasks = newTasks; |
| lazyTasksCopyNeeded = false; |
| } |
| } |
| tasks.put(task.getTaskId(), task); |
| // TODO Metrics |
| //metrics.waitingTask(task); |
| } |
| |
| void setFinishTime() { |
| finishTime = clock.getTime(); |
| } |
| |
| |
| void logJobHistoryVertexStartedEvent() { |
| VertexStartedEvent startEvt = new VertexStartedEvent(vertexId, |
| vertexName, initTime, startTime, numTasks, getProcessorName()); |
| this.eventHandler.handle(new DAGHistoryEvent(startEvt)); |
| } |
| |
| void logJobHistoryVertexFinishedEvent() { |
| this.setFinishTime(); |
| VertexFinishedEvent finishEvt = new VertexFinishedEvent(vertexId, |
| vertexName, startTime, finishTime, VertexStatus.State.SUCCEEDED, "", |
| getAllCounters()); |
| this.eventHandler.handle(new DAGHistoryEvent(finishEvt)); |
| } |
| |
| void logJobHistoryVertexFailedEvent(VertexStatus.State state) { |
| VertexFinishedEvent finishEvt = new VertexFinishedEvent(vertexId, |
| vertexName, startTime, clock.getTime(), state, |
| StringUtils.join(LINE_SEPARATOR, getDiagnostics()), |
| getAllCounters()); |
| this.eventHandler.handle(new DAGHistoryEvent(finishEvt)); |
| } |
| |
| static VertexState checkVertexForCompletion(VertexImpl vertex) { |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Checking for vertex completion" |
| + ", failedTaskCount=" + vertex.failedTaskCount |
| + ", killedTaskCount=" + vertex.killedTaskCount |
| + ", successfulTaskCount=" + vertex.succeededTaskCount |
| + ", completedTaskCount=" + vertex.completedTaskCount |
| + ", terminationCause=" + vertex.terminationCause); |
| } |
| |
| //check for vertex failure first |
| if (vertex.completedTaskCount > vertex.tasks.size()) { |
| LOG.error("task completion accounting issue: completedTaskCount > nTasks:" |
| + ", failedTaskCount=" + vertex.failedTaskCount |
| + ", killedTaskCount=" + vertex.killedTaskCount |
| + ", successfulTaskCount=" + vertex.succeededTaskCount |
| + ", completedTaskCount=" + vertex.completedTaskCount |
| + ", terminationCause=" + vertex.terminationCause); |
| } |
| |
| if (vertex.completedTaskCount == vertex.tasks.size()) { |
| //Only succeed if tasks complete successfully and no terminationCause is registered. |
| if(vertex.succeededTaskCount == vertex.tasks.size() && vertex.terminationCause == null) { |
| try { |
| if (!vertex.committed.getAndSet(true)) { |
| // commit only once |
| vertex.committer.commitVertex(); |
| } |
| } catch (IOException e) { |
| LOG.error("Failed to do commit on vertex, name=" + vertex.getName(), e); |
| vertex.trySetTerminationCause(VertexTerminationCause.COMMIT_FAILURE); |
| return vertex.finished(VertexState.FAILED); |
| } |
| return vertex.finished(VertexState.SUCCEEDED); |
| } |
| else if(vertex.terminationCause == VertexTerminationCause.DAG_KILL ){ |
| vertex.setFinishTime(); |
| String diagnosticMsg = "Vertex killed due to user-initiated job kill. " |
| + "failedTasks:" |
| + vertex.failedTaskCount; |
| LOG.info(diagnosticMsg); |
| vertex.addDiagnostic(diagnosticMsg); |
| vertex.abortVertex(VertexStatus.State.KILLED); |
| return vertex.finished(VertexState.KILLED); |
| } |
| else if(vertex.terminationCause == VertexTerminationCause.OTHER_VERTEX_FAILURE ){ |
| vertex.setFinishTime(); |
| String diagnosticMsg = "Vertex killed as other vertex failed. " |
| + "failedTasks:" |
| + vertex.failedTaskCount; |
| LOG.info(diagnosticMsg); |
| vertex.addDiagnostic(diagnosticMsg); |
| vertex.abortVertex(VertexStatus.State.KILLED); |
| return vertex.finished(VertexState.KILLED); |
| } |
| else if(vertex.terminationCause == VertexTerminationCause.OWN_TASK_FAILURE ){ |
| if(vertex.failedTaskCount == 0){ |
| LOG.error("task failure accounting error. terminationCause=TASK_FAILURE but vertex.failedTaskCount == 0"); |
| } |
| vertex.setFinishTime(); |
| String diagnosticMsg = "Vertex killed as one or more tasks failed. " |
| + "failedTasks:" |
| + vertex.failedTaskCount; |
| LOG.info(diagnosticMsg); |
| vertex.addDiagnostic(diagnosticMsg); |
| vertex.abortVertex(VertexStatus.State.FAILED); |
| return vertex.finished(VertexState.FAILED); |
| } |
| else { |
| //should never occur |
| throw new TezUncheckedException("All tasks complete, but cannot determine final state of vertex" |
| + ", failedTaskCount=" + vertex.failedTaskCount |
| + ", killedTaskCount=" + vertex.killedTaskCount |
| + ", successfulTaskCount=" + vertex.succeededTaskCount |
| + ", completedTaskCount=" + vertex.completedTaskCount |
| + ", terminationCause=" + vertex.terminationCause); |
| } |
| } |
| |
| //return the current state, Vertex not finished yet |
| return vertex.getInternalState(); |
| } |
| |
| /** |
| * Set the terminationCause and send a kill-message to all tasks. |
| * The task-kill messages are only sent once. |
| * @param the trigger that is causing the Vertex to transition to KILLED/FAILED |
| * @param event The type of kill event to send to the vertices. |
| */ |
| void enactKill(VertexTerminationCause trigger, TaskTerminationCause taskterminationCause) { |
| if(trySetTerminationCause(trigger)){ |
| for (Task task : tasks.values()) { |
| eventHandler.handle( |
| new TaskEventTermination(task.getTaskId(), taskterminationCause)); |
| } |
| } |
| } |
| |
| VertexState finished(VertexState finalState) { |
| if (finishTime == 0) setFinishTime(); |
| |
| switch (finalState) { |
| case KILLED: |
| eventHandler.handle(new DAGEventVertexCompleted(getVertexId(), |
| finalState)); |
| logJobHistoryVertexFailedEvent(VertexStatus.State.KILLED); |
| break; |
| case ERROR: |
| eventHandler.handle(new DAGEvent(getDAGId(), |
| DAGEventType.INTERNAL_ERROR)); |
| logJobHistoryVertexFailedEvent(VertexStatus.State.FAILED); |
| break; |
| case FAILED: |
| eventHandler.handle(new DAGEventVertexCompleted(getVertexId(), |
| finalState)); |
| logJobHistoryVertexFailedEvent(VertexStatus.State.FAILED); |
| break; |
| case SUCCEEDED: |
| eventHandler.handle(new DAGEventVertexCompleted(getVertexId(), |
| finalState)); |
| logJobHistoryVertexFinishedEvent(); |
| break; |
| default: |
| throw new TezUncheckedException("Unexpected VertexState: " + finalState); |
| } |
| return finalState; |
| } |
| |
| public static class InitTransition |
| implements MultipleArcTransition<VertexImpl, VertexEvent, VertexState> { |
| |
| @Override |
| public VertexState transition(VertexImpl vertex, VertexEvent event) { |
| try { |
| |
| // TODODAGAM |
| // TODO: Splits? |
| |
| vertex.numTasks = vertex.getVertexPlan().getTaskConfig().getNumTasks(); |
| |
| /* |
| TaskSplitMetaInfo[] taskSplitMetaInfo = createSplits(job, job.jobId); |
| job.numMapTasks = taskSplitMetaInfo.length; |
| */ |
| |
| if (vertex.numTasks == 0) { |
| vertex.addDiagnostic("No tasks for vertex " + vertex.getVertexId()); |
| vertex.trySetTerminationCause(VertexTerminationCause.ZERO_TASKS); |
| vertex.abortVertex(VertexStatus.State.FAILED); |
| return vertex.finished(VertexState.FAILED); |
| } |
| |
| checkTaskLimits(); |
| |
| // create the Tasks but don't start them yet |
| createTasks(vertex); |
| |
| // TODO get this from API |
| boolean hasBipartite = false; |
| if (vertex.sourceVertices != null) { |
| for (Edge edge : vertex.sourceVertices.values()) { |
| if (edge.getEdgeProperty().getDataMovementType() == |
| DataMovementType.SCATTER_GATHER) { |
| hasBipartite = true; |
| break; |
| } |
| } |
| } |
| |
| if (hasBipartite) { |
| // setup vertex scheduler |
| // TODO this needs to consider data size and perhaps API. |
| // Currently implicitly BIPARTITE is the only edge type |
| vertex.vertexScheduler = new ShuffleVertexManager(vertex); |
| } else { |
| // schedule all tasks upon vertex start |
| vertex.vertexScheduler = new ImmediateStartVertexScheduler(vertex); |
| } |
| |
| vertex.vertexScheduler.initialize(vertex.conf); |
| |
| // FIXME how do we decide vertex needs a committer? |
| // Answer: Do commit for every vertex |
| // for now, only for leaf vertices |
| // TODO TEZ-41 make commmitter type configurable per vertex |
| if (vertex.targetVertices.isEmpty()) { |
| vertex.committer = new MRVertexOutputCommitter(); |
| } |
| vertex.committer.init(vertex.vertexContext); |
| vertex.committer.setupVertex(); |
| |
| // TODO: Metrics |
| //vertex.metrics.endPreparingJob(job); |
| vertex.initTime = vertex.clock.getTime(); |
| return VertexState.INITED; |
| |
| } catch (IOException e) { |
| LOG.warn("Vertex init failed", e); |
| vertex.addDiagnostic("Job init failed : " |
| + StringUtils.stringifyException(e)); |
| vertex.trySetTerminationCause(VertexTerminationCause.INIT_FAILURE); |
| vertex.abortVertex(VertexStatus.State.FAILED); |
| // TODO: Metrics |
| //job.metrics.endPreparingJob(vertex); |
| return vertex.finished(VertexState.FAILED); |
| } |
| } |
| |
| |
| private void createTasks(VertexImpl vertex) { |
| Configuration conf = vertex.conf; |
| boolean useNullLocationHint = true; |
| if (vertex.vertexLocationHint != null |
| && vertex.vertexLocationHint.getTaskLocationHints() != null |
| && vertex.vertexLocationHint.getTaskLocationHints().size() == |
| vertex.numTasks) { |
| useNullLocationHint = false; |
| } |
| for (int i=0; i < vertex.numTasks; ++i) { |
| TaskLocationHint locHint = null; |
| if (!useNullLocationHint) { |
| locHint = vertex.vertexLocationHint.getTaskLocationHints().get(i); |
| } |
| TaskImpl task = |
| new TaskImpl(vertex.getVertexId(), i, |
| vertex.eventHandler, |
| conf, |
| vertex.taskAttemptListener, |
| vertex.clock, |
| vertex.taskHeartbeatHandler, |
| vertex.appContext, |
| vertex.targetVertices.isEmpty(), |
| locHint, vertex.taskResource, |
| vertex.containerContext); |
| vertex.addTask(task); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Created task for vertex " + vertex.getVertexId() + ": " + |
| task.getTaskId()); |
| } |
| } |
| |
| } |
| |
| /** |
| * If the number of tasks are greater than the configured value |
| * throw an exception that will fail job initialization |
| */ |
| private void checkTaskLimits() { |
| // no code, for now |
| } |
| } // end of InitTransition |
| |
| // Temporary to maintain topological order while starting vertices. Not useful |
| // since there's not much difference between the INIT and RUNNING states. |
| public static class SourceVertexStartedTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventSourceVertexStarted startEvent = |
| (VertexEventSourceVertexStarted) event; |
| int distanceFromRoot = startEvent.getSourceDistanceFromRoot() + 1; |
| if(vertex.distanceFromRoot < distanceFromRoot) { |
| vertex.distanceFromRoot = distanceFromRoot; |
| } |
| vertex.numStartedSourceVertices++; |
| if (vertex.numStartedSourceVertices == vertex.sourceVertices.size()) { |
| // Consider inlining this. |
| LOG.info("Starting vertex: " + vertex.getVertexId() + |
| " with name: " + vertex.getName() + |
| " with distanceFromRoot: " + vertex.distanceFromRoot ); |
| vertex.eventHandler.handle(new VertexEvent(vertex.vertexId, |
| VertexEventType.V_START)); |
| } |
| } |
| } |
| |
| public static class StartTransition |
| implements SingleArcTransition<VertexImpl, VertexEvent> { |
| /** |
| * This transition executes in the event-dispatcher thread, though it's |
| * triggered in MRAppMaster's startJobs() method. |
| */ |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| vertex.startTime = vertex.clock.getTime(); |
| vertex.vertexScheduler.onVertexStarted(); |
| vertex.logJobHistoryVertexStartedEvent(); |
| |
| // TODO: Metrics |
| //job.metrics.runningJob(job); |
| |
| // default behavior is to start immediately. so send information about us |
| // starting to downstream vertices. If the connections/structure of this |
| // vertex is not fully defined yet then we could send this event later |
| // when we are ready |
| for (Vertex targetVertex : vertex.targetVertices.keySet()) { |
| vertex.eventHandler.handle( |
| new VertexEventSourceVertexStarted(targetVertex.getVertexId(), |
| vertex.distanceFromRoot)); |
| } |
| |
| } |
| } |
| |
| private void abortVertex(VertexStatus.State finalState) { |
| try { |
| committer.abortVertex(finalState); |
| } catch (IOException e) { |
| LOG.warn("Could not abort vertex, name=" + getName(), e); |
| } |
| |
| if (finishTime == 0) { |
| setFinishTime(); |
| } |
| } |
| |
| private void mayBeConstructFinalFullCounters() { |
| // Calculating full-counters. This should happen only once for the vertex. |
| synchronized (this.fullCountersLock) { |
| if (this.fullCounters != null) { |
| // Already constructed. Just return. |
| return; |
| } |
| this.constructFinalFullcounters(); |
| } |
| } |
| |
| @Private |
| public void constructFinalFullcounters() { |
| this.fullCounters = new TezCounters(); |
| for (Task t : this.tasks.values()) { |
| TezCounters counters = t.getCounters(); |
| this.fullCounters.incrAllCounters(counters); |
| } |
| } |
| |
| // Task-start has been moved out of InitTransition, so this arc simply |
| // hardcodes 0 for both map and reduce finished tasks. |
| private static class TerminateNewVertexTransition |
| implements SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventTermination vet = (VertexEventTermination) event; |
| vertex.trySetTerminationCause(vet.getTerminationCause()); |
| vertex.setFinishTime(); |
| vertex.addDiagnostic("Vertex received Kill in NEW state."); |
| vertex.finished(VertexState.KILLED); |
| } |
| } |
| |
| private static class TerminateInitedVertexTransition |
| implements SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventTermination vet = (VertexEventTermination) event; |
| vertex.trySetTerminationCause(vet.getTerminationCause()); |
| vertex.abortVertex(VertexStatus.State.KILLED); |
| vertex.addDiagnostic("Vertex received Kill in INITED state."); |
| vertex.finished(VertexState.KILLED); |
| } |
| } |
| |
| private static class VertexKilledTransition |
| implements SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| vertex.addDiagnostic("Vertex received Kill while in RUNNING state."); |
| VertexEventTermination vet = (VertexEventTermination) event; |
| VertexTerminationCause trigger = vet.getTerminationCause(); |
| switch(trigger){ |
| case DAG_KILL : vertex.enactKill(trigger, TaskTerminationCause.DAG_KILL); break; |
| case OTHER_VERTEX_FAILURE: vertex.enactKill(trigger, TaskTerminationCause.OTHER_VERTEX_FAILURE); break; |
| case OWN_TASK_FAILURE: vertex.enactKill(trigger, TaskTerminationCause.OTHER_TASK_FAILURE); break; |
| default://should not occur |
| throw new TezUncheckedException("VertexKilledTransition: event.terminationCause is unexpected: " + trigger); |
| } |
| |
| // TODO: Metrics |
| //job.metrics.endRunningJob(job); |
| } |
| } |
| |
| /** |
| * Here, the Vertex is being told that one of his source task-attempts |
| * completed. |
| */ |
| private static class SourceTaskAttemptCompletedEventTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventTaskAttemptCompleted completionEvent = |
| ((VertexEventSourceTaskAttemptCompleted) event).getCompletionEvent(); |
| LOG.info("Source task attempt completed for vertex: " + vertex.getVertexId() |
| + " attempt: " + completionEvent.getTaskAttemptId() |
| + " with state: " + completionEvent.getTaskAttemptState()); |
| |
| if (TaskAttemptStateInternal.SUCCEEDED.equals(completionEvent |
| .getTaskAttemptState())) { |
| vertex.numSuccessSourceAttemptCompletions++; |
| vertex.vertexScheduler.onSourceTaskCompleted(completionEvent |
| .getTaskAttemptId()); |
| } |
| |
| } |
| } |
| |
| private static class TaskAttemptCompletedEventTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventTaskAttemptCompleted completionEvent = |
| ((VertexEventTaskAttemptCompleted) event); |
| |
| // If different tasks were connected to different destination vertices |
| // then this would need to be sent via the edges |
| // Notify all target vertices |
| if (vertex.targetVertices != null) { |
| for (Vertex targetVertex : vertex.targetVertices.keySet()) { |
| vertex.eventHandler.handle( |
| new VertexEventSourceTaskAttemptCompleted( |
| targetVertex.getVertexId(), completionEvent) |
| ); |
| } |
| } |
| } |
| } |
| |
| private static class TaskCompletedTransition implements |
| MultipleArcTransition<VertexImpl, VertexEvent, VertexState> { |
| |
| @Override |
| public VertexState transition(VertexImpl vertex, VertexEvent event) { |
| boolean forceTransitionToKillWait = false; |
| vertex.completedTaskCount++; |
| LOG.info("Num completed Tasks: " + vertex.completedTaskCount); |
| VertexEventTaskCompleted taskEvent = (VertexEventTaskCompleted) event; |
| Task task = vertex.tasks.get(taskEvent.getTaskID()); |
| if (taskEvent.getState() == TaskState.SUCCEEDED) { |
| taskSucceeded(vertex, task); |
| } else if (taskEvent.getState() == TaskState.FAILED) { |
| vertex.enactKill(VertexTerminationCause.OWN_TASK_FAILURE, TaskTerminationCause.OTHER_TASK_FAILURE); |
| forceTransitionToKillWait = true; |
| taskFailed(vertex, task); |
| } else if (taskEvent.getState() == TaskState.KILLED) { |
| taskKilled(vertex, task); |
| } |
| |
| VertexState state = VertexImpl.checkVertexForCompletion(vertex); |
| if(state == VertexState.RUNNING && forceTransitionToKillWait){ |
| return VertexState.TERMINATING; |
| } |
| |
| return state; |
| } |
| |
| private void taskSucceeded(VertexImpl vertex, Task task) { |
| vertex.succeededTaskCount++; |
| // TODO Metrics |
| // job.metrics.completedTask(task); |
| } |
| |
| private void taskFailed(VertexImpl vertex, Task task) { |
| vertex.failedTaskCount++; |
| vertex.addDiagnostic("Task failed " + task.getTaskId()); |
| // TODO Metrics |
| //vertex.metrics.failedTask(task); |
| } |
| |
| private void taskKilled(VertexImpl vertex, Task task) { |
| vertex.killedTaskCount++; |
| // TODO Metrics |
| //job.metrics.killedTask(task); |
| } |
| } |
| |
| private static class TaskRescheduledTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| //succeeded task is restarted back |
| vertex.completedTaskCount--; |
| vertex.succeededTaskCount--; |
| } |
| } |
| |
| private static class TaskRescheduledAfterVertexSuccessTransition implements |
| MultipleArcTransition<VertexImpl, VertexEvent, VertexState> { |
| |
| @Override |
| public VertexState transition(VertexImpl vertex, VertexEvent event) { |
| if (vertex.committer instanceof NullVertexOutputCommitter) { |
| LOG.info(vertex.getVertexId() + " back to running due to rescheduling " |
| + ((VertexEventTaskReschedule)event).getTaskID()); |
| (new TaskRescheduledTransition()).transition(vertex, event); |
| // inform the DAG that we are re-running |
| vertex.eventHandler.handle(new DAGEventVertexReRunning(vertex.getVertexId())); |
| return VertexState.RUNNING; |
| } |
| |
| LOG.info(vertex.getVertexId() + " failed due to post-commit rescheduling of " |
| + ((VertexEventTaskReschedule)event).getTaskID()); |
| // terminate any running tasks |
| vertex.enactKill(VertexTerminationCause.OWN_TASK_FAILURE, |
| TaskTerminationCause.OWN_TASK_FAILURE); |
| // since the DAG thinks this vertex is completed it must be notified of |
| // an error |
| vertex.eventHandler.handle(new DAGEvent(vertex.getDAGId(), |
| DAGEventType.INTERNAL_ERROR)); |
| return VertexState.FAILED; |
| } |
| } |
| |
| private void addDiagnostic(String diag) { |
| diagnostics.add(diag); |
| } |
| |
| private static void checkEventSourceMetadata(Vertex vertex, EventMetaData sourceMeta) { |
| if (!sourceMeta.getTaskVertexName().equals(vertex.getName())) { |
| throw new TezUncheckedException("Bad routing of event" |
| + ", Event-vertex=" + sourceMeta.getTaskVertexName() |
| + ", Expected=" + vertex.getName()); |
| } |
| } |
| |
| private static class RouteEventTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| VertexEventRouteEvent rEvent = (VertexEventRouteEvent) event; |
| List<TezEvent> tezEvents = rEvent.getEvents(); |
| for(TezEvent tezEvent : tezEvents) { |
| LOG.info("Vertex: " + vertex.getName() + " routing event: " |
| + tezEvent.getEventType()); |
| EventMetaData sourceMeta = tezEvent.getSourceInfo(); |
| checkEventSourceMetadata(vertex, sourceMeta); |
| switch(tezEvent.getEventType()) { |
| case DATA_MOVEMENT_EVENT: |
| { |
| TezTaskAttemptID srcTaId = sourceMeta.getTaskAttemptID(); |
| DataMovementEvent dmEvent = (DataMovementEvent) tezEvent.getEvent(); |
| dmEvent.setVersion(srcTaId.getId()); |
| Edge destEdge = vertex.targetVertices.get(vertex.getDAG().getVertex( |
| sourceMeta.getEdgeVertexName())); |
| destEdge.sendTezEventToDestinationTasks(tezEvent); |
| } |
| break; |
| case INPUT_FAILED_EVENT: |
| { |
| TezTaskAttemptID srcTaId = sourceMeta.getTaskAttemptID(); |
| InputFailedEvent ifEvent = (InputFailedEvent) tezEvent.getEvent(); |
| ifEvent.setVersion(srcTaId.getId()); |
| Edge destEdge = vertex.targetVertices.get(vertex.getDAG().getVertex( |
| sourceMeta.getEdgeVertexName())); |
| destEdge.sendTezEventToDestinationTasks(tezEvent); |
| } |
| break; |
| case INPUT_READ_ERROR_EVENT: |
| { |
| Edge srcEdge = vertex.sourceVertices.get(vertex.getDAG().getVertex( |
| sourceMeta.getEdgeVertexName())); |
| srcEdge.sendTezEventToSourceTasks(tezEvent); |
| } |
| break; |
| case TASK_STATUS_UPDATE_EVENT: |
| { |
| TaskStatusUpdateEvent sEvent = |
| (TaskStatusUpdateEvent) tezEvent.getEvent(); |
| vertex.getEventHandler().handle( |
| new TaskAttemptEventStatusUpdate(sourceMeta.getTaskAttemptID(), |
| sEvent)); |
| } |
| break; |
| case TASK_ATTEMPT_COMPLETED_EVENT: |
| { |
| vertex.getEventHandler().handle( |
| new TaskAttemptEvent(sourceMeta.getTaskAttemptID(), |
| TaskAttemptEventType.TA_DONE)); |
| } |
| break; |
| case TASK_ATTEMPT_FAILED_EVENT: |
| { |
| TaskAttemptFailedEvent taskFailedEvent = |
| (TaskAttemptFailedEvent) tezEvent.getEvent(); |
| vertex.getEventHandler().handle( |
| new TaskAttemptEventAttemptFailed(sourceMeta.getTaskAttemptID(), |
| TaskAttemptEventType.TA_FAILED, |
| "Error: " + taskFailedEvent.getDiagnostics())); |
| } |
| break; |
| default: |
| throw new TezUncheckedException("Unhandled tez event type: " |
| + tezEvent.getEventType()); |
| } |
| } |
| } |
| } |
| |
| private static class InternalErrorTransition implements |
| SingleArcTransition<VertexImpl, VertexEvent> { |
| @Override |
| public void transition(VertexImpl vertex, VertexEvent event) { |
| LOG.error("Invalid event " + event.getType() + " on Vertex " |
| + vertex.getVertexId()); |
| vertex.eventHandler.handle(new DAGEventDiagnosticsUpdate( |
| vertex.getDAGId(), "Invalid event " + event.getType() |
| + " on Vertex " + vertex.getVertexId())); |
| vertex.setFinishTime(); |
| vertex.finished(VertexState.ERROR); |
| } |
| } |
| |
| @Override |
| public void setInputVertices(Map<Vertex, Edge> inVertices) { |
| this.sourceVertices = inVertices; |
| } |
| |
| @Override |
| public void setOutputVertices(Map<Vertex, Edge> outVertices) { |
| this.targetVertices = outVertices; |
| } |
| |
| @Override |
| public int compareTo(Vertex other) { |
| return this.vertexId.compareTo(other.getVertexId()); |
| } |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (this == obj) { |
| return true; |
| } |
| if (obj == null) { |
| return false; |
| } |
| if (getClass() != obj.getClass()) { |
| return false; |
| } |
| Vertex other = (Vertex) obj; |
| return this.vertexId.equals(other.getVertexId()); |
| } |
| |
| @Override |
| public int hashCode() { |
| final int prime = 11239; |
| return prime + prime * this.vertexId.hashCode(); |
| } |
| |
| @Override |
| public Map<Vertex, Edge> getInputVertices() { |
| return Collections.unmodifiableMap(this.sourceVertices); |
| } |
| |
| @Override |
| public Map<Vertex, Edge> getOutputVertices() { |
| return Collections.unmodifiableMap(this.targetVertices); |
| } |
| |
| @Override |
| public int getInputVerticesCount() { |
| return this.sourceVertices.size(); |
| } |
| |
| @Override |
| public int getOutputVerticesCount() { |
| return this.targetVertices.size(); |
| } |
| |
| @Override |
| public ProcessorDescriptor getProcessorDescriptor() { |
| return processorDescriptor; |
| } |
| |
| @Override |
| public DAG getDAG() { |
| return appContext.getCurrentDAG(); |
| } |
| |
| private TezDAGID getDAGId() { |
| return getDAG().getID(); |
| } |
| |
| private ApplicationAttemptId getApplicationAttemptId() { |
| return appContext.getApplicationAttemptId(); |
| } |
| |
| public Resource getTaskResource() { |
| return taskResource; |
| } |
| |
| @VisibleForTesting |
| String getProcessorName() { |
| return this.processorDescriptor.getClassName(); |
| } |
| |
| @VisibleForTesting |
| String getJavaOpts() { |
| return this.javaOpts; |
| } |
| |
| // TODO Eventually remove synchronization. |
| @Override |
| public synchronized List<InputSpec> getInputSpecList(int taskIndex) { |
| inputSpecList = new ArrayList<InputSpec>( |
| this.getInputVerticesCount()); |
| for (Entry<Vertex, Edge> entry : this.getInputVertices().entrySet()) { |
| InputSpec inputSpec = entry.getValue().getDestinationSpec(taskIndex); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("For vertex : " + this.getName() |
| + ", Using InputSpec : " + inputSpec); |
| } |
| // TODO DAGAM This should be based on the edge type. |
| inputSpecList.add(inputSpec); |
| } |
| return inputSpecList; |
| } |
| |
| // TODO Eventually remove synchronization. |
| @Override |
| public synchronized List<OutputSpec> getOutputSpecList(int taskIndex) { |
| if (this.outputSpecList == null) { |
| outputSpecList = new ArrayList<OutputSpec>(this.getOutputVerticesCount()); |
| for (Entry<Vertex, Edge> entry : this.getOutputVertices().entrySet()) { |
| OutputSpec outputSpec = entry.getValue().getSourceSpec(taskIndex); |
| outputSpecList.add(outputSpec); |
| } |
| } |
| return outputSpecList; |
| } |
| |
| @VisibleForTesting |
| VertexOutputCommitter getVertexOutputCommitter() { |
| return this.committer; |
| } |
| |
| @VisibleForTesting |
| // Only to be used for testing |
| void setVertexOutputCommitter(VertexOutputCommitter committer) { |
| this.committer = committer; |
| } |
| |
| @VisibleForTesting |
| VertexScheduler getVertexScheduler() { |
| return this.vertexScheduler; |
| } |
| |
| private static void logLocationHints(VertexLocationHint locationHint) { |
| Multiset<String> hosts = HashMultiset.create(); |
| Multiset<String> racks = HashMultiset.create(); |
| int counter = 0; |
| for (TaskLocationHint taskLocationHint : locationHint |
| .getTaskLocationHints()) { |
| StringBuilder sb = new StringBuilder(); |
| sb.append("Hosts: "); |
| for (String host : taskLocationHint.getDataLocalHosts()) { |
| hosts.add(host); |
| sb.append(host).append(", "); |
| } |
| sb.append("Racks: "); |
| for (String rack : taskLocationHint.getRacks()) { |
| racks.add(rack); |
| sb.append(rack).append(", "); |
| } |
| LOG.debug("Location: " + counter + " : " + sb.toString()); |
| counter++; |
| } |
| |
| LOG.debug("Host Counts"); |
| for (Multiset.Entry<String> host : hosts.entrySet()) { |
| LOG.debug("host: " + host.toString()); |
| } |
| |
| LOG.debug("Rack Counts"); |
| for (Multiset.Entry<String> rack : racks.entrySet()) { |
| LOG.debug("rack: " + rack.toString()); |
| } |
| } |
| } |
