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 : #include "alg_template_multi_deter_pipeline.h"
11 : namespace hccl {
12 0 : MultiDeterPipeline::MultiDeterPipeline(const HcclDispatcher dispatcher)
13 0 : : AlgTemplateBase(dispatcher) {}
14 :
15 0 : MultiDeterPipeline::~MultiDeterPipeline() {}
16 :
17 0 : HcclResult MultiDeterPipeline::RunAsync()
18 : {
19 0 : return HCCL_SUCCESS;
20 : }
21 :
22 : // ReduceScatterDeterPipeline
23 0 : HcclResult MultiDeterPipeline::Prepare(HcomCollOpInfo *opInfo, DeviceMem &buffer, const u64 count,
24 : const u64 offset, const std::vector<Slice> &slices, const SubCommInfo &level0CommInfo,
25 : const SubCommInfo &level1CommInfo, Stream &mainStream, std::vector<Stream> &subStream,
26 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub)
27 : {
28 0 : return HCCL_SUCCESS;
29 : }
30 :
31 : // AllReduceDeterPipeline
32 0 : HcclResult MultiDeterPipeline::Prepare(HcomCollOpInfo *opInfo, DeviceMem &inBuffer, DeviceMem &outBuffer, const u64 count,
33 : const std::vector<Slice> &slices, const SubCommInfo &level0CommInfo,
34 : const SubCommInfo &level1CommInfo, Stream &mainStream, std::vector<Stream> &subStream,
35 : std::vector<std::shared_ptr<LocalNotify>> ¬ifyMain, std::vector<std::shared_ptr<LocalNotify>> ¬ifySub)
36 : {
37 0 : return HCCL_SUCCESS;
38 : }
39 :
40 0 : HcclResult MultiDeterPipeline::MainWaitSub(u32 begin, u32 end)
41 : {
42 0 : for (u32 signalIndex = begin; signalIndex < end; signalIndex++) {
43 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, streamNotifyMain_[signalIndex], INVALID_VALUE_STAGE));
44 : }
45 0 : return HCCL_SUCCESS;
46 : }
47 :
48 0 : HcclResult MultiDeterPipeline::SubRecordMain(u32 begin, u32 end)
49 : {
50 0 : for (u32 streamIndex = begin; streamIndex < end; streamIndex++) {
51 0 : CHK_RET(LocalNotify::Post(subStreams_[streamIndex], dispatcher_, streamNotifyMain_[streamIndex], -1));
52 : }
53 0 : return HCCL_SUCCESS;
54 : }
55 :
56 0 : HcclResult MultiDeterPipeline::MainRecordSub(u32 begin, u32 end)
57 : {
58 0 : for (u32 signalIndex = begin; signalIndex < end; signalIndex++) {
59 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, streamNotifySub_[signalIndex], -1));
60 : }
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 : // begin max = 7, end max = 11
65 0 : HcclResult MultiDeterPipeline::SubWaitMain(u32 begin, u32 end)
66 : {
67 0 : for (u32 streamIndex = begin; streamIndex < end; streamIndex++) {
68 0 : CHK_RET(LocalNotify::Wait(subStreams_[streamIndex], dispatcher_, streamNotifySub_[streamIndex],
69 : INVALID_VALUE_STAGE));
70 : }
71 0 : return HCCL_SUCCESS;
72 : }
73 :
74 0 : HcclResult MultiDeterPipeline::GetRemoteCclbufferDeviceMem(u32 inputSliceIndex, LINK link,
75 : u32 outputSliceIndex, DeviceMem &remoteMem)
76 : {
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : HcclResult MultiDeterPipeline::GetLocalUserInDeviceMem(u32 rankIdInAllRanks, DeviceMem &locaMem)
81 : {
82 0 : return HCCL_SUCCESS;
83 : }
84 :
85 0 : HcclResult MultiDeterPipeline::GetLocalUserOutDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem)
86 : {
87 0 : return HCCL_SUCCESS;
88 : }
89 :
90 0 : HcclResult MultiDeterPipeline::GetLocalInCclbufferDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem, bool ifUseLastSize)
91 : {
92 0 : return HCCL_SUCCESS;
93 : }
94 :
95 0 : HcclResult MultiDeterPipeline::GetLocalOutCclbufferDeviceMem(u32 rankIdInAllRanks, DeviceMem &localMem, bool ifUseLastSize)
96 : {
97 0 : return HCCL_SUCCESS;
98 : }
99 :
100 0 : HcclResult MultiDeterPipeline::RunLocalCopy()
101 : {
102 0 : return HCCL_SUCCESS;
103 : }
104 :
105 0 : HcclResult MultiDeterPipeline::RunIntraAlltoallPreSync(u32 step)
106 : {
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 0 : HcclResult MultiDeterPipeline::RunIntraAlltoall(u32 step)
111 : {
112 0 : u32 recvServerId = GetPreServerIdByStep(step); // 从上一个收 2
113 0 : u32 sendServerId = GetNextServerIdByStep(step); // 发给发下一个 1
114 : // 机内alltoall full mesh收集数据，是为了收集机内第sendServerId整块的内存（包含intraRankId_块）
115 : // 该索引是为了计算第sendServerId整块内的内存序号块（0~intraRankId_-1）
116 0 : std::vector<u32> localUsrInIndex;
117 0 : for (u32 i = intraRankId_ + 1; i < intraRankSize_ + intraRankId_; ++i) {
118 0 : localUsrInIndex.push_back(i % intraRankSize_);
119 : }
120 0 : for (u32 i = 0; i < intraRankSize_ - 1; ++i) {
121 0 : u32 sendIntraRankId = GetNextIntraRankIdByStep(i + 1);
122 0 : LINK sendIntraLink = intraLinks_[sendIntraRankId];
123 0 : DeviceMem srcMem;
124 0 : DeviceMem dstMem;
125 : // 从usrin收集发给下一个cclbufer的数据, 收集的所有数据需要发送给机间序号为sendServerId的server
126 0 : u32 needSendInputndex = GetRankIdx(sendServerId, localUsrInIndex[i]);
127 : // 发送数据到cclbufer 索引为[intraRankId_, localUsrInIndex[i]]
128 0 : u32 recvIntraRankIdx = alltoallRecvBlockIdxMap_[intraRankId_][localUsrInIndex[i]];
129 0 : u32 recvCclbufferIndex = GetRankIdx(recvServerId, recvIntraRankIdx);
130 0 : CHK_RET(GetLocalUserInDeviceMem(needSendInputndex, srcMem));
131 0 : CHK_RET(GetRemoteCclbufferDeviceMem(needSendInputndex, sendIntraLink, recvCclbufferIndex, dstMem));
132 : // 发送给机内rank索引 [serverId_, sendIntraRankId]
133 0 : u32 remoteUserRank = GetRankIdx(serverId_, sendIntraRankId);
134 : // SDMA copy write语义，因为只能写到cclbuffer
135 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, subStreams_[i], remoteUserRank, sendIntraLink->GetLinkType()));
136 0 : CHK_RET(sendIntraLink->TxDataSignal(subStreams_[i]));
137 0 : CHK_RET(sendIntraLink->RxDataSignal(subStreams_[i]));
138 0 : HCCL_DEBUG("[%s] intra-server SDMA send, intraRank: [%u] -> [%u]",
139 : __func__, intraRankId_, sendIntraRankId);
140 0 : HCCL_DEBUG("[%s] intra-server SDMA send, mem: inputMem[%u, %u] -> cclbuffer[%u, %u]; cclbufferNo[%u] -> [%u]",
141 : __func__, sendServerId, localUsrInIndex[i], recvServerId, recvIntraRankIdx, needSendInputndex, recvCclbufferIndex);
142 0 : }
143 0 : HCCL_INFO("[%s] intra-server step[%u] run alltoall success" , __func__, step);
144 0 : return HCCL_SUCCESS;
145 0 : }
146 :
147 0 : HcclResult MultiDeterPipeline::GroupTasksByStream(u32 activeCount, const std::vector<bool>& isReduceBlock,
148 : u32 retIndex, std::vector<std::vector<std::vector<std::pair<u32, u32>>>>& batchStreamTasks, // 输出：批次→流→任务
149 : std::vector<bool>& processed, std::vector<u32>& origIdxMap, u32& newActiveCount)
150 : {
151 0 : batchStreamTasks.clear(); // 清空批次任务
152 0 : processed.assign(activeCount, false);
153 0 : newActiveCount = 0;
154 :
155 0 : const u32 mergeStep = 2;
156 0 : const u32 batchSize = MAX_REDUCE_STREAM_NUM;
157 0 : const u32 totalGroups = (activeCount + mergeStep - 1) / mergeStep;
158 0 : const u32 batchNum = (totalGroups + batchSize - 1) / batchSize;
159 :
160 : // 逐批生成流任务
161 0 : for (u32 batch = 0; batch < batchNum; batch++) {
162 : // 初始化当前批次的流任务（MAX_REDUCE_STREAM_NUM条流）
163 0 : std::vector<std::vector<std::pair<u32, u32>>> streamTasks(MAX_REDUCE_STREAM_NUM);
164 0 : u32 startGroup = batch * batchSize;
165 0 : u32 endGroup = std::min((batch + 1) * batchSize, totalGroups);
166 :
167 : // 处理当前批次的分组
168 0 : for (u32 group = startGroup; group < endGroup; group++) {
169 0 : u32 idx0 = group * mergeStep;
170 0 : u32 idx1 = idx0 + 1;
171 0 : if (idx1 >= activeCount) {
172 0 : processed[idx0] = false;
173 0 : continue;
174 : }
175 :
176 : // 选择dst/src（原有优先级逻辑不变）
177 : u32 dstIdx, srcIdx;
178 0 : if (origIdxMap[idx0] == retIndex) { // 当前idx0的原始索引是目标块，强制为dst
179 0 : dstIdx = idx0;
180 0 : srcIdx = idx1;
181 0 : } else if (origIdxMap[idx1] == retIndex) { // 当前idx1的原始索引是目标块，强制为dst
182 0 : dstIdx = idx1;
183 0 : srcIdx = idx0;
184 0 : } else if (isReduceBlock[idx0] && !isReduceBlock[idx1]) {
185 0 : dstIdx = idx0;
186 0 : srcIdx = idx1;
187 0 : } else if (!isReduceBlock[idx0] && isReduceBlock[idx1]) {
188 0 : dstIdx = idx1;
189 0 : srcIdx = idx0;
190 : } else {
191 0 : dstIdx = std::max(idx0, idx1);
192 0 : srcIdx = std::min(idx0, idx1);
193 : }
194 :
195 : // 分配到当前批次的流任务中
196 0 : u32 batchInnerGroupIdx = group - startGroup;
197 0 : u32 streamId = batchInnerGroupIdx % MAX_REDUCE_STREAM_NUM;
198 0 : streamTasks[streamId].emplace_back(srcIdx, dstIdx);
199 :
200 0 : processed[srcIdx] = true;
201 0 : processed[dstIdx] = false;
202 :
203 0 : HCCL_DEBUG("[%s] batch[%u] group[%u] merge src[%u] -> dst[%u] on stream[%u]", __func__,
204 : batch, group, srcIdx, dstIdx, streamId + reduceStreamBegin_);
205 : }
206 :
207 : // 将当前批次的流任务加入总批次列表
208 0 : batchStreamTasks.push_back(streamTasks);
209 0 : }
210 :
211 0 : newActiveCount = std::count(processed.begin(), processed.end(), false);
212 0 : return HCCL_SUCCESS;
213 : }
214 :
215 0 : HcclResult MultiDeterPipeline::BatchPostNotifyForStreams(
216 : const std::vector<std::vector<std::pair<u32, u32>>>& streamTasks, bool isStartPhase, bool useMainStream)
217 : {
218 0 : return HCCL_SUCCESS;
219 : }
220 :
221 0 : HcclResult MultiDeterPipeline::ExecuteStreamTasks(const std::vector<std::vector<std::pair<u32, u32>>>& streamTasks,
222 : const std::vector<DeviceMem>& validMem, std::vector<u32>& origIdxMap, bool useMainStream)
223 : {
224 0 : for (u32 s = 0; s < MAX_REDUCE_STREAM_NUM; s++) {
225 0 : if (streamTasks[s].empty()) continue;
226 :
227 0 : u32 streamIdx = reduceStreamBegin_ + s;
228 0 : Stream& subStream = subStreams_[streamIdx];
229 0 : Stream& stream = useMainStream ? mainStream_ : subStream;
230 :
231 0 : for (const auto& task : streamTasks[s]) {
232 0 : u32 srcIdx = task.first;
233 0 : u32 dstIdx = task.second;
234 0 : const DeviceMem& dstMem = validMem[dstIdx];
235 0 : const DeviceMem& srcMem = validMem[srcIdx];
236 0 : u64 count = srcMem.size() / unitSize_;
237 0 : CHK_RET(HcclReduceAsync(
238 : dispatcher_, srcMem.ptr(), count, dataType_, reductionOp_,
239 : stream, dstMem.ptr(), INVALID_VALUE_RANKID, LinkType::LINK_ONCHIP, INLINE_REDUCE_BIT
240 : ));
241 0 : HCCL_DEBUG("[%s] stream[%u] execute task: merge src[%u] -> dst[%u], origSrc[%u] -> origDst[%u]",
242 : __func__, useMainStream ? 0 : streamIdx, srcIdx, dstIdx, origIdxMap[srcIdx], origIdxMap[dstIdx]);
243 : }
244 : }
245 0 : return HCCL_SUCCESS;
246 : }
247 :
248 0 : void MultiDeterPipeline::CompressActiveSet(std::vector<DeviceMem> &validMem,
249 : std::vector<bool> &isReduceBlock, std::vector<u32> &origIdxMap, const std::vector<bool> &processed,
250 : u32 &trackedTargetIdx, const u32 origRetIndex)
251 : {
252 0 : std::vector<DeviceMem> newValidMem;
253 0 : std::vector<bool> newIsReduceBlock;
254 0 : std::vector<u32> newOrigIdxMap;
255 0 : u32 newTrackedTargetIdx = 0;
256 0 : bool foundTarget = false;
257 :
258 : // 保留未处理的块（processed=false，即dst块）
259 0 : for (u32 i = 0; i < validMem.size(); i++) {
260 0 : if (!processed[i]) { // 仅保留dst块，移除src块（processed=true）
261 0 : newValidMem.push_back(validMem[i]);
262 0 : newIsReduceBlock.push_back(isReduceBlock[i]);
263 0 : newOrigIdxMap.push_back(origIdxMap[i]);
264 :
265 : // 追踪原始目标块（retIndex）的新索引
266 0 : if (!foundTarget && origIdxMap[i] == origRetIndex) {
267 0 : newTrackedTargetIdx = newValidMem.size() - 1;
268 0 : foundTarget = true;
269 : }
270 : }
271 : }
272 :
273 0 : validMem.swap(newValidMem);
274 0 : isReduceBlock.swap(newIsReduceBlock);
275 0 : origIdxMap.swap(newOrigIdxMap);
276 : // 未找到目标块时，默认指向最后一个块
277 0 : trackedTargetIdx = foundTarget ? newTrackedTargetIdx : (validMem.empty() ? 0 : validMem.size() - 1);
278 0 : HCCL_DEBUG("[%s] compressed: old size[%llu], new size[%llu], trackedTargetIdx[%u]", __func__,
279 : processed.size(), validMem.size(), trackedTargetIdx);
280 0 : }
281 :
282 0 : HcclResult MultiDeterPipeline::LocalReduce(std::vector<DeviceMem> &reduceMem,
283 : std::vector<bool> &isReduceBlock, u32 retIndex, bool useMainStream)
284 : {
285 0 : const u32 totalBlockCount = reduceMem.size();
286 : // 校验1：容器大小匹配 + retIndex越界
287 0 : if (reduceMem.size() != isReduceBlock.size() || retIndex >= totalBlockCount) {
288 0 : HCCL_ERROR("[%s] Invalid param (size mismatch: %llu vs %llu, retIndex: %u >= %u)",
289 : __func__, reduceMem.size(), isReduceBlock.size(), retIndex, totalBlockCount);
290 0 : return HCCL_E_PARA;
291 : }
292 : // 校验2：目标内存块有效
293 0 : const DeviceMem& targetCCLBuffer = reduceMem[retIndex];
294 0 : if (targetCCLBuffer.ptr() == nullptr || targetCCLBuffer.size() == 0) {
295 0 : HCCL_ERROR("[%s] Target CCLBuffer invalid (ptr: %p, size: %llu)",
296 : __func__, targetCCLBuffer.ptr(), targetCCLBuffer.size());
297 0 : return HCCL_E_MEMORY;
298 : }
299 :
300 0 : std::vector<DeviceMem> validMem = std::move(reduceMem); // 外层不使用reduceMem
301 0 : std::vector<bool> validIsReduceBlock = std::move(isReduceBlock);
302 : // 1. 动态追踪目标块索引 2. 原索引→当前索引的映射表
303 0 : std::vector<u32> origIdxMap(validMem.size());
304 0 : for (size_t i = 0; i < origIdxMap.size(); ++i) {
305 0 : origIdxMap[i] = i;
306 : }
307 :
308 0 : if (validMem.size() == 1) {
309 0 : HCCL_ERROR("[%s] validMem size is one, only target block valid", __func__);
310 0 : return HCCL_E_PARA;
311 : }
312 :
313 0 : u32 trackedTargetIdx = retIndex;
314 0 : u32 activeCount = validMem.size();
315 0 : const u32 origRetIndex = retIndex;
316 :
317 0 : while (activeCount > 1) {
318 0 : std::vector<std::vector<std::vector<std::pair<u32, u32>>>> batchStreamTasks; // 批次→流→任务
319 0 : std::vector<bool> processed(activeCount, false);
320 0 : u32 newActiveCount = 0;
321 :
322 : // 2.1 生成按批次组织的流任务
323 0 : CHK_RET(GroupTasksByStream(activeCount, validIsReduceBlock, origRetIndex,
324 : batchStreamTasks, processed, origIdxMap, newActiveCount));
325 :
326 : // 2.2 逐批执行流任务（串行处理每批）
327 0 : for (const auto& streamTasks : batchStreamTasks) {
328 : // a. 执行start phase notify（仅处理有任务的流）
329 0 : CHK_RET(BatchPostNotifyForStreams(streamTasks, true, useMainStream));
330 : // b. 执行当前批次的流任务（src→dst归约）
331 0 : CHK_RET(ExecuteStreamTasks(streamTasks, validMem, origIdxMap, useMainStream));
332 : // c. 执行sync phase notify（等待当前批次完成）
333 0 : CHK_RET(BatchPostNotifyForStreams(streamTasks, false, useMainStream));
334 : }
335 : // 2.3 压缩活跃块（移除src块，保留dst块）
336 0 : CompressActiveSet(validMem, validIsReduceBlock, origIdxMap, processed, trackedTargetIdx, origRetIndex);
337 0 : activeCount = validMem.size();
338 0 : HCCL_DEBUG("[LocalReduce] round done: activeCount=%u -> %u", newActiveCount, activeCount);
339 0 : }
340 0 : HCCL_DEBUG("[%s] Local reduce success (merge to retIndex[%u] CCLBuffer, final tracked idx[%u])",
341 : __func__, retIndex, trackedTargetIdx);
342 0 : return HCCL_SUCCESS;
343 0 : }
344 :
345 0 : HcclResult MultiDeterPipeline::RunIntraLocalReduce(u32 step)
346 : {
347 0 : return HCCL_SUCCESS;
348 : }
349 :
350 0 : HcclResult MultiDeterPipeline::RunInterSend(u32 step)
351 : {
352 0 : return HCCL_SUCCESS;
353 : }
354 :
355 0 : HcclResult MultiDeterPipeline::RunFinalReduce()
356 : {
357 0 : return HCCL_SUCCESS;
358 : }
359 :
360 0 : HcclResult MultiDeterPipeline::AlltoallSync(u32 step, bool isStartPhase)
361 : {
362 0 : return HCCL_SUCCESS;
363 : }
364 :
365 0 : HcclResult MultiDeterPipeline::LocalReduceSync(u32 step, bool isStartPhase)
366 : {
367 0 : return HCCL_SUCCESS;
368 : }
369 :
370 0 : HcclResult MultiDeterPipeline::AlltoallLocalReduceSync(u32 step, bool isStartPhase)
371 : {
372 0 : bool alltoallStep = (step < allSteps_);
373 0 : bool localReduceStep = (step > 1 && step < allSteps_ + 1);
374 0 : if (alltoallStep) {
375 0 : CHK_RET(AlltoallSync(step, isStartPhase));
376 : }
377 0 : if (localReduceStep) {
378 0 : CHK_RET(LocalReduceSync(step, isStartPhase));
379 : }
380 0 : return HCCL_SUCCESS;
381 : }
382 :
383 0 : HcclResult MultiDeterPipeline::RunAsyncLocalReduceSerial()
384 : {
385 0 : HCCL_INFO("[MultiDeterPipeline] run begin: rank[%u] ranksize[%u] inputMem[%p] outputMem[%p]",
386 : userRank_, userRankSize_, usrInMemPtr_, usrOutMemPtr_);
387 0 : CHK_SMART_PTR_NULL(dispatcher_);
388 : // 以机内8卡为例主流 + 从流 = 1 + 7 + 4 = 12
389 0 : allSteps_ = serverSize_;
390 : // #1 机间发送，#2 local reduce，#3 机内alltoall
391 : // 以机间的pairwise来分步，#n表示pairwise的第n步
392 0 : for (u32 step = 1; step < allSteps_ + 1; step++) {
393 0 : HCCL_DEBUG("[%s] userRank[%u], intraRankId[%u], serverId[%u], step[%u/%u] begin", __func__,
394 : userRank_, intraRankId_, serverId_, step, allSteps_);
395 : // alltoall 主从流同步+前同步+拉齐
396 0 : CHK_RET(AlltoallSync(step, true));
397 0 : if (step < allSteps_ - 1) {
398 0 : CHK_RET(RunIntraAlltoallPreSync(step));
399 : }
400 0 : if (step == allSteps_ - 1) {
401 0 : CHK_RET(RunIntraAlltoallPreSync(0));
402 : }
403 0 : if (step == 1) {
404 0 : CHK_RET(RunLocalCopy());
405 : }
406 0 : if (step > 1) {
407 0 : CHK_RET(RunInterSend(step - 1));
408 : }
409 : // #1 机内RS alltoall + local reduce串行
410 0 : if (step < allSteps_) {
411 0 : CHK_RET(RunIntraAlltoall(step));
412 0 : CHK_RET(AlltoallSync(step, false));
413 0 : CHK_RET(LocalReduceSync(step, true));
414 0 : CHK_RET(RunIntraLocalReduce(step));
415 0 : CHK_RET(LocalReduceSync(step, false));
416 : } else {
417 : // #0 机内RS alltoall + local reduce串行，#0不需要向其他机发送数据
418 0 : CHK_RET(RunIntraAlltoall(0));
419 0 : CHK_RET(AlltoallSync(0, false));
420 0 : CHK_RET(LocalReduceSync(0, true));
421 0 : CHK_RET(RunIntraLocalReduce(0));
422 0 : CHK_RET(LocalReduceSync(0, false));
423 : }
424 : }
425 : // 总local reduce
426 0 : CHK_RET(RunFinalReduce());
427 0 : HCCL_INFO("[MultiDeterPipeline] MultiDeterPipeline success userRank[%u] ", userRank_);
428 0 : return HCCL_SUCCESS;
429 : }
430 :
431 : // 每个server内首先要进行alltoall full mesh收集数据，再进行机内local reduce，最后发送给指定server
432 0 : HcclResult MultiDeterPipeline::RunAsyncReduceScatterPipeline()
433 : {
434 0 : constexpr u64 HCCL_MEDIUM_COUNT_2_MB = 2 * 1024 * 1024;
435 : // 2机或者数据量小于2MB走localreduce串行算法
436 0 : if (serverSize_ <= LOCAL_REDUCE_SERIIAL_ALG_SERVER_NUM || GetLocalReduceSerialThresh() < HCCL_MEDIUM_COUNT_2_MB) {
437 0 : CHK_RET(RunAsyncLocalReduceSerial());
438 0 : return HCCL_SUCCESS;
439 : }
440 0 : HCCL_INFO("[MultiDeterPipeline] [%s] begin, userRank[%u]", __func__, userRank_);
441 : // 以机内8卡为例主流 + 从流 = 1 + 7 + 4 = 12
442 : // pairwise总共需要serverSize_步，#1机内alltoall只能自己执行，无法和其他步骤并行，所以总共需要serverSize_ + 1个步骤
443 0 : allSteps_ = serverSize_ + 1;
444 : // #1 机间发送，#2 local reduce，#3 机内alltoall
445 : // 以机间的pairwise来分步，#n表示pairwise的第n步
446 0 : for (u32 step = 1; step < allSteps_ + 1; step++) {
447 0 : HCCL_DEBUG("[%s] userRank[%u], intraRankId[%u], serverId[%u], step[%u/%u] begin", __func__,
448 : userRank_, intraRankId_, serverId_, step, allSteps_);
449 0 : CHK_RET(AlltoallLocalReduceSync(step, true));
450 0 : if (step == 1) {
451 0 : CHK_RET(RunLocalCopy());
452 : }
453 0 : if (step < allSteps_ - 1) {
454 0 : CHK_RET(RunIntraAlltoallPreSync(step));
455 : }
456 0 : if (step == allSteps_ - 1) {
457 0 : CHK_RET(RunIntraAlltoallPreSync(0));
458 : }
459 0 : if (step > STEP_OFFSET_TWO) {
460 0 : CHK_RET(RunInterSend(step - STEP_OFFSET_TWO));
461 : }
462 : // allSteps_最小为3
463 0 : if (step > 1 && step < allSteps_) {
464 0 : CHK_RET(RunIntraLocalReduce(step - 1));
465 : }
466 : // 最后一步，需要额外执行 #0 机内RS local reduce，#0不需要向其他机发送数据
467 0 : if (step == allSteps_) {
468 0 : CHK_RET(RunIntraLocalReduce(0));
469 : }
470 0 : if (step < allSteps_ - 1) {
471 0 : CHK_RET(RunIntraAlltoall(step));
472 : }
473 0 : if (step == allSteps_ - 1) {
474 : // #0 机内RS alltoall
475 0 : CHK_RET(RunIntraAlltoall(0));
476 : }
477 0 : CHK_RET(AlltoallLocalReduceSync(step, false));
478 : }
479 : // 总local reduce
480 0 : CHK_RET(RunFinalReduce());
481 0 : HCCL_INFO("[MultiDeterPipeline] [%s] end, userRank[%u]", __func__, userRank_);
482 0 : return HCCL_SUCCESS;
483 : }
484 :
485 : // 遍历所有发送方rank和发送方块索引，预计算映射关系，目的是构造出下属矩阵
486 : // srcRank\srcBlockIdx 0 1 2
487 : // 0 MAX 0 0
488 : // 1 0 MAX 1
489 : // 2 1 1 MAX
490 : // 例1，发送方是rank0，接收端是rank1，那么接收端非自身 rank 列表为[0, 2]，那么rank0的索引为0，所以就发到rank1的第0块内存
491 : // 例2，发送方是rank0，接收端是rank2，那么接收端非自身 rank 列表为[0, 1]，那么rank0的索引为0，所以就发到rank2的第0块内存
492 : // 例3，发送方是rank1，接收端是rank2，那么接收端非自身 rank 列表为[0, 1]，那么rank1的索引为1，所以就发到rank2的第1块内存
493 0 : void MultiDeterPipeline::InitAlltoallRecvBlockIdxMap()
494 : {
495 0 : const u32 rankSize = intraRankSize_;
496 0 : alltoallRecvBlockIdxMap_.resize(rankSize, std::vector<u32>(rankSize, UINT32_MAX));
497 0 : if (rankSize <= 1) {
498 0 : return;
499 : }
500 : // 规则一：接收端的块索引 = 发送方 rank 在「接收端非自身 rank 列表」中的索引
501 0 : std::vector<std::vector<u32>> dstRankToRankIndex(rankSize, std::vector<u32>(rankSize, UINT32_MAX));
502 0 : for (u32 dstRank = 0; dstRank < rankSize; ++dstRank) {
503 0 : for (u32 srcRank = 0; srcRank < rankSize; ++srcRank) {
504 0 : if (srcRank == dstRank) {
505 0 : continue;
506 : }
507 : // 若 srcRank < dstRank：索引 = srcRank, 若 srcRank > dstRank：索引 = srcRank - 1
508 0 : const u32 idx = (srcRank < dstRank) ? srcRank : (srcRank - 1);
509 0 : dstRankToRankIndex[dstRank][srcRank] = idx;
510 : }
511 : }
512 0 : for (u32 srcRank = 0; srcRank < rankSize; ++srcRank) {
513 0 : for (u32 srcBlockIdx = 0; srcBlockIdx < rankSize; ++srcBlockIdx) {
514 : // 规则二：发送方块索引 = 接收端 rank
515 0 : const u32 dstRank = srcBlockIdx;
516 : // 跳过自身发送的无效场景
517 0 : if (srcRank == dstRank) {
518 0 : continue;
519 : }
520 : // 查找发送方rank在列表中的索引，存入映射表
521 0 : const u32 dstBlockIdx = dstRankToRankIndex[dstRank][srcRank];
522 0 : alltoallRecvBlockIdxMap_[srcRank][srcBlockIdx] = dstBlockIdx;
523 0 : HCCL_DEBUG("[%s] srcRank[%u], srcBlockIdx[%u] -> dstRank[%u], dstBlockIdx[%u]", __func__,
524 : srcRank, srcBlockIdx, dstRank, dstBlockIdx);
525 : }
526 : }
527 0 : }
528 :
529 0 : HcclResult MultiDeterPipeline::PrepareTopoInfo(const SubCommInfo &level0CommInfo,
530 : const SubCommInfo &level1CommInfo)
531 : {
532 0 : serverSize_ = level1CommInfo.localRankSize;
533 0 : CHK_PRT_RET(serverSize_ < MIN_SERVER_NUM,
534 : HCCL_ERROR("[%s] Unexpected inter rank size[%u], which should >= 2.", __func__, serverSize_),
535 : HCCL_E_PARA);
536 :
537 0 : intraRankSize_ = level0CommInfo.localRankSize;
538 0 : CHK_PRT_RET(intraRankSize_ < MIN_INTRA_RANK_NUM,
539 : HCCL_ERROR("[%s] Unexpected intra rank size[%u], which should >= 3.", __func__, intraRankSize_),
540 : HCCL_E_PARA);
541 0 : intraRankId_ = level0CommInfo.localRank;
542 0 : serverId_ = level1CommInfo.localRank;
543 0 : userRankSize_ = intraRankSize_ * serverSize_;
544 0 : userRank_ = intraRankId_ + serverId_ * intraRankSize_;
545 :
546 0 : intraLinks_ = level0CommInfo.links; // 节点内
547 0 : serverLinks_ = level1CommInfo.links; // 节点间
548 0 : HCCL_INFO("[%s] opInfo: dataType[%u], unitSize[%u], memSliceSize[%u], usrInMem[%p], usrOutMem[%p], reductionOp[%u]",
549 : __func__, dataType_, unitSize_, memSliceSize_, usrInMemPtr_, usrOutMemPtr_, reductionOp_);
550 0 : HCCL_INFO("[%s] topoInfo: userRank[%u], intraRankId[%u], intraRankSize[%u], serverId[%u], interRankSize[%u]",
551 : __func__, userRank_, intraRankId_, intraRankSize_, serverId_, serverSize_);
552 0 : HCCL_INFO("[%s] topoInfo: severLinksNum[%zu], intraLinksNum[%zu]", __func__, serverLinks_.size(), intraLinks_.size());
553 0 : InitAlltoallRecvBlockIdxMap();
554 0 : return HCCL_SUCCESS;
555 : }
556 : }
|
