LCOV - code coverage report
Current view: top level - legacy/ascend910/framework/device/aicpu_kfc/algorithm - aicpu_allreduce.cc (source / functions) Coverage Total Hit
Test: coverage.info Lines: 58.8 % 439 258
Test Date: 2026-08-04 10:52:23 Functions: 73.3 % 30 22

            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 "aicpu_allreduce.h"
      12              : #include <cmath>
      13              : #include <algorithm>
      14              : #include "common/aicpu_hccl_common.h"
      15              : 
      16           41 : HcclResult AicpuAllreduce::RunAlgorithm(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
      17              :     HcclDataType dataType, u64 strideLen, AivAicpuOpParam * /* nextTask */)
      18              : {
      19           41 :     CHK_PTR_NULL(ctx_);
      20           41 :     if (CC_EXE_ONE_SHOT_8_STREAM == ctx_->commOpType) {
      21           34 :         return RunAllReduceReduceBcast(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
      22            7 :     } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_1_STREAM) {
      23            1 :         return RunAllReduceOneShot1Stream(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
      24            6 :     } else if (ctx_->commOpType == CC_EXE_TWO_SHOT_1_STREAM) {
      25            1 :         return RunAllReduceTwoShot1Stream(opType, sendBuffer, recvBuffer, dataCount, dataType);
      26            5 :     } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_HD) {
      27            1 :         return RunAllReduceOneshotHD(opType, sendBuffer, recvBuffer, dataCount * ctx_->unitSize, dataType);
      28            4 :     } else if (ctx_->commOpType == CC_EXE_ONE_SHOT_SINGLE_RING) {
      29            2 :         return RunAllReduceRing(opType, sendBuffer, recvBuffer, dataCount, dataType);
      30              :     }
      31              : 
      32            2 :     if (ctx_->useBufferType == MC2_BUFFER_TYPE_WINDOW_IN) {
      33            1 :         if (HCCL_SUCCESS == RunAllReduceAlignWin2Win(opType, recvBuffer, dataCount, dataType)) {
      34            0 :             return HCCL_SUCCESS;
      35              :         }
      36              :     } else {
      37            1 :         if (HCCL_SUCCESS == RunAllReduceAlign(opType, sendBuffer, recvBuffer, dataCount, dataType)) {
      38            0 :             return HCCL_SUCCESS;
      39              :         }
      40              :     }
      41              : 
      42            2 :     return RunAllReduce(opType, sendBuffer, recvBuffer, dataCount, dataType);
      43              : }
      44              : 
      45            1 : int64_t AicpuAllreduce::RoundUpWithDivisor(u64 value, u64 divisor) const
      46              : {
      47            1 :     if ((value == 0) || (divisor == 0)) {
      48            0 :         return divisor;
      49              :     }
      50              :     // divisor必须大于等于1, 返回value向上取divisor的整数倍的值
      51            1 :     return ((value + (divisor - 1)) / divisor) * divisor;
      52              : }
      53              : 
      54            1 : HcclResult AicpuAllreduce::PrepareSlice(u64 dataCount, HcclDataType dataType, u32 sliceNum,
      55              :     std::vector<Slice> &dataSlice) const
      56              : {
      57            1 :     Slice temp;
      58            1 :     u32 unitSize = DataUnitSize(dataType);
      59            1 :     u64 totalSize = dataCount * unitSize;
      60            1 :     dataSlice.clear();
      61            1 :     dataSlice.reserve(sliceNum);
      62            1 :     if (sliceNum == 0) {
      63            0 :         HCCL_ERROR("[Prepare][SliceData]data slice prepare, sliceNum is 0.");
      64            0 :         return HCCL_E_PARA;
      65              :     }
      66            1 :     u64 sizePerSlice = (totalSize + sliceNum - 1) / sliceNum; /* 1是为了向上取整 */
      67            1 :     sizePerSlice = RoundUpWithDivisor(sizePerSlice, HCCL_MIN_SLICE_ALIGN);
      68            1 :     u64 residueSize = totalSize;
      69            1 :     u32 i = 0;
      70            3 :     while (residueSize > 0) {
      71            2 :         u64 sliceSize = sizePerSlice < residueSize ? sizePerSlice : residueSize;
      72            2 :         temp.size = sliceSize;
      73            2 :         temp.offset = totalSize - residueSize;
      74            2 :         i++;
      75            2 :         if (sliceSize <= 0) {
      76            0 :             HCCL_ERROR("[Prepare][SliceData]data_slice_prepare sliceSize[%llu]", sliceSize);
      77            0 :             return HCCL_E_PARA;
      78              :         }
      79            2 :         residueSize -= sliceSize;
      80            2 :         dataSlice.push_back(temp);
      81              :     }
      82            1 :     while (i < sliceNum) {
      83            0 :         temp.size = 0;
      84            0 :         temp.offset = totalSize;
      85            0 :         i++;
      86            0 :         dataSlice.push_back(temp);
      87              :     }
      88            1 :     return HCCL_SUCCESS;
      89              : }
      90              : 
      91            3 : void AicpuAllreduce::GetDataSizes16K(std::vector<u64> &dataSizes, u64 allDataSize) const
      92              : {
      93            3 :     u64 num16k = allDataSize / HCCL_COPY_ALIGN;
      94            3 :     u64 tailSize = allDataSize % HCCL_COPY_ALIGN;
      95              : 
      96            3 :     u64 baseNum = num16k / ctx_->rankNum;
      97            3 :     u64 tailNum = num16k % ctx_->rankNum;
      98              : 
      99            7 :     for (u32 i = 0; i < ctx_->rankNum; i++) {
     100            4 :         dataSizes[i] = baseNum * HCCL_COPY_ALIGN;
     101              :     }
     102            3 :     for (u32 i = 0; i < tailNum; i++) {
     103            0 :         dataSizes[i] += HCCL_COPY_ALIGN;
     104              :     }
     105            3 :     dataSizes[ctx_->rankNum - 1] += tailSize;
     106            3 : }
     107              : 
     108            1 : HcclResult AicpuAllreduce::RunAllReduceAlignWin2Win(HcclReduceOp opType, void *recvBuffer, u64 dataCount,
     109              :     HcclDataType dataType) const
     110              : {
     111            1 :     u32 unitSize = ctx_->unitSize;
     112            1 :     u64 allDataSize = dataCount * unitSize;
     113            1 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     114              : 
     115            1 :     if (ctx_->rankNum == 0) {
     116            0 :         return HCCL_E_PARA;
     117              :     }
     118            1 :     std::vector<u64> dataSizes(ctx_->rankNum, 0);
     119            1 :     GetDataSizes16K(dataSizes, allDataSize);
     120              : 
     121            2 :     if (dataSizes[0] > ctx_->windowSize || dataSizes[ctx_->rankNum - 1] > ctx_->windowSize ||
     122            1 :         (ctx_->rankNum - 1) * HCCL_COPY_ALIGN >= allDataSize) {
     123            1 :         return HCCL_E_PARA;
     124              :     }
     125              : 
     126            0 :     std::vector<u64> dataOffsets(ctx_->rankNum, 0);
     127            0 :     for (u32 i = 1; i < ctx_->rankNum; i++) {
     128            0 :         dataOffsets[i] = dataOffsets[i - 1] + dataSizes[i - 1];
     129              :     }
     130              : 
     131            0 :     u64 winOffset = ctx_->winOffset;
     132              : 
     133              :     // 1. 前同步
     134            0 :     TaskOrchestrator::DoPreSync();
     135              : 
     136              :     // 2. 跨片SDMA,分批拷贝 + 分批结束同步
     137            0 :     TaskOrchestrator::IpcCpyWin2Win(dataSizes, winOffset, dataOffsets, opType, dataType);
     138              : 
     139              :     // 3. 后同步
     140            0 :     TaskOrchestrator::DoPostSync();
     141              : 
     142              :     // 4. 前同步
     143            0 :     TaskOrchestrator::DoPreSync();
     144              : 
     145              :     // 5. 片内数据 Win拷贝到Rcv
     146            0 :     TaskOrchestrator::SelfCpyWin2RcvEx1(curOutputPtr, dataSizes[ctx_->rankId], dataOffsets[ctx_->rankId], winOffset,
     147              :         HCCL_REDUCE_RESERVED, dataType);
     148              : 
     149              :     // 6. 跨片SDMA,分批拷贝 + 分批结束同步
     150            0 :     TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, dataSizes, dataOffsets, winOffset, HCCL_REDUCE_RESERVED, dataType);
     151              : 
     152              :     // 7. 后同步
     153            0 :     TaskOrchestrator::DoPostSync();
     154              : 
     155            0 :     TaskOrchestrator::LaunchTasks();
     156              : 
     157            0 :     return HCCL_SUCCESS;
     158            1 : }
     159              : 
     160            1 : HcclResult AicpuAllreduce::RunAllReduceAlign(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
     161              :     HcclDataType dataType) const
     162              : {
     163            1 :     u32 unitSize = ctx_->unitSize;
     164            1 :     u64 allDataSize = dataCount * unitSize;
     165              : 
     166            1 :     u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
     167            1 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     168              : 
     169            1 :     if (ctx_->rankNum == 0) {
     170            0 :         return HCCL_E_PARA;
     171              :     }
     172            1 :     std::vector<u64> dataSizes(ctx_->rankNum, 0);
     173            1 :     GetDataSizes16K(dataSizes, allDataSize);
     174              : 
     175            2 :     if (dataSizes[0] > ctx_->windowSize || dataSizes[ctx_->rankNum - 1] > ctx_->windowSize ||
     176            1 :         (ctx_->rankNum - 1) * HCCL_COPY_ALIGN >= allDataSize) {
     177            1 :         return HCCL_E_PARA;
     178              :     }
     179              : 
     180            0 :     std::vector<u64> dataOffsets(ctx_->rankNum, 0);
     181            0 :     for (u32 i = 1; i < ctx_->rankNum; i++) {
     182            0 :         dataOffsets[i] = dataOffsets[i - 1] + dataSizes[i - 1];
     183              :     }
     184              : 
     185              :     // 1. 片内数据 Snd拷贝到Window
     186            0 :     TaskOrchestrator::SelfCpySnd2Win(curInputPtr, dataSizes[ctx_->rankId], dataOffsets[ctx_->rankId], 0,
     187              :         HCCL_REDUCE_RESERVED, dataType);
     188              : 
     189              :     // 2. 前同步
     190            0 :     TaskOrchestrator::DoPreSync();
     191              : 
     192              :     // 3. 跨片SDMA,分批拷贝 + 分批结束同步
     193            0 :     TaskOrchestrator::IpcCpySnd2Win(curInputPtr, dataSizes, dataOffsets, nullptr, opType, dataType);
     194              : 
     195              :     // 4. 后同步
     196            0 :     TaskOrchestrator::DoPostSync();
     197              : 
     198              :     // 5. 前同步
     199            0 :     TaskOrchestrator::DoPreSync();
     200              : 
     201              :     // 6. 片内数据 Win拷贝到Rcv
     202            0 :     TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, dataSizes[ctx_->rankId], 0, dataOffsets[ctx_->rankId],
     203              :         HCCL_REDUCE_RESERVED, dataType);
     204              : 
     205              :     // 7. 跨片SDMA,分批拷贝 + 分批结束同步
     206            0 :     TaskOrchestrator::IpcCpyWin2Rcv(curOutputPtr, dataSizes, nullptr, dataOffsets, HCCL_REDUCE_RESERVED, dataType);
     207              : 
     208              :     // 8. 后同步
     209            0 :     TaskOrchestrator::DoPostSync();
     210              : 
     211            0 :     TaskOrchestrator::LaunchTasks();
     212              : 
     213            0 :     return HCCL_SUCCESS;
     214            1 : }
     215              : 
     216            3 : HcclResult AicpuAllreduce::RunAllReduce(HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
     217              :     HcclDataType dataType) const
     218              : {
     219            3 :     u64 windowSize = ctx_->windowSize; // window size default is 200M, maybe need read from cfg/env.
     220            3 :     u32 unitSize = ctx_->unitSize;
     221            3 :     u64 maxCountPerLoop = (windowSize / unitSize) * ctx_->rankNum; // 中转内存单次最多能够接受的output count
     222              : 
     223            3 :     u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
     224            3 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     225            3 :     u64 inputOffset = 0;
     226            3 :     u64 outputOffset = 0;
     227            3 :     u64 countLeft = dataCount;
     228              : 
     229            3 :     u64 dataSlice[AC_MAX_RANK_NUM] = {0};
     230            3 :     u64 sliceSize[AC_MAX_RANK_NUM] = {0};
     231            3 :     if (ctx_->rankNum <= 0) {
     232            1 :         return HCCL_E_UNAVAIL;
     233              :     }
     234              : 
     235            2 :     while (countLeft > 0) {
     236            0 :         curInputPtr += inputOffset;
     237            0 :         curOutputPtr += outputOffset;
     238            0 :         u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
     239            0 :         u64 curSize = curCount * unitSize; // 单位 byte
     240            0 :         u64 curRankCnt = curCount / ctx_->rankNum;
     241            0 :         for (u32 i = 0; i < ctx_->rankNum; i++) {
     242            0 :             dataSlice[i] = i * curRankCnt * unitSize;
     243            0 :             sliceSize[i] = curRankCnt * unitSize;
     244              :         }
     245            0 :         sliceSize[ctx_->rankNum - 1] += (curCount - curRankCnt * ctx_->rankNum) * unitSize;
     246              : 
     247            0 :         HCCL_DEBUG("RunAllReducev:curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr,
     248              :             curOutputPtr, curCount, curSize);
     249              : 
     250            0 :         if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
     251            0 :             RunAllReduceSlice(curOutputPtr, curInputPtr, sliceSize, dataSlice, opType, dataType);
     252              :         } else {
     253            0 :             RunAllReduceSliceWin2Win(curOutputPtr, sliceSize, dataSlice, opType, dataType);
     254              :         }
     255              : 
     256            0 :         countLeft -= curCount;
     257            0 :         inputOffset = curSize;
     258            0 :         outputOffset = curSize;
     259              :     }
     260              : 
     261            2 :     return HCCL_SUCCESS;
     262              : }
     263              : 
     264            1 : void AicpuAllreduce::RunAllReduceSliceWin2Win(u8 *curOutputPtr, u64 *sliceSize, u64 *dataSlice, HcclReduceOp opType,
     265              :     HcclDataType dataType) const
     266              : {
     267            1 :     u64 winOffset = ctx_->winOffset;
     268              :     // 1. 前同步
     269            1 :     TaskOrchestrator::DoPreSync();
     270              : 
     271              :     // 2. 跨片SDMA,分批拷贝 + 分批结束同步
     272            1 :     TaskOrchestrator::IpcCpyWin2Win(sliceSize, dataSlice, opType, winOffset, dataType);
     273              : 
     274              :     // 3. 后同步
     275            1 :     TaskOrchestrator::DoPostSync();
     276              : 
     277              :     // 4. 前同步
     278            1 :     TaskOrchestrator::DoPreSync();
     279              : 
     280              :     // 5. 片内数据 Win拷贝到Rcv
     281            1 :     TaskOrchestrator::SelfCpyWin2RcvEx1(curOutputPtr, sliceSize[ctx_->rankId], dataSlice[ctx_->rankId], winOffset,
     282              :         HCCL_REDUCE_RESERVED, dataType);
     283              : 
     284              :     // 6. 跨片SDMA,分批拷贝 + 分批结束同步
     285            1 :     TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, sliceSize, dataSlice, winOffset, HCCL_REDUCE_RESERVED, dataType);
     286              : 
     287              :     // 7. 后同步
     288            1 :     TaskOrchestrator::DoPostSync();
     289              : 
     290            1 :     TaskOrchestrator::LaunchTasks();
     291            1 : }
     292              : 
     293            1 : void AicpuAllreduce::RunAllReduceSlice(u8 *curOutputPtr, u8 *curInputPtr, u64 *sliceSize, u64 *dataSlice,
     294              :     HcclReduceOp opType, HcclDataType dataType) const
     295              : {
     296              :     // 1. 片内数据 Snd拷贝到Window
     297            1 :     TaskOrchestrator::SelfCpySnd2Win(curInputPtr, sliceSize[ctx_->rankId], dataSlice[ctx_->rankId], 0,
     298              :         HCCL_REDUCE_RESERVED, dataType);
     299              : 
     300              :     // 2. 前同步
     301            1 :     TaskOrchestrator::DoPreSync();
     302              : 
     303              :     // 3. 跨片SDMA,分批拷贝 + 分批结束同步
     304            1 :     TaskOrchestrator::IpcCpySnd2Win(curInputPtr, sliceSize, dataSlice, nullptr, opType, dataType);
     305              : 
     306              :     // 4. 后同步
     307            1 :     TaskOrchestrator::DoPostSync();
     308              : 
     309              :     // 5. 前同步
     310            1 :     TaskOrchestrator::DoPreSync();
     311              : 
     312              :     // 6. 片内数据 Win拷贝到Rcv
     313            1 :     TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, sliceSize[ctx_->rankId], 0, dataSlice[ctx_->rankId],
     314              :         HCCL_REDUCE_RESERVED, dataType);
     315              : 
     316              :     // 7. 跨片SDMA,分批拷贝 + 分批结束同步
     317            1 :     TaskOrchestrator::IpcCpyWin2Rcv(curOutputPtr, sliceSize, nullptr, dataSlice, HCCL_REDUCE_RESERVED, dataType);
     318              : 
     319              :     // 8. 后同步
     320            1 :     TaskOrchestrator::DoPostSync();
     321              : 
     322            1 :     TaskOrchestrator::LaunchTasks();
     323            1 : }
     324              : 
     325            1 : HcclResult AicpuAllreduce::RunAllReduceOneShot4Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
     326              :     u64 dataSize, HcclDataType dataType) const
     327              : {
     328              :     // 第一轮第一组
     329            1 :     u32 mainRankId = ctx_->rankId;
     330            1 :     u32 maxStreamNum = ctx_->rankNum / 2;
     331            1 :     u32 startRank = 0;
     332            1 :     u32 endRank = maxStreamNum - 1;
     333              : 
     334            1 :     if (mainRankId >= maxStreamNum) {
     335            1 :         startRank = maxStreamNum;
     336            1 :         endRank = ctx_->rankNum - 1;
     337              :     }
     338              : 
     339              :     // 第1轮
     340              :     // 1. 片内数据 拷贝到Window
     341            1 :     TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
     342              :         maxStreamNum);
     343            1 :     TaskOrchestrator::MainSubPreSync(mainRankId, startRank, endRank, maxStreamNum);
     344              : 
     345            1 :     TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, false);
     346              :     // 2. 跨片SDMA 片内Send拷贝到对端Window
     347            1 :     TaskOrchestrator::IpcCpySnd2WinEx(sendBuffer, dataSize, nullptr, nullptr, opType, dataType, startRank, endRank,
     348              :         maxStreamNum, false);
     349            1 :     TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, false);
     350              : 
     351            1 :     TaskOrchestrator::MainSubPostSync(mainRankId, startRank, endRank, maxStreamNum);
     352              : 
     353              :     // 第2轮
     354            1 :     u32 remoteRank = (ctx_->rankNum - 1) - mainRankId; // 0-7; 1-6; 2-5; 3-4
     355              :     // 3. 片内数据 拷贝到recv
     356            1 :     TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
     357              :         maxStreamNum);
     358            1 :     TaskOrchestrator::IpcPreSyncEx(remoteRank, remoteRank, maxStreamNum, true);
     359              :     // 4. 跨片SDMA Window拷贝到对端Recv
     360            1 :     TaskOrchestrator::IpcCpyWin2RcvEx(recvBuffer, dataSize, nullptr, nullptr, opType, dataType, remoteRank, remoteRank,
     361              :         maxStreamNum, true);
     362            1 :     TaskOrchestrator::IpcPostSyncEx(remoteRank, remoteRank, maxStreamNum, true);
     363              : 
     364              :     // 5. 下发sqe
     365            1 :     TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
     366              : 
     367            1 :     return HCCL_SUCCESS;
     368              : }
     369              : 
     370           34 : HcclResult AicpuAllreduce::RunReduceBcastOnMainSq(u32 mainRankId, u32 maxStreamNum, u32 /* startRank */,
     371              :     u32 /* endRank */, void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
     372              : {
     373           34 :     HCCL_DEBUG("run RunReduceBcastOnMainSq start");
     374           34 :     if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
     375              :         // 1. reduce
     376           34 :         TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType,
     377              :             maxStreamNum);
     378              :     }
     379              : 
     380              :     // 2. 前同步
     381           34 :     TaskOrchestrator::MainSubPreSync();
     382           34 :     TaskOrchestrator::IpcPreRecordEx(0, maxStreamNum - 1, maxStreamNum, false);
     383              :     // 3. 后同步
     384           34 :     TaskOrchestrator::IpcPostWaitEx(0, maxStreamNum - 1, maxStreamNum, true);
     385              : 
     386              :     // 4. 前同步
     387           34 :     TaskOrchestrator::MainSubPreSync();
     388           34 :     TaskOrchestrator::IpcPreRecordEx(0, maxStreamNum - 1, maxStreamNum, false);
     389              :     // 5. bcast
     390           34 :     TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize,
     391           34 :         ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN ? 0 : ctx_->winOffset, 0, HCCL_REDUCE_RESERVED, dataType,
     392              :         maxStreamNum);
     393              :     // 6. 后同步
     394           34 :     TaskOrchestrator::IpcPostWaitEx(0, maxStreamNum - 1, maxStreamNum, true);
     395              : 
     396           34 :     HCCL_DEBUG("run RunReduceBcastOnMainSq end");
     397           34 :     return HCCL_SUCCESS;
     398              : }
     399              : 
     400            0 : HcclResult AicpuAllreduce::RunReduceBcastOnOtherSq(HcclReduceOp opType, u32 mainRankId, u32 maxStreamNum,
     401              :     void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
     402              : {
     403            0 :     HCCL_DEBUG("run RunReduceBcastOnOtherSq start");
     404              : 
     405              :     // reduce
     406            0 :     TaskOrchestrator::IpcPreWaitEx(mainRankId, mainRankId, maxStreamNum, true);
     407            0 :     if (ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN) {
     408            0 :         TaskOrchestrator::SelfCpySnd2WinEx(mainRankId, sendBuffer, dataSize, 0, 0, opType, dataType, maxStreamNum);
     409              :     } else {
     410            0 :         TaskOrchestrator::IpcCpyWin2WinEx(mainRankId, dataSize, ctx_->winOffset, opType, dataType, maxStreamNum);
     411              :     }
     412            0 :     TaskOrchestrator::IpcPostRecordEx(mainRankId, mainRankId, maxStreamNum, true);
     413              : 
     414              :     // bcast
     415            0 :     TaskOrchestrator::IpcPreWaitEx(mainRankId, mainRankId, maxStreamNum, true);
     416            0 :     TaskOrchestrator::SelfCpyWin2RcvEx(mainRankId, recvBuffer, dataSize,
     417            0 :         ctx_->useBufferType != MC2_BUFFER_TYPE_WINDOW_IN ? 0 : ctx_->winOffset, 0, HCCL_REDUCE_RESERVED, dataType,
     418              :         maxStreamNum);
     419            0 :     TaskOrchestrator::IpcPostRecordEx(mainRankId, mainRankId, maxStreamNum, true);
     420            0 :     HCCL_DEBUG("run RunReduceBcastOnOtherSq end");
     421            0 :     return HCCL_SUCCESS;
     422              : }
     423              : 
     424           34 : HcclResult AicpuAllreduce::RunAllReduceReduceBcast(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
     425              :     u64 dataSize, HcclDataType dataType) const
     426              : {
     427           34 :     HCCL_DEBUG("run RunAllReduceReduceBcast start");
     428              : 
     429           34 :     u32 mainRankId = 0;
     430           34 :     u32 maxStreamNum = ctx_->rankNum;
     431           34 :     u32 startRank = 0;
     432           34 :     u32 endRank = ctx_->rankNum - 1;
     433              : 
     434           34 :     if (ctx_->rankId == mainRankId) {
     435           34 :         RunReduceBcastOnMainSq(mainRankId, maxStreamNum, startRank, endRank, sendBuffer, recvBuffer, dataSize,
     436              :             dataType);
     437              :     } else {
     438            0 :         RunReduceBcastOnOtherSq(opType, mainRankId, maxStreamNum, sendBuffer, recvBuffer, dataSize, dataType);
     439              :     }
     440              : 
     441              :     // 下发sqe
     442           34 :     TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
     443              : 
     444           34 :     HCCL_DEBUG("run RunAllReduceReduceBcast end");
     445           34 :     return HCCL_SUCCESS;
     446              : }
     447              : 
     448            1 : HcclResult AicpuAllreduce::RunAllReduceOneShot1Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
     449              :     u64 dataSize, HcclDataType dataType) const
     450              : {
     451            1 :     HCCL_INFO("run RunAllReduceOneShot1Stream start");
     452            1 :     u32 maxStreamNum = ctx_->rankNum;
     453            1 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     454            1 :     u32 startRank = 0;
     455            1 :     u32 endRank = maxStreamNum - 1;
     456              : 
     457            1 :     TaskOrchestrator::SelfCpySnd2WinEx1(sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType, maxStreamNum);
     458              : 
     459            1 :     TaskOrchestrator::SelfCpySnd2RcvEx(sendBuffer, recvBuffer, 0, 0, dataSize, HCCL_REDUCE_RESERVED, dataType);
     460              : 
     461            1 :     TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, true);
     462              : 
     463            1 :     TaskOrchestrator::IpcCpyWin2RcvEx(curOutputPtr, dataSize, nullptr, nullptr, opType, dataType, startRank, endRank,
     464              :         maxStreamNum, true);
     465              : 
     466            1 :     TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, true);
     467              : 
     468              :     // 下发sqe
     469            1 :     TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
     470              : 
     471            1 :     HCCL_INFO("run RunAllReduceOneShot1Stream end");
     472            1 :     return HCCL_SUCCESS;
     473              : }
     474              : 
     475            2 : HcclResult AicpuAllreduce::RunAllReduceTwoShot1Stream(HcclReduceOp opType, void *sendBuffer, void *recvBuffer,
     476              :     u64 dataCount, HcclDataType dataType) const
     477              : {
     478            2 :     HCCL_INFO("run RunAllReduceTwoShot1Stream start");
     479            2 :     u32 maxStreamNum = ctx_->rankNum;
     480            2 :     u32 startRank = 0;
     481            2 :     u32 endRank = maxStreamNum - 1;
     482            2 :     u8 *curInputPtr = static_cast<u8 *>(sendBuffer);
     483            2 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     484            2 :     u32 unitSize = ctx_->unitSize;
     485            2 :     u64 inputOffset = 0;
     486            2 :     u64 outputOffset = 0;
     487            2 :     u64 windowSize = ctx_->windowSize; // window size default is 200M, maybe need read from cfg/env.
     488            2 :     u64 maxCountPerLoop = windowSize / unitSize * ctx_->rankNum; // 中转内存单次最多能够接受的output count
     489            2 :     u64 countLeft = dataCount;
     490              : 
     491            2 :     while (countLeft > 0) {
     492            0 :         curInputPtr += inputOffset;
     493            0 :         curOutputPtr += outputOffset;
     494            0 :         u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
     495            0 :         u64 curSize = curCount * unitSize; // 单位 byte
     496            0 :         std::vector<Slice> dataSlice;
     497            0 :         PrepareSlice(curCount, dataType, maxStreamNum, dataSlice);
     498            0 :         u64 sliceSize = dataSlice[ctx_->rankId].size;
     499              : 
     500            0 :         HCCL_INFO("RunAllReducev:curInputPtr[%p], curOutputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr,
     501              :             curOutputPtr, curCount, curSize);
     502              : 
     503            0 :         TaskOrchestrator::SelfCpySnd2WinEx1(curInputPtr, sliceSize, dataSlice[ctx_->rankId].offset, 0,
     504              :             HCCL_REDUCE_RESERVED, dataType, maxStreamNum);
     505              : 
     506            0 :         TaskOrchestrator::IpcPreSyncEx(startRank, endRank, maxStreamNum, true);
     507              : 
     508            0 :         TaskOrchestrator::IpcCpySnd2WinSliceEx(curInputPtr, dataSlice, nullptr, opType, dataType, startRank, endRank,
     509              :             maxStreamNum, true);
     510              : 
     511            0 :         TaskOrchestrator::IpcCpyWin2RcvSliceEx(curOutputPtr, dataSlice, nullptr, HCCL_REDUCE_RESERVED, dataType,
     512              :             startRank, endRank, maxStreamNum, true);
     513              : 
     514            0 :         TaskOrchestrator::IpcPostSyncEx(startRank, endRank, maxStreamNum, true);
     515              : 
     516            0 :         TaskOrchestrator::SelfCpyWin2Rcv(curOutputPtr, sliceSize, 0, dataSlice[ctx_->rankId].offset,
     517              :             HCCL_REDUCE_RESERVED, dataType);
     518              : 
     519              :         // 下发sqe
     520            0 :         TaskOrchestrator::LaunchTasksEx(0, maxStreamNum - 1, maxStreamNum);
     521              : 
     522            0 :         countLeft -= curCount;
     523            0 :         inputOffset = curSize;
     524            0 :         outputOffset = curSize;
     525            0 :     }
     526              : 
     527            2 :     HCCL_INFO("run RunAllReduceTwoShot1Stream end");
     528            2 :     return HCCL_SUCCESS;
     529              : }
     530              : 
     531              : // 计算HD算法给定轮中给定rank的对端的rank号
     532            1 : u32 AicpuAllreduce::GetHdPeer(const u32 hdRound, const u32 curRank) const
     533              : {
     534              :     // 将所有的设备分成若干组,相邻的组之间的对位节点相互通信
     535              :     // 每个组中的设备数量为 2^当前轮次,即:1,2,4,8 ....
     536            1 :     u32 groupSize = std::pow(2, hdRound);
     537              :     // 获取当前节点在组内的位置 (同时也是对端节点在组内的位置)
     538            1 :     u32 rankOffset = curRank % groupSize;
     539              :     // 获取当前节点在第几组
     540            1 :     u32 curGroupIdx = curRank / groupSize;
     541              :     // 偶数号组内的节点与下一组的节点通信,对应的,奇数号组内的节点与上一组的节点通信
     542            1 :     u32 peerGroupIdx = (curGroupIdx % 2 == 0) ? curGroupIdx + 1 : curGroupIdx - 1;
     543              : 
     544            1 :     u32 peerRank = peerGroupIdx * groupSize + rankOffset;
     545            1 :     return peerRank;
     546              : }
     547              : 
     548              : // OneshotHD 算法, 使用了DMA消减
     549              : // 只支持卡数大于 2 且为 2 的幂数的场景
     550              : // datasize 需小于 ccl buffer 大小
     551            1 : HcclResult AicpuAllreduce::RunAllReduceOneshotHD(
     552              :     HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataSize, HcclDataType dataType) const
     553              : {
     554              :     /* 分3个阶段:
     555              :      * 第一阶段:将 input 的数据拷贝到 window 上
     556              :      * 第二阶段:将 input 数据发送到对端 window 进行 reduce 并将 reduce 完的
     557              :      * window 拷贝到 output
     558              :      * 第三阶段:不断将对端 window 数据读取到本端 output,
     559              :      * 如果不是最后一轮,则将 reduce 完的 output 拷贝到 window
     560              :      */
     561            1 :     HCCL_INFO("run RunAllReduceOneshotHD start");
     562              : 
     563            1 :     u8 *curOutputPtr = static_cast<u8 *>(recvBuffer);
     564            1 :     const u32 curRank = ctx_->rankId;
     565              :     // 第一阶段:
     566              :     // 将输入数据拷贝到 Window
     567            1 :     CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
     568              :         sendBuffer, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
     569              :     // 第二阶段:
     570              :     // 片内拷贝 片内Send拷贝到对端Window - 卡内双 die 间 allreduce
     571            1 :     u32 hdRound = 0;
     572            1 :     u32 peerRank = GetHdPeer(hdRound, curRank);
     573            1 :     CHK_RET(TaskOrchestrator::IpcPreSyncEx(peerRank, peerRank, ctx_->rankNum, true));
     574              : 
     575              :     // 将输入数据写到对端
     576            1 :     CHK_RET(TaskOrchestrator::IpcCpySnd2WinP2P(
     577              :         sendBuffer, peerRank, dataSize, 0, 0, opType, dataType));
     578            1 :     TaskOrchestrator::IpcPostSyncEx(peerRank, peerRank, ctx_->rankNum, true);
     579              : 
     580              :     // 片内拷贝 Win拷贝到Rcv
     581            1 :     CHK_RET(TaskOrchestrator::SelfCpyWin2Rcv(
     582              :         curOutputPtr, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
     583              :     // 第三阶段:
     584              :     // 循环 log2(rankNum) - 1 次
     585            1 :     u32 remainingHdRounds = ctx_->rankNum >> 1;
     586            1 :     while (remainingHdRounds >>= 1) { // 使用位移代替 log2
     587            0 :         hdRound++;
     588            0 :         peerRank = GetHdPeer(hdRound, curRank);
     589              :         // 跨片拷贝 对端 window 拷贝到 rcv
     590            0 :         CHK_RET(TaskOrchestrator::IpcPreSyncEx(peerRank, peerRank, ctx_->rankNum, true));
     591            0 :         CHK_RET(TaskOrchestrator::IpcCpyWin2RcvP2PMainStream(
     592              :             curOutputPtr, peerRank, dataSize, 0, 0, opType, dataType));
     593            0 :         CHK_RET(TaskOrchestrator::IpcPostSyncEx(peerRank, peerRank, ctx_->rankNum, true));
     594              :         // 如果这不是最后一轮,则需要将rcv里的数据同步到 win 里
     595            0 :         if (remainingHdRounds > 1) {
     596            0 :             CHK_RET(TaskOrchestrator::SelfCpyRcv2Win(
     597              :                 curOutputPtr, dataSize, 0, 0, HCCL_REDUCE_RESERVED, dataType));
     598              :         }
     599              :     }
     600              :     // 下发sqe
     601            1 :     TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum);
     602            1 :     HCCL_INFO("run RunAllReduceOneshotHD end");
     603            1 :     return HCCL_SUCCESS;
     604              : }
     605              : 
     606              : // 将 oriValue 对调整为 alignValue 的倍数
     607            2 : u64 AicpuAllreduce::AlignWith(u64 oriValue, u64 alignValue) const
     608              : {
     609            2 :     if (oriValue <= alignValue || alignValue == 0) {
     610            2 :         return oriValue;
     611              :     }
     612            0 :     u64 remain = oriValue % alignValue;
     613            0 :     return oriValue - remain;
     614              : }
     615              : 
     616              : // 按照 cclBuffer 大小将数据切分
     617            2 : HcclResult AicpuAllreduce::GetBurstDataCounts(
     618              :     u64 windowSize, u64 dataCount, std::vector<u64> &burstDataCounts) const
     619              : {
     620            2 :     u64 alignedWindowSize = AlignWith(windowSize, HCCL_COPY_ALIGN);
     621            2 :     u32 unitSize = ctx_->unitSize;
     622            2 :     CHK_PRT_RET(unitSize == 0, HCCL_ERROR("UnitSize is 0"), HCCL_E_UNAVAIL);
     623            2 :     u32 maxDataPerBurst = alignedWindowSize / unitSize;
     624            2 :     CHK_PRT_RET(maxDataPerBurst == 0, HCCL_ERROR("maxDataPerBurst is 0"), HCCL_E_UNAVAIL);
     625            0 :     burstDataCounts.insert(burstDataCounts.end(), dataCount / maxDataPerBurst, maxDataPerBurst);
     626            0 :     u64 tailSize = dataCount % maxDataPerBurst;
     627            0 :     if (tailSize != 0) {
     628            0 :         burstDataCounts.push_back(tailSize);
     629              :     }
     630            0 :     return HCCL_SUCCESS;
     631              : }
     632              : 
     633            1 : std::vector<std::vector<u32>> AicpuAllreduce::GetRingOrders() const
     634              : {
     635              :     // simple ring
     636            1 :     std::vector<std::vector<u32>> ringOrders;
     637            1 :     std::vector<u32> ringOrder(ctx_->rankNum);
     638            3 :     for (u32 i = 0; i < ctx_->rankNum; i++) {
     639            2 :         ringOrder[i] = i;
     640              :     }
     641            1 :     ringOrders.push_back(ringOrder);
     642            1 :     return ringOrders;
     643            1 : }
     644              : 
     645              : // 当前只支持从0开始的rankID
     646            1 : HcclResult AicpuAllreduce::reorderRingSlice(const std::vector<u32> &ringOrder, const std::vector<Slice> &ringSlices,
     647              :     std::vector<Slice> &orderedRingSlices) const
     648              : {
     649            1 :     size_t ringSize = ringOrder.size();
     650            1 :     CHK_PRT_RET(ringSize == 0, HCCL_ERROR("ringSize is 0"), HCCL_E_UNAVAIL);
     651            1 :     orderedRingSlices.resize(ringSize);
     652            3 :     for (size_t rankIdx = 0; rankIdx < ringSize; rankIdx++) {
     653            2 :         u32 currentRank = ringOrder[rankIdx];
     654            2 :         u32 previousRankIdx = (rankIdx + ringSize - 1) % ringSize;
     655            2 :         u32 previousRank = ringOrder[previousRankIdx];
     656            2 :         Slice currentSlice = ringSlices[previousRank];
     657            2 :         orderedRingSlices[currentRank] = currentSlice;
     658              :     }
     659            1 :     return HCCL_SUCCESS;
     660              : }
     661              : 
     662            1 : HcclResult AicpuAllreduce::PrepareRingSlice(const std::vector<std::vector<u32>> &ringOrders, u64 dataCount,
     663              :     HcclDataType dataType, std::vector<std::vector<Slice>> &orderedAllRingSlice) const
     664              : {
     665            1 :     u32 ringNum = ringOrders.size();
     666            1 :     u32 sliceNum = ctx_->rankNum * ringNum;
     667            1 :     std::vector<Slice> dataSlices;
     668            1 :     PrepareSlice(dataCount, dataType, sliceNum, dataSlices);
     669            1 :     std::vector<std::vector<Slice>> allRingSlices(ringNum);
     670            1 :     orderedAllRingSlice.resize(ringNum);
     671              : 
     672            3 :     for (size_t i = 0; i < dataSlices.size(); i++) {
     673            2 :         allRingSlices[i % ringNum].push_back(dataSlices[i]);
     674              :     }
     675            2 :     for (size_t i = 0; i < ringNum; i++) {
     676            1 :         reorderRingSlice(ringOrders[i], allRingSlices[i],
     677            1 :                          orderedAllRingSlice[i]);
     678              :     }
     679            1 :     return HCCL_SUCCESS;
     680            1 : }
     681              : 
     682            0 : HcclResult AicpuAllreduce::RingIPCPreSync(const u32 stream, const u32 prevRank, const u32 nextRank) const
     683              : {
     684              :     // 通知下游
     685            0 :     CHK_RET(AicpuDispatcher::SignalRecord(stream, nextRank, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
     686              :     // 等待上游通知
     687            0 :     CHK_RET(AicpuDispatcher::SignalWait(stream, prevRank, AicpuDispatcher::IPC, AicpuDispatcher::PRE_SYNC));
     688            0 :     return HCCL_SUCCESS;
     689              : }
     690              : 
     691            0 : HcclResult AicpuAllreduce::RingIPCPostSync(const u32 stream, const u32 prevRank, const u32 nextRank) const
     692              : {
     693              :     // 回复上游通知
     694            0 :     CHK_RET(AicpuDispatcher::SignalRecord(stream, prevRank, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
     695              :     // 等待下游回复,回收 notify
     696            0 :     CHK_RET(AicpuDispatcher::SignalWait(stream, nextRank, AicpuDispatcher::IPC, AicpuDispatcher::POST_SYNC));
     697            0 :     return HCCL_SUCCESS;
     698              : }
     699              : 
     700            0 : HcclResult AicpuAllreduce::GetPrevRankList(const std::vector<u32> &ringOrder,
     701              :     std::vector<u32> &previousRankList) const
     702              : {
     703            0 :     size_t ringSize = ringOrder.size();
     704            0 :     for (size_t rankIdx = 0; rankIdx < ringSize; rankIdx++) {
     705            0 :         u32 currRank = ringOrder[rankIdx];
     706            0 :         u32 previousRankIdx = (rankIdx + ringSize - 1) % ringSize;
     707            0 :         previousRankList[currRank] = ringOrder[previousRankIdx];
     708              :     }
     709            0 :     return HCCL_SUCCESS;
     710              : }
     711              : 
     712            0 : size_t AicpuAllreduce::FindNextRank(const std::vector<u32> &previousRankList, const u32 localRank) const
     713              : {
     714            0 :     return std::find(previousRankList.begin(), previousRankList.end(),
     715            0 :                      localRank) -
     716            0 :            previousRankList.begin();
     717              : }
     718              : 
     719            0 : Slice* AicpuAllreduce::GetNextRingSlice(const std::vector<u32> &previousRankList,
     720              :     std::vector<Slice> &orderedRingSlices, u32 &curSliceIdx) const
     721              : {
     722            0 :     curSliceIdx = previousRankList[curSliceIdx];
     723            0 :     Slice* nextSlice = &orderedRingSlices[curSliceIdx];
     724            0 :     return nextSlice;
     725              : }
     726              : 
     727            0 : HcclResult AicpuAllreduce::RunAllReduceRingAlg(
     728              :     HcclReduceOp opType, void *sendBuffer, void *recvBuffer, std::vector<Slice> &orderedRingSlices,
     729              :     std::vector<u32> &ringOrder, HcclDataType dataType) const
     730              : {
     731            0 :     size_t ringSize = ringOrder.size();
     732            0 :     std::vector<u32> previousRankList(ringSize);
     733              :     // 计算每一个rank在ring环上的前一个rank
     734            0 :     CHK_RET(GetPrevRankList(ringOrder, previousRankList));
     735            0 :     const u32 prevRank = previousRankList[ctx_->rankId];
     736            0 :     const u32 nextRank = FindNextRank(previousRankList, ctx_->rankId);
     737            0 :     const u32 subStream = prevRank;
     738            0 :     u32 curSliceIdx = ctx_->rankId;
     739            0 :     Slice *localSlice = &orderedRingSlices[curSliceIdx];
     740            0 :     Slice *remoteSlice = nullptr;
     741              :     // 第一轮:主流准备前2片数据
     742            0 :     CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
     743              :         sendBuffer, localSlice->size, localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
     744            0 :     localSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx);
     745            0 :     CHK_RET(TaskOrchestrator::SelfCpySnd2Win(
     746              :         sendBuffer, localSlice->size, localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
     747            0 :     for (size_t rankOffset = 0; rankOffset < ringSize - 1; rankOffset++) {
     748            0 :         remoteSlice = localSlice;
     749              :         // 从流开始 reduce 操作
     750            0 :         CHK_RET(TaskOrchestrator::MainSubPreSync(subStream)); // 主流启动从流
     751              :         // 从流跨片 reduce
     752            0 :         CHK_RET(RingIPCPreSync(subStream, prevRank, nextRank));
     753            0 :         CHK_RET(TaskOrchestrator::IpcCpyWin2WinP2P(prevRank, remoteSlice->size, remoteSlice->offset,
     754              :             remoteSlice->offset, opType, dataType));
     755            0 :         CHK_RET(RingIPCPostSync(subStream, prevRank, nextRank));
     756            0 :         if (rankOffset <  ringSize - 2) { // ringSize - 2: 最后一轮不进行本地搬运
     757              :             // 主流继续准备数据
     758            0 :             localSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx);
     759            0 :             CHK_RET(TaskOrchestrator::SelfCpySnd2Win(sendBuffer, localSlice->size,
     760              :                 localSlice->offset, localSlice->offset, HCCL_REDUCE_RESERVED, dataType));
     761              :         }
     762            0 :         CHK_RET(TaskOrchestrator::MainSubPostSync(subStream)); // 从流通知主流,回收 notify,主流继续执行
     763            0 :         TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum); // 下发sqe
     764              :     }
     765            0 :     for (size_t rankOffset = 0; rankOffset < ringSize - 1; rankOffset++) {
     766            0 :         localSlice = remoteSlice; // 主流搬运上一轮从流准备的数据
     767            0 :         remoteSlice = GetNextRingSlice(previousRankList, orderedRingSlices, curSliceIdx); // 从流继续往下循环,搬运reduce好的数据
     768            0 :         CHK_RET(TaskOrchestrator::MainSubPreSync(subStream)); // 主流通知从流回收notify资源
     769            0 :         TaskOrchestrator::SelfCpyWin2Rcv(recvBuffer, localSlice->size, localSlice->offset,
     770              :             localSlice->offset, HCCL_REDUCE_RESERVED, dataType);
     771            0 :         CHK_RET(RingIPCPreSync(subStream, prevRank, nextRank));
     772            0 :         if (rankOffset < ringSize - 2) { // < RingSize - 2: 不是最后一轮,搬到 window 上,让下游读
     773            0 :             CHK_RET(TaskOrchestrator::IpcCpyWin2WinP2P(prevRank, remoteSlice->size, remoteSlice->offset,
     774              :                 remoteSlice->offset, HCCL_REDUCE_RESERVED, dataType));
     775              :         } else {  // 最后一轮,dma 消减,直接从对端读入 rcv
     776            0 :             CHK_RET(TaskOrchestrator::IpcCpyWin2RcvP2P(recvBuffer, prevRank, remoteSlice->size,
     777              :                 remoteSlice->offset, remoteSlice->offset, HCCL_REDUCE_RESERVED, dataType));
     778              :         }
     779            0 :         CHK_RET(RingIPCPostSync(subStream, prevRank, nextRank));
     780            0 :         CHK_RET(TaskOrchestrator::MainSubPostSync(subStream)); // 通知主流继续
     781            0 :         TaskOrchestrator::LaunchTasksEx(0, ctx_->rankNum - 1, ctx_->rankNum); // 下发sqe
     782              :     }
     783            0 :     return HCCL_SUCCESS;
     784            0 : }
     785              : 
     786            0 : HcclResult AicpuAllreduce::RunAllReduceRingSingleBurst(
     787              :     HcclReduceOp opType, void *sendBuffer, void *recvBuffer, u64 dataCount,
     788              :     HcclDataType dataType, std::vector<std::vector<u32>> &ringOrders) const
     789              : {
     790            0 :     std::vector<std::vector<Slice>> orderedRingSlices;
     791            0 :     PrepareRingSlice(ringOrders, dataCount, dataType, orderedRingSlices);
     792            0 :     for (size_t i = 0; i < ringOrders.size(); i++) {
     793            0 :         CHK_RET(RunAllReduceRingAlg(opType, sendBuffer, recvBuffer, orderedRingSlices[i],
     794              :             ringOrders[i], dataType));
     795              :     }
     796            0 :     return HCCL_SUCCESS;
     797            0 : }
     798              : 
     799              : // 单 ring 算法, 使用了DMA消减
     800            2 : HcclResult AicpuAllreduce::RunAllReduceRing(HcclReduceOp opType, void *sendBuffer,
     801              :     void *recvBuffer, u64 dataCount, HcclDataType dataType) const
     802              : {
     803            2 :     HCCL_INFO("run RunAllReduceRing start");
     804              :     // 数据准备阶段
     805            2 :     std::vector<u64> burstDataCounts;
     806            2 :     u64 windowSize = ctx_->windowSize;
     807            2 :     u32 unitSize = ctx_->unitSize;
     808            2 :     CHK_RET(GetBurstDataCounts(windowSize, dataCount, burstDataCounts));
     809            0 :     u8 *currSendBuffer = static_cast<u8 *>(sendBuffer);
     810            0 :     u8 *currRecvBuffer = static_cast<u8 *>(recvBuffer);
     811              :     u64 burstSize;
     812            0 :     std::vector<std::vector<u32>> ringOrders = GetRingOrders();
     813            0 :     for (u64 burstDataCount : burstDataCounts) {
     814            0 :         burstSize = burstDataCount * unitSize;
     815            0 :         CHK_RET(RunAllReduceRingSingleBurst(
     816              :             opType, currSendBuffer, currRecvBuffer, burstDataCount, dataType, ringOrders));
     817            0 :         currSendBuffer += burstSize;
     818            0 :         currRecvBuffer += burstSize;
     819              :     }
     820            0 :     HCCL_INFO("run RunAllReduceRing end");
     821            0 :     return HCCL_SUCCESS;
     822            2 : }
        

Generated by: LCOV version 2.0-1