[test] Refactor testrunner (4)

 - Removed duplication and unnecessary indirection from all suites testcfgs.
 - Introduced a more comprehensive context to cover both command context and other platform specific concerns.
 - Propagated above context to TestLoader to allow for test counting command execution on all platforms.
 - Wrapped original pool with another class to give it a new interface and allow injecting different implementations in the future.
 - Consolidated progress indicators under a single processor in the pipeline.
 - Consolidated result retention requirements calculation outside of pipeline chain.
 - Refactored LoaderProc and got it under tests.
 - Added some more tests for the standard runner.
 - Extracted BuildConfig class.


Bug: v8:12785
Change-Id: I87be040e91f792a983662bb5a10d55b36a14ea7f
Reviewed-on: https://chromium-review.googlesource.com/c/v8/v8/+/3701595Reviewed-by: Michael Achenbach <machenbach@chromium.org>
Commit-Queue: Liviu Rau <liviurau@chromium.org>
Cr-Commit-Position: refs/heads/main@{#81770}

test/benchmarks/testcfg.py

@@ -98,8 +98,9 @@ class TestLoader(testsuite.TestLoader):
 
 
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
     self.testroot = os.path.join(self.root, "data")
 
   def _test_loader_class(self):
@@ -135,7 +136,3 @@ class TestCase(testcase.D8TestCase):
 
   def _get_source_path(self):
     return os.path.join(self.suite.testroot, self.path + self._get_suffix())
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/bigint/testcfg.py

@@ -25,10 +25,10 @@ class TestLoader(testsuite.TestLoader):
       os.path.join(self.test_config.shell_dir, SHELL))
     if utils.IsWindows():
       shell += ".exe"
-    cmd = command.Command(
-      cmd_prefix=self.test_config.command_prefix,
-      shell=shell,
-      args=['--list'] + self.test_config.extra_flags)
+    cmd = self.ctx.command(
+        cmd_prefix=self.test_config.command_prefix,
+        shell=shell,
+        args=['--list'] + self.test_config.extra_flags)
     output = cmd.execute()
 
     if output.exit_code != 0:
@@ -57,7 +57,3 @@ class TestCase(testcase.TestCase):
 
   def get_shell(self):
     return SHELL
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/cctest/testcfg.py

@@ -44,7 +44,7 @@ class TestLoader(testsuite.TestLoader):
     shell = os.path.abspath(os.path.join(self.test_config.shell_dir, SHELL))
     if utils.IsWindows():
       shell += ".exe"
-    cmd = command.Command(
+    cmd = self.ctx.command(
         cmd_prefix=self.test_config.command_prefix,
         shell=shell,
         args=["--list"] + self.test_config.extra_flags)
@@ -87,7 +87,3 @@ class TestCase(testcase.TestCase):
 
   def _get_files_params(self):
     return [self.path]
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/debugger/testcfg.py

@@ -73,7 +73,3 @@ class TestCase(testcase.D8TestCase):
     if os.path.exists(base_path + self._get_suffix()):
       return base_path + self._get_suffix()
     return base_path + '.mjs'
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/debugging/testcfg.py

@@ -94,7 +94,3 @@ class TestSuite(testsuite.TestSuite):
 
   def _test_class(self):
     return TestCase
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/fuzzer/testcfg.py

@@ -65,7 +65,3 @@ class TestCase(testcase.TestCase):
   def get_shell(self):
     group, _ = self.path.split(os.path.sep, 1)
     return 'v8_simple_%s_fuzzer' % group
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/fuzzilli/testcfg.py

@@ -52,7 +52,3 @@ class TestCase(testcase.TestCase):
 
   def get_shell(self):
     return 'v8_fuzzilli_test'
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/inspector/testcfg.py

@@ -67,7 +67,3 @@ class TestCase(testcase.TestCase):
         self.expected_outcomes,
         os.path.join(self.suite.root, self.path) + EXPECTED_SUFFIX,
         self.suite.test_config.regenerate_expected_files)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/intl/testcfg.py

@@ -92,7 +92,3 @@ class TestCase(testcase.D8TestCase):
 
   def _get_source_path(self):
     return os.path.join(self.suite.root, self.path + self._get_suffix())
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/message/testcfg.py

@@ -100,7 +100,3 @@ class TestCase(testcase.D8TestCase):
                            self._expected_fail(),
                            self._base_path + '.out',
                            self.suite.test_config.regenerate_expected_files)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/mjsunit/testcfg.py

@@ -275,7 +275,3 @@ class CombinedTest(testcase.D8TestCase):
     # Combine flags from all status file entries.
     return self._get_combined_flags(
         test._get_statusfile_flags() for test in self._tests)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/mkgrokdump/testcfg.py

@@ -18,8 +18,9 @@ class TestLoader(testsuite.TestLoader):
 
 #TODO(tmrts): refactor the test creation logic to migrate to TestLoader
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
 
     v8_path = os.path.dirname(os.path.dirname(os.path.abspath(self.root)))
     self.expected_path = os.path.join(v8_path, 'tools', 'v8heapconst.py')
@@ -47,7 +48,3 @@ class TestCase(testcase.TestCase):
   @property
   def output_proc(self):
     return mkgrokdump.OutProc(self.expected_outcomes, self.suite.expected_path)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/mozilla/testcfg.py

@@ -75,8 +75,9 @@ class TestLoader(testsuite.JSTestLoader):
 
 
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
     self.test_root = os.path.join(self.root, "data")
     self._test_loader.test_root = self.test_root
 
@@ -117,7 +118,3 @@ class TestCase(testcase.D8TestCase):
     if self.path.endswith('-n'):
       return mozilla.NegOutProc(self.expected_outcomes)
     return mozilla.OutProc(self.expected_outcomes)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/test262/testcfg.py

@@ -123,8 +123,9 @@ class TestLoader(testsuite.JSTestLoader):
 
 
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
     self.test_root = os.path.join(self.root, *TEST_262_SUITE_PATH)
     # TODO: this makes the TestLoader mutable, refactor it.
     self._test_loader.test_root = self.test_root
@@ -252,7 +253,3 @@ class TestCase(testcase.D8TestCase):
     if self.expected_outcomes == outproc.OUTCOMES_PASS:
       return test262.PASS_NO_EXCEPTION
     return test262.NoExceptionOutProc(self.expected_outcomes)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/unittests/testcfg.py

@@ -24,10 +24,10 @@ class TestLoader(testsuite.TestLoader):
 
     output = None
     for i in range(3): # Try 3 times in case of errors.
-      cmd = command.Command(
-        cmd_prefix=self.test_config.command_prefix,
-        shell=shell,
-        args=['--gtest_list_tests'] + self.test_config.extra_flags)
+      cmd = self.ctx.command(
+          cmd_prefix=self.test_config.command_prefix,
+          shell=shell,
+          args=['--gtest_list_tests'] + self.test_config.extra_flags)
       output = cmd.execute()
       if output.exit_code == 0:
         break
@@ -89,7 +89,3 @@ class TestCase(testcase.TestCase):
       if os.path.exists(expectation_file):
         return [expectation_file]
     return []
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/wasm-api-tests/testcfg.py

@@ -24,10 +24,10 @@ class TestLoader(testsuite.TestLoader):
 
     output = None
     for i in range(3): # Try 3 times in case of errors.
