tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bi...@apache.org
Subject tez git commit: TEZ-2011. InputReadyVertexManager not resilient to updates in parallelism (bikas) (cherry picked from commit b7268699e684fc4d1ebc00b47d8f1f7e4163bf97)
Date Sat, 31 Jan 2015 01:48:35 GMT
Repository: tez
Updated Branches:
  refs/heads/branch-0.6 8e5d18e63 -> 8cdd988b2


TEZ-2011. InputReadyVertexManager not resilient to updates in parallelism (bikas)
(cherry picked from commit b7268699e684fc4d1ebc00b47d8f1f7e4163bf97)


Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/8cdd988b
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/8cdd988b
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/8cdd988b

Branch: refs/heads/branch-0.6
Commit: 8cdd988b2549f0aec243a40d3bf7b78b2c9fb4e1
Parents: 8e5d18e
Author: Bikas Saha <bikas@apache.org>
Authored: Fri Jan 30 17:46:27 2015 -0800
Committer: Bikas Saha <bikas@apache.org>
Committed: Fri Jan 30 17:47:48 2015 -0800

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../apache/tez/dag/app/dag/impl/VertexImpl.java |  1 +
 .../vertexmanager/InputReadyVertexManager.java  | 66 ++++++++++++---
 .../TestInputReadyVertexManager.java            | 87 +++++++++++++++++++-
 4 files changed, 140 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/tez/blob/8cdd988b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 6ecc8e7..796ec9a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -125,6 +125,7 @@ TEZ-UI CHANGES (TEZ-8):
 Release 0.5.4: Unreleased
 
 ALL CHANGES:
+  TEZ-2011. InputReadyVertexManager not resilient to updates in parallelism 
   TEZ-1934. TestAMRecovery may fail due to the execution order is not determined.
   TEZ-1642. TestAMRecovery sometimes fail.
   TEZ-1931. Publish tez version info to Timeline.

http://git-wip-us.apache.org/repos/asf/tez/blob/8cdd988b/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 b1c93da..a467e4b 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
@@ -1560,6 +1560,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
       Preconditions.checkState(vertexToBeReconfiguredByManager, "doneReconfiguringVertex()
can be "
           + "invoked only after vertexReconfigurationPlanned() is invoked");
       this.vertexToBeReconfiguredByManager = false;
