ray.rllib.evaluation.worker_set.WorkerSet#

class ray.rllib.evaluation.worker_set.WorkerSet(*, env_creator: Optional[Callable[[EnvContext], Optional[Any]]] = None, validate_env: Optional[Callable[[Any], None]] = None, default_policy_class: Optional[Type[ray.rllib.policy.policy.Policy]] = None, config: Optional[AlgorithmConfig] = None, num_workers: int = 0, local_worker: bool = True, logdir: Optional[str] = None, _setup: bool = True, policy_class=- 1, trainer_config=- 1)[source]#

Bases: object

Set of RolloutWorkers with n @ray.remote workers and zero or one local worker.

Where: n >= 0.

Methods

__init__(*[, env_creator, validate_env, ...])

Initializes a WorkerSet instance.

add_policy(policy_id[, policy_cls, policy, ...])

Adds a policy to this WorkerSet's workers or a specific list of workers.

add_workers(num_workers[, validate])

Creates and adds a number of remote workers to this worker set.

fetch_ready_async_reqs(*[, timeout_seconds, ...])

Get esults from outstanding asynchronous requests that are ready.

foreach_env(func)

Calls func with all workers' sub-environments as args.

foreach_env_with_context(func)

Calls func with all workers' sub-environments and env_ctx as args.

foreach_policy(func)

Calls func with each worker's (policy, PolicyID) tuple.

foreach_policy_to_train(func)

Apply func to all workers' Policies iff in policies_to_train.

foreach_worker(func, *[, local_worker, ...])

Calls the given function with each worker instance as the argument.

foreach_worker_async(func, *[, ...])

Calls the given function asynchronously with each worker as the argument.

foreach_worker_with_id(func, *[, ...])

Similar to foreach_worker(), but calls the function with id of the worker too.

healthy_worker_ids()

Returns the list of remote worker IDs.

is_policy_to_train(policy_id[, batch])

Whether given PolicyID (optionally inside some batch) is trainable.

local_worker()

Returns the local rollout worker.

num_healthy_remote_workers()

Returns the number of healthy workers, including local and remote workers.

num_healthy_workers()

Returns the number of healthy workers, including local and remote workers.

num_in_flight_async_reqs()

Returns the number of in-flight async requests.

num_remote_worker_restarts()

Total number of times managed remote workers get restarted.

num_remote_workers()

Returns the number of remote rollout workers.

probe_unhealthy_workers()

Checks the unhealth workers, and try restoring their states.

reset(new_remote_workers)

Hard overrides the remote workers in this set with the given one.

stop()

Calls stop on all rollout workers (including the local one).

sync_weights([policies, ...])

Syncs model weights from the given weight source to all remote workers.