code: def __init__(
      self,
      num_workers,
      worker_cores,
      worker_cls=None,
      worker_param=None,
      cpu_binding=True,
  ):
      if not ray.is_initialized():
          raise RuntimeError(
              "Ray is not initialize. Please initialize ray."
          )
  
      self.num_workers = num_workers
      self.worker_cores = worker_cores
      self.worker_cls = make_worker(worker_cls)
      self.work_param = worker_param
      if sys.platform == "linux":
          self.cpu_binding = cpu_binding
      else:
          if cpu_binding:
              log.warn(
                  f"cpu_binding is only support in linux, detectiong os {sys.platform}, "
                  "set cpu_binding to False"
              )
  
          self.cpu_binding = False
  
      self.worker_class = ray.remote(num_cpus=self.worker_cores)(
          self.worker_cls
      )
      self.remote_workers = [
          self.worker_class.remote(**worker_param)
          for i in range(0, self.num_workers)
      ]
  
      if self.cpu_binding:
          hosts = ray.get(
              [
                  worker.ip_addr.remote()
                  for worker in self.remote_workers
              ]
          )
          ip2workers = {}
          for ip, worker in zip(hosts, self.remote_workers):
              if ip not in ip2workers:
                  ip2workers[ip] = []
              ip2workers[ip].append(worker)
          ips = ip2workers.keys()
  
          cpu_binding_refs = []
          for ip in ips:
              ref = ip2workers[ip][0].run.remote(
                  schedule_workers,
                  len(ip2workers[ip]),
                  self.worker_cores,
              )
              cpu_binding_refs.append(ref)
          cpu_bindings = ray.get(cpu_binding_refs)
  
          result = []
          for ip, core_lists in zip(ips, cpu_bindings):
              for worker, core_list in zip(ip2workers[ip], core_lists):
                  log.debug(
                      f"Setting thread affinity for worker in {ip}: {core_list}"
                  )
                  result.append(
                      worker.set_cpu_affinity.remote(core_list)
                  )
  
          ray.get(result)
      else:
          ray.get(
              [
                  worker.disable_cpu_affinity.remote(self.worker_cores)
                  for worker in self.remote_workers
              ]
          )
  
file_path: analytics-zoo/pyzoo/zoo/orca/learn/dl_cluster.py
imports: import ray
  from zoo.orca.cpu_info import schedule_workers
  import os
  import sys
  import logging
label_desc: Use image node_id 1 to create a new RayDLCluster object  with example: obj = RayDLCluster(num_workers, worker_cores, worker_cls, worker_param, cpu_binding)