Repository: Lightning-AI/pytorch-lightning
File: src/lightning/pytorch/callbacks/model_checkpoint.py (Python)

Code before:
            step = monitor_candidates["step"]
            rank_zero_info(
                f"Epoch {epoch:d}, global step {step:d}: {self.monitor!r} reached {current:0.5f}"
                f" (best {self.best_model_score:0.5f}), saving model to {filepath!r} as top {k}"
            )
        self._save_checkpoint(trainer, filepath)

        if del_filepath and self._should_remove_checkpoint(trainer, del_filepath, filepath):
            self._remove_checkpoint(trainer, del_filepath)

    def to_yaml(self, filepath: Optional[_PATH] = None) -> None:
        """Saves the `best_k_models` dict containing the checkpoint paths with the corresponding scores to a YAML
        file."""
        best_k = {k: v.item() for k, v in self.best_k_models.items()}
        if filepath is None:
            assert self.dirpath
            filepath = os.path.join(self.dirpath, "best_k_models.yaml")
        with self._fs.open(filepath, "w") as fp:
            yaml.dump(best_k, fp)

    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."""
        # Single-process or strategies without distributed world size: no need for coordination
        if trainer.world_size == 1:
            return self._fs.exists(filepath)

        # 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.

        A checkpoint won't be deleted if any of the cases apply:
        - The previous checkpoint is the same as the current checkpoint (means the old was already overwritten by new)
        - The previous checkpoint is not in the current checkpoint directory and the filesystem is local
        - The previous checkpoint is the checkpoint the Trainer resumed from and the filesystem is local

        """
        if previous == current:
            return False
        if not _is_local_file_protocol(previous):
            return True
        previous = Path(previous).absolute()
        resume_path = Path(trainer.ckpt_path).absolute() if trainer.ckpt_path is not None else None
        if resume_path is not None and previous == resume_path:
            return False
        assert self.dirpath is not None
        dirpath = Path(self.dirpath).absolute()

Review comment (suggestion):
in these cases the `reduce_boolean_decision` should be a no-op, so could we just do it always for consistency

Code after:
            step = monitor_candidates["step"]
            rank_zero_info(
                f"Epoch {epoch:d}, global step {step:d}: {self.monitor!r} reached {current:0.5f}"
                f" (best {self.best_model_score:0.5f}), saving model to {filepath!r} as top {k}"
            )
        self._save_checkpoint(trainer, filepath)

        if del_filepath and self._should_remove_checkpoint(trainer, del_filepath, filepath):
            self._remove_checkpoint(trainer, del_filepath)

    def to_yaml(self, filepath: Optional[_PATH] = None) -> None:
        """Saves the `best_k_models` dict containing the checkpoint paths with the corresponding scores to a YAML
        file."""
        best_k = {k: v.item() for k, v in self.best_k_models.items()}
        if filepath is None:
            assert self.dirpath
            filepath = os.path.join(self.dirpath, "best_k_models.yaml")
        with self._fs.open(filepath, "w") as fp:
            yaml.dump(best_k, fp)

    def file_exists(self, filepath: _PATH, trainer: "pl.Trainer") -> bool:
        """Checks if a file exists on rank 0 and synchronizes the result to all other ranks, preventing the internal
        state to diverge between ranks."""
        # 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.

        A checkpoint won't be deleted if any of the cases apply:
        - The previous checkpoint is the same as the current checkpoint (means the old was already overwritten by new)
        - The previous checkpoint is not in the current checkpoint directory and the filesystem is local
        - The previous checkpoint is the checkpoint the Trainer resumed from and the filesystem is local

        """
        if previous == current:
            return False
        if not _is_local_file_protocol(previous):
            return True
        previous = Path(previous).absolute()
        resume_path = Path(trainer.ckpt_path).absolute() if trainer.ckpt_path is not None else None
        if resume_path is not None and previous == resume_path:
            return False
        assert self.dirpath is not None
        dirpath = Path(self.dirpath).absolute()
        return dirpath in previous.parents

    def _remove_checkpoint(self, trainer: "pl.Trainer", filepath: str) -> None:
        """Calls the strategy to remove the checkpoint file."""