rllib/agents/ppo/ddppo.py
def execution_plan(
        workers: WorkerSet, config: TrainerConfigDict, **kwargs
    ) -> LocalIterator[dict]:
        """Execution plan of the DD-PPO algorithm. Defines the distributed dataflow.

        Args:
            workers (WorkerSet): The WorkerSet for training the Polic(y/ies)
                of the Trainer.
            config (TrainerConfigDict): The trainer's configuration dict.

        Returns:
            LocalIterator[dict]: The Policy class to use with PGTrainer.
                If None, use `get_default_policy_class()` provided by Trainer.
        """
        assert (
            len(kwargs) == 0
        ), "DDPPO execution_plan does NOT take any additional parameters"

        rollouts = ParallelRollouts(workers, mode="raw")

        # Setup the distributed processes.
        ip = ray.get(workers.remote_workers()[0].get_node_ip.remote())
        port = ray.get(workers.remote_workers()[0].find_free_port.remote())
        address = "tcp://{ip}:{port}".format(ip=ip, port=port)
        logger.info("Creating torch process group with leader {}".format(address))

        # Get setup tasks in order to throw errors on failure.
        ray.get(
            [
                worker.setup_torch_data_parallel.remote(
                    url=address,
                    world_rank=i,
                    world_size=len(workers.remote_workers()),
                    backend=config["torch_distributed_backend"],
                )
                for i, worker in enumerate(workers.remote_workers())
            ]
        )
        logger.info("Torch process group init completed")

        # This function is applied remotely on each rollout worker.
        def train_torch_distributed_allreduce(batch):
            expected_batch_size = (
                config["rollout_fragment_length"] * config["num_envs_per_worker"]
            )
            this_worker = get_global_worker()
            assert batch.count == expected_batch_size, (
                "Batch size possibly out of sync between workers, expected:",
                expected_batch_size,
                "got:",
                batch.count,
            )
            logger.info(
                "Executing distributed minibatch SGD "
                "with epoch size {}, minibatch size {}".format(
                    batch.count, config["sgd_minibatch_size"]
                )
            )
            info = do_minibatch_sgd(
                batch,
                this_worker.policy_map,
                this_worker,
                config["num_sgd_iter"],
                config["sgd_minibatch_size"],
                ["advantages"],
            )
            return info, batch.count

        # Broadcast the local set of global vars.
        def update_worker_global_vars(item):
            global_vars = _get_global_vars()
            for w in workers.remote_workers():
                w.set_global_vars.remote(global_vars)
            return item

        # Have to manually record stats since we are using "raw" rollouts mode.
        class RecordStats:
            def _on_fetch_start(self):
                self.fetch_start_time = time.perf_counter()

            def __call__(self, items):
                assert len(items) == config["num_workers"]
                for item in items:
                    info, count = item
                    metrics = _get_shared_metrics()
                    metrics.counters[STEPS_TRAINED_THIS_ITER_COUNTER] = count
                    metrics.counters[STEPS_SAMPLED_COUNTER] += count
                    metrics.counters[STEPS_TRAINED_COUNTER] += count
                    metrics.info[LEARNER_INFO] = info
                # Since SGD happens remotely, the time delay between fetch and
                # completion is approximately the SGD step time.
                metrics.timers[LEARN_ON_BATCH_TIMER].push(
                    time.perf_counter() - self.fetch_start_time
                )

        train_op = (
            rollouts.for_each(train_torch_distributed_allreduce)  # allreduce
            .batch_across_shards()  # List[(grad_info, count)]
            .for_each(RecordStats())
        )

        train_op = train_op.for_each(update_worker_global_vars)

        # Sync down the weights. As with the sync up, this is not really
        # needed unless the user is reading the local weights.
        if config["keep_local_weights_in_sync"]:

            def download_weights(item):
                workers.local_worker().set_weights(
                    ray.get(workers.remote_workers()[0].get_weights.remote())
                )
                return item

            train_op = train_op.for_each(download_weights)

        # In debug mode, check the allreduce successfully synced the weights.
        if logger.isEnabledFor(logging.DEBUG):

            def check_sync(item):
                weights = ray.get(
                    [w.get_weights.remote() for w in workers.remote_workers()]
                )
                sums = []
                for w in weights:
                    acc = 0
                    for p in w.values():
                        for k, v in p.items():
                            acc += v.sum()
                    sums.append(float(acc))
                logger.debug("The worker weight sums are {}".format(sums))
                assert len(set(sums)) == 1, sums

            train_op = train_op.for_each(check_sync)

        return StandardMetricsReporting(train_op, workers, config)
Execution plan of the DD-PPO algorithm. Defines the distributed dataflow.

        Args:
            workers (WorkerSet): The WorkerSet for training the Polic(y/ies)
                of the Trainer.
            config (TrainerConfigDict): The trainer's configuration dict.

        Returns:
            LocalIterator[dict]: The Policy class to use with PGTrainer.
                If None, use `get_default_policy_class()` provided by Trainer.
        
Execution plan of the DD-PPO algorithm. Defines the distributed dataflow.
Execution plan of the DD - PPO algorithm . Defines the distributed dataflow .
def execution_plan(
        workers: WorkerSet, config: TrainerConfigDict, **kwargs
    ) -> LocalIterator[dict]:
        assert (
            len(kwargs) == 0
        ), "DDPPO execution_plan does NOT take any additional parameters"
        rollouts = ParallelRollouts(workers, mode="raw")
        ip = ray.get(workers.remote_workers()[0].get_node_ip.remote())
        port = ray.get(workers.remote_workers()[0].find_free_port.remote())
        address = "tcp://{ip}:{port}".format(ip=ip, port=port)
        logger.info("Creating torch process group with leader {}".format(address))
        ray.get(
            [
                worker.setup_torch_data_parallel.remote(
                    url=address,
                    world_rank=i,
                    world_size=len(workers.remote_workers()),
                    backend=config["torch_distributed_backend"],
                )
                for i, worker in enumerate(workers.remote_workers())
            ]
        )
        logger.info("Torch process group init completed")
        def train_torch_distributed_allreduce(batch):
            expected_batch_size = (
                config["rollout_fragment_length"] * config["num_envs_per_worker"]
            )
            this_worker = get_global_worker()
            assert batch.count == expected_batch_size, (
                "Batch size possibly out of sync between workers, expected:",
                expected_batch_size,
                "got:",
                batch.count,
            )
            logger.info(
                "Executing distributed minibatch SGD "
                "with epoch size {}, minibatch size {}".format(
                    batch.count, config["sgd_minibatch_size"]
                )
            )
            info = do_minibatch_sgd(
                batch,
                this_worker.policy_map,
                this_worker,
                config["num_sgd_iter"],
                config["sgd_minibatch_size"],
                ["advantages"],
            )
            return info, batch.count
        def update_worker_global_vars(item):
            global_vars = _get_global_vars()
            for w in workers.remote_workers():
                w.set_global_vars.remote(global_vars)
            return item
        class RecordStats:
            def _on_fetch_start(self):
                self.fetch_start_time = time.perf_counter()
            def __call__(self, items):
                assert len(items) == config["num_workers"]
                for item in items:
                    info, count = item
                    metrics = _get_shared_metrics()
                    metrics.counters[STEPS_TRAINED_THIS_ITER_COUNTER] = count
                    metrics.counters[STEPS_SAMPLED_COUNTER] += count
                    metrics.counters[STEPS_TRAINED_COUNTER] += count
                    metrics.info[LEARNER_INFO] = info
                metrics.timers[LEARN_ON_BATCH_TIMER].push(
                    time.perf_counter() - self.fetch_start_time
                )
        train_op = (
            rollouts.for_each(train_torch_distributed_allreduce)  
            .batch_across_shards()  
            .for_each(RecordStats())
        )
        train_op = train_op.for_each(update_worker_global_vars)
        if config["keep_local_weights_in_sync"]:
            def download_weights(item):
                workers.local_worker().set_weights(
                    ray.get(workers.remote_workers()[0].get_weights.remote())
                )
                return item
            train_op = train_op.for_each(download_weights)
        if logger.isEnabledFor(logging.DEBUG):
            def check_sync(item):
                weights = ray.get(
                    [w.get_weights.remote() for w in workers.remote_workers()]
                )
                sums = []
                for w in weights:
                    acc = 0
                    for p in w.values():
                        for k, v in p.items():
                            acc += v.sum()
                    sums.append(float(acc))
                logger.debug("The worker weight sums are {}".format(sums))
                assert len(set(sums)) == 1, sums
            train_op = train_op.for_each(check_sync)
        return StandardMetricsReporting(train_op, workers, config)
def execution_plan ( workers : WorkerSet , config : TrainerConfigDict , ** kwargs ) -> LocalIterator [ dict ] : assert ( len ( kwargs ) == 0 ) , "DDPPO execution_plan does NOT take any additional parameters" rollouts = ParallelRollouts ( workers , mode = "raw" ) ip = ray . get ( workers . remote_workers ( ) [ 0 ] . get_node_ip . remote ( ) ) port = ray . get ( workers . remote_workers ( ) [ 0 ] . find_free_port . remote ( ) ) address = "tcp://{ip}:{port}" . format ( ip = ip , port = port ) logger . info ( "Creating torch process group with leader {}" . format ( address ) ) ray . get ( [ worker . setup_torch_data_parallel . remote ( url = address , world_rank = i , world_size = len ( workers . remote_workers ( ) ) , backend = config [ "torch_distributed_backend" ] , ) for i , worker in enumerate ( workers . remote_workers ( ) ) ] ) logger . info ( "Torch process group init completed" ) def train_torch_distributed_allreduce ( batch ) : expected_batch_size = ( config [ "rollout_fragment_length" ] * config [ "num_envs_per_worker" ] ) this_worker = get_global_worker ( ) assert batch . count == expected_batch_size , ( "Batch size possibly out of sync between workers, expected:" , expected_batch_size , "got:" , batch . count , ) logger . info ( "Executing distributed minibatch SGD " "with epoch size {}, minibatch size {}" . format ( batch . count , config [ "sgd_minibatch_size" ] ) ) info = do_minibatch_sgd ( batch , this_worker . policy_map , this_worker , config [ "num_sgd_iter" ] , config [ "sgd_minibatch_size" ] , [ "advantages" ] , ) return info , batch . count def update_worker_global_vars ( item ) : global_vars = _get_global_vars ( ) for w in workers . remote_workers ( ) : w . set_global_vars . remote ( global_vars ) return item class RecordStats : def _on_fetch_start ( self ) : self . fetch_start_time = time . perf_counter ( ) def __call__ ( self , items ) : assert len ( items ) == config [ "num_workers" ] for item in items : info , count = item metrics = _get_shared_metrics ( ) metrics . counters [ STEPS_TRAINED_THIS_ITER_COUNTER ] = count metrics . counters [ STEPS_SAMPLED_COUNTER ] += count metrics . counters [ STEPS_TRAINED_COUNTER ] += count metrics . info [ LEARNER_INFO ] = info metrics . timers [ LEARN_ON_BATCH_TIMER ] . push ( time . perf_counter ( ) - self . fetch_start_time ) train_op = ( rollouts . for_each ( train_torch_distributed_allreduce ) . batch_across_shards ( ) . for_each ( RecordStats ( ) ) ) train_op = train_op . for_each ( update_worker_global_vars ) if config [ "keep_local_weights_in_sync" ] : def download_weights ( item ) : workers . local_worker ( ) . set_weights ( ray . get ( workers . remote_workers ( ) [ 0 ] . get_weights . remote ( ) ) ) return item train_op = train_op . for_each ( download_weights ) if logger . isEnabledFor ( logging . DEBUG ) : def check_sync ( item ) : weights = ray . get ( [ w . get_weights . remote ( ) for w in workers . remote_workers ( ) ] ) sums = [ ] for w in weights : acc = 0 for p in w . values ( ) : for k , v in p . items ( ) : acc += v . sum ( ) sums . append ( float ( acc ) ) logger . debug ( "The worker weight sums are {}" . format ( sums ) ) assert len ( set ( sums ) ) == 1 , sums train_op = train_op . for_each ( check_sync ) return StandardMetricsReporting ( train_op , workers , config )
Execution plan of the DD-PPO algorithm.
Execution plan of the DD - PPO algorithm .
"""Execution plan of the DD-PPO algorithm. Defines the distributed dataflow.

        Args:
            workers (WorkerSet): The WorkerSet for training the Polic(y/ies)
                of the Trainer.
            config (TrainerConfigDict): The trainer's configuration dict.

        Returns:
            LocalIterator[dict]: The Policy class to use with PGTrainer.
                If None, use `get_default_policy_class()` provided by Trainer.
        """ # Setup the distributed processes. # Get setup tasks in order to throw errors on failure. # This function is applied remotely on each rollout worker. # Broadcast the local set of global vars. # Have to manually record stats since we are using "raw" rollouts mode. # Since SGD happens remotely, the time delay between fetch and # completion is approximately the SGD step time. # allreduce # List[(grad_info, count)] # Sync down the weights. As with the sync up, this is not really # needed unless the user is reading the local weights. # In debug mode, check the allreduce successfully synced the weights.
The Policy class to use with PGTrainer.
If None, use `get_default_policy_class()` provided by Trainer.
The Policy class to use with PGTrainer . If None use ` get_default_policy_class () ` provided by Trainer .
LocalIterator[dict]
workers
WorkerSet
The WorkerSet for training the Polic(y/ies)
of the Trainer.
The WorkerSet for training the Polic ( y / ies ) of the Trainer .
config
TrainerConfigDict
The trainer's configuration dict.
The trainer ' s configuration dict .