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