More fixes for hvd

Author: vfdev-5

ignite/distributed/comp_models/horovod.py

@@ -1,3 +1,4 @@
+import os
 import warnings
 from typing import Any, Callable, cast, List, Mapping, Optional, Tuple
 
@@ -23,6 +24,9 @@
 if has_hvd_support:
     HOROVOD = "horovod"
 
+    # Enables dynamic process sets: new_group methods and passing group into collective ops
+    os.environ["HOROVOD_DYNAMIC_PROCESS_SETS"] = "1"
+
     class _HorovodDistModel(ComputationModel):
         """Private class for `Horovod <https://horovod.readthedocs.io/en/stable/>`_ distributed computation model."""
 
@@ -155,6 +159,15 @@ def spawn(
                 **kwargs,
             )
 
+        def _setup_group(self, group: Any) -> hvd.ProcessSet:
+            if isinstance(group, list) and all(isinstance(item, int) for item in group):
+                group = self._do_new_group(group)
+            if not isinstance(group, hvd.ProcessSet):
+                raise ValueError(
+                    f"Argument group should be list of int or hvd.ProcessSet, got {type(group)}, group={group}"
+                )
+            return group
+
         _reduce_op_map = {
             "SUM": hvd.mpi_ops.Sum,
             "AVERAGE": hvd.mpi_ops.Average,
@@ -186,21 +199,25 @@ def _do_manual_all_reduce(self, tensor: torch.Tensor, op: Any) -> torch.Tensor:
             return reduced_res[0]
 
         def _do_all_gather(self, tensor: torch.Tensor, group: Optional[Any] = None) -> torch.Tensor:
+            if group is not None:
+                group = self._setup_group(group)
+            if self._rank_not_in_group(group):
+                return tensor
             if tensor.ndimension() == 0:
                 tensor = tensor.unsqueeze(0)
-            if group is None:
-                return hvd.allgather(tensor)
-            else:
+            if group is not None:
                 return hvd.allgather(tensor, process_set=group)
+            else:
+                return hvd.allgather(tensor)
 
         def _do_all_gather_object(self, tensor: Any, group: Optional[Any] = None) -> List[Any]:
             if group is not None:
                 raise NotImplementedError("all_gather with group for horovod is not implemented")
 
             return hvd.allgather_object(tensor)
 
-        def _do_new_group(self, ranks: List[int], **kwargs: Any) -> Any:
-            return hvd.ProcessSet(ranks)
+        def _do_new_group(self, ranks: List[int], **kwargs: Any) -> hvd.ProcessSet:
+            return hvd.add_process_set(ranks)
 
         def _do_broadcast(self, tensor: torch.Tensor, src: int) -> torch.Tensor:
             return hvd.broadcast(tensor, root_rank=src)
@@ -210,5 +227,7 @@ def barrier(self) -> None:
             # hvd.allreduce(torch.tensor(0, device=self.device()), name="barrier")
             hvd.allreduce(torch.tensor(0, device="cpu"), name="barrier")
 
-        def _rank_not_in_group(self, group: Any) -> bool:
+        def _rank_not_in_group(self, group: Optional[Any]) -> bool:
+            if group is None:
+                return False
             return not group.included()

ignite/distributed/comp_models/native.py

@@ -408,7 +408,7 @@ def spawn(
                 **spawn_kwargs,
             )
 
-        def _setup_group(self, group: Optional[Any]) -> dist.ProcessGroup:
+        def _setup_group(self, group: Any) -> dist.ProcessGroup:
             if isinstance(group, list) and all(isinstance(item, int) for item in group):
                 group = self._do_new_group(group)
             if not (isinstance(group, dist.ProcessGroup) or group == dist.GroupMember.NON_GROUP_MEMBER):
@@ -442,7 +442,7 @@ def _do_all_reduce(self, tensor: torch.Tensor, op: str = "SUM", group: Optional[
         def _do_all_gather(self, tensor: torch.Tensor, group: Optional[Any] = None) -> torch.Tensor:
             if group is not None:
                 group = self._setup_group(group)
-            if group == dist.GroupMember.NON_GROUP_MEMBER:
+            if self._rank_not_in_group(group):
                 return tensor
             if group is None:
                 group_size = self.get_world_size()
@@ -466,7 +466,7 @@ def _do_all_gather_object(self, tensor: Any, group: Optional[Any] = None) -> Lis
                 )
             if group is not None:
                 group = self._setup_group(group)
-            if group == dist.GroupMember.NON_GROUP_MEMBER:
+            if self._rank_not_in_group(group):
                 return tensor
             if group is None:
                 group_size = self.get_world_size()
@@ -491,7 +491,7 @@ def _do_broadcast(self, tensor: torch.Tensor, src: int) -> torch.Tensor:
         def barrier(self) -> None:
             dist.barrier()
 
-        def _rank_not_in_group(self, group: Any) -> bool:
+        def _rank_not_in_group(self, group: Optional[Any]) -> bool:
             return dist._rank_not_in_group(group)
 
     def _expand_hostlist(nodelist: str) -> List[str]:

tests/ignite/distributed/utils/__init__.py

@@ -238,7 +238,10 @@ def _test_distrib_all_gather_group(device):
         assert res == t
 
     t = torch.tensor([rank], device=device)
-    res = idist.all_gather(t, group=ranks)
+    if bnd == "horovod":
+        res = idist.all_gather(t, group=group)
+    else:
+        res = idist.all_gather(t, group=ranks)
     if rank in ranks:
         assert torch.equal(res, torch.tensor(sorted(ranks), device=device))
     else:
@@ -252,6 +255,9 @@ def _test_distrib_all_gather_group(device):
     if bnd in ("xla-tpu"):
         with pytest.raises(NotImplementedError, match=r"all_gather on object is not implemented for xla"):
             res = idist.all_gather(t, group=ranks)
+    elif bnd in ("horovod"):
+        with pytest.raises(NotImplementedError, match=r"all_gather with group for horovod is not implemented"):
+            res = idist.all_gather(t, group=group)
     else:
         res = idist.all_gather(t, group=ranks)
         if rank in ranks:
@@ -273,15 +279,13 @@ def _test_distrib_all_gather_group(device):
         else:
             assert res == t
 
+    t = torch.tensor([rank], device=device)
     if bnd in ("nccl", "gloo", "mpi", "horovod"):
-        with pytest.raises(ValueError, match=r"Argument group should be list of int or ProcessGroup"):
+        with pytest.raises(ValueError, match=r"Argument group should be list of int"):
             res = idist.all_gather(t, group="abc")
     elif bnd in ("xla-tpu"):
         with pytest.raises(ValueError, match=r"Argument group should be list of int"):
             res = idist.all_gather(t, group="abc")
-    elif bnd in ("horovod"):
-        with pytest.raises(NotImplementedError, match=r"all_gather with group for horovod is not implemented"):
-            res = idist.all_gather(t, group="abc")
 
 
 def _test_idist_all_gather_tensors_with_shapes(device):
@@ -309,9 +313,8 @@ def _test_idist_all_gather_tensors_with_shapes_group(device):
     torch.manual_seed(41)
 
     rank = idist.get_rank()
-    ranks = list(range(1, idist.get_world_size()))
+    ranks = sorted(range(idist.get_world_size() - 1, 0, -1))  # [0, 1, 2, 3] -> [1, 2, 3]
     ws = idist.get_world_size()
-    bnd = idist.backend()
     if rank in ranks:
         reference = torch.randn(ws * (ws + 1) // 2, ws * (ws + 3) // 2, ws * (ws + 5) // 2, device=device)
         rank_tensor = reference[
@@ -321,17 +324,18 @@ def _test_idist_all_gather_tensors_with_shapes_group(device):
         ]
     else:
         rank_tensor = torch.tensor([rank], device=device)
-        tensors = all_gather_tensors_with_shapes(rank_tensor, [[r + 1, r + 2, r + 3] for r in ranks], ranks)
-        if rank in ranks:
-            for r in ranks:
-                r_tensor = reference[
-                    r * (r + 1) // 2 : r * (r + 1) // 2 + r + 1,
-                    r * (r + 3) // 2 : r * (r + 3) // 2 + r + 2,
-                    r * (r + 5) // 2 : r * (r + 5) // 2 + r + 3,
-                ]
-                assert (r_tensor == tensors[r - 1]).all()
-        else:
-            assert [rank_tensor] == tensors
+
+    tensors = all_gather_tensors_with_shapes(rank_tensor, [[r + 1, r + 2, r + 3] for r in ranks], ranks)
+    if rank in ranks:
+        for r in ranks:
+            r_tensor = reference[
+                r * (r + 1) // 2 : r * (r + 1) // 2 + r + 1,
+                r * (r + 3) // 2 : r * (r + 3) // 2 + r + 2,
+                r * (r + 5) // 2 : r * (r + 5) // 2 + r + 3,
+            ]
+            assert torch.equal(r_tensor, tensors[r - 1])
+    else:
+        assert [rank_tensor] == tensors
 
 
 def _test_distrib_broadcast(device):