Method run

test/parallel_testsuite.py:117–200

(self, result)

    test.is_parallel = True

  def run(self, result):
    # The 'spawn' method is used on windows and it can be useful to set this on
    # all platforms when debugging multiprocessing issues.  Without this we
    # default to 'fork' on unix which is better because global state is
    # inherited by the child process, but can lead to hard-to-debug windows-only
    # issues.
    # multiprocessing.set_start_method('spawn')

    # No need to worry about stdout/stderr buffering since are a not
    # actually running the test here, only setting the results.
    buffer = result.buffer
    result.buffer = False

    result.core_time = 0
    tests = self.get_sorted_tests()
    self.num_tests = self.countTestCases()
    contains_browser_test = any(test.is_browser_test() for test in tests)
    use_cores = cap_max_workers_in_pool(min(self.num_tests, num_cores()), contains_browser_test)
    errlog(f'Using {use_cores} parallel test processes')
    with multiprocessing.Manager() as manager:
      # Give each worker a unique ID.
      worker_id_counter = manager.Value('i', 0) # 'i' for integer, starting at 0
      worker_id_lock = manager.Lock()
      with multiprocessing.Pool(
        processes=use_cores,
        initializer=browser_common.init_worker,
        initargs=(worker_id_counter, worker_id_lock),
      ) as pool:
        if python_multiprocessing_structures_are_buggy():
          # When multiprocessing shared structures are buggy we don't support failfast
          # or the progress bar.
          allowed_failures_counter = None
          if self.max_failures < 2**31 - 1:
            errlog('The version of python being used is not compatible with --failfast and --max-failures options. See https://github.com/python/cpython/issues/71936')
            sys.exit(1)
        else:
          allowed_failures_counter = manager.Value('i', self.max_failures)

        results = []
        args = ((t, allowed_failures_counter, buffer) for t in tests)
        for res in pool.imap_unordered(run_test, args, chunksize=1):
          # results may be be None if # of allowed errors was exceeded
          # and the harness aborted.
          if res:
            if res.test_result not in {'success', 'skipped'} and allowed_failures_counter is not None:
              # Signal existing multiprocess pool runners so that they can exit early if needed.
              allowed_failures_counter.value -= 1
            res.integrate_result(result)
            results.append(res)

        # Send a task to each worker to tear down the browser and server. This
        # relies on the implementation detail in the worker pool that all workers
        # are cycled through once.
        num_tear_downs = sum(pool.apply(tear_down, ()) for _ in range(use_cores))
        # Assert the assumed behavior above hasn't changed.
        if num_tear_downs != use_cores and not buffer:
          errlog(f'Expected {use_cores} teardowns, got {num_tear_downs}')