Assure thread affinity

CHANGES.txt

@@ -1,4 +1,10 @@
 Current
+New  : TestNG now guarantees thread-affinity for methods that use either preserve-order (or) dependsOnMethods (Krishnan Mahadevan) 
+Fixed: GITHUB-1185: DependsOnMethods made parallel class use several threads (Krishnan Mahadevan)
+Fixed: GITHUB-1173: Parallel="classes" executes methods of test class in different threads (Krishnan Mahadevan)
+Fixed: GITHUB-1066: Regression is in priority. It broke parallel mode (Krishnan Mahadevan)
+Fixed: GITHUB-1050: Parallel classes runs methods from one class in different threads, interleaves two classes in one thread (Krishnan Mahadevan)
+Fixed: GITHUB-89: parallel="classes" is not forcing test methods from the same testClass to be run in the same thread as it is suposed to (Krishnan Mahadevan)
 Fixed: GITHUB-1719: successPercentage does not work correctly for tests with dataProvider (Krishnan Mahadevan)
 Fixed: GITHUB-1241: Streamline Retry Analyzer usage when same test is run multiple times (Krishnan Mahadevan)
 Fixed: GITHUB-1777: ITestListener.onTestStart() not called after fail or skip from @BeforeMethod (Krishnan Mahadevan)

src/main/java/org/testng/TestRunner.java

@@ -57,6 +57,8 @@
 import com.google.inject.Injector;
 import com.google.inject.Module;
 
+import javax.annotation.Nonnull;
+
 import static org.testng.internal.MethodHelper.fixMethodsWithClass;
 
 /**
@@ -147,7 +149,6 @@ public class TestRunner
   private ClassMethodMap m_classMethodMap;
   private TestNGClassFinder m_testClassFinder;
   private IConfiguration m_configuration;
-  private IMethodInterceptor builtinInterceptor;
 
   public enum PriorityWeight {
     groupByInstance, preserveOrder, priority, dependsOnGroups, dependsOnMethods
@@ -212,7 +213,7 @@ private void init(IConfiguration configuration,
     setVerbose(test.getVerbose());
 
     boolean preserveOrder = test.getPreserveOrder();
-    builtinInterceptor = preserveOrder ? new PreserveOrderMethodInterceptor() : new InstanceOrderingMethodInterceptor();
+    IMethodInterceptor builtinInterceptor = preserveOrder ? new PreserveOrderMethodInterceptor() : new InstanceOrderingMethodInterceptor();
     m_methodInterceptors = new ArrayList<>();
     //Add the built in interceptor as the first interceptor. That way we let our users determine the final order
     //by plugging in their own custom interceptors as well.
@@ -370,7 +371,7 @@ private void initMethods() {
     List<ITestNGMethod> afterXmlTestMethods = Lists.newArrayList();
 
     ClassInfoMap classMap = new ClassInfoMap(m_testClassesFromXml);
-    m_testClassFinder= new TestNGClassFinder(classMap,Maps.<Class<?>, List<Object>>newHashMap(),
+    m_testClassFinder= new TestNGClassFinder(classMap,Maps.newHashMap(),
                                              m_configuration, this, m_dataProviderListeners);
     ITestMethodFinder testMethodFinder = new TestNGMethodFinder(m_runInfo, m_annotationFinder, comparator);
 
@@ -570,7 +571,7 @@ public void run() {
           IJUnitTestRunner tr= ClassHelper.createTestRunner(TestRunner.this);
           tr.setInvokedMethodListeners(m_invokedMethodListeners);
           try {
-            tr.run(tc, methods.toArray(new String[methods.size()]));
+            tr.run(tc, methods.toArray(new String[0]));
           }
           catch(Exception ex) {
             ex.printStackTrace();
@@ -596,13 +597,13 @@ public int getPriority() {
       }
 
       @Override
-      public int compareTo(IWorker<ITestNGMethod> other) {
+      public int compareTo(@Nonnull IWorker<ITestNGMethod> other) {
         return getPriority() - other.getPriority();
       }
     });
 
     runJUnitWorkers(workers);
-    m_allTestMethods= runMethods.toArray(new ITestNGMethod[runMethods.size()]);
+    m_allTestMethods= runMethods.toArray(new ITestNGMethod[0]);
   }
 
   /**
@@ -625,7 +626,7 @@ private void privateRun(XmlTest xmlTest) {
           GraphThreadPoolExecutor<ITestNGMethod> executor =
                   new GraphThreadPoolExecutor<>("test=" + xmlTest.getName(), graph, this,
                           threadCount, threadCount, 0, TimeUnit.MILLISECONDS,
-                          new LinkedBlockingQueue<Runnable>());
+                          new LinkedBlockingQueue<>());
           executor.run();
           try {
             long timeOut = m_xmlTest.getTimeOut(XmlTest.DEFAULT_TIMEOUT_MS);
@@ -675,7 +676,7 @@ private ITestNGMethod[] intercept(ITestNGMethod[] methods) {
     //so let's update the current classMethodMap object with the list of methods obtained from the interceptor.
     this.m_classMethodMap = new ClassMethodMap(result, null);
 
-    ITestNGMethod[] resultArray = result.toArray(new ITestNGMethod[result.size()]);
+    ITestNGMethod[] resultArray = result.toArray(new ITestNGMethod[0]);
 
     //Check if an interceptor had altered the effective test method count. If yes, then we need to
     //update our configurationGroupMethod object with that information.
@@ -822,13 +823,13 @@ public IResultMap getFailedButWithinSuccessPercentageTests() {
   @Override
   public String[] getIncludedGroups() {
     Map<String, String> ig= m_xmlMethodSelector.getIncludedGroups();
-    return ig.values().toArray(new String[ig.size()]);
+    return ig.values().toArray(new String[0]);
   }
 
   @Override
   public String[] getExcludedGroups() {
     Map<String, String> eg= m_xmlMethodSelector.getExcludedGroups();
-    return eg.values().toArray(new String[eg.size()]);
+    return eg.values().toArray(new String[0]);
   }
 
   @Override

src/main/java/org/testng/internal/DynamicGraph.java

@@ -83,6 +83,14 @@ public List<T> getFreeNodes() {
     return finalResult;
   }
 
+  public List<T> getDependenciesFor(T node) {
+    Map<T, Integer> data = m_edges.to(node);
+    if (data == null) {
+      return Lists.newArrayList();
+    }
+    return Lists.newArrayList(data.keySet());
+  }
+
   /**
    * Set the status for a set of nodes.
    */
