Method _dist_run

tests/unit/common.py:295–340

(self, local_rank, num_procs, master_port, init_method, skip_msg="")

            self._launch_daemonic_procs(num_procs, init_method)

    def _dist_run(self, local_rank, num_procs, master_port, init_method, skip_msg=""):
        if dist.is_initialized():
            if get_accelerator().is_available():
                # local_rank might not match the rank in the previous run if you are reusing the environment
                get_accelerator().set_device(dist.get_rank())
        else:
            """ Initialize deepspeed.comm and execute the user function. """
            if self.set_dist_env:
                os.environ['MASTER_ADDR'] = '127.0.0.1'
                os.environ['MASTER_PORT'] = str(master_port)
                os.environ['LOCAL_RANK'] = str(local_rank)
                # NOTE: unit tests don't support multi-node so local_rank == global rank
                os.environ['RANK'] = str(local_rank)
                # In case of multiprocess launching LOCAL_SIZE should be same as WORLD_SIZE
                # DeepSpeed single node launcher would also set LOCAL_SIZE accordingly
                os.environ['LOCAL_SIZE'] = str(num_procs)
                os.environ['WORLD_SIZE'] = str(num_procs)

            # turn off NCCL logging if set
            os.environ.pop('NCCL_DEBUG', None)

            if get_accelerator().is_available():
                set_accelerator_visible()

            if get_accelerator().is_available():
                get_accelerator().set_device(local_rank)

            if self.init_distributed:
                deepspeed.init_distributed(dist_backend=self.backend,
                                           init_method=init_method,
                                           rank=local_rank,
                                           world_size=num_procs)
                dist.barrier()

        try:
            self.run(**self._fixture_kwargs)
        except BaseException as e:
            if isinstance(e, Skipped):
                if self.non_daemonic_procs:
                    skip_msg.put(e.msg)
                else:
                    skip_msg = e.msg
            else:
                raise e

        return skip_msg

    def _launch_with_file_store(self, request, world_size):
        tmpdir = request.getfixturevalue("tmpdir")