class_docstr: Tests for bluefog/torch/mpi_ops.py
class_docstr_tok: ["Tests","for","bluefog\/torch\/mpi_ops.py"]
docstr: Test that the neighbor all reduce (avg) 1D, 2D, 3D tensors correctly.
code: def test_neighbor_allreduce_dynamic_topo_check(self):
      """Test that the neighbor all reduce (avg) 1D, 2D, 3D tensors correctly."""
      size = bf.size()
      rank = bf.rank()
      if size <= 1:
          fname = inspect.currentframe().f_code.co_name
          warnings.warn("Skip {} due to size 1".format(fname))
          return
      dtypes = [torch.HalfTensor, torch.FloatTensor, torch.DoubleTensor]
      if TEST_ON_GPU:
          dtypes += [torch.cuda.FloatTensor, torch.cuda.DoubleTensor]
  
      # By default, we use exponential two ring topology.
      self_weight = 0.0
      neighbor_weights = {(rank - 1) % size: 1.0}
      send_ranks = [(rank + 2) % size]
  
      dims = [1, 2, 3]
      for dtype, dim in itertools.product(dtypes, dims):
          tensor = torch.FloatTensor(*([23] * dim)).fill_(1).mul_(rank)
          tensor = self.cast_and_place(tensor, dtype)
          name = "neighbor_allreduce_{}_{}".format(dim, dtype)
          with pytest.raises(ValueError):
              bf.neighbor_allreduce(
                  tensor,
                  name=name,
                  self_weight=self_weight,
                  src_weights=neighbor_weights,
                  dst_weights=send_ranks,
              )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import inspect
  import itertools
  import unittest
  import warnings
  import numpy
  import pytest
  import torch
  import networkx
  import bluefog.torch
  from bluefog.common import topology_util
  from common import is_openmpi_built
label_desc: Use image node_id 16 for calling the OpsTests obj's underlying member method code with example usage: obj.test_neighbor_allreduce_dynamic_topo_check() without return types