code: def _save(self, path: pathlib.Path) -> None:
      if self.context.distributed.local_rank == 0:
          path.mkdir(parents=True, exist_ok=True)
      _ = self.context.distributed.gather_local(None)  # sync
  
      if self.is_chief:
          # We assume these stateful objects should be the same across slots and only have
          # the chief save them.
          util.write_user_code(path, self.env.on_cluster)
  
          if self.wlsq is not None:
              with path.joinpath("workload_sequencer.pkl").open(
                  "wb"
              ) as f:
                  pickle.dump(self.wlsq.get_state(), f)
  
      # Save per rank Determined checkpoint.
      rng_state = {
          "cpu_rng_state": torch.random.get_rng_state(),
          "np_rng_state": np.random.get_state(),
          "random_rng_state": random.getstate(),
      }
  
      if torch.cuda.device_count():
          rng_state["gpu_rng_state"] = torch.cuda.get_rng_state(
              self.context.distributed.get_local_rank()
          )
      checkpoint = {"rng_state": rng_state}
  
      # PyTorch uses optimizer objects that take the model parameters to
      # optimize on construction, so we store and reload the `state_dict()`
      # of the model and optimizer explicitly (instead of dumping the entire
      # objects) to avoid breaking the connection between the model and the
      # optimizer.
      checkpoint["callbacks"] = {
          name: callback.state_dict()
          for name, callback in self.callbacks.items()
      }
  
      for callback in self.callbacks.values():
          callback.on_checkpoint_save_start(checkpoint)
  
      ckpt_name = (
          f"det_state_dict_rank{self.context.distributed.rank}.pth"
      )
      torch.save(checkpoint, str(path.joinpath(ckpt_name)))
  
      # We allow users to override save behavior if needed, but we default to using
      # the save method provided by DeepSpeed.
      self.trial.save(self.context, path)
  
      for callback in self.callbacks.values():
          # TODO(DET-7912): remove on_checkpoint_end once it has been deprecated long enough.
          callback.on_checkpoint_end(str(path))
          callback.on_checkpoint_write_end(str(path))
  
imports: import abc
  import contextlib
  import logging
  import os
  import pathlib
  import pickle
  import random
  import time
  import warnings
  from typing import Any, Callable, Dict, Iterator, List, Optional, Type, Union, cast
  import deepspeed
  import numpy
  import torch
  from deepspeed.runtime import dataloader
  import determined
  from determined import layers, pytorch, util, workload
  from determined.pytorch import deepspeed
  from determined.pytorch import dsat
label_desc: Use image node_id 12 for calling the DeepSpeedTrialController obj's underlying member method code with example usage: obj._save(path) without return types