def METHOD_NAME(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