class_docstr: A callback which periodically saves the application state during training using `Distributed Checkpoint <https://pytorch.org/docs/stable/distributed.checkpoint.html/>`_.
  
  This callback supplements the application state provided by :class:`torchtnt.unit.AppStateMixin`
  with the train progress, and train dataloader (if applicable).
  
  If used with :func:`torchtnt.framework.fit`, this class will also save the evaluation progress state.
  
  Checkpoints will be saved under ``dirpath/epoch_{epoch}_step_{step}`` where step is the *total* number of training steps completed across all epochs.
  
  Args:
      dirpath: Parent directory to save snapshots to.
      save_every_n_train_steps: Frequency of steps with which to save snapshots during the train epoch. If None, no intra-epoch snapshots are generated.
      save_every_n_epochs: Frequency of epochs with which to save snapshots during training. If None, no end-of-epoch snapshots are generated.
      keep_last_n_checkpoints: Number of most recent checkpoints to keep. If None, all checkpoints are kept. If an excess of existing checkpoints are present, the oldest ones will be deleted to clean the difference.
      process_group: The process group on which the ranks will communicate on. default: ``None`` (the entire world)
  
  Note:
      If torch.distributed is available and default process group is initialized, dcp's `no_dist <https://pytorch.org/docs/stable/distributed.checkpoint.html#torch.distributed.checkpoint.load_state_dict/>_`
      argument is automatically set to False. Otherwise it's set to True.
  
  Note:
      If checkpointing FSDP model, you can set state_dict type calling `set_state_dict_type <https://pytorch.org/docs/stable/fsdp.html#torch.distributed.fsdp.FullyShardedDataParallel.set_state_dict_type>`_ prior to starting training.
class_docstr_tok: ["A","callback","which","periodically","saves","the","application","state","during","training","using","`","Distributed","Checkpoint","<","https",":","\/\/pytorch.org\/docs\/stable\/distributed.checkpoint.html\/",">","`","_",".","This","callback","supplements","the","application","state","provided","by",":","class",":","`","torchtnt.unit.AppStateMixin","`","with","the","train","progress",",","and","train","dataloader","(","if","applicable",")",".","If","used","with",":","func",":","`","torchtnt.framework.fit","`",",","this","class","will","also","save","the","evaluation","progress","state",".","Checkpoints","will","be","saved","under","``","dirpath\/epoch_","{","epoch","}","_step_","{","step","}","``","where","step","is","the","*","total","*","number","of","training","steps","completed","across","all","epochs",".","Args",":","dirpath",":","Parent","directory","to","save","snapshots","to",".","save_every_n_train_steps",":","Frequency","of","steps","with","which","to","save","snapshots","during","the","train","epoch",".","If","None",",","no","intra-epoch","snapshots","are","generated",".","save_every_n_epochs",":","Frequency","of","epochs","with","which","to","save","snapshots","during","training",".","If","None",",","no","end-of-epoch","snapshots","are","generated",".","keep_last_n_checkpoints",":","Number","of","most","recent","checkpoints","to","keep",".","If","None",",","all","checkpoints","are","kept",".","If","an","excess","of","existing","checkpoints","are","present",",","the","oldest","ones","will","be","deleted","to","clean","the","difference",".","process_group",":","The","process","group","on","which","the","ranks","will","communicate","on",".","default",":","``","None","``","(","the","entire","world",")","Note",":","If","torch.distributed","is","available","and","default","process","group","is","initialized",",","dcp","'s","`","no_dist","<","https",":","\/\/pytorch.org\/docs\/stable\/distributed.checkpoint.html","#","torch.distributed.checkpoint.load_state_dict\/",">","_","`","argument","is","automatically","set","to","False",".","Otherwise","it","'s","set","to","True",".","Note",":","If","checkpointing","FSDP","model",",","you","can","set","state_dict","type","calling","`","set_state_dict_type","<","https",":","\/\/pytorch.org\/docs\/stable\/fsdp.html","#","torch.distributed.fsdp.FullyShardedDataParallel.set_state_dict_type",">","`","_","prior","to","starting","training","."]
code: def _checkpoint_impl(
      self,
      state: State,
      unit: AppStateMixin,
      *,
      checkpoint_path: str,
      hook: str,
  ) -> bool:
      intra_epoch = False
      if hook == "on_train_step_end":
          intra_epoch = True
  
      storage_writer = FsspecWriter(checkpoint_path)
  
      app_state = _prepare_app_state_for_checkpoint(
          state, unit, intra_epoch
      )
      # flag to indicate whether distributed is available
      # determines what to set ``no_dist`` arg in DCP apis
      pg_available: bool = dist.is_initialized()
      with get_timing_context(
          state, f"{self.__class__.__name__}.save_state_dict"
      ):
          dist_cp.save_state_dict(
              {"app_state": MultiStateful(app_state).state_dict()},
              storage_writer=storage_writer,
              process_group=self._process_group,
              no_dist=not pg_available,
          )
      return True
  
file_path: tnt/torchtnt/framework/callbacks/dcp_saver.py
imports: import logging
  from typing import Iterable, Optional
  import torch.distributed
  from torch.distributed import checkpoint
  from torch.distributed.checkpoint._fsspec_filesystem import FsspecReader, FsspecWriter
  from torchtnt.framework.callbacks._checkpoint_utils import _prepare_app_state_for_checkpoint, _prepare_app_state_for_restore, _TRAIN_DL_STATE_KEY
  from torchtnt.framework.callbacks.base_checkpointer import BaseCheckpointer
  from torchtnt.framework.callbacks.checkpointer_types import RestoreOptions
  from torchtnt.framework.state import State
  from torchtnt.framework.unit import AppStateMixin, TTrainData
  from torchtnt.framework.utils import get_timing_context
  from torchtnt.utils.rank_zero_log import rank_zero_info, rank_zero_warn
  from torchtnt.utils.stateful import MultiStateful, Stateful
label_desc: Use image node_id 2 for calling the DistributedCheckpointSaver obj's underlying member method code with example usage: obj._checkpoint_impl(state, unit) and returns: True