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: Gets the remote values while handling for worker failures.
  
  This method should be called instead of ``ray.get()`` directly in
  order to handle worker failures.
  
  If a worker failure is identified, backend specific failure handling
  is executed and a ``TrainingWorkerError`` is raised.
  
  Args:
      remote_values: List of object refs representing functions
          that may fail in the middle of execution. For example, running
          a Train training loop in multiple parallel actor calls.
  Returns:
      The resolved objects represented by the passed in ObjectRefs.
docstr_tok: ["Gets","the","remote","values","while","handling","for","worker","failures",".","This","method","should","be","called","instead","of","``","ray.get","(",")","``","directly","in","order","to","handle","worker","failures",".","If","a","worker","failure","is","identified",",","backend","specific","failure","handling","is","executed","and","a","``","TrainingWorkerError","``","is","raised",".","Args",":","remote_values",":","List","of","object","refs","representing","functions","that","may","fail","in","the","middle","of","execution",".","For","example",",","running","a","Train","training","loop","in","multiple","parallel","actor","calls",".","Returns",":","The","resolved","objects","represented","by","the","passed","in","ObjectRefs","."]
code: def get_with_failure_handling(self, remote_values):
      """Gets the remote values while handling for worker failures.
  
      This method should be called instead of ``ray.get()`` directly in
      order to handle worker failures.
  
      If a worker failure is identified, backend specific failure handling
      is executed and a ``TrainingWorkerError`` is raised.
  
      Args:
          remote_values: List of object refs representing functions
              that may fail in the middle of execution. For example, running
              a Train training loop in multiple parallel actor calls.
      Returns:
          The resolved objects represented by the passed in ObjectRefs.
      """
      success, exception = check_for_failure(remote_values)
      if success:
          return ray.get(remote_values)
      else:
          self._last_failure = exception
          self._increment_failures()
          logger.warning(
              "Failure identified during training. Restarting all workers and "
              "continuing training from latest checkpoint."
          )
          self._restart()
          raise TrainingWorkerError
  
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 12 for calling the BackendExecutor obj's underlying member method code with example usage: obj.get_with_failure_handling(remote_values) and returns: ray