class_docstr: Tests for bluefog/torch/mpi_ops.py on one-sided communication.
code: def test_win_mutex_full(self):
      size = bf.size()
      rank = bf.rank()
      if size <= 2:
          fname = inspect.currentframe().f_code.co_name
          warnings.warn(
              "Skip {} because it only supports test over at least 3 nodes".format(
                  fname
              )
          )
          return
      bf.set_topology(topology_util.FullyConnectedGraph(size))
  
      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_full_{}".format(dtype)
          bf.win_create(tensor, window_name)
  
          if rank == 0:
              with bf.win_mutex(window_name, for_self=True):
                  bf.barrier()
                  time.sleep(1.01)
          else:
              bf.barrier()
              t_start = time.time()
              with bf.win_mutex(window_name):
                  time.sleep(0.001)
              t_end = time.time()
              assert (
                  t_end - t_start
              ) > 1, "The mutex acquire time should be longer than 1 second"
              assert (
                  t_end - t_start
              ) < 2, "The mutex acquire time should be shorter than 2 second"
  
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 22 for calling the WinOpsTests obj's underlying member method code with example usage: obj.test_win_mutex_full() without return types