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: Executes a training function on all workers in a separate thread.
  
  ``finish_training`` should be called after this.
  
  Args:
      train_func: The training function to run on each worker.
      datasets: The base datasets.
      data_config: The config object for creating dataset shards for workers.
      checkpoint: The checkpoint data that
          should be loaded onto each worker and accessed by the
          training function via ``session.get_checkpoint()``. If this
          is ``None`` then no checkpoint will be loaded.
docstr_tok: ["Executes","a","training","function","on","all","workers","in","a","separate","thread",".","``","finish_training","``","should","be","called","after","this",".","Args",":","train_func",":","The","training","function","to","run","on","each","worker",".","datasets",":","The","base","datasets",".","data_config",":","The","config","object","for","creating","dataset","shards","for","workers",".","checkpoint",":","The","checkpoint","data","that","should","be","loaded","onto","each","worker","and","accessed","by","the","training","function","via","``","session.get_checkpoint","(",")","``",".","If","this","is","``","None","``","then","no","checkpoint","will","be","loaded","."]
code: def start_training(
      self,
      train_func: Callable[[], T],
      datasets: Dict[str, Dataset],
      metadata: Dict[str, Any],
      data_config: DataConfig,
      storage: StorageContext,
      checkpoint: Optional[Checkpoint] = None,
      on_session_init: Callable[[], None] = None,
  ) -> None:
      """Executes a training function on all workers in a separate thread.
  
      ``finish_training`` should be called after this.
  
      Args:
          train_func: The training function to run on each worker.
          datasets: The base datasets.
          data_config: The config object for creating dataset shards for workers.
          checkpoint: The checkpoint data that
              should be loaded onto each worker and accessed by the
              training function via ``session.get_checkpoint()``. If this
              is ``None`` then no checkpoint will be loaded.
      """
      use_detailed_autofilled_metrics = env_integer(
          ENABLE_DETAILED_AUTOFILLED_METRICS_ENV, 0
      )
  
      # First initialize the session.
      def initialize_session(
          train_func,
          world_rank,
          local_rank,
          node_rank,
          local_world_size,
          world_size,
          trial_info,
          checkpoint,
          dataset_shard,
          metadata,
          storage,
      ):
          try:
              init_session(
                  training_func=train_func,
                  world_rank=world_rank,
                  local_rank=local_rank,
                  node_rank=node_rank,
                  local_world_size=local_world_size,
                  world_size=world_size,
                  trial_info=trial_info,
                  dataset_shard=dataset_shard,
                  metadata=metadata,
                  checkpoint=checkpoint,
                  detailed_autofilled_metrics=use_detailed_autofilled_metrics,
                  storage=storage,
              )
          except ValueError:
              raise TrainBackendError(
                  "Attempting to start training but a "
                  "previous training run is still ongoing. "
                  "You must call `finish_training` before "
                  "calling `start_training` again."
              )
  
      if self.dataset_shards is None:
          actors = [
              worker.actor for worker in self.worker_group.workers
          ]
          node_ids = [
              worker.metadata.node_id
              for worker in self.worker_group.workers
          ]
          self.dataset_shards = data_config.configure(
              datasets,
              world_size=len(self.worker_group),
              worker_handles=actors,
              worker_node_ids=node_ids,
          )
  
      (
          local_rank_map,
          local_world_size_map,
          node_rank_map,
      ) = self._create_rank_world_size_mappings()
  
      futures = []
      for index in range(len(self.worker_group)):
          futures.append(
              self.worker_group.execute_single_async(
                  index,
                  initialize_session,
                  world_rank=index,
                  local_rank=local_rank_map[index],
                  node_rank=node_rank_map[index],
                  local_world_size=local_world_size_map[index],
                  world_size=len(self.worker_group),
                  trial_info=self._trial_info,
                  train_func=train_func,
                  dataset_shard=self.dataset_shards[index],
                  metadata=metadata,
                  checkpoint=checkpoint,
                  storage=storage,
              )
          )
  
      self._backend.on_training_start(
          self.worker_group, self._backend_config
      )
  
      self.get_with_failure_handling(futures)
  
      if on_session_init:
          on_session_init()
  
      # Run the training function asynchronously in its own thread.
      def train_async():
          session = get_session()
          session.start()
  
      self.worker_group.execute_async(train_async)
  
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 8 for calling the BackendExecutor obj's underlying member method code with example usage: obj.start_training(train_func, datasets, metadata, data_config, storage, checkpoint, on_session_init) without return types