class_docstr: parallel.Execution is a tool for writing easy threading-based parallel tests.
  
  Execution.run is the main helper function, but there are a few magic getters that return
  correct thread-specific values when called from within an Execution.run-wrapped function.
  
  Example usage:
  
      SIZE = 10
      with parallel.Execution(SIZE) as pex:
          @pex.run
          def all_ranks():
              return pex.rank
  
          assert all_ranks == list(range(SIZE))
class_docstr_tok: ["parallel.Execution","is","a","tool","for","writing","easy","threading-based","parallel","tests",".","Execution.run","is","the","main","helper","function",",","but","there","are","a","few","magic","getters","that","return","correct","thread-specific","values","when","called","from","within","an","Execution.run-wrapped","function",".","Example","usage",":","SIZE","=","10","with","parallel.Execution","(","SIZE",")","as","pex",":","@","pex.run","def","all_ranks","(",")",":","return","pex.rank","assert","all_ranks","==","list","(","range","(","SIZE",")",")"]
docstr: Run the same function on one-thread-per-rank, assert there were no exceptions, and return
  the results from each rank.
  
  run can be used as a decorator or called directly.
docstr_tok: ["Run","the","same","function","on","one-thread-per-rank",",","assert","there","were","no","exceptions",",","and","return","the","results","from","each","rank",".","run","can","be","used","as","a","decorator","or","called","directly","."]
code: def run(self, fn: Callable[[], T]) -> List[T]:
      """
      Run the same function on one-thread-per-rank, assert there were no exceptions, and return
      the results from each rank.
  
      run can be used as a decorator or called directly.
      """
      results = [None] * self.size  # type: List
      errors = [None] * self.size  # type: List
      threads = []
  
      for cross_rank, local_rank in itertools.product(
          range(self.cross_size), range(self.local_size)
      ):
          rank = cross_rank * self.local_size + local_rank
  
          def _fn(rank: int, cross_rank: int, local_rank: int) -> None:
              thread_id = threading.get_ident()
              self._info[thread_id] = (rank, cross_rank, local_rank)
              try:
                  results[rank] = fn()
              # Catch anything, including a pytest.Fail (so we can preserve it).
              except BaseException as e:
                  # Print the error to stderr immediately, in case it results in a hang.
                  traceback.print_exc()
                  errors[rank] = (rank, e, sys.exc_info())
              finally:
                  del self._info[thread_id]
  
          threads.append(
              threading.Thread(
                  target=_fn, args=(rank, cross_rank, local_rank)
              )
          )
  
      for thread in threads:
          thread.start()
  
      for thread in threads:
          thread.join()
  
      # Filter empty errors.
      errors = [e for e in errors if e is not None]
      if len(errors) > 1:
          # In multi-errors situations, print all of them
          for rank, _, exc_info in errors:
              print(f"\nERROR ON RANK={rank}:", file=sys.stderr)
              traceback.print_exception(*exc_info)
          print(file=sys.stderr)
      if errors:
          # Reraise just the first exception.
          _, e, _ = errors[0]
          raise e
  
      return results
  
file_path: determined/harness/tests/parallel.py
imports: import contextlib
  import itertools
  import sys
  import threading
  import traceback
  from typing import Any, Callable, Dict, Iterator, List, Tuple, TypeVar
  import pytest
  from determined import core
label_desc: Use image node_id 4 for calling the Execution obj's underlying member method code with example usage: obj.run(fn) and returns: results