-      cmd = command.Command(
-        cmd_prefix=self.test_config.command_prefix,
-        shell=shell,
-        args=['--gtest_list_tests'] + self.test_config.extra_flags)
+      cmd = self.ctx.command(
+          cmd_prefix=self.test_config.command_prefix,
+          shell=shell,
+          args=['--gtest_list_tests'] + self.test_config.extra_flags)
       output = cmd.execute()
       if output.exit_code == 0:
         break
@@ -77,7 +77,3 @@ class TestCase(testcase.TestCase):
 
   def get_shell(self):
     return "wasm_api_tests"
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/wasm-js/testcfg.py

@@ -38,8 +38,9 @@ class TestLoader(testsuite.JSTestLoader):
 
 
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
     self.mjsunit_js = os.path.join(os.path.dirname(self.root), "mjsunit",
                                    "mjsunit.js")
     self.test_root = os.path.join(self.root, "tests")
@@ -114,7 +115,3 @@ class TestCase(testcase.D8TestCase):
   def _get_source_path(self):
     # All tests are named `path/name.any.js`
     return os.path.join(self.suite.test_root, self.path + ANY_JS)
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/wasm-spec-tests/testcfg.py

@@ -28,8 +28,9 @@ class TestLoader(testsuite.JSTestLoader):
   pass
 
 class TestSuite(testsuite.TestSuite):
-  def __init__(self, *args, **kwargs):
-    super(TestSuite, self).__init__(*args, **kwargs)
+
+  def __init__(self, ctx, *args, **kwargs):
+    super(TestSuite, self).__init__(ctx, *args, **kwargs)
     self.test_root = os.path.join(self.root, "tests")
     self._test_loader.test_root = self.test_root
 
@@ -48,7 +49,3 @@ class TestCase(testcase.D8TestCase):
       if os.sep.join(['proposals', proposal['name']]) in self.path:
         return proposal['flags']
     return []
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

test/webkit/testcfg.py

@@ -96,7 +96,3 @@ class TestCase(testcase.D8TestCase):
     return webkit.OutProc(
         self.expected_outcomes,
         os.path.join(self.suite.root, self.path) + '-expected.txt')
-
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/run_perf.py

@@ -871,7 +871,7 @@ class AndroidPlatform(Platform):  # pragma: no cover
 
   def __init__(self, args):
     super(AndroidPlatform, self).__init__(args)
-    self.driver = android.android_driver(args.device)
+    self.driver = android.Driver.instance(args.device)
 
   def PreExecution(self):
     self.driver.set_high_perf_mode()

tools/testrunner/build_config.py

+# Copyright 2022 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.
+
+from testrunner.local import utils
+
+# Increase the timeout for these:
+SLOW_ARCHS = [
+    "arm", "arm64", "mips", "mipsel", "mips64", "mips64el", "s390", "s390x",
+    "riscv64", "loong64"
+]
+
+
+class BuildConfig(object):
+
+  def __init__(self, build_config, options):
+    self.options = options
+    # In V8 land, GN's x86 is called ia32.
+    if build_config['v8_target_cpu'] == 'x86':
+      self.arch = 'ia32'
+    else:
+      self.arch = build_config['v8_target_cpu']
+
+    self.asan = build_config['is_asan']
+    self.cfi_vptr = build_config['is_cfi']
+    self.control_flow_integrity = build_config['v8_control_flow_integrity']
+    self.concurrent_marking = build_config['v8_enable_concurrent_marking']
+    self.single_generation = build_config['v8_enable_single_generation']
+    self.dcheck_always_on = build_config['dcheck_always_on']
+    self.gcov_coverage = build_config['is_gcov_coverage']
+    self.is_android = build_config['is_android']
+    self.is_clang = build_config['is_clang']
+    self.is_debug = build_config['is_debug']
+    self.is_full_debug = build_config['is_full_debug']
+    self.msan = build_config['is_msan']
+    self.no_i18n = not build_config['v8_enable_i18n_support']
+    self.predictable = build_config['v8_enable_verify_predictable']
+    self.simulator_run = (
+        build_config['target_cpu'] != build_config['v8_target_cpu'])
+    self.tsan = build_config['is_tsan']
+    # TODO(machenbach): We only have ubsan not ubsan_vptr.
+    self.ubsan_vptr = build_config['is_ubsan_vptr']
+    self.verify_csa = build_config['v8_enable_verify_csa']
+    self.lite_mode = build_config['v8_enable_lite_mode']
+    self.pointer_compression = build_config['v8_enable_pointer_compression']
+    self.pointer_compression_shared_cage = build_config[
+        'v8_enable_pointer_compression_shared_cage']
+    self.shared_ro_heap = build_config['v8_enable_shared_ro_heap']
+    self.sandbox = build_config['v8_enable_sandbox']
+    self.third_party_heap = build_config['v8_enable_third_party_heap']
+    self.webassembly = build_config['v8_enable_webassembly']
+    self.dict_property_const_tracking = build_config[
+        'v8_dict_property_const_tracking']
+    # Export only for MIPS target
+    if self.arch in ['mips', 'mipsel', 'mips64', 'mips64el']:
+      self._mips_arch_variant = build_config['mips_arch_variant']
+      self.mips_use_msa = build_config['mips_use_msa']
+
+  @property
+  def use_sanitizer(self):
+    return (self.asan or self.cfi_vptr or self.msan or self.tsan or
+            self.ubsan_vptr)
+
+  @property
+  def no_js_shared_memory(self):
+    return (not self.shared_ro_heap) or (
+        self.pointer_compression and not self.pointer_compression_shared_cage)
+
+  @property
+  def is_mips_arch(self):
+    return self.arch in ['mipsel', 'mips', 'mips64', 'mips64el']
+
+  @property
+  def simd_mips(self):
+    return (self.is_mips_arch and self._mips_arch_variant == "r6" and
+            self.mips_use_msa)
+
+  @property
+  def mips_arch_variant(self):
+    return (self.is_mips_arch and self._mips_arch_variant)
+
+  @property
+  def no_simd_hardware(self):
+    # TODO(liviurau): Add some tests and refactor the logic here.
+    # We try to find all the reasons why we have no_simd.
+    no_simd_hardware = any(i in self.options.extra_flags for i in [
+        '--noenable-sse3', '--no-enable-sse3', '--noenable-ssse3',
+        '--no-enable-ssse3', '--noenable-sse4-1', '--no-enable-sse4_1'
+    ])
+
+    # Set no_simd_hardware on architectures without Simd enabled.
+    if self.arch == 'mips64el' or \
+       self.arch == 'mipsel':
+      no_simd_hardware = not self.simd_mips
+
+    if self.arch == 'loong64':
+      no_simd_hardware = True
+
+    # S390 hosts without VEF1 do not support Simd.
+    if self.arch == 's390x' and \
+       not self.simulator_run and \
+       not utils.IsS390SimdSupported():
+      no_simd_hardware = True
+
+    # Ppc64 processors earlier than POWER9 do not support Simd instructions
+    if self.arch == 'ppc64' and \
+       not self.simulator_run and \
+       utils.GuessPowerProcessorVersion() < 9:
+      no_simd_hardware = True
+
+    return no_simd_hardware
+
+  def timeout_scalefactor(self, initial_factor):
+    """Increases timeout for slow build configurations."""
+    factors = dict(
+        lite_mode=2,
+        predictable=4,
+        tsan=2,
+        use_sanitizer=1.5,
+        is_full_debug=4,
+    )
+    result = initial_factor
+    for k, v in factors.items():
+      if getattr(self, k, False):
+        result *= v
+    if self.arch in SLOW_ARCHS:
+      result *= 4.5
+    return result
+
+  def __str__(self):
+    attrs = [
+        'asan',
+        'cfi_vptr',
+        'control_flow_integrity',
+        'dcheck_always_on',
+        'gcov_coverage',
+        'msan',
+        'no_i18n',
+        'predictable',
+        'tsan',
+        'ubsan_vptr',
+        'verify_csa',
+        'lite_mode',
+        'pointer_compression',
+        'pointer_compression_shared_cage',
+        'sandbox',
+        'third_party_heap',
+        'webassembly',
+        'dict_property_const_tracking',
+    ]
+    detected_options = [attr for attr in attrs if getattr(self, attr, False)]
+    return '\n'.join(detected_options)