+      // TEZ-2015 VM may not have configured everything eg. input edge. maybeSendConfiguredEvent()
       if (completelyConfiguredSent.compareAndSet(false, true)) {
         // vertex already started and at that time this event was not sent. Send now.
         stateChangeNotifier.stateChanged(vertexId, new VertexStateUpdate(vertexName,

http://git-wip-us.apache.org/repos/asf/tez/blob/8cdd988b/tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/InputReadyVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/InputReadyVertexManager.java
b/tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/InputReadyVertexManager.java
index 8f30276..0e3a3ce 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/InputReadyVertexManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/dag/library/vertexmanager/InputReadyVertexManager.java
@@ -18,6 +18,8 @@
 
 package org.apache.tez.dag.library.vertexmanager;
 
+import java.util.Collection;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
 
@@ -32,11 +34,16 @@ import org.apache.tez.dag.api.TaskLocationHint;
 import org.apache.tez.dag.api.VertexManagerPlugin;
 import org.apache.tez.dag.api.VertexManagerPluginContext;
 import org.apache.tez.dag.api.VertexManagerPluginContext.TaskWithLocationHint;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
 import org.apache.tez.runtime.api.Event;
 import org.apache.tez.runtime.api.events.VertexManagerEvent;
 
+import com.google.common.base.Preconditions;
+import com.google.common.collect.LinkedListMultimap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Multimap;
 
 @Private
 public class InputReadyVertexManager extends VertexManagerPlugin {
@@ -48,6 +55,8 @@ public class InputReadyVertexManager extends VertexManagerPlugin {
   int oneToOneSrcTasksDoneCount[];
   TaskLocationHint oneToOneLocationHints[];
   int numOneToOneEdges;
+  int numSignalsToWaitFor;
+  Multimap<String, Integer> pendingCompletions = LinkedListMultimap.create();
 
   public InputReadyVertexManager(VertexManagerPluginContext context) {
     super(context);
@@ -67,12 +76,10 @@ public class InputReadyVertexManager extends VertexManagerPlugin {
     }
   }
   
-  @Override
-  public void initialize() {
-  }
-
-  @Override
-  public void onVertexStarted(Map<String, List<Integer>> completions) {
+  void start() {
+    if (!ready()) {
+      return;
+    }
     int numManagedTasks = getContext().getVertexNumTasks(getContext().getVertexName());
     LOG.info("Managing " + numManagedTasks + " tasks for vertex: " + getContext().getVertexName());
     taskIsStarted = new boolean[numManagedTasks];
@@ -117,24 +124,61 @@ public class InputReadyVertexManager extends VertexManagerPlugin {
       oneToOneLocationHints = new TaskLocationHint[oneToOneSrcTaskCount];
     }
 
-    for (Map.Entry<String, List<Integer>> entry : completions.entrySet()) {
+    for (Map.Entry<String, Collection<Integer>> entry :  pendingCompletions.asMap().entrySet())
{
       for (Integer task : entry.getValue()) {
         handleSourceTaskFinished(entry.getKey(), task);
       }
     }
   }
+  
+  boolean ready() {
+    int target = getContext().getInputVertexEdgeProperties().size() + 1;
+    Preconditions.checkState(numSignalsToWaitFor <= target);
+    return (numSignalsToWaitFor == target);
+  }
+  
+  @Override
+  public void initialize() {
+    Map<String, EdgeProperty> edges = getContext().getInputVertexEdgeProperties();
+    // wait for sources and self to start
+    numSignalsToWaitFor = 0;
+    for (String entry : edges.keySet()) {
+      getContext().registerForVertexStateUpdates(entry, EnumSet.of(VertexState.CONFIGURED));
+    }
+  }
+  
+  @Override
+  public synchronized void onVertexStateUpdated(VertexStateUpdate stateUpdate) throws Exception
{
+    numSignalsToWaitFor++;
+    LOG.info("Received configured signal from: " + stateUpdate.getVertexName() + 
+        " numConfiguredSources: " + numSignalsToWaitFor);
+    start();
+  }
 
   @Override
-  public void onSourceTaskCompleted(String srcVertexName, Integer taskId) {
-    handleSourceTaskFinished(srcVertexName, taskId);
+  public synchronized void onVertexStarted(Map<String, List<Integer>> completions)
{
+    for (Map.Entry<String, List<Integer>> entry : completions.entrySet()) {
+      pendingCompletions.putAll(entry.getKey(), entry.getValue());
+    }
+    numSignalsToWaitFor++;
+    start();
+  }
+
+  @Override
+  public synchronized void onSourceTaskCompleted(String srcVertexName, Integer taskId) {
+    if (ready()) {
+      handleSourceTaskFinished(srcVertexName, taskId);
+    } else {
+      pendingCompletions.put(srcVertexName, taskId);
+    }
   }
 
   @Override
-  public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) {
+  public synchronized void onVertexManagerEventReceived(VertexManagerEvent vmEvent) {
   }
 
   @Override
-  public void onRootVertexInitialized(String inputName,
+  public synchronized void onRootVertexInitialized(String inputName,
       InputDescriptor inputDescriptor, List<Event> events) {
   }
   

http://git-wip-us.apache.org/repos/asf/tez/blob/8cdd988b/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestInputReadyVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestInputReadyVertexManager.java
b/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestInputReadyVertexManager.java
index c6981ed..9a83a51 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestInputReadyVertexManager.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/dag/library/vertexmanager/TestInputReadyVertexManager.java
@@ -34,6 +34,8 @@ import org.apache.tez.dag.api.TezUncheckedException;
 import org.apache.tez.dag.api.VertexManagerPluginContext;
 import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
 import org.apache.tez.dag.api.VertexManagerPluginContext.TaskWithLocationHint;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -55,7 +57,7 @@ public class TestInputReadyVertexManager {
   }
 
   @Test (timeout=5000)
-  public void testBasicScatterGather() {
+  public void testBasicScatterGather() throws Exception {
     HashMap<String, EdgeProperty> mockInputVertices = 
         new HashMap<String, EdgeProperty>();
     String mockSrcVertexId1 = "Vertex1";
@@ -80,6 +82,9 @@ public class TestInputReadyVertexManager {
     
     InputReadyVertexManager manager = new InputReadyVertexManager(mockContext);
     manager.initialize();
+    // first source vertex configured
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    // then own vertex started
     manager.onVertexStarted(initialCompletions);
     manager.onSourceTaskCompleted(mockSrcVertexId1, 1);
     verify(mockContext, times(0)).scheduleVertexTasks(anyList());
@@ -89,7 +94,7 @@ public class TestInputReadyVertexManager {
   }
   
   @Test (timeout=5000)
-  public void testBasicOneToOne() {
+  public void testBasicOneToOne() throws Exception {
     HashMap<String, EdgeProperty> mockInputVertices = 
         new HashMap<String, EdgeProperty>();
     String mockSrcVertexId1 = "Vertex1";
@@ -114,7 +119,10 @@ public class TestInputReadyVertexManager {
     
     InputReadyVertexManager manager = new InputReadyVertexManager(mockContext);
     manager.initialize();
+    // first own vertex started
     manager.onVertexStarted(initialCompletions);
+    // then source vertex configured
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
     verify(mockContext, times(1)).scheduleVertexTasks(requestCaptor.capture());
     Assert.assertEquals(1, requestCaptor.getValue().size());
     Assert.assertEquals(0, requestCaptor.getValue().get(0).getTaskIndex().intValue());
@@ -139,9 +147,56 @@ public class TestInputReadyVertexManager {
     Assert.assertEquals(2, requestCaptor.getValue().get(0)
         .getTaskLocationHint().getAffinitizedTask().getTaskIndex());
   }
+  
+  @Test (timeout=5000)
+  public void testDelayedConfigureOneToOne() throws Exception {
+    HashMap<String, EdgeProperty> mockInputVertices = 
+        new HashMap<String, EdgeProperty>();
+    String mockSrcVertexId1 = "Vertex1";
+    EdgeProperty eProp1 = EdgeProperty.create(
+        EdgeProperty.DataMovementType.ONE_TO_ONE,
+        EdgeProperty.DataSourceType.PERSISTED,
+        SchedulingType.SEQUENTIAL,
+        OutputDescriptor.create("out"),
+        InputDescriptor.create("in"));
+    
+    String mockManagedVertexId = "Vertex";
+    
+    VertexManagerPluginContext mockContext = mock(VertexManagerPluginContext.class);
+    when(mockContext.getInputVertexEdgeProperties()).thenReturn(mockInputVertices);
+    when(mockContext.getVertexName()).thenReturn(mockManagedVertexId);
+    when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(3);
+    when(mockContext.getVertexNumTasks(mockSrcVertexId1)).thenReturn(3);
+    mockInputVertices.put(mockSrcVertexId1, eProp1);
+    
+    Map<String, List<Integer>> initialCompletions = Maps.newHashMap();
+    initialCompletions.put(mockSrcVertexId1, Collections.singletonList(0));
+    
+    InputReadyVertexManager manager = new InputReadyVertexManager(mockContext);
+    manager.initialize();
+    // first own vertex started
+    manager.onVertexStarted(initialCompletions);
+    verify(mockContext, times(0)).scheduleVertexTasks(requestCaptor.capture());
+    manager.onSourceTaskCompleted(mockSrcVertexId1, 1);
+    verify(mockContext, times(0)).scheduleVertexTasks(requestCaptor.capture());
+    // then source vertex configured. now we start
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    verify(mockContext, times(2)).scheduleVertexTasks(requestCaptor.capture());
+    Assert.assertEquals(2, requestCaptor.getAllValues().size());
+    Assert.assertEquals(1, requestCaptor.getValue().size());
+    Assert.assertEquals(1, requestCaptor.getValue().get(0).getTaskIndex().intValue());
+    manager.onSourceTaskCompleted(mockSrcVertexId1, 2);
+    verify(mockContext, times(3)).scheduleVertexTasks(requestCaptor.capture());
+    Assert.assertEquals(1, requestCaptor.getValue().size());
+    Assert.assertEquals(2, requestCaptor.getValue().get(0).getTaskIndex().intValue());
+    Assert.assertEquals(mockSrcVertexId1, requestCaptor.getValue().get(0)
+        .getTaskLocationHint().getAffinitizedTask().getVertexName());
+    Assert.assertEquals(2, requestCaptor.getValue().get(0)
+        .getTaskLocationHint().getAffinitizedTask().getTaskIndex());
+  }
 
   @Test (timeout=5000)
-  public void testComplex() {
+  public void testComplex() throws Exception {
     HashMap<String, EdgeProperty> mockInputVertices = 
         new HashMap<String, EdgeProperty>();
     String mockSrcVertexId1 = "Vertex1";
@@ -192,22 +247,43 @@ public class TestInputReadyVertexManager {
     
     Map<String, List<Integer>> initialCompletions = Maps.newHashMap();
     
-    // 1-1 sources do not match managed tasks
+    // 1-1 sources do not match managed tasks before vertex started
     when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(4);
     InputReadyVertexManager manager = new InputReadyVertexManager(mockContext);
     manager.initialize();
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId2, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId3, VertexState.CONFIGURED));
     try {
       manager.onVertexStarted(initialCompletions);
       Assert.assertTrue("Should have exception", false);
     } catch (TezUncheckedException e) {
       e.getMessage().contains("Managed task number must equal 1-1 source");
     }
+
+    // 1-1 sources do not match managed tasks after vertex started
+    when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(3);
+    manager = new InputReadyVertexManager(mockContext);
+    manager.initialize();
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId2, VertexState.CONFIGURED));
+    manager.onVertexStarted(initialCompletions);
+    when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(4);
+    try {
+      manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId3, VertexState.CONFIGURED));
+      Assert.assertTrue("Should have exception", false);
+    } catch (TezUncheckedException e) {
+      e.getMessage().contains("Managed task number must equal 1-1 source");
+    }
     
     // 1-1 sources do not match
     when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(3);
     when(mockContext.getVertexNumTasks(mockSrcVertexId3)).thenReturn(4);
     manager = new InputReadyVertexManager(mockContext);
     manager.initialize();
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId2, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId3, VertexState.CONFIGURED));
     try {
       manager.onVertexStarted(initialCompletions);
       Assert.assertTrue("Should have exception", false);
@@ -220,6 +296,9 @@ public class TestInputReadyVertexManager {
     when(mockContext.getVertexNumTasks(mockSrcVertexId3)).thenReturn(3);
     manager = new InputReadyVertexManager(mockContext);
     manager.initialize();
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId2, VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId3, VertexState.CONFIGURED));
     manager.onVertexStarted(initialCompletions);
     // all 1-1 0's done but not scheduled because v1 is not done
     manager.onSourceTaskCompleted(mockSrcVertexId3, 0);


Mime
View raw message