class_docstr: Coordinator of n (possibly remote) Learner workers.
  
  Each Learner worker has a copy of the RLModule, the loss function(s), and
  one or more optimizers.
code: def _update(
      self,
      *,
      batch: Optional[MultiAgentBatch] = None,
      episodes: Optional[List[EpisodeType]] = None,
      async_update: bool = False,
      reduce_fn: Optional[
          Callable[[List[Dict[str, Any]]], ResultDict]
      ] = (_reduce_mean_results),
      minibatch_size: Optional[int] = None,
      num_iters: int = 1,
  ) -> Union[
      Dict[str, Any], List[Dict[str, Any]], List[List[Dict[str, Any]]]
  ]:
      # Define function to be called on all Learner actors (or the local learner).
      def _learner_update(
          learner: Learner, batch_shard=None, episodes_shard=None
      ):
          if batch_shard is not None:
              return learner.update_from_batch(
                  batch=batch_shard,
                  reduce_fn=reduce_fn,
                  minibatch_size=minibatch_size,
                  num_iters=num_iters,
              )
          else:
              return learner.update_from_episodes(
                  episodes=episodes_shard,
                  reduce_fn=reduce_fn,
                  minibatch_size=minibatch_size,
                  num_iters=num_iters,
              )
  
      if self.is_local:
          if async_update:
              raise ValueError(
                  "Cannot call `update_from_batch(update_async=True)` when running in"
                  " local mode! Try setting `config.num_learner_workers > 0`."
              )
  
          results = [
              _learner_update(
                  learner=self._learner,
                  batch_shard=batch,
                  episodes_shard=episodes,
              )
          ]
      else:
          if episodes is None:
              partials = [
                  partial(_learner_update, batch_shard=batch_shard)
                  for batch_shard in ShardBatchIterator(
                      batch, len(self._workers)
                  )
              ]
          else:
              partials = [
                  partial(
                      _learner_update, episodes_shard=episodes_shard
                  )
                  for episodes_shard in ShardEpisodesIterator(
                      episodes, len(self._workers)
                  )
              ]
  
          if async_update:
              # Retrieve all ready results (kicked off by prior calls to this method).
              results = None
              if self._update_request_tags:
                  results = self._worker_manager.fetch_ready_async_reqs(
                      tags=list(self._update_request_tags)
                  )
  
              update_tag = str(uuid.uuid4())
  
              num_sent_requests = (
                  self._worker_manager.foreach_actor_async(
                      partials, tag=update_tag
                  )
              )
  
              if num_sent_requests:
                  self._update_request_tags[
                      update_tag
                  ] = num_sent_requests
  
              # Some requests were dropped, record lost ts/data.
              if num_sent_requests != len(self._workers):
                  # assert num_sent_requests == 0, num_sent_requests
                  factor = 1 - (num_sent_requests / len(self._workers))
                  if episodes is None:
                      self._ts_dropped += factor * len(batch)
                  else:
                      self._ts_dropped += factor * sum(
                          len(e) for e in episodes
                      )
              # NOTE: There is a strong assumption here that the requests launched to
              # learner workers will return at the same time, since they are have a
              # barrier inside of themselves for gradient aggregation. Therefore
              # results should be a list of lists where each inner list should be the
              # length of the number of learner workers, if results from an
              # non-blocking update are ready.
              results = self._get_async_results(results)
  
          else:
              results = self._get_results(
                  self._worker_manager.foreach_actor(partials)
              )
  
      # TODO (sven): Move reduce_fn to the training_step
      if reduce_fn is None:
          return results
      elif not async_update:
          return reduce_fn(results)
      else:
          return [reduce_fn(r) for r in results]
  
imports: from collections import defaultdict, Counter
  from functools import partial
  import pathlib
  from typing import Any, Callable, Dict, List, Optional, Set, Type, Union
  import uuid
  import ray
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.core.learner.learner import Learner
  from ray.rllib.core.learner.reduce_result_dict_fn import _reduce_mean_results
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec, RLMODULE_STATE_DIR_NAME
  from ray.rllib.policy.sample_batch import MultiAgentBatch
  from ray.rllib.utils.actor_manager import FaultTolerantActorManager
  from ray.rllib.utils.deprecation import Deprecated, deprecation_warning
  from ray.rllib.utils.minibatch_utils import ShardBatchIterator, ShardEpisodesIterator
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.typing import EpisodeType, ModuleID, ResultDict, RLModuleSpec, ShouldModuleBeUpdatedFn
  from ray.train._internal.backend_executor import BackendExecutor
  from ray.tune.utils.file_transfer import sync_dir_between_nodes
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 7 for calling the LearnerGroup obj's underlying member method code with example usage: obj._update() and returns: results, learner, learner, reduce_fn, unknown