tools/testrunner/local/android.py

@@ -29,7 +29,15 @@ class CommandFailedException(Exception):
     self.output = output
 
 
-class _Driver(object):
+class Driver(object):
+  __instance = None
+
+  @staticmethod
+  def instance(device):
+    if not Driver.__instance:
+      Driver.__instance = Driver(device)
+    return Driver.__instance
+
   """Helper class to execute shell commands on an Android device."""
   def __init__(self, device=None):
     assert os.path.exists(ANDROID_DIR)
@@ -188,12 +196,3 @@ class _Driver(object):
     """Set device into default performance mode."""
     perf = perf_control.PerfControl(self.device)
     perf.SetDefaultPerfMode()
-
-
-_ANDROID_DRIVER = None
-def android_driver(device=None):
-  """Singleton access method to the driver class."""
-  global _ANDROID_DRIVER
-  if not _ANDROID_DRIVER:
-    _ANDROID_DRIVER = _Driver(device)
-  return _ANDROID_DRIVER

tools/testrunner/local/command.py

@@ -11,9 +11,10 @@ import sys
 import threading
 import time
 
-from ..local.android import (
-    android_driver, CommandFailedException, TimeoutException)
+from ..local.android import (Driver, CommandFailedException, TimeoutException)
 from ..objects import output
+from ..local.pool import DefaultExecutionPool, AbortException,\
+  taskkill_windows
 
 BASE_DIR = os.path.normpath(
     os.path.join(os.path.dirname(os.path.abspath(__file__)), '..' , '..', '..'))
@@ -29,12 +30,6 @@ def setup_testing():
   """
   signal.signal = lambda *_: None
 
-
-class AbortException(Exception):
-  """Indicates early abort on SIGINT, SIGTERM or internal hard timeout."""
-  pass
-
-
 @contextmanager
 def handle_sigterm(process, abort_fun, enabled):
   """Call`abort_fun` on sigterm and restore previous handler to prevent
@@ -208,22 +203,6 @@ class PosixCommand(BaseCommand):
     os.killpg(process.pid, signal.SIGKILL)
 
 
-def taskkill_windows(process, verbose=False, force=True):
-  force_flag = ' /F' if force else ''
-  tk = subprocess.Popen(
-      'taskkill /T%s /PID %d' % (force_flag, process.pid),
-      stdout=subprocess.PIPE,
-      stderr=subprocess.PIPE,
-  )
-  stdout, stderr = tk.communicate()
-  if verbose:
-    print('Taskkill results for %d' % process.pid)
-    print(stdout)
-    print(stderr)
-    print('Return code: %d' % tk.returncode)
-    sys.stdout.flush()
-
-
 class WindowsCommand(BaseCommand):
   def _start_process(self, **kwargs):
     # Try to change the error mode to avoid dialogs on fatal errors. Don't
@@ -325,54 +304,67 @@ class AndroidCommand(BaseCommand):
         duration,
     )
 
-
 Command = None
 
-class CommandContext():
-  def __init__(self, command):
+
+class DefaultOSContext():
+
+  def __init__(self, command, pool=None):
     self.command = command
+    self.pool = pool or DefaultExecutionPool()
 
   @contextmanager
-  def context(self, device):
+  def context(self, options):
     yield
 
-class AndroidContext():
+
+class AndroidOSContext(DefaultOSContext):
+
   def __init__(self):
-    self.command = AndroidCommand
+    super(AndroidOSContext, self).__init__(AndroidCommand)
 
   @contextmanager
-  def context(self, device):
+  def context(self, options):
     try:
-      AndroidCommand.driver = android_driver(device)
+      AndroidCommand.driver = Driver.instance(options.device)
       yield
     finally:
       AndroidCommand.driver.tear_down()
 
+
+# TODO(liviurau): Add documentation with diagrams to describe how context and
+# its components gets initialized and eventually teared down and how does it
+# interact with both tests and underlying platform specific concerns.
+def find_os_context_factory(target_os):
+  registry = dict(
+      android=AndroidOSContext,
+      windows=lambda: DefaultOSContext(WindowsCommand))
+  default = lambda: DefaultOSContext(PosixCommand)
+  return registry.get(target_os, default)
+
+
 @contextmanager
-def command_context(target_os, device):
-  factory = dict(
-    android=AndroidContext(),
-    windows=CommandContext(WindowsCommand),
-  )
-  context = factory.get(target_os, CommandContext(PosixCommand))
-  with context.context(device):
-    global Command
-    Command = context.command
-    yield
+def os_context(target_os, options):
+  factory = find_os_context_factory(target_os)
+  context = factory()
+  with context.context(options):
+    yield context
 
-# Deprecated : use command_context
+
+# Deprecated : use os_context
 def setup(target_os, device):
   """Set the Command class to the OS-specific version."""
   global Command
   if target_os == 'android':
-    AndroidCommand.driver = android_driver(device)
+    AndroidCommand.driver = Driver.instance(device)
     Command = AndroidCommand
   elif target_os == 'windows':
     Command = WindowsCommand
   else:
     Command = PosixCommand
 
-# Deprecated : use command_context
+
+# Deprecated : use os_context
 def tear_down():
   """Clean up after using commands."""
   if Command == AndroidCommand:

tools/testrunner/local/fake_testsuite/testcfg.py

@@ -26,6 +26,3 @@ class TestSuite(testsuite.TestSuite):
 
   def _test_class(self):
     return TestCase
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/testrunner/local/pool.py

@@ -3,14 +3,15 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
-from contextlib import contextmanager
-from multiprocessing import Process, Queue
+import collections
 import os
 import signal
+import subprocess
 import traceback
 
+from contextlib import contextmanager
+from multiprocessing import Process, Queue
 from queue import Empty
-from . import command
 from . import utils
 
 
@@ -31,6 +32,27 @@ def setup_testing():
   Process.pid = property(lambda self: None)
 
 
+def taskkill_windows(process, verbose=False, force=True):
+  force_flag = ' /F' if force else ''
+  tk = subprocess.Popen(
+      'taskkill /T%s /PID %d' % (force_flag, process.pid),
+      stdout=subprocess.PIPE,
+      stderr=subprocess.PIPE,
+  )
+  stdout, stderr = tk.communicate()
+  if verbose:
+    print('Taskkill results for %d' % process.pid)
+    print(stdout)
+    print(stderr)
+    print('Return code: %d' % tk.returncode)
+    sys.stdout.flush()
+
+
+class AbortException(Exception):
+  """Indicates early abort on SIGINT, SIGTERM or internal hard timeout."""
+  pass
+
+
 class NormalResult():
   def __init__(self, result):
     self.result = result
