docstr: Update the priorities of the sample batches with new priorities that are
  computed by the learner thread.
docstr_tok: ["Update","the","priorities","of","the","sample","batches","with","new","priorities","that","are","computed","by","the","learner","thread","."]
code: def update_replay_sample_priority(self) -> None:
      """Update the priorities of the sample batches with new priorities that are
      computed by the learner thread.
      """
      num_samples_trained_this_itr = 0
      for _ in range(self.learner_thread.outqueue.qsize()):
          if self.learner_thread.is_alive():
              (
                  replay_actor_id,
                  priority_dict,
                  env_steps,
                  agent_steps,
              ) = self.learner_thread.outqueue.get(timeout=0.001)
              if (
                  self.config.replay_buffer_config.get(
                      "prioritized_replay_alpha"
                  )
                  > 0
              ):
                  self._replay_actor_manager.foreach_actor(
                      func=lambda actor: actor.update_priorities(
                          priority_dict
                      ),
                      remote_actor_ids=[replay_actor_id],
                      timeout_seconds=0,  # Do not wait for results.
                  )
              num_samples_trained_this_itr += env_steps
              self.update_target_networks(env_steps)
              self._counters[NUM_ENV_STEPS_TRAINED] += env_steps
              self._counters[NUM_AGENT_STEPS_TRAINED] += agent_steps
              self.workers.local_worker().set_global_vars(
                  {"timestep": self._counters[NUM_ENV_STEPS_TRAINED]}
              )
          else:
              raise RuntimeError(
                  "The learner thread died while training"
              )
  
      self._timers["learner_dequeue"] = self.learner_thread.queue_timer
      self._timers["learner_grad"] = self.learner_thread.grad_timer
      self._timers[
          "learner_overall"
      ] = self.learner_thread.overall_timer
  
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 8 for calling the ApexDQN obj's underlying member method code with example usage: obj.update_replay_sample_priority() without return types