code: def main(args=None):
      args = parse_args(args)
  
      if args.elastic_training:
          assert (
              args.master_addr != ""
          ), "Master Addr is required when elastic training is enabled"
  
      resource_pool = fetch_hostfile(args.hostfile)
  
      # respect CUDA_VISIBLE_DEVICES for a single node and no explicit resource filters
      cuda_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES", "")
      if not resource_pool and len(cuda_visible_devices):
          detected_str = (
              f"Detected CUDA_VISIBLE_DEVICES={cuda_visible_devices}"
          )
          if (
              len(args.include)
              or len(args.exclude)
              or args.num_nodes > 1
              or args.num_gpus > 0
          ):
              print(
                  f"{detected_str} but ignoring it because one or several of --include/--exclude/--num_gpus/--num_nodes cl args were used. If you want to use CUDA_VISIBLE_DEVICES don't pass any of these arguments to deepspeed."
              )
          else:
              args.include = f"localhost:{cuda_visible_devices}"
              print(f"{detected_str}: setting --include={args.include}")
          del os.environ["CUDA_VISIBLE_DEVICES"]
  
      if args.num_nodes >= 0 or args.num_gpus >= 0:
          if args.include != "" or args.exclude != "":
              raise ValueError(
                  "Cannot specify num_nodes/gpus with include/exclude"
              )
  
      multi_node_exec = True
      if not resource_pool:
          resource_pool = {}
          device_count = torch.cuda.device_count()
          if device_count == 0:
              raise RuntimeError(
                  "Unable to proceed, no GPU resources available"
              )
          resource_pool["localhost"] = device_count
          args.master_addr = "127.0.0.1"
          multi_node_exec = False
  
      if not multi_node_exec and args.num_nodes > 1:
          raise ValueError(
              "Num nodes is >1 but no extra nodes available via hostfile"
          )
  
      active_resources = parse_inclusion_exclusion(
          resource_pool, args.include, args.exclude
      )
      env = os.environ.copy()
  
      # validate that passwordless-ssh is workly properly with this hostfile
      if multi_node_exec and not args.no_ssh_check:
          first_host = list(active_resources.keys())[0]
          try:
              subprocess.check_call(
                  f"ssh -o PasswordAuthentication=no {first_host} hostname",
                  stderr=subprocess.DEVNULL,
                  stdout=subprocess.DEVNULL,
                  shell=True,
              )
          except subprocess.CalledProcessError:
              raise RuntimeError(
                  f"Using hostfile at {args.hostfile} but host={first_host} was not reachable via ssh. If you are running with a single node please remove {args.hostfile} or setup passwordless ssh."
              )
  
      if not args.master_addr:
          assert multi_node_exec
          first_host = list(active_resources.keys())[0]
          hostname_cmd = [f"ssh {first_host} hostname -I"]
          result = subprocess.check_output(hostname_cmd, shell=True)
          args.master_addr = result.decode("utf-8").split()[0]
          logger.info(
              f"Using IP address of {args.master_addr} for node {first_host}"
          )
  
      if args.autotuning != "":
          run_autotuning(args, active_resources)
          return
  
      if args.num_nodes > 0:
          updated_active_resources = collections.OrderedDict()
          for count, hostname in enumerate(active_resources.keys()):
              if args.num_nodes == count:
                  break
              updated_active_resources[hostname] = active_resources[
                  hostname
              ]
          active_resources = updated_active_resources
  
      if args.num_gpus > 0:
          updated_active_resources = collections.OrderedDict()
          for hostname in active_resources.keys():
              updated_active_resources[hostname] = list(
                  range(args.num_gpus)
              )
          active_resources = updated_active_resources
  
      if args.elastic_training:
          assert (
              not args.no_local_rank
          ), "--no_local_rank argument is not supported in Elastic training"
  
      # encode world info as base64 to make it easier to pass via command line
      world_info_base64 = encode_world_info(active_resources)
  
      multi_node_exec = args.force_multi or len(active_resources) > 1
  
      if not multi_node_exec:
          deepspeed_launch = [
              sys.executable,
              "-u",
              "-m",
              "deepspeed.launcher.launch",
              f"--world_info={world_info_base64}",
              f"--master_addr={args.master_addr}",
              f"--master_port={args.master_port}",
          ]
          if args.no_python:
              deepspeed_launch.append("--no_python")
          if args.module:
              deepspeed_launch.append("--module")
          if args.no_local_rank:
              deepspeed_launch.append("--no_local_rank")
          if args.save_pid:
              deepspeed_launch += ["--save_pid", f"{os.getpid()}"]
          if args.elastic_training:
              deepspeed_launch.append("--enable_elastic_training")
              deepspeed_launch.append(
                  f"--max_elastic_nodes={args.max_elastic_nodes}"
              )
              deepspeed_launch.append(
                  f"--min_elastic_nodes={args.min_elastic_nodes}"
              )
          cmd = deepspeed_launch + [args.user_script] + args.user_args
      else:
          args.launcher = args.launcher.lower()
          if args.launcher == PDSH_LAUNCHER:
              runner = PDSHRunner(args, world_info_base64)
          elif args.launcher == OPENMPI_LAUNCHER:
              runner = OpenMPIRunner(
                  args, world_info_base64, resource_pool
              )
          elif args.launcher == MVAPICH_LAUNCHER:
              runner = MVAPICHRunner(
                  args, world_info_base64, resource_pool
              )
          elif args.launcher == SLURM_LAUNCHER:
              runner = SlurmRunner(
                  args, world_info_base64, resource_pool
              )
          else:
              raise NotImplementedError(
                  f"Unknown launcher {args.launcher}"
              )
  
          if not runner.backend_exists():
              raise RuntimeError(
                  f"launcher '{args.launcher}' not installed."
              )
  
          curr_path = os.path.abspath(".")
          if "PYTHONPATH" in env:
              env["PYTHONPATH"] = curr_path + ":" + env["PYTHONPATH"]
          else:
              env["PYTHONPATH"] = curr_path
  
          exports = ""
          for var in env.keys():
              if any([var.startswith(name) for name in EXPORT_ENVS]):
                  runner.add_export(var, env[var])
  
          for environ_path in DEEPSPEED_ENVIRONMENT_PATHS:
              environ_file = os.path.join(
                  environ_path, DEEPSPEED_ENVIRONMENT_NAME
              )
              if os.path.isfile(environ_file):
                  with open(environ_file, "r") as fd:
                      for var in fd.readlines():
                          key, val = var.split("=", maxsplit=1)
                          runner.add_export(key, val)
  
          if args.launcher == PDSH_LAUNCHER:
              cmd, kill_cmd = runner.get_cmd(env, active_resources)
          else:
              cmd = runner.get_cmd(env, active_resources)
  
      logger.info(f"cmd = {' '.join(cmd)}")
      result = subprocess.Popen(cmd, env=env)
  
      def sigkill_handler(signum, frame):
          result.send_signal(signal.SIGINT)
          time.sleep(0.1)
          result.send_signal(signal.SIGTERM)
          result_kill = subprocess.Popen(kill_cmd, env=env)
          result_kill.wait()
          time.sleep(1)
          sys.exit(1)
  
      if args.launcher == PDSH_LAUNCHER:
          signal.signal(signal.SIGINT, sigkill_handler)
  
      result.wait()
  
      # In case of failure must propagate the error-condition back to the caller (usually shell). The
      # actual error and traceback should have been printed in the subprocess, so in order to avoid
      # unnecessary noise we just quietly exit here with the same code as the subprocess
      if result.returncode > 0:
          sys.exit(result.returncode)
  
imports: import os
  import sys
  import json
  import base64
  import argparse
  import subprocess
  import collections
  from copy import deepcopy
  import signal
  import time
  import torch.cuda
  from .multinode_runner import PDSHRunner, OpenMPIRunner, MVAPICHRunner, SlurmRunner
  from .constants import PDSH_LAUNCHER, OPENMPI_LAUNCHER, MVAPICH_LAUNCHER, SLURM_LAUNCHER
  from ..constants import TORCH_DISTRIBUTED_DEFAULT_PORT
  from ..nebula.constants import NEBULA_EXPORT_ENVS
  from ..utils import logger
  from ..autotuning import Autotuner
label_desc: Use image node_id 9 for calling a global function with example usage: main(args) without return types