code: def _load(self, load_path: pathlib.Path) -> None:
      # Right now we will load all checkpoint shards on each node regardless of which
      # checkpoints are needed.
      # TODO (Liam): revisit later to optimize sharded checkpoint loading.
  
      # Load stateful things tracked by Determined on all slots.
      ckpt_path = (
          f"det_state_dict_rank{self.context.distributed.rank}.pth"
      )
      maybe_ckpt = load_path.joinpath(ckpt_path)
  
      if not maybe_ckpt.exists():
          return
  
      checkpoint = torch.load(str(maybe_ckpt), map_location="cpu")  # type: ignore
      if not isinstance(checkpoint, dict):
          raise det.errors.InvalidExperimentException(
              f"Expected checkpoint at {maybe_ckpt} to be a dict "
              f"but got {type(checkpoint).__name__}."
          )
  
      for callback in self.callbacks.values():
          callback.on_checkpoint_load_start(checkpoint)
  
      # We allow users to override load behavior if needed, but we default to using
      # the load method provided by DeepSpeed.
      self.trial.load(self.context, load_path)
  
      if "rng_state" in checkpoint:
          rng_state = checkpoint["rng_state"]
          np.random.set_state(rng_state["np_rng_state"])
          random.setstate(rng_state["random_rng_state"])
          torch.random.set_rng_state(rng_state["cpu_rng_state"])
  
          if torch.cuda.device_count():
              if "gpu_rng_state" in rng_state:
                  torch.cuda.set_rng_state(
                      rng_state["gpu_rng_state"],
                      device=self.context.distributed.get_local_rank(),
                  )
              else:
                  logger.warning(
                      "The system has a gpu but no gpu_rng_state exists in the checkpoint."
                  )
          else:
              if "gpu_rng_state" in rng_state:
                  logger.warning(
                      "There exists gpu_rng_state in checkpoint but the system has no gpu."
                  )
      else:
          logger.warning(
              "The checkpoint has no random state to restore."
          )
  
      callback_state = checkpoint.get("callbacks", {})
      for name in self.callbacks:
          if name in callback_state:
              self.callbacks[name].load_state_dict(callback_state[name])
          elif util.is_overridden(
              self.callbacks[name].load_state_dict,
              pytorch.PyTorchCallback,
          ):
              logger.warning(
                  "Callback '{}' implements load_state_dict(), but no callback state "
                  "was found for that name when restoring from checkpoint. This "
                  "callback will be initialized from scratch"
              )
  
      # Load workload sequencer state.
      wlsq_path = load_path.joinpath("workload_sequencer.pkl")
      if self.wlsq is not None and wlsq_path.exists():
          with wlsq_path.open("rb") as f:
              self.wlsq.load_state(pickle.load(f))
  
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 11 for calling the DeepSpeedTrialController obj's underlying member method code with example usage: obj._load(load_path) without return types