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","."]
code: def _launch_procs(self, num_procs):
      mp.set_start_method("forkserver", force=True)
      skip_msg = (
          mp.Queue()
      )  # Allows forked processes to share pytest.skip reason
      processes = []
      for local_rank in range(num_procs):
          p = Process(
              target=self._dist_init,
              args=(local_rank, num_procs, skip_msg),
          )
          p.start()
          processes.append(p)
  
      # Now loop and wait for a test to complete. The spin-wait here isn't a big
      # deal because the number of processes will be O(#GPUs) << O(#CPUs).
      any_done = False
      while not any_done:
          for p in processes:
              if not p.is_alive():
                  any_done = True
                  break
  
      # Wait for all other processes to complete
      for p in processes:
          p.join(DEEPSPEED_UNIT_WORKER_TIMEOUT)
  
      failed = [
          (rank, p)
          for rank, p in enumerate(processes)
          if p.exitcode != 0
      ]
      for rank, p in failed:
          # If it still hasn't terminated, kill it because it hung.
          if p.exitcode is None:
              p.terminate()
              pytest.fail(f"Worker {rank} hung.", pytrace=False)
          if p.exitcode < 0:
              pytest.fail(
                  f"Worker {rank} killed by signal {-p.exitcode}",
                  pytrace=False,
              )
          if p.exitcode > 0:
              pytest.fail(
                  f"Worker {rank} exited with code {p.exitcode}",
                  pytrace=False,
              )
  
      if not skip_msg.empty():
          # This assumed all skip messages are the same, it may be useful to
          # add a check here to assert all exit messages are equal
          pytest.skip(skip_msg.get())
  
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 4 for calling the DistributedExec obj's underlying member method code with example usage: obj._launch_procs(num_procs) without return types