code: def _step(self) -> None:
      s = self.multi_gpu_learner_thread
      policy_map = s.policy_map
  
      # Get a new batch from the data (inqueue).
      with self.queue_timer:
          batch = s.inqueue.get()
  
      # Get next idle stack for loading.
      buffer_idx = s.idle_tower_stacks.get()
  
      # Load the batch into the idle stack.
      with self.load_timer:
          for pid in policy_map.keys():
              if (
                  s.local_worker.is_policy_to_train is not None
                  and not s.local_worker.is_policy_to_train(pid, batch)
              ):
                  continue
              policy = policy_map[pid]
              if isinstance(batch, SampleBatch):
                  policy.load_batch_into_buffer(
                      batch=batch,
                      buffer_index=buffer_idx,
                  )
              elif pid in batch.policy_batches:
                  policy.load_batch_into_buffer(
                      batch=batch.policy_batches[pid],
                      buffer_index=buffer_idx,
                  )
  
      # Tag just-loaded stack as "ready".
      s.ready_tower_stacks.put(buffer_idx)
  
imports: import logging
  import queue
  import threading
  from ray.util.timer import _Timer
  from ray.rllib.execution.learner_thread import LearnerThread
  from ray.rllib.execution.minibatch_buffer import MinibatchBuffer
  from ray.rllib.policy.sample_batch import SampleBatch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.deprecation import deprecation_warning
  from ray.rllib.utils.framework import try_import_tf
  from ray.rllib.utils.metrics.learner_info import LearnerInfoBuilder
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
label_desc: Use image node_id 3 for calling the _MultiGPULoaderThread obj's underlying member method code with example usage: obj._step() without return types