docstr: Get samples from the replay buffer and place them on the learner queue.
  
  Args:
      num_worker_samples_collected: A mapping from ActorHandle (RolloutWorker) to
          number of samples returned by the remote worker. This is used to
          implement training intensity which is the concept of triggering a
          certain amount of training based on the number of samples that have
          been collected since the last time that training was triggered.
docstr_tok: ["Get","samples","from","the","replay","buffer","and","place","them","on","the","learner","queue",".","Args",":","num_worker_samples_collected",":","A","mapping","from","ActorHandle","(","RolloutWorker",")","to","number","of","samples","returned","by","the","remote","worker",".","This","is","used","to","implement","training","intensity","which","is","the","concept","of","triggering","a","certain","amount","of","training","based","on","the","number","of","samples","that","have","been","collected","since","the","last","time","that","training","was","triggered","."]
code: def sample_from_replay_buffer_place_on_learner_queue_non_blocking(
      self, num_worker_samples_collected: Dict[ActorHandle, int]
  ) -> None:
      """Get samples from the replay buffer and place them on the learner queue.
  
      Args:
          num_worker_samples_collected: A mapping from ActorHandle (RolloutWorker) to
              number of samples returned by the remote worker. This is used to
              implement training intensity which is the concept of triggering a
              certain amount of training based on the number of samples that have
              been collected since the last time that training was triggered.
  
      """
  
      def wait_on_replay_actors() -> List[Tuple[int, SampleBatchType]]:
          """Wait for the replay actors to finish sampling for timeout seconds.
  
          If the timeout is None, then block on the actors indefinitely.
          """
          results = self._replay_actor_manager.fetch_ready_async_reqs(
              timeout_seconds=self._replay_req_timeout_s
          )
          handle_remote_call_result_errors(
              results, self.config["ignore_worker_failures"]
          )
          return [
              (r.actor_id, r.get()) for r in results.ignore_errors()
          ]
  
      num_samples_collected = sum(num_worker_samples_collected.values())
      self.curr_num_samples_collected += num_samples_collected
      # Fetch replayed batched from last round.
      replay_sample_batches = wait_on_replay_actors()
      if (
          self.curr_num_samples_collected
          >= self.config.train_batch_size
          and
          # There are at least 1 healthy replay actor.
          self._replay_actor_manager.num_healthy_actors() > 0
      ):
          training_intensity = int(self.config.training_intensity or 1)
          num_requests_to_launch = (
              self.curr_num_samples_collected
              / self.config.train_batch_size
          ) * training_intensity
          num_requests_to_launch = max(1, round(num_requests_to_launch))
  
          self.curr_num_samples_collected = 0
          train_batch_size = self.config.train_batch_size
          healthy_worker_ids = (
              self._replay_actor_manager.healthy_actor_ids()
          )
  
          # Make num_requests_to_launch calls to the underlying replay actors.
          worker_ids_to_call = [
              random.choice(healthy_worker_ids)
              for _ in range(num_requests_to_launch)
          ]
          self._replay_actor_manager.foreach_actor_async(
              func=lambda actor: actor.sample(train_batch_size),
              remote_actor_ids=worker_ids_to_call,
          )
          # Fetch anything that is already ready.
          replay_sample_batches.extend(wait_on_replay_actors())
  
      # Add all the tuples of (ActorHandle, SampleBatchType) to the learner queue.
      for item in replay_sample_batches:
          # Setting block = True prevents the learner thread,
          # the main thread, and the gpu loader threads from
          # thrashing when there are more samples than the
          # learner can reasonable process.
          # see https://github.com/ray-project/ray/pull/26581#issuecomment-1187877674  # noqa
          self.learner_thread.inqueue.put(item, block=True)
      del replay_sample_batches
  
imports: import copy
  import platform
  import random
  from collections import defaultdict
  from typing import Any, Dict, List, Optional, Tuple, Union
  import ray
  from ray._private.dict import merge_dicts
  from ray.actor import ActorHandle
  from ray.rllib.algorithms import Algorithm
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig, NotProvided
  from ray.rllib.algorithms.dqn.dqn import DQN, DQNConfig
  from ray.rllib.algorithms.dqn.learner_thread import LearnerThread
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.evaluation.worker_set import handle_remote_call_result_errors
  from ray.rllib.utils.actor_manager import FaultTolerantActorManager
  from ray.rllib.utils.actors import create_colocated_actors
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.deprecation import DEPRECATED_VALUE
  from ray.rllib.utils.metrics import LAST_TARGET_UPDATE_TS, NUM_AGENT_STEPS_SAMPLED, NUM_AGENT_STEPS_TRAINED, NUM_ENV_STEPS_SAMPLED, NUM_ENV_STEPS_TRAINED, NUM_TARGET_UPDATES, SAMPLE_TIMER, SYNCH_WORKER_WEIGHTS_TIMER, TARGET_NET_UPDATE_TIMER
  from ray.rllib.utils.typing import PartialAlgorithmConfigDict, ResultDict, SampleBatchType
  from ray.tune.execution.placement_groups import PlacementGroupFactory
  from ray.tune.trainable import Trainable
label_desc: Use image node_id 7 for calling the ApexDQN obj's underlying member method code with example usage: obj.sample_from_replay_buffer_place_on_learner_queue_non_blocking(num_worker_samples_collected) and returns: unknown