@@ -67,7 +89,7 @@ def Worker(fn, work_queue, done_queue,
     for args in iter(work_queue.get, "STOP"):
       try:
         done_queue.put(NormalResult(fn(*args, **kwargs)))
-      except command.AbortException:
+      except AbortException:
         # SIGINT, SIGTERM or internal hard timeout.
         break
       except Exception as e:
@@ -235,7 +257,7 @@ class Pool():
   def _terminate_processes(self):
     for p in self.processes:
       if utils.IsWindows():
-        command.taskkill_windows(p, verbose=True, force=False)
+        taskkill_windows(p, verbose=True, force=False)
       else:
         os.kill(p.pid, signal.SIGTERM)
 
@@ -299,3 +321,32 @@ class Pool():
         return MaybeResult.create_result(result.result)
       except Empty:
         return MaybeResult.create_heartbeat()
+
+
+# Global function for multiprocessing, because pickling a static method doesn't
+# work on Windows.
+def run_job(job, process_context):
+  return job.run(process_context)
+
+
+ProcessContext = collections.namedtuple('ProcessContext', ['result_reduction'])
+
+
+class DefaultExecutionPool():
+
+  def init(self, jobs, notify_fun):
+    self._pool = Pool(jobs, notify_fun=notify_fun)
+
+  def add_jobs(self, jobs):
+    self._pool.add(jobs)
+
+  def results(self, requirement):
+    return self._pool.imap_unordered(
+        fn=run_job,
+        gen=[],
+        process_context_fn=ProcessContext,
+        process_context_args=[requirement],
+    )
+
+  def abort(self):
+    self._pool.abort()

tools/testrunner/local/testsuite.py

@@ -81,7 +81,8 @@ class TestLoader(object):
   """Base class for loading TestSuite tests after applying test suite
   transformations."""
 
-  def __init__(self, suite, test_class, test_config, test_root):
+  def __init__(self, ctx, suite, test_class, test_config, test_root):
+    self.ctx = ctx
     self.suite = suite
     self.test_class = test_class
     self.test_config = test_config
@@ -246,12 +247,12 @@ def _load_testsuite_module(name, root):
 
 class TestSuite(object):
   @staticmethod
-  def Load(root, test_config, framework_name):
+  def Load(ctx, root, test_config, framework_name):
     name = root.split(os.path.sep)[-1]
     with _load_testsuite_module(name, root) as module:
-      return module.GetSuite(name, root, test_config, framework_name)
+      return module.TestSuite(ctx, name, root, test_config, framework_name)
 
-  def __init__(self, name, root, test_config, framework_name):
+  def __init__(self, ctx, name, root, test_config, framework_name):
     self.name = name  # string
     self.root = root  # string containing path
     self.test_config = test_config
@@ -259,8 +260,8 @@ class TestSuite(object):
     self.tests = None  # list of TestCase objects
     self.statusfile = None
 
-    self._test_loader = self._test_loader_class()(
-      self, self._test_class(), self.test_config, self.root)
+    self._test_loader = self._test_loader_class()(ctx, self, self._test_class(),
+                                                  self.test_config, self.root)
 
   def status_file(self):
     return "%s/%s.status" % (self.root, self.name)

tools/testrunner/local/testsuite_test.py

@@ -12,6 +12,7 @@ TOOLS_PATH = os.path.dirname(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 sys.path.append(TOOLS_PATH)
 
+from testrunner.local.command import DefaultOSContext, PosixCommand
 from testrunner.local.testsuite import TestSuite
 from testrunner.test_config import TestConfig
 
@@ -35,8 +36,9 @@ class TestSuiteTest(unittest.TestCase):
         verbose=False,
     )
 
-    self.suite = TestSuite.Load(self.test_root, self.test_config,
-                                "standard_runner")
+    self.suite = TestSuite.Load(
+        DefaultOSContext(PosixCommand), self.test_root, self.test_config,
+        "standard_runner")
 
   def testLoadingTestSuites(self):
     self.assertEqual(self.suite.name, "fake_testsuite")

tools/testrunner/num_fuzzer.py

@@ -20,7 +20,7 @@ from testrunner.testproc.loader import LoadProc
 from testrunner.utils import random_utils
 from testrunner.testproc.rerun import RerunProc
 from testrunner.testproc.timeout import TimeoutProc
-from testrunner.testproc.progress import ResultsTracker
+from testrunner.testproc.progress import ResultsTracker, ProgressProc
 from testrunner.testproc.shard import ShardProc
 
 
@@ -145,31 +145,33 @@ class NumFuzzer(base_runner.BaseTestRunner):
     })
     return variables
 
-  def _do_execute(self, tests, args):
+  def _do_execute(self, tests, args, ctx):
     loader = LoadProc(tests)
     combiner = CombinerProc.create(self.options)
     results = ResultsTracker.create(self.options)
-    execproc = ExecutionProc(self.options.j)
+    execproc = ExecutionProc(ctx, self.options.j)
     sigproc = self._create_signal_proc()
-    indicators = self._create_progress_indicators(
-      tests.test_count_estimate)
+    progress = ProgressProc(self.options, self.framework_name,
+                            tests.test_count_estimate)
     procs = [
-      loader,
-      NameFilterProc(args) if args else None,
-      StatusFileFilterProc(None, None),
-      # TODO(majeski): Improve sharding when combiner is present. Maybe select
-      # different random seeds for shards instead of splitting tests.
-      ShardProc.create(self.options),
-      ExpectationProc(),
-      combiner,
-      fuzzer.FuzzerProc.create(self.options),
-      sigproc,
-    ] + indicators + [
-      results,
-      TimeoutProc.create(self.options),
-      RerunProc.create(self.options),
-      execproc,
+        loader,
+        NameFilterProc(args) if args else None,
+        StatusFileFilterProc(None, None),
+        # TODO(majeski): Improve sharding when combiner is present. Maybe select
+        # different random seeds for shards instead of splitting tests.
+        ShardProc.create(self.options),
+        ExpectationProc(),
+        combiner,
+        fuzzer.FuzzerProc.create(self.options),
+        sigproc,
+        progress,
+        results,
+        TimeoutProc.create(self.options),
+        RerunProc.create(self.options),
+        execproc,
     ]
+    procs = [p for p in procs if p]
+
     self._prepare_procs(procs)
     loader.load_initial_tests()
 
@@ -181,8 +183,7 @@ class NumFuzzer(base_runner.BaseTestRunner):
     # processed.
     execproc.run()
 
-    for indicator in indicators:
-      indicator.finished()
+    progress.finished()
 
     print('>>> %d tests ran' % results.total)
     if results.failed:

tools/testrunner/objects/testcase.py

@@ -290,7 +290,7 @@ class TestCase(object):
   def only_standard_variant(self):
     return statusfile.NO_VARIANTS in self._statusfile_outcomes
 
-  def get_command(self):
+  def get_command(self, ctx):
     params = self._get_cmd_params()
     env = self._get_cmd_env()
     shell = self.get_shell()
@@ -298,7 +298,7 @@ class TestCase(object):
       shell += '.exe'
     shell_flags = self._get_shell_flags()
     timeout = self._get_timeout(params)
