code: def create_placement_group(
      resources_per_bundle: Dict[str, int],
      num_bundles: int,
      pg_timeout: int,
      pg_strategy: str,
  ):
      bundles = [
          resources_per_bundle.copy() for _ in range(num_bundles)
      ]
      pg = ray.util.placement_group(bundles, strategy=pg_strategy)
      logger.debug("Waiting for placement group to start.")
      ready, _ = ray.wait([pg.ready()], timeout=pg_timeout)
      if ready:
          logger.debug("Placement group has started.")
      else:
          raise TimeoutError(
              "Placement group creation timed out. Make sure "
              "your cluster either has enough resources or use "
              "an autoscaling cluster. Current resources "
              "available: {}, resources requested by the "
              "placement group: {}".format(
                  ray.available_resources(), pg.bundle_specs
              )
          )
  
      return pg, bundles
  
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 1 for calling a global function with example usage: create_placement_group(resources_per_bundle, num_bundles, pg_timeout, pg_strategy) and returns: pg,  bundles