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: Starts the worker group.
docstr_tok: ["Starts","the","worker","group","."]
code: def start(
      self,
      initialization_hook: Optional[Callable[[], None]] = None,
      train_cls: Optional[Type] = None,
      train_cls_args: Optional[Tuple] = None,
      train_cls_kwargs: Optional[Dict] = None,
  ):
      """Starts the worker group."""
      self._create_placement_group()
      placement_group = self._placement_group or "default"
      self.worker_group = WorkerGroup(
          num_workers=self._num_workers,
          num_cpus_per_worker=self._num_cpus_per_worker,
          num_gpus_per_worker=self._num_gpus_per_worker,
          additional_resources_per_worker=self._additional_resources_per_worker,
          actor_cls=train_cls,
          actor_cls_args=train_cls_args,
          actor_cls_kwargs=train_cls_kwargs,
          placement_group=placement_group,
      )
      # Hack to avoid OOMs.
      # This is just a temporary solution for Train loading entire checkpoints
      # into memory by ensuring that the rank 0 worker is on the same node as
      # trainable, thus allowing for lazy checkpoint transfer to be used.
      # See https://github.com/ray-project/ray/issues/33073
      # for more context.
      # TODO remove passing in trial_driver_ip.
  
      trial_driver_ip = (
          self._trial_info.driver_ip if self._trial_info else None
      )
      self.worker_group.sort_workers_by_ip_and_gpu_id(trial_driver_ip)
  
      try:
          if initialization_hook:
              self._initialization_hook = initialization_hook
              self.worker_group.execute(initialization_hook)
  
          # Always propagate the driver's DataContext to each worker in the group.
          from ray.data import DataContext
  
          def _set_driver_dataset_context(ctx: DataContext):
              DataContext._set_current(ctx)
  
          self.worker_group.execute(
              _set_driver_dataset_context,
              DataContext.get_current(),
          )
  
          share_cuda_visible_devices_enabled = bool(
              env_integer(
                  ENABLE_SHARE_CUDA_VISIBLE_DEVICES_ENV,
                  self._backend.share_cuda_visible_devices,
              )
          )
  
          if (
              self._num_gpus_per_worker > 0
              and share_cuda_visible_devices_enabled
          ):
              self._share_cuda_visible_devices()
          elif self._additional_resources_per_worker:
              for resource_config in self._resource_configs:
                  if self._is_share_resources_enabled(
                      resource_config.resource_name,
                      resource_config.resource_enable_sharing_env_var,
                  ):
                      self._share_resource_ids(
                          resource_config.resource_name,
                          resource_config.share_resource_ids_env_var,
                      )
          self._backend.on_start(
              self.worker_group, self._backend_config
          )
      except RayActorError as exc:
          logger.exception(str(exc))
          logger.warning(
              "Failure occurred during startup. Restarting all workers and "
              "attempting to startup again."
          )
          self._increment_failures()
          self._restart()
  
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 2 for calling the BackendExecutor obj's underlying member method code with example usage: obj.start(initialization_hook, train_cls, train_cls_args, train_cls_kwargs) without return types