From 8b52eedfa339d9ed7522cdf113dfa3570ac6fc24 Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 12:18:38 +0800 Subject: [PATCH 1/7] test(mindtorch_v2): add hccl ddp subgroup integration coverage --- ...hccl_ddp_subgroup_enumeration_multicard.py | 92 +++++++++++++++++++ ...hccl_ddp_subgroup_single_rank_multicard.py | 88 ++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 tests/mindtorch_v2/test_hccl_ddp_subgroup_enumeration_multicard.py create mode 100644 tests/mindtorch_v2/test_hccl_ddp_subgroup_single_rank_multicard.py diff --git a/tests/mindtorch_v2/test_hccl_ddp_subgroup_enumeration_multicard.py b/tests/mindtorch_v2/test_hccl_ddp_subgroup_enumeration_multicard.py new file mode 100644 index 000000000..14e4e4f3c --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_ddp_subgroup_enumeration_multicard.py @@ -0,0 +1,92 @@ +"""HCCL DDP with enumeration-created subgroup on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.nn as nn +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +subgroup, groups = dist.new_subgroups_by_enumeration([[0, 1]], backend="hccl") +assert subgroup is not dist.GroupMember.NON_GROUP_MEMBER +assert len(groups) == 1 +assert dist.get_world_size(subgroup) == 2 + +# Deterministic init, then DDP on subgroup. +torch.manual_seed(1234) +model = nn.Linear(4, 2).to(device) +ddp = nn.parallel.DistributedDataParallel(model, process_group=subgroup) + +x = torch.ones((3, 4), device=device) +loss = ddp(x).sum() +loss.backward() + +# Verify synchronized grads across subgroup. +w_ref = model.weight.grad.clone() +b_ref = model.bias.grad.clone() +dist.broadcast(w_ref, src=0, group=subgroup) +dist.broadcast(b_ref, src=0, group=subgroup) +wdiff = (model.weight.grad - w_ref).abs().sum().to("cpu").item() +bdiff = (model.bias.grad - b_ref).abs().sum().to("cpu").item() +assert wdiff < 1e-5, f"rank={rank} subgroup weight grad mismatch {wdiff}" +assert bdiff < 1e-5, f"rank={rank} subgroup bias grad mismatch {bdiff}" + +dist.barrier() +dist.destroy_process_group() +print(f"[rank {rank}] HCCL DDP enumeration subgroup PASS") +''' + + +def test_hccl_ddp_enumeration_subgroup_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29705" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_ddp_enumeration_subgroup_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=300) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL DDP enumeration subgroup failed on ranks: {failed}") diff --git a/tests/mindtorch_v2/test_hccl_ddp_subgroup_single_rank_multicard.py b/tests/mindtorch_v2/test_hccl_ddp_subgroup_single_rank_multicard.py new file mode 100644 index 000000000..2047f0d10 --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_ddp_subgroup_single_rank_multicard.py @@ -0,0 +1,88 @@ +"""HCCL DDP with single-rank subgroup should behave as local training.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.nn as nn +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +subgroup = dist.new_group(ranks=[0], backend="hccl", group_desc="hccl_ddp_single_rank_subgroup") +if rank == 0: + assert subgroup is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(subgroup) == 1 + + model = nn.Linear(4, 2).to(device) + ddp = nn.parallel.DistributedDataParallel(model, process_group=subgroup) + + x = torch.ones((3, 4), device=device) + loss = ddp(x).sum() + loss.backward() + + assert model.weight.grad is not None + assert model.bias.grad is not None + # No cross-rank sync for single-rank subgroup; grads stay finite and local. + assert model.weight.grad.abs().sum().to("cpu").item() > 0 +else: + assert subgroup is dist.GroupMember.NON_GROUP_MEMBER + +# WORLD barrier for clean shutdown. +dist.barrier() +dist.destroy_process_group() +print(f"[rank {rank}] HCCL DDP single-rank subgroup PASS") +''' + + +def test_hccl_ddp_single_rank_subgroup_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29704" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_ddp_single_rank_subgroup_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=300) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL DDP single-rank subgroup failed on ranks: {failed}") From 1131d44b003ba504baff729ca583f7971dd7c20d Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 12:26:17 +0800 Subject: [PATCH 2/7] test(mindtorch_v2): add hccl new_group semantics coverage --- ...est_hccl_new_group_non_member_multicard.py | 75 +++++++++++++++++ ...test_hccl_new_group_semantics_multicard.py | 83 +++++++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 tests/mindtorch_v2/test_hccl_new_group_non_member_multicard.py create mode 100644 tests/mindtorch_v2/test_hccl_new_group_semantics_multicard.py diff --git a/tests/mindtorch_v2/test_hccl_new_group_non_member_multicard.py b/tests/mindtorch_v2/test_hccl_new_group_non_member_multicard.py new file mode 100644 index 000000000..c7eab104a --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_new_group_non_member_multicard.py @@ -0,0 +1,75 @@ +"""HCCL new_group non-member behavior on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +pg = dist.new_group(ranks=[0], backend="hccl", group_desc="hccl_group_rank0_only") +if rank == 0: + assert pg is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(pg) == 1 + assert dist.get_process_group_ranks(pg) == [0] +else: + assert pg is dist.GroupMember.NON_GROUP_MEMBER + +dist.barrier() +dist.destroy_process_group() +print(f"[rank {rank}] HCCL new_group non-member PASS") +''' + + +def test_hccl_new_group_non_member_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29707" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_new_group_non_member_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=300) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL new_group non-member failed on ranks: {failed}") diff --git a/tests/mindtorch_v2/test_hccl_new_group_semantics_multicard.py b/tests/mindtorch_v2/test_hccl_new_group_semantics_multicard.py new file mode 100644 index 000000000..2748e8af8 --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_new_group_semantics_multicard.py @@ -0,0 +1,83 @@ +"""HCCL new_group semantics on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +# Explicit non-default rank order. +pg = dist.new_group(ranks=[1, 0], backend="hccl", group_desc="hccl_new_group_10") +assert pg is not dist.GroupMember.NON_GROUP_MEMBER +assert dist.get_world_size(pg) == 2 +assert dist.get_group_rank(pg, 1) == 0 +assert dist.get_group_rank(pg, 0) == 1 +assert dist.get_global_rank(pg, 0) == 1 +assert dist.get_global_rank(pg, 1) == 0 + +# API returns sorted global ranks. +assert dist.get_process_group_ranks(pg) == [0, 1] + +# Collective on explicit group. +x = torch.tensor([float(rank + 1)], device=device) +dist.all_reduce(x, group=pg) +assert float(x.to("cpu").item()) == 3.0 + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL new_group semantics PASS") +''' + + +def test_hccl_new_group_semantics_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29706" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_new_group_semantics_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=300) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL new_group semantics failed on ranks: {failed}") From 13e5b885e6b192751ba43106ad87531a933121e7 Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 12:34:02 +0800 Subject: [PATCH 3/7] test(mindtorch_v2): add hccl split_group stress coverage --- ..._hccl_split_group_interleaved_multicard.py | 110 ++++++++++++++++++ ...est_hccl_split_group_then_ddp_multicard.py | 95 +++++++++++++++ 2 files changed, 205 insertions(+) create mode 100644 tests/mindtorch_v2/test_hccl_split_group_interleaved_multicard.py create mode 100644 tests/mindtorch_v2/test_hccl_split_group_then_ddp_multicard.py diff --git a/tests/mindtorch_v2/test_hccl_split_group_interleaved_multicard.py b/tests/mindtorch_v2/test_hccl_split_group_interleaved_multicard.py new file mode 100644 index 000000000..86496182c --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_split_group_interleaved_multicard.py @@ -0,0 +1,110 @@ +"""HCCL split_group interleaved calls stress on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +# Build two parent groups with different rank orders. +parent_a = dist.new_group(ranks=[1, 0], backend="hccl", group_desc="hccl_parent_a_10") +parent_b = dist.new_group(ranks=[0, 1], backend="hccl", group_desc="hccl_parent_b_01") +assert parent_a is not dist.GroupMember.NON_GROUP_MEMBER +assert parent_b is not dist.GroupMember.NON_GROUP_MEMBER + +for step in range(5): + # 1) WORLD split: both ranks included. + w = dist.split_group(dist.group.WORLD, color=step, key=0) + assert w is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(w) == 2 + + # 2) parent_a split with equal keys: tie-break by parent rank mapping. + ga = dist.split_group(parent_a, color=100 + step, key=0) + assert ga is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(ga) == 2 + assert dist.get_group_rank(ga, 1) == 0 + assert dist.get_group_rank(ga, 0) == 1 + + # 3) parent_b split with rank-specific key ordering. + gb = dist.split_group(parent_b, color=200 + step, key=(0 if rank == 0 else 1)) + assert gb is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(gb) == 2 + assert dist.get_group_rank(gb, 0) == 0 + assert dist.get_group_rank(gb, 1) == 1 + + # 4) Subset split on WORLD with alternating non-member rank. + s = dist.split_group(dist.group.WORLD, color=(7 if rank == (step % 2) else -1), key=0) + if rank == (step % 2): + assert s is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(s) == 1 + assert dist.get_process_group_ranks(s) == [rank] + else: + assert s is dist.GroupMember.NON_GROUP_MEMBER + + # collective sanity on ga/gb + xa = torch.tensor([float(rank + 1)], device=device) + xb = torch.tensor([float(rank + 1)], device=device) + dist.all_reduce(xa, group=ga) + dist.all_reduce(xb, group=gb) + assert float(xa.to("cpu").item()) == 3.0 + assert float(xb.to("cpu").item()) == 3.0 + +dist.barrier() +dist.destroy_process_group() +print(f"[rank {rank}] HCCL split_group interleaved PASS") +''' + + +def test_hccl_split_group_interleaved_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29708" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_split_group_interleaved_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=360) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL split_group interleaved failed on ranks: {failed}") diff --git a/tests/mindtorch_v2/test_hccl_split_group_then_ddp_multicard.py b/tests/mindtorch_v2/test_hccl_split_group_then_ddp_multicard.py new file mode 100644 index 000000000..bf8e122df --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_split_group_then_ddp_multicard.py @@ -0,0 +1,95 @@ +"""HCCL split_group repeated calls followed by DDP subgroup training.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.nn as nn +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +# Repeated split_group invocations to stress sequence bookkeeping and communicator setup. +for step in range(8): + g = dist.split_group(dist.group.WORLD, color=10 + step, key=0) + assert g is not dist.GroupMember.NON_GROUP_MEMBER + assert dist.get_world_size(g) == 2 + +# Use a split-created group for DDP to verify post-split training stability. +train_group = dist.split_group(dist.group.WORLD, color=999, key=0) +assert train_group is not dist.GroupMember.NON_GROUP_MEMBER + +model = nn.Linear(4, 2).to(device) +ddp = nn.parallel.DistributedDataParallel(model, process_group=train_group) + +for step in range(3): + x = torch.ones((3, 4), device=device) + loss = ddp(x).sum() + loss.backward() + + ref = model.weight.grad.clone() + dist.broadcast(ref, src=0, group=train_group) + diff = (model.weight.grad - ref).abs().sum().to("cpu").item() + assert diff < 1e-5, f"step={step} rank={rank} grad mismatch {diff}" + + with torch.no_grad(): + model.weight.grad = None + model.bias.grad = None + +dist.barrier() +dist.destroy_process_group() +print(f"[rank {rank}] HCCL split->DDP PASS") +''' + + +def test_hccl_split_group_then_ddp_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29709" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_split_group_then_ddp_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=420) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL split->DDP failed on ranks: {failed}") From 23e90df64d8c627acb1e5d572a3f5e5e73337bdf Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 12:44:31 +0800 Subject: [PATCH 4/7] test(mindtorch_v2): add hccl monitored_barrier stress coverage --- .../test_hccl_monitored_barrier_multicard.py | 148 ++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 tests/mindtorch_v2/test_hccl_monitored_barrier_multicard.py diff --git a/tests/mindtorch_v2/test_hccl_monitored_barrier_multicard.py b/tests/mindtorch_v2/test_hccl_monitored_barrier_multicard.py new file mode 100644 index 000000000..60841a908 --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_monitored_barrier_multicard.py @@ -0,0 +1,148 @@ +"""HCCL monitored_barrier behavior on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT_STRESS = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +sub = dist.new_group(ranks=[0, 1], backend="hccl", group_desc="hccl_barrier_sub") +assert sub is not dist.GroupMember.NON_GROUP_MEMBER + +for step in range(8): + dist.monitored_barrier(group=dist.group.WORLD) + dist.monitored_barrier(group=sub) + +# Collective sanity after repeated barriers. +x = torch.tensor([float(rank + 1)], device=device) +dist.all_reduce(x, group=sub) +assert float(x.to("cpu").item()) == 3.0 + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL monitored_barrier stress PASS") +''' + + +SCRIPT_REJECT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +try: + dist.monitored_barrier(wait_all_ranks=True) +except NotImplementedError as exc: + assert "wait_all_ranks" in str(exc) +else: + raise AssertionError("expected NotImplementedError for wait_all_ranks=True on hccl") + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL monitored_barrier reject PASS") +''' + + +SCRIPT_TIMEOUT = r''' +import os, sys +from datetime import timedelta +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +for _ in range(4): + dist.monitored_barrier(timeout=timedelta(seconds=5)) + +# Check collectives still work. +x = torch.tensor([float(rank + 1)], device=device) +dist.all_reduce(x) +assert float(x.to("cpu").item()) == 3.0 + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL monitored_barrier timeout PASS") +''' + + +def _run_two_rank(script_text: str, worker_name: str, master_port: int, timeout_sec: int = 300): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = str(master_port) + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = f"/tmp/{worker_name}.py" + with open(worker_file, "w") as f: + f.write(script_text) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=timeout_sec) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"{worker_name} failed on ranks: {failed}") + + +def test_hccl_monitored_barrier_stress_2card(): + _run_two_rank(SCRIPT_STRESS, "_hccl_monitored_barrier_stress_2card", master_port=29710, timeout_sec=360) + + +def test_hccl_monitored_barrier_reject_wait_all_ranks_2card(): + _run_two_rank(SCRIPT_REJECT, "_hccl_monitored_barrier_reject_2card", master_port=29711, timeout_sec=240) + + +def test_hccl_monitored_barrier_timeout_2card(): + _run_two_rank(SCRIPT_TIMEOUT, "_hccl_monitored_barrier_timeout_2card", master_port=29712, timeout_sec=240) From 28557564c214b0378e8c8b08955ef6f3beacb76b Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 12:59:44 +0800 Subject: [PATCH 5/7] test(mindtorch_v2): add hccl subgroup lifecycle stress --- .../test_hccl_group_lifecycle_multicard.py | 91 +++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 tests/mindtorch_v2/test_hccl_group_lifecycle_multicard.py diff --git a/tests/mindtorch_v2/test_hccl_group_lifecycle_multicard.py b/tests/mindtorch_v2/test_hccl_group_lifecycle_multicard.py new file mode 100644 index 000000000..615211c53 --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_group_lifecycle_multicard.py @@ -0,0 +1,91 @@ +"""HCCL subgroup lifecycle stress: create/use/destroy loops on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +for step in range(8): + # A: explicit new_group lifecycle. + g_explicit = dist.new_group(ranks=[1, 0], backend="hccl", group_desc=f"hccl_life_explicit_{step}") + assert g_explicit is not dist.GroupMember.NON_GROUP_MEMBER + x = torch.tensor([float(rank + 1)], device=device) + dist.all_reduce(x, group=g_explicit) + assert float(x.to("cpu").item()) == 3.0 + dist.destroy_process_group(g_explicit) + + # B: split_group-created lifecycle. + g_split = dist.split_group(dist.group.WORLD, color=1000 + step, key=0) + assert g_split is not dist.GroupMember.NON_GROUP_MEMBER + y = torch.tensor([float(rank + 1)], device=device) + dist.all_reduce(y, group=g_split) + assert float(y.to("cpu").item()) == 3.0 + dist.destroy_process_group(g_split) + + # WORLD barrier between iterations. + dist.barrier() + +# WORLD should remain usable after repeated subgroup destroy. +z = torch.tensor([float(rank + 1)], device=device) +dist.all_reduce(z) +assert float(z.to("cpu").item()) == 3.0 + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL group lifecycle PASS") +''' + + +def test_hccl_group_lifecycle_create_use_destroy_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29713" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_group_lifecycle_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=480) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL group lifecycle failed on ranks: {failed}") From 101402330e11ffa8c745637a520e9bea200bc044 Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 13:24:57 +0800 Subject: [PATCH 6/7] fix(mindtorch_v2): support hccl unequal all_to_all_single path --- src/mindtorch_v2/distributed/__init__.py | 20 ++-- .../distributed/_process_group.py | 106 +++++++++-------- ...l_to_all_single_async_unequal_multicard.py | 108 ++++++++++++++++++ 3 files changed, 174 insertions(+), 60 deletions(-) create mode 100644 tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py diff --git a/src/mindtorch_v2/distributed/__init__.py b/src/mindtorch_v2/distributed/__init__.py index b96dfe35d..7cde7dd11 100644 --- a/src/mindtorch_v2/distributed/__init__.py +++ b/src/mindtorch_v2/distributed/__init__.py @@ -682,29 +682,29 @@ def all_to_all_single(output, input, output_split_sizes=None, if not async_op: work.wait() dst_base = output.storage().data_ptr() - offset = 0 - for t in output_list: - nbytes = t.numel() * itemsize + dst_offset = 0 + for t, out_size in zip(output_list, output_split_sizes): + nbytes = out_size * output.dtype.itemsize ret = npu_runtime.acl.rt.memcpy( - dst_base + offset, nbytes, + dst_base + dst_offset, nbytes, t.storage().data_ptr(), nbytes, ACL_MEMCPY_D2D, ) if ret != 0: raise RuntimeError(f"D2D memcpy failed: {ret}") - offset += nbytes + dst_offset += nbytes return None def _writeback_npu_output(): dst_base = output.storage().data_ptr() - offset = 0 - for t in output_list: - nbytes = t.numel() * itemsize + dst_offset = 0 + for t, out_size in zip(output_list, output_split_sizes): + nbytes = out_size * output.dtype.itemsize ret = npu_runtime.acl.rt.memcpy( - dst_base + offset, nbytes, + dst_base + dst_offset, nbytes, t.storage().data_ptr(), nbytes, ACL_MEMCPY_D2D, ) if ret != 0: raise RuntimeError(f"D2D memcpy failed: {ret}") - offset += nbytes + dst_offset += nbytes work._on_wait = _writeback_npu_output return work else: diff --git a/src/mindtorch_v2/distributed/_process_group.py b/src/mindtorch_v2/distributed/_process_group.py index ca7150731..5cbe1370a 100644 --- a/src/mindtorch_v2/distributed/_process_group.py +++ b/src/mindtorch_v2/distributed/_process_group.py @@ -308,57 +308,63 @@ def all_to_all(self, output_tensors, input_tensors): stream = self._stream() ACL_MEMCPY_D2D = 3 - # Fast path for 2 ranks: use native HcclAlltoAll + # Fast path for 2 ranks + equal split: use native HcclAlltoAll. + # Unequal split must use per-peer P2P so each shard keeps its own size. if self._size == 2: - import mindtorch_v2 as torch - count_per_rank = input_tensors[0].numel() - dtype = input_tensors[0].dtype - itemsize = dtype.itemsize - - # Pack into contiguous buffers - total_count = count_per_rank * 2 - send_flat = torch.empty(total_count, dtype=dtype, device=input_tensors[0].device) - recv_flat = torch.empty(total_count, dtype=dtype, device=output_tensors[0].device) - - dst_base = send_flat.storage().data_ptr() - for i, t in enumerate(input_tensors): - ret = npu_runtime.acl.rt.memcpy( - dst_base + i * count_per_rank * itemsize, - count_per_rank * itemsize, - t.storage().data_ptr(), - count_per_rank * itemsize, - ACL_MEMCPY_D2D) - if ret != 0: - raise RuntimeError(f"D2D memcpy pack failed: {ret}") - - # Call HcclAlltoAll - ret = bindings.all_to_all( - ctypes.c_void_p(send_flat.storage().data_ptr()), - ctypes.c_uint64(count_per_rank), - ctypes.c_int32(dtype_to_hccl(dtype)), - ctypes.c_void_p(recv_flat.storage().data_ptr()), - ctypes.c_uint64(count_per_rank), - ctypes.c_int32(dtype_to_hccl(dtype)), - self._comm, - ctypes.c_void_p(int(stream))) - _check(ret, "HcclAlltoAll") - - # Sync and unpack - dev_id = self._device_id if self._device_id is not None else 0 - npu_runtime.get_runtime(dev_id).synchronize_stream(stream) - - src_base = recv_flat.storage().data_ptr() - for i, t in enumerate(output_tensors): - ret = npu_runtime.acl.rt.memcpy( - t.storage().data_ptr(), - count_per_rank * itemsize, - src_base + i * count_per_rank * itemsize, - count_per_rank * itemsize, - ACL_MEMCPY_D2D) - if ret != 0: - raise RuntimeError(f"D2D memcpy unpack failed: {ret}") - - return self._make_work(stream) + equal_split = ( + len({t.numel() for t in input_tensors}) == 1 and + len({t.numel() for t in output_tensors}) == 1 + ) + if equal_split: + import mindtorch_v2 as torch + count_per_rank = input_tensors[0].numel() + dtype = input_tensors[0].dtype + itemsize = dtype.itemsize + + # Pack into contiguous buffers + total_count = count_per_rank * 2 + send_flat = torch.empty(total_count, dtype=dtype, device=input_tensors[0].device) + recv_flat = torch.empty(total_count, dtype=dtype, device=output_tensors[0].device) + + dst_base = send_flat.storage().data_ptr() + for i, t in enumerate(input_tensors): + ret = npu_runtime.acl.rt.memcpy( + dst_base + i * count_per_rank * itemsize, + count_per_rank * itemsize, + t.storage().data_ptr(), + count_per_rank * itemsize, + ACL_MEMCPY_D2D) + if ret != 0: + raise RuntimeError(f"D2D memcpy pack failed: {ret}") + + # Call HcclAlltoAll + ret = bindings.all_to_all( + ctypes.c_void_p(send_flat.storage().data_ptr()), + ctypes.c_uint64(count_per_rank), + ctypes.c_int32(dtype_to_hccl(dtype)), + ctypes.c_void_p(recv_flat.storage().data_ptr()), + ctypes.c_uint64(count_per_rank), + ctypes.c_int32(dtype_to_hccl(dtype)), + self._comm, + ctypes.c_void_p(int(stream))) + _check(ret, "HcclAlltoAll") + + # Sync and unpack + dev_id = self._device_id if self._device_id is not None else 0 + npu_runtime.get_runtime(dev_id).synchronize_stream(stream) + + src_base = recv_flat.storage().data_ptr() + for i, t in enumerate(output_tensors): + ret = npu_runtime.acl.rt.memcpy( + t.storage().data_ptr(), + count_per_rank * itemsize, + src_base + i * count_per_rank * itemsize, + count_per_rank * itemsize, + ACL_MEMCPY_D2D) + if ret != 0: + raise RuntimeError(f"D2D memcpy unpack failed: {ret}") + + return self._make_work(stream) # Fallback for >2 ranks: use P2P for peer in range(self._size): diff --git a/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py b/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py new file mode 100644 index 000000000..b2514789d --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py @@ -0,0 +1,108 @@ +"""HCCL all_to_all_single async unequal-split semantics on 2 NPUs.""" + +import os +import subprocess +import sys + + +SCRIPT = r''' +import os, sys +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +assert world_size == 2 +device = torch.Device(f"npu:{rank}") +dist.init_process_group("hccl", device_id=device) + +# Use legal unequal splits: pairwise send/recv sizes must match across ranks. +# rank0: input_split=[1,3], output_split=[1,3] +# rank1: input_split=[3,1], output_split=[3,1] +if rank == 0: + input_split = [1, 3] + output_split = [1, 3] + inp = torch.tensor([0.0, 10.0, 11.0, 12.0], device=device) + expected = [0.0, 100.0, 101.0, 102.0] +else: + input_split = [3, 1] + output_split = [3, 1] + inp = torch.tensor([100.0, 101.0, 102.0, 110.0], device=device) + expected = [10.0, 11.0, 12.0, 110.0] + +out = torch.zeros(4, device=device) + +w = dist.all_to_all_single( + out, + inp, + output_split_sizes=output_split, + input_split_sizes=input_split, + async_op=True, +) +assert w is not None +w.wait() + +actual = list(out.to("cpu")._numpy_view()) +assert actual == expected, f"rank={rank} actual={actual}, expected={expected}" + +# Repeat once to verify reusable stability. +out2 = torch.zeros(4, device=device) +w2 = dist.all_to_all_single( + out2, + inp, + output_split_sizes=output_split, + input_split_sizes=input_split, + async_op=True, +) +w2.wait() +actual2 = list(out2.to("cpu")._numpy_view()) +assert actual2 == expected, f"rank={rank} repeat actual={actual2}, expected={expected}" + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL all_to_all_single async unequal PASS") +''' + + +def test_hccl_all_to_all_single_async_unequal_2card(): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = "29714" + env["WORLD_SIZE"] = "2" + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = "/tmp/_hccl_all_to_all_single_async_unequal_2card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + for r in range(2): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + for r, p in enumerate(procs): + out, _ = p.communicate(timeout=360) + txt = out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + if failed: + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError(f"HCCL all_to_all_single async unequal failed on ranks: {failed}") From 9b24d6ecc1a0ab2a0f898d93a3fd9e68c6d650fe Mon Sep 17 00:00:00 2001 From: lvyufeng Date: Sun, 8 Mar 2026 14:54:40 +0800 Subject: [PATCH 7/7] fix hccl all_to_all_single split validation and expand 2/4/8 tests --- src/mindtorch_v2/distributed/__init__.py | 69 ++++++++- ...l_to_all_single_async_unequal_multicard.py | 106 +++++++++---- ..._to_all_single_invalid_splits_multicard.py | 140 ++++++++++++++++++ 3 files changed, 283 insertions(+), 32 deletions(-) create mode 100644 tests/mindtorch_v2/test_hccl_all_to_all_single_invalid_splits_multicard.py diff --git a/src/mindtorch_v2/distributed/__init__.py b/src/mindtorch_v2/distributed/__init__.py index 7cde7dd11..ed8411055 100644 --- a/src/mindtorch_v2/distributed/__init__.py +++ b/src/mindtorch_v2/distributed/__init__.py @@ -30,6 +30,7 @@ _pg_group_ranks = {} # ProcessGroup -> {global_rank: group_rank} _group_count = 0 _split_group_seq = {} +_all_to_all_single_seq = {} default_pg_timeout = timedelta(minutes=30) @@ -130,7 +131,7 @@ def _raise_with_context(exc, *, stage, backend, rank, world_size, device_id=None def init_process_group(backend=None, init_method=None, timeout=None, world_size=-1, rank=-1, store=None, group_name="", pg_options=None, device_id=None): - global _default_pg, _group_count, _split_group_seq + global _default_pg, _group_count, _split_group_seq, _all_to_all_single_seq if _default_pg is not None: raise RuntimeError( @@ -226,10 +227,11 @@ def init_process_group(backend=None, init_method=None, timeout=None, _pg_group_ranks[pg] = {i: i for i in range(world_size)} _group_count += 1 _split_group_seq = {} + _all_to_all_single_seq = {} def destroy_process_group(group=None): - global _default_pg, _group_count, _split_group_seq + global _default_pg, _group_count, _split_group_seq, _all_to_all_single_seq if group is None or group is _default_pg: # Destroy all groups @@ -242,11 +244,13 @@ def destroy_process_group(group=None): GroupMember.WORLD = None _group_count = 0 _split_group_seq = {} + _all_to_all_single_seq = {} else: group.destroy() _pg_map.pop(group, None) _pg_names.pop(group, None) _pg_group_ranks.pop(group, None) + _all_to_all_single_seq.pop(group, None) # --------------------------------------------------------------------------- @@ -624,6 +628,62 @@ def all_to_all(output_tensor_list, input_tensor_list, group=None, return work +def _validate_all_to_all_single_splits(pg, input_split_sizes, output_split_sizes): + if len(input_split_sizes) != pg.size(): + raise ValueError( + f"input_split_sizes length {len(input_split_sizes)} must equal world_size {pg.size()}" + ) + if len(output_split_sizes) != pg.size(): + raise ValueError( + f"output_split_sizes length {len(output_split_sizes)} must equal world_size {pg.size()}" + ) + if any(int(s) < 0 for s in input_split_sizes + output_split_sizes): + raise ValueError("all_to_all_single split sizes must be non-negative") + + +def _validate_hccl_all_to_all_single_pairwise(pg, input_split_sizes, output_split_sizes): + if pg not in _pg_map: + raise ValueError("The given group is not registered") + + _, store = _pg_map[pg] + rank = pg.rank() + world_size = pg.size() + + # Use a per-process-group call sequence so every rank writes/reads + # the same store keys for the same collective invocation. + call_id = _all_to_all_single_seq.get(pg, 0) + _all_to_all_single_seq[pg] = call_id + 1 + key_prefix = f"all_to_all_single_splits/{call_id}" + + local_in = ",".join(str(int(x)) for x in input_split_sizes) + local_out = ",".join(str(int(x)) for x in output_split_sizes) + store.set(f"{key_prefix}/in/{rank}", local_in.encode("utf-8")) + store.set(f"{key_prefix}/out/{rank}", local_out.encode("utf-8")) + store.wait([f"{key_prefix}/in/{r}" for r in range(world_size)]) + store.wait([f"{key_prefix}/out/{r}" for r in range(world_size)]) + + all_in = [] + all_out = [] + for peer in range(world_size): + raw_in = store.get(f"{key_prefix}/in/{peer}") + raw_out = store.get(f"{key_prefix}/out/{peer}") + peer_in = [int(x) for x in (raw_in.decode("utf-8") if isinstance(raw_in, bytes) else str(raw_in)).split(",") if x != ""] + peer_out = [int(x) for x in (raw_out.decode("utf-8") if isinstance(raw_out, bytes) else str(raw_out)).split(",") if x != ""] + if len(peer_in) != world_size or len(peer_out) != world_size: + raise ValueError("all_to_all_single split sizes have inconsistent world_size") + all_in.append(peer_in) + all_out.append(peer_out) + + # Matrix consistency: for every (src, dst) pair, + # src.input_split[dst] must equal dst.output_split[src]. + for src in range(world_size): + for dst in range(world_size): + if int(all_in[src][dst]) != int(all_out[dst][src]): + raise ValueError( + "all_to_all_single split mismatch: input_split_sizes[dst] must match peer output_split_sizes[src]" + ) + + def all_to_all_single(output, input, output_split_sizes=None, input_split_sizes=None, group=None, async_op=False): import mindtorch_v2 as torch @@ -639,6 +699,11 @@ def all_to_all_single(output, input, output_split_sizes=None, chunk_size = output.numel() // world_size output_split_sizes = [chunk_size] * world_size + _validate_all_to_all_single_splits(pg, input_split_sizes, output_split_sizes) + + if isinstance(pg, ProcessGroupHCCL): + _validate_hccl_all_to_all_single_pairwise(pg, input_split_sizes, output_split_sizes) + equal_split = (len(set(input_split_sizes)) == 1 and len(set(output_split_sizes)) == 1) diff --git a/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py b/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py index b2514789d..466797cbe 100644 --- a/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py +++ b/tests/mindtorch_v2/test_hccl_all_to_all_single_async_unequal_multicard.py @@ -1,12 +1,15 @@ -"""HCCL all_to_all_single async unequal-split semantics on 2 NPUs.""" +"""HCCL all_to_all_single async unequal-split semantics on 2/4/8 NPUs.""" import os import subprocess import sys +import time + +import pytest SCRIPT = r''' -import os, sys +import os, sys, time src_dir = os.environ.get("MINDTORCH_V2_SRC") if src_dir: sys.path.insert(0, src_dir) @@ -17,25 +20,23 @@ rank = int(os.environ["RANK"]) world_size = int(os.environ["WORLD_SIZE"]) -assert world_size == 2 device = torch.Device(f"npu:{rank}") +# Reduce HCCL init burst on large-card jobs. +time.sleep(0.05 * rank) dist.init_process_group("hccl", device_id=device) -# Use legal unequal splits: pairwise send/recv sizes must match across ranks. -# rank0: input_split=[1,3], output_split=[1,3] -# rank1: input_split=[3,1], output_split=[3,1] -if rank == 0: - input_split = [1, 3] - output_split = [1, 3] - inp = torch.tensor([0.0, 10.0, 11.0, 12.0], device=device) - expected = [0.0, 100.0, 101.0, 102.0] -else: - input_split = [3, 1] - output_split = [3, 1] - inp = torch.tensor([100.0, 101.0, 102.0, 110.0], device=device) - expected = [10.0, 11.0, 12.0, 110.0] - -out = torch.zeros(4, device=device) +# Pairwise-consistent unequal split profile for all ranks: +# send 1 item to self, 2 items to every other peer. +input_split = [1 if i == rank else 2 for i in range(world_size)] +output_split = [1 if j == rank else 2 for j in range(world_size)] + +vals = [] +for dst in range(world_size): + cnt = input_split[dst] + for k in range(cnt): + vals.append(float(rank * 1000 + dst * 10 + k)) +inp = torch.tensor(vals, device=device) +out = torch.zeros(sum(output_split), device=device) w = dist.all_to_all_single( out, @@ -48,10 +49,15 @@ w.wait() actual = list(out.to("cpu")._numpy_view()) +expected = [] +for src in range(world_size): + cnt = 1 if src == rank else 2 + for k in range(cnt): + expected.append(float(src * 1000 + rank * 10 + k)) assert actual == expected, f"rank={rank} actual={actual}, expected={expected}" -# Repeat once to verify reusable stability. -out2 = torch.zeros(4, device=device) +# Repeat once for stability. +out2 = torch.zeros(sum(output_split), device=device) w2 = dist.all_to_all_single( out2, inp, @@ -64,28 +70,29 @@ assert actual2 == expected, f"rank={rank} repeat actual={actual2}, expected={expected}" dist.destroy_process_group() -print(f"[rank {rank}] HCCL all_to_all_single async unequal PASS") +print(f"[rank {rank}] HCCL all_to_all_single async unequal {world_size}card PASS") ''' -def test_hccl_all_to_all_single_async_unequal_2card(): +def _run_once(world_size, master_port): env = os.environ.copy() env["MASTER_ADDR"] = "127.0.0.1" - env["MASTER_PORT"] = "29714" - env["WORLD_SIZE"] = "2" + env["MASTER_PORT"] = str(master_port) + env["WORLD_SIZE"] = str(world_size) src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) env["MINDTORCH_V2_SRC"] = src_dir env["PYTHONPATH"] = src_dir + \ (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") - worker_file = "/tmp/_hccl_all_to_all_single_async_unequal_2card.py" + worker_file = f"/tmp/_hccl_all_to_all_single_async_unequal_{world_size}card.py" with open(worker_file, "w") as f: f.write(SCRIPT) failed = [] outputs = [] procs = [] - for r in range(2): + + for r in range(world_size): p = subprocess.Popen( [sys.executable, worker_file], env={**env, "RANK": str(r)}, @@ -94,15 +101,54 @@ def test_hccl_all_to_all_single_async_unequal_2card(): ) procs.append(p) + timeout = 420 if world_size <= 4 else 900 for r, p in enumerate(procs): - out, _ = p.communicate(timeout=360) - txt = out.decode("utf-8", errors="replace") + try: + out, _ = p.communicate(timeout=timeout) + txt = out.decode("utf-8", errors="replace") + except subprocess.TimeoutExpired: + p.kill() + out, _ = p.communicate() + txt = "TIMEOUT\n" + out.decode("utf-8", errors="replace") outputs.append(txt) if p.returncode != 0: failed.append(r) - if failed: + return failed, outputs + + +def _run_case(world_size, master_port): + retries = 3 + for attempt in range(1, retries + 1): + failed, outputs = _run_once(world_size, master_port) + if not failed: + return + + joined = "\n".join(outputs) + transient = "resource unavailable" in joined + if transient and attempt < retries: + print( + f"HCCL transient init failure on {world_size} cards, " + f"retry {attempt}/{retries}" + ) + time.sleep(5) + continue + for r, txt in enumerate(outputs): print(f"=== RANK {r} ===") print(txt) - raise AssertionError(f"HCCL all_to_all_single async unequal failed on ranks: {failed}") + raise AssertionError( + f"HCCL all_to_all_single async unequal {world_size}card failed on ranks: {failed}" + ) + + +@pytest.mark.parametrize( + "world_size,master_port", + [ + (2, 29714), + (4, 29724), + (8, 29734), + ], +) +def test_hccl_all_to_all_single_async_unequal_multicard(world_size, master_port): + _run_case(world_size, master_port) diff --git a/tests/mindtorch_v2/test_hccl_all_to_all_single_invalid_splits_multicard.py b/tests/mindtorch_v2/test_hccl_all_to_all_single_invalid_splits_multicard.py new file mode 100644 index 000000000..c061a6cab --- /dev/null +++ b/tests/mindtorch_v2/test_hccl_all_to_all_single_invalid_splits_multicard.py @@ -0,0 +1,140 @@ +"""HCCL all_to_all_single invalid split-size validation on 2/4/8 NPUs.""" + +import os +import subprocess +import sys +import time + +import pytest + + +SCRIPT = r''' +import os, sys, time +src_dir = os.environ.get("MINDTORCH_V2_SRC") +if src_dir: + sys.path.insert(0, src_dir) + +import mindtorch_v2 as torch +import mindtorch_v2.distributed as dist + +rank = int(os.environ["RANK"]) +world_size = int(os.environ["WORLD_SIZE"]) + +device = torch.Device(f"npu:{rank}") +# Reduce HCCL init burst on large-card jobs. +time.sleep(0.05 * rank) +dist.init_process_group("hccl", device_id=device) + +# Baseline profile: send 1 item to self, 2 items to other peers. +input_split = [1 if i == rank else 2 for i in range(world_size)] +output_split = [1 if j == rank else 2 for j in range(world_size)] + +# Break pairwise compatibility only on rank 0: +# rank0 -> rank1 send count becomes 3, but rank1's output from rank0 remains 2. +if rank == 0: + input_split[1] = 3 + +vals = [] +for dst in range(world_size): + cnt = input_split[dst] + for k in range(cnt): + vals.append(float(rank * 1000 + dst * 10 + k)) +inp = torch.tensor(vals, device=device) +out = torch.zeros(sum(output_split), device=device) + +try: + dist.all_to_all_single( + out, + inp, + output_split_sizes=output_split, + input_split_sizes=input_split, + async_op=True, + ) +except ValueError as exc: + assert "split mismatch" in str(exc) +else: + raise AssertionError("expected ValueError for invalid all_to_all_single split pairing") + +dist.destroy_process_group() +print(f"[rank {rank}] HCCL invalid split validation {world_size}card PASS") +''' + + +def _run_once(world_size, master_port): + env = os.environ.copy() + env["MASTER_ADDR"] = "127.0.0.1" + env["MASTER_PORT"] = str(master_port) + env["WORLD_SIZE"] = str(world_size) + src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "src")) + env["MINDTORCH_V2_SRC"] = src_dir + env["PYTHONPATH"] = src_dir + \ + (":" + env["PYTHONPATH"] if "PYTHONPATH" in env else "") + + worker_file = f"/tmp/_hccl_all_to_all_single_invalid_split_{world_size}card.py" + with open(worker_file, "w") as f: + f.write(SCRIPT) + + failed = [] + outputs = [] + procs = [] + + for r in range(world_size): + p = subprocess.Popen( + [sys.executable, worker_file], + env={**env, "RANK": str(r)}, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + ) + procs.append(p) + + timeout = 420 if world_size <= 4 else 900 + for r, p in enumerate(procs): + try: + out, _ = p.communicate(timeout=timeout) + txt = out.decode("utf-8", errors="replace") + except subprocess.TimeoutExpired: + p.kill() + out, _ = p.communicate() + txt = "TIMEOUT\n" + out.decode("utf-8", errors="replace") + outputs.append(txt) + if p.returncode != 0: + failed.append(r) + + return failed, outputs + + +def _run_case(world_size, master_port): + retries = 3 + for attempt in range(1, retries + 1): + failed, outputs = _run_once(world_size, master_port) + if not failed: + return + + joined = "\n".join(outputs) + transient = "resource unavailable" in joined + if transient and attempt < retries: + print( + f"HCCL transient init failure on {world_size} cards, " + f"retry {attempt}/{retries}" + ) + time.sleep(5) + continue + + for r, txt in enumerate(outputs): + print(f"=== RANK {r} ===") + print(txt) + raise AssertionError( + f"HCCL invalid split validation {world_size}card failed on ranks: {failed}" + ) + + +@pytest.mark.parametrize( + "world_size,master_port", + [ + (2, 29715), + (4, 29725), + (8, 29735), + ], +) +def test_hccl_all_to_all_single_invalid_split_pairing_multicard(world_size, master_port): + _run_case(world_size, master_port)