已合并
[sync] PR-38965: 在A5机器调用div_时加参数rounding_mode=trunc #39163
ascend-robot创建于 6月24日
[sync] PR-38965: 在A5机器调用div_时加参数rounding_mode=trunc #39163
已合并
共 3 个文件变更+23-14
| @@ -85,7 +85,7 @@ class HcclAlltoAllSingleTest(TestCase): | |||
| 85 | c2p = ctx.Queue(2) | 85 | c2p = ctx.Queue(2) |
| 86 | p2c = ctx.Queue(2) | 86 | p2c = ctx.Queue(2) |
| 87 | expected = [] | 87 | expected = [] |
| 88 | - expectError = "Split sizes dosen't match total dim 0 size" | 88 | + expectError = "Split sizes doesn't match total dim 0 size" |
| 89 | ps = [] | 89 | ps = [] |
| 90 | for i in range(ws): | 90 | for i in range(ws): |
| 91 | p = ctx.Process( | 91 | p = ctx.Process( |
| @@ -1,14 +1,12 @@ | |||
| 1 | import unittest | 1 | import unittest |
| 2 | -import os | ||
| 3 | 2 | ||
| 4 | import numpy as np | 3 | import numpy as np |
| 5 | import torch | 4 | import torch |
| 6 | import torch.distributed as dist | 5 | import torch.distributed as dist |
| 7 | -import torch.multiprocessing as mp | ||
| 8 | import torch_npu | 6 | import torch_npu |
| 9 | 7 | ||
| 10 | from torch_npu.testing.testcase import TestCase, run_tests | 8 | from torch_npu.testing.testcase import TestCase, run_tests |
| 11 | -from torch_npu.testing.common_utils import create_common_tensor | 9 | +from torch_npu.testing.common_utils import create_common_tensor, SupportedDevices |
| 12 | from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU | 10 | from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU |
| 13 | 11 | ||
| 14 | from test_reduce_scatter import HcclReduceScatterTestBase | 12 | from test_reduce_scatter import HcclReduceScatterTestBase |
 | |||
| @@ -60,6 +58,7 @@ class HcclReduceScatterTensorTest(HcclReduceScatterTestBase): | |||
| 60 | with test_case.assertRaisesRegex(RuntimeError, error_expect): | 58 | with test_case.assertRaisesRegex(RuntimeError, error_expect): |
| 61 | pg.reduce_scatter_tensor(output, input_tensor) | 59 | pg.reduce_scatter_tensor(output, input_tensor) |
| 62 | 60 | ||
| 61 | + | ||
| 63 | 62 | ||
| 64 | def test_reduce_scatter_tensor(self): | 63 | def test_reduce_scatter_tensor(self): |
| 65 | ranks = [2] | 64 | ranks = [2] |
| @@ -92,6 +91,7 @@ class HcclReduceScatterTensorTest(HcclReduceScatterTestBase): | |||
| 92 | dist.barrier() | 91 | dist.barrier() |
| 93 | p2c.get() | 92 | p2c.get() |
| 94 | 93 | ||
| 94 | + | ||
| 95 | 95 | ||
| 96 | def test_reduce_scatter_tensor_uneven(self): | 96 | def test_reduce_scatter_tensor_uneven(self): |
| 97 | ranks = [2] | 97 | ranks = [2] |
| @@ -108,7 +108,8 @@ class HcclReduceScatterTensorTest(HcclReduceScatterTestBase): | |||
| 108 | for _ in range(world_size): | 108 | for _ in range(world_size): |
| 109 | _, input1 = create_common_tensor(shape, -10, 10) | 109 | _, input1 = create_common_tensor(shape, -10, 10) |
| 110 | input_list.append(input1.cpu()) | 110 | input_list.append(input1.cpu()) |
| 111 | - expected = self._construct_excepted_result(input_list, world_size, torch_npu.distributed.reduce_scatter_tensor_uneven) | 111 | + expected = self._construct_excepted_result( |
| 112 | + input_list, world_size, torch_npu.distributed.reduce_scatter_tensor_uneven) | ||
| 112 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor_uneven, | 113 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor_uneven, |
| 113 | HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, world_size) | 114 | HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, world_size) |
| 114 | 115 | ||
| @@ -127,7 +128,8 @@ class HcclReduceScatterTensorTest(HcclReduceScatterTestBase): | |||
| 127 | input_list.append(input1.cpu()) | 128 | input_list.append(input1.cpu()) |
| 128 | expected = self._construct_excepted_result(input_list, world_size, dist.reduce_scatter_tensor, dist.ReduceOp.AVG) | 129 | expected = self._construct_excepted_result(input_list, world_size, dist.reduce_scatter_tensor, dist.ReduceOp.AVG) |
| 129 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor, | 130 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor, |
| 130 | - HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, world_size, dist.ReduceOp.AVG) | 131 | + HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, |
| 132 | + world_size, dist.ReduceOp.AVG) | ||
| 131 | 133 | ||
| 132 | 134 | ||
| 133 | def test_reduce_scatter_tensor_with_input_internal_format_and_offset(self): | 135 | def test_reduce_scatter_tensor_with_input_internal_format_and_offset(self): |
| @@ -169,9 +171,11 @@ class HcclReduceScatterTensorTest(HcclReduceScatterTestBase): | |||
| 169 | for _ in range(world_size): | 171 | for _ in range(world_size): |
| 170 | _, input1 = create_common_tensor(shape, -10, 10) | 172 | _, input1 = create_common_tensor(shape, -10, 10) |
| 171 | input_list.append(input1.cpu()) | 173 | input_list.append(input1.cpu()) |
| 172 | - expected = self._construct_excepted_result(input_list, world_size, dist.reduce_scatter_tensor_uneven, dist.ReduceOp.AVG) | 174 | + expected = self._construct_excepted_result( |
| 175 | + input_list, world_size, dist.reduce_scatter_tensor_uneven, dist.ReduceOp.AVG) | ||
| 173 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor_uneven, | 176 | self._test_multiprocess(HcclReduceScatterTensorTest._test_reduce_scatter_tensor_uneven, |
| 174 | - HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, world_size, dist.ReduceOp.AVG) | 177 | + HcclReduceScatterTensorTest._init_dist_hccl, expected, input_list, |
| 178 | + world_size, dist.ReduceOp.AVG) | ||
| 175 | 179 | ||
| 176 | 180 | ||
| 177 | def test_reduce_scatter_tensor_pre_mul(self): | 181 | def test_reduce_scatter_tensor_pre_mul(self): |
| @@ -409,7 +409,7 @@ void check_split_sizes(const std::vector<int64_t>& split_sizes, const at::Tensor | |||||||||||||||||||
| 409 | split_sizes.size() == static_cast<size_t>(group_size), "Number of tensor splits not equal to group size", | 409 | split_sizes.size() == static_cast<size_t>(group_size), "Number of tensor splits not equal to group size", | ||||||||||||||||
| 410 | DIST_ERROR(ErrCode::TYPE)); | 410 | DIST_ERROR(ErrCode::TYPE)); | ||||||||||||||||
| 411 | const auto sum = c10::sum_integers(split_sizes); | 411 | const auto sum = c10::sum_integers(split_sizes); | ||||||||||||||||
| 412 | - TORCH_CHECK(sum == tensor.size(0), "Split sizes dosen't match total dim 0 size", DIST_ERROR(ErrCode::TYPE)); | 412 | + TORCH_CHECK(sum == tensor.size(0), "Split sizes doesn't match total dim 0 size", DIST_ERROR(ErrCode::TYPE)); | ||||||||||||||||
| 413 | } | 413 | } | ||||||||||||||||
| 414 | } | 414 | } | ||||||||||||||||
| 415 | 415 | ||||||||||||||||||
| @@ -1711,7 +1711,7 @@ void ProcessGroupHCCL::heartbeatMonitor() | |||||||||||||||||||
| 1711 | // 1. The local rank is the first to observe a timeout.shouldDump_ will be | 1711 | // 1. The local rank is the first to observe a timeout.shouldDump_ will be | ||||||||||||||||
| 1712 | // set to true. | 1712 | // set to true. | ||||||||||||||||
| 1713 | // 2. other ranks detected the timeout and signal the local rank to dump | 1713 | // 2. other ranks detected the timeout and signal the local rank to dump | ||||||||||||||||
| 1714 | - // In addtion, monitor threads will dump if watchdog threads has no | 1714 | + // In addition, monitor threads will dump if watchdog threads has no | ||||||||||||||||
| 1715 | // heartbeat or dumpPipe is not empty. | 1715 | // heartbeat or dumpPipe is not empty. | ||||||||||||||||
| 1716 | if (shouldDump_.load()) { | 1716 | if (shouldDump_.load()) { | ||||||||||||||||
| 1717 | errorMsg = c10::str( | 1717 | errorMsg = c10::str( | ||||||||||||||||
| @@ -3084,9 +3084,9 @@ int64_t ProcessGroupHCCL::getP2PStreamId( | |||||||||||||||||||
| 3084 | void ProcessGroupHCCL::windowRegisterAndExchange(int64_t windowSize, std::vector<uint32_t>& peerRanks) | 3084 | void ProcessGroupHCCL::windowRegisterAndExchange(int64_t windowSize, std::vector<uint32_t>& peerRanks) | ||||||||||||||||
| 3085 | { | 3085 | { | ||||||||||||||||
| 3086 | TORCH_CHECK(windowSize > 0, "Window memory must be greater than 0.", DIST_ERROR(ErrCode::PARAM)); | 3086 | TORCH_CHECK(windowSize > 0, "Window memory must be greater than 0.", DIST_ERROR(ErrCode::PARAM)); | ||||||||||||||||
| 3087 | - TORCH_CHECK(!windowMem_, "Window memory cannnot be registered repeatedly.", DIST_ERROR(ErrCode::UNAVAIL)); | 3087 | + TORCH_CHECK(!windowMem_, "Window memory cannot be registered repeatedly.", DIST_ERROR(ErrCode::UNAVAIL)); | ||||||||||||||||
| 3088 | TORCH_CHECK(!c10_npu::option::OptionsManager::IsHcclZeroCopyEnable(), | 3088 | TORCH_CHECK(!c10_npu::option::OptionsManager::IsHcclZeroCopyEnable(), | ||||||||||||||||
| 3089 | - "Window memory register unsupport set HCCL_ZERO_COPY=1", DIST_ERROR(ErrCode::UNAVAIL)); | 3089 | + "Window memory register unsupported set HCCL_ZERO_COPY=1", DIST_ERROR(ErrCode::UNAVAIL)); | ||||||||||||||||
| 3090 | 3090 | ||||||||||||||||||
| 3091 | auto options = at::TensorOptions(c10::DeviceType::PrivateUse1).dtype(at::kChar); | 3091 | auto options = at::TensorOptions(c10::DeviceType::PrivateUse1).dtype(at::kChar); | ||||||||||||||||
| 3092 | windowMem_ = at::empty({windowSize}, options); | 3092 | windowMem_ = at::empty({windowSize}, options); | ||||||||||||||||
| @@ -3364,7 +3364,7 @@ c10::intrusive_ptr<ProcessGroupHCCL::WorkHCCL> ProcessGroupHCCL::initWork( | |||||||||||||||||||
| 3364 | // - initially, moved record() into workEnqueue(), but found that makes it | 3364 | // - initially, moved record() into workEnqueue(), but found that makes it | ||||||||||||||||
| 3365 | // hard to get access to profilingTitle, | 3365 | // hard to get access to profilingTitle, | ||||||||||||||||
| 3366 | // inputs, and outputs for metadata recording, and we don't want to attach | 3366 | // inputs, and outputs for metadata recording, and we don't want to attach | ||||||||||||||||
| 3367 | - // these objects to the Work becuase it has implications for keeping those | 3367 | + // these objects to the Work because it has implications for keeping those | ||||||||||||||||
| 3368 | // tensors alive longer and adds overhead when copying Work objects | 3368 | // tensors alive longer and adds overhead when copying Work objects | ||||||||||||||||
| 3369 | // between threads | 3369 | // between threads | ||||||||||||||||
| 3370 | r->trace_id_ = HCCLTraceBuffer::get()->record( | 3370 | r->trace_id_ = HCCLTraceBuffer::get()->record( | ||||||||||||||||
| @@ -6110,8 +6110,13 @@ c10::intrusive_ptr<c10d::Work> ProcessGroupHCCL::_reduce_scatter_base( | |||||||||||||||||||
| 6110 | [&](std::vector<c10_npu::NPUStream>& hcclStreams, c10::intrusive_ptr<ProcessGroupHCCL::WorkHCCL>&) { | 6110 | [&](std::vector<c10_npu::NPUStream>& hcclStreams, c10::intrusive_ptr<ProcessGroupHCCL::WorkHCCL>&) { | ||||||||||||||||
| 6111 | if (opts.reduceOp == c10d::ReduceOp::AVG) { | 6111 | if (opts.reduceOp == c10d::ReduceOp::AVG) { | ||||||||||||||||
| 6112 | c10_npu::NPUStreamGuard guard(hcclStreams[0]); | 6112 | c10_npu::NPUStreamGuard guard(hcclStreams[0]); | ||||||||||||||||
| 6113 | + bool is_atlas_a5 = c10_npu::GetSocVersion() >= c10_npu::SocVersion::Ascend950; | ||||||||||||||||||
| 6113 | for (auto& tensor : outputs) { | 6114 | for (auto& tensor : outputs) { | ||||||||||||||||
| 6114 | - tensor.div_(getSize()); | 6115 | + if (is_atlas_a5) { | ||||||||||||||||
| 6116 | + tensor.div_(getSize(), "trunc"); | ||||||||||||||||||
| 6117 | + } else { | ||||||||||||||||||
| 6118 | + tensor.div_(getSize()); | ||||||||||||||||||
| 6119 | + } | ||||||||||||||||||
🟠 High Priority 在 由于 触发条件：在 Ascend950（A5）机器上，对 float 类型 tensor 执行 证据链：
建议：在调用 改动建议
  | |||||||||||||||||||
| 6115 | } | 6120 | } | ||||||||||||||||
🟡 Medium Priority 此 diff 仅在
如果 A5 机器的整数除法默认行为发生了变化（需要显式 证据链：
建议：需要确认原始 issue（#2441）的范围：是所有 collective 操作都需要此修复，还是仅 | |||||||||||||||||||
| 6116 | } | 6121 | } | ||||||||||||||||
| 6117 | }, | 6122 | }, | ||||||||||||||||
🔵 Low Priority
test_reduce_scatter_tensor.py中删除了import os和import torch.multiprocessing as mp。经检查，该文件的代码中确实没有直接使用os或mp（多进程功能继承自HcclReduceScatterTestBase），删除是安全的。test_all_to_all_single.py中 typo 修复 "dosen't" → "doesn't" 与 C++ 代码中的同款修复一致，正确。test_reduce_scatter_tensor.py新增@SupportedDevices(['Ascend910A', 'Ascend910B', 'Ascend910_93'])装饰器到test_reduce_scatter_tensor和test_reduce_scatter_tensor_uneven。这些测试使用 SUM op（非 AVG），排除 A5 设备不会影响新增的div_("trunc")代码路径的测试覆盖（AVG 测试test_reduce_scatter_tensor_avg和test_reduce_scatter_tensor_uneven_avg未加限制，仍可在 A5 上运行）。这些均属正常的清理和测试策略调整，无问题。
建议：无需修改，此为正常的 import 清理和测试设备范围调整。