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: Fetches the next ``_TrainingResult`` from each worker.
  
  Each ``_TrainingResult`` is expected to correspond to the same step from
  each worker (e.g. the same call to ``train.report()``).
  
  Returns:
      A list of ``_TrainingResult``s or ``None`` if there are no more results
      since the training function has exited on all workers.
docstr_tok: ["Fetches","the","next","``","_TrainingResult","``","from","each","worker",".","Each","``","_TrainingResult","``","is","expected","to","correspond","to","the","same","step","from","each","worker","(","e.g",".","the","same","call","to","``","train.report","(",")","``",")",".","Returns",":","A","list","of","``","_TrainingResult","``","s","or","``","None","``","if","there","are","no","more","results","since","the","training","function","has","exited","on","all","workers","."]
code: def get_next_results(self) -> Optional[List[_TrainingResult]]:
      """Fetches the next ``_TrainingResult`` from each worker.
  
      Each ``_TrainingResult`` is expected to correspond to the same step from
      each worker (e.g. the same call to ``train.report()``).
  
      Returns:
          A list of ``_TrainingResult``s or ``None`` if there are no more results
          since the training function has exited on all workers.
      """
  
      def get_next():
          session = _get_session("get_next_results")
          try:
              result = session.get_next()
          except RuntimeError:
              # Training thread has not been started yet.
              raise TrainBackendError(
                  "`get_next_results` has been called "
                  "before `start_training`. Please call "
                  "`start_training` before "
                  "`get_next_results`."
              )
  
          return result
  
      # Get next result from each worker.
      futures = self.worker_group.execute_async(get_next)
      results = self.get_with_failure_handling(futures)
  
      # Check if any worker returned None.
      if any(r is None for r in results):
          # Either all workers have results or none of them do.
          if not all(r is None for r in results):
              raise RuntimeError(
                  "Some workers returned results while "
                  "others didn't. Make sure that "
                  "`session.report()` are called the "
                  "same number of times on all workers."
              )
          else:
              # Return None if all results are None.
              return None
  
      return results
  
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 9 for calling the BackendExecutor obj's underlying member method code with example usage: obj.get_next_results() and returns: results, result, None