class_docstr: Learner that can use multiple GPUs and parallel loading.
  
  This class is used for async sampling algorithms.
  
  Example workflow: 2 GPUs and 3 multi-GPU tower stacks.
  -> On each GPU, there are 3 slots for batches, indexed 0, 1, and 2.
  
  Workers collect data from env and push it into inqueue:
  Workers -> (data) -> self.inqueue
  
  We also have two queues, indicating, which stacks are loaded and which
  are not.
  - idle_tower_stacks = [0, 1, 2]  <- all 3 stacks are free at first.
  - ready_tower_stacks = []  <- None of the 3 stacks is loaded with data.
  
  `ready_tower_stacks` is managed by `ready_tower_stacks_buffer` for
  possible minibatch-SGD iterations per loaded batch (this avoids a reload
  from CPU to GPU for each SGD iter).
  
  n _MultiGPULoaderThreads: self.inqueue -get()->
  policy.load_batch_into_buffer() -> ready_stacks = [0 ...]
  
  This thread: self.ready_tower_stacks_buffer -get()->
  policy.learn_on_loaded_batch() -> if SGD-iters done,
  put stack index back in idle_tower_stacks queue.
class_docstr_tok: ["Learner","that","can","use","multiple","GPUs","and","parallel","loading",".","This","class","is","used","for","async","sampling","algorithms",".","Example","workflow",":","2","GPUs","and","3","multi-GPU","tower","stacks",".","-",">","On","each","GPU",",","there","are","3","slots","for","batches",",","indexed","0",",","1",",","and","2",".","Workers","collect","data","from","env","and","push","it","into","inqueue",":","Workers","-",">","(","data",")","-",">","self.inqueue","We","also","have","two","queues",",","indicating",",","which","stacks","are","loaded","and","which","are","not",".","-","idle_tower_stacks","=","[","0",",","1",",","2","]","<","-","all","3","stacks","are","free","at","first",".","-","ready_tower_stacks","=","[","]","<","-","None","of","the","3","stacks","is","loaded","with","data",".","`","ready_tower_stacks","`","is","managed","by","`","ready_tower_stacks_buffer","`","for","possible","minibatch-SGD","iterations","per","loaded","batch","(","this","avoids","a","reload","from","CPU","to","GPU","for","each","SGD","iter",")",".","n","_MultiGPULoaderThreads",":","self.inqueue","-get","(",")","-",">","policy.load_batch_into_buffer","(",")","-",">","ready_stacks","=","[","0","...","]","This","thread",":","self.ready_tower_stacks_buffer","-get","(",")","-",">","policy.learn_on_loaded_batch","(",")","-",">","if","SGD-iters","done",",","put","stack","index","back","in","idle_tower_stacks","queue","."]
code: def step(self) -> None:
      if not self.loader_thread.is_alive():
          raise RuntimeError(
              "The `_MultiGPULoaderThread` has died! Will therefore also terminate "
              "the `MultiGPULearnerThread`."
          )
  
      with self.load_wait_timer:
          buffer_idx, released = self.ready_tower_stacks_buffer.get()
  
      get_num_samples_loaded_into_buffer = 0
      with self.grad_timer:
          # Use LearnerInfoBuilder as a unified way to build the final
          # results dict from `learn_on_loaded_batch` call(s).
          # This makes sure results dicts always have the same structure
          # no matter the setup (multi-GPU, multi-agent, minibatch SGD,
          # tf vs torch).
          learner_info_builder = LearnerInfoBuilder(
              num_devices=len(self.devices)
          )
  
          for pid in self.policy_map.keys():
              # Not a policy-to-train.
              if (
                  self.local_worker.is_policy_to_train is not None
                  and not self.local_worker.is_policy_to_train(pid)
              ):
                  continue
              policy = self.policy_map[pid]
              default_policy_results = policy.learn_on_loaded_batch(
                  offset=0, buffer_index=buffer_idx
              )
              learner_info_builder.add_learn_on_batch_results(
                  default_policy_results, policy_id=pid
              )
              self.policy_ids_updated.append(pid)
              get_num_samples_loaded_into_buffer += (
                  policy.get_num_samples_loaded_into_buffer(buffer_idx)
              )
  
          self.learner_info = learner_info_builder.finalize()
  
      if released:
          self.idle_tower_stacks.put(buffer_idx)
  
      # Put tuple: env-steps, agent-steps, and learner info into the queue.
      self.outqueue.put(
          (
              get_num_samples_loaded_into_buffer,
              get_num_samples_loaded_into_buffer,
              self.learner_info,
          )
      )
      self.learner_queue_size.push(self.inqueue.qsize())
  
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 2 for calling the MultiGPULearnerThread obj's underlying member method code with example usage: obj.step() without return types