code: def __init__(
      self,
      ray_ctx,
      worker_cls=None,
      worker_param=None,
      workers_per_node=1,
  ):
      self.cores_per_node = (
          ray_ctx.ray_node_cpu_cores // workers_per_node
      )
      self.num_nodes = ray_ctx.num_ray_nodes * workers_per_node
      if worker_param is None:
          worker_param = {}
      worker_cls = make_worker(worker_cls, HorovodWorker)
      self.worker_class = ray.remote(num_cpus=self.cores_per_node)(
          worker_cls
      )
      self.remote_workers = [
          self.worker_class.remote(**worker_param)
          for i in range(0, self.num_nodes)
      ]
      hosts = ray.get(
          [worker.ip_addr.remote() for worker in self.remote_workers]
      )
      hosts_spec, name_rank_to_id, host_to_size = _hosts_to_hosts_spec(
          hosts
      )
  
      major, minor, patch, version_str = get_horovod_version()
  
      if major == 0 and minor < 19:
          raise RuntimeError(
              f"We only support horovod versions newer "
              f"than 0.19.0, but got {version_str}"
          )
      if major == 0 and minor == 19:
          from horovod.run.gloo_run import RendezvousServer, _allocate
  
          self.host_alloc_plan = _allocate(
              ",".join(hosts_spec), self.num_nodes
          )
          self.global_rendezv = RendezvousServer(True)
          global_rendezv_port = self.global_rendezv.start_server(
              self.host_alloc_plan
          )
      else:
          from horovod.runner.gloo_run import (
              RendezvousServer,
              parse_hosts,
              get_host_assignments,
          )
  
          self.host_alloc_plan = get_host_assignments(
              parse_hosts(",".join(hosts_spec)), self.num_nodes
          )
          self.global_rendezv = RendezvousServer(True)
          global_rendezv_port = self.global_rendezv.start()
          self.global_rendezv.init(self.host_alloc_plan)
  
      driver_ip = ray._private.services.get_node_ip_address()
  
      common_envs = {
          "HOROVOD_GLOO_RENDEZVOUS_ADDR": driver_ip,
          "HOROVOD_GLOO_RENDEZVOUS_PORT": str(global_rendezv_port),
          "HOROVOD_CONTROLLER": "gloo",
          "HOROVOD_CPU_OPERATIONS": "gloo",
          "PYTHONUNBUFFERED": "1",
          "OMP_NUM_THREADS": str(self.cores_per_node),
      }
  
      for key in os.environ:
          if key.startswith("HOROVOD"):
              common_envs[key] = os.environ[key]
  
      # todo support other Horovod envs
      self.per_worker_envs = [
          common_envs.copy() for _ in range(self.num_nodes)
      ]
      for alloc_info in self.host_alloc_plan:
          key = (alloc_info.hostname, alloc_info.local_rank)
          local_envs = self.per_worker_envs[name_rank_to_id[key]]
          local_envs["HOROVOD_HOSTNAME"] = str(alloc_info.hostname)
          local_envs["HOROVOD_RANK"] = str(alloc_info.rank)
          local_envs["HOROVOD_SIZE"] = str(alloc_info.size)
          local_envs["HOROVOD_LOCAL_RANK"] = str(alloc_info.local_rank)
          local_envs["HOROVOD_LOCAL_SIZE"] = str(alloc_info.local_size)
          local_envs["HOROVOD_CROSS_RANK"] = str(alloc_info.cross_rank)
          local_envs["HOROVOD_CROSS_SIZE"] = str(alloc_info.cross_size)
  
      ray.get(
          [
              worker.set_gloo_iface.remote()
              for worker in self.remote_workers
          ]
      )
      self.run(lambda: print("horovod worker initialized"))
  
file_path: analytics-zoo/pyzoo/zoo/orca/learn/horovod/horovod_ray_runner.py
imports: import ray
  import os
label_desc: Use image node_id 1 to create a new HorovodRayRunner object  with example: obj = HorovodRayRunner(ray_ctx, worker_cls, worker_param, workers_per_node)