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 : AllGatherNHRV1::~AllGatherNHRV1() {}
18 :
19 : // 服务器间allgather的入口函数
20 0 : HcclResult AllGatherNHRV1::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
21 : {
22 0 : CHK_SMART_PTR_NULL(dispatcher_);
23 0 : CHK_PTR_NULL(stream_.ptr());
24 0 : HCCL_INFO(
25 : "[AllGatherNHRV1] run_async rank[%u] ranksize[%u] inputMem[%p] outputMem[%p] count[%llu]", rank, rankSize,
26 : inputMem_.ptr(), outputMem_.ptr(), count_);
27 :
28 : // 判断rank_size == 1
29 0 : if (rankSize == 1) {
30 0 : if (inputMem_ != outputMem_) {
31 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, outputMem_, inputMem_, stream_));
32 : }
33 0 : return HCCL_SUCCESS;
34 : }
35 :
36 0 : CHK_PRT_RET(
37 : links.size() < rankSize, HCCL_ERROR("[AllGatherNHRV1][RunAsync]rank[%u] linkSize is less than rankSize", rank),
38 : HCCL_E_INTERNAL);
39 :
40 0 : u32 unitSize = DataUnitSize(dataType_);
41 0 : CHK_PRT_RET(unitSize == 0, HCCL_ERROR("[AllGatherNHRV1][RunAsync]unitSize is zero"), HCCL_E_INTERNAL);
42 :
43 : // 处理和检查Slices
44 0 : if (slices_.size() != 0) {
45 0 : HCCL_WARNING("[AllGatherNHRV1][RunAsync]AllGatherNHRV1 not supported passing in parameter slice_, "
46 : "otherwise will be cleared");
47 0 : slices_.clear();
48 : }
49 0 : std::vector<Slice> inputSlices(slices_);
50 0 : if (slices_.size() == 0) {
51 0 : slices_.resize(rankSize);
52 0 : inputSlices.resize(rankSize);
53 0 : u64 sliceSize = count_ * unitSize;
54 0 : HCCL_DEBUG("[AllGatherNHRV1][RunAsync]sliceSize is %llu, rankSize is %u", sliceSize, rankSize);
55 0 : for (u32 i = 0; i < rankSize; i++) {
56 0 : slices_[i].size = sliceSize;
57 0 : slices_[i].offset = sliceSize * i;
58 0 : inputSlices[i].size = sliceSize;
59 0 : inputSlices[i].offset = (inputMem_.size() < outputMem_.size()) ? 0 : (sliceSize * i);
60 0 : HCCL_DEBUG(
61 : "rank[%u], slices[%u].offset=%llu, slices[%u].size=%llu", rank, i, slices_[i].offset, i,
62 : slices_[i].size);
63 : }
64 : }
65 :
66 : // 双buffer下, 先将input拷贝到output的合适位置
67 0 : if (inputMem_ != outputMem_) {
68 0 : DeviceMem dst = outputMem_.range(slices_[rank].offset, slices_[rank].size);
69 0 : DeviceMem src = inputMem_.range(inputSlices[rank].offset, inputSlices[rank].size);
70 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
71 0 : }
72 :
73 : HcclResult ret;
74 : // 获取通信关系
75 0 : RingInfo info = GetRingInfo(rankSize);
76 :
77 : // 水平方向做ring
78 0 : ret = RunAllGatherOnHorizontal(rank, links, info);
79 0 : CHK_PRT_RET(
80 : ret != HCCL_SUCCESS,
81 : HCCL_ERROR(
82 : "[AllGatherNHRV1][RunAsync]rank[%u] count[%llu] failed in "
83 : "RunAllGatherOnHorizontal step",
84 : rank, count_),
85 : ret);
86 :
87 : // 垂直方向做ring
88 0 : ret = RunAllGatherOnVertical(rank, links, info);
89 0 : CHK_PRT_RET(
90 : ret != HCCL_SUCCESS,
91 : HCCL_ERROR(
92 : "[AllGatherNHRV1][RunAsync]rank[%u] count[%llu] failed in "
93 : "RunAllGatherOnVertical step",
94 : rank, count_),
95 : ret);
96 :
97 0 : HCCL_INFO("[AllGatherNHRV1] finished: rank[%u]", rank);
98 0 : return HCCL_SUCCESS;
99 0 : }
100 :
101 0 : HcclResult AllGatherNHRV1::RunAllGatherOnHorizontal(u32 rank, const std::vector<LINK>& links, const RingInfo& info)
102 : {
103 0 : u32 ringRank = info.GetHIndex(rank); // 查找自己位于第几列，也即处于Ring中的第几个rank
104 0 : u32 ringSize = info.GetHSizeByRank(rank); // 查找自己所处的行长度，也即Ring的大小
105 0 : u32 vIndex = info.GetVIndex(rank); // 查找自己位于第几行
106 :
107 : // 收集本列各rank号，构建新的links、slices数组
108 0 : std::vector<Slice> hSlices;
109 0 : std::vector<LINK> hLinks;
110 0 : for (u32 hIdx = 0; hIdx < ringSize; hIdx++) {
111 0 : u32 oldRank = info.GetRank(vIndex, hIdx);
112 0 : CHK_PRT_RET(
113 : oldRank >= links.size(),
114 : HCCL_ERROR(
115 : "[AllGatherNHRV1] rank[%u] out of range, "
116 : "oldRank=%u, links.size=%u",
117 : rank, oldRank, links.size()),
118 : HCCL_E_INTERNAL);
119 0 : hSlices.push_back(slices_[oldRank]);
120 0 : hLinks.push_back(links[oldRank]);
121 : }
122 :
123 : // 长度不足2，直接跳过
124 0 : if (hLinks.size() < 2) {
125 0 : return HCCL_SUCCESS;
126 : }
127 :
128 : std::unique_ptr<AlgTemplateBase> tempAlg
129 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
130 0 : CHK_SMART_PTR_NULL(tempAlg);
131 0 : HCCL_INFO(
132 : "rank[%u] tempAlg AllGathering inputMem[%p] outputMem[%p] mem_size[%llu] "
133 : "count[%llu] planeID:[%d]",
134 : rank, inputMem_.ptr(), outputMem_.ptr(), inputMem_.size(), count_, profilerInput_.planeID);
135 :
136 : // 判断是否关闭AllGather的barrier
137 0 : if (!barrierSwitchOn_) {
138 0 : tempAlg->CloseBarrier();
139 : }
140 :
141 : // 调用AllGather ring的算法执行
142 0 : CHK_RET(tempAlg->Prepare(
143 : outputMem_, outputMem_, outputMem_, count_, dataType_, stream_, reductionOp_, root_, hSlices, baseOffset_));
144 :
145 0 : CHK_RET(tempAlg->RegisterProfiler(profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
146 :
147 0 : HCCL_DEBUG("[AllGatherNHRV1][Horizontal] rank[%u], ringRank=%u, ringSize=%u", rank, ringRank, ringSize);
148 0 : return tempAlg->RunAsync(ringRank, ringSize, hLinks);
149 0 : }
150 :
151 0 : HcclResult AllGatherNHRV1::RunAllGatherOnVertical(u32 rank, const std::vector<LINK>& links, const RingInfo& info)
152 : {
153 0 : u32 hIndex = info.GetHIndex(rank); // 查找自己位于第几列
154 :
155 0 : u32 hIndexForRing = (hIndex < info.GetRowSize()) ? hIndex : info.GetVIndex(rank); // 属于第几个垂直Ring
156 0 : u32 vSizeForRing = info.GetVSizeByHIndex(hIndexForRing); // 所属垂直Ring的大小
157 :
158 : // 收集本列各rank号，构建新的links、slices数组
159 0 : std::vector<LINK> vLinks;
160 0 : std::vector<Slice> vSlices;
161 0 : for (u32 vIdx = 0; vIdx < vSizeForRing; vIdx++) {
162 0 : u32 oldLRank = info.GetRank(vIdx, hIndexForRing);
163 0 : CHK_PRT_RET(
164 : oldLRank >= links.size(),
165 : HCCL_ERROR(
166 : "[AllGatherNHRV1] rank[%u] out of range, "
167 : "oldLRank=%u, links.size=%u",
168 : rank, oldLRank, links.size()),
169 : HCCL_E_INTERNAL);
170 0 : vLinks.push_back(links[oldLRank]);
171 0 : Slice slice;
172 0 : slice.size = slices_[vIdx].size * info.GetHSizeByVIndex(vIdx);
173 0 : u32 oldSRank = info.GetRank(vIdx, 0);
174 0 : CHK_PRT_RET(
175 : oldSRank >= links.size(),
176 : HCCL_ERROR(
177 : "[AllGatherNHRV1] rank[%u] out of range, "
178 : "oldSRank=%u, links.size=%u",
179 : rank, oldSRank, links.size()),
180 : HCCL_E_INTERNAL);
181 0 : slice.offset = slices_[oldSRank].offset;
182 0 : vSlices.push_back(slice);
183 : }
184 :
185 : // -- 可能还涉及跳跃的一个链接，比如8节点
186 : // ---- 0 1 2
187 : // ---- 3 4 5
188 : // ---- 6 7
189 : // -- 两个垂直Ring分别是{0,3,6,2}和{1,4,7,5}，而不是{0,3,6}和{1,4,7}
190 0 : if (info.GetHSizeByVIndex(hIndexForRing) > info.GetRowSize()) {
191 0 : u32 oldLRank = info.GetRank(hIndexForRing, info.GetSqrtRankSize());
192 0 : CHK_PRT_RET(
193 : oldLRank >= links.size(),
194 : HCCL_ERROR(
195 : "[AllGatherNHRV1] rank[%u] out of range, "
196 : "oldLRank=%u, links.size=%u",
197 : rank, oldLRank, links.size()),
198 : HCCL_E_INTERNAL);
199 0 : vLinks.push_back(links[oldLRank]);
200 0 : Slice slice;
201 0 : slice.offset = 0;
202 0 : slice.size = 0;
203 0 : vSlices.push_back(slice);
204 : }
205 :
206 : // 长度不足2，直接跳过
207 0 : if (vLinks.size() < 2) {
208 0 : return HCCL_SUCCESS;
209 : }
210 :
211 : std::unique_ptr<AlgTemplateBase> tempAlg
212 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
213 0 : CHK_SMART_PTR_NULL(tempAlg);
214 0 : HCCL_INFO(
215 : "rank[%u] tempAlg AllGathering inputMem[%p] outputMem[%p] mem_size[%llu] "
216 : "count[%llu] planeID:[%d]",
217 : rank, inputMem_.ptr(), outputMem_.ptr(), inputMem_.size(), count_, profilerInput_.planeID);
218 : // 判断是否关闭allgather的barrier
219 0 : if (!barrierSwitchOn_) {
220 0 : tempAlg->CloseBarrier();
221 : }
222 : // 调用allgather ring的算法执行
223 0 : CHK_RET(tempAlg->Prepare(
224 : outputMem_, outputMem_, outputMem_, count_, dataType_, stream_, reductionOp_, root_, vSlices, baseOffset_));
225 :
226 0 : CHK_RET(tempAlg->RegisterProfiler(profilerInput_.planeID, profilerInput_.stage, profilerInput_.step, stream_));
227 :
228 : // 计算在垂直Ring中的rank号
229 0 : u32 subRank = (hIndex == hIndexForRing) ? info.GetVIndex(rank) : vSizeForRing;
230 :
231 0 : HCCL_DEBUG("[AllGatherNHR][Vertical] rank[%u], subRank=%u, ringSize=%u", rank, subRank, vLinks.size());
232 0 : return tempAlg->RunAsync(subRank, vLinks.size(), vLinks);
233 0 : }
234 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_NHRV1, AllGatherNHRV1);
235 : } // namespace hccl
|
