code: def run_tests(
      selected_tests: List[ShardedTest],
      test_directory: str,
      options,
      failures: List[TestFailure],
  ) -> None:
      if len(selected_tests) == 0:
          return
  
      # parallel = in parallel with other files
      # serial = this file on it's own.  The file might still be run in parallel with itself (ex test_ops)
      selected_tests_parallel = [
          x for x in selected_tests if not must_serial(x)
      ]
      selected_tests_serial = [
          x for x in selected_tests if x not in selected_tests_parallel
      ]
  
      # See Note [ROCm parallel CI testing]
      pool = get_context("spawn").Pool(
          NUM_PROCS, maxtasksperchild=None if torch.version.hip else 1
      )
  
      # NB: This is a hack to make conftest.py available on CPP_TESTS_DIR. We should
      # see if the file could be turned into a full-fledge ptest plugin instead
      cpp_conftest_file = os.path.join(CPP_TESTS_DIR, "conftest.py")
      if (
          options.cpp
          and os.path.exists(CPP_TESTS_DIR)
          and os.path.isdir(CPP_TESTS_DIR)
          and not os.path.exists(cpp_conftest_file)
      ):
          # Take the conftest file from the test directory
          shutil.copy(
              os.path.join(test_directory, "conftest.py"),
              cpp_conftest_file,
          )
  
      def handle_error_messages(failure: Optional[TestFailure]):
          if failure is None:
              return False
          failures.append(failure)
          print_to_stderr(failure.message)
          return True
  
      def parallel_test_completion_callback(failure):
          test_failed = handle_error_messages(failure)
          if (
              test_failed
              and not options.continue_through_error
              and not RERUN_DISABLED_TESTS
          ):
              pool.terminate()
  
      try:
          os.environ["NUM_PARALLEL_PROCS"] = str(NUM_PROCS)
          for test in selected_tests_parallel:
              options_clone = copy.deepcopy(options)
              if can_run_in_pytest(test):
                  options_clone.pytest = True
              pool.apply_async(
                  run_test_module,
                  args=(test, test_directory, options_clone),
                  callback=parallel_test_completion_callback,
              )
          pool.close()
          pool.join()
          del os.environ["NUM_PARALLEL_PROCS"]
  
          if (
              not options.continue_through_error
              and not RERUN_DISABLED_TESTS
              and len(failures) != 0
          ):
              raise RuntimeError(
                  "\n".join(x.message for x in failures)
                  + "\n\nTip: You can keep running tests even on failure by "
                  "passing --keep-going to run_test.py.\n"
                  "If running on CI, add the 'keep-going' label to "
                  "your PR and rerun your jobs."
              )
  
          for test in selected_tests_serial:
              options_clone = copy.deepcopy(options)
              if can_run_in_pytest(test):
                  options_clone.pytest = True
              failure = run_test_module(
                  test, test_directory, options_clone
              )
              test_failed = handle_error_messages(failure)
              if (
                  test_failed
                  and not options.continue_through_error
                  and not RERUN_DISABLED_TESTS
              ):
                  raise RuntimeError(failure.message)
  
      finally:
          pool.terminate()
          pool.join()
  
      return
  
imports: import argparse
  import copy
  import glob
  import json
  import os
  import pathlib
  import re
  import shutil
  import signal
  import subprocess
  import sys
  import tempfile
  import time
  from collections import defaultdict
  from contextlib import ExitStack
  from datetime import datetime
  from typing import Any, cast, Dict, List, NamedTuple, Optional, Sequence, Tuple, Union
  import pkg_resources
  import torch
  import torch.distributed
  from torch.multiprocessing import current_process, get_context
  from torch.testing._internal.common_utils import FILE_SCHEMA, get_report_path, IS_CI, parser, retry_shell, set_cwd, shell, TEST_WITH_ASAN, TEST_WITH_ROCM, TEST_WITH_SLOW_GRADCHECK
  from tools.stats.import_test_stats import ADDITIONAL_CI_FILES_FOLDER, TEST_CLASS_TIMES_FILE, TEST_TIMES_FILE
  from tools.stats.upload_metrics import add_global_metric, emit_metric
  from tools.testing.target_determination.determinator import AggregatedHeuristics, get_prediction_confidences, get_test_prioritizations
  from tools.testing.test_run import TestRun
  from tools.testing.test_selections import calculate_shards, get_test_case_configs, NUM_PROCS, ShardedTest, THRESHOLD
label_desc: Use image node_id 29 for calling a global function with example usage: run_tests(selected_tests, test_directory, options, failures) and returns: True, False