allenact.algorithms.onpolicy_sync.vector_sampled_tasks#
VectorSampledTasks#
class VectorSampledTasks(object)
Vectorized collection of tasks. Creates multiple processes where each process runs its own TaskSampler. Each process generates one Task from its TaskSampler at a time and this class allows for interacting with these tasks in a vectorized manner. When a task on a process completes, the process samples another task from its task sampler. All the tasks are synchronized (for step and new_task methods).
Attributes
make_sampler_fn: function which creates a single TaskSampler.sampler_fn_args: sequence of dictionaries describing the args to pass to make_sampler_fn on each individual process.auto_resample_when_done: automatically sample a new Task from the TaskSampler when the Task completes. If False, a new Task will not be resampled until all Tasks on all processes have completed. This functionality is provided for seamless training of vectorized Tasks.multiprocessing_start_method: the multiprocessing method used to spawn worker processes. Valid methods are{'spawn', 'forkserver', 'fork'}'forkserver'is the recommended method as it works well with CUDA. If'fork'is used, the subproccess must be started before any other GPU useage.
VectorSampledTasks.is_closed#
| @property
| is_closed() -> bool
Has the vector task been closed.
VectorSampledTasks.num_unpaused_tasks#
| @property
| num_unpaused_tasks() -> int
Number of unpaused processes.
Returns
Number of unpaused processes.
VectorSampledTasks.mp_ctx#
| @property
| mp_ctx()
Get the multiprocessing process used by the vector task.
Returns
The multiprocessing context.
VectorSampledTasks.next_task#
| next_task(**kwargs)
Move to the the next Task for all TaskSamplers.
Parameters
- kwargs : key word arguments passed to the
next_taskfunction of the samplers.
Returns
List of initial observations for each of the new tasks.
VectorSampledTasks.get_observations#
| get_observations()
Get observations for all unpaused tasks.
Returns
List of observations for each of the unpaused tasks.
VectorSampledTasks.command_at#
| command_at(sampler_index: int, command: str, data: Optional[Any] = None) -> Any
Runs the command on the selected task and returns the result.
Parameters
Returns
Result of the command.
VectorSampledTasks.call_at#
| call_at(sampler_index: int, function_name: str, function_args: Optional[List[Any]] = None) -> Any
Calls a function (which is passed by name) on the selected task and returns the result.
Parameters
- index : Which task to call the function on.
- function_name : The name of the function to call on the task.
- function_args : Optional function args.
Returns
Result of calling the function.
VectorSampledTasks.next_task_at#
| next_task_at(sampler_index: int) -> List[RLStepResult]
Move to the the next Task from the TaskSampler in index_process process in the vector.
Parameters
- index_process : Index of the process to be reset.
Returns
List of length one containing the observations the newly sampled task.
VectorSampledTasks.step_at#
| step_at(sampler_index: int, action: Any) -> List[RLStepResult]
Step in the index_process task in the vector.
Parameters
- sampler_index : Index of the sampler to be reset.
- action : The action to take.
Returns
List containing the output of step method on the task in the indexed process.
VectorSampledTasks.async_step#
| async_step(actions: Sequence[Any]) -> None
Asynchronously step in the vectorized Tasks.
Parameters
- actions : actions to be performed in the vectorized Tasks.
VectorSampledTasks.wait_step#
| wait_step() -> List[Dict[str, Any]]
Wait until all the asynchronized processes have synchronized.
VectorSampledTasks.step#
| step(actions: Sequence[Any])
Perform actions in the vectorized tasks.
Parameters
- actions: List of size _num_samplers containing action to be taken in each task.
Returns
List of outputs from the step method of tasks.
VectorSampledTasks.reset_all#
| reset_all()
Reset all task samplers to their initial state (except for the RNG seed).
VectorSampledTasks.set_seeds#
| set_seeds(seeds: List[int])
Sets new tasks' RNG seeds.
Parameters
- seeds: List of size _num_samplers containing new RNG seeds.
VectorSampledTasks.pause_at#
| pause_at(sampler_index: int) -> None
Pauses computation on the Task in process index without destroying
the Task. This is useful for not needing to call steps on all Tasks
when only some are active (for example during the last samples of
running eval).
Parameters
- index : which process to pause. All indexes after this one will be shifted down by one.
VectorSampledTasks.resume_all#
| resume_all() -> None
Resumes any paused processes.
VectorSampledTasks.call#
| call(function_names: Union[str, List[str]], function_args_list: Optional[List[Any]] = None) -> List[Any]
Calls a list of functions (which are passed by name) on the corresponding task (by index).
Parameters
- function_names : The name of the functions to call on the tasks.
- function_args_list : List of function args for each function. If provided, len(function_args_list) should be as long as len(function_names).
Returns
List of results of calling the functions.
VectorSampledTasks.attr_at#
| attr_at(sampler_index: int, attr_name: str) -> Any
Gets the attribute (specified by name) on the selected task and returns it.
Parameters
- index : Which task to call the function on.
- attr_name : The name of the function to call on the task.
Returns
Result of calling the function.
VectorSampledTasks.attr#
| attr(attr_names: Union[List[str], str]) -> List[Any]
Gets the attributes (specified by name) on the tasks.
Parameters
- attr_names : The name of the functions to call on the tasks.
Returns
List of results of calling the functions.
VectorSampledTasks.render#
| render(mode: str = "human", *args, **kwargs) -> Union[np.ndarray, None, List[np.ndarray]]
Render observations from all Tasks in a tiled image or list of images.
SingleProcessVectorSampledTasks#
class SingleProcessVectorSampledTasks(object)
Vectorized collection of tasks.
Simultaneously handles the state of multiple TaskSamplers and their associated tasks. Allows for interacting with these tasks in a vectorized manner. When a task completes, another task is sampled from the appropriate task sampler. All the tasks are synchronized (for step and new_task methods).
Attributes
make_sampler_fn: function which creates a single TaskSampler.sampler_fn_args: sequence of dictionaries describing the args to pass to make_sampler_fn on each individual process.auto_resample_when_done: automatically sample a new Task from the TaskSampler when the Task completes. If False, a new Task will not be resampled until all Tasks on all processes have completed. This functionality is provided for seamless training of vectorized Tasks.
SingleProcessVectorSampledTasks.is_closed#
| @property
| is_closed() -> bool
Has the vector task been closed.
SingleProcessVectorSampledTasks.num_unpaused_tasks#
| @property
| num_unpaused_tasks() -> int
Number of unpaused processes.
Returns
Number of unpaused processes.
SingleProcessVectorSampledTasks.next_task#
| next_task(**kwargs)
Move to the the next Task for all TaskSamplers.
Parameters
- kwargs : key word arguments passed to the
next_taskfunction of the samplers.
Returns
List of initial observations for each of the new tasks.
SingleProcessVectorSampledTasks.get_observations#
| get_observations()
Get observations for all unpaused tasks.
Returns
List of observations for each of the unpaused tasks.
SingleProcessVectorSampledTasks.next_task_at#
| next_task_at(index_process: int) -> List[RLStepResult]
Move to the the next Task from the TaskSampler in index_process process in the vector.
Parameters
- index_process : Index of the generator to be reset.
Returns
List of length one containing the observations the newly sampled task.
SingleProcessVectorSampledTasks.step_at#
| step_at(index_process: int, action: int) -> List[RLStepResult]
Step in the index_process task in the vector.
Parameters
- index_process : Index of the process to be reset.
- action : The action to take.
Returns
List containing the output of step method on the task in the indexed process.
SingleProcessVectorSampledTasks.step#
| step(actions: List[List[int]])
Perform actions in the vectorized tasks.
Parameters
- actions: List of size _num_samplers containing action to be taken in each task.
Returns
List of outputs from the step method of tasks.
SingleProcessVectorSampledTasks.reset_all#
| reset_all()
Reset all task samplers to their initial state (except for the RNG seed).
SingleProcessVectorSampledTasks.set_seeds#
| set_seeds(seeds: List[int])
Sets new tasks' RNG seeds.
Parameters
- seeds: List of size _num_samplers containing new RNG seeds.
SingleProcessVectorSampledTasks.pause_at#
| pause_at(sampler_index: int) -> None
Pauses computation on the Task in process index without destroying
the Task. This is useful for not needing to call steps on all Tasks
when only some are active (for example during the last samples of
running eval).
Parameters
- index : which process to pause. All indexes after this one will be shifted down by one.
SingleProcessVectorSampledTasks.resume_all#
| resume_all() -> None
Resumes any paused processes.
SingleProcessVectorSampledTasks.command_at#
| command_at(sampler_index: int, command: str, data: Optional[Any] = None) -> Any
Calls a function (which is passed by name) on the selected task and returns the result.
Parameters
- index : Which task to call the function on.
- function_name : The name of the function to call on the task.
- function_args : Optional function args.
Returns
Result of calling the function.
SingleProcessVectorSampledTasks.call_at#
| call_at(sampler_index: int, function_name: str, function_args: Optional[List[Any]] = None) -> Any
Calls a function (which is passed by name) on the selected task and returns the result.
Parameters
- index : Which task to call the function on.
- function_name : The name of the function to call on the task.
- function_args : Optional function args.
Returns
Result of calling the function.
SingleProcessVectorSampledTasks.call#
| call(function_names: Union[str, List[str]], function_args_list: Optional[List[Any]] = None) -> List[Any]
Calls a list of functions (which are passed by name) on the corresponding task (by index).
Parameters
- function_names : The name of the functions to call on the tasks.
- function_args_list : List of function args for each function. If provided, len(function_args_list) should be as long as len(function_names).
Returns
List of results of calling the functions.
SingleProcessVectorSampledTasks.attr_at#
| attr_at(sampler_index: int, attr_name: str) -> Any
Gets the attribute (specified by name) on the selected task and returns it.
Parameters
- index : Which task to call the function on.
- attr_name : The name of the function to call on the task.
Returns
Result of calling the function.
SingleProcessVectorSampledTasks.attr#
| attr(attr_names: Union[List[str], str]) -> List[Any]
Gets the attributes (specified by name) on the tasks.
Parameters
- attr_names : The name of the functions to call on the tasks.
Returns
List of results of calling the functions.
SingleProcessVectorSampledTasks.render#
| render(mode: str = "human", *args, **kwargs) -> Union[np.ndarray, None, List[np.ndarray]]
Render observations from all Tasks in a tiled image or a list of images.