docstr: Test that async style updates converge to the same result as sync.
docstr_tok: ["Test","that","async","style","updates","converge","to","the","same","result","as","sync","."]
code: def test_async_update(self):
      """Test that async style updates converge to the same result as sync."""
      fws = ["torch", "tf2"]
      # async_update only needs to be tested for the most complex case.
      # so we'll only test it for multi-gpu-ddp.
      scaling_modes = ["multi-gpu-ddp", "remote-gpu"]
      test_iterator = itertools.product(fws, scaling_modes)
  
      for fw, scaling_mode in test_iterator:
          print(
              f"Testing framework: {fw}, scaling mode: {scaling_mode}."
          )
          env = gym.make("CartPole-v1")
          config_overrides = REMOTE_CONFIGS[scaling_mode]
          config = BaseTestingAlgorithmConfig().update_from_dict(
              config_overrides
          )
          learner_group = config.build_learner_group(env=env)
          reader = get_cartpole_dataset_reader(batch_size=512)
          min_loss = float("inf")
          batch = reader.next()
          timer_sync = _Timer()
          timer_async = _Timer()
          with timer_sync:
              learner_group.update_from_batch(
                  batch=batch.as_multi_agent(),
                  async_update=False,
                  reduce_fn=None,
              )
          with timer_async:
              result_async = learner_group.update_from_batch(
                  batch=batch.as_multi_agent(),
                  async_update=True,
                  reduce_fn=None,
              )
          # ideally the the first async update will return nothing, and an easy
          # way to check that is if the time for an async update call is faster
          # than the time for a sync update call.
          self.assertLess(timer_async.mean, timer_sync.mean)
          self.assertIsInstance(result_async, list)
          self.assertEqual(len(result_async), 0)
          iter_i = 0
          while True:
              batch = reader.next()
              async_results = learner_group.update_from_batch(
                  batch.as_multi_agent(),
                  async_update=True,
                  reduce_fn=None,
              )
              if not async_results:
                  continue
              losses = [
                  np.mean(
                      [
                          res[ALL_MODULES][Learner.TOTAL_LOSS_KEY]
                          for res in results
                      ]
                  )
                  for results in async_results
              ]
              min_loss_this_iter = min(losses)
              min_loss = min(min_loss_this_iter, min_loss)
              print(
                  f"[iter = {iter_i}] Loss: {min_loss_this_iter:.3f}, Min Loss: "
                  f"{min_loss:.3f}"
              )
              # The loss is initially around 0.69 (ln2). When it gets to around
              # 0.57 the return of the policy gets to around 100.
              if min_loss < 0.57:
                  break
              for results in async_results:
                  for res1, res2 in zip(results, results[1:]):
                      self.assertEqual(
                          res1[DEFAULT_POLICY_ID]["mean_weight"],
                          res2[DEFAULT_POLICY_ID]["mean_weight"],
                      )
              iter_i += 1
          learner_group.shutdown()
          self.assertLess(min_loss, 0.57)
  
imports: import gymnasium
  import itertools
  import numpy
  from typing import Any, Dict, List
  import tempfile
  import unittest
  import ray
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.algorithms.ppo.tests.test_ppo_learner import FAKE_BATCH
  from ray.rllib.core.learner.learner import Learner
  from ray.rllib.core.rl_module.marl_module import MultiAgentRLModule
  from ray.rllib.core.rl_module.rl_module import SingleAgentRLModuleSpec
  from ray.rllib.core.testing.tf.bc_learner import BCTfLearner
  from ray.rllib.core.testing.tf.bc_module import DiscreteBCTFModule
  from ray.rllib.core.testing.utils import add_module_to_learner_or_learner_group
  from ray.rllib.core.testing.testing_learner import BaseTestingAlgorithmConfig
  from ray.rllib.examples.env.multi_agent import MultiAgentCartPole
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID, SampleBatch, MultiAgentBatch
  from ray.rllib.utils.test_utils import check, get_cartpole_dataset_reader
  from ray.rllib.utils.metrics import ALL_MODULES
  from ray.util.timer import _Timer
label_desc: Use image node_id 3 for calling the TestLearnerGroupAsyncUpdate obj's underlying member method code with example usage: obj.test_async_update() without return types