WorkerSet¶

A set of RolloutWorker containing n ray remote workers as well as a single “local” RolloutWorker .

WorkerSet exposes some convenience methods to make calls on its individual workers’ own methods in parallel using e.g. ray.get().

class ray.rllib.evaluation.worker_set.WorkerSet(*, env_creator: Optional[Callable[[ray.rllib.env.env_context.EnvContext], Optional[Any]]] = None, validate_env: Optional[Callable[[Any], None]] = None, policy_class: Optional[Type[ray.rllib.policy.policy.Policy]] = None, trainer_config: Optional[dict] = None, num_workers: int = 0, local_worker: bool = True, logdir: Optional[str] = None, _setup: bool = True)[source]¶

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

Where: n >= 0.

__init__(*, env_creator: Optional[Callable[[ray.rllib.env.env_context.EnvContext], Optional[Any]]] = None, validate_env: Optional[Callable[[Any], None]] = None, policy_class: Optional[Type[ray.rllib.policy.policy.Policy]] = None, trainer_config: Optional[dict] = None, num_workers: int = 0, local_worker: bool = True, logdir: Optional[str] = None, _setup: bool = True)[source]¶

Initializes a WorkerSet instance.

Parameters

  • env_creator – Function that returns env given env config.

  • validate_env – Optional callable to validate the generated environment (only on worker=0).

  • policy_class – An optional Policy class. If None, PolicySpecs can be generated automatically by using the Algorithm’s default class of via a given multi-agent policy config dict.

  • trainer_config – Optional dict that extends the common config of the Algorithm class.

  • 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 setup workers. This is only for testing.

local_worker() → ray.rllib.evaluation.rollout_worker.RolloutWorker[source]¶

Returns the local rollout worker.

remote_workers() → List[ray.actor.ActorHandle][source]¶

Returns a list of remote rollout workers.

sync_weights(policies: Optional[List[str]] = None, from_worker: Optional[ray.rllib.evaluation.rollout_worker.RolloutWorker] = None, global_vars: Optional[Dict[str, Union[numpy.array, tf.Tensor, torch.Tensor]]] = None) → None[source]¶

Syncs model weights from the local worker to all remote workers.

Parameters

  • policies – Optional list of PolicyIDs to sync weights for. If None (default), sync weights to/from all policies.

  • from_worker – Optional RolloutWorker instance to sync from. If None (default), sync from this WorkerSet’s local worker.

  • global_vars – An optional global vars dict to set this worker to. If None, do not update the global_vars.

add_workers(num_workers: int, validate: bool = False) → None[source]¶

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

Can be called several times on the same WorkerSet to add more RolloutWorkers to the set.

Parameters

  • num_workers – The number of remote Workers to add to this WorkerSet.

  • validate – Whether to validate remote workers after their construction process.

Raises

  • RayError – If any of the constructed remote workers is not up and running

  • properly. –

reset(new_remote_workers: List[ray.actor.ActorHandle]) → None[source]¶

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

Parameters

new_remote_workers – A list of new RolloutWorkers (as ActorHandles) to use as remote workers.

recreate_failed_workers(local_worker_for_synching: ray.rllib.evaluation.rollout_worker.RolloutWorker) → Tuple[List[ray.actor.ActorHandle], List[ray.actor.ActorHandle]][source]¶

Recreates any failed workers (after health check).

Parameters

local_worker_for_synching – RolloutWorker to use to synchronize the weights after recreation.

Returns

The list of removed workers and the list of newly added ones.

Return type

A tuple consisting of two items

stop() → None[source]¶

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

is_policy_to_train(policy_id: str, batch: Optional[Union[SampleBatch, MultiAgentBatch]] = None) → bool[source]¶

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

foreach_worker(func: Callable[[ray.rllib.evaluation.rollout_worker.RolloutWorker], ray.rllib.evaluation.worker_set.T]) → List[ray.rllib.evaluation.worker_set.T][source]¶

Calls the given function with each worker instance as arg.

Parameters

func – The function to call for each worker (as only arg).

Returns

The list of return values of all calls to func([worker]).

foreach_worker_with_index(func: Callable[[ray.rllib.evaluation.rollout_worker.RolloutWorker, int], ray.rllib.evaluation.worker_set.T]) → List[ray.rllib.evaluation.worker_set.T][source]¶

Calls func with each worker instance and worker idx as args.

The index will be passed as the second arg to the given function.

Parameters

func – The function to call for each worker and its index (as args). The local worker has index 0, all remote workers have indices > 0.

Returns

The list of return values of all calls to func([worker, idx]).

The first entry in this list are the results of the local worker, followed by all remote workers’ results.

foreach_policy(func: Callable[[ray.rllib.policy.policy.Policy, str], ray.rllib.evaluation.worker_set.T]) → List[ray.rllib.evaluation.worker_set.T][source]¶

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

Note that in the multi-agent case, each worker may have more than one policy.

Parameters

func – A function - taking a Policy and its ID - that is called on all workers’ Policies.

Returns

The list of return values of func over all workers’ policies. The

length of this list is: (num_workers + 1 (local-worker)) * [num policies in the multi-agent config dict]. The local workers’ results are first, followed by all remote workers’ results

foreach_policy_to_train(func: Callable[[ray.rllib.policy.policy.Policy, str], ray.rllib.evaluation.worker_set.T]) → List[ray.rllib.evaluation.worker_set.T][source]¶

Apply func to all workers’ Policies iff in policies_to_train.

Parameters

func – A function - taking a Policy and its ID - that is called on all workers’ Policies, for which worker.is_policy_to_train() returns True.

Returns

The list of n return values of all

func([trainable policy], [ID])-calls.

Return type

List[any]

foreach_env(func: Callable[[Any], List[ray.rllib.evaluation.worker_set.T]]) → List[List[ray.rllib.evaluation.worker_set.T]][source]¶

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

An “underlying sub environment” is a single clone of an env within a vectorized environment. func takes a single underlying sub environment as arg, e.g. a gym.Env object.

Parameters

func – A function - taking an EnvType (normally a gym.Env object) as arg and returning a list of lists of return values, one value per underlying sub-environment per each worker.

Returns

The list (workers) of lists (sub environments) of results.

foreach_env_with_context(func: Callable[[ray.rllib.env.base_env.BaseEnv, ray.rllib.env.env_context.EnvContext], List[ray.rllib.evaluation.worker_set.T]]) → List[List[ray.rllib.evaluation.worker_set.T]][source]¶

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

An “underlying sub environment” is a single clone of an env within a vectorized environment. func takes a single underlying sub environment and the env_context as args.

Parameters

func – A function - taking a BaseEnv object and an EnvContext as arg - and returning a list of lists of return values over envs of the worker.

Returns

The list (1 item per workers) of lists (1 item per sub-environment)

of results.

previous

Sample Batches

next

Environment Samplers