class_docstr: ``CheckpointContext`` gives access to checkpoint-related features of a Determined cluster.
class_docstr_tok: ["``","CheckpointContext","``","gives","access","to","checkpoint-related","features","of","a","Determined","cluster","."]
code: def _upload_sharded(
      self,
      ckpt_dir: Optional[str],
      storage_id: str,
      metadata: Optional[Dict[str, Any]] = None,
      *,
      selector: Optional[Callable[[str], bool]] = None,
  ) -> str:
      logger.debug(
          f"Uploading sharded content from checkpoint directory {ckpt_dir} to storage "
          f"(id: {storage_id}, metadata={metadata})"
      )
      if selector is not None and ckpt_dir is None:
          raise RuntimeError(
              "ckpt_dir has to be provided if selector is not None"
          )
  
      ckpt_dir_mask = self._dist.allgather(ckpt_dir is not None)
      if not any(ckpt_dir_mask):
          raise RuntimeError(
              "cannot call .upload(ckpt_dir=None, shard=True), from all ranks; "
              "at least one rank must have a valid ckpt_dir"
          )
  
      # Deconflict locally-shared directories; if multiple ranks (with no selectors) upload
      # /tmp/ckpt then only the lowest rank on each node will actually upload this directory.
      # If multiple ranks (with selectors) want to upload selected files from /tmp/ckpt, then
      # each rank will upload its selected files.
      if ckpt_dir is None:
          file_uid = None
      else:
          st = os.stat(ckpt_dir)
          file_uid = (st.st_dev, st.st_ino)
  
      all_file_uids = self._dist.allgather(file_uid)
  
      # Rank uploads resources only if (1) it has a selector; or
      # (2) it is the lowest local rank requested to upload this ckpt_dir.
      want_upload = selector is not None or (
          file_uid is not None
          and all_file_uids.index(file_uid) == self._dist.rank
      )
      # Collect and filter resources for all uploading ranks.
      if want_upload:
          assert ckpt_dir
          resources = self._storage_manager._list_directory(ckpt_dir)
          if selector is not None:
              resources = {
                  key: resources[key]
                  for key in resources
                  if selector(key)
              }
      else:
          resources = {}
  
      all_resources = self._dist.allgather(resources)
      merged_resources, conflicts = merge_resources(all_resources)
      resources = self._resolve_conflicts(
          resources, conflicts, ckpt_dir
      )
  
      # Merge and upload metadata.
      all_metadata = self._merge_metadata(metadata)
      upload_mask = self._dist.allgather(want_upload)
      metadata_upload_rank = upload_mask.index(True)
      if self._dist.rank == metadata_upload_rank:
          assert ckpt_dir
          self._write_metadata_file(ckpt_dir, all_metadata)
          resources["metadata.json"] = os.path.getsize(
              os.path.join(ckpt_dir, "metadata.json")
          )
  
      if want_upload:
          assert ckpt_dir
          paths = set(resources.keys())
          self._storage_manager.upload(
              src=ckpt_dir, dst=storage_id, paths=paths
          )
  
      # Synchronize workers.
      _ = self._dist.allgather(None)
  
      if self._dist.rank == 0:
          self._report_checkpoint(
              storage_id, merged_resources, all_metadata
          )
      return storage_id
  
imports: import contextlib
  import datetime
  import enum
  import hashlib
  import json
  import logging
  import os
  import pathlib
  import uuid
  from typing import Any, Callable, Dict, Iterator, List, Optional, Set, Tuple, Union
  from determined import core, tensorboard
  from determined.common import api, storage
  from determined.common.api import bindings
label_desc: Use image node_id 4 for calling the CheckpointContext obj's underlying member method code with example usage: obj._upload_sharded(ckpt_dir, storage_id, metadata) and returns: storage_id