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 "scatter_ring_concurrent_direct.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : ScatterRingConcurrentDirect::ScatterRingConcurrentDirect(const HcclDispatcher dispatcher)
16 0 : : AlgTemplateBase(dispatcher)
17 : {
18 0 : }
19 :
20 0 : ScatterRingConcurrentDirect::~ScatterRingConcurrentDirect()
21 : {
22 0 : }
23 :
24 0 : HcclResult ScatterRingConcurrentDirect::Prepare(HcomCollOpInfo *opInfo, const u32 userRank,
25 : std::vector<Stream> &subStreams, const std::vector<std::shared_ptr<LocalNotify>> &mainSignals,
26 : const std::vector<std::shared_ptr<LocalNotify>> &subSignals, const std::vector<u32> &ringsOrder,
27 : const std::vector<Slice> &userMemSlices, bool isSdma)
28 : {
29 0 : opInfo_ = opInfo;
30 0 : userRank_ = userRank;
31 0 : subStreams_ = subStreams;
32 0 : mainSignals_ = mainSignals;
33 0 : subSignals_ = subSignals;
34 0 : ringsOrder_ = ringsOrder;
35 0 : userMemInputSlices_ = userMemSlices;
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 : // reduce scatter ring direct算法的函数入口
40 0 : HcclResult ScatterRingConcurrentDirect::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK> &links)
41 : {
42 : // 基本的检查
43 0 : CHK_RET(CheckParameters(rank, rankSize, links));
44 :
45 : // 判断rank_size == 1, 若inputMem_ != outputMem_，才需要搬运
46 0 : if (rankSize == 1) {
47 0 : CHK_RET(OneRankMemcpy());
48 0 : return HCCL_SUCCESS;
49 : }
50 : // 收集邻居信息
51 0 : CHK_RET(GetInitializedNeighborLinks(rank, rankSize, links));
52 : // 填充slice_
53 0 : CHK_RET(SetSlices(rank, rankSize));
54 :
55 : // 运行scatter, ring算法
56 0 : CHK_RET(RunScatter(rank, rankSize));
57 :
58 0 : if (barrierSwitchOn_) {
59 : // 执行barrier，保证数据收发完成
60 0 : CHK_RET(ExecuteBarrier(leftLink_, rightLink_));
61 : }
62 0 : CHK_RET(LaunchTaskExtend(dispatcher_, stream_, subStreams_));
63 :
64 0 : HCCL_INFO("ScatterRingConcurrentDirect finished: rank[%u]", rank);
65 0 : return HCCL_SUCCESS;
66 : }
67 :
68 0 : HcclResult ScatterRingConcurrentDirect::CheckParameters(const u32 rank, const u32 rankSize,
69 : const std::vector<LINK> &links)
70 : {
71 0 : CHK_PTR_NULL(opInfo_);
72 0 : CHK_RET(CheckConcurrentDirectParameters(rank, rankSize, links));
73 : // 判断subStreams数量是否正确
74 0 : CHK_PRT_RET(subStreams_.size() < 1,
75 : HCCL_ERROR("[ScatterRingConcurrentDirect] subStreams size[%u] is less than 1", subStreams_.size()),
76 : HCCL_E_PARA);
77 0 : for (auto &s : subStreams_) {
78 0 : CHK_PTR_NULL(s.ptr());
79 : }
80 : // 判断mainSignals数量是否正确
81 0 : CHK_PRT_RET(mainSignals_.size() < 1,
82 : HCCL_ERROR("[ScatterRingConcurrentDirect] mainSignals size[%u] is less than 1", mainSignals_.size()),
83 : HCCL_E_PARA);
84 : // 判断subSignals数量是否正确
85 0 : CHK_PRT_RET(subSignals_.size() < 1,
86 : HCCL_ERROR("[ScatterRingConcurrentDirect] subSignals size[%u] is less than 1", subSignals_.size()),
87 : HCCL_E_PARA);
88 : // 判断ringsOrder数量是否正确
89 0 : CHK_PRT_RET(ringsOrder_.size() != rankSize,
90 : HCCL_ERROR("[ScatterRingConcurrentDirect] ringsOrder size[%u] is not equal to rank size[%u]",
91 : ringsOrder_.size(), rankSize),
92 : HCCL_E_PARA);
93 : // 判断userMemInputSlices数量是否正确
94 0 : CHK_PRT_RET(userMemInputSlices_.size() != rankSize,
95 : HCCL_ERROR("[ScatterRingConcurrentDirect] userMemInputSlices size[%u] is not equal to rank size[%u]",
96 : userMemInputSlices_.size(), rankSize),
97 : HCCL_E_PARA);
98 0 : HCCL_INFO("ScatterRingConcurrentDirect CheckParameters success");
99 0 : return HCCL_SUCCESS;
100 : }
101 :
102 0 : HcclResult ScatterRingConcurrentDirect::OneRankMemcpy()
103 : {
104 0 : const Slice &srcSlice = userMemInputSlices_[0];
105 0 : const Slice &dstSlice = slices_[0];
106 0 : DeviceMem src = DeviceMem::create(static_cast<u8 *>(opInfo_->inputAddr) + srcSlice.offset, srcSlice.size);
107 0 : DeviceMem dst;
108 0 : if (opInfo_->outputAddr != nullptr) {
109 : // opInfo_->outputAddr != nullptr指示要将输出发送至user output
110 0 : u64 stepOffset = slices_[ringsOrder_[0]].offset;
111 0 : HCCL_DEBUG("[OneRankMemcpy]Memcpy operation: stream[main], rank[%u] starts to rcv offset[%llu], size[%llu] at userMemOut_",
112 : userRank_, stepOffset, dstSlice.size);
113 0 : dst = DeviceMem::create(static_cast<u8 *>(opInfo_->outputAddr) + stepOffset, dstSlice.size);
114 : } else {
115 : // opInfo_->outputAddr == nullptr指示要将输出发送至CCL buffer
116 0 : HCCL_DEBUG("[OneRankMemcpy]Memcpy operation: stream[main], rank[%u] starts to rcv offset[%llu], size[%llu] at outputMem_",
117 : userRank_, dstSlice.offset, dstSlice.size);
118 0 : dst = outputMem_.range(dstSlice.offset, dstSlice.size);
119 : }
120 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
121 0 : return HCCL_SUCCESS;
122 0 : }
123 :
124 0 : HcclResult ScatterRingConcurrentDirect::GetInitializedNeighborLinks(const u32 rank, const u32 rankSize,
125 : const std::vector<LINK> &links)
126 : {
127 : // 收集左邻居信息
128 0 : leftLink_ = links[(rank + rankSize - 1) % rankSize];
129 0 : CHK_SMART_PTR_NULL(leftLink_);
130 :
131 : // 收集右邻居信息
132 0 : rightLink_ = links[(rank + 1) % rankSize];
133 0 : CHK_SMART_PTR_NULL(rightLink_);
134 0 : HCCL_INFO("ScatterRingConcurrentDirect finished to GetInitializedNeighborLinks");
135 0 : return HCCL_SUCCESS;
136 : }
137 :
138 0 : HcclResult ScatterRingConcurrentDirect::SetSlices(const u32 rank, const u32 rankSize)
139 : {
140 0 : if (slices_.size() == 0) {
141 0 : slices_.resize(rankSize);
142 :
143 : // 生成std::vector<Slice> slices_
144 0 : u64 sliceSize = count_ * SIZE_TABLE[dataType_];
145 : ;
146 :
147 0 : for (u32 i = 0; i < rankSize; i++) {
148 0 : slices_[i].size = sliceSize;
149 : // 用于DMA消减过程中，消除src与dst不对位的风险
150 0 : slices_[i].offset = RoundUpWithDivisor(i * sliceSize, HCCL_MIN_SLICE_ALIGN);
151 :
152 0 : HCCL_DEBUG("[SetSlices]rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu]", rank, i, slices_[i].offset, i,
153 : slices_[i].size);
154 : }
155 : }
156 0 : if (UNLIKELY(HcclCheckLogLevel(DLOG_DEBUG))) {
157 0 : for (u32 i = 0; i < slices_.size(); i++) {
158 0 : HCCL_DEBUG("[ScatterRingConcurrentDirect][SetSlices]rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu]",
159 : rank, i, slices_[i].offset, i, slices_[i].size);
160 : }
161 : }
162 : // 最后一步搬到userMemOut_的offset, 不同的ring环offset不一样
163 0 : lastStepOffset_ = slices_[ringsOrder_[0]].offset;
164 0 : HCCL_INFO("ScatterRingConcurrentDirect finished to SetSlices");
165 0 : return HCCL_SUCCESS;
166 : }
167 :
168 0 : HcclResult ScatterRingConcurrentDirect::RunInitStep(const u32 rank, const u32 rankSize)
169 : {
170 : // 例如rank[0,1,2,3]中，rank0的rxSliceIdx = 2，txSliceIdx = 3
171 0 : u32 initSlice0Idx = 0;
172 0 : initSlice0Idx = (rank + rankSize - 1) % rankSize;
173 : // 第-1步，片内将部分数据从userIn搬到cclIn
174 0 : if (rank == root_) {
175 0 : CHK_RET(MainRecordSub()); // 主流通知从流开始通信
176 0 : CHK_RET(SubWaitMain()); // 从流等待主流通知
177 0 : const Slice &srcInitSlice0 = userMemInputSlices_[initSlice0Idx];
178 : DeviceMem srcInit
179 0 : = DeviceMem::create(static_cast<u8 *>(opInfo_->inputAddr) + srcInitSlice0.offset, srcInitSlice0.size);
180 0 : const Slice &dstInitSlice0 = slices_[initSlice0Idx];
181 0 : DeviceMem dstInit = inputMem_.range(dstInitSlice0.offset, dstInitSlice0.size);
182 0 : HCCL_DEBUG("Memcpy operation: step[-1] stream[sub] src rank[%u] starts to copy(rcv) offset[%llu], size[%llu] "
183 : "on userMemInput to offset[%llu], size[%llu] on CCL",
184 : userRank_, srcInitSlice0.offset, srcInitSlice0.size, dstInitSlice0.offset, dstInitSlice0.size);
185 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, subStreams_[0]));
186 0 : CHK_RET(SubRecordMain()); // 从流通知主流通信完成
187 0 : CHK_RET(MainWaitSub()); // 主流等待从流通知
188 0 : }
189 0 : return HCCL_SUCCESS;
190 : }
191 :
192 0 : HcclResult ScatterRingConcurrentDirect::RunMainStream(const u32 stepsFromRank2Root, const u32 step,
193 : const Slice &txSlice, const Slice &rxSlice, const u32 rankSize)
194 : {
195 0 : bool needReceive = stepsFromRank2Root > 0 && stepsFromRank2Root <= (step + 1);
196 0 : bool needSend = stepsFromRank2Root <= step;
197 0 : DeviceMem src;
198 0 : DeviceMem dst;
199 : // Ack
200 0 : if (needReceive) {
201 0 : CHK_RET(leftLink_->TxAck(stream_));
202 : }
203 0 : if (needSend) {
204 0 : CHK_RET(rightLink_->RxAck(stream_));
205 : }
206 :
207 : // 不同的rank会在不同的step开始持续发送操作，距离root节点越近，越早step开始发送操作
208 0 : if (needSend) {
209 0 : src = inputMem_.range(txSlice.offset, txSlice.size);
210 0 : CHK_RET(rightLink_->TxAsync(UserMemType::INPUT_MEM, txSlice.offset + baseOffset_, src.ptr(), txSlice.size,
211 : stream_));
212 : }
213 : // 不同的rank会在不同的step开始持续发送操作，距离root节点越近，越早step开始发送操作
214 0 : if (needReceive) {
215 0 : HCCL_DEBUG("MemcpyAsync operation: step[%u] stream[main], src rank[%u] starts to send offset[%llu] size[%llu] "
216 : "from leftMem_",
217 : step, leftLink_->GetRemoteRank(), rxSlice.offset, rxSlice.size);
218 0 : if (step == rankSize - DMA_REDUCE_TWO_OFFSET && opInfo_->outputAddr != nullptr) {
219 0 : HCCL_DEBUG("MemcpyAsync operation: step[%u] stream[main], dst rank[%u] starts to rcv offset[%llu], "
220 : "size[%llu] "
221 : "at userMemOut_ .",
222 : step, userRank_, lastStepOffset_, rxSlice.size);
223 0 : dst = DeviceMem::create(static_cast<u8 *>(opInfo_->outputAddr) + lastStepOffset_, rxSlice.size);
224 : } else {
225 0 : HCCL_DEBUG("MemcpyAsync operation: step[%u] stream[main], dst rank[%u] starts to rcv offset[%llu], "
226 : "size[%llu] "
227 : "at inputMem_ .",
228 : step, userRank_, rxSlice.offset, rxSlice.size);
229 0 : dst = inputMem_.range(rxSlice.offset, rxSlice.size);
230 : }
231 0 : CHK_RET(
232 : leftLink_->RxAsync(UserMemType::INPUT_MEM, rxSlice.offset + baseOffset_, dst.ptr(), rxSlice.size, stream_));
233 : }
234 0 : return HCCL_SUCCESS;
235 0 : }
236 :
237 0 : HcclResult ScatterRingConcurrentDirect::RunSubStream(const u32 step, const Slice &subSlice, const Slice &cclSlice,
238 : const u32 rank, const u32 rankSize)
239 : {
240 0 : if (rank == root_) {
241 0 : HCCL_DEBUG("Memcpy operation: step[%u] stream[sub], src rank[%u] starts to send offset[%llu], size[%llu] "
242 : "from userMemIn_",
243 : step, userRank_, subSlice.offset, subSlice.size);
244 0 : DeviceMem src = DeviceMem::create(static_cast<u8 *>(opInfo_->inputAddr) + subSlice.offset, subSlice.size);
245 0 : DeviceMem dst;
246 0 : if (step == rankSize - DMA_REDUCE_TWO_OFFSET && opInfo_->outputAddr != nullptr) {
247 0 : HCCL_DEBUG("Memcpy operation: step[%u] stream[sub], dst rank[%u] starts to rcv offset[%llu], size[%llu] "
248 : "to userMemOut_",
249 : step, userRank_, lastStepOffset_, subSlice.size);
250 0 : dst = DeviceMem::create(static_cast<u8 *>(opInfo_->outputAddr) + lastStepOffset_, subSlice.size);
251 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStreams_[0]));
252 0 : } else {
253 0 : HCCL_DEBUG("Memcpy operation: step[%u] stream[sub], dst rank[%u] starts to rcv offset[%llu], size[%llu] "
254 : "to inputMem_",
255 : step, userRank_, cclSlice.offset, cclSlice.size);
256 0 : dst = inputMem_.range(cclSlice.offset, cclSlice.size);
257 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStreams_[0]));
258 : }
259 0 : }
260 0 : return HCCL_SUCCESS;
261 : }
262 :
263 0 : HcclResult ScatterRingConcurrentDirect::RunScatter(const u32 rank, const u32 rankSize)
264 : {
265 0 : HCCL_INFO("ScatterRingConcurrentDirect starts, the input param rank[%u]", rank);
266 : // 空拷贝用于后续操作附着
267 0 : CHK_RET(ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
268 :
269 0 : CHK_RET(RunInitStep(rank, rankSize));
270 :
271 : // 例如rank[0,1,2,3]中，rank0的rxSliceIdx = 2，txSliceIdx = 3, subSliceIdx = 1
272 0 : u32 txSliceIdx = (rank + rankSize - 1) % rankSize;
273 0 : u32 rxSliceIdx = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize;
274 0 : u32 subSliceIdx = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize; // 只存在于根节点
275 :
276 0 : u32 stepsFromRank2Root = (rank + rankSize - root_) % rankSize;
277 0 : for (u32 step = 0; step < rankSize - 1; step++) {
278 0 : const Slice &subSlice = userMemInputSlices_[subSliceIdx];
279 0 : const Slice &cclSlice = slices_[subSliceIdx];
280 0 : const Slice &txSlice = slices_[txSliceIdx];
281 0 : const Slice &rxSlice = slices_[rxSliceIdx];
282 :
283 : // 并发
284 0 : CHK_RET(MainRecordSub()); // 主流通知从流开始通信
285 0 : CHK_RET(SubWaitMain()); // 从流等待主流通知
286 :
287 : // 空拷贝用于主从流任务并发
288 0 : CHK_RET(ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
289 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, subStreams_[0], dispatcher_));
290 :
291 : // 主流
292 0 : CHK_RET(RunMainStream(stepsFromRank2Root, step, txSlice, rxSlice, rankSize));
293 :
294 : // 从流
295 0 : CHK_RET(RunSubStream(step, subSlice, cclSlice, rank, rankSize));
296 :
297 0 : CHK_RET(SubRecordMain()); // 从流通知主流通信完成
298 0 : CHK_RET(MainWaitSub()); // 主流等待从流通知
299 :
300 : // 更新索引
301 0 : subSliceIdx = (subSliceIdx + rankSize - 1) % rankSize;
302 0 : txSliceIdx = (txSliceIdx + rankSize - 1) % rankSize;
303 0 : rxSliceIdx = (rxSliceIdx + rankSize - 1) % rankSize;
304 : }
305 0 : HCCL_INFO("ScatterRingConcurrentDirect finished to RunScatter");
306 0 : return HCCL_SUCCESS;
307 : }
308 : // 主流通知从流干活
309 0 : HcclResult ScatterRingConcurrentDirect::MainRecordSub()
310 : {
311 0 : for (u32 signalIndex = 0; signalIndex < subSignals_.size(); signalIndex++) {
312 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[signalIndex],
313 : profilerInput_.stage));
314 : }
315 0 : return HCCL_SUCCESS;
316 : }
317 : // 从流等待主流
318 0 : HcclResult ScatterRingConcurrentDirect::SubWaitMain()
319 : {
320 0 : for (u32 streamIndex = 0; streamIndex < subSignals_.size(); streamIndex++) {
321 0 : CHK_RET(LocalNotify::Wait(subStreams_[streamIndex], dispatcher_, subSignals_[streamIndex],
322 : profilerInput_.stage));
323 : }
324 0 : return HCCL_SUCCESS;
325 : }
326 : // 主流等待从流
327 0 : HcclResult ScatterRingConcurrentDirect::MainWaitSub()
328 : {
329 0 : for (u32 signalIndex = 0; signalIndex < mainSignals_.size(); signalIndex++) {
330 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[signalIndex], profilerInput_.stage));
331 : }
332 0 : return HCCL_SUCCESS;
333 : }
334 : // 从流告诉主流活干完了
335 0 : HcclResult ScatterRingConcurrentDirect::SubRecordMain()
336 : {
337 0 : for (u32 streamIndex = 0; streamIndex < mainSignals_.size(); streamIndex++) {
338 0 : CHK_RET(LocalNotify::Post(subStreams_[streamIndex], dispatcher_, mainSignals_[streamIndex],
339 : profilerInput_.stage));
340 : }
341 0 : return HCCL_SUCCESS;
342 : }
343 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_SCATTER_RING_CONCURRENT_DIRECT, ScatterRingConcurrentDirect);
344 : } // namespace hccl
|
