code: def setup_workers(self):
      ray.get(
          [
              worker.setup_address.remote()
              for worker in self.remote_instances
          ]
      )
      ps_ips = []
      chief_ips = []
      worker_ips = []
      for worker in self.remote_instances:
          ip = ray.get(worker.get_address.remote())
          task_type = ray.get(worker.get_task_type.remote())
          if task_type == "ps":
              ps_ips.append(ip)
          elif task_type == "chief":
              chief_ips.append(ip)
          else:
              worker_ips.append(ip)
  
      cluster_info = {
          "ps": ps_ips,
          "chief": chief_ips,
          "worker": worker_ips,
      }
      print(cluster_info)
      ps_chief_refs = [
          worker.setup_distributed.remote(cluster_info, self.protocol)
          for worker in self.remote_ps
      ]
      ps_chief_refs += [
          self.remote_workers[0].setup_distributed.remote(
              cluster_info, self.protocol
          )
      ]
      # Use ray.wait since ps server.join process won't terminate by itself and
      # thus using ray.get on all results would hang.
      # By using ray.wait, we wait for the chief to finish the distributed setting and proceed.
      # In this case, we don't need to manually kill the ps process at the very end.
      # This also tries to make sure that ps and chief is launched earlier than ordinary workers.
      finished, unfinished = ray.wait(ps_chief_refs, num_returns=1)
      print(ray.get(finished))
      worker_refs = [
          worker.setup_distributed.remote(cluster_info, self.protocol)
          for worker in self.remote_workers[1:]
      ]
      print(ray.get(worker_refs))
  
imports: import time
  import argparse
  import tensorflow
  import os
  import sys
  import math
  import collections
  from tensorflow.python.client import timeline
  import json
  from tensorflow.python.ops import partitioned_variables
  import ray
  from bigdl.dllib.utils.log4Error import *
  from bigdl.orca import init_orca_context, stop_orca_context, OrcaContext
  from bigdl.orca.data.utils import process_spark_xshards
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards
  from bigdl.friesian.feature import FeatureTable
label_desc: Use image node_id 8 for calling the RayDeepRecCluster obj's underlying member method code with example usage: obj.setup_workers() without return types