docstr: Update the remote workers that have samples ready.
  
  Args:
      _num_samples_ready: A mapping from ActorHandle (RolloutWorker) to
          the number of samples returned by the remote worker.
  
  Returns:
      The number of remote workers whose weights were updated.
docstr_tok: ["Update","the","remote","workers","that","have","samples","ready",".","Args",":","_num_samples_ready",":","A","mapping","from","ActorHandle","(","RolloutWorker",")","to","the","number","of","samples","returned","by","the","remote","worker",".","Returns",":","The","number","of","remote","workers","whose","weights","were","updated","."]
code: def update_workers(
      self, _num_samples_ready: Dict[ActorHandle, int]
  ) -> int:
      """Update the remote workers that have samples ready.
  
      Args:
          _num_samples_ready: A mapping from ActorHandle (RolloutWorker) to
              the number of samples returned by the remote worker.
  
      Returns:
          The number of remote workers whose weights were updated.
      """
      max_steps_weight_sync_delay = self.config.optimizer[
          "max_weight_sync_delay"
      ]
      # Update our local copy of the weights if the learner thread has updated
      # the learner worker's weights
      policy_ids_updated = self.learner_thread.policy_ids_updated.copy()
      self.learner_thread.policy_ids_updated.clear()
      if policy_ids_updated:
          weights = self.workers.local_worker().get_weights(
              policies=policy_ids_updated
          )
          self.curr_learner_weights = ray.put(weights)
  
      num_workers_updated = 0
  
      with self._timers[SYNCH_WORKER_WEIGHTS_TIMER]:
          curr_weights = self.curr_learner_weights
          timestep = self._counters[
              NUM_AGENT_STEPS_TRAINED
              if self.config.count_steps_by == "agent_steps"
              else NUM_ENV_STEPS_TRAINED
          ]
          for (
              remote_sampler_worker_id,
              num_samples_collected,
          ) in _num_samples_ready.items():
              self.steps_since_update[
                  remote_sampler_worker_id
              ] += num_samples_collected
              if (
                  self.steps_since_update[remote_sampler_worker_id]
                  >= max_steps_weight_sync_delay
              ):
                  self.workers.foreach_worker(
                      func=lambda w: w.set_weights(
                          ray.get(curr_weights), {"timestep": timestep}
                      ),
                      healthy_only=True,
                      local_worker=False,
                      timeout_seconds=0,  # Do not wait for results.
                  )
                  self.steps_since_update[remote_sampler_worker_id] = 0
                  num_workers_updated += 1
  
              self._counters["num_weight_syncs"] += 1
  
      return num_workers_updated
  
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 6 for calling the ApexDQN obj's underlying member method code with example usage: obj.update_workers(_num_samples_ready) and returns: num_workers_updated