class_docstr: Tests for bluefog/torch/mpi_ops.py on one-sided communication.
code: def test_asscoicated_with_p(self):
      size = bf.size()
      rank = bf.rank()
      if size <= 3:
          fname = inspect.currentframe().f_code.co_name
          warnings.warn(
              "Skip {} because it only supports test over at least 3 nodes".format(
                  fname
              )
          )
          return
  
      dtypes = [torch.FloatTensor, torch.DoubleTensor]
      if TEST_ON_GPU and not bf.nccl_built():
          dtypes += [torch.cuda.FloatTensor, torch.cuda.DoubleTensor]
  
      bf.set_topology(topology_util.RingGraph(size))
      bf.turn_on_win_ops_with_associated_p()
      for dtype, send_rank in itertools.product(dtypes, range(size)):
          tensor = torch.FloatTensor([23]).fill_(1).mul_(rank)
          tensor = self.cast_and_place(tensor, dtype)
          window_name = "win_asscoicate_with_p_{}_{}".format(
              dtype, send_rank
          )
          bf.win_create(tensor, window_name)
          left_neighbor_rank = (send_rank - 1) % size
          right_neighbor_rank = (send_rank + 1) % size
          if rank == send_rank:
              bf.win_accumulate(
                  tensor,
                  name=window_name,
                  self_weight=0.5,
                  dst_weights={
                      left_neighbor_rank: 0.5,
                      right_neighbor_rank: 0.5,
                  },
              )
          bf.barrier()
          bf.win_update_then_collect(name=window_name)
          associated_p = bf.win_associated_p(name=window_name)
          if rank == send_rank:
              assert (
                  associated_p == 0.5
              ), "associated_p for sender {} is wrong. Get {}".format(
                  rank, associated_p
              )
          elif (rank == left_neighbor_rank) or (
              rank == right_neighbor_rank
          ):
              assert (
                  associated_p - 1.5
              ) < EPSILON, "associated_p for received neighbor {} is wrong. Get {}".format(
                  rank, associated_p
              )
          else:
              assert (
                  associated_p == 1.0
              ), "associated_p for untouched node {} is wrong. Get {}".format(
                  rank, associated_p
              )
      bf.turn_off_win_ops_with_associated_p()
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import inspect
  import itertools
  import time
  import warnings
  import unittest
  from bluefog.common import topology_util
  import bluefog.torch
  import torch
  import numpy
label_desc: Use image node_id 24 for calling the WinOpsTests obj's underlying member method code with example usage: obj.test_asscoicated_with_p() without return types