class_docstr: Packs workers together but does not guarantee balanced hosts.
  
  Will use the current placement group if one is available.
class_docstr_tok: ["Packs","workers","together","but","does","not","guarantee","balanced","hosts",".","Will","use","the","current","placement","group","if","one","is","available","."]
code: def create_workers(self):
      if not self.placement_group:
          self.placement_group, _ = create_placement_group(
              resources_per_bundle=self.resources_per_worker(),
              num_bundles=self.num_workers,
              pg_strategy="PACK",
              pg_timeout=self.settings.placement_group_timeout_s,
          )
          self._created_placement_group = True
  
      # Placement group has started. Now create the workers.
      self.workers = []
      remote_cls = ray.remote(BaseHorovodWorker)
  
      for worker_index in range(self.num_workers):
          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=worker_index
              if self._created_placement_group
              else -1,
          )
          worker = remote_cls_with_options.remote(
              world_rank=worker_index, world_size=self.num_workers
          )
  
          self.workers.append(worker)
  
      if self.use_gpu:
          node_ids = ray.get(
              [worker.node_id.remote() for worker in self.workers]
          )
          gpus = ray.get(
              [worker.get_gpu_ids.remote() for worker in self.workers]
          )
          node_workers = defaultdict(list)
          node_id_to_gpus = defaultdict(list)
          for worker, node_id, worker_gpu_ids in zip(
              self.workers, node_ids, gpus
          ):
              node_workers[node_id].append(worker)
              node_id_to_gpus[node_id].extend(worker_gpu_ids)
  
          futures = []
          for node_id, gpu_ids in node_id_to_gpus.items():
              all_ids = ",".join([str(gpu_id) for gpu_id in gpu_ids])
  
              for worker in node_workers[node_id]:
                  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 PGStrategy obj's underlying member method code with example usage: obj.create_workers() and returns: self,  self