Diff Coverage

        together either.
    """
    if storage_reader is None:
        storage_reader = FileSystemReader(checkpoint_id)
    elif checkpoint_id:
        storage_reader.initialize_reader(checkpoint_id)

    try:
        return storage_reader, storage_reader.load_metadata(), True
    except FileNotFoundError:

    use_collectives = False if no_dist else use_collectives

    # Check the arguments
    if broadcast_groups and not broadcast_replicated_tensors:
        raise ValueError("broadcast_groups is only used when broadcast_replicated_tensors is True.")
    if storage_reader is None and checkpoint_id is None:
        raise ValueError("Either storage_reader or checkpoint_id must be provided")

    # Set up planner

        else:
            # Scalar entries (rank-0 tensors such as the AdamW ``step``) have no slices to
            # narrow by, and safetensors before 0.4.3 rejects the empty index with
            # "too many indices for tensor of dimension 0" - read the whole tensor instead.
            tensor = tensor_file.get_tensor(tensor_key)
        fetched.append((req, tensor))
    return fetched

        reqs: list[ReadItem],
        storage_data: dict[MetadataIndex, StorageInfo],
) -> list[tuple[ReadItem, Any]]:
    """Narrow the region each item asked for out of a file the ms adapter has read in."""
    fetched: list[tuple[ReadItem, Any]] = []
    for req in reqs:
        storage_info = _get_storage_info(req, storage_data)
        tensor_key = storage_info.tensor_key or req.storage_index.fqn
        if tensor_key not in param_dict:
            raise KeyError(f"Key {tensor_key} not found in checkpoint file")
        fetched.append((req, narrow_tensor_by_index(
            param_dict[tensor_key],
            req.storage_offsets,
            req.lengths,
        )))
    return fetched


def _fetch_tensor_file(
        tensor_file: Any,

        list[tuple[ReadItem, Any]]: Each item with the region it asked for.
    """
    if platform.platform_type == PlatformType.PYTORCH:
        return _fetch_torch_tensor_file(tensor_file, reqs, storage_data)
    return _fetch_ms_tensor_file(tensor_file, reqs, storage_data)


def _apply_fetched(
        fetched: list[tuple[ReadItem, Any]],

            _TORCH_FILES_KEPT,
            lambda path: safe_open(path, framework="pt", device="cpu"),
            lambda reader: reader.__exit__(None, None, None),
        )
    return _OpenFiles(_MS_FILES_KEPT, platform.load_checkpoint, lambda reader: None)


def _broadcast_batch_bytes(requested: int) -> int:
    """

        reach the buffer behind them; there every shard is sent on its own, as before.
    """
    if platform.platform_type == PlatformType.PYTORCH:
        return requested
    return 0


class FileSystemReader(StorageReader):
    """

        ValueError: If the gathered list holds no plan for this rank.
    """
    own_index = rank if len(all_plans) > 1 else 0
    if not 0 <= own_index < len(all_plans):
        raise ValueError(
            f"Rank {rank} has no plan of its own among the {len(all_plans)} gathered "
            "plans; the gathered list must hold one plan per rank, in rank order."
        )
    return own_index

    while len(in_flight) >= _MAX_BROADCASTS_IN_FLIGHT:
        _finish_broadcast(in_flight.popleft())
    handle = platform.broadcast_async(buffer, source.src_rank, groups[source.group_ranks])
    if handle is None:
        if after is not None:
            after()
        return
    in_flight.append((handle, after))


def _finish_broadcast(pending: tuple) -> None:

        """Send one batch, and arrange for it to be dealt out once it has landed."""
        buffers = self._batches.pop(key, [])
        self._pending_bytes.pop(key, None)
        if not buffers:
            return

        group_ranks, src_rank = key[:2]
        source = BroadcastSource(group_ranks=group_ranks, src_rank=src_rank)
        self.sent += 1

        Any: A tensor view over the shard, writable in place by a collective.
    """
    if isinstance(obj, DTensor):
        if obj.layout is not None and obj.layout.ragged_shard is not None:
            return get_ragged_box_tensor(obj, index).detach()
        return obj.to_local().detach()
    return obj.detach()

        Args:
            dests (list): Tensors to copy into, one per source.
            srcs (list): Tensors to copy from, matching ``dests`` in length and shape.
        """
        for dest, src in zip(dests, srcs):
            if hasattr(dest, "copy_"):
                dest.copy_(src)
            else:
                dest[...] = src

    @staticmethod
    def differentiable_all_reduce(data, op, group):
        """Perform differentiable all-reduce operation.

        Returns:
            A work handle to wait on, or None if the backend completed the call inline.
        """
        raise NotImplementedError("Platform subclasses must implement broadcast_async")

    @staticmethod
    def scatter(output, scatter_list, src=None, group=None, async_op=False, group_src=None):
        """Scatter tensor list from source rank to all ranks in group."""

        Returns:
            The process group of every rank.
        """
        raise NotImplementedError("Platform subclasses must implement get_world_group")

    @staticmethod
    def get_created_group(rank_list: Union[list[int], tuple[int]]):
        """Get an existing process group by rank list.

        Measured on an Ascend card, five hundred copies of 64 KiB take 11.8 ms one at a
        time and 1.4 ms fused: nearly all of what a small copy costs is starting it.
        """
        if dests:
            torch._foreach_copy_(dests, srcs)  # pylint: disable=protected-access

    @staticmethod
    def parameters_dict(cell: Module):
        return cell.named_parameters()

            handle.wait()

    @staticmethod
    def broadcast_async(data, src=None, group=None, group_src=None):
        if group_src is not None:
            src = dist.get_global_rank(group, group_src)
        return dist.broadcast(data, src, group, async_op=True)

    @staticmethod
    def scatter(output, scatter_list, src=None, group=None, async_op=False, group_src=None):
        if group_src is not None:

    @staticmethod
    def get_world_group() -> ProcessGroup:
        """The default process group, which holds every rank."""
        return _get_default_group()

    @staticmethod
    def destroy_process_group(group: Optional[ProcessGroup] = None) -> None:
        """