-    return self._create_cmd(shell, shell_flags + params, env, timeout)
+    return self._create_cmd(ctx, shell, shell_flags + params, env, timeout)
 
   def _get_cmd_params(self):
     """Gets all command parameters and combines them in the following order:
@@ -393,16 +393,16 @@ class TestCase(object):
   def _get_suffix(self):
     return '.js'
 
-  def _create_cmd(self, shell, params, env, timeout):
-    return command.Command(
-      cmd_prefix=self._test_config.command_prefix,
-      shell=os.path.abspath(os.path.join(self._test_config.shell_dir, shell)),
-      args=params,
-      env=env,
-      timeout=timeout,
-      verbose=self._test_config.verbose,
-      resources_func=self._get_resources,
-      handle_sigterm=True,
+  def _create_cmd(self, ctx, shell, params, env, timeout):
+    return ctx.command(
+        cmd_prefix=self._test_config.command_prefix,
+        shell=os.path.abspath(os.path.join(self._test_config.shell_dir, shell)),
+        args=params,
+        env=env,
+        timeout=timeout,
+        verbose=self._test_config.verbose,
+        resources_func=self._get_resources,
+        handle_sigterm=True,
     )
 
   def _parse_source_flags(self, source=None):

tools/testrunner/standard_runner.py

@@ -13,7 +13,7 @@ import sys
 import tempfile
 from testrunner.testproc.rerun import RerunProc
 from testrunner.testproc.timeout import TimeoutProc
-from testrunner.testproc.progress import ResultsTracker
+from testrunner.testproc.progress import ResultsTracker, ProgressProc
 from testrunner.testproc.shard import ShardProc
 
 # Adds testrunner to the path hence it has to be imported at the beggining.
@@ -22,7 +22,6 @@ sys.path.append(TOOLS_PATH)
 
 import testrunner.base_runner as base_runner
 
-from testrunner.local import utils
 from testrunner.local.variants import ALL_VARIANTS
 from testrunner.objects import predictable
 from testrunner.testproc.execution import ExecutionProc
@@ -284,71 +283,55 @@ class StandardTestRunner(base_runner.BaseTestRunner):
     """Create processor for sequencing heavy tests on swarming."""
     return SequenceProc(self.options.max_heavy_tests) if self.options.swarming else None
 
-  def _do_execute(self, tests, args):
+  def _do_execute(self, tests, args, ctx):
     jobs = self.options.j
 
     print('>>> Running with test processors')
     loader = LoadProc(tests, initial_batch_size=self.options.j * 2)
     results = ResultsTracker.create(self.options)
-    indicators = self._create_progress_indicators(
-        tests.test_count_estimate)
-
     outproc_factory = None
     if self.build_config.predictable:
       outproc_factory = predictable.get_outproc
-    execproc = ExecutionProc(jobs, outproc_factory)
+    execproc = ExecutionProc(ctx, jobs, outproc_factory)
     sigproc = self._create_signal_proc()
-
+    progress = ProgressProc(self.options, self.framework_name,
+                            tests.test_count_estimate)
     procs = [
-      loader,
-      NameFilterProc(args) if args else None,
-      StatusFileFilterProc(self.options.slow_tests, self.options.pass_fail_tests),
-      VariantProc(self._variants),
-      StatusFileFilterProc(self.options.slow_tests, self.options.pass_fail_tests),
-      self._create_predictable_filter(),
-      ShardProc.create(self.options),
-      self._create_seed_proc(),
-      self._create_sequence_proc(),
-      sigproc,
-    ] + indicators + [
-      results,
-      TimeoutProc.create(self.options),
-      RerunProc.create(self.options),
-      execproc,
+        loader,
+        NameFilterProc(args) if args else None,
+        VariantProc(self._variants),
+        StatusFileFilterProc(self.options.slow_tests,
+                             self.options.pass_fail_tests),
+        self._create_predictable_filter(),
+        ShardProc.create(self.options),
+        self._create_seed_proc(),
+        self._create_sequence_proc(),
+        sigproc,
+        progress,
+        results,
+        TimeoutProc.create(self.options),
+        RerunProc.create(self.options),
+        execproc,
     ]
+    procs = [p for p in procs if p]
 
     self._prepare_procs(procs)
     loader.load_initial_tests()
 
     # This starts up worker processes and blocks until all tests are
     # processed.
-    execproc.run()
-
-    for indicator in indicators:
-      indicator.finished()
-
-    if tests.test_count_estimate:
-      percentage = float(results.total) / tests.test_count_estimate * 100
-    else:
-      percentage = 0
+    requirement = max(p._requirement for p in procs)
+    execproc.run(requirement)
 
-    print (('>>> %d base tests produced %d (%d%s)'
-           ' non-filtered tests') % (
-        tests.test_count_estimate, results.total, percentage, '%'))
+    progress.finished()
 
-    print('>>> %d tests ran' % (results.total - results.remaining))
+    results.standard_show(tests)
 
-    exit_code = utils.EXIT_CODE_PASS
-    if results.failed:
-      exit_code = utils.EXIT_CODE_FAILURES
-    if not results.total:
-      exit_code = utils.EXIT_CODE_NO_TESTS
 
     if self.options.time:
       self._print_durations()
 
-    # Indicate if a SIGINT or SIGTERM happened.
-    return max(exit_code, sigproc.exit_code)
+    return sigproc.worst_exit_code(results)
 
   def _print_durations(self):
 

tools/testrunner/standard_runner_test.py

@@ -17,20 +17,22 @@ with different test suite extensions and build configurations.
 # TODO(machenbach): Coverage data from multiprocessing doesn't work.
 # TODO(majeski): Add some tests for the fuzzers.
 
+from collections import deque
 import os
 import sys
 import unittest
 from os.path import dirname as up
+from mock import patch
 
 TOOLS_ROOT = up(up(os.path.abspath(__file__)))
 sys.path.append(TOOLS_ROOT)
-
 from testrunner import standard_runner
 from testrunner import num_fuzzer
 from testrunner.utils.test_utils import (
-  temp_base,
-  TestRunnerTest,
-  with_json_output,
+    temp_base,
+    TestRunnerTest,
+    with_json_output,
+    FakeOSContext,
 )
 
 class StandardRunnerTest(TestRunnerTest):
@@ -215,6 +217,39 @@ class StandardRunnerTest(TestRunnerTest):
     # TODO(machenbach): Test some more implications of the auto-detected
     # options, e.g. that the right env variables are set.
 
+  def testLimitedPreloading(self):
+    result = self.run_tests('--progress=verbose', '--variants=default', '-j1',
+                            'sweet/*')
+    result.stdout_includes("sweet/mangoes default: PASS")
+    result.stdout_includes("sweet/cherries default: FAIL")
+    result.stdout_includes("sweet/apples default: FAIL")
+    result.stdout_includes("sweet/strawberries default: FAIL")
+    result.stdout_includes("sweet/bananas default: PASS")
+    result.stdout_includes("sweet/blackberries default: FAIL")
+    result.stdout_includes("sweet/raspberries default: PASS")
+
+  def testWithFakeContext(self):
+    with patch(
+        'testrunner.local.command.find_os_context_factory',
+        return_value=FakeOSContext):
+      result = self.run_tests(
+          '--progress=verbose',
+          'sweet/cherries',
+      )
+      result.stdout_includes('===>Starting stuff\n'
+                             '>>> Running tests for x64.release\n'
+                             '>>> Running with test processors\n')
+      result.stdout_includes('--- stdout ---\nfake stdout 1')
+      result.stdout_includes('--- stderr ---\nfake stderr 1')
+      result.stdout_includes('Command: fake_wrapper ')
+      result.stdout_includes(
+          '===\n'
+          '=== 1 tests failed\n'
+          '===\n'
+          '>>> 7 base tests produced 1 (14%) non-filtered tests\n'
+          '>>> 1 tests ran\n'
+          '<===Stopping stuff\n')
+
   def testSkips(self):
     """Test skipping tests in status file for a specific variant."""
     result = self.run_tests(
@@ -404,6 +439,48 @@ class StandardRunnerTest(TestRunnerTest):
     result.stdout_includes('3 tests ran')
     result.has_returncode(1)
 
+  def testHeavySequence(self):
+    """Test a configuration with 2 heavy tests.
+    One heavy test will get buffered before being run.
+    """
+    appended = 0
+    popped = 0
+
+    def mock_deque():
+
+      class MockQ():
+
+        def __init__(self):
+          self.buffer = deque()
+
+        def append(self, test):
+          nonlocal appended
+          self.buffer.append(test)
+          appended += 1
+
+        def popleft(self):
+          nonlocal popped
+          popped += 1
+          return self.buffer.popleft()
+
+        def __bool__(self):
+          return bool(self.buffer)
+
+      return MockQ()
+
+    # Swarming option will trigger a cleanup routine on the bot
+    def mock_kill():
+      pass
+
+    with patch('testrunner.testproc.sequence.deque', side_effect=mock_deque), \
+         patch('testrunner.testproc.util.kill_processes_linux', side_effect=mock_kill):
+      result = self.run_tests(
+          '--variants=default', '--swarming', 'fat', baseroot="testroot6")
+
+      result.has_returncode(1)
+      self.assertEqual(1, appended)
+      self.assertEqual(1, popped)
+
   def testRunnerFlags(self):
     """Test that runner-specific flags are passed to tests."""
     result = self.run_tests(

tools/testrunner/testdata/testroot1/test/sweet/testcfg.py

@@ -31,6 +31,3 @@ class TestCase(testcase.D8TestCase):
 
   def _get_files_params(self):
     return [self.name]
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/testrunner/testdata/testroot2/test/sweet/testcfg.py

@@ -29,6 +29,3 @@ class TestCase(testcase.TestCase):
 
   def _get_files_params(self):
     return [self.name]
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/testrunner/testdata/testroot3/test/sweet/testcfg.py

@@ -31,6 +31,3 @@ class TestCase(testcase.D8TestCase):
 
   def _get_files_params(self):
     return [self.name]
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/testrunner/testdata/testroot5/test/sweet/sweet.status

-# Copyright 2017 the V8 project authors. All rights reserved.
+# Copyright 2022 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.
 

tools/testrunner/testdata/testroot5/test/sweet/testcfg.py

-# Copyright 2017 the V8 project authors. All rights reserved.
+# Copyright 2022 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.
 
@@ -31,6 +31,3 @@ class TestCase(testcase.D8TestCase):
 
   def _get_files_params(self):
     return [self.name]
-
-def GetSuite(*args, **kwargs):
-  return TestSuite(*args, **kwargs)

tools/testrunner/testdata/testroot6/out/build/d8_mocked.py

+# Copyright 2017 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.
+"""
+Dummy d8 replacement. Just passes all test, except if 'berries' is in args.
+"""
+
+# for py2/py3 compatibility
+from __future__ import print_function
+
+import sys
+
+args = ' '.join(sys.argv[1:])
+print(args)
+# Let all berries fail.
+if 'berries' in args:
+  sys.exit(1)
+sys.exit(0)

tools/testrunner/testdata/testroot6/out/build/v8_build_config.json

+{
+  "current_cpu": "x64",
+  "dcheck_always_on": false,
+  "is_android": false,
+  "is_asan": false,
+  "is_cfi": false,
+  "is_clang": true,
+  "is_component_build": false,
+  "is_debug": false,
+  "is_full_debug": false,
+  "is_gcov_coverage": false,
+  "is_ubsan_vptr": false,
+  "is_msan": false,
+  "is_tsan": false,
+  "target_cpu": "x64",
+  "v8_current_cpu": "x64",
+  "v8_enable_i18n_support": true,
+  "v8_enable_verify_predictable": false,
+  "v8_target_cpu": "x64",
+  "v8_enable_concurrent_marking": true,
+  "v8_enable_verify_csa": false,
+  "v8_enable_lite_mode": false,
+  "v8_enable_pointer_compression": true,
+  "v8_enable_pointer_compression_shared_cage": true,
+  "v8_enable_sandbox": false,
+  "v8_enable_shared_ro_heap": true,
+  "v8_control_flow_integrity": false,
+  "v8_enable_single_generation": false,
+  "v8_enable_third_party_heap": false,
+  "v8_enable_webassembly": true,
+  "v8_dict_property_const_tracking": false
+}

tools/testrunner/testdata/testroot6/test/fat/fat.status

+# Copyright 2017 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.
+
+[
+[ALWAYS, {
+  'raspberries': [FAIL, HEAVY],
+  'strawberries': [PASS, ['mode == release', SLOW], ['mode == debug', NO_VARIANTS]],
+  'mangoes': [PASS, SLOW, HEAVY],
+
+  # Both cherries and apples are to test how PASS an FAIL from different
+  # sections are merged.
+  'cherries': [PASS, SLOW],
+  'apples': [FAIL],
+
+  # Unused rule.
+  'carrots': [PASS, FAIL],
+}],
+]

tools/testrunner/testdata/testroot6/test/fat/testcfg.py

+# Copyright 2017 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.
+"""
+Dummy test suite extension with some fruity tests.
+"""
+
+from testrunner.local import testsuite
+from testrunner.objects import testcase
+
+
+class TestLoader(testsuite.TestLoader):
+
+  def _list_test_filenames(self):
+    return [
+        'bananas',
+        'apples',
+        'cherries',
+        'mangoes',
+        'strawberries',
+        'blackberries',
+        'raspberries',
+    ]
+
+
+class TestSuite(testsuite.TestSuite):
+
+  def _test_loader_class(self):
+    return TestLoader
+
+  def _test_class(self):
+    return TestCase
+
+
+class TestCase(testcase.D8TestCase):
+
+  def get_shell(self):
+    return 'd8_mocked.py'
+
+  def _get_files_params(self):
+    return [self.name]

tools/testrunner/testproc/base.py

@@ -42,37 +42,11 @@ class TestProc(object):
     self._next_proc = None
     self._stopped = False
     self._requirement = DROP_RESULT
-    self._prev_requirement = None
-    self._reduce_result = lambda result: result
 
   def connect_to(self, next_proc):
     """Puts `next_proc` after itself in the chain."""
     next_proc._prev_proc = self
     self._next_proc = next_proc
-
-  def remove_from_chain(self):
-    if self._prev_proc:
-      self._prev_proc._next_proc = self._next_proc
-    if self._next_proc:
-      self._next_proc._prev_proc = self._prev_proc
-
-  def setup(self, requirement=DROP_RESULT):
-    """
-    Method called by previous processor or processor pipeline creator to let
-    the processors know what part of the result can be ignored.
-    """
-    self._prev_requirement = requirement
-    if self._next_proc:
-      self._next_proc.setup(max(requirement, self._requirement))
-
-    # Since we're not winning anything by droping part of the result we are
-    # dropping the whole result or pass it as it is. The real reduction happens
-    # during result creation (in the output processor), so the result is
-    # immutable.
-    if (self._prev_requirement < self._requirement and
-        self._prev_requirement == DROP_RESULT):
-      self._reduce_result = lambda _: None
-
   def next_test(self, test):
     """
     Method called by previous processor whenever it produces new test.
