class_docstr: Ensures that the workers are balanced across all hosts.
class_docstr_tok: ["Ensures","that","the","workers","are","balanced","across","all","hosts","."]
code: def create_workers(self):
      self.placement_group, bundles = create_placement_group(
          resources_per_bundle=self._resources_per_host(),
          num_bundles=self.num_hosts,
          pg_timeout=self.settings.placement_group_timeout_s,
          pg_strategy="STRICT_SPREAD",
      )
  
      # Placement group has started. Now create the workers.
      self.workers = []
  
      # STRICT_SPREAD guarantees each bundle is on a different node.
      # Create num_workers_per_host workers per bundle, i.e. per machine.
      for bundle_index in range(len(bundles)):
          gpu_id_futures = []
          curr_node_workers = []
          remote_cls = ray.remote(BaseHorovodWorker)
          for i in range(self.num_workers_per_host):
              remote_cls_with_options = remote_cls.options(
                  num_cpus=self.cpus_per_worker,
                  num_gpus=self.gpus_per_worker * int(self.use_gpu),
                  placement_group_capture_child_tasks=False,
                  placement_group=self.placement_group,
                  placement_group_bundle_index=bundle_index,
              )
              worker = remote_cls_with_options.remote(
                  world_rank=self.num_workers_per_host * bundle_index
                  + i,
                  world_size=self.num_workers,
              )
              if self.use_gpu:
                  gpu_id_futures.append(worker.get_gpu_ids.remote())
              self.workers.append(worker)
              curr_node_workers.append(worker)
          if len(gpu_id_futures) > 0:
              # By setting CUDA VISIBLE DEVICES to ALL GPUs,
              # CUDA will be able to detect adjacent devices and use IPC
              # allowing for better performance.
              gpu_ids = sum(ray.get(gpu_id_futures), [])
              # Make sure that each worker on the node has unique device.
              assert (
                  len(gpu_ids)
                  == len(set(gpu_ids))
                  == self.num_workers_per_host
              ), gpu_ids
              all_ids = ",".join([str(gpu_id) for gpu_id in gpu_ids])
              futures = []
              for worker in curr_node_workers:
                  futures.append(
                      worker.update_env_vars.remote(
                          {"CUDA_VISIBLE_DEVICES": all_ids}
                      )
                  )
              ray.get(futures)
  
      return self.workers, self.get_node_workers(self.workers)
  
imports: from collections import defaultdict
  import logging
  from typing import Dict
  import ray
  from ray.util.placement_group import get_current_placement_group
  from horovod.ray.utils import map_blocking
  from horovod.ray.worker import BaseHorovodWorker
label_desc: Use image node_id 4 for calling the ColocatedStrategy obj's underlying member method code with example usage: obj.create_workers() and returns: self,  self