code: def run_test_observation_aggregator(
      comm, xp, use_chainer_variable, communicate_interval, use_gpu
  ):
      model = DummyChain()
  
      if use_gpu:
          # Use CuPy's Device class to force call cudaSetDevice()
          chainer.cuda.get_device_from_id(comm.intra_rank).use()
  
      device = get_device(
          comm.intra_rank if use_gpu else None, xp == chainerx
      )
  
      if xp == chainerx:
          train = xp.array(np.random.rand(10, 1).astype(np.float32))
      else:
          train = xp.random.rand(10, 1).astype(np.float32)
  
      model.to_device(device)
  
      train_iter = chainer.iterators.SerialIterator(
          train, batch_size=1, repeat=True, shuffle=True
      )
  
      optimizer = chainermn.create_multi_node_optimizer(
          chainer.optimizers.Adam(), comm
      )
      optimizer.setup(model)
  
      updater = chainer.training.StandardUpdater(
          train_iter, optimizer, device=device
      )
  
      trainer = chainer.training.Trainer(updater, (1, "epoch"))
  
      @extension.make_extension(
          trigger=(1, "iteration"), priority=extension.PRIORITY_WRITER
      )
      def rank_reporter(trainer_):
          tmp = xp.asarray(comm.rank, dtype=np.float32)
          if use_chainer_variable:
              tmp = chainer.Variable(tmp)
          trainer_.observation["rank"] = tmp
  
      @extension.make_extension(
          trigger=(communicate_interval, "iteration"),
          priority=extension.PRIORITY_READER,
      )
      def aggregated_rank_checker(trainer_):
          actual = trainer_.observation["rank-aggregated"]
          if use_chainer_variable:
              actual = actual.data
          expected = (comm.size - 1) / 2
          chainer.testing.assert_allclose(actual, expected)
  
      trainer.extend(rank_reporter)
      trainer.extend(
          ObservationAggregator(
              comm,
              "rank",
              "rank-aggregated",
              comm_trigger=(communicate_interval, "iteration"),
          )
      )
      trainer.extend(aggregated_rank_checker)
  
      trainer.run()
  
file_path: chainer/tests/chainermn_tests/extensions_tests/test_observation_aggregator.py
imports: from __future__ import division
  import pytest
  import numpy
  import chainer
  import chainer.testing
  from chainer.training import extension
  from chainer.backend import cuda
  import chainermn
  from chainermn.testing import get_device
  from chainermn.extensions import ObservationAggregator
  import chainerx
label_desc: Use image node_id 4 for calling a global function with example usage: run_test_observation_aggregator(comm, xp, use_chainer_variable, communicate_interval, use_gpu) without return types