class_docstr: Importance weighted actor/learner architecture (IMPALA) Algorithm
  
  == Overview of data flow in IMPALA ==
  1. Policy evaluation in parallel across `num_workers` actors produces
     batches of size `rollout_fragment_length * num_envs_per_worker`.
  2. If enabled, the replay buffer stores and produces batches of size
     `rollout_fragment_length * num_envs_per_worker`.
  3. If enabled, the minibatch ring buffer stores and replays batches of
     size `train_batch_size` up to `num_sgd_iter` times per batch.
  4. The learner thread executes data parallel SGD across `num_gpus` GPUs
     on batches of size `train_batch_size`.
class_docstr_tok: ["Importance","weighted","actor\/learner","architecture","(","IMPALA",")","Algorithm","==","Overview","of","data","flow","in","IMPALA","==","1",".","Policy","evaluation","in","parallel","across","`","num_workers","`","actors","produces","batches","of","size","`","rollout_fragment_length","*","num_envs_per_worker","`",".","2",".","If","enabled",",","the","replay","buffer","stores","and","produces","batches","of","size","`","rollout_fragment_length","*","num_envs_per_worker","`",".","3",".","If","enabled",",","the","minibatch","ring","buffer","stores","and","replays","batches","of","size","`","train_batch_size","`","up","to","`","num_sgd_iter","`","times","per","batch",".","4",".","The","learner","thread","executes","data","parallel","SGD","across","`","num_gpus","`","GPUs","on","batches","of","size","`","train_batch_size","`","."]
code: def setup(self, config: AlgorithmConfig):
      super().setup(config)
  
      # Queue of batches to be sent to the Learner.
      self.batches_to_place_on_learner = []
  
      # Create extra aggregation workers and assign each rollout worker to
      # one of them.
      self.batch_being_built = []
      if self.config.num_aggregation_workers > 0:
          # This spawns `num_aggregation_workers` actors that aggregate
          # experiences coming from RolloutWorkers in parallel. We force
          # colocation on the same node (localhost) to maximize data bandwidth
          # between them and the learner.
          localhost = platform.node()
          assert localhost != "", (
              "ERROR: Cannot determine local node name! "
              "`platform.node()` returned empty string."
          )
          all_co_located = create_colocated_actors(
              actor_specs=[
                  # (class, args, kwargs={}, count=1)
                  (
                      AggregatorWorker,
                      [
                          self.config,
                      ],
                      {},
                      self.config.num_aggregation_workers,
                  )
              ],
              node=localhost,
          )
          aggregator_workers = [
              actor
              for actor_groups in all_co_located
              for actor in actor_groups
          ]
          self._aggregator_actor_manager = FaultTolerantActorManager(
              aggregator_workers,
              max_remote_requests_in_flight_per_actor=(
                  self.config.max_requests_in_flight_per_aggregator_worker
              ),
          )
          self._timeout_s_aggregator_manager = (
              self.config.timeout_s_aggregator_manager
          )
      else:
          # Create our local mixin buffer if the num of aggregation workers is 0.
          self.local_mixin_buffer = MixInMultiAgentReplayBuffer(
              capacity=(
                  self.config.replay_buffer_num_slots
                  if self.config.replay_buffer_num_slots > 0
                  else 1
              ),
              replay_ratio=self.config.replay_ratio,
              replay_mode=ReplayMode.LOCKSTEP,
          )
          self._aggregator_actor_manager = None
  
      # This variable is used to keep track of the statistics from the most recent
      # update of the learner group
      self._results = {}
      self._timeout_s_sampler_manager = (
          self.config.timeout_s_sampler_manager
      )
  
      if not self.config._enable_new_api_stack:
          # Create and start the learner thread.
          self._learner_thread = make_learner_thread(
              self.workers.local_worker(), self.config
          )
          self._learner_thread.start()
  
imports: import copy
  from functools import partial
  import logging
  import platform
  import queue
  import random
  from typing import Callable, List, Optional, Set, Tuple, Type, Union
  import numpy
  import tree
  import ray
  from ray import ObjectRef
  from ray.rllib import SampleBatch
  from ray.rllib.algorithms.algorithm import Algorithm
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig, NotProvided
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec
  from ray.rllib.evaluation.worker_set import handle_remote_call_result_errors
  from ray.rllib.execution.buffers.mixin_replay_buffer import MixInMultiAgentReplayBuffer
  from ray.rllib.execution.learner_thread import LearnerThread
  from ray.rllib.execution.multi_gpu_learner_thread import MultiGPULearnerThread
  from ray.rllib.policy.policy import Policy
  from ray.rllib.policy.sample_batch import concat_samples
  from ray.rllib.utils.actor_manager import FaultAwareApply, FaultTolerantActorManager, RemoteCallResults
  from ray.rllib.utils.actors import create_colocated_actors
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.metrics import ALL_MODULES
  from ray.rllib.utils.deprecation import DEPRECATED_VALUE, deprecation_warning
  from ray.rllib.utils.metrics import NUM_AGENT_STEPS_SAMPLED, NUM_AGENT_STEPS_TRAINED, NUM_ENV_STEPS_SAMPLED, NUM_ENV_STEPS_TRAINED, NUM_SYNCH_WORKER_WEIGHTS, NUM_TRAINING_STEP_CALLS_SINCE_LAST_SYNCH_WORKER_WEIGHTS, SYNCH_WORKER_WEIGHTS_TIMER, SAMPLE_TIMER
  from ray.rllib.utils.metrics.learner_info import LearnerInfoBuilder
  from ray.rllib.utils.replay_buffers.multi_agent_replay_buffer import ReplayMode
  from ray.rllib.utils.replay_buffers.replay_buffer import _ALL_POLICIES
  from ray.rllib.utils.schedules.scheduler import Scheduler
  from ray.rllib.utils.typing import PartialAlgorithmConfigDict, PolicyID, ResultDict, SampleBatchType
  from ray.tune.execution.placement_groups import PlacementGroupFactory
label_desc: Use image node_id 3 for calling the Impala obj's underlying member method code with example usage: obj.setup(config) without return types