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