code: def training_step(self) -> ResultDict:
      # Shortcut.
      local_worker = self.workers.local_worker()
  
      # Define the function executed in parallel by all RolloutWorkers to collect
      # samples + compute and return gradients (and other information).
  
      def sample_and_compute_grads(
          worker: RolloutWorker,
      ) -> Dict[str, Any]:
          """Call sample() and compute_gradients() remotely on workers."""
          samples = worker.sample()
          grads, infos = worker.compute_gradients(samples)
          return {
              "grads": grads,
              "infos": infos,
              "agent_steps": samples.agent_steps(),
              "env_steps": samples.env_steps(),
          }
  
      # Perform rollouts and gradient calculations asynchronously.
      with self._timers[GRAD_WAIT_TIMER]:
          # Results are a mapping from ActorHandle (RolloutWorker) to their
          # returned gradient calculation results.
          self.workers.foreach_worker_async(
              func=sample_and_compute_grads,
              healthy_only=True,
          )
          async_results = self.workers.fetch_ready_async_reqs()
  
      # Loop through all fetched worker-computed gradients (if any)
      # and apply them - one by one - to the local worker's model.
      # After each apply step (one step per worker that returned some gradients),
      # update that particular worker's weights.
      global_vars = None
      learner_info_builder = LearnerInfoBuilder(num_devices=1)
      to_sync_workers = set()
      for worker_id, result in async_results:
          # Apply gradients to local worker.
          with self._timers[APPLY_GRADS_TIMER]:
              local_worker.apply_gradients(result["grads"])
          self._timers[APPLY_GRADS_TIMER].push_units_processed(
              result["agent_steps"]
          )
  
          # Update all step counters.
          self._counters[NUM_AGENT_STEPS_SAMPLED] += result[
              "agent_steps"
          ]
          self._counters[NUM_ENV_STEPS_SAMPLED] += result["env_steps"]
          self._counters[NUM_AGENT_STEPS_TRAINED] += result[
              "agent_steps"
          ]
          self._counters[NUM_ENV_STEPS_TRAINED] += result["env_steps"]
  
          learner_info_builder.add_learn_on_batch_results_multi_agent(
              result["infos"]
          )
  
          # Create current global vars.
          global_vars = {
              "timestep": self._counters[NUM_AGENT_STEPS_SAMPLED],
          }
  
          # Add this worker to be synced.
          to_sync_workers.add(worker_id)
  
      # Synch updated weights back to the particular worker
      # (only those policies that are trainable).
      with self._timers[SYNCH_WORKER_WEIGHTS_TIMER]:
          self.workers.sync_weights(
              policies=local_worker.get_policies_to_train(),
              to_worker_indices=list(to_sync_workers),
              global_vars=global_vars,
          )
  
      return learner_info_builder.finalize()
  
imports: import logging
  from typing import Any, Dict, List, Optional, Type, Union
  from ray.rllib.algorithms.algorithm import Algorithm
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig, NotProvided
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.policy.policy import Policy
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.metrics import APPLY_GRADS_TIMER, GRAD_WAIT_TIMER, NUM_AGENT_STEPS_SAMPLED, NUM_AGENT_STEPS_TRAINED, NUM_ENV_STEPS_SAMPLED, NUM_ENV_STEPS_TRAINED, SYNCH_WORKER_WEIGHTS_TIMER
  from ray.rllib.utils.metrics.learner_info import LearnerInfoBuilder
  from ray.rllib.utils.typing import ResultDict
label_desc: Use image node_id 3 for calling the A3C obj's underlying member method code with example usage: obj.training_step() and returns: learner_info_builder, dict