Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "all_gather_nhr_v1.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : AllGatherNHRV1::AllGatherNHRV1(const HcclDispatcher dispatcher) : NHRV1Base(dispatcher)
16 : {
17 0 : }
18 :
19 0 : AllGatherNHRV1::~AllGatherNHRV1()
20 : {
21 0 : }
22 :
23 : // 服务器间allgather的入口函数
24 0 : HcclResult AllGatherNHRV1::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
25 : {
26 0 : CHK_SMART_PTR_NULL(dispatcher_);
27 0 : CHK_PTR_NULL(stream_.ptr());
28 0 : HCCL_INFO("[AllGatherNHRV1] run_async rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", \
29 : rank, rankSize, inputMem_.ptr(), outputMem_.ptr(), count_);
30 :
31 : // 判断rank_size == 1
32 0 : if (rankSize == 1) {
33 0 : if (inputMem_ != outputMem_) {
34 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_));
35 : }
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 0 : CHK_PRT_RET(links.size() < rankSize, HCCL_ERROR("[AllGatherNHRV1][RunAsync]rank[%u] linkSize is less than rankSize",
40 : rank), HCCL_E_INTERNAL);
41 :
42 0 : u32 unitSize = DataUnitSize(dataType_);
43 0 : CHK_PRT_RET(unitSize == 0, HCCL_ERROR("[AllGatherNHRV1][RunAsync]unitSize is zero"), HCCL_E_INTERNAL);
44 :
45 : // 处理和检查Slices
46 0 : if (slices_.size() != 0) {
47 0 : HCCL_WARNING("[AllGatherNHRV1][RunAsync]AllGatherNHRV1 not supported passing in parameter slice_, "\
48 : "otherwise will be cleared");
49 0 : slices_.clear();
50 : }
51 0 : std::vector<Slice> inputSlices(slices_);
52 0 : if (slices_.size() == 0) {
53 0 : slices_.resize(rankSize);
54 0 : inputSlices.resize(rankSize);
55 0 : u64 sliceSize = count_ * unitSize;
56 0 : HCCL_DEBUG("[AllGatherNHRV1][RunAsync]sliceSize is %llu, rankSize is %u", sliceSize, rankSize);
57 0 : for (u32 i = 0; i < rankSize; i++) {
58 0 : slices_[i].size = sliceSize;
59 0 : slices_[i].offset = sliceSize * i;
60 0 : inputSlices[i].size = sliceSize;
61 0 : inputSlices[i].offset = (inputMem_.size() < outputMem_.size()) ? 0 : (sliceSize * i);
62 0 : HCCL_DEBUG("rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu", \
63 : rank, i, slices_[i].offset, i, slices_[i].size);
64 : }
65 : }
66 :
67 : // 双buffer下, 先将input拷贝到output的合适位置
68 0 : if (inputMem_ != outputMem_) {
69 0 : DeviceMem dst = outputMem_.range(slices_[rank].offset, slices_[rank].size);
70 0 : DeviceMem src = inputMem_.range(inputSlices[rank].offset, inputSlices[rank].size);
71 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
72 0 : }
73 :
74 : HcclResult ret;
75 : // 获取通信关系
76 0 : RingInfo info = GetRingInfo(rankSize);
77 :
78 : // 水平方向做ring
79 0 : ret = RunAllGatherOnHorizontal(rank, links, info);
80 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllGatherNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
81 : "RunAllGatherOnHorizontal step", rank, count_), ret);
82 :
83 : // 垂直方向做ring
84 0 : ret = RunAllGatherOnVertical(rank, links, info);
85 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[AllGatherNHRV1][RunAsync]rank[%u] count[%llu] failed in "\
86 : "RunAllGatherOnVertical step", rank, count_), ret);
87 :
88 0 : HCCL_INFO("[AllGatherNHRV1] finished: rank[%u]", rank);
89 0 : return HCCL_SUCCESS;
90 0 : }
91 :
92 0 : HcclResult AllGatherNHRV1::RunAllGatherOnHorizontal(u32 rank, const std::vector<LINK> &links, const RingInfo &info)
93 : {
94 0 : u32 ringRank = info.GetHIndex(rank); // 查找自己位于第几列，也即处于Ring中的第几个rank
95 0 : u32 ringSize = info.GetHSizeByRank(rank); // 查找自己所处的行长度，也即Ring的大小
96 0 : u32 vIndex = info.GetVIndex(rank); // 查找自己位于第几行
97 :
98 : // 收集本列各rank号，构建新的links、slices数组
99 0 : std::vector<Slice> hSlices;
100 0 : std::vector<LINK> hLinks;
101 0 : for (u32 hIdx = 0; hIdx < ringSize; hIdx++) {
102 0 : u32 oldRank = info.GetRank(vIndex, hIdx);
103 0 : CHK_PRT_RET(oldRank >= links.size(), HCCL_ERROR("[AllGatherNHRV1] rank[%u] out of range, "\
104 : "oldRank=%u, links.size=%u", rank, oldRank, links.size()), HCCL_E_INTERNAL);
105 0 : hSlices.push_back(slices_[oldRank]);
106 0 : hLinks.push_back(links[oldRank]);
107 : }
108 :
109 : // 长度不足2，直接跳过
110 0 : if (hLinks.size() < 2) {
111 0 : return HCCL_SUCCESS;
112 : }
113 :
114 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
115 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
116 0 : CHK_SMART_PTR_NULL(tempAlg);
117 0 : HCCL_INFO("rank[%u] tempAlg AllGathering inputMem[%p] outputMem[%p] mem_size[%llu] "\
118 : "count[%llu] planeID:[%d]", rank, inputMem_.ptr(), outputMem_.ptr(), inputMem_.size(),
119 : count_, profilerInput_.planeID);
120 :
121 : // 判断是否关闭AllGather的barrier
122 0 : if (!barrierSwitchOn_) {
123 0 : tempAlg->CloseBarrier();
124 : }
125 :
126 : // 调用AllGather ring的算法执行
127 0 : CHK_RET(tempAlg->Prepare(outputMem_, outputMem_, outputMem_, count_, dataType_, stream_,
128 : reductionOp_, root_, hSlices, baseOffset_));
129 :
130 0 : CHK_RET(tempAlg->RegisterProfiler(
131 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
132 :
133 0 : HCCL_DEBUG("[AllGatherNHRV1][Horizontal] rank[%u], ringRank=%u, ringSize=%u", rank, ringRank, ringSize);
134 0 : return tempAlg->RunAsync(ringRank, ringSize, hLinks);
135 0 : }
136 :
137 0 : HcclResult AllGatherNHRV1::RunAllGatherOnVertical(u32 rank, const std::vector<LINK> &links, const RingInfo &info)
138 : {
139 0 : u32 hIndex = info.GetHIndex(rank); // 查找自己位于第几列
140 :
141 0 : u32 hIndexForRing = (hIndex < info.GetRowSize()) ? hIndex : info.GetVIndex(rank); // 属于第几个垂直Ring
142 0 : u32 vSizeForRing = info.GetVSizeByHIndex(hIndexForRing); // 所属垂直Ring的大小
143 :
144 : // 收集本列各rank号，构建新的links、slices数组
145 0 : std::vector<LINK> vLinks;
146 0 : std::vector<Slice> vSlices;
147 0 : for (u32 vIdx = 0; vIdx < vSizeForRing; vIdx++) {
148 0 : u32 oldLRank = info.GetRank(vIdx, hIndexForRing);
149 0 : CHK_PRT_RET(oldLRank >= links.size(), HCCL_ERROR("[AllGatherNHRV1] rank[%u] out of range, "\
150 : "oldLRank=%u, links.size=%u", rank, oldLRank, links.size()), HCCL_E_INTERNAL);
151 0 : vLinks.push_back(links[oldLRank]);
152 0 : Slice slice;
153 0 : slice.size = slices_[vIdx].size * info.GetHSizeByVIndex(vIdx);
154 0 : u32 oldSRank = info.GetRank(vIdx, 0);
155 0 : CHK_PRT_RET(oldSRank >= links.size(), HCCL_ERROR("[AllGatherNHRV1] rank[%u] out of range, "\
156 : "oldSRank=%u, links.size=%u", rank, oldSRank, links.size()), HCCL_E_INTERNAL);
157 0 : slice.offset = slices_[oldSRank].offset;
158 0 : vSlices.push_back(slice);
159 : }
160 :
161 : // -- 可能还涉及跳跃的一个链接，比如8节点
162 : // ---- 0 1 2
163 : // ---- 3 4 5
164 : // ---- 6 7
165 : // -- 两个垂直Ring分别是{0,3,6,2}和{1,4,7,5}，而不是{0,3,6}和{1,4,7}
166 0 : if (info.GetHSizeByVIndex(hIndexForRing) > info.GetRowSize()) {
167 0 : u32 oldLRank = info.GetRank(hIndexForRing, info.GetSqrtRankSize());
168 0 : CHK_PRT_RET(oldLRank >= links.size(), HCCL_ERROR("[AllGatherNHRV1] rank[%u] out of range, "\
169 : "oldLRank=%u, links.size=%u", rank, oldLRank, links.size()), HCCL_E_INTERNAL);
170 0 : vLinks.push_back(links[oldLRank]);
171 0 : Slice slice;
172 0 : slice.offset = 0;
173 0 : slice.size = 0;
174 0 : vSlices.push_back(slice);
175 : }
176 :
177 : // 长度不足2，直接跳过
178 0 : if (vLinks.size() < 2) {
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
183 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
184 0 : CHK_SMART_PTR_NULL(tempAlg);
185 0 : HCCL_INFO("rank[%u] tempAlg AllGathering inputMem[%p] outputMem[%p] mem_size[%llu] "\
186 : "count[%llu] planeID:[%d]", rank, inputMem_.ptr(), outputMem_.ptr(), inputMem_.size(),
187 : count_, profilerInput_.planeID);
188 : // 判断是否关闭allgather的barrier
189 0 : if (!barrierSwitchOn_) {
190 0 : tempAlg->CloseBarrier();
191 : }
192 : // 调用allgather ring的算法执行
193 0 : CHK_RET(tempAlg->Prepare(outputMem_, outputMem_, outputMem_, count_, dataType_, stream_,
194 : reductionOp_, root_, vSlices, baseOffset_));
195 :
196 0 : CHK_RET(tempAlg->RegisterProfiler(
197 : profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
198 :
199 : // 计算在垂直Ring中的rank号
200 0 : u32 subRank = (hIndex == hIndexForRing) ? info.GetVIndex(rank) : vSizeForRing;
201 :
202 0 : HCCL_DEBUG("[AllGatherNHR][Vertical] rank[%u], subRank=%u, ringSize=%u", rank, subRank, vLinks.size());
203 0 : return tempAlg->RunAsync(subRank, vLinks.size(), vLinks);
204 0 : }
205 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_NHRV1, AllGatherNHRV1);
206 : } // namespace hccl
207 :
|
