code: def main(
      hvd_args: List[str], script: List[str], autohorovod: bool
  ) -> int:
      hvd_args = hvd_args or []
  
      info = det.get_cluster_info()
      assert info is not None, "must be run on-cluster"
      assert (
          info.task_type == "TRIAL"
      ), f'must be run with task_type="TRIAL", not "{info.task_type}"'
  
      # When --autohorovod was set, detect single-slot and zero-slot trials.
      if (
          autohorovod
          and len(info.container_addrs) == 1
          and len(info.slot_ids) <= 1
      ):
          p = subprocess.Popen(script)
          with det.util.forward_signals(p):
              return p.wait()
  
      # Hack: get the resources id from the environment.
      resources_id = os.environ.get("DET_RESOURCES_ID")
      assert (
          resources_id is not None
      ), "Unable to run with DET_RESOURCES_ID unset"
  
      # Hack: read the full config.  The experiment config is not a stable API!
      experiment_config = info.trial._config
  
      debug = experiment_config.get("debug", False)
      if debug:
          logging.getLogger().setLevel(logging.DEBUG)
  
      # TODO: refactor websocket and profiling to to not use the cli_cert.
      cert = certs.default_load(info.master_url)
      certs.cli_cert = cert
  
      # The launch layer should provide the chief_ip to the training code, so that the training code
      # can function with a different launch layer in a different environment.  Inside Determined, the
      # easiest way to get the chief_ip is with container_addrs.
      chief_ip = info.container_addrs[0]
  
      # Chief IP is set as an environment variable to support nested launch layers
      os.environ["DET_CHIEF_IP"] = chief_ip
  
      if info.container_rank > 0:
          # Non-chief machines just run sshd.
  
          # Mark sshd containers as daemon resources that the master should kill when all non-daemon
          # containers (horovodrun, in this case) have exited.
          api.post(
              info.master_url,
              path=f"/api/v1/allocations/{info.allocation_id}/resources/{resources_id}/daemon",
              cert=cert,
          )
  
          pid_server_cmd, run_sshd_command = create_sshd_worker_cmd(
              info.allocation_id, len(info.slot_ids), debug=debug
          )
  
          logger.debug(
              f"Non-chief [{info.container_rank}] training process launch "
              f"command: {run_sshd_command}."
          )
          p = subprocess.Popen(pid_server_cmd + run_sshd_command)
          with det.util.forward_signals(p):
              return p.wait()
  
      # Chief machine waits for every worker's sshd to be available.  All machines should be pretty
      # close to in-step by now because all machines just finished synchronizing rendezvous info.
      deadline = time.time() + 20
      for peer_addr in info.container_addrs[1:]:
          util.check_sshd(peer_addr, deadline, DTRAIN_SSH_PORT)
  
      # The chief has several layers of wrapper processes:
      # - a top-level pid_server, which causes the whole container to exit if any local worker dies.
      # - horovodrun, which launches $slots_per_trial copies of the following layers:
      #     - a pid_client process to contact the local pid_server
      #     - wrap_rank, which redirects stdin/stdout to the local container
      #     - harness.py, which actually does the training for the worker
      #
      # It is a bug in horovod that causes us to have this pid_server/pid_client pair of layers.
      # We can remove these layers when the upstream fix has been around for long enough that we can
      # reasonably require user images to have patched horovod installations.
  
      pid_server_cmd = create_hvd_pid_server_cmd(
          info.allocation_id, len(info.slot_ids)
      )
  
      # TODO: remove this (very old) hack when we have a configurable launch layer.
      hvd_optional_args = experiment_config.get("data", {}).get(
          "__det_dtrain_args", []
      )
      hvd_optional_args += hvd_args
      if debug:
          hvd_optional_args += ["--mpi-args=-v --display-map"]
  
      hvd_cmd = horovod.create_run_command(
          host_slot_counts=info.container_slot_counts,
          ip_addresses=info.container_addrs,
          inter_node_network_interface=info.trial._inter_node_network_interface,
          optimizations=experiment_config["optimizations"],
          debug=debug,
          optional_args=hvd_optional_args,
      )
  
      worker_wrapper_cmd = create_worker_wrapper_cmd(info.allocation_id)
  
      logger.debug(
          f"chief worker calling horovodrun with args: {hvd_cmd[1:]} ..."
      )
  
      os.environ["USE_HOROVOD"] = "1"
  
      # We now have environment images with built-in OpenMPI.   When invoked the
      # SLURM_JOBID variable triggers integration with SLURM, however, we are
      # running in a singularity container and SLURM may or may not have
      # compatible configuration enabled.  We therefore clear the SLURM_JOBID variable
      # before invoking mpi so that mpirun will honor the args passed via horvod
      # run to it describing the hosts and process topology, otherwise mpi ends
      # up wanting to launch all -np# processes on the local causing an oversubscription
      # error ("There are not enough slots available in the system").
      os.environ.pop("SLURM_JOBID", None)
      p = subprocess.Popen(
          pid_server_cmd + hvd_cmd + worker_wrapper_cmd + script
      )
      with det.util.forward_signals(p):
          return p.wait()
  
file_path: determined/harness/determined/launch/horovod.py
imports: import argparse
  import logging
  import os
  import subprocess
  import sys
  import time
  from typing import List, Tuple
  import determined
  from determined import horovod, util
  from determined.common import api
  from determined.common.api import certs
  from determined.constants import DTRAIN_SSH_PORT
label_desc: Use image node_id 4 for calling a global function with example usage: main(hvd_args, script, autohorovod) and returns: p, p, p