[testrunner] remove recursive result calls in chain processors

Procs return the result by increasing recursion through result_for.

This CL eliminates that mechanism from the Processor interface and uses boolen
return values for sending tests to signal success or the failure to load the
test into the execution queue.

R=machenbach@chromium.org
CC=​​yangguo@chromium.org,sergiyb@chromium.org

Bug: v8:8174,v8:8731
Change-Id: I073a86ca84bcf88da11132b90013d4c8455bc61e
Reviewed-on: https://chromium-review.googlesource.com/c/1439239
Commit-Queue: Tamer Tas <tmrts@chromium.org>
Reviewed-by: Michael Achenbach <machenbach@chromium.org>
Reviewed-by: Sergiy Belozorov <sergiyb@chromium.org>
Cr-Commit-Position: refs/heads/master@{#59201}

tools/testrunner/PRESUBMIT.py

@@ -5,4 +5,6 @@
 def CheckChangeOnCommit(input_api, output_api):
   tests = input_api.canned_checks.GetUnitTestsInDirectory(
       input_api, output_api, '../unittests', whitelist=['run_tests_test.py$'])
+  tests = input_api.canned_checks.GetUnitTestsInDirectory(
+    input_api, output_api, 'testproc', whitelist=['variant_unittest.py$'])
   return input_api.RunTests(tests)

tools/testrunner/testproc/base.py

@@ -79,6 +79,8 @@ class TestProc(object):
     """
     Method called by previous processor whenever it produces new test.
     This method shouldn't be called by anyone except previous processor.
+    Returns a boolean value to signal whether the test was loaded into the
+    execution queue successfully or not.
     """
     raise NotImplementedError()
 
@@ -109,7 +111,7 @@ class TestProc(object):
 
   def _send_test(self, test):
     """Helper method for sending test to the next processor."""
-    self._next_proc.next_test(test)
+    return self._next_proc.next_test(test)
 
   def _send_result(self, test, result):
     """Helper method for sending result to the previous processor."""
@@ -126,7 +128,7 @@ class TestProcObserver(TestProc):
 
   def next_test(self, test):
     self._on_next_test(test)
-    self._send_test(test)
+    return self._send_test(test)
 
   def result_for(self, test, result):
     self._on_result_for(test, result)
@@ -158,7 +160,7 @@ class TestProcProducer(TestProc):
     self._name = name
 
   def next_test(self, test):
-    self._next_test(test)
+    return self._next_test(test)
 
   def result_for(self, subtest, result):
     self._result_for(subtest.origin, subtest, result)
@@ -190,9 +192,9 @@ class TestProcFilter(TestProc):
 
   def next_test(self, test):
     if self._filter(test):
-      self._send_result(test, SKIPPED)
-    else:
-      self._send_test(test)
+      return False
+
+    return self._send_test(test)
 
   def result_for(self, test, result):
     self._send_result(test, result)

tools/testrunner/testproc/combiner.py

@@ -45,9 +45,10 @@ class CombinerProc(base.TestProc):
     group_key = self._get_group_key(test)
     if not group_key:
       # Test not suitable for combining
-      return
+      return False
 
     self._groups[test.suite.name].add_test(group_key, test)
+    return True
 
   def _get_group_key(self, test):
     combiner =  self._get_combiner(test.suite)
@@ -66,17 +67,17 @@ class CombinerProc(base.TestProc):
 
   def _send_next_test(self):
     if self.is_stopped:
-      return
+      return False
 
     if self._count and self._current_num >= self._count:
-      return
+      return False
 
     combined_test = self._create_new_test()
     if not combined_test:
       # Not enough tests
-      return
+      return False
 
-    self._send_test(combined_test)
+    return self._send_test(combined_test)
 
   def _create_new_test(self):
     suite, combiner = self._select_suite()

tools/testrunner/testproc/execution.py

@@ -64,7 +64,7 @@ class ExecutionProc(base.TestProc):
 
   def next_test(self, test):
     if self.is_stopped:
-      return
+      return False
 
     test_id = test.procid
     cmd = test.get_command()
@@ -73,6 +73,8 @@ class ExecutionProc(base.TestProc):
     outproc = self._outproc_factory(test)
     self._pool.add([Job(test_id, cmd, outproc, test.keep_output)])
 
+    return True
+
   def result_for(self, test, result):
     assert False, 'ExecutionProc cannot receive results'
 

tools/testrunner/testproc/expectation.py

@@ -21,7 +21,8 @@ class ForgiveTimeoutProc(base.TestProcProducer):
     elif statusfile.TIMEOUT not in subtest.expected_outcomes:
       subtest.expected_outcomes = (
           subtest.expected_outcomes + [statusfile.TIMEOUT])
-    self._send_test(subtest)
+
+    return self._send_test(subtest)
 
   def _result_for(self, test, subtest, result):
     self._send_result(test, result)

tools/testrunner/testproc/fuzzer.py

@@ -69,14 +69,14 @@ class FuzzerProc(base.TestProcProducer):
 
   def _next_test(self, test):
     if self.is_stopped:
-      return
+      return False
 
     analysis_subtest = self._create_analysis_subtest(test)
     if analysis_subtest:
-      self._send_test(analysis_subtest)
-    else:
-      self._gens[test.procid] = self._create_gen(test)
-      self._try_send_next_test(test)
+      return self._send_test(analysis_subtest)
+
+    self._gens[test.procid] = self._create_gen(test)
+    return self._try_send_next_test(test)
 
   def _create_analysis_subtest(self, test):
     if self._disable_analysis:
@@ -100,6 +100,7 @@ class FuzzerProc(base.TestProcProducer):
         if result.has_unexpected_output:
           self._send_result(test, None)
           return
+
         self._gens[test.procid] = self._create_gen(test, result)
 
     self._try_send_next_test(test)
@@ -146,11 +147,11 @@ class FuzzerProc(base.TestProcProducer):
   def _try_send_next_test(self, test):
     if not self.is_stopped:
       for subtest in self._gens[test.procid]:
-        self._send_test(subtest)
-        return
+        if self._send_test(subtest):
+          return True
 
     del self._gens[test.procid]
-    self._send_result(test, None)
+    return False
 
   def _next_seed(self):
     seed = None

tools/testrunner/testproc/rerun.py

@@ -19,7 +19,7 @@ class RerunProc(base.TestProcProducer):
     self._rerun_total_left = rerun_max_total
 
   def _next_test(self, test):
-    self._send_next_subtest(test)
+    return self._send_next_subtest(test)
 
   def _result_for(self, test, subtest, result):
     # First result
@@ -52,7 +52,7 @@ class RerunProc(base.TestProcProducer):
 
   def _send_next_subtest(self, test, run=0):
     subtest = self._create_subtest(test, str(run + 1), keep_output=(run != 0))
-    self._send_test(subtest)
+    return self._send_test(subtest)
 
   def _finalize_test(self, test):
     del self._rerun[test.procid]

tools/testrunner/testproc/seed.py

@@ -34,12 +34,19 @@ class SeedProc(base.TestProcProducer):
     assert requirement == base.DROP_RESULT
 
   def _next_test(self, test):
+    is_loaded = False
     for _ in xrange(0, self._parallel_subtests):
-      self._try_send_next_test(test)
+      is_loaded |= self._try_send_next_test(test)
+
+    return is_loaded
 
   def _result_for(self, test, subtest, result):
     self._todo[test.procid] -= 1
-    self._try_send_next_test(test)
+    if not self._try_send_next_test(test):
+      if not self._todo.get(test.procid):
+        del self._last_idx[test.procid]
+        del self._todo[test.procid]
+        self._send_result(test, None)
 
   def _try_send_next_test(self, test):
     def create_subtest(idx):
@@ -49,10 +56,8 @@ class SeedProc(base.TestProcProducer):
     num = self._last_idx[test.procid]
     if not self._count or num < self._count:
       num += 1
-      self._send_test(create_subtest(num))
       self._todo[test.procid] += 1
       self._last_idx[test.procid] = num
-    elif not self._todo.get(test.procid):
-      del self._last_idx[test.procid]
-      del self._todo[test.procid]
-      self._send_result(test, None)
+      return self._send_test(create_subtest(num))
+
+    return False

tools/testrunner/testproc/variant.py

@@ -39,21 +39,22 @@ class VariantProc(base.TestProcProducer):
   def _next_test(self, test):
     gen = self._variants_gen(test)
     self._next_variant[test.procid] = gen
-    self._try_send_new_subtest(test, gen)
+    return self._try_send_new_subtest(test, gen)
 
   def _result_for(self, test, subtest, result):
     gen = self._next_variant[test.procid]
-    self._try_send_new_subtest(test, gen)
+    if not self._try_send_new_subtest(test, gen):
+      self._send_result(test, None)
 
   def _try_send_new_subtest(self, test, variants_gen):
     for variant, flags, suffix in variants_gen:
       subtest = self._create_subtest(test, '%s-%s' % (variant, suffix),
                                      variant=variant, flags=flags)
-      self._send_test(subtest)
-      return
+      if self._send_test(subtest):
+        return True
 
     del self._next_variant[test.procid]
-    self._send_result(test, None)
+    return False
 
   def _variants_gen(self, test):
     """Generator producing (variant, flags, procid suffix) tuples."""

tools/testrunner/testproc/variant_unittest.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python
+# Copyright 2019 the V8 project authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import os
+import sys
+import tempfile
+import unittest
+
+# Needed because the test runner contains relative imports.
+TOOLS_PATH = os.path.dirname(os.path.dirname(os.path.dirname(
+    os.path.abspath(__file__))))
+sys.path.append(TOOLS_PATH)
+
+from testrunner.testproc import base
+from testrunner.testproc.variant import VariantProc
+
+
+class FakeResultObserver(base.TestProcObserver):
+  def __init__(self):
+    super(FakeResultObserver, self).__init__()
+
+    self.results = set()
+
+  def result_for(self, test, result):
+    self.results.add((test, result))
+
+
+class FakeFilter(base.TestProcFilter):
+  def __init__(self, filter_predicate):
+    super(FakeFilter, self).__init__()
+
+    self._filter_predicate = filter_predicate
+
+    self.loaded = set()
+    self.call_counter = 0
+
+  def next_test(self, test):
+    self.call_counter += 1
+
+    if self._filter_predicate(test):
+      return False
+
+    self.loaded.add(test)
+    return True
+
+
+class FakeSuite(object):
+  def __init__(self, name):
+    self.name = name
+
+
+class FakeTest(object):
+  def __init__(self, procid):
+    self.suite = FakeSuite("fake_suite")
+    self.procid = procid
+
+    self.keep_output = False
+
+  def create_subtest(self, proc, subtest_id, **kwargs):
+    variant = kwargs['variant']
+
+    variant.origin = self
+    return variant
+
+
+class FakeVariantGen(object):
+  def __init__(self, variants):
+    self._variants = variants
+
+  def gen(self, test):
+    for variant in self._variants:
+      yield variant, [], "fake_suffix"
+
+
+class TestVariantProcLoading(unittest.TestCase):
+  def setUp(self):
+    self.test = FakeTest("test")
+
+  def _simulate_proc(self, variants):
+    """Expects the list of instantiated test variants to load into the
+    VariantProc."""
+    variants_mapping = {self.test: variants}
+
+    # Creates a Variant processor containing the possible types of test
+    # variants.
+    self.variant_proc = VariantProc(variants=["to_filter", "to_load"])
+    self.variant_proc._variant_gens = {
+      "fake_suite": FakeVariantGen(variants)}
+
+    # FakeFilter only lets tests passing the predicate to be loaded.
+    self.fake_filter = FakeFilter(
+      filter_predicate=(lambda t: t.procid == "to_filter"))
+
+    # FakeResultObserver to verify that VariantProc calls result_for correctly.
+    self.fake_result_observer = FakeResultObserver()
+
+    # Links up processors together to form a test processing pipeline.
+    self.variant_proc._prev_proc = self.fake_result_observer
+    self.fake_filter._prev_proc = self.variant_proc
+    self.variant_proc._next_proc = self.fake_filter
+
+    # Injects the test into the VariantProc
+    is_loaded = self.variant_proc.next_test(self.test)
+
+    # Verifies the behavioral consistency by using the instrumentation in
+    # FakeFilter
+    loaded_variants = list(self.fake_filter.loaded)
+    self.assertEqual(is_loaded, any(loaded_variants))
+    return self.fake_filter.loaded, self.fake_filter.call_counter
+
+  def test_filters_first_two_variants(self):
+    variants = [
+      FakeTest('to_filter'),
+      FakeTest('to_filter'),
+      FakeTest('to_load'),
+      FakeTest('to_load'),
+    ]
+    expected_load_results = {variants[2]}
+
+    load_results, call_count = self._simulate_proc(variants)
+
+    self.assertSetEqual(expected_load_results, load_results)
+    self.assertEqual(call_count, 3)
+
+  def test_stops_loading_after_first_successful_load(self):
+    variants = [
+      FakeTest('to_load'),
+      FakeTest('to_load'),
+      FakeTest('to_filter'),
+    ]
+    expected_load_results = {variants[0]}
+
+    loaded_tests, call_count = self._simulate_proc(variants)
+
+    self.assertSetEqual(expected_load_results, loaded_tests)
+    self.assertEqual(call_count, 1)
+
+  def test_return_result_when_out_of_variants(self):
+    variants = [
+      FakeTest('to_filter'),
+      FakeTest('to_load'),
+    ]
+
+    self._simulate_proc(variants)
+
+    self.variant_proc.result_for(variants[1], None)
+
+    expected_results = {(self.test, None)}
+
+    self.assertSetEqual(expected_results, self.fake_result_observer.results)
+
+  def test_return_result_after_running_variants(self):
+    variants = [
+      FakeTest('to_filter'),
+      FakeTest('to_load'),
+      FakeTest('to_load'),
+    ]
+
+    self._simulate_proc(variants)
+    self.variant_proc.result_for(variants[1], None)
+
+    self.assertSetEqual(set(variants[1:]), self.fake_filter.loaded)
+
+    self.variant_proc.result_for(variants[2], None)
+
+    expected_results = {(self.test, None)}
+    self.assertSetEqual(expected_results, self.fake_result_observer.results)
+
+if __name__ == '__main__':
+  unittest.main()

tools/v8_presubmit.py

@@ -652,6 +652,7 @@ def PyTests(workspace):
   for script in [
       join(workspace, 'tools', 'release', 'test_scripts.py'),
       join(workspace, 'tools', 'unittests', 'run_tests_test.py'),
+      join(workspace, 'tools', 'testrunner', 'testproc', 'variant_unittest.py'),
     ]:
     print 'Running ' + script
     result &= subprocess.call(