docstr: Save a state to disk.
  
  Modified from brentyi/fannypack.
  
  Args:
    save_path (str): The name of the checkpoint to save.
    global_step (int): The iteration number which will be used
       to name the checkpoint.
code: def save(self, save_path: str, global_step: int):
      """Save a state to disk.
  
      Modified from brentyi/fannypack.
  
      Args:
        save_path (str): The name of the checkpoint to save.
        global_step (int): The iteration number which will be used
           to name the checkpoint.
      """
      if self.is_local_rank_0():
          state = {
              "global_step": global_step,
              "model_weights": self.get_model_state_dict(),
          }
          if self.optimizer is not None:
              state["optim_state"] = self.optimizer.state_dict()
          if self.scheduler is not None:
              state["scheduler_state"] = self.scheduler.state_dict()
  
          # ignore ctrl+c while saving
          try:
              orig_handler = signal.getsignal(signal.SIGINT)
              signal.signal(signal.SIGINT, lambda _sig, _frame: None)
          except ValueError:
              # signal throws a ValueError if we're not in the main thread
              orig_handler = None
  
          try:
              # atomic save
              with tempfile.TemporaryDirectory() as tmpdir:
                  # Save to a temporary directory outside of the checkpoint dir so
                  # async processes do not try and copy a partially-written checkpoint.
                  # See Ray Tune and MLFlow for examples of background processes that
                  # are affected by this.
                  tmp_path = os.path.join(tmpdir, "temp.ckpt")
                  torch.save(state, tmp_path)
  
                  self.safe_move_file(tmp_path, save_path)
                  logger.debug(f"Saved checkpoint at {save_path}.")
          finally:
              # restore SIGINT handler
              if orig_handler is not None:
                  signal.signal(signal.SIGINT, orig_handler)
      self.distributed.barrier()
  
imports: import errno
  import logging
  import os
  import re
  import shutil
  import signal
  import tempfile
  import uuid
  from abc import ABC, abstractmethod
  from glob import glob
  from typing import Any, Dict, Mapping, Optional, Tuple, TYPE_CHECKING
  import torch
  from torch.optim import Optimizer
  from ludwig.api_annotations import DeveloperAPI
  from ludwig.modules.lr_scheduler import LRScheduler
label_desc: Use image node_id 4 for calling the MultiNodeCheckpoint obj's underlying member method code with example usage: obj.save(save_path, global_step) without return types