Skip to content

Commit af4f5c0

Browse files
jeffkbkimfacebook-github-bot
authored andcommitted
[torchrec][LocalShardsWrapper] Implement tensor padding for local shards wrapper (pytorch#163183)
Summary: X-link: pytorch/torchrec#3382 This diff implements the constant padding functionality (aten.constant_pad_nd.default) for `LocalShardsWrapper`. The method applies constant padding to the local shards based on the provided padding specification. Depending on the sharding type (RW, CW), the padding on [left, right, top, bottom] directions will be either applied to the first/last shard, or all local shards. New unit tests cover: - 1D (RW) top/bottom paddings - 2D (CW) left, right, top, bottom paddings - empty shards, number of dimensions > 2 Test Plan: ``` buck2 test fbcode//mode/opt fbcode//torchrec/distributed/tests:test_shards_wrapper <...> Buck UI: https://www.internalfb.com/buck2/9fff7732-346a-43eb-b1a0-f0e43e2e8815 Test UI: https://www.internalfb.com/intern/testinfra/testrun/18014398620870153 Network: Up: 110KiB Down: 95KiB (reSessionID-c0cdcb56-f82e-4f42-9fb8-54d8a3fb74eb) Analyzing targets. Remaining 0/191 Executing actions. Remaining 0/12849 7.6s exec time total Command: test. Finished 5 local Time elapsed: 1:40.1s Test execution completed but tests were skipped Tests finished: Pass 14. Fail 0. Fatal 0. Skip 3. Build failure 0 ``` Rollback Plan: Differential Revision: D82663766
1 parent 4660e38 commit af4f5c0

File tree

1 file changed

+180
-0
lines changed

1 file changed

+180
-0
lines changed

torch/distributed/tensor/_shards_wrapper.py

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@ def __torch_dispatch__(cls, func, types, args=(), kwargs=None): # type: ignore[
109109
aten.detach.default: cls.handle_detach,
110110
aten.clone.default: cls.handle_clone,
111111
aten.new_empty.default: cls.handle_new_empty,
112+
aten.constant_pad_nd.default: cls.handle_constant_pad_nd,
112113
}
113114

114115
if func in dispatcher:
@@ -223,6 +224,185 @@ def handle_new_empty(args, kwargs) -> "LocalShardsWrapper":
223224
self_ls.local_offsets(),
224225
)
225226

227+
@staticmethod
228+
def handle_constant_pad_nd(args, kwargs) -> "LocalShardsWrapper":
229+
"""
230+
Apply constant padding to LocalShardsWrapper.
231+
232+
The padding is based off of the following ideas:
233+
- The resulting wrapper represents the padded version of the logical tensor.
234+
- Each shard is padded based on the sharding type + dimension that is padded.
235+
- For instance, CW shards padded on the left most col will have only padding on the first CW shard.
236+
- Padding the top row will apply to all CW shards.
237+
"""
238+
self_lsw = args[0]
239+
pad_spec = args[1]
240+
pad_value = args[2] if len(args) > 2 else 0.0
241+
242+
if len(self_lsw.local_shards()) == 0:
243+
raise NotImplementedError("Padding empty LocalShardsWrapper is not supported.")
244+
245+
local_shards = self_lsw.local_shards()
246+
247+
if len(local_shards) == 1:
248+
padded_shard = torch.nn.functional.pad(
249+
local_shards[0], pad_spec, mode="constant", value=pad_value
250+
)
251+
return LocalShardsWrapper([padded_shard], self_lsw.local_offsets())
252+
253+
padded_shards = list(local_shards)
254+
255+
if local_shards[0].ndim == 2:
256+
# 2D Column-wise sharding: [pad_left, pad_right, pad_top, pad_bottom]
257+
pad_left, pad_right, pad_top, pad_bottom = pad_spec[0], pad_spec[1], pad_spec[2], pad_spec[3]
258+
259+
if pad_top > 0:
260+
padded_shards = [
261+
torch.nn.functional.pad(
262+
shard, [0, 0, pad_top, 0], mode="constant", value=pad_value
263+
)
264+
for shard in padded_shards
265+
]
266+
if pad_bottom > 0:
267+
padded_shards = [
268+
torch.nn.functional.pad(
269+
shard, [0, 0, 0, pad_bottom], mode="constant", value=pad_value
270+
)
271+
for shard in padded_shards
272+
]
273+
if pad_left > 0:
274+
padded_shards[0] = torch.nn.functional.pad(
275+
padded_shards[0],
276+
[pad_left, 0, 0, 0],
277+
mode="constant",
278+
value=pad_value
279+
)
280+
if pad_right > 0:
281+
padded_shards[-1] = torch.nn.functional.pad(
282+
padded_shards[-1],
283+
[0, pad_right, 0, 0],
284+
mode="constant",
285+
value=pad_value
286+
)
287+
elif local_shards[0].ndim == 1:
288+
# 1D Row-wise sharding: [pad_top, pad_bottom]
289+
pad_top, pad_bottom = pad_spec[0], pad_spec[1]
290+
291+
if pad_top > 0:
292+
padded_shards[0] = torch.nn.functional.pad(
293+
padded_shards[0], [pad_top, 0], mode="constant", value=pad_value
294+
)
295+
if pad_bottom > 0:
296+
padded_shards[-1] = torch.nn.functional.pad(
297+
padded_shards[-1], [0, pad_bottom], mode="constant", value=pad_value
298+
)
299+
else:
300+
raise NotImplementedError(
301+
f"Padding for {local_shards[0].ndim}D tensors is not supported. "
302+
f"Only 1D and 2D tensors are currently supported."
303+
)
304+
305+
# Update offsets and storage metadata
306+
original_storage = self_lsw.storage_metadata()
307+
updated_offsets, updated_storage = LocalShardsWrapper._compute_updated_metadata(
308+
original_storage,
309+
self_lsw.local_offsets(),
310+
pad_spec, local_shards[0].ndim,
311+
padded_shards
312+
)
313+
314+
result = LocalShardsWrapper(padded_shards, updated_offsets)
315+
result._storage_meta = updated_storage
316+
return result
317+
318+
@staticmethod
319+
def _compute_updated_metadata(
320+
original_storage: TensorStorageMetadata,
321+
original_offsets: list[torch.Size],
322+
pad_spec: list[int],
323+
ndim: int,
324+
padded_shards: list[torch.Tensor],
325+
) -> tuple[list[torch.Size], TensorStorageMetadata]:
326+
"""
327+
Compute updated offsets and storage metadata after padding is applied.
328+
329+
Args:
330+
original_storage: Original storage metadata
331+
original_offsets: Original shard offsets
332+
pad_spec: Padding specification
333+
ndim: Number of dimensions (1=RW or 2=CW)
334+
padded_shards: Padded shard tensors
335+
336+
Returns:
337+
Tuple of (updated_offsets, updated_storage_metadata)
338+
"""
339+
if ndim == 1: # 1D RW
340+
pad_top, pad_bottom = pad_spec[0], pad_spec[1]
341+
342+
updated_offsets = []
343+
for i, offset in enumerate(original_offsets):
344+
if i == 0:
345+
# First shard: offset stays the same (absorbs top padding)
346+
updated_offsets.append(offset)
347+
else:
348+
# Subsequent shards: shift by top padding amount
349+
new_offset = (offset[0] + pad_top,)
350+
updated_offsets.append(torch.Size(new_offset))
351+
352+
new_global_size = torch.Size(
353+
[original_storage.size[0] + pad_top + pad_bottom]
354+
)
355+
356+
elif ndim == 2: # 2D CW
357+
pad_left, pad_right, pad_top, pad_bottom = (
358+
pad_spec[0],
359+
pad_spec[1],
360+
pad_spec[2],
361+
pad_spec[3]
362+
)
363+
364+
updated_offsets = []
365+
for i, offset in enumerate(original_offsets):
366+
row_offset = offset[0]
367+
col_offset = offset[1]
368+
369+
# Top/bottom padding doesn't affect offsets
370+
# Left padding affects column offsets
371+
if i == 0:
372+
# First shard: column offset stays the same (absorbs left padding)
373+
new_offset = (row_offset, col_offset)
374+
else:
375+
# Subsequent shards: shift column offset by left padding amount
376+
new_offset = (row_offset, col_offset + pad_left)
377+
378+
updated_offsets.append(torch.Size(new_offset))
379+
380+
new_global_size = torch.Size(
381+
[
382+
original_storage.size[0] + pad_top + pad_bottom,
383+
original_storage.size[1] + pad_left + pad_right
384+
]
385+
)
386+
387+
else:
388+
raise NotImplementedError(f"Metadata computation for {ndim}D not supported")
389+
390+
updated_chunks = [
391+
ChunkStorageMetadata(
392+
offsets=offset,
393+
sizes=shard.size(),
394+
)
395+
for offset, shard in zip(updated_offsets, padded_shards)
396+
]
397+
398+
updated_storage = TensorStorageMetadata(
399+
properties=original_storage.properties,
400+
size=new_global_size,
401+
chunks=updated_chunks,
402+
)
403+
404+
return updated_offsets, updated_storage
405+
226406
@property
227407
def device(self) -> torch._C.device: # type: ignore[override]
228408
return (

0 commit comments

Comments
 (0)