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_ins_preprocessor.h"
12 : #include "null_ptr_exception.h"
13 : #include "orion_adapter_rts.h"
14 : #include "stl_util.h"
15 : #include "aicpu_res_package_helper.h"
16 : #include "alg_topo_package_helper.h"
17 :
18 : namespace Hccl {
19 :
20 : constexpr u8 QUEUE_NOTIFY_POST_QID_POS = 0;
21 : constexpr u8 QUEUE_NOTIFY_WAIT_QID_POS = 1;
22 : constexpr u8 QUEUE_NOTIFY_TOPIC_ID_POS = 2;
23 :
24 0 : void AicpuInsPreprocessor::Preprocess(std::shared_ptr<InsQueue> &insQueue)
25 : {
26 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] insQueue Preprocess start.", __func__);
27 :
28 : // 对每个queue中每个aicpuIns进行预处理
29 0 : for (auto slaveIter = insQueue->IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
30 0 : for (auto ins = slaveIter->Iter(); ins.HasNext(); ++ins) {
31 0 : if (ins->GetType() != InstructionType::AICPU_INS) { // todo:InstructionType
32 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] slave insQueue ins type[%s] not aicpu type.", __func__,
33 : ins->GetType().Describe().c_str());
34 0 : continue;
35 0 : }
36 0 : InsPreprocess(ins);
37 0 : }
38 0 : }
39 :
40 : // 对每主queue中每个aicpuIns进行预处理
41 0 : for (auto ins = insQueue->Iter(); ins.HasNext(); ++ins) {
42 0 : if (ins->GetType() != InstructionType::AICPU_INS) {
43 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] master insQueue ins type[%s] not aicpu type.", __func__,
44 : ins->GetType().Describe().c_str());
45 0 : continue;
46 0 : }
47 0 : InsPreprocess(ins);
48 0 : }
49 :
50 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] insQueue Preprocess end.", __func__);
51 0 : }
52 :
53 2 : bool AicpuInsPreprocessor::IsAicpuResExisted(const std::string &algName)
54 : {
55 2 : if (aicpuResExistedMap.find(algName) == aicpuResExistedMap.end()
56 2 : || aicpuResMap.find(algName) == aicpuResMap.end()) {
57 1 : THROW<NullPtrException>(
58 3 : StringFormat("[AicpuInsPreprocessor::%s] aicpuRes for algName[%s] is not exited on aicpuResExistedMap.",
59 : __func__, algName.c_str()));
60 : }
61 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end, aicpuResExisted [%d].", __func__, aicpuResExistedMap[algName]);
62 1 : return aicpuResExistedMap[algName];
63 : }
64 :
65 2 : DevBuffer *AicpuInsPreprocessor::GetAicpuResBuffer(const std::string &algName)
66 : {
67 6 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
68 :
69 2 : if (aicpuResMap.find(algName) == aicpuResMap.end()) {
70 1 : THROW<NullPtrException>(
71 3 : StringFormat("[AicpuInsPreprocessor::%s] aicpuRes for algName[%s] is not exited on device buffer.",
72 : __func__, algName.c_str()));
73 : }
74 :
75 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
76 1 : return aicpuResMap[algName].get();
77 : }
78 :
79 0 : void AicpuInsPreprocessor::InsPreprocess(InsIterator &insIter)
80 : {
81 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
82 :
83 0 : const AicpuInstruction &aicpuIns = dynamic_cast<const AicpuInstruction &>(*insIter);
84 :
85 0 : CollAlgResReq collAlgResReq = aicpuIns.GetCollAlgResReq();
86 0 : AllocWorkStream(collAlgResReq.primQueueNum);
87 0 : AllocQueueNotify(collAlgResReq.queueNotifys);
88 0 : AllocBcastPostCntNotify(collAlgResReq.localBcastPostCntNotify);
89 0 : AllocWaitGroupCntNotify(collAlgResReq.localWaitGroupCntNotify);
90 0 : AllocInterRankNotifies(collAlgResReq.links);
91 :
92 : // 创建MemTransport并建链、交换
93 0 : BatchBuildTransports(collAlgResReq.links);
94 :
95 0 : std::string algName = aicpuIns.GetAlgName();
96 0 : if (aicpuResMap.find(algName) != aicpuResMap.end()) { // 已经向Device Mem写过资源
97 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] aicpuRes for algName[%s] has existed.", __func__, algName.c_str());
98 0 : return;
99 : }
100 :
101 0 : PackResAndCopyToDev(algName, collAlgResReq);
102 :
103 0 : AllocAlltoallVOpMem();
104 :
105 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
106 0 : }
107 :
108 0 : void AicpuInsPreprocessor::AllocWorkStream(u32 workStreamNum) const
109 : {
110 0 : comm->GetAicpuStreamManager().AllocStreams(workStreamNum);
111 0 : }
112 :
113 0 : void AicpuInsPreprocessor::AllocQueueNotify(std::vector<std::tuple<QId, QId, u32>> &queueNotifyReq) const
114 : {
115 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
116 :
117 0 : QueueNotifyManager &queueNotifyMgr = comm->GetAicpuQueueNotifyManager();
118 :
119 0 : std::for_each(queueNotifyReq.begin(), queueNotifyReq.end(), [&queueNotifyMgr](auto item) {
120 0 : queueNotifyMgr.ApplyFor(std::get<QUEUE_NOTIFY_POST_QID_POS>(item), std::get<QUEUE_NOTIFY_WAIT_QID_POS>(item),
121 0 : std::get<QUEUE_NOTIFY_TOPIC_ID_POS>(item));
122 0 : });
123 :
124 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
125 0 : }
126 :
127 0 : void AicpuInsPreprocessor::AllocBcastPostCntNotify(std::vector<std::pair<QId, u32>> &bcastPostCntNotifyReq) const
128 : {
129 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
130 :
131 0 : QueueBcastPostCntNotifyManager &bcastPostCntNotifyMgr = comm->GetBcastPostCntNotifyManager();
132 :
133 0 : std::for_each(bcastPostCntNotifyReq.begin(), bcastPostCntNotifyReq.end(), [&bcastPostCntNotifyMgr](auto item) {
134 0 : bcastPostCntNotifyMgr.ApplyFor(item.first, item.second);
135 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] qid[%u] topicId[%u]", __func__, item.first, item.second);
136 0 : });
137 :
138 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
139 0 : }
140 :
141 0 : void AicpuInsPreprocessor::AllocWaitGroupCntNotify(std::vector<std::pair<QId, u32>> &waitGroupCntNotifyReq) const
142 : {
143 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
144 :
145 0 : QueueWaitGroupCntNotifyManager &waitGroupCntNotifyMgr = comm->GetQueueWaitGroupCntNotifyManager();
146 :
147 0 : std::for_each(waitGroupCntNotifyReq.begin(), waitGroupCntNotifyReq.end(), [&waitGroupCntNotifyMgr](auto item) {
148 0 : waitGroupCntNotifyMgr.ApplyFor(item.first, item.second);
149 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] qid[%u] topicId[%u]", __func__, item.first, item.second);
150 0 : });
151 :
152 0 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
153 0 : }
154 :
155 1 : void AicpuInsPreprocessor::AllocInterRankNotifies(const vector<LinkData> &links)
156 : {
157 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
158 :
159 1 : vector<LinkData> pendingLinks;
160 2 : for (auto &link : links) {
161 1 : if (Contain(availableLinks, link)) {
162 0 : continue;
163 : }
164 1 : pendingLinks.emplace_back(link);
165 : }
166 :
167 1 : if (pendingLinks.empty()) {
168 0 : return;
169 : }
170 :
171 2 : for (auto &link : pendingLinks) {
172 : // 待修改: 申请数量
173 1 : comm->GetConnLocalNotifyManager().ApplyFor(link.GetRemoteRankId(), link);
174 : }
175 :
176 1 : availableLinks.insert(pendingLinks.begin(), pendingLinks.end());
177 :
178 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
179 1 : }
180 :
181 2 : void AicpuInsPreprocessor::BatchBuildTransports(const vector<LinkData> &links)
182 : {
183 6 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
184 :
185 2 : std::string opTag = comm->GetCurrentCollOperator()->opTag;
186 :
187 : // 创建RmaConnectiuon
188 2 : auto connBuilderPair = connectionsBuilders.emplace(opTag, make_unique<ConnectionsBuilder>(*comm));
189 2 : connBuilderPair.first->second->BatchBuild(opTag, links);
190 :
191 : // 创建MemTransport并进行异步建链、交换
192 2 : auto op = comm->GetCurrentCollOperator();
193 2 : if (op->opMode == OpMode::OPBASE) {
194 1 : comm->GetMemTransportManager()->BatchBuildOpbasedTransports(links);
195 1 : } else if (op->opMode == OpMode::OFFLOAD) {
196 1 : comm->GetMemTransportManager()->BatchBuildOffloadTransports(opTag, links);
197 : }
198 :
199 : // 等待异步建链完成
200 2 : comm->GetCollService()->WaitTransportReady(opTag);
201 :
202 6 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
203 2 : }
204 :
205 0 : static void SetModuleName(ModuleData &module, const std::string &name)
206 : {
207 0 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
208 0 : if (ret != 0) {
209 0 : THROW<InternalException>(StringFormat("strcpy_s name %s failed", name.c_str()));
210 : }
211 0 : }
212 :
213 0 : std::vector<char> AicpuInsPreprocessor::PackOpData(const std::string &opTag, const std::string &algName,
214 : const CollAlgResReq &resReq)
215 : {
216 0 : std::vector<ModuleData> dataVec;
217 0 : dataVec.resize(AicpuResMgrType::__COUNT__);
218 :
219 0 : AicpuResMgrType resType = AicpuResMgrType::STREAM;
220 0 : SetModuleName(dataVec[resType], "StreamManager");
221 0 : dataVec[resType].data = comm->GetAicpuStreamManager().GetPackedData();
222 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
223 :
224 0 : resType = AicpuResMgrType::QUEUE_NOTIFY;
225 0 : SetModuleName(dataVec[resType], "QueueNotifyManager");
226 0 : dataVec[resType].data = comm->GetAicpuQueueNotifyManager().GetPackedData();
227 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
228 :
229 0 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
230 0 : SetModuleName(dataVec[resType], "QueueWaitGroupCntNotifyManager");
231 0 : dataVec[resType].data = comm->GetQueueWaitGroupCntNotifyManager().GetPackedData();
232 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
233 :
234 0 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
235 0 : SetModuleName(dataVec[resType], "GetBcastPostCntNotifyManager");
236 0 : dataVec[resType].data = comm->GetBcastPostCntNotifyManager().GetPackedData();
237 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
238 :
239 0 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
240 0 : SetModuleName(dataVec[resType], "HostDeviceSyncNotifyManager");
241 0 : dataVec[resType].data = comm->GetHostDeviceSyncNotifyManager().GetPackedData();
242 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
243 :
244 0 : resType = AicpuResMgrType::TRANSPORT;
245 0 : SetModuleName(dataVec[resType], "MemTransportManager");
246 0 : auto op = comm->GetCurrentCollOperator();
247 0 : if (op->opMode == OpMode::OPBASE) { // 单算子模式
248 0 : dataVec[resType].data = comm->GetMemTransportManager()->GetOpbasedPackedData();
249 0 : } else if (op->opMode == OpMode::OFFLOAD) { // 图下沉模式
250 0 : dataVec[resType].data = comm->GetMemTransportManager()->GetOffloadPackedData(opTag);
251 : } else {
252 0 : THROW<InternalException>(StringFormat("opMode=%s failed", op->opMode.Describe().c_str()));
253 : }
254 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
255 :
256 0 : resType = AicpuResMgrType::ALG_TOPO;
257 0 : SetModuleName(dataVec[resType], algName);
258 : AlgTopoPackageHelper algTopoHelper;
259 0 : dataVec[resType].data = algTopoHelper.GetPackedData(resReq.topoInfo);
260 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
261 :
262 0 : resType = AicpuResMgrType::CONNECTD_MGR;
263 0 : SetModuleName(dataVec[resType], "ConnectedManager");
264 0 : dataVec[resType].data = comm->GetRankGraph()->GetPackedData(resReq.levelRankPairs);
265 0 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
266 :
267 : AicpuResPackageHelper helper;
268 0 : return helper.GetPackedData(dataVec);
269 0 : }
270 :
271 1 : void AicpuInsPreprocessor::PackResAndCopyToDev(const std::string &algName, const CollAlgResReq &resReq)
272 : {
273 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
274 :
275 1 : std::string opTag = comm->GetCurrentCollOperator()->opTag;
276 1 : auto buffer = PackOpData(opTag, algName, resReq);
277 1 : shared_ptr<DevBuffer> devMem = make_shared<DevBuffer>(buffer.size()); // 申请device内存
278 1 : HrtMemcpy(reinterpret_cast<void *>(devMem->GetAddr()), devMem->GetSize(), buffer.data(), buffer.size(),
279 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到device内存
280 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] PackedData %s", __func__, Bytes2hex(buffer.data(), buffer.size()).c_str());
281 :
282 1 : aicpuResMap.insert(std::make_pair(algName, devMem));
283 1 : aicpuResExistedMap.insert(std::make_pair(algName, false));
284 :
285 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
286 1 : }
287 :
288 1 : void AicpuInsPreprocessor::AllocAlltoallVOpMem()
289 : {
290 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
291 :
292 1 : auto op = comm->GetCurrentCollOperator();
293 1 : if (op->opType != OpType::ALLTOALLV) {
294 0 : HCCL_INFO("[AllocAlltoallVOpMem] op->opType[%d]", op->opType);
295 0 : return;
296 : }
297 :
298 1 : size_t size = static_cast<size_t>(comm->GetRankSize() * sizeof(u64)); // counts内存大小
299 1 : if (!isCountMemInited) {
300 65 : for (u32 i = 0; i < MAX_ALLTOALLV_MEM_NUM; i++) { // 64: 初始化countMem
301 64 : shared_ptr<DevBuffer> sendMem = make_shared<DevBuffer>(size); // 申请senddevice内存
302 64 : sendCountsMem.push_back(sendMem);
303 :
304 64 : shared_ptr<DevBuffer> recvMem = make_shared<DevBuffer>(size); // 申请recvdevice内存
305 64 : recvCountsMem.push_back(recvMem);
306 :
307 64 : shared_ptr<DevBuffer> sdisplMem = make_shared<DevBuffer>(size); // 申请sdisplsdevice内存
308 64 : sdisplsMem.push_back(sdisplMem);
309 :
310 64 : shared_ptr<DevBuffer> rdisplMem = make_shared<DevBuffer>(size); // 申请rdisplsdevice内存
311 64 : rdisplsMem.push_back(rdisplMem);
312 64 : }
313 1 : isCountMemInited = true;
314 : }
315 :
316 1 : HrtMemcpy(reinterpret_cast<void *>(sendCountsMem[resIndex].get()->GetAddr()),
317 1 : sendCountsMem[resIndex].get()->GetSize(), op->all2AllVDataDes.sendCounts, size,
318 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到SEND内存
319 1 : HrtMemcpy(reinterpret_cast<void *>(recvCountsMem[resIndex].get()->GetAddr()),
320 1 : recvCountsMem[resIndex].get()->GetSize(), op->all2AllVDataDes.recvCounts, size,
321 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到RECV内存
322 1 : HrtMemcpy(reinterpret_cast<void *>(sdisplsMem[resIndex].get()->GetAddr()), sdisplsMem[resIndex].get()->GetSize(),
323 1 : op->all2AllVDataDes.sdispls, size,
324 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到SDISPLS内存
325 1 : HrtMemcpy(reinterpret_cast<void *>(rdisplsMem[resIndex].get()->GetAddr()), rdisplsMem[resIndex].get()->GetSize(),
326 1 : op->all2AllVDataDes.rdispls, size,
327 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到RDISPLS内存
328 :
329 1 : resIndex++;
330 1 : if (resIndex >= MAX_ALLTOALLV_MEM_NUM) { // MAX_ALLTOALLV_MEM_NUM: 初始化countMem
331 0 : resIndex = 0;
332 : }
333 :
334 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
335 : }
336 :
337 1 : void AicpuInsPreprocessor::SetAicpuKernelLaunchParam(HcclKernelLaunchParam ¶m)
338 : {
339 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] start.", __func__);
340 :
341 1 : auto op = comm->GetCurrentCollOperator();
342 1 : if (op->opType != OpType::ALLTOALLV) {
343 0 : HCCL_INFO("[SetAicpuKernelLaunchParam] op->opType[%d]", op->opType);
344 0 : return;
345 : }
346 :
347 : param.kernel.op.algOperator.all2AllVDataDes.sendCounts
348 1 : = reinterpret_cast<void *>(sendCountsMem[launchResIndex].get()->GetAddr());
349 : param.kernel.op.algOperator.all2AllVDataDes.recvCounts
350 1 : = reinterpret_cast<void *>(recvCountsMem[launchResIndex].get()->GetAddr());
351 : param.kernel.op.algOperator.all2AllVDataDes.sdispls
352 1 : = reinterpret_cast<void *>(sdisplsMem[launchResIndex].get()->GetAddr());
353 : param.kernel.op.algOperator.all2AllVDataDes.rdispls
354 1 : = reinterpret_cast<void *>(rdisplsMem[launchResIndex].get()->GetAddr());
355 1 : param.kernel.op.algOperator.all2AllVDataDes.sendType = op->all2AllVDataDes.sendType;
356 1 : param.kernel.op.algOperator.all2AllVDataDes.recvType = op->all2AllVDataDes.recvType;
357 :
358 3 : HCCL_INFO("AicpuKernelLauncher::SetHcclKernelLaunchParam param.kernel.op.algOperator.sendCounts[%p] "
359 : "param.kernel.op.algOperator.recvCounts[%p] param.kernel.op.algOperator.sdispls[%p] "
360 : "param.kernel.op.algOperator.rdispls[%p], launchResIndex[%u]",
361 : param.kernel.op.algOperator.all2AllVDataDes.sendCounts,
362 : param.kernel.op.algOperator.all2AllVDataDes.recvCounts,
363 : param.kernel.op.algOperator.all2AllVDataDes.sdispls, param.kernel.op.algOperator.all2AllVDataDes.rdispls,
364 : launchResIndex);
365 :
366 1 : launchResIndex++;
367 1 : if (launchResIndex >= MAX_ALLTOALLV_MEM_NUM) { // MAX_ALLTOALLV_MEM_NUM: 初始化countMem
368 0 : launchResIndex = 0;
369 : }
370 :
371 3 : HCCL_INFO("[AicpuInsPreprocessor::%s] end.", __func__);
372 : }
373 :
374 2 : void AicpuInsPreprocessor::SetAicpuResExisted(const std::string &algName)
375 : {
376 2 : if (aicpuResExistedMap.find(algName) == aicpuResExistedMap.end()) {
377 1 : THROW<NullPtrException>(
378 2 : StringFormat("[AicpuInsPreprocessor::%s] aicpuRes for algName[%s] is not exited on aicpuResExistedMap.",
379 : __func__, algName.c_str()));
380 : }
381 1 : aicpuResExistedMap[algName] = true;
382 1 : }
383 :
384 : } // namespace Hccl
|