@@ -130,6 +138,10 @@ public void setStatus(T node, Status status) {
 
         m_edges.removeNode(node);
         break;
+      case READY:
+        m_nodesReady.add(node);
+        m_nodesRunning.remove(node);
+        break;
       default:
         throw new IllegalArgumentException("Unsupported status: " + status);
     }
@@ -139,28 +151,30 @@ public void setStatus(T node, Status status) {
    * @return the number of nodes in this graph.
    */
   public int getNodeCount() {
-    int result = m_nodesReady.size() + m_nodesRunning.size() + m_nodesFinished.size();
-    return result;
+    return m_nodesReady.size() + m_nodesRunning.size() + m_nodesFinished.size();
   }
 
   public int getNodeCountWithStatus(Status status) {
+    return getNodesWithStatus(status).size();
+  }
+
+  public Set<T> getNodesWithStatus(Status status) {
     switch(status) {
-      case READY: return m_nodesReady.size();
-      case RUNNING: return m_nodesRunning.size();
-      case FINISHED: return m_nodesFinished.size();
+      case READY: return m_nodesReady;
+      case RUNNING: return m_nodesRunning;
+      case FINISHED: return m_nodesFinished;
       default: throw new IllegalArgumentException();
     }
+
   }
 
   @Override
   public String toString() {
-    StringBuilder result = new StringBuilder("[DynamicGraph ");
-    result.append("\n  Ready:").append(m_nodesReady);
-    result.append("\n  Running:").append(m_nodesRunning);
-    result.append("\n  Finished:").append(m_nodesFinished);
-    result.append("\n  Edges:\n").append(m_edges);
-    result.append("]");
-    return result.toString();
+    return "[DynamicGraph " + "\n  Ready:" + m_nodesReady +
+            "\n  Running:" + m_nodesRunning +
+            "\n  Finished:" + m_nodesFinished +
+            "\n  Edges:\n" + m_edges +
+            "]";
   }
 
   private static <T> String dotShortName(T t) {
@@ -255,8 +269,8 @@ Map<T, Integer> to(T node) {
      * Return the weight of the edge in the graph that is the reversed direction of edge. For example, if
      * edge a -> b exists, and edge b -> a is passed in, then return a -> b.
      *
-     * @param from
-     * @param to
+     * @param from - the from edge
+     * @param to - the to edge
      * @return the weight of the reversed edge or null if edge does not exist
      */
     private Integer findReversedEdge(T from, T to) {

src/main/java/org/testng/internal/RuntimeBehavior.java

@@ -8,6 +8,8 @@
 public final class RuntimeBehavior {
 
     public static final String TESTNG_LISTENERS_ALWAYSRUN = "testng.listeners.alwaysrun";
+    public static final String TESTNG_THREAD_AFFINITY = "testng.thread.affinity";
+    public static final String TESTNG_MODE_DRYRUN = "testng.mode.dryrun";
 
     private RuntimeBehavior() {
     }
@@ -17,7 +19,7 @@ private RuntimeBehavior() {
      * <code>false</code> otherwise.
      */
     public static boolean isDryRun() {
-        String value = System.getProperty("testng.mode.dryrun", "false");
+        String value = System.getProperty(TESTNG_MODE_DRYRUN, "false");
         return Boolean.parseBoolean(value);
     }
 
@@ -42,4 +44,16 @@ public static TimeZone getTimeZone() {
   public static boolean invokeListenersForSkippedTests() {
     return Boolean.parseBoolean(System.getProperty(TESTNG_LISTENERS_ALWAYSRUN, "false"));
   }
+
+  /**
+   * @return - <code>true</code> if we would like to enforce Thread affinity when dealing with the
+   *     below two variants of execution models:
+   *     <ul>
+   *       <li>Ordering priority
+   *       <li>Ordering by dependsOnMethods (will not work with dependency on multiple methods)
+   *     </ul>
+   */
+  public static boolean enforceThreadAffinity() {
+    return Boolean.parseBoolean(System.getProperty(TESTNG_THREAD_AFFINITY, "false"));
+  }
 }

src/main/java/org/testng/internal/TestMethodWorker.java

@@ -11,11 +11,14 @@
 import org.testng.internal.thread.graph.IWorker;
 import org.testng.xml.XmlSuite;
 
+import javax.annotation.Nonnull;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 
 /**
  * FIXME: reduce contention when this class is used through parallel invocation due to
@@ -39,6 +42,9 @@ public class TestMethodWorker implements IWorker<ITestNGMethod> {
   private final ClassMethodMap m_classMethodMap;
   private final ITestContext m_testContext;
   private final List<IClassListener> m_listeners;
+  private long currentThreadId;
+  private long threadIdToRunOn = -1;
+  private boolean completed = true;
 
   public TestMethodWorker(IInvoker invoker,
                           List<IMethodInstance> testMethods,
@@ -98,6 +104,12 @@ public String toString() {
    */
   @Override
   public void run() {
+    this.currentThreadId = Thread.currentThread().getId();
+    if (RuntimeBehavior.enforceThreadAffinity() && doesTaskHavePreRequistes() && currentThreadId != threadIdToRunOn) {
+      completed = false;
+      return;
+    }
+
     for (IMethodInstance testMthdInst : m_methodInstances) {
       ITestNGMethod testMethod = testMthdInst.getMethod();
       ITestClass testClass = testMethod.getTestClass();
@@ -113,6 +125,10 @@ public void run() {
     }
   }
 
+  private boolean doesTaskHavePreRequistes() {
+    return threadIdToRunOn != -1;
+  }
+
   protected void invokeTestMethods(ITestNGMethod tm, Object instance,
       ITestContext testContext)
   {
@@ -136,8 +152,6 @@ protected void invokeTestMethods(ITestNGMethod tm, Object instance,
 
   /**
    * Invoke the @BeforeClass methods if not done already
-   * @param testClass
-   * @param mi
    */
   protected void invokeBeforeClassMethods(ITestClass testClass, IMethodInstance mi) {
     // if no BeforeClass than return immediately
@@ -177,8 +191,6 @@ protected void invokeBeforeClassMethods(ITestClass testClass, IMethodInstance mi
 
   /**
    * Invoke the @AfterClass methods if not done already
-   * @param testClass
-   * @param mi
    */
   protected void invokeAfterClassMethods(ITestClass testClass, IMethodInstance mi) {
     // if no BeforeClass than return immediately
@@ -244,7 +256,7 @@ public List<ITestNGMethod> getTasks()
   }
 
   @Override
-  public int compareTo(IWorker<ITestNGMethod> other) {
+  public int compareTo(@Nonnull IWorker<ITestNGMethod> other) {
     return getPriority() - other.getPriority();
   }
 
@@ -257,6 +269,22 @@ public int getPriority() {
         ? m_methodInstances.get(0).getMethod().getPriority()
         : 0;
   }
+
+  @Override
+  public long getCurrentThreadId() {
+    return currentThreadId;
+  }
+
+  @Override
+  public void setThreadIdToRunOn(long threadIdToRunOn) {
+    this.threadIdToRunOn = threadIdToRunOn;
+  }
+
+  @Override
+  public boolean completed() {
+    return this.completed;
+  }
+
 }
 
 /**
@@ -265,8 +293,7 @@ public int getPriority() {
  */
 class SingleTestMethodWorker extends TestMethodWorker {
   private static final ConfigurationGroupMethods EMPTY_GROUP_METHODS =
-    new ConfigurationGroupMethods(new ITestNGMethod[0],
-        new HashMap<String, List<ITestNGMethod>>(), new HashMap<String, List<ITestNGMethod>>());
+    new ConfigurationGroupMethods(new ITestNGMethod[0], new HashMap<>(), new HashMap<>());
 
   public SingleTestMethodWorker(IInvoker invoker,
                                 IMethodInstance testMethod,
@@ -276,7 +303,7 @@ public SingleTestMethodWorker(IInvoker invoker,
                                 List<IClassListener> listeners)
   {
     super(invoker,
-            asList(testMethod),
+            Collections.singletonList(testMethod),
           suite,
           parameters,
           EMPTY_GROUP_METHODS,
@@ -285,11 +312,5 @@ public SingleTestMethodWorker(IInvoker invoker,
           listeners);
   }
 
-  //TODO Resorted to introducing this method to keep JDK7 happy. Can be removed once we move to JDK8
-  private static List<IMethodInstance> asList(IMethodInstance testMethod) {
-    List<IMethodInstance> methods = Lists.newLinkedList();
-    methods.add(testMethod);
-    return methods;
-  }
 
 }