torch: sync values_handle firstly in sparse_allreduce_async (#2965)

Root cause hasn't been found out. But if values_handle is not synced
firstly, sparse_allreduce_async will be hanging in unit test.

Signed-off-by: Chongxiao Cao <chongxiaoc@uber.com>

horovod/torch/mpi_ops.py

@@ -518,8 +518,11 @@ def sparse_allreduce_async(tensor, name, op):
     values_handle = allgather_async(t._values(), name=f'{name}.values')
 
     def handle():
-        indices = synchronize(indices_handle)
+        # We need to sync values handle firstly for torch nightly >= 10.0
+        # Issue: https://github.com/horovod/horovod/issues/2961
         values = synchronize(values_handle)
+        indices = synchronize(indices_handle)
+
         values = (values / size()) if op == Average else values
 
         if indices.dim() == 0 or values.dim() == 0:

test/parallel/test_torch.py

@@ -2421,8 +2421,6 @@ def forward(self, x):
             loss.backward()
             opt.step()
 
-    @pytest.mark.skipif(LooseVersion(torch.__version__) > LooseVersion('1.10.0'),
-                        reason='https://github.com/horovod/horovod/issues/2961')
     def test_async_sparse_allreduce(self):
         """Test that allgather over indices and values is equivalent to allreduce."""
         hvd.init()