@@ -126,11 +100,7 @@ class TestProc(object):
 
   def _send_result(self, test, result):
     """Helper method for sending result to the previous processor."""
-    if not test.keep_output:
-      result = self._reduce_result(result)
     self._prev_proc.result_for(test, result)
-
-
 class TestProcObserver(TestProc):
   """Processor used for observing the data."""
   def __init__(self):

tools/testrunner/testproc/combiner.py

@@ -23,7 +23,7 @@ class CombinerProc(base.TestProc):
       count: how many tests to generate. 0 means infinite running
     """
     super(CombinerProc, self).__init__()
-
+    self._requirement = base.DROP_RESULT
     self._rng = rng
     self._min_size = min_group_size
     self._max_size = max_group_size
@@ -38,12 +38,6 @@ class CombinerProc(base.TestProc):
     # {suite name: instance of TestCombiner}
     self._combiners = {}
 
-  def setup(self, requirement=base.DROP_RESULT):
-    # Combiner is not able to pass results (even as None) to the previous
-    # processor.
-    assert requirement == base.DROP_RESULT
-    self._next_proc.setup(base.DROP_RESULT)
-
   def next_test(self, test):
     group_key = self._get_group_key(test)
     if not group_key:

tools/testrunner/testproc/execution.py

@@ -5,22 +5,8 @@
 import collections
 
 from . import base
-from ..local import pool
-
-
-# Global function for multiprocessing, because pickling a static method doesn't
-# work on Windows.
-def run_job(job, process_context):
-  return job.run(process_context)
-
-
-def create_process_context(result_reduction):
-  return ProcessContext(result_reduction)
-
 
 JobResult = collections.namedtuple('JobResult', ['id', 'result'])
-ProcessContext = collections.namedtuple('ProcessContext', ['result_reduction'])
-
 
 class Job(object):
   def __init__(self, test_id, cmd, outproc, keep_output):
@@ -42,23 +28,19 @@ class ExecutionProc(base.TestProc):
   sends results to the previous processor.
   """
 
