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