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_send_executor.h"
15 : #include "alg_data_trans_wrapper.h"
16 :
17 : using namespace std;
18 :
19 : namespace Hccl {
20 0 : InsSendExecutor::InsSendExecutor() : InsCollAlgBase()
21 : {
22 0 : }
23 :
24 0 : InsSendExecutor::~InsSendExecutor()
25 : {
26 0 : }
27 :
28 0 : HcclResult InsSendExecutor::Orchestrate(const RankGraph *rankGraph,
29 : const CollAlgOperator &op,
30 : const CollAlgParams ¶ms,
31 : InsQuePtr insQue)
32 : {
33 0 : dmaMode_ = DmaMode::DEFAULT;
34 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Begin to Generate Instruction Queue for SEND.");
35 0 : CHK_RET(Init(op, params, insQue));
36 : // 从集合通信算子op中获得remote rank和data type/count相关信息
37 0 : RankId remoteRank = op.sendRecvRemoteRank;
38 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
39 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
40 0 : if (totalDataSize == 0) {
41 0 : HCCL_WARNING("[InsCollAlgFactory][InsSendExecutor][Orchestrate] totalDataSize is 0, do nothing.");
42 0 : return HcclResult::HCCL_SUCCESS;
43 : }
44 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
45 0 : if (myRank_ == remoteRank) {
46 0 : HCCL_WARNING("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Self send, Self recv, Do nothing");
47 0 : return HcclResult::HCCL_SUCCESS;
48 : }
49 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Self send, Other recv");
50 :
51 : // 从rankGraph里面拿到 link，转成LinkData格式
52 0 : const std::vector<NetInstance::Path> sendPath = GetPathsFromRankGraph(rankGraph, myRank_, remoteRank);
53 0 : CHK_PRT_RET(sendPath.size() == 0,
54 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_,
55 : remoteRank), HcclResult::HCCL_E_INTERNAL);
56 0 : LinkData sendLinkData(sendPath[0]);
57 0 : if (sendLinkData.GetType() == PortDeploymentType::P2P
58 0 : && sendLinkData.GetLinkProtocol() == LinkProtocol::PCIE) {
59 0 : dmaMode_ = DmaMode::GET;
60 : }
61 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size [%u].", totalDataSize, params.maxTmpMemSize);
62 :
63 : // 初始化循环参数
64 0 : u64 resDataSize = totalDataSize;
65 0 : u64 currentOffset = 0;
66 0 : u32 roundIdx = 0;
67 : // 模式判断
68 0 : HCCL_DEBUG("[InsCollAlgFactory] opmode_ is [%d], ", opMode_);
69 0 : if (opMode_ == OpMode::OFFLOAD) {
70 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
71 0 : u64 transferSize = resDataSize;
72 : // 根据本轮数据搬运量声明相关DataSlice--图模式一次搬运全部数据 从inputbuffer直接send到对面outputbuffer
73 0 : DataSlice inputBuffer(BufferType::INPUT, currentOffset, transferSize);
74 0 : DataSlice remoteOutputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
75 0 : SlicesList sendSlicesList({inputBuffer}, {remoteOutputBuffer});
76 0 : DataInfo sendInfo(sendLinkData, sendSlicesList);
77 0 : CHK_RET(Send(sendInfo, insQue, 0, true, dmaMode_));
78 0 : }else{
79 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
80 : // 当需要多轮搬运时，需保证一次数据的搬运量需为单个数据size的整数倍
81 0 : u64 maxRoundTransferSize = params.maxTmpMemSize - params.maxTmpMemSize % dataElemSize;
82 0 : while(resDataSize > 0) {
83 : // 判断本轮需搬运的数据量
84 0 : u64 transferSize = resDataSize > params.maxTmpMemSize ? maxRoundTransferSize : resDataSize;
85 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Send round [%u], transfer data size [%llu].", roundIdx, transferSize);
86 : // 根据本轮数据搬运量声明相关DataSlice
87 0 : DataSlice inputBuffer(BufferType::INPUT, currentOffset, transferSize);
88 0 : DataSlice scratchBuffer(BufferType::SCRATCH, 0, transferSize);
89 0 : DataSlice remoteScratchBuffer(BufferType::SCRATCH, 0, transferSize);
90 : // local copy
91 0 : CHK_RET(LocalCopy(insQue, inputBuffer, scratchBuffer));
92 0 : SlicesList sendSlicesList({scratchBuffer}, {remoteScratchBuffer});
93 0 : DataInfo sendInfo(sendLinkData, sendSlicesList);
94 0 : CHK_RET(Send(sendInfo, insQue, 0, true, dmaMode_));
95 :
96 0 : currentOffset = currentOffset + transferSize;
97 0 : resDataSize = resDataSize - transferSize;
98 0 : roundIdx = roundIdx + 1;
99 0 : }
100 : }
101 0 : return HcclResult::HCCL_SUCCESS;
102 0 : }
103 :
104 0 : HcclResult InsSendExecutor::CalcResOffload(const RankGraph *rankGraph,
105 : const u64 &dataSize,
106 : CollOffloadOpResReq &resReq)
107 : {
108 : (void)rankGraph;
109 : (void)dataSize;
110 0 : resReq.requiredScratchMemSize = 0; //图模式不用scratchmemory
111 0 : resReq.requiredSubQueNum = 0;
112 :
113 0 : return HcclResult::HCCL_SUCCESS;
114 : }
115 :
116 0 : HcclResult InsSendExecutor::CalcRes(const RankGraph *rankGraph,
117 : CollAlgResReq &algResReq)
118 : {
119 0 : u32 linkNumBtwPeers = 1;
120 0 : algResReq.primQueueNum = 1;
121 0 : AlgTempResReq tempResReq;
122 0 : tempResReq.queNum = 1;
123 0 : tempResReq.streamNum = tempResReq.streamNum;
124 0 : if (static_cast<u32>(sendRecvRemoteRank_) > rankSize_ - 1) {
125 0 : HCCL_ERROR("[InsCollAlgFactory][InsSendExecutor][CalcRes] Rank[%d] get dest[%d] is invalid", myRank_, sendRecvRemoteRank_);
126 0 : return HcclResult::HCCL_E_PARA;
127 : }
128 0 : tempResReq.links[sendRecvRemoteRank_] = linkNumBtwPeers;
129 0 : uint32_t linkNum = GetPathsFromRankGraph(rankGraph, myRank_, sendRecvRemoteRank_).size();
130 0 : if (linkNum == 0) {
131 0 : HCCL_ERROR("[InsCollAlgFactory][InsSendExecutor][CalcRes] Rank[%d] get path num to dest[%d] is zero", myRank_, sendRecvRemoteRank_);
132 : }
133 0 : CHK_RET(CalcResLinks(myRank_, rankGraph, linkPriority_, tempResReq.links, algResReq.links));
134 0 : CHK_RET(CalcLinkInfo(myRank_, rankGraph, tempResReq.links, algResReq.levelRankPairs));
135 0 : return HcclResult::HCCL_SUCCESS;
136 0 : }
137 :
138 0 : HcclResult InsSendExecutor::Orchestrate(const AlgTopoInfo &topoInfo, // aicpu
139 : const CollAlgOperator &op,
140 : const CollAlgParams ¶ms,
141 : ConnectedLinkMgr *linkMgr,
142 : InsQuePtr insQue)
143 : {
144 0 : dmaMode_ = DmaMode::DEFAULT;
145 : (void)topoInfo;
146 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Begin to Generate Instruction Queue for SEND AICPU mode.");
147 0 : CHK_RET(Init(op, params, insQue));
148 : // 从集合通信算子op中获得remote rank和data type/count相关信息
149 0 : RankId remoteRank = op.sendRecvRemoteRank;
150 0 : u32 dataElemSize = DATA_TYPE_SIZE_MAP.at(op.dataType);
151 0 : u64 totalDataSize = static_cast<u64>(dataElemSize) * op.dataCount;
152 0 : if (totalDataSize == 0) {
153 0 : HCCL_WARNING("[InsCollAlgFactory][InsSendExecutor][Orchestrate] totalDataSize is 0, do nothing.");
154 0 : return HcclResult::HCCL_SUCCESS;
155 : }
156 : // 判断是不是自发自收这种情况，若是，则什么都不做，直接返回
157 0 : if (myRank_ == remoteRank) {
158 0 : HCCL_WARNING("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Self send, Self recv, Do nothing");
159 0 : return HcclResult::HCCL_SUCCESS;
160 : }
161 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Self send, Other recv.");
162 :
163 : // 从linkMgr里面拿到 linkData
164 0 : const vector<LinkData> sendPath = linkMgr->GetLinks(remoteRank);
165 0 : CHK_PRT_RET(sendPath.size() == 0,
166 : HCCL_ERROR("[InsCollAlgFactory] Unable to obtain valid link, srcRank [%d], dstRank [%d].", myRank_,
167 : remoteRank), HcclResult::HCCL_E_INTERNAL);
168 0 : LinkData sendLinkData(sendPath[0]);
169 0 : if (sendLinkData.GetType() == PortDeploymentType::P2P
170 0 : && sendLinkData.GetLinkProtocol() == LinkProtocol::PCIE) {
171 0 : dmaMode_ = DmaMode::GET;
172 : }
173 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Total transfer data size [%llu], Max scratch buffer size [%u].", totalDataSize, params.maxTmpMemSize);
174 :
175 : // 初始化循环参数
176 0 : u64 resDataSize = totalDataSize;
177 0 : u64 currentOffset = 0;
178 0 : u32 roundIdx = 0;
179 : // 模式判断
180 0 : if (opMode_ == OpMode::OFFLOAD) {
181 0 : u64 maxLoopOutputSize = 256 * 1024 * 1024; // 256m为一轮
182 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OFFLOAD Mode for HOST.", myRank_);
183 0 : while (resDataSize > 0) {
184 0 : u64 transferSize = resDataSize > maxLoopOutputSize ? maxLoopOutputSize : resDataSize;
185 0 : DataSlice inputBuffer(BufferType::INPUT, currentOffset, transferSize);
186 0 : DataSlice remoteOutputBuffer(BufferType::OUTPUT, currentOffset, transferSize);
187 0 : SlicesList sendSlicesList({inputBuffer}, {remoteOutputBuffer});
188 0 : DataInfo sendInfo(sendLinkData, sendSlicesList);
189 0 : CHK_RET(Send(sendInfo, insQue, 0, true, dmaMode_));
190 0 : currentOffset = currentOffset + transferSize;
191 0 : resDataSize = resDataSize - transferSize;
192 0 : roundIdx = roundIdx + 1;
193 0 : }
194 : } else {
195 0 : HCCL_DEBUG("[InsCollAlgFactory] Rank[%d], Generating Instruction Queues in OPBASE Mode for HOST.", myRank_);
196 : // 当需要多轮搬运时，需保证一次数据的搬运量需为单个数据size的整数倍
197 0 : u64 maxRoundTransferSize = params.maxTmpMemSize - params.maxTmpMemSize % dataElemSize;
198 0 : while(resDataSize > 0) {
199 : // 判断本轮需搬运的数据量
200 0 : u64 transferSize = resDataSize > params.maxTmpMemSize ? maxRoundTransferSize : resDataSize;
201 0 : HCCL_DEBUG("[InsCollAlgFactory][InsSendExecutor][Orchestrate] Send round [%u], transfer data size [%llu]", roundIdx, transferSize);
202 : // 根据本轮数据搬运量声明相关DataSlice
203 0 : DataSlice inputBuffer(BufferType::INPUT, currentOffset, transferSize);
204 0 : DataSlice scratchBuffer(BufferType::SCRATCH, 0, transferSize);
205 0 : DataSlice remoteScratchBuffer(BufferType::SCRATCH, 0, transferSize);
206 : // local copy
207 0 : CHK_RET(LocalCopy(insQue, inputBuffer, scratchBuffer));
208 0 : SlicesList sendSlicesList({scratchBuffer}, {remoteScratchBuffer});
209 0 : DataInfo sendInfo(sendLinkData, sendSlicesList);
210 0 : CHK_RET(Send(sendInfo, insQue, 0, true, dmaMode_));
211 :
212 0 : currentOffset = currentOffset + transferSize;
213 0 : resDataSize = resDataSize - transferSize;
214 0 : roundIdx = roundIdx + 1;
215 0 : }
216 : }
217 0 : return HcclResult::HCCL_SUCCESS;
218 0 : }
219 :
220 : // 注册
221 : INS_REGISTER_IMPL(OpType::SEND, InsSend, InsSendExecutor);
222 :
223 : } // namespace Hccl
|
