Add is_distributed to Strategy API and remove accl_conn is_distributed property

pytorch_lightning/lite/lite.py

@@ -432,7 +432,7 @@ def _move_model_to_device(self, model: nn.Module, optimizers: List[Optimizer]) -
 
     def _requires_distributed_sampler(self, dataloader: DataLoader) -> bool:
         return (
-            self._accelerator_connector.is_distributed
+            self._strategy.is_distributed
             and not isinstance(dataloader.sampler, DistributedSampler)
             and not has_iterable_dataset(dataloader)
         )

pytorch_lightning/strategies/ddp.py

@@ -96,10 +96,6 @@ def __init__(
         self._sync_dir: Optional[str] = None
         self._rank_0_will_call_children_scripts: bool = False
 
-    @property
-    def is_distributed(self) -> bool:
-        return True
-
     @property
     def root_device(self) -> torch.device:
         return self.parallel_devices[self.local_rank]

pytorch_lightning/strategies/dp.py

@@ -63,6 +63,10 @@ def node_rank(self) -> int:
     def world_size(self) -> int:
         return 1
 
+    @property
+    def is_distributed(self) -> bool:
+        return False
+
     def setup(self, trainer: "pl.Trainer") -> None:
         # model needs to be moved to the device before it is wrapped
         self.model_to_device()

pytorch_lightning/strategies/parallel.py

@@ -80,6 +80,10 @@ def parallel_devices(self):
     def parallel_devices(self, parallel_devices):
         self._parallel_devices = parallel_devices
 
+    @property
+    def is_distributed(self) -> bool:
+        return True
+
     @property
     def distributed_sampler_kwargs(self):
         distributed_sampler_kwargs = dict(num_replicas=len(self.parallel_devices), rank=self.global_rank)

pytorch_lightning/strategies/single_device.py

@@ -71,6 +71,10 @@ def setup(self, trainer: pl.Trainer) -> None:
         self.model_to_device()
         super().setup(trainer)
 
+    @property
+    def is_distributed(self) -> bool:
+        return False
+
     @property
     def is_global_zero(self) -> bool:
         return True

pytorch_lightning/strategies/strategy.py

@@ -252,6 +252,11 @@ def model_to_device(self) -> None:
     def is_global_zero(self) -> bool:
         """Whether the current process is the rank zero process not only on the local node, but for all nodes."""
 
+    @property
+    @abstractmethod
+    def is_distributed(self) -> bool:
+        """Whether the strategy is distributed strategy."""
+
     @abstractmethod
     def reduce(
         self,

pytorch_lightning/trainer/connectors/accelerator_connector.py

@@ -48,7 +48,6 @@
 )
 from pytorch_lightning.strategies import (
     DataParallelStrategy,
-    DDP2Strategy,
     DDPFullyShardedStrategy,
     DDPShardedStrategy,
     DDPSpawnShardedStrategy,
@@ -818,28 +817,6 @@ def gpus(self) -> Optional[Union[List[int], str, int]]:
     def parallel_device_ids(self) -> List[int]:
         return [i for i in range(len(self.parallel_devices))] if isinstance(self.accelerator, GPUAccelerator) else []
 
-    @property
-    def is_distributed(self) -> bool:
-        # Used for custom plugins.
-        # Custom plugins should implement is_distributed property.
-        if hasattr(self.strategy, "is_distributed") and not isinstance(self.accelerator, TPUAccelerator):
-            return self.strategy.is_distributed
-        distributed_strategy = (
-            DDP2Strategy,
-            DDPStrategy,
-            DDPSpawnShardedStrategy,
-            DDPShardedStrategy,
-            DDPFullyShardedStrategy,
-            DDPSpawnStrategy,
-            DeepSpeedStrategy,
-            TPUSpawnStrategy,
-            HorovodStrategy,
-        )
-        is_distributed = isinstance(self.strategy, distributed_strategy)
-        if isinstance(self.accelerator, TPUAccelerator):
-            is_distributed |= self.strategy.is_distributed
-        return is_distributed
-
     @property
     def has_ipu(self) -> bool:
         return isinstance(self.accelerator, IPUAccelerator) and isinstance(self.strategy, IPUStrategy)

pytorch_lightning/trainer/connectors/data_connector.py

@@ -266,7 +266,7 @@ def _worker_check(self, dataloader: DataLoader, name: str) -> None:
     def _requires_distributed_sampler(self, dataloader) -> bool:
         return (
             self.trainer._accelerator_connector.replace_sampler_ddp
-            and self.trainer._accelerator_connector.is_distributed
+            and self.trainer.strategy.is_distributed
             and not isinstance(dataloader.sampler, DistributedSampler)
             and not has_iterable_dataset(dataloader)
         )

tests/lite/test_lite.py

@@ -268,7 +268,7 @@ def test_setup_dataloaders_replace_custom_sampler(strategy):
 
     # explicitly asking to replace when a custom sampler is already configured raises an exception
     lite = EmptyLite(accelerator="cpu", strategy=strategy, devices=2)
-    if lite._accelerator_connector.is_distributed:
+    if lite._strategy.is_distributed:
         with pytest.raises(MisconfigurationException, match="You seem to have configured a sampler in your DataLoader"):
             lite.setup_dataloaders(dataloader, replace_sampler=True)
 
@@ -292,7 +292,7 @@ def test_setup_dataloaders_replace_custom_sampler(strategy):
 def test_setup_dataloaders_replace_standard_sampler(shuffle, strategy):
     """Test that Lite replaces the default samplers with DistributedSampler automatically."""
     lite = EmptyLite(accelerator="cpu", strategy=strategy, devices=2)
-    is_distributed = lite._accelerator_connector.is_distributed
+    is_distributed = lite._strategy.is_distributed
     lite_dataloader = lite.setup_dataloaders(DataLoader(range(3), shuffle=shuffle))
     assert not is_distributed or isinstance(lite_dataloader.sampler, DistributedSampler)
 

tests/trainer/test_trainer.py

@@ -1393,15 +1393,15 @@ def write_on_batch_end(self, trainer, pl_module, prediction, batch_indices, *arg
         self.write_on_batch_end_called = True
 
     def write_on_epoch_end(self, trainer, pl_module, predictions, batch_indices):
-        expected = 1 if trainer._accelerator_connector.is_distributed else 2
+        expected = 1 if trainer.strategy.is_distributed else 2
         assert len(predictions) == 2
         assert len(predictions[0]) == expected
         assert len(batch_indices) == 2
         assert len(batch_indices[0]) == expected
         self.write_on_epoch_end_called = True
 
     def on_predict_epoch_end(self, trainer, pl_module, outputs):
-        if trainer._accelerator_connector.is_distributed:
+        if trainer.strategy.is_distributed:
             for idx in range(2):
                 assert isinstance(trainer.predict_dataloaders[idx].batch_sampler.sampler, UnrepeatedDistributedSampler)
                 assert isinstance(trainer.predict_dataloaders[idx].batch_sampler, IndexBatchSamplerWrapper)