Repository: Lightning-AI/pytorch-lightning
File: src/lightning/pytorch/loops/training_epoch_loop.py (Python)

Code before:
                loader = trainer.fit_loop._combined_loader
                assert loader is not None
                is_resumable_loader = all(isinstance(loader, _Stateful) for loader in loader.flattened)
                if self.global_step % expected_steps != 0 and not is_resumable_loader:
                    rank_zero_warn(
                        "You're resuming from a checkpoint that ended before the epoch ended and your dataloader is"
                        " not resumable. This can cause unreliable results if further training is done."
                        " Consider using an end-of-epoch checkpoint or make your dataloader resumable by implementing"
                        " the `state_dict` / `load_state_dict` interface.",
                        category=PossibleUserWarning,
                    )
        else:
            self.batch_progress.reset_on_run()
            self.scheduler_progress.reset_on_run()
            self.automatic_optimization.optim_progress.reset_on_run()
            # when the epoch starts, the total val batch progress should be reset as it's supposed to count the batches
            # seen per epoch, this is useful for tracking when validation is run multiple times per epoch
            self.val_loop.batch_progress.total.reset()

    def on_run_start(self, data_fetcher: _DataFetcher) -> None:
        # `iter()` was called once in `FitLoop.setup_data()` already
        # Call `iter()` again only when:
        #       1. Not restarting
        #       2. Not resuming from checkpoint (not is_resuming)
        #       3. Past first epoch (current_epoch > 0)
        if (self.trainer.current_epoch > 0 and not self.trainer.fit_loop.is_resuming) and not self.restarting:
            iter(data_fetcher)  # creates the iterator inside the fetcher

        # add the previous `fetched` value to properly track `is_last_batch` with no prefetching
        data_fetcher.fetched += self.batch_progress.current.ready
        data_fetcher._start_profiler = self._on_before_fetch
        data_fetcher._stop_profiler = self._on_after_fetch

    def _on_before_fetch(self) -> None:
        self.trainer.profiler.start(f"[{self.__class__.__name__}].train_dataloader_next")

    def _on_after_fetch(self) -> None:
        self.trainer.profiler.stop(f"[{self.__class__.__name__}].train_dataloader_next")

    def _broadcast_sigterm_tensor(self) -> None:
        try:
            sigterm_tensor = torch.tensor(
                [1 if getattr(self.trainer, "received_sigterm", False) else 0],
                device=self.trainer.strategy.root_device,
            )
            torch.distributed.broadcast(sigterm_tensor, src=0)
        except Exception:
            sigterm_tensor = torch.tensor([0], device=self.trainer.strategy.root_device)

        if sigterm_tensor.item() == 1:
            with contextlib.suppress(Exception):

Review comment (suggestion):
[nitpick] The conditional logic has unnecessary parentheses and could be simplified for better readability. Consider restructuring as: `if self.trainer.current_epoch > 0 and not self.trainer.fit_loop.is_resuming and not self.restarting:`
```suggestion
        if self.trainer.current_epoch > 0 and not self.trainer.fit_loop.is_resuming and not self.restarting:
```

Code after:
                loader = trainer.fit_loop._combined_loader
                assert loader is not None
                is_resumable_loader = all(isinstance(loader, _Stateful) for loader in loader.flattened)
                if self.global_step % expected_steps != 0 and not is_resumable_loader:
                    rank_zero_warn(
                        "You're resuming from a checkpoint that ended before the epoch ended and your dataloader is"
                        " not resumable. This can cause unreliable results if further training is done."
                        " Consider using an end-of-epoch checkpoint or make your dataloader resumable by implementing"
                        " the `state_dict` / `load_state_dict` interface.",
                        category=PossibleUserWarning,
                    )
        else:
            self.batch_progress.reset_on_run()
            self.scheduler_progress.reset_on_run()
            self.automatic_optimization.optim_progress.reset_on_run()
            # when the epoch starts, the total val batch progress should be reset as it's supposed to count the batches
            # seen per epoch, this is useful for tracking when validation is run multiple times per epoch
            self.val_loop.batch_progress.total.reset()

    def on_run_start(self, data_fetcher: _DataFetcher) -> None:
        # `iter()` was called once in `FitLoop.setup_data()` already
        # Call `iter()` again only when:
        #       1. Not restarting
        #       2. Not resuming from checkpoint (not is_resuming)
        #       3. Past first epoch (current_epoch > 0)
        if self.trainer.current_epoch > 0 and not self.trainer.fit_loop.is_resuming and not self.restarting:
            iter(data_fetcher)  # creates the iterator inside the fetcher

        # add the previous `fetched` value to properly track `is_last_batch` with no prefetching
        data_fetcher.fetched += self.batch_progress.current.ready
        data_fetcher._start_profiler = self._on_before_fetch
        data_fetcher._stop_profiler = self._on_after_fetch

    def _on_before_fetch(self) -> None:
        self.trainer.profiler.start(f"[{self.__class__.__name__}].train_dataloader_next")

    def _on_after_fetch(self) -> None:
        self.trainer.profiler.stop(f"[{self.__class__.__name__}].train_dataloader_next")

    def _broadcast_sigterm_tensor(self) -> None:
        try:
            sigterm_tensor = torch.tensor(
                [1 if getattr(self.trainer, "received_sigterm", False) else 0],
                device=self.trainer.strategy.root_device,
            )
            torch.distributed.broadcast(sigterm_tensor, src=0)
        except Exception:
            sigterm_tensor = torch.tensor([0], device=self.trainer.strategy.root_device)

        if sigterm_tensor.item() == 1:
            with contextlib.suppress(Exception):