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 _store_path_sharded(
      self, metadata: Optional[Dict[str, Any]] = None
  ) -> Iterator[Tuple[pathlib.Path, str]]:
      logger.debug(
          f"Getting path for sharded storage (metadata={metadata})"
      )
      storage_id = None
      if self._dist.rank == 0:
          storage_id = str(uuid.uuid4())
      storage_id = self._dist.broadcast(storage_id)
      assert storage_id
  
      path = self._storage_manager.pre_store_path(storage_id)
      yield path, storage_id
  
      ckpt_dir = os.fspath(path)
  
      if self._storage_manager.store_path_is_direct_access():
          # Each rank saves files directly to ckpt_dir which means there is no conflict
          # detection on upload. Metadata still needs to be merged and saved,
          # and checkpoint has to be reported.
          all_metadata = self._merge_metadata(metadata)
  
          if self._dist.rank == 0:
              self._write_metadata_file(os.fspath(path), all_metadata)
              resources = self._storage_manager._list_directory(
                  ckpt_dir
              )
              self._report_checkpoint(
                  storage_id, resources, all_metadata
              )
  
          return
  
      # Deconflict locally-shared directories; if every worker uploads /tmp/ckpt, then only
      # the lowest rank on each node will actually upload this directory.
      st = os.stat(ckpt_dir)
      file_uid = (st.st_dev, st.st_ino)
      all_file_uids = self._dist.allgather(file_uid)
      # Decide if our rank is the lowest rank trying to upload this ckpt_dir.
      want_upload = all_file_uids.index(file_uid) == self._dist.rank
  
      # Decide what we are going to upload.
      if want_upload:
          assert ckpt_dir
          resources = self._storage_manager._list_directory(ckpt_dir)
      else:
          resources = {}
  
      # Merge resources, detect conflicts.
      all_resources = self._dist.allgather(resources)
  
      merged_resources, conflicts = merge_resources(all_resources)
      self._resolve_conflicts(resources, conflicts, ckpt_dir)
  
      all_metadata = self._merge_metadata(metadata)
      if self._dist.rank == 0:
          self._write_metadata_file(ckpt_dir, all_metadata)
  
      if want_upload:
          # Use post_store_path to upload and clean up ckpt_dir after uploading.
          self._storage_manager.post_store_path(
              src=ckpt_dir, dst=storage_id
          )
  
      if self._dist.rank == 0:
          self._report_checkpoint(
              storage_id, merged_resources, all_metadata
          )
  
      # Synchronize workers.
      _ = self._dist.allgather(None)
  
      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 11 for calling the CheckpointContext obj's underlying member method code with example usage: obj._store_path_sharded(metadata) and returns: storage_id