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