class_docstr: Base class for distributed execution of functions/methods. Contains common
  methods needed for DistributedTest and DistributedFixture.
class_docstr_tok: ["Base","class","for","distributed","execution","of","functions\/methods",".","Contains","common","methods","needed","for","DistributedTest","and","DistributedFixture","."]
docstr: Initialize deepspeed.comm and execute the user function. 
docstr_tok: ["Initialize","deepspeed.comm","and","execute","the","user","function","."]
code: def _dist_init(self, local_rank, num_procs, skip_msg):
      """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"] = get_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)
          os.environ["WORLD_SIZE"] = str(num_procs)
  
      # turn off NCCL logging if set
      os.environ.pop("NCCL_DEBUG", None)
  
      set_cuda_visibile()
  
      if self.init_distributed:
          deepspeed.init_distributed(dist_backend=self.backend)
          dist.barrier()
  
      if torch.cuda.is_available():
          torch.cuda.set_device(local_rank)
  
      try:
          self.run(**self._fixture_kwargs)
      except BaseException as e:
          if isinstance(e, Skipped):
              skip_msg.put(e.msg)
          else:
              raise e
  
      if self.init_distributed or dist.is_initialized():
          # make sure all ranks finish at the same time
          dist.barrier()
          # tear down after test completes
          dist.destroy_process_group()
  
imports: import os
  import time
  import inspect
  from abc import ABC, abstractmethod
  from pathlib import Path
  import torch
  import torch.multiprocessing
  import deepspeed
  import deepspeed.comm
  from torch.multiprocessing import Process
  import pytest
  from _pytest.outcomes import Skipped
  from _pytest.fixtures import FixtureLookupError, FixtureFunctionMarker
label_desc: Use image node_id 5 for calling the DistributedExec obj's underlying member method code with example usage: obj._dist_init(local_rank, num_procs, skip_msg) without return types