class_docstr: Tests for bluefog/torch/mpi_ops.py on one-sided communication.
code: def test_win_mutex_given_ranks(self):
      size = bf.size()
      rank = bf.rank()
      if size < 4:
          fname = inspect.currentframe().f_code.co_name
          warnings.warn(
              "Skip {} because it only supports test above 4 nodes".format(
                  fname
              )
          )
          return
  
      dtypes = [torch.FloatTensor, torch.DoubleTensor]
      if TEST_ON_GPU:
          dtypes += [torch.cuda.FloatTensor, torch.cuda.DoubleTensor]
  
      for dtype in dtypes:
          tensor = torch.FloatTensor([DIM_SIZE]).fill_(1).mul_(rank)
          tensor = self.cast_and_place(tensor, dtype)
          window_name = "win_mutex_given_ranks_{}".format(dtype)
          bf.win_create(tensor, window_name)
          if rank == 0:
              with bf.win_mutex(window_name, for_self=True, ranks=[1]):
                  bf.barrier()
                  time.sleep(1.01)
          elif rank == 1:
              bf.barrier()
              t_start = time.time()
              with bf.win_mutex(window_name, ranks=[0]):
                  time.sleep(0.001)
              t_end = time.time()
              assert (t_end - t_start) > 1
          elif rank == 2:
              bf.barrier()
              t_start = time.time()
              with bf.win_mutex(window_name, ranks=[0]):
                  time.sleep(0.001)
              t_end = time.time()
              assert (t_end - t_start) < 0.1
          else:
              bf.barrier()
  
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 23 for calling the WinOpsTests obj's underlying member method code with example usage: obj.test_win_mutex_given_ranks() without return types