We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Sharded checkpointing fails on load for certain tensor sizes
### 🐛 Describe the bug

Sharded checkpointing (particularly with FSDP for the optimizer state) uses ChunkShardingSpec to save/load tensors. ChunkShardingSpec's behavior is similar to torch.chunk and will result in some chunks of size 0.

This can be reproduced by trying to save a tensor of size 6 with 4 gpus. This tensor is sharded across the first 3 gpus. The resulting size of the chunks will look like [2, 2, 2, 0]. On save, it seems like ChunkShardingSpec is aware of which gpus contain shards, so it saves the tensor with shard metadata showing the size of chunks to be [2, 2, 2].

The problem occurs when attempting to load the sharded checkpoint. ChunkShardingSpec attempts to rebuild the metadata, this time being unaware of how many gpus originally contained shards. It knows that there is a tensor of size 6 and 4 gpus though, so it generates shard metadata with chunk sizes [2, 2, 2], [skipping the last gpu since it has size 0](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L94). Then when attempting to shard the tensor, the 4th gpu has no shard metadata, so a [local_tensor is never created](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L165), resulting in an [assertion error on the 4th rank](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L172) and a [type error on all other ranks](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/distributed_c10d.py#L808) because None is contained in the scatter_list.

There are a couple possible solutions to this.
1. Add shardmetadata for all gpus that allow for a tensor to be size 0
2. Change ChunkShardingSpec to distribute a tensor evenly across gpus (e.g. [2, 2, 1, 1] instead of [2, 2, 2, 0])

I've implemented and tested both solutions and both are backwards compatible with previously saved sharded checkpoints on versions 2.0.1, 2.1.0-rc3, and 8/27 nightly (e.g. save on 2.0.1 with the old ChunkShardingSpec and load on 2.0.1 with the new ChunkShardingSpec). Both solutions are also cross-version compatible for 2.0.1->2.1.0-rc3, 2.0.1->8/27 nightly, and 8/27 nightly->2.0.1 (e.g. save on 2.0.1 with the old ChunkShardingSpec and load on 2.1.0 with the new ChunkShardingSpec). The solutions might be version/cross-version compatible for other combinations, but I haven't tested those.

### Versions

This happens with pytorch 2.0.1, 2.1.0-rc3, and 8/27 nightly.

cc @mrshenli @pritamdamania87 @zhaojuanmao @satgera @rohan-varma @gqchen @aazzolini @osalpekar @jiayisuse @H-Huang @kwen2501 @awgu @penguinwu

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py`
Content:
```
1 from dataclasses import dataclass
2 import torch
3 import torch.distributed._shard.sharded_tensor.metadata as sharded_tensor_meta
4 from torch.distributed._shard.metadata import ShardMetadata
5 from torch.distributed._shard.sharded_tensor.shard import Shard
6 from torch.distributed._shard.sharded_tensor.utils import (
7     _parse_and_validate_remote_device
8 )
9 from torch.distributed._shard._utils import narrow_tensor
10 import torch.distributed as dist
11 import torch.distributed.distributed_c10d as distributed_c10d
12 from typing import List, Union, TYPE_CHECKING
13 from ._internals import (
14     get_chunked_dim_size,
15     get_split_size,
16 )
17 
18 from .api import ShardingSpec
19 
20 if TYPE_CHECKING:
21     # Only include ShardedTensor when do type checking, exclude it
22     # from run-time to resolve circular dependency.
23     from torch.distributed._shard.sharded_tensor import ShardedTensor
24 
25 @dataclass
26 class ChunkShardingSpec(ShardingSpec):
27     """
28     This is a type of PlacementSpec that defines the placement as being sharded
29     across multiple devices. In particular, it represents sharding a Tensor
30     along a single dimension into equal chunks (similar to :meth:`torch.chunk`).
31 
32     The semantics of how a tensor is partitioned is inline with
33     :meth:`torch.chunk`, where ``dim`` in torch.chunk corresponds to the
34     specified ``dim`` and ``chunks`` in torch.chunk is the number of elements
35     in the placement specified.
36 
37     Args:
38         dim (int or str):
39             The dimension to shard on, could be an integer representing the
40             dimension or a string in case of named tensors where dimensions are
41             named. Note that named tensor support is not added yet.
42         placement(List[Union[_remote_device, str]]):
43             Specifies the placement of each shard of the Tensor. The size of
44             the list represents the number of shards to be created. This could
45             be a list of
46             :class:`torch.distributed._remote_device`'s. This list
47             could also contain a string which represents remote
48             device as accepted by
49             :class:`torch.distributed._remote_device`
50     """
51 
52     ShardingDim = Union[int, str]
53 
54     dim: ShardingDim
55     placements: List[Union[torch.distributed._remote_device, str]]
56 
57     def __post_init__(self):
58         self._verify_dim(self.dim)
59         for i, remote_device in enumerate(self.placements):
60             if not isinstance(remote_device, torch.distributed._remote_device):
61                 self.placements[i] = torch.distributed._remote_device(remote_device)
62 
63     @staticmethod
64     def _verify_dim(dim):
65         # Validate the sharding spec.
66         # TODO: support named dimension
67         if isinstance(dim, str):
68             raise NotImplementedError(
69                 "ChunkShardingSpec does not support named dimension yet!"
70             )
71 
72         if not isinstance(dim, int):
73             raise ValueError(
74                 f"Sharding dim needs to be an integer, found: {dim}"
75             )
76 
77     def build_metadata(self,
78                        tensor_sizes: torch.Size,
79                        tensor_properties: sharded_tensor_meta.TensorProperties,
80                        ) -> sharded_tensor_meta.ShardedTensorMetadata:
81         tensor_num_dim = len(tensor_sizes)
82 
83         self._verify_dim(self.dim)
84         if self.dim >= tensor_num_dim or self.dim < -tensor_num_dim:  # type: ignore[operator]
85             raise ValueError(f"Invalid sharding dim: {self.dim}")
86 
87         shards_metadata = []
88         sharding_dim_size = tensor_sizes[self.dim]  # type: ignore[index]
89         chunks = len(self.placements)
90         split_size = get_split_size(sharding_dim_size, chunks)
91         for idx, placement in enumerate(self.placements):
92             # generate ShardMetadata for each placement device
93             chunked_dim_size = get_chunked_dim_size(sharding_dim_size, split_size, idx)
94             if chunked_dim_size > 0:
95                 shard_size = list(tensor_sizes)
96                 current_offsets = [0] * tensor_num_dim
97                 current_offsets[self.dim] = split_size * idx  # type: ignore[index]
98                 shard_size[self.dim] = chunked_dim_size  # type: ignore[index]
99 
100                 shard_metadata = ShardMetadata(
101                     shard_offsets=current_offsets,
102                     shard_sizes=shard_size,
103                     placement=placement,
104                 )
105                 shards_metadata.append(shard_metadata)
106 
107                 # current_offsets[self.dim] += chunked_dim_size  # type: ignore[index]
108 
109         return sharded_tensor_meta.ShardedTensorMetadata(
110             shards_metadata,
111             tensor_sizes,
112             tensor_properties
113         )
114 
115 
116     def shard(self, tensor: torch.Tensor, src_rank: int = 0, process_group=None) -> "ShardedTensor":
117         """
118         Args:
119             src_rank: group rank relative to ``process_group``
120 
121             N.B. If ``process_group`` is None, ``src_rank`` is a global rank.
122         """
123         # relative imports to avoid circular dependency
124         from torch.distributed._shard.sharded_tensor import (
125             ShardedTensor
126         )
127         tensor_properties = sharded_tensor_meta.TensorProperties(
128             dtype=tensor.dtype,
129             layout=tensor.layout,
130             requires_grad=tensor.requires_grad,
131             memory_format=torch.contiguous_format,
132             pin_memory=tensor.is_pinned()
133         )
134         current_rank = dist.get_rank(process_group)
135         tensor_meta = self.build_metadata(tensor.size(), tensor_properties)
136         local_shards = []
137         local_tensor = None
138         local_metadata = None
139         tensors_to_scatter = [None] * dist.get_world_size(process_group)
140 
141         sharding_dim_size = tensor.size()[self.dim]  # type: ignore[index]
142         chunks = len(self.placements)
143         split_size = get_split_size(sharding_dim_size, chunks)
144         scatter_shape = list(tensor.size())
145         scatter_shape[self.dim] = split_size  # type: ignore[index]
146 
147         for shard_meta in tensor_meta.shards_metadata:
148             rank, device = _parse_and_validate_remote_device(process_group, shard_meta.placement)
149             if current_rank == src_rank:
150                 # Reshape to get shard for this rank and we don't want autograd
151                 # recording here for the narrow op and 'local_shard' should be a
152                 # leaf variable in the autograd graph.
153                 narrowed_tensor = narrow_tensor(tensor, shard_meta)
154                 if shard_meta.shard_sizes[self.dim] < split_size:  # type: ignore[index]
155                     # for the last shard that might be smaller to other shards
156                     # resize the narrowed tensor to the same size and use it for
157                     # the scatter collective as dist.scatter requires same size
158                     # inputs on every rank
159                     tensor_to_scatter = narrowed_tensor.detach().clone().resize_(scatter_shape)
160                 else:
161                     tensor_to_scatter = narrowed_tensor.detach().clone().contiguous()
162 
163                 tensors_to_scatter[rank] = tensor_to_scatter
164 
165             if current_rank == rank:
166                 local_tensor = torch.empty(
167                     scatter_shape, dtype=tensor.dtype, layout=tensor.layout, device=device)
168                 local_metadata = shard_meta
169 
170         # each rank should have local_tensor and local_metadata initialized if we build
171         # the metadata list in a correct way.
172         assert local_tensor is not None
173         assert local_metadata is not None
174 
175         # Scatter the shards to all ranks in the pg
176         # scatter takes the global rank as ``src``
177         src_for_scatter = src_rank
178         if process_group is not None and process_group is not distributed_c10d._get_default_group():
179             src_for_scatter = distributed_c10d.get_global_rank(process_group, src_for_scatter)
180 
181         dist.scatter(
182             local_tensor,
183             scatter_list=tensors_to_scatter if current_rank == src_rank else None,
184             src=src_for_scatter,
185             group=process_group
186         )
187 
188         if list(local_tensor.size()) != local_metadata.shard_sizes:
189             # detach again after receiving to ensure local shards remain a leaf node
190             local_tensor = local_tensor.resize_(local_metadata.shard_sizes).detach()
191 
192         # Sync requires_grad to local_shard.
193         local_tensor.requires_grad = tensor.requires_grad
194 
195         local_shards.append(Shard(tensor=local_tensor, metadata=local_metadata))
196 
197         st = ShardedTensor._init_from_local_shards_and_global_metadata(
198             local_shards,
199             tensor_meta,
200             process_group=process_group)
201 
202         # Manually set sharding_spec
203         st._sharding_spec = self
204 
205         return st
206
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py b/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py
--- a/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py
+++ b/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py
@@ -91,20 +91,17 @@
         for idx, placement in enumerate(self.placements):
             # generate ShardMetadata for each placement device
             chunked_dim_size = get_chunked_dim_size(sharding_dim_size, split_size, idx)
-            if chunked_dim_size > 0:
-                shard_size = list(tensor_sizes)
-                current_offsets = [0] * tensor_num_dim
-                current_offsets[self.dim] = split_size * idx  # type: ignore[index]
-                shard_size[self.dim] = chunked_dim_size  # type: ignore[index]
-
-                shard_metadata = ShardMetadata(
-                    shard_offsets=current_offsets,
-                    shard_sizes=shard_size,
-                    placement=placement,
-                )
-                shards_metadata.append(shard_metadata)
-
-                # current_offsets[self.dim] += chunked_dim_size  # type: ignore[index]
+            shard_size = list(tensor_sizes)
+            current_offsets = [0] * tensor_num_dim
+            current_offsets[self.dim] = split_size * idx  # type: ignore[index]
+            shard_size[self.dim] = chunked_dim_size  # type: ignore[index]
+
+            shard_metadata = ShardMetadata(
+                shard_offsets=current_offsets,
+                shard_sizes=shard_size,
+                placement=placement,
+            )
+            shards_metadata.append(shard_metadata)
 
         return sharded_tensor_meta.ShardedTensorMetadata(
             shards_metadata,

{"golden_diff": "diff --git a/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py b/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py\n--- a/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py\n+++ b/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py\n@@ -91,20 +91,17 @@\n         for idx, placement in enumerate(self.placements):\n             # generate ShardMetadata for each placement device\n             chunked_dim_size = get_chunked_dim_size(sharding_dim_size, split_size, idx)\n-            if chunked_dim_size > 0:\n-                shard_size = list(tensor_sizes)\n-                current_offsets = [0] * tensor_num_dim\n-                current_offsets[self.dim] = split_size * idx  # type: ignore[index]\n-                shard_size[self.dim] = chunked_dim_size  # type: ignore[index]\n-\n-                shard_metadata = ShardMetadata(\n-                    shard_offsets=current_offsets,\n-                    shard_sizes=shard_size,\n-                    placement=placement,\n-                )\n-                shards_metadata.append(shard_metadata)\n-\n-                # current_offsets[self.dim] += chunked_dim_size  # type: ignore[index]\n+            shard_size = list(tensor_sizes)\n+            current_offsets = [0] * tensor_num_dim\n+            current_offsets[self.dim] = split_size * idx  # type: ignore[index]\n+            shard_size[self.dim] = chunked_dim_size  # type: ignore[index]\n+\n+            shard_metadata = ShardMetadata(\n+                shard_offsets=current_offsets,\n+                shard_sizes=shard_size,\n+                placement=placement,\n+            )\n+            shards_metadata.append(shard_metadata)\n \n         return sharded_tensor_meta.ShardedTensorMetadata(\n             shards_metadata,\n", "issue": "Sharded checkpointing fails on load for certain tensor sizes\n### \ud83d\udc1b Describe the bug\n\nSharded checkpointing (particularly with FSDP for the optimizer state) uses ChunkShardingSpec to save/load tensors. ChunkShardingSpec's behavior is similar to torch.chunk and will result in some chunks of size 0.\r\n\r\nThis can be reproduced by trying to save a tensor of size 6 with 4 gpus. This tensor is sharded across the first 3 gpus. The resulting size of the chunks will look like [2, 2, 2, 0]. On save, it seems like ChunkShardingSpec is aware of which gpus contain shards, so it saves the tensor with shard metadata showing the size of chunks to be [2, 2, 2].\r\n\r\nThe problem occurs when attempting to load the sharded checkpoint. ChunkShardingSpec attempts to rebuild the metadata, this time being unaware of how many gpus originally contained shards. It knows that there is a tensor of size 6 and 4 gpus though, so it generates shard metadata with chunk sizes [2, 2, 2], [skipping the last gpu since it has size 0](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L94). Then when attempting to shard the tensor, the 4th gpu has no shard metadata, so a [local_tensor is never created](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L165), resulting in an [assertion error on the 4th rank](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py#L172) and a [type error on all other ranks](https://github.com/pytorch/pytorch/blob/ddbaad6d746d34b56f2d2cc0e02d7e0d0301e626/torch/distributed/distributed_c10d.py#L808) because None is contained in the scatter_list.\r\n\r\nThere are a couple possible solutions to this.\r\n1. Add shardmetadata for all gpus that allow for a tensor to be size 0\r\n2. Change ChunkShardingSpec to distribute a tensor evenly across gpus (e.g. [2, 2, 1, 1] instead of [2, 2, 2, 0])\r\n\r\nI've implemented and tested both solutions and both are backwards compatible with previously saved sharded checkpoints on versions 2.0.1, 2.1.0-rc3, and 8/27 nightly (e.g. save on 2.0.1 with the old ChunkShardingSpec and load on 2.0.1 with the new ChunkShardingSpec). Both solutions are also cross-version compatible for 2.0.1->2.1.0-rc3, 2.0.1->8/27 nightly, and 8/27 nightly->2.0.1 (e.g. save on 2.0.1 with the old ChunkShardingSpec and load on 2.1.0 with the new ChunkShardingSpec). The solutions might be version/cross-version compatible for other combinations, but I haven't tested those.\n\n### Versions\n\nThis happens with pytorch 2.0.1, 2.1.0-rc3, and 8/27 nightly.\n\ncc @mrshenli @pritamdamania87 @zhaojuanmao @satgera @rohan-varma @gqchen @aazzolini @osalpekar @jiayisuse @H-Huang @kwen2501 @awgu @penguinwu\n", "before_files": [{"content": "from dataclasses import dataclass\nimport torch\nimport torch.distributed._shard.sharded_tensor.metadata as sharded_tensor_meta\nfrom torch.distributed._shard.metadata import ShardMetadata\nfrom torch.distributed._shard.sharded_tensor.shard import Shard\nfrom torch.distributed._shard.sharded_tensor.utils import (\n    _parse_and_validate_remote_device\n)\nfrom torch.distributed._shard._utils import narrow_tensor\nimport torch.distributed as dist\nimport torch.distributed.distributed_c10d as distributed_c10d\nfrom typing import List, Union, TYPE_CHECKING\nfrom ._internals import (\n    get_chunked_dim_size,\n    get_split_size,\n)\n\nfrom .api import ShardingSpec\n\nif TYPE_CHECKING:\n    # Only include ShardedTensor when do type checking, exclude it\n    # from run-time to resolve circular dependency.\n    from torch.distributed._shard.sharded_tensor import ShardedTensor\n\n@dataclass\nclass ChunkShardingSpec(ShardingSpec):\n    \"\"\"\n    This is a type of PlacementSpec that defines the placement as being sharded\n    across multiple devices. In particular, it represents sharding a Tensor\n    along a single dimension into equal chunks (similar to :meth:`torch.chunk`).\n\n    The semantics of how a tensor is partitioned is inline with\n    :meth:`torch.chunk`, where ``dim`` in torch.chunk corresponds to the\n    specified ``dim`` and ``chunks`` in torch.chunk is the number of elements\n    in the placement specified.\n\n    Args:\n        dim (int or str):\n            The dimension to shard on, could be an integer representing the\n            dimension or a string in case of named tensors where dimensions are\n            named. Note that named tensor support is not added yet.\n        placement(List[Union[_remote_device, str]]):\n            Specifies the placement of each shard of the Tensor. The size of\n            the list represents the number of shards to be created. This could\n            be a list of\n            :class:`torch.distributed._remote_device`'s. This list\n            could also contain a string which represents remote\n            device as accepted by\n            :class:`torch.distributed._remote_device`\n    \"\"\"\n\n    ShardingDim = Union[int, str]\n\n    dim: ShardingDim\n    placements: List[Union[torch.distributed._remote_device, str]]\n\n    def __post_init__(self):\n        self._verify_dim(self.dim)\n        for i, remote_device in enumerate(self.placements):\n            if not isinstance(remote_device, torch.distributed._remote_device):\n                self.placements[i] = torch.distributed._remote_device(remote_device)\n\n    @staticmethod\n    def _verify_dim(dim):\n        # Validate the sharding spec.\n        # TODO: support named dimension\n        if isinstance(dim, str):\n            raise NotImplementedError(\n                \"ChunkShardingSpec does not support named dimension yet!\"\n            )\n\n        if not isinstance(dim, int):\n            raise ValueError(\n                f\"Sharding dim needs to be an integer, found: {dim}\"\n            )\n\n    def build_metadata(self,\n                       tensor_sizes: torch.Size,\n                       tensor_properties: sharded_tensor_meta.TensorProperties,\n                       ) -> sharded_tensor_meta.ShardedTensorMetadata:\n        tensor_num_dim = len(tensor_sizes)\n\n        self._verify_dim(self.dim)\n        if self.dim >= tensor_num_dim or self.dim < -tensor_num_dim:  # type: ignore[operator]\n            raise ValueError(f\"Invalid sharding dim: {self.dim}\")\n\n        shards_metadata = []\n        sharding_dim_size = tensor_sizes[self.dim]  # type: ignore[index]\n        chunks = len(self.placements)\n        split_size = get_split_size(sharding_dim_size, chunks)\n        for idx, placement in enumerate(self.placements):\n            # generate ShardMetadata for each placement device\n            chunked_dim_size = get_chunked_dim_size(sharding_dim_size, split_size, idx)\n            if chunked_dim_size > 0:\n                shard_size = list(tensor_sizes)\n                current_offsets = [0] * tensor_num_dim\n                current_offsets[self.dim] = split_size * idx  # type: ignore[index]\n                shard_size[self.dim] = chunked_dim_size  # type: ignore[index]\n\n                shard_metadata = ShardMetadata(\n                    shard_offsets=current_offsets,\n                    shard_sizes=shard_size,\n                    placement=placement,\n                )\n                shards_metadata.append(shard_metadata)\n\n                # current_offsets[self.dim] += chunked_dim_size  # type: ignore[index]\n\n        return sharded_tensor_meta.ShardedTensorMetadata(\n            shards_metadata,\n            tensor_sizes,\n            tensor_properties\n        )\n\n\n    def shard(self, tensor: torch.Tensor, src_rank: int = 0, process_group=None) -> \"ShardedTensor\":\n        \"\"\"\n        Args:\n            src_rank: group rank relative to ``process_group``\n\n            N.B. If ``process_group`` is None, ``src_rank`` is a global rank.\n        \"\"\"\n        # relative imports to avoid circular dependency\n        from torch.distributed._shard.sharded_tensor import (\n            ShardedTensor\n        )\n        tensor_properties = sharded_tensor_meta.TensorProperties(\n            dtype=tensor.dtype,\n            layout=tensor.layout,\n            requires_grad=tensor.requires_grad,\n            memory_format=torch.contiguous_format,\n            pin_memory=tensor.is_pinned()\n        )\n        current_rank = dist.get_rank(process_group)\n        tensor_meta = self.build_metadata(tensor.size(), tensor_properties)\n        local_shards = []\n        local_tensor = None\n        local_metadata = None\n        tensors_to_scatter = [None] * dist.get_world_size(process_group)\n\n        sharding_dim_size = tensor.size()[self.dim]  # type: ignore[index]\n        chunks = len(self.placements)\n        split_size = get_split_size(sharding_dim_size, chunks)\n        scatter_shape = list(tensor.size())\n        scatter_shape[self.dim] = split_size  # type: ignore[index]\n\n        for shard_meta in tensor_meta.shards_metadata:\n            rank, device = _parse_and_validate_remote_device(process_group, shard_meta.placement)\n            if current_rank == src_rank:\n                # Reshape to get shard for this rank and we don't want autograd\n                # recording here for the narrow op and 'local_shard' should be a\n                # leaf variable in the autograd graph.\n                narrowed_tensor = narrow_tensor(tensor, shard_meta)\n                if shard_meta.shard_sizes[self.dim] < split_size:  # type: ignore[index]\n                    # for the last shard that might be smaller to other shards\n                    # resize the narrowed tensor to the same size and use it for\n                    # the scatter collective as dist.scatter requires same size\n                    # inputs on every rank\n                    tensor_to_scatter = narrowed_tensor.detach().clone().resize_(scatter_shape)\n                else:\n                    tensor_to_scatter = narrowed_tensor.detach().clone().contiguous()\n\n                tensors_to_scatter[rank] = tensor_to_scatter\n\n            if current_rank == rank:\n                local_tensor = torch.empty(\n                    scatter_shape, dtype=tensor.dtype, layout=tensor.layout, device=device)\n                local_metadata = shard_meta\n\n        # each rank should have local_tensor and local_metadata initialized if we build\n        # the metadata list in a correct way.\n        assert local_tensor is not None\n        assert local_metadata is not None\n\n        # Scatter the shards to all ranks in the pg\n        # scatter takes the global rank as ``src``\n        src_for_scatter = src_rank\n        if process_group is not None and process_group is not distributed_c10d._get_default_group():\n            src_for_scatter = distributed_c10d.get_global_rank(process_group, src_for_scatter)\n\n        dist.scatter(\n            local_tensor,\n            scatter_list=tensors_to_scatter if current_rank == src_rank else None,\n            src=src_for_scatter,\n            group=process_group\n        )\n\n        if list(local_tensor.size()) != local_metadata.shard_sizes:\n            # detach again after receiving to ensure local shards remain a leaf node\n            local_tensor = local_tensor.resize_(local_metadata.shard_sizes).detach()\n\n        # Sync requires_grad to local_shard.\n        local_tensor.requires_grad = tensor.requires_grad\n\n        local_shards.append(Shard(tensor=local_tensor, metadata=local_metadata))\n\n        st = ShardedTensor._init_from_local_shards_and_global_metadata(\n            local_shards,\n            tensor_meta,\n            process_group=process_group)\n\n        # Manually set sharding_spec\n        st._sharding_spec = self\n\n        return st\n", "path": "torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py"}], "after_files": [{"content": "from dataclasses import dataclass\nimport torch\nimport torch.distributed._shard.sharded_tensor.metadata as sharded_tensor_meta\nfrom torch.distributed._shard.metadata import ShardMetadata\nfrom torch.distributed._shard.sharded_tensor.shard import Shard\nfrom torch.distributed._shard.sharded_tensor.utils import (\n    _parse_and_validate_remote_device\n)\nfrom torch.distributed._shard._utils import narrow_tensor\nimport torch.distributed as dist\nimport torch.distributed.distributed_c10d as distributed_c10d\nfrom typing import List, Union, TYPE_CHECKING\nfrom ._internals import (\n    get_chunked_dim_size,\n    get_split_size,\n)\n\nfrom .api import ShardingSpec\n\nif TYPE_CHECKING:\n    # Only include ShardedTensor when do type checking, exclude it\n    # from run-time to resolve circular dependency.\n    from torch.distributed._shard.sharded_tensor import ShardedTensor\n\n@dataclass\nclass ChunkShardingSpec(ShardingSpec):\n    \"\"\"\n    This is a type of PlacementSpec that defines the placement as being sharded\n    across multiple devices. In particular, it represents sharding a Tensor\n    along a single dimension into equal chunks (similar to :meth:`torch.chunk`).\n\n    The semantics of how a tensor is partitioned is inline with\n    :meth:`torch.chunk`, where ``dim`` in torch.chunk corresponds to the\n    specified ``dim`` and ``chunks`` in torch.chunk is the number of elements\n    in the placement specified.\n\n    Args:\n        dim (int or str):\n            The dimension to shard on, could be an integer representing the\n            dimension or a string in case of named tensors where dimensions are\n            named. Note that named tensor support is not added yet.\n        placement(List[Union[_remote_device, str]]):\n            Specifies the placement of each shard of the Tensor. The size of\n            the list represents the number of shards to be created. This could\n            be a list of\n            :class:`torch.distributed._remote_device`'s. This list\n            could also contain a string which represents remote\n            device as accepted by\n            :class:`torch.distributed._remote_device`\n    \"\"\"\n\n    ShardingDim = Union[int, str]\n\n    dim: ShardingDim\n    placements: List[Union[torch.distributed._remote_device, str]]\n\n    def __post_init__(self):\n        self._verify_dim(self.dim)\n        for i, remote_device in enumerate(self.placements):\n            if not isinstance(remote_device, torch.distributed._remote_device):\n                self.placements[i] = torch.distributed._remote_device(remote_device)\n\n    @staticmethod\n    def _verify_dim(dim):\n        # Validate the sharding spec.\n        # TODO: support named dimension\n        if isinstance(dim, str):\n            raise NotImplementedError(\n                \"ChunkShardingSpec does not support named dimension yet!\"\n            )\n\n        if not isinstance(dim, int):\n            raise ValueError(\n                f\"Sharding dim needs to be an integer, found: {dim}\"\n            )\n\n    def build_metadata(self,\n                       tensor_sizes: torch.Size,\n                       tensor_properties: sharded_tensor_meta.TensorProperties,\n                       ) -> sharded_tensor_meta.ShardedTensorMetadata:\n        tensor_num_dim = len(tensor_sizes)\n\n        self._verify_dim(self.dim)\n        if self.dim >= tensor_num_dim or self.dim < -tensor_num_dim:  # type: ignore[operator]\n            raise ValueError(f\"Invalid sharding dim: {self.dim}\")\n\n        shards_metadata = []\n        sharding_dim_size = tensor_sizes[self.dim]  # type: ignore[index]\n        chunks = len(self.placements)\n        split_size = get_split_size(sharding_dim_size, chunks)\n        for idx, placement in enumerate(self.placements):\n            # generate ShardMetadata for each placement device\n            chunked_dim_size = get_chunked_dim_size(sharding_dim_size, split_size, idx)\n            shard_size = list(tensor_sizes)\n            current_offsets = [0] * tensor_num_dim\n            current_offsets[self.dim] = split_size * idx  # type: ignore[index]\n            shard_size[self.dim] = chunked_dim_size  # type: ignore[index]\n\n            shard_metadata = ShardMetadata(\n                shard_offsets=current_offsets,\n                shard_sizes=shard_size,\n                placement=placement,\n            )\n            shards_metadata.append(shard_metadata)\n\n        return sharded_tensor_meta.ShardedTensorMetadata(\n            shards_metadata,\n            tensor_sizes,\n            tensor_properties\n        )\n\n\n    def shard(self, tensor: torch.Tensor, src_rank: int = 0, process_group=None) -> \"ShardedTensor\":\n        \"\"\"\n        Args:\n            src_rank: group rank relative to ``process_group``\n\n            N.B. If ``process_group`` is None, ``src_rank`` is a global rank.\n        \"\"\"\n        # relative imports to avoid circular dependency\n        from torch.distributed._shard.sharded_tensor import (\n            ShardedTensor\n        )\n        tensor_properties = sharded_tensor_meta.TensorProperties(\n            dtype=tensor.dtype,\n            layout=tensor.layout,\n            requires_grad=tensor.requires_grad,\n            memory_format=torch.contiguous_format,\n            pin_memory=tensor.is_pinned()\n        )\n        current_rank = dist.get_rank(process_group)\n        tensor_meta = self.build_metadata(tensor.size(), tensor_properties)\n        local_shards = []\n        local_tensor = None\n        local_metadata = None\n        tensors_to_scatter = [None] * dist.get_world_size(process_group)\n\n        sharding_dim_size = tensor.size()[self.dim]  # type: ignore[index]\n        chunks = len(self.placements)\n        split_size = get_split_size(sharding_dim_size, chunks)\n        scatter_shape = list(tensor.size())\n        scatter_shape[self.dim] = split_size  # type: ignore[index]\n\n        for shard_meta in tensor_meta.shards_metadata:\n            rank, device = _parse_and_validate_remote_device(process_group, shard_meta.placement)\n            if current_rank == src_rank:\n                # Reshape to get shard for this rank and we don't want autograd\n                # recording here for the narrow op and 'local_shard' should be a\n                # leaf variable in the autograd graph.\n                narrowed_tensor = narrow_tensor(tensor, shard_meta)\n                if shard_meta.shard_sizes[self.dim] < split_size:  # type: ignore[index]\n                    # for the last shard that might be smaller to other shards\n                    # resize the narrowed tensor to the same size and use it for\n                    # the scatter collective as dist.scatter requires same size\n                    # inputs on every rank\n                    tensor_to_scatter = narrowed_tensor.detach().clone().resize_(scatter_shape)\n                else:\n                    tensor_to_scatter = narrowed_tensor.detach().clone().contiguous()\n\n                tensors_to_scatter[rank] = tensor_to_scatter\n\n            if current_rank == rank:\n                local_tensor = torch.empty(\n                    scatter_shape, dtype=tensor.dtype, layout=tensor.layout, device=device)\n                local_metadata = shard_meta\n\n        # each rank should have local_tensor and local_metadata initialized if we build\n        # the metadata list in a correct way.\n        assert local_tensor is not None\n        assert local_metadata is not None\n\n        # Scatter the shards to all ranks in the pg\n        # scatter takes the global rank as ``src``\n        src_for_scatter = src_rank\n        if process_group is not None and process_group is not distributed_c10d._get_default_group():\n            src_for_scatter = distributed_c10d.get_global_rank(process_group, src_for_scatter)\n\n        dist.scatter(\n            local_tensor,\n            scatter_list=tensors_to_scatter if current_rank == src_rank else None,\n            src=src_for_scatter,\n            group=process_group\n        )\n\n        if list(local_tensor.size()) != local_metadata.shard_sizes:\n            # detach again after receiving to ensure local shards remain a leaf node\n            local_tensor = local_tensor.resize_(local_metadata.shard_sizes).detach()\n\n        # Sync requires_grad to local_shard.\n        local_tensor.requires_grad = tensor.requires_grad\n\n        local_shards.append(Shard(tensor=local_tensor, metadata=local_metadata))\n\n        st = ShardedTensor._init_from_local_shards_and_global_metadata(\n            local_shards,\n            tensor_meta,\n            process_group=process_group)\n\n        # Manually set sharding_spec\n        st._sharding_spec = self\n\n        return st\n", "path": "torch/distributed/_shard/sharding_spec/chunk_sharding_spec.py"}]}