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: Sets the given env_var on all workers.
  
  For each worker, the cores/devices are visible to all the
  workers on that worker's node.This allows workers on the
  same node to communicate with one another.
  
  Example:
  
      Setup:
      - Node1:
          - Worker1: {0, 1}
          - Worker2: {2, 3}
      - Node2:
          - Worker3: {0, 1}
  
      NEURON_RT_VISIBLE_CORES/TPU_VISIBLE_CHIPS/...:
      - Worker1: "0,1,2,3"
      - Worker2: "0,1,2,3"
      - Worker2: "0,1"
  
  Args:
      resource: The name of the resource/accelerator.
      env_var: The name of the environment variable to set.
docstr_tok: ["Sets","the","given","env_var","on","all","workers",".","For","each","worker",",","the","cores\/devices","are","visible","to","all","the","workers","on","that","worker","'s","node.This","allows","workers","on","the","same","node","to","communicate","with","one","another",".","Example",":","Setup",":","-","Node1",":","-","Worker1",":","{","0",",","1","}","-","Worker2",":","{","2",",","3","}","-","Node2",":","-","Worker3",":","{","0",",","1","}","NEURON_RT_VISIBLE_CORES\/TPU_VISIBLE_CHIPS\/","...",":","-","Worker1",":","``","0,1,2,3","''","-","Worker2",":","``","0,1,2,3","''","-","Worker2",":","``","0,1","''","Args",":","resource",":","The","name","of","the","resource\/accelerator",".","env_var",":","The","name","of","the","environment","variable","to","set","."]
code: def _share_resource_ids(self, resource: str, env_var: str):
      """Sets the given env_var on all workers.
  
      For each worker, the cores/devices are visible to all the
      workers on that worker's node.This allows workers on the
      same node to communicate with one another.
  
      Example:
  
          Setup:
          - Node1:
              - Worker1: {0, 1}
              - Worker2: {2, 3}
          - Node2:
              - Worker3: {0, 1}
  
          NEURON_RT_VISIBLE_CORES/TPU_VISIBLE_CHIPS/...:
          - Worker1: "0,1,2,3"
          - Worker2: "0,1,2,3"
          - Worker2: "0,1"
  
      Args:
          resource: The name of the resource/accelerator.
          env_var: The name of the environment variable to set.
      """
      node_ids_and_resource_ids = [
          (
              w.metadata.node_id,
              w.metadata.resource_ids[resource],
          )
          for w in self.worker_group.workers
      ]
      node_id_to_worker_id = defaultdict(set)
      node_id_to_resource_ids = defaultdict(set)
  
      for worker_id, (node_id, resource_ids) in enumerate(
          node_ids_and_resource_ids
      ):
          node_id_to_worker_id[node_id].add(worker_id)
          node_id_to_resource_ids[node_id].update(resource_ids)
  
      futures = []
      for node_id, resource_ids in node_id_to_resource_ids.items():
          resource_ids = sorted(resource_ids)
          all_resource_ids = ",".join(resource_ids)
  
          def set_resource_ids():
              os.environ[env_var] = all_resource_ids
  
          for worker_id in node_id_to_worker_id[node_id]:
              futures.append(
                  self.worker_group.execute_single_async(
                      worker_id, set_resource_ids
                  )
              )
      ray.get(futures)
  
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 5 for calling the BackendExecutor obj's underlying member method code with example usage: obj._share_resource_ids(resource, env_var) without return types