-  def __init__(self, jobs, outproc_factory=None):
+  def __init__(self, ctx, jobs, outproc_factory=None):
     super(ExecutionProc, self).__init__()
-    self._pool = pool.Pool(jobs, notify_fun=self.notify_previous)
+    self.ctx = ctx
+    self.ctx.pool.init(jobs, notify_fun=self.notify_previous)
     self._outproc_factory = outproc_factory or (lambda t: t.output_proc)
     self._tests = {}
 
   def connect_to(self, next_proc):
-    assert False, 'ExecutionProc cannot be connected to anything'
-
-  def run(self):
-    it = self._pool.imap_unordered(
-        fn=run_job,
-        gen=[],
-        process_context_fn=create_process_context,
-        process_context_args=[self._prev_requirement],
-    )
-    for pool_result in it:
+    assert False, \
+        'ExecutionProc cannot be connected to anything' # pragma: no cover
+
+  def run(self, requirement=None):
+    for pool_result in self.ctx.pool.results(requirement):
       self._unpack_result(pool_result)
 
   def next_test(self, test):
@@ -66,20 +48,21 @@ class ExecutionProc(base.TestProc):
       return False
 
     test_id = test.procid
-    cmd = test.get_command()
+    cmd = test.get_command(self.ctx)
     self._tests[test_id] = test, cmd
 
     outproc = self._outproc_factory(test)
-    self._pool.add([Job(test_id, cmd, outproc, test.keep_output)])
+    self.ctx.pool.add_jobs([Job(test_id, cmd, outproc, test.keep_output)])
 
     return True
 
   def result_for(self, test, result):
-    assert False, 'ExecutionProc cannot receive results'
+    assert False, \
+        'ExecutionProc cannot receive results' # pragma: no cover
 
   def stop(self):
     super(ExecutionProc, self).stop()
-    self._pool.abort()
+    self.ctx.pool.abort()
 
   def _unpack_result(self, pool_result):
     if pool_result.heartbeat:

tools/testrunner/testproc/fuzzer.py

@@ -157,18 +157,12 @@ class FuzzerProc(base.TestProcProducer):
         set, processor passes None as analysis result to fuzzers
     """
     super(FuzzerProc, self).__init__('Fuzzer')
-
     self._rng = rng
     self._count = count
     self._fuzzer_configs = fuzzers
     self._disable_analysis = disable_analysis
     self._gens = {}
 
-  def setup(self, requirement=base.DROP_RESULT):
-    # Fuzzer is optimized to not store the results
-    assert requirement == base.DROP_RESULT
-    super(FuzzerProc, self).setup(requirement)
-
   def _next_test(self, test):
     if self.is_stopped:
       return False

tools/testrunner/testproc/loader.py

@@ -16,28 +16,21 @@ class LoadProc(base.TestProc):
 
 
   def load_initial_tests(self):
-    """
-    Args:
-      exec_proc: execution processor that the tests are being loaded into
-      initial_batch_size: initial number of tests to load
-    """
-    loaded_tests = 0
-    while loaded_tests < self.initial_batch_size:
-      try:
-        t = next(self.tests)
-      except StopIteration:
-        return
-
+    """Send an initial batch of tests down to executor"""
+    if not self.initial_batch_size:
+      return
+    to_load = self.initial_batch_size
+    for t in self.tests:
       if self._send_test(t):
-        loaded_tests += 1
+        to_load -= 1
+      if not to_load:
+        break
 
   def next_test(self, test):
-    assert False, 'Nothing can be connected to the LoadProc'
+    assert False, \
+        'Nothing can be connected to the LoadProc' # pragma: no cover
 
   def result_for(self, test, result):
-    try:
-      while not self._send_test(next(self.tests)):
-        pass
-    except StopIteration:
-      # No more tests to load.
-      pass
+    for t in self.tests:
+      if self._send_test(t):
+        break

tools/testrunner/testproc/loader_test.py

+#!/usr/bin/env python3
+# Copyright 2022 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 unittest
+from unittest.mock import patch
+from contextlib import contextmanager
+
+TOOLS_PATH = os.path.dirname(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.append(TOOLS_PATH)
+
+from testrunner.testproc.loader import LoadProc
+
+
+class LoadProcTest(unittest.TestCase):
+  # TODO(liviurau): test interaction between load_initialtests and results_for.
+
+  def setUp(self):
+    self.loader = LoadProc(iter(range(4)), 2)
+
+  @contextmanager
+  def send_test_return_values(self, l):
+
+    def do_pop(*args):
+      return l.pop()
+
+    with patch(
+        'testrunner.testproc.loader.LoadProc._send_test', side_effect=do_pop):
+      yield
+
+  def test_react_to_2_results(self):
+    with self.send_test_return_values([True] * 2):
+      self.loader.result_for(None, None)
+      self.loader.result_for(None, None)
+      self.assertEqual(2, next(self.loader.tests))
+
+  def test_react_to_result_but_fail_to_send(self):
+    with self.send_test_return_values([False] * 4):
+      self.loader.result_for(None, None)
+      self.assertEqual("empty", next(self.loader.tests, "empty"))
+
+  def test_init(self):
+    with self.send_test_return_values([True] * 4):
+      self.loader.load_initial_tests()
+      self.assertEqual(2, next(self.loader.tests))
+
+  def test_init_fully_filtered(self):
+    with self.send_test_return_values([False] * 4):
+      self.loader.load_initial_tests()
+      self.assertEqual("empty", next(self.loader.tests, "empty"))
+
+  def test_init_filter_1(self):
+    with self.send_test_return_values([True, False, True]):
+      self.loader.load_initial_tests()
+      self.assertEqual(3, next(self.loader.tests))
+
+  def test_init_infinity(self):
+    self.loader = LoadProc(iter(range(500)))
+    with self.send_test_return_values(([False] * 100) + ([True] * 400)):
+      self.loader.load_initial_tests()
+      self.assertEqual("empty", next(self.loader.tests, "empty"))
+
+  def test_init_0(self):
+    self.loader = LoadProc(iter(range(10)), 0)
+    with self.send_test_return_values([]):  # _send_test never gets called
+      self.loader.load_initial_tests()
+      self.assertEqual(0, next(self.loader.tests))
+
+
+if __name__ == '__main__':
+  unittest.main()

tools/testrunner/testproc/result.py

@@ -25,6 +25,17 @@ class Result(ResultBase):
     self.output = output
     self.cmd = cmd
 
+  def status(self):
+    if self.has_unexpected_output:
+      if not hasattr(self.output, "HasCrashed"):
+        raise Exception(type(self))
+      if self.output.HasCrashed():
+        return 'CRASH'
+      else:
+        return 'FAIL'
+    else:
+      return 'PASS'
+
 
 class GroupedResult(ResultBase):
   """Result consisting of multiple results. It can be used by processors that
@@ -95,3 +106,6 @@ class RerunResult(Result):
   @property
   def is_rerun(self):
     return True
+
+  def status(self):
+    return ' '.join(r.status() for r in self.results)

tools/testrunner/testproc/seed.py

@@ -18,6 +18,7 @@ class SeedProc(base.TestProcProducer):
       parallel_subtests: How many subtest of each test to run at the same time.
     """
     super(SeedProc, self).__init__('Seed')
+    self._requirement = base.DROP_RESULT
     self._count = count
     self._seed = seed
     self._last_idx = defaultdict(int)
@@ -26,12 +27,6 @@ class SeedProc(base.TestProcProducer):
     if count:
       self._parallel_subtests = min(self._parallel_subtests, count)
 
-  def setup(self, requirement=base.DROP_RESULT):
-    super(SeedProc, self).setup(requirement)
-
-    # SeedProc is optimized for dropping the result
-    assert requirement == base.DROP_RESULT
-
   def _next_test(self, test):
     is_loaded = False
     for _ in range(0, self._parallel_subtests):

tools/testrunner/testproc/sigproc.py

@@ -12,11 +12,6 @@ class SignalProc(base.TestProcObserver):
   def __init__(self):
     super(SignalProc, self).__init__()
     self.exit_code = utils.EXIT_CODE_PASS
-
-  def setup(self, *args, **kwargs):
-    super(SignalProc, self).setup(*args, **kwargs)
-    # It should be called after processors are chained together to not loose
-    # catched signal.
     signal.signal(signal.SIGINT, self._on_ctrlc)
     signal.signal(signal.SIGTERM, self._on_sigterm)
 
@@ -29,3 +24,8 @@ class SignalProc(base.TestProcObserver):
     print('>>> SIGTERM received, early abort...')
     self.exit_code = utils.EXIT_CODE_TERMINATED
     self.stop()
+
+  def worst_exit_code(self, results):
+    exit_code = results.exit_code()
+    # Indicate if a SIGINT or SIGTERM happened.
+    return max(exit_code, self.exit_code)

tools/testrunner/testproc/variant.py

@@ -23,17 +23,11 @@ class VariantProc(base.TestProcProducer):
 
   def __init__(self, variants):
     super(VariantProc, self).__init__('VariantProc')
+    self._requirement = base.DROP_RESULT
     self._next_variant = {}
     self._variant_gens = {}
     self._variants = variants
 
-  def setup(self, requirement=base.DROP_RESULT):
-    super(VariantProc, self).setup(requirement)
-
-    # VariantProc is optimized for dropping the result and it should be placed
-    # in the chain where it's possible.
-    assert requirement == base.DROP_RESULT
-
   def _next_test(self, test):
     gen = self._variants_gen(test)
     self._next_variant[test.procid] = gen

tools/testrunner/testproc/variant_test.py

@@ -86,7 +86,6 @@ class TestVariantProcLoading(unittest.TestCase):
   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.

tools/testrunner/utils/test_utils.py

@@ -11,10 +11,14 @@ import sys
 import tempfile
 import unittest
 
+from contextlib import contextmanager
 from dataclasses import dataclass
 from io import StringIO
 from os.path import dirname as up
 
+from testrunner.local.command import BaseCommand, DefaultOSContext
+from testrunner.objects import output
+
 TOOLS_ROOT = up(up(up(os.path.abspath(__file__))))
 sys.path.append(TOOLS_ROOT)
 
@@ -172,10 +176,60 @@ class TestRunnerTest(unittest.TestCase):
           sys_args.append('--infra-staging')
         else:
           sys_args.append('--no-infra-staging')
-        code = self.get_runner_class()(basedir=basedir).execute(sys_args)
+        runner = self.get_runner_class()(basedir=basedir)
+        code = runner.execute(sys_args)
         json_out = clean_json_output(json_out_path, basedir)
         return TestResult(stdout.getvalue(), stderr.getvalue(), code, json_out, self)
 
     def get_runner_class():
       """Implement to return the runner class"""
       return None
+
+
+class FakeOSContext(DefaultOSContext):
+
+  def __init__(self):
+    super(FakeOSContext, self).__init__(FakeCommand)
+
+  @contextmanager
+  def context(self, device):
+    print("===>Starting stuff")
+    yield
+    print("<===Stopping stuff")
+
+  def on_load(self):
+    print("<==>Loading stuff")
+
+
+class FakeCommand(BaseCommand):
+  counter = 0
+
+  def __init__(self,
+               shell,
+               args=None,
+               cmd_prefix=None,
+               timeout=60,
+               env=None,
+               verbose=False,
+               resources_func=None,
+               handle_sigterm=False):
+    f_prefix = ['fake_wrapper'] + cmd_prefix
+    super(FakeCommand, self).__init__(
+        shell,
+        args=args,
+        cmd_prefix=f_prefix,
+        timeout=timeout,
+        env=env,
+        verbose=verbose,
+        handle_sigterm=handle_sigterm)
+
+  def execute(self):
+    FakeCommand.counter += 1
+    return output.Output(
+        0,  #return_code,
+        False,  # TODO: Figure out timeouts.
+        f'fake stdout {FakeCommand.counter}',
+        f'fake stderr {FakeCommand.counter}',
+        -1,  # No pid available.
+        99,
+    )

tools/unittests/run_perf_test.py

@@ -3,7 +3,6 @@
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
-from collections import namedtuple
 import json
 import os
 import platform
@@ -644,7 +643,7 @@ class PerfTest(unittest.TestCase):
         'run_perf.AndroidPlatform.Run',
         return_value=(Output(stdout='Richards: 1.234\nDeltaBlue: 10657567\n'),
                       NULL_OUTPUT)).start()
-    mock.patch('testrunner.local.android._Driver', autospec=True).start()
+    mock.patch('testrunner.local.android.Driver', autospec=True).start()
     mock.patch(
         'run_perf.Platform.ReadBuildConfig',
         return_value={'is_android': True}).start()