code: def launch_multi_processes(args: Config) -> None:
      """
      Launch multiple processes on a single node or multiple nodes.
  
      The overall logic can be summarized as the pseudo code below:
  
          if hostfile given:
              hostinfo = parse_hostfile(hostfile)
              hostinfo = include_or_exclude_hosts(hostinfo)
              launch_on_multi_nodes(hostinfo)
          elif hosts given:
              hostinfo = parse_hosts(hosts)
              launch_on_multi_nodes(hostinfo)
          else:
              launch_on_current_node()
  
      Args:
          args (Config): the arguments taken from command line
  
      """
      assert isinstance(args, Config)
  
      if args.nproc_per_node is None:
          click.echo("--nproc_per_node did not receive any value")
          exit()
  
      # cannot accept hosts and hostfile at the same time
      if args.host and args.hostfile:
          click.echo(
              "Error: hostfile and hosts are mutually exclusive, only one is required"
          )
  
      # check if hostfile is given
      if args.hostfile:
          device_pool = fetch_hostfile(
              args.hostfile, ssh_port=args.ssh_port
          )
          active_device_pool = parse_device_filter(
              device_pool, args.include, args.exclude
          )
  
          if args.num_nodes > 0:
              # only keep the first num_nodes to execute jobs
              updated_active_device_pool = HostInfoList()
              for count, hostinfo in enumerate(active_device_pool):
                  if args.num_nodes == count:
                      break
                  updated_active_device_pool.append(hostinfo)
              active_device_pool = updated_active_device_pool
      else:
          active_device_pool = None
  
      env = os.environ.copy()
  
      # use hosts if hostfile is not given
      if args.host and active_device_pool is None:
          active_device_pool = HostInfoList()
          host_list = args.host.strip().split(NODE_SEP)
          for hostname in host_list:
              hostinfo = HostInfo(hostname=hostname, port=args.ssh_port)
              active_device_pool.append(hostinfo)
  
      if not active_device_pool:
          # run on local node if not hosts or hostfile is given
          # add local node to host info list
          active_device_pool = HostInfoList()
          localhost_info = HostInfo(
              hostname="127.0.0.1", port=args.ssh_port
          )
          active_device_pool.append(localhost_info)
  
      # launch distributed processes
      runner = MultiNodeRunner()
      curr_path = os.path.abspath(".")
  
      # collect current path env
      env = dict()
      for k, v in os.environ.items():
          # do not support multi-line env var
          if v and "\n" not in v:
              env[k] = v
  
      # establish remote connection
      runner.connect(
          host_info_list=active_device_pool, workdir=curr_path, env=env
      )
  
      # overwrite master addr when num_nodes > 1 and not specified
      if (
          len(active_device_pool) > 1
          and args.master_addr == "127.0.0.1"
      ):
          args.master_addr = active_device_pool.hostinfo_list[
              0
          ].hostname
  
      # execute distributed launching command
      for node_id, hostinfo in enumerate(active_device_pool):
          cmd = get_launch_command(
              master_addr=args.master_addr,
              master_port=args.master_port,
              nproc_per_node=args.nproc_per_node,
              user_script=args.user_script,
              user_args=args.user_args,
              node_rank=node_id,
              num_nodes=len(active_device_pool),
              extra_launch_args=args.extra_launch_args,
          )
          runner.send(hostinfo=hostinfo, cmd=cmd)
  
      # start training
      msg_from_node = runner.recv_from_all()
      has_error = False
  
      # print node status
      click.echo("\n====== Training on All Nodes =====")
      for hostname, msg in msg_from_node.items():
          click.echo(f"{hostname}: {msg}")
  
          # check if a process failed
          if msg == "failure":
              has_error = True
  
      # stop all nodes
      runner.stop_all()
  
      # receive the stop status
      msg_from_node = runner.recv_from_all()
  
      # print node status
      click.echo("\n====== Stopping All Nodes =====")
      for hostname, msg in msg_from_node.items():
          click.echo(f"{hostname}: {msg}")
  
      # give the process an exit code
      # so that it behaves like a normal process
      if has_error:
          sys.exit(1)
      else:
          sys.exit(0)
  
file_path: colossalai/colossalai/cli/launcher/run.py
imports: import os
  import sys
  from typing import List
  import click
  import torch
  from packaging import version
  from colossalai.context import Config
  from .hostinfo import HostInfo, HostInfoList
  from .multinode_runner import MultiNodeRunner
label_desc: Use image node_id 4 for calling a global function with example usage: launch_multi_processes(args) without return types