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 "log.h"
12 :
13 : #include "ins_coll_alg_registry.h"
14 : #include "ins_v2_send_executor.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : #include "hccl_aiv_utils.h"
18 : #include "aiv_ins.h"
19 : #include "executor_utils.h"
20 :
21 : using namespace std;
22 :
23 : namespace Hccl {
24 0 : InsV2SendExecutor::InsV2SendExecutor() : InsCollAlgBase() {}
25 :
26 0 : InsV2SendExecutor::~InsV2SendExecutor() {}
27 :
28 : HcclResult
29 0 : InsV2SendExecutor::CalcResOffload(const RankGraph* rankGraph, const u64& dataSize, CollOffloadOpResReq& resReq)
30 : {
31 : (void)rankGraph;
32 : (void)dataSize;
33 : (void)resReq;
34 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2SendExecutor][CalcResOffload] offload is not support");
35 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
36 : }
37 :
38 0 : HcclResult InsV2SendExecutor::CalNumBlocks(u32& numBlocks, u64 dataSize, u32 numBlocksLimit)
39 : {
40 : (void)dataSize;
41 :
42 0 : if (numBlocksLimit < 1) {
43 0 : HCCL_ERROR("[InsV2SendExecutor] core num[%u] is less than 1", numBlocksLimit);
44 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
45 : }
46 :
47 0 : numBlocks = numBlocksLimit;
48 0 : HCCL_INFO("[InsV2SendExecutor] Actually use core num[%u]", numBlocks);
49 :
50 0 : return HcclResult::HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult InsV2SendExecutor::CalcRes(const RankGraph* rankGraph, CollAlgResReq& algResReq)
54 : {
55 0 : u32 linkNumBtwPeers = 1;
56 0 : algResReq.primQueueNum = 1;
57 0 : AlgTempResReq tempResReq;
58 0 : tempResReq.queNum = 1;
59 0 : tempResReq.streamNum = tempResReq.queNum;
60 0 : if (static_cast<u32>(sendRecvRemoteRank_) > rankSize_ - 1) {
61 0 : HCCL_ERROR(
62 : "[InsCollAlgFactory][InsV2SendExecutor][CalcRes] Rank[%d] get dest[%d] is invalid", myRank_,
63 : sendRecvRemoteRank_);
64 0 : return HcclResult::HCCL_E_PARA;
65 : }
66 0 : tempResReq.links[sendRecvRemoteRank_] = linkNumBtwPeers;
67 0 : uint32_t linkNum = GetPathsFromRankGraph(rankGraph, myRank_, sendRecvRemoteRank_).size();
68 0 : if (linkNum == 0) {
69 0 : HCCL_ERROR(
70 : "[InsCollAlgFactory][InsV2SendExecutor][CalcRes] Rank[%d] get path num to dest[%d] is zero", myRank_,
71 : sendRecvRemoteRank_);
72 : }
73 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
74 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
75 0 : return HcclResult::HCCL_SUCCESS;
76 0 : }
77 :
78 : // host
79 0 : HcclResult InsV2SendExecutor::Orchestrate(
80 : const RankGraph* rankGraph, const CollAlgOperator& op, const CollAlgParams& params, InsQuePtr insQue)
81 : {
82 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Begin to Generate Instruction Queue for SEND.");
83 0 : CHK_RET(Init(op, params, insQue));
84 : // 从集合通信算子op中获得remote rank和data type/count相关信息
85 0 : RankId remoteRank = op.sendRecvRemoteRank;
86 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
87 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
88 :
89 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
90 0 : if (myRank_ == remoteRank) {
91 0 : HCCL_WARNING("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Self send, Self recv, Do nothing");
92 0 : return HcclResult::HCCL_SUCCESS;
93 : }
94 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Self send, Other recv");
95 :
96 : // 从rankGraph里面拿到 link，转成LinkData格式
97 0 : const std::vector<NetInstance::Path> sendPath = GetPathsFromRankGraph(rankGraph, myRank_, remoteRank);
98 0 : CHK_PRT_RET(
99 : sendPath.size() == 0,
100 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
101 : HcclResult::HCCL_E_INTERNAL);
102 0 : LinkData sendLinkData(sendPath[0]);
103 0 : HCCL_DEBUG(
104 : "[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
105 : "[%u].",
106 : totalDataSize, params.maxTmpMemSize);
107 :
108 : // 模式判断
109 0 : HCCL_DEBUG("[InsCollAlgFactory] opmode_ is [%d], ", opMode_);
110 0 : if (opMode_ == OpMode::OFFLOAD) {
111 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] offload is not support");
112 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
113 : }
114 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
115 :
116 0 : CHK_RET(ExecAiv(op, params, sendLinkData, insQue));
117 0 : return HcclResult::HCCL_SUCCESS;
118 0 : }
119 :
120 : // aicpu
121 0 : HcclResult InsV2SendExecutor::Orchestrate(
122 : const AlgTopoInfo& topoInfo, const CollAlgOperator& op, const CollAlgParams& params, ConnectedLinkMgr* linkMgr,
123 : InsQuePtr insQue)
124 : {
125 : (void)topoInfo;
126 0 : HCCL_DEBUG(
127 : "[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Begin to Generate Instruction Queue for SEND AICPU mode.");
128 0 : CHK_RET(Init(op, params, insQue));
129 : // 从集合通信算子op中获得remote rank和data type/count相关信息
130 0 : RankId remoteRank = op.sendRecvRemoteRank;
131 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
132 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
133 :
134 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
135 0 : if (myRank_ == remoteRank) {
136 0 : HCCL_WARNING("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Self send, Self recv, Do nothing");
137 0 : return HcclResult::HCCL_SUCCESS;
138 : }
139 0 : HCCL_DEBUG("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Self send, Other recv.");
140 :
141 : // 从linkMgr里面拿到 linkData
142 0 : const vector<LinkData> sendPath = linkMgr->GetLinks(remoteRank);
143 0 : CHK_PRT_RET(
144 : sendPath.size() == 0,
145 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_, remoteRank),
146 : HcclResult::HCCL_E_INTERNAL);
147 0 : LinkData sendLinkData(sendPath[0]);
148 0 : HCCL_DEBUG(
149 : "[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size "
150 : "[%u].",
151 : totalDataSize, params.maxTmpMemSize);
152 :
153 : // 模式判断
154 0 : if (opMode_ == OpMode::OFFLOAD) {
155 0 : HCCL_ERROR("[InsCollAlgFactory][InsV2SendExecutor][Orchestrate] offload is not support");
156 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
157 : }
158 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
159 :
160 0 : CHK_RET(ExecAiv(op, params, sendLinkData, insQue));
161 0 : return HcclResult::HCCL_SUCCESS;
162 0 : }
163 :
164 0 : HcclResult InsV2SendExecutor::ExecAiv(
165 : const CollAlgOperator& op, const CollAlgParams& params, LinkData& sendLinkData, InsQuePtr insQue)
166 : {
167 0 : HCCL_INFO(
168 : "[InsV2SendExecutor][ExecAiv] start: rank is %d, count is %u, dataType is %u, destRank is %d", myRank_,
169 : op.dataCount, static_cast<u32>(op.dataType), op.sendRecvRemoteRank);
170 :
171 0 : u64 transportBoundDataSize = UB_MAX_DATA_SIZE;
172 0 : u64 maxScratchDataSize = std::min(transportBoundDataSize, params.maxTmpMemSize);
173 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
174 0 : u64 maxScratchDataCount = maxScratchDataSize / dataElemSize;
175 0 : CHK_PRT_RET(
176 : maxScratchDataCount == 0, HCCL_ERROR("[InsV2SendExecutor][Orchestrate] maxScratchDataCount is 0"),
177 : HCCL_E_INTERNAL);
178 :
179 0 : std::vector<LinkData> allLinks;
180 0 : allLinks.emplace_back(sendLinkData);
181 :
182 0 : u64 loopTimes = op.dataCount / maxScratchDataCount + static_cast<u64>(op.dataCount % maxScratchDataCount != 0);
183 0 : u64 processedDataCount = 0;
184 0 : for (u64 loop = 0; loop < loopTimes; loop++) {
185 0 : sliceId_++; // 自动增长sliceId，传入aivTag
186 0 : u64 currDataCount = (loop == loopTimes - 1) ? op.dataCount - processedDataCount : maxScratchDataCount;
187 0 : HCCL_INFO(
188 : "[InsV2SendExecutor][ExecAiv] myRank[%d], loop[%llu] sliceId_[%llu] currDataCount[%llu], "
189 : "processedDataCount[%llu]",
190 : myRank_, loop, sliceId_, currDataCount, processedDataCount);
191 :
192 0 : AivOpArgs aivSendArgs;
193 0 : aivSendArgs.cmdType = HcclCMDType::HCCL_CMD_SEND;
194 0 : aivSendArgs.input = processedDataCount * dataElemSize;
195 0 : aivSendArgs.output = 0;
196 0 : aivSendArgs.rank = u32(myRank_);
197 0 : aivSendArgs.sendRecvRemoteRank = op.sendRecvRemoteRank;
198 0 : aivSendArgs.rankSize = rankSize_;
199 0 : aivSendArgs.count = currDataCount; // 需要传输的数据量
200 0 : aivSendArgs.dataType = op.dataType;
201 0 : aivSendArgs.aivTag = sliceId_; // 传入aivTag，Lauch时重新组装为aivTag
202 0 : aivSendArgs.isOpBase = (opMode_ == OpMode::OPBASE);
203 0 : aivSendArgs.xRankSize = rankSize_;
204 0 : aivSendArgs.yRankSize = 0;
205 0 : aivSendArgs.zRankSize = 0;
206 0 : CHK_RET(CalNumBlocks(aivSendArgs.numBlocks, 0, op.numBlocksLimit));
207 :
208 0 : aivSendArgs.inputSliceStride = 0;
209 0 : aivSendArgs.outputSliceStride = 0;
210 0 : aivSendArgs.repeatNum = 1; // 不重复
211 0 : aivSendArgs.inputRepeatStride = 0;
212 0 : aivSendArgs.outputRepeatStride = 0;
213 :
214 0 : std::unique_ptr<Instruction> aivInsSendMesh1D = std::make_unique<AivInstruction>(allLinks, aivSendArgs);
215 :
216 0 : insQue->Append(std::move(aivInsSendMesh1D));
217 0 : processedDataCount += currDataCount;
218 0 : }
219 :
220 0 : HCCL_INFO("[InsV2SendExecutor][ExecAiv] end: rank[%d]", myRank_);
221 0 : return HcclResult::HCCL_SUCCESS;
222 0 : }
223 :
224 : // 注册
225 : INS_REGISTER_IMPL(OpType::SEND, AivSend, InsV2SendExecutor);
226 :
227 : } // namespace Hccl
|
