Fix ModelCheckpoint.file_exists OOM in DDP

Author: littlebullGit

src/lightning/pytorch/callbacks/model_checkpoint.py

@@ -999,8 +999,10 @@ def to_yaml(self, filepath: Optional[_PATH] = None) -> None:
     def file_exists(self, filepath: _PATH, trainer: "pl.Trainer") -> bool:
         """Checks if a file exists on rank 0 and broadcasts the result to all other ranks, preventing the internal
         state to diverge between ranks."""
-        exists = self._fs.exists(filepath)
-        return trainer.strategy.broadcast(exists)
+        # In distributed setups, only global rank 0 touches the filesystem
+        local_decision = self._fs.exists(filepath) if trainer.is_global_zero else False
+        # Reduce the decision across ranks using an "any"-style reduction to decide if the file exists anywhere
+        return trainer.strategy.reduce_boolean_decision(local_decision, all=False)
 
     def _should_remove_checkpoint(self, trainer: "pl.Trainer", previous: str, current: str) -> bool:
         """Checks if the previous checkpoint should be deleted.

tests/tests_pytorch/checkpointing/test_checkpoint_callback_frequency.py

@@ -121,3 +121,28 @@ def on_train_epoch_end(self):
     trainer.fit(model)
     if os.getenv("LOCAL_RANK") == "0":
         assert save_mock.call_count == expected
+
+
+@RunIf(min_cuda_gpus=2, standalone=True)
+def test_model_checkpoint_ddp_monitor_none(tmp_path):
+    """Ensure that ModelCheckpoint with monitor=None works correctly under DDP and exercises the file_exists path."""
+
+    model = BoringModel()
+    checkpoint = callbacks.ModelCheckpoint(dirpath=tmp_path, monitor=None, save_top_k=1)
+
+    trainer = Trainer(
+        default_root_dir=tmp_path,
+        callbacks=[checkpoint],
+        enable_progress_bar=False,
+        enable_model_summary=False,
+        max_epochs=1,
+        strategy="ddp",
+        accelerator="gpu",
+        devices=2,
+        limit_train_batches=2,
+        limit_val_batches=0,
+    )
+
+    trainer.fit(model)
+    if os.getenv("LOCAL_RANK") == "0":
+        assert checkpoint.best_model_path

tests/tests_pytorch/checkpointing/test_model_checkpoint.py

@@ -2180,3 +2180,36 @@ def on_validation_epoch_end(self):
     assert len(checkpoint_files) == expected_files, (
         f"Expected {expected_files} files, got {len(checkpoint_files)}: {checkpoint_names}"
     )
+
+
+def test_model_checkpoint_file_exists_distributed_branch(tmp_path):
+    """Ensure the distributed branch of ModelCheckpoint.file_exists uses reduce_boolean_decision."""
+
+    checkpoint = ModelCheckpoint(dirpath=tmp_path)
+    calls = []
+
+    class DummyStrategy:
+        def reduce_boolean_decision(self, decision, all=True):
+            calls.append((decision, all))
+            return decision
+
+    class DummyTrainer:
+        def __init__(self, is_global_zero: bool):
+            self.world_size = 2
+            self.is_global_zero = is_global_zero
+            self.strategy = DummyStrategy()
+
+    # global rank 0: filesystem is touched and decision=True is reduced with all=False
+    checkpoint._fs.exists = Mock(return_value=True)
+    trainer = DummyTrainer(is_global_zero=True)
+    assert checkpoint.file_exists("ignored", trainer)
+    checkpoint._fs.exists.assert_called_once_with("ignored")
+    assert calls == [(True, False)]
+
+    # non-global ranks: filesystem is not touched and local decision is False
+    calls.clear()
+    checkpoint._fs.exists = Mock(return_value=True)
+    trainer = DummyTrainer(is_global_zero=False)
+    assert not checkpoint.file_exists("ignored", trainer)
+    checkpoint._fs.exists.assert_not_called()
+    assert calls == [(False, False)]