code: def training_step(self) -> ResultDict:
      self.workers.foreach_worker_async(
          func=self._sample_and_train_torch_distributed,
          healthy_only=False,
      )
      sample_and_update_results = self.workers.fetch_ready_async_reqs(
          timeout_seconds=0.03
      )
  
      # For all results collected:
      # - Update our counters and timers.
      # - Update the worker's global_vars.
      # - Build info dict using a LearnerInfoBuilder object.
      learner_info_builder = LearnerInfoBuilder(num_devices=1)
      sampled_workers = set()
      for worker_id, result in sample_and_update_results:
          sampled_workers.add(worker_id)
  
          self._counters[NUM_AGENT_STEPS_SAMPLED] += result[
              "agent_steps"
          ]
          self._counters[NUM_AGENT_STEPS_TRAINED] += result[
              "agent_steps"
          ]
          self._counters[NUM_ENV_STEPS_SAMPLED] += result["env_steps"]
          self._counters[NUM_ENV_STEPS_TRAINED] += result["env_steps"]
          self._timers[LEARN_ON_BATCH_TIMER].push(
              result["learn_on_batch_time"]
          )
          self._timers[SAMPLE_TIMER].push(result["sample_time"])
  
          # Add partial learner info to builder object.
          learner_info_builder.add_learn_on_batch_results_multi_agent(
              result["info"]
          )
  
      # Broadcast the local set of global vars to this worker.
      global_vars = {
          "timestep": self._counters[NUM_AGENT_STEPS_SAMPLED]
      }
      self.workers.foreach_worker(
          func=lambda w: w.set_global_vars(global_vars),
          local_worker=False,
          remote_worker_ids=list(sampled_workers),
          timeout_seconds=0,  # Don't wait for workers to finish.
      )
  
      # Sync down the weights from 1st remote worker (only if we have received
      # some results from it).
      # As with the sync up, this is not really needed unless the user is
      # reading the local weights.
      worker_ids = self.workers.healthy_worker_ids()
      assert worker_ids, "No healthy rollout workers?"
      if (
          self.config.keep_local_weights_in_sync
          and worker_ids[0] in sampled_workers
      ):
          weights = self.workers.foreach_worker(
              func=lambda w: w.get_weights(),
              local_worker=False,
              remote_worker_ids=[worker_ids[0]],
          )
          self.workers.local_worker().set_weights(weights[0])
      # Return merged laarner into results.
      new_learner_info = learner_info_builder.finalize()
      if new_learner_info:
          self._curr_learner_info = new_learner_info
      return self._curr_learner_info
  
imports: import logging
  import time
  from typing import Optional
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig, NotProvided
  from ray.rllib.algorithms.ppo import PPO, PPOConfig
  from ray.rllib.evaluation.postprocessing import Postprocessing
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.metrics import LEARN_ON_BATCH_TIMER, NUM_AGENT_STEPS_SAMPLED, NUM_AGENT_STEPS_TRAINED, NUM_ENV_STEPS_SAMPLED, NUM_ENV_STEPS_TRAINED, SAMPLE_TIMER
  from ray.rllib.utils.metrics.learner_info import LearnerInfoBuilder
  from ray.rllib.utils.sgd import do_minibatch_sgd
  from ray.rllib.utils.typing import ResultDict
label_desc: Use image node_id 3 for calling the DDPPO obj's underlying member method code with example usage: obj.training_step() and returns: self