class_docstr: Set of EnvRunners with n @ray.remote workers and zero or one local worker.
  
  Where: n >= 0.
class_docstr_tok: ["Set","of","EnvRunners","with","n","@","ray.remote","workers","and","zero","or","one","local","worker",".","Where",":","n",">","=","0","."]
docstr: Initializes a WorkerSet instance.
  
  Args:
      env_creator: Function that returns env given env config.
      validate_env: Optional callable to validate the generated
          environment (only on worker=0). This callable should raise
          an exception if the environment is invalid.
      default_policy_class: An optional default Policy class to use inside
          the (multi-agent) `policies` dict. In case the PolicySpecs in there
          have no class defined, use this `default_policy_class`.
          If None, PolicySpecs will be using the Algorithm's default Policy
          class.
      config: Optional AlgorithmConfig (or config dict).
      num_workers: Number of remote rollout workers to create.
      local_worker: Whether to create a local (non @ray.remote) worker
          in the returned set as well (default: True). If `num_workers`
          is 0, always create a local worker.
      logdir: Optional logging directory for workers.
      _setup: Whether to actually set up workers. This is only for testing.
docstr_tok: ["Initializes","a","WorkerSet","instance",".","Args",":","env_creator",":","Function","that","returns","env","given","env","config",".","validate_env",":","Optional","callable","to","validate","the","generated","environment","(","only","on","worker=0",")",".","This","callable","should","raise","an","exception","if","the","environment","is","invalid",".","default_policy_class",":","An","optional","default","Policy","class","to","use","inside","the","(","multi-agent",")","`","policies","`","dict",".","In","case","the","PolicySpecs","in","there","have","no","class","defined",",","use","this","`","default_policy_class","`",".","If","None",",","PolicySpecs","will","be","using","the","Algorithm","'s","default","Policy","class",".","config",":","Optional","AlgorithmConfig","(","or","config","dict",")",".","num_workers",":","Number","of","remote","rollout","workers","to","create",".","local_worker",":","Whether","to","create","a","local","(","non","@","ray.remote",")","worker","in","the","returned","set","as","well","(","default",":","True",")",".","If","`","num_workers","`","is","0",",","always","create","a","local","worker",".","logdir",":","Optional","logging","directory","for","workers",".","_setup",":","Whether","to","actually","set","up","workers",".","This","is","only","for","testing","."]
code: def __init__(
      self,
      *,
      env_creator: Optional[EnvCreator] = None,
      validate_env: Optional[Callable[[EnvType], None]] = None,
      default_policy_class: Optional[Type[Policy]] = None,
      config: Optional["AlgorithmConfig"] = None,
      num_workers: int = 0,
      local_worker: bool = True,
      logdir: Optional[str] = None,
      _setup: bool = True,
  ):
      """Initializes a WorkerSet instance.
  
      Args:
          env_creator: Function that returns env given env config.
          validate_env: Optional callable to validate the generated
              environment (only on worker=0). This callable should raise
              an exception if the environment is invalid.
          default_policy_class: An optional default Policy class to use inside
              the (multi-agent) `policies` dict. In case the PolicySpecs in there
              have no class defined, use this `default_policy_class`.
              If None, PolicySpecs will be using the Algorithm's default Policy
              class.
          config: Optional AlgorithmConfig (or config dict).
          num_workers: Number of remote rollout workers to create.
          local_worker: Whether to create a local (non @ray.remote) worker
              in the returned set as well (default: True). If `num_workers`
              is 0, always create a local worker.
          logdir: Optional logging directory for workers.
          _setup: Whether to actually set up workers. This is only for testing.
      """
      from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  
      # Make sure `config` is an AlgorithmConfig object.
      if not config:
          config = AlgorithmConfig()
      elif isinstance(config, dict):
          config = AlgorithmConfig.from_dict(config)
  
      self._env_creator = env_creator
      self._policy_class = default_policy_class
      self._remote_config = config
      self._remote_args = {
          "num_cpus": self._remote_config.num_cpus_per_worker,
          "num_gpus": self._remote_config.num_gpus_per_worker,
          "resources": self._remote_config.custom_resources_per_worker,
          "max_restarts": config.max_num_worker_restarts,
      }
  
      # Set the EnvRunner subclass to be used as "workers". Default: RolloutWorker.
      self.env_runner_cls = (
          RolloutWorker
          if config.env_runner_cls is None
          else config.env_runner_cls
      )
      self._cls = ray.remote(**self._remote_args)(
          self.env_runner_cls
      ).remote
  
      self._logdir = logdir
      self._ignore_worker_failures = config["ignore_worker_failures"]
  
      # Create remote worker manager.
      # Note(jungong) : ID 0 is used by the local worker.
      # Starting remote workers from ID 1 to avoid conflicts.
      self.__worker_manager = FaultTolerantActorManager(
          max_remote_requests_in_flight_per_actor=(
              config["max_requests_in_flight_per_sampler_worker"]
          ),
          init_id=1,
      )
  
      if _setup:
          try:
              self._setup(
                  validate_env=validate_env,
                  config=config,
                  num_workers=num_workers,
                  local_worker=local_worker,
              )
          # WorkerSet creation possibly fails, if some (remote) workers cannot
          # be initialized properly (due to some errors in the EnvRunners's
          # constructor).
          except RayActorError as e:
              # In case of an actor (remote worker) init failure, the remote worker
              # may still exist and will be accessible, however, e.g. calling
              # its `sample.remote()` would result in strange "property not found"
              # errors.
              if e.actor_init_failed:
                  # Raise the original error here that the EnvRunners raised
                  # during its construction process. This is to enforce transparency
                  # for the user (better to understand the real reason behind the
                  # failure).
                  # - e.args[0]: The RayTaskError (inside the caught RayActorError).
                  # - e.args[0].args[2]: The original Exception (e.g. a ValueError due
                  # to a config mismatch) thrown inside the actor.
                  raise e.args[0].args[2]
              # In any other case, raise the RayActorError as-is.
              else:
                  raise e
  
imports: import functools
  import gymnasium
  import logging
  import importlib.util
  import os
  from typing import Callable, Container, Dict, List, Optional, Tuple, Type, TYPE_CHECKING, TypeVar, Union
  import ray
  from ray.actor import ActorHandle
  from ray.exceptions import RayActorError
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.utils.actor_manager import RemoteCallResults
  from ray.rllib.env.base_env import BaseEnv
  from ray.rllib.env.env_context import EnvContext
  from ray.rllib.env.env_runner import EnvRunner
  from ray.rllib.offline import get_dataset_and_shards
  from ray.rllib.policy.policy import Policy, PolicyState
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID
  from ray.rllib.utils.actor_manager import FaultTolerantActorManager
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.deprecation import Deprecated, deprecation_warning, DEPRECATED_VALUE
  from ray.rllib.utils.framework import try_import_tf
  from ray.rllib.utils.policy import validate_policy_id
  from ray.rllib.utils.typing import AgentID, EnvCreator, EnvType, EpisodeID, PartialAlgorithmConfigDict, PolicyID, SampleBatchType, TensorType
label_desc: Use image node_id 1 to create a new WorkerSet object  with example: obj = WorkerSet()