Skip to content

Commit fd7d03b

Browse files
Sunt-ingksivaman
andauthored
Bound symmetric all-reduce scratch storage across input sizes (#3709)
* Bound symmetric all-reduce scratch storage across input sizes Signed-off-by: Ting Sun <suntcrick@gmail.com> * Skip symmetric-memory regression on unsupported devices Signed-off-by: Ting Sun <suntcrick@gmail.com> * 2x -> 25% memory overhead Signed-off-by: Kirthi Shankar Sivamani <ksivamani@nvidia.com> --------- Signed-off-by: Ting Sun <suntcrick@gmail.com> Signed-off-by: Kirthi Shankar Sivamani <ksivamani@nvidia.com> Co-authored-by: Kirthi Shankar Sivamani <ksivamani@nvidia.com>
1 parent f0b1096 commit fd7d03b

3 files changed

Lines changed: 80 additions & 2 deletions

File tree

‎qa/L1_pytorch_distributed_unittest/test.sh‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ fi
4545

4646
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_sanity.xml $TE_PATH/tests/pytorch/distributed/test_sanity.py || test_fail "test_sanity.py"
4747
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_parallel_cross_entropy.xml $TE_PATH/tests/pytorch/distributed/test_parallel_cross_entropy.py || test_fail "test_parallel_cross_entropy.py"
48+
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_symmetric_memory.xml $TE_PATH/tests/pytorch/distributed/test_symmetric_memory.py || test_fail "test_symmetric_memory.py"
4849
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_numerics.xml $TE_PATH/tests/pytorch/distributed/test_numerics.py || test_fail "test_numerics.py"
4950
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_numerics_exact.xml $TE_PATH/tests/pytorch/distributed/test_numerics_exact.py || test_fail "test_numerics_exact.py"
5051
python3 -m pytest -v -s --junitxml=$XML_LOG_DIR/pytest_test_fusible_ops.xml $TE_PATH/tests/pytorch/distributed/test_fusible_ops.py || test_fail "test_fusible_ops.py"
Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
# Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2+
#
3+
# See LICENSE for license information.
4+
5+
import os
6+
import tempfile
7+
8+
import pytest
9+
import torch
10+
import torch.distributed as dist
11+
import torch.multiprocessing as mp
12+
13+
from transformer_engine.pytorch import distributed as te_dist
14+
15+
16+
def _check_cache(rank, init_file):
17+
torch.cuda.set_device(rank)
18+
dist.init_process_group("nccl", init_method=f"file://{init_file}", rank=rank, world_size=2)
19+
try:
20+
group = dist.group.WORLD
21+
stream = torch.cuda.Stream()
22+
stream.wait_stream(torch.cuda.current_stream())
23+
with torch.cuda.stream(stream):
24+
# Ordinary all-reduces with varying row counts must not retain one buffer per size.
25+
for rows in range(1, 33):
26+
tensor = torch.full((rows, 256), rank + 1, device="cuda", dtype=torch.bfloat16)
27+
te_dist.symmetric_all_reduce(tensor, group, all_reduce_type="two_shot")
28+
assert torch.equal(tensor, torch.full_like(tensor, 3))
29+
buffers = list(te_dist.symmetric_mem_cache.values())
30+
assert sum(t.numel() for t in buffers) < 8 * tensor.numel()
31+
32+
# Capture two sizes sharing a capacity, then grow before replaying the old graphs.
33+
graphs = []
34+
for rows in (23, 24):
35+
inp = torch.empty(rows, 256, device="cuda", dtype=torch.bfloat16)
36+
out = torch.empty_like(inp)
37+
for _ in range(3):
38+
inp.fill_(rank + 1)
39+
te_dist.symmetric_all_reduce(inp, group, all_reduce_type="two_shot")
40+
torch.cuda.synchronize()
41+
graph = torch.cuda.CUDAGraph()
42+
with torch.cuda.graph(graph, stream=stream):
43+
inp.fill_(rank + 1)
44+
te_dist.symmetric_all_reduce(inp, group, all_reduce_type="two_shot")
45+
out.copy_(inp)
46+
graphs.append((graph, inp, out))
47+
large = torch.full((128, 256), rank + 1, device="cuda", dtype=torch.bfloat16)
48+
te_dist.symmetric_all_reduce(large, group, all_reduce_type="two_shot")
49+
for _ in range(3):
50+
for graph, _, out in reversed(graphs):
51+
graph.replay()
52+
torch.cuda.synchronize()
53+
assert torch.equal(out, torch.full_like(out, 3))
54+
torch.cuda.current_stream().wait_stream(stream)
55+
torch.cuda.synchronize()
56+
finally:
57+
dist.destroy_process_group()
58+
59+
60+
@pytest.mark.skipif(torch.cuda.device_count() < 2, reason="Requires two peer-accessible GPUs")
61+
def test_symmetric_cache_capacity_and_graph_replay():
62+
"""Exercise cache growth and graph replay in fresh distributed processes."""
63+
if not te_dist.HAS_TORCH_SYMMETRIC:
64+
pytest.skip("Requires PyTorch symmetric memory support")
65+
if not (torch.cuda.can_device_access_peer(0, 1) and torch.cuda.can_device_access_peer(1, 0)):
66+
pytest.skip("Requires peer access between GPUs 0 and 1")
67+
with tempfile.TemporaryDirectory() as directory:
68+
mp.spawn(_check_cache, args=(os.path.join(directory, "rendezvous"),), nprocs=2, join=True)

‎transformer_engine/pytorch/distributed.py‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2018,6 +2018,9 @@ def symmetric_all_reduce(
20182018
created symmetric memory tensors not in the cache currently will not be able to
20192019
avoid the extra copies.
20202020
2021+
Note: Scratch buffers for uncached inputs are rounded up to a bucketed capacity, so
2022+
each one may use up to 25% more symmetric memory than the input tensor.
2023+
20212024
Parameters
20222025
----------
20232026
inp : torch.Tensor
@@ -2089,8 +2092,14 @@ def symmetric_all_reduce(
20892092
group_name,
20902093
)
20912094
else:
2092-
# Get symmetric memory tensor. Build or retrieve from cache.
2093-
msg = get_symmetric_memory_tensor(tensor_numel, tensor_dtype, tensor_device, tp_group)
2095+
# Round up to one of four capacities per power of two. This bounds retained
2096+
# scratch space by the largest request while padding each buffer by at most 25%.
2097+
# Keep older capacities alive because captured graphs still reference them.
2098+
numel = max(1, tensor_numel)
2099+
step = 1 << max(0, numel.bit_length() - 3)
2100+
capacity = -(-numel // step) * step
2101+
msg = get_symmetric_memory_tensor(capacity, tensor_dtype, tensor_device, tp_group)
2102+
msg = msg[:tensor_numel]
20942103

20952104
msg.copy_(inp.reshape(-1))
20962105

0 commit comments

Comments
 (0)