From daa03ba6e69b1ff3a2965dda12fc4904f21137af Mon Sep 17 00:00:00 2001 From: Eric Liang Date: Mon, 21 Sep 2020 23:03:06 -0700 Subject: [PATCH] [rllib] Add execution module to package ref (#10941) * add init * add * update --- rllib/examples/saving_experiences.py | 1 + rllib/execution/__init__.py | 38 ++++++++++++++++++++++++++++ rllib/execution/concurrency_ops.py | 20 +++++++-------- rllib/execution/rollout_ops.py | 15 +++++------ 4 files changed, 56 insertions(+), 18 deletions(-) diff --git a/rllib/examples/saving_experiences.py b/rllib/examples/saving_experiences.py index 544127b89..c066097cd 100644 --- a/rllib/examples/saving_experiences.py +++ b/rllib/examples/saving_experiences.py @@ -42,6 +42,7 @@ if __name__ == "__main__": obs=prep.transform(obs), actions=action, action_prob=1.0, # put the true action probability here + action_logp=0.0, rewards=rew, prev_actions=prev_action, prev_rewards=prev_reward, diff --git a/rllib/execution/__init__.py b/rllib/execution/__init__.py index e69de29bb..0b007c023 100644 --- a/rllib/execution/__init__.py +++ b/rllib/execution/__init__.py @@ -0,0 +1,38 @@ +from ray.rllib.execution.concurrency_ops import Concurrently, Enqueue, Dequeue +from ray.rllib.execution.metric_ops import StandardMetricsReporting, \ + CollectMetrics, OncePerTimeInterval, OncePerTimestepsElapsed +from ray.rllib.execution.replay_buffer import ReplayBuffer, \ + PrioritizedReplayBuffer +from ray.rllib.execution.replay_ops import StoreToReplayBuffer, Replay, \ + SimpleReplayBuffer, MixInReplay +from ray.rllib.execution.rollout_ops import ParallelRollouts, AsyncGradients, \ + ConcatBatches, SelectExperiences, StandardizeFields +from ray.rllib.execution.train_ops import TrainOneStep, TrainTFMultiGPU, \ + ComputeGradients, ApplyGradients, AverageGradients, UpdateTargetNetwork + +__all__ = [ + "ApplyGradients", + "AsyncGradients", + "AverageGradients", + "CollectMetrics", + "ComputeGradients", + "ConcatBatches", + "Concurrently", + "Dequeue", + "Enqueue", + "MixInReplay", + "OncePerTimeInterval", + "OncePerTimestepsElapsed", + "ParallelRollouts", + "PrioritizedReplayBuffer", + "Replay", + "ReplayBuffer", + "SelectExperiences", + "SimpleReplayBuffer", + "StandardMetricsReporting", + "StandardizeFields", + "StoreToReplayBuffer", + "TrainOneStep", + "TrainTFMultiGPU", + "UpdateTargetNetwork", +] diff --git a/rllib/execution/concurrency_ops.py b/rllib/execution/concurrency_ops.py index 3979cb65c..7b057852e 100644 --- a/rllib/execution/concurrency_ops.py +++ b/rllib/execution/concurrency_ops.py @@ -13,21 +13,21 @@ def Concurrently(ops: List[LocalIterator], """Operator that runs the given parent iterators concurrently. Args: - mode (str): One of {'round_robin', 'async'}. - - In 'round_robin' mode, we alternate between pulling items from - each parent iterator in order deterministically. - - In 'async' mode, we pull from each parent iterator as fast as - they are produced. This is non-deterministic. + mode (str): One of 'round_robin', 'async'. In 'round_robin' mode, + we alternate between pulling items from each parent iterator in + order deterministically. In 'async' mode, we pull from each parent + iterator as fast as they are produced. This is non-deterministic. output_indexes (list): If specified, only output results from the - given ops. For example, if output_indexes=[0], only results from - the first op in ops will be returned. + given ops. For example, if ``output_indexes=[0]``, only results + from the first op in ops will be returned. round_robin_weights (list): List of weights to use for round robin - mode. For example, [2, 1] will cause the iterator to pull twice - as many items from the first iterator as the second. [2, 1, *] will - cause as many items to be pulled as possible from the third + mode. For example, ``[2, 1]`` will cause the iterator to pull twice + as many items from the first iterator as the second. ``[2, 1, *]`` + will cause as many items to be pulled as possible from the third iterator without blocking. This is only allowed in round robin mode. + Examples: >>> sim_op = ParallelRollouts(...).for_each(...) >>> replay_op = LocalReplay(...).for_each(...) >>> combined_op = Concurrently([sim_op, replay_op], mode="async") diff --git a/rllib/execution/rollout_ops.py b/rllib/execution/rollout_ops.py index ed07c7480..818254c2d 100644 --- a/rllib/execution/rollout_ops.py +++ b/rllib/execution/rollout_ops.py @@ -27,14 +27,13 @@ def ParallelRollouts(workers: WorkerSet, *, mode="bulk_sync", Args: workers (WorkerSet): set of rollout workers to use. - mode (str): One of {'async', 'bulk_sync', 'raw'}. - - In 'async' mode, batches are returned as soon as they are - computed by rollout workers with no order guarantees. - - In 'bulk_sync' mode, we collect one batch from each worker - and concatenate them together into a large batch to return. - - In 'raw' mode, the ParallelIterator object is returned directly - and the caller is responsible for implementing gather and - updating the timesteps counter. + mode (str): One of 'async', 'bulk_sync', 'raw'. In 'async' mode, + batches are returned as soon as they are computed by rollout + workers with no order guarantees. In 'bulk_sync' mode, we collect + one batch from each worker and concatenate them together into a + large batch to return. In 'raw' mode, the ParallelIterator object + is returned directly and the caller is responsible for implementing + gather and updating the timesteps counter. num_async (int): In async mode, the max number of async requests in flight per actor.