class_docstr: Main execution class for training backends.
  
  This class holds a worker group and is responsible for executing the
  training function on the workers, and collecting intermediate results
  from ``session.report()``.
  
  Args:
      backend_config: The configurations for this
          specific backend.
      num_workers: Number of workers to use for training.
      num_cpus_per_worker: Number of CPUs to use per worker.
      num_gpus_per_worker: Number of GPUs to use per worker.
      additional_resources_per_worker (Optional[Dict[str, float]]):
          Dictionary specifying the extra resources that will be
          requested for each worker in addition to ``num_cpus_per_worker``
          and ``num_gpus_per_worker``.
      max_retries: Number of retries when Ray actors fail.
          Defaults to 3. Set to -1 for unlimited retries.
class_docstr_tok: ["Main","execution","class","for","training","backends",".","This","class","holds","a","worker","group","and","is","responsible","for","executing","the","training","function","on","the","workers",",","and","collecting","intermediate","results","from","``","session.report","(",")","``",".","Args",":","backend_config",":","The","configurations","for","this","specific","backend",".","num_workers",":","Number","of","workers","to","use","for","training",".","num_cpus_per_worker",":","Number","of","CPUs","to","use","per","worker",".","num_gpus_per_worker",":","Number","of","GPUs","to","use","per","worker",".","additional_resources_per_worker","(","Optional","[","Dict","[","str",",","float","]","]",")",":","Dictionary","specifying","the","extra","resources","that","will","be","requested","for","each","worker","in","addition","to","``","num_cpus_per_worker","``","and","``","num_gpus_per_worker","``",".","max_retries",":","Number","of","retries","when","Ray","actors","fail",".","Defaults","to","3",".","Set","to","-1","for","unlimited","retries","."]
docstr: Creates a placement group if it does not exist.
  
  If a placement group is already detected (Tune) this will be a no-op.
  
  By default the placement group will be created with PACK strategy.
  This is optimized for colocating GPUs on a minimal number of nodes.
  This behavior can be overridden to use the SPREAD strategy by defining
  ``TRAIN_ENABLE_WORKER_SPREAD_ENV``
  
  If a placement group is created it will be stored as
  self._placement_group.
docstr_tok: ["Creates","a","placement","group","if","it","does","not","exist",".","If","a","placement","group","is","already","detected","(","Tune",")","this","will","be","a","no-op",".","By","default","the","placement","group","will","be","created","with","PACK","strategy",".","This","is","optimized","for","colocating","GPUs","on","a","minimal","number","of","nodes",".","This","behavior","can","be","overridden","to","use","the","SPREAD","strategy","by","defining","``","TRAIN_ENABLE_WORKER_SPREAD_ENV","``","If","a","placement","group","is","created","it","will","be","stored","as","self._placement_group","."]
code: def _create_placement_group(self):
      """Creates a placement group if it does not exist.
  
      If a placement group is already detected (Tune) this will be a no-op.
  
      By default the placement group will be created with PACK strategy.
      This is optimized for colocating GPUs on a minimal number of nodes.
      This behavior can be overridden to use the SPREAD strategy by defining
      ``TRAIN_ENABLE_WORKER_SPREAD_ENV``
  
      If a placement group is created it will be stored as
      self._placement_group.
      """
      current_placement_group = get_current_placement_group()
      worker = ray._private.worker.global_worker
      should_capture_child_tasks_in_placement_group = (
          worker.should_capture_child_tasks_in_placement_group
      )
      should_create_placement_group = (
          current_placement_group is None
          or not should_capture_child_tasks_in_placement_group
      )
  
      if should_create_placement_group:
          additional_resources_per_worker = (
              self._additional_resources_per_worker or {}
          )
          bundle = {
              "CPU": self._num_cpus_per_worker,
              "GPU": self._num_gpus_per_worker,
              **additional_resources_per_worker,
          }
          bundles = [bundle.copy() for _ in range(self._num_workers)]
  
          use_spread = bool(
              env_integer(TRAIN_ENABLE_WORKER_SPREAD_ENV, 0)
          )
          strategy = "SPREAD" if use_spread else "PACK"
  
          placement_group = ray.util.placement_group(
              bundles, strategy=strategy
          )
          logger.debug("Waiting for placement group to start.")
          timeout = env_integer(
              TRAIN_PLACEMENT_GROUP_TIMEOUT_S_ENV, 100
          )
          ready, _ = ray.wait(
              [placement_group.ready()], timeout=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. If you are running on a cluster, "
                  "make sure you specify an address in `ray.init()`, for example, "
                  '`ray.init("auto")`. You can also increase the timeout by setting '
                  "the TRAIN_PLACEMENT_GROUP_TIMEOUT_S environment variable. "
                  "Current resources available: {}, resources requested by the "
                  "placement group: {}".format(
                      ray.available_resources(),
                      placement_group.bundle_specs,
                  )
              )
          self._placement_group = placement_group
  
imports: import logging
  import os
  from collections import defaultdict
  from dataclasses import dataclass
  from typing import Any, Callable, Dict, List, Optional, Tuple, Type, TypeVar
  import ray
  import ray._private.ray_constants
  from ray._private.ray_constants import env_integer
  from ray.data import Dataset
  from ray.exceptions import RayActorError
  from ray.train import Checkpoint, DataConfig
  from ray.train._internal.session import TrialInfo, _TrainingResult, get_session, init_session, shutdown_session
  from ray.train._internal.storage import StorageContext
  from ray.train._internal.utils import check_for_failure
  from ray.train._internal.worker_group import WorkerGroup
  from ray.train.backend import BackendConfig
  from ray.train.constants import ENABLE_DETAILED_AUTOFILLED_METRICS_ENV, ENABLE_SHARE_CUDA_VISIBLE_DEVICES_ENV, ENABLE_SHARE_NEURON_CORES_ACCELERATOR_ENV, TRAIN_ENABLE_WORKER_SPREAD_ENV, TRAIN_PLACEMENT_GROUP_TIMEOUT_S_ENV
  from ray.util.placement_group import get_current_placement_group, remove_placement_group
label_desc: Use image node_id 3 for calling the BackendExecutor obj's underlying member method code with example usage: obj._create_placement_group() without return types