class_docstr: Adapter for executing Ray calls for non-elastic Horovod jobs.
  
  Args:
      settings (horovod.Settings): Configuration for job setup. You can
          use a standard Horovod Settings object or create one directly
          from RayExecutor.create_settings.
      num_workers (int): Number of workers to use for training.
      cpus_per_worker (int): Number of CPU resources to allocate to
          each worker.
      use_gpu (bool): Whether to use GPU for allocation. TODO: this
          can be removed.
      gpus_per_worker (int): Number of GPU resources to allocate to
          each worker.
      num_hosts (int): Alternative API to ``num_workers``. Number of
          machines to execute the job on. Used to enforce equal number of
          workers on each machine.
      num_workers_per_host (int): Alternative API to
          ``num_workers``. Number of workers to be placed on each machine.
          Used to enforce equal number of workers on each machine. Only
          used in conjunction with `num_hosts`.
      use_current_placement_group (bool): Whether to use the current
          placement group instead of creating a new one. Defaults to True.
class_docstr_tok: ["Adapter","for","executing","Ray","calls","for","non-elastic","Horovod","jobs",".","Args",":","settings","(","horovod.Settings",")",":","Configuration","for","job","setup",".","You","can","use","a","standard","Horovod","Settings","object","or","create","one","directly","from","RayExecutor.create_settings",".","num_workers","(","int",")",":","Number","of","workers","to","use","for","training",".","cpus_per_worker","(","int",")",":","Number","of","CPU","resources","to","allocate","to","each","worker",".","use_gpu","(","bool",")",":","Whether","to","use","GPU","for","allocation",".","TODO",":","this","can","be","removed",".","gpus_per_worker","(","int",")",":","Number","of","GPU","resources","to","allocate","to","each","worker",".","num_hosts","(","int",")",":","Alternative","API","to","``","num_workers","``",".","Number","of","machines","to","execute","the","job","on",".","Used","to","enforce","equal","number","of","workers","on","each","machine",".","num_workers_per_host","(","int",")",":","Alternative","API","to","``","num_workers","``",".","Number","of","workers","to","be","placed","on","each","machine",".","Used","to","enforce","equal","number","of","workers","on","each","machine",".","Only","used","in","conjunction","with","`","num_hosts","`",".","use_current_placement_group","(","bool",")",":","Whether","to","use","the","current","placement","group","instead","of","creating","a","new","one",".","Defaults","to","True","."]
docstr: Starts the workers and colocates them on all machines.
  
  We implement a node grouping because it seems like
  our implementation doesn't quite work for imbalanced nodes.
  Also, colocation performance is typically much better than
  non-colocated workers.
  
  Args:
      executable_cls (type): The class that will be created within
          an actor (BaseHorovodWorker). This will allow Horovod
          to establish its connections and set env vars.
      executable_args (List): Arguments to be passed into the
          worker class upon initialization.
      executable_kwargs (Dict): Keyword arguments to be passed into the
          worker class upon initialization.
      extra_env_vars (Dict): Environment variables to be set
          on the actors (worker processes) before initialization.
docstr_tok: ["Starts","the","workers","and","colocates","them","on","all","machines",".","We","implement","a","node","grouping","because","it","seems","like","our","implementation","does","n't","quite","work","for","imbalanced","nodes",".","Also",",","colocation","performance","is","typically","much","better","than","non-colocated","workers",".","Args",":","executable_cls","(","type",")",":","The","class","that","will","be","created","within","an","actor","(","BaseHorovodWorker",")",".","This","will","allow","Horovod","to","establish","its","connections","and","set","env","vars",".","executable_args","(","List",")",":","Arguments","to","be","passed","into","the","worker","class","upon","initialization",".","executable_kwargs","(","Dict",")",":","Keyword","arguments","to","be","passed","into","the","worker","class","upon","initialization",".","extra_env_vars","(","Dict",")",":","Environment","variables","to","be","set","on","the","actors","(","worker","processes",")","before","initialization","."]
code: def start(
      self,
      executable_cls: type = None,
      executable_args: Optional[List] = None,
      executable_kwargs: Optional[Dict] = None,
      extra_env_vars: Optional[Dict] = None,
  ):
      """Starts the workers and colocates them on all machines.
  
      We implement a node grouping because it seems like
      our implementation doesn't quite work for imbalanced nodes.
      Also, colocation performance is typically much better than
      non-colocated workers.
  
      Args:
          executable_cls (type): The class that will be created within
              an actor (BaseHorovodWorker). This will allow Horovod
              to establish its connections and set env vars.
          executable_args (List): Arguments to be passed into the
              worker class upon initialization.
          executable_kwargs (Dict): Keyword arguments to be passed into the
              worker class upon initialization.
          extra_env_vars (Dict): Environment variables to be set
              on the actors (worker processes) before initialization.
  
      """
      extra_env_vars = extra_env_vars or {}
  
      self.strategy = self._create_strategy()
      self.coordinator = Coordinator(self.settings)
      executable_args = executable_args or []
      self.workers, node_workers = self.strategy.create_workers()
      # Get all the hostnames of all workers
      node_ids = map_blocking(
          lambda w: w.node_id.remote(), self.workers
      )
      hostnames = map_blocking(
          lambda w: w.hostname.remote(), self.workers
      )
      # Register each hostname to the coordinator. assumes the hostname
      # ordering is the same.
      for rank, (hostname, node_id) in enumerate(
          zip(hostnames, node_ids)
      ):
          self.coordinator.register(hostname, node_id, rank)
      all_info = self.coordinator.finalize_registration()
  
      indexed_runners = dict(enumerate(self.workers))
      for rank, local_cross_env_var in all_info.items():
          indexed_runners[rank].update_env_vars.remote(
              local_cross_env_var
          )
  
      coordinator_envs = self.coordinator.establish_rendezvous()
      coordinator_envs.update(extra_env_vars)
      nics = detect_nics(
          self.settings,
          all_host_names=list(self.coordinator.hostnames),
          node_workers=node_workers,
      )
      coordinator_envs.update(nics_to_env_var(nics))
  
      map_blocking(
          lambda w: w.update_env_vars.remote(coordinator_envs),
          self.workers,
      )
  
      self._start_executables(
          executable_cls, executable_args, executable_kwargs
      )
  
imports: import ray
  from ray.util.placement_group import get_current_placement_group
  from collections import defaultdict
  from dataclasses import dataclass, asdict
  import os
  from typing import Dict, Callable, Any, Optional, List
  import logging
  import ray.exceptions
  from horovod.ray.adapter import Adapter, BaseParams
  from horovod.runner.common.util import secret, timeout, hosts
  from horovod.runner.http.http_server import RendezvousServer
  from horovod.ray.utils import detect_nics, nics_to_env_var, map_blocking
  from horovod.ray.strategy import ColocatedStrategy, PGStrategy
  from horovod.ray.elastic_v2 import ElasticParams
label_desc: Use image node_id 4 for calling the StaticAdapter obj's underlying member method code with example usage: obj.start(executable_cls, executable_args, executable_kwargs, extra_env_vars) without return types