class_docstr: Executor for elastic jobs using Ray.
  
  Leverages the Ray global state to detect available hosts and
  slots. Assumes that the entire Ray cluster is available for
  the Executor to use.
  
  Args:
      settings: Configuration for the elastic job
          setup. You can use a standard Horovod ElasticSettings
          object or create one directly from
          ElasticRayExecutor.create_settings.
      use_gpu (bool): Whether to use GPU for allocation.
      cpus_per_slot (int): Number of CPU resources to allocate to
          each worker.
      gpus_per_slot (int): Number of GPU resources to allocate to
          each worker.
      env_vars (Dict): Environment variables to be set
          on the actors (worker processes) before initialization.
      override_discovery (bool): Whether for the ElasticRayExecutor to
          automatically provide a discovery mechanism for ElasticSettings.
  
  Example:
  
  .. code-block:: python
  
      import ray
      ray.init(address="auto")
      settings = ElasticRayExecutor.create_settings(verbose=True)
      executor = ElasticRayExecutor(
          settings, use_gpu=True, cpus_per_slot=2)
      executor.start()
      executor.run(train_fn)
  
  warning:: .. deprecated:: 0.25.0
docstr: Executes the provided function on all workers.
  
  Args:
      worker_fn: Target elastic function that can be executed.
      callbacks: List of callables. Each callback must either
          be a callable function or a class that implements __call__.
          Every callback will be invoked on every value logged
          by the rank 0 worker.
  
  Returns:
      List of return values from every completed worker.
code: def run(
      self,
      worker_fn: Callable,
      callbacks: Optional[List[Callable]] = None,
  ) -> List[Any]:
      """Executes the provided function on all workers.
  
      Args:
          worker_fn: Target elastic function that can be executed.
          callbacks: List of callables. Each callback must either
              be a callable function or a class that implements __call__.
              Every callback will be invoked on every value logged
              by the rank 0 worker.
  
      Returns:
          List of return values from every completed worker.
      """
      return_values = []
      from ray.util.queue import Queue
      import inspect
  
      args = inspect.getfullargspec(Queue).args
      if "actor_options" not in args:
          # Ray 1.1 and less
          _queue = Queue()
      else:
          _queue = Queue(
              actor_options={
                  "num_cpus": 0,
                  "resources": {ray.state.current_node_id(): 0.001},
              }
          )
      self.driver.start(
          self.settings.num_proc,
          self._create_spawn_worker_fn(
              return_values, worker_fn, _queue
          ),
      )
  
      def _process_calls(queue, callbacks, event):
          if not callbacks:
              return
          while queue.actor:
              if not queue.empty():
                  result = queue.get_nowait()
                  for c in callbacks:
                      c(result)
                  # avoid slamming the CI
              elif event.is_set():
                  break
              time.sleep(0.1)
  
      try:
          event = threading.Event()
          _callback_thread = threading.Thread(
              target=_process_calls,
              args=(_queue, callbacks, event),
              daemon=True,
          )
          _callback_thread.start()
          res = self.driver.get_results()
          event.set()
          if _callback_thread:
              _callback_thread.join(timeout=60)
      finally:
          if hasattr(_queue, "shutdown"):
              _queue.shutdown()
          else:
              done_ref = _queue.actor.__ray_terminate__.remote()
              done, not_done = ray.wait([done_ref], timeout=5)
              if not_done:
                  ray.kill(_queue.actor)
      self.driver.stop()
  
      if res.error_message is not None:
          raise RuntimeError(res.error_message)
  
      for name, value in sorted(
          res.worker_results.items(), key=lambda item: item[1][1]
      ):
          exit_code, timestamp = value
          if exit_code != 0:
              raise RuntimeError(
                  "Horovod detected that one or more processes "
                  "exited with non-zero "
                  "status, thus causing the job to be terminated. "
                  "The first process "
                  "to do so was:\nProcess name: {name}\nExit code: {code}\n".format(
                      name=name, code=exit_code
                  )
              )
  
      return_values = [
          value
          for k, value in sorted(return_values, key=lambda kv: kv[0])
      ]
      return return_values
  
imports: from typing import Callable, List, Any, Dict, Optional
  import logging
  import socket
  import time
  import os
  import random
  import math
  import threading
  import warnings
  from horovod.runner.common.util import timeout, secret
  from horovod.runner.http.http_server import RendezvousServer
  from horovod.runner.gloo_run import create_slot_env_vars, create_run_env_vars, _get_min_start_hosts
  from horovod.runner.elastic.settings import ElasticSettings
  from horovod.runner.elastic.rendezvous import create_rendezvous_handler
  from horovod.runner.elastic.discovery import HostDiscovery
  from horovod.runner.elastic.driver import ElasticDriver
  import ray
  import ray.exceptions
  from horovod.ray.worker import BaseHorovodWorker
  from horovod.ray.utils import detect_nics
label_desc: Use image node_id 7 for calling the ElasticRayExecutor obj's underlying member method code with example usage: obj.run(worker_fn, callbacks) and returns: return_values