class_docstr: Background thread that updates the local model from sample trajectories.
  
  The learner thread communicates with the main thread through Queues. This
  is needed since Ray operations can only be run on the main thread. In
  addition, moving heavyweight gradient ops session runs off the main thread
  improves overall throughput.
class_docstr_tok: ["Background","thread","that","updates","the","local","model","from","sample","trajectories",".","The","learner","thread","communicates","with","the","main","thread","through","Queues",".","This","is","needed","since","Ray","operations","can","only","be","run","on","the","main","thread",".","In","addition",",","moving","heavyweight","gradient","ops","session","runs","off","the","main","thread","improves","overall","throughput","."]
code: def step(self) -> Optional[_NextValueNotReady]:
      with self.queue_timer:
          try:
              batch, _ = self.minibatch_buffer.get()
          except queue.Empty:
              return _NextValueNotReady()
      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=1)
          if self.local_worker.config.policy_states_are_swappable:
              self.local_worker.lock()
          multi_agent_results = self.local_worker.learn_on_batch(batch)
          if self.local_worker.config.policy_states_are_swappable:
              self.local_worker.unlock()
          self.policy_ids_updated.extend(
              list(multi_agent_results.keys())
          )
          for pid, results in multi_agent_results.items():
              learner_info_builder.add_learn_on_batch_results(
                  results, pid
              )
          self.learner_info = learner_info_builder.finalize()
  
      self.num_steps += 1
      # Put tuple: env-steps, agent-steps, and learner info into the queue.
      self.outqueue.put(
          (batch.count, batch.agent_steps(), self.learner_info)
      )
      self.learner_queue_size.push(self.inqueue.qsize())
  
file_path: ray/python/ray/rllib/execution/learner_thread.py
imports: import copy
  import queue
  import threading
  from typing import Dict, Optional
  from ray.util.timer import _Timer
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.execution.minibatch_buffer import MinibatchBuffer
  from ray.rllib.utils.framework import try_import_tf
  from ray.rllib.utils.metrics.learner_info import LearnerInfoBuilder, LEARNER_INFO
  from ray.rllib.utils.metrics.window_stat import WindowStat
  from ray.util.iter import _NextValueNotReady
label_desc: Use image node_id 3 for calling the LearnerThread obj's underlying member method code with example usage: obj.step() and returns: _NextValueNotReady