code: def test_top_k_ddp(
      save_mock, tmpdir, k, epochs, val_check_interval, expected
  ):
      class TestModel(BoringModel):
          def training_step(self, batch, batch_idx):
              local_rank = int(os.getenv("LOCAL_RANK"))
              self.log(
                  "my_loss", batch_idx * (1 + local_rank), on_epoch=True
              )
              return super().training_step(batch, batch_idx)
  
          def on_train_epoch_end(self):
              local_rank = int(os.getenv("LOCAL_RANK"))
              if self.trainer.is_global_zero:
                  self.log(
                      "my_loss_2",
                      (1 + local_rank),
                      on_epoch=True,
                      rank_zero_only=True,
                  )
              data = str(self.global_rank)
              obj = [[data], (data,), set(data)]
              out = self.trainer.strategy.broadcast(obj)
              assert obj == [
                  [str(self.global_rank)],
                  (str(self.global_rank),),
                  set(str(self.global_rank)),
              ]
              assert out == [["0"], ("0",), set("0")]
  
      model = TestModel()
      trainer = Trainer(
          callbacks=[
              callbacks.ModelCheckpoint(
                  dirpath=tmpdir,
                  monitor="my_loss_step",
                  save_top_k=k,
                  mode="max",
              )
          ],
          default_root_dir=tmpdir,
          enable_progress_bar=False,
          max_epochs=epochs,
          enable_model_summary=False,
          val_check_interval=val_check_interval,
          strategy="ddp",
          accelerator="gpu",
          devices=2,
          limit_train_batches=64,
          limit_val_batches=32,
      )
      trainer.fit(model)
      if os.getenv("LOCAL_RANK") == "0":
          assert save_mock.call_count == expected
  
file_path: pytorch-lightning/tests/tests_pytorch/checkpointing/test_checkpoint_callback_frequency.py
imports: import os
  from unittest import mock
  import pytest
  import torch
  from lightning.pytorch import Trainer, callbacks
  from lightning.pytorch.demos.boring_classes import BoringModel
  from tests_pytorch.helpers.runif import RunIf
label_desc: Use image node_id 4 for calling a global function with example usage: test_top_k_ddp(save_mock, tmpdir, k, epochs, val_check_interval, expected) and returns: super