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 : #include "coll_service_ai_cpu_impl.h"
11 : #include <memory>
12 : #include "communicator_impl.h"
13 : #include "internal_exception.h"
14 : #include "env_config_v2.h"
15 : #include "stl_util.h"
16 : #include "stream_utils.h"
17 :
18 : #include "aicpu_res_package_helper.h"
19 : #include "alg_topo_package_helper.h"
20 : #include "dlprof_function_v2.h"
21 : #include "task_exception_handler.h"
22 : #include "exception_util.h"
23 : #include "runtime_api_exception.h"
24 :
25 : namespace Hccl {
26 :
27 : template <class T, class U>
28 : u16 CalcFieldOffset(T* target, U* base)
29 : {
30 : return static_cast<u16>(reinterpret_cast<const char*>(target) - reinterpret_cast<const char*>(base));
31 : }
32 :
33 193 : void CollServiceAiCpuImpl::Init()
34 : {
35 : // 算子执行次数计数器buffer申请
36 193 : AddOpCounterMems();
37 193 : }
38 :
39 3 : static std::string GetTagKey(CollOperator& op, std::string algName, u32 bsrRemoteRanksHashValue)
40 : {
41 3 : std::string tmp{};
42 3 : tmp = (op.opMode == OpMode::OPBASE) ? algName : op.opTag;
43 3 : if (op.opType == OpType::BATCHSENDRECV) {
44 0 : tmp = tmp + std::to_string(bsrRemoteRanksHashValue);
45 3 : } else if (op.opType == OpType::SEND || op.opType == OpType::RECV) {
46 0 : tmp = tmp + std::to_string(op.sendRecvRemoteRank);
47 : }
48 3 : return tmp;
49 0 : }
50 :
51 3 : DevBuffer* CollServiceAiCpuImpl::OpBasedCollProcess(CollOperator& op, const std::string& algName)
52 : {
53 3 : auto req = comm->GetCollAlgComponent()->GetCollAlgOpReq(op, algName);
54 9 : HCCL_INFO(
55 : "CollServiceAiCpuImpl::OpBasedCollProcess GetCollAlgOpReq OrchestMode::INSTRUCTION, algName %s",
56 : req.algName.c_str());
57 :
58 3 : for (auto& it : req.resReq.levelRankPairs) {
59 0 : HCCL_INFO("CollServiceAiCpuImpl::level=%u, rank=%d", it.first, it.second);
60 : }
61 :
62 9 : HCCL_INFO("CollServiceAiCpuImpl::OpBasedCollProcess req.resReq.primQueueNum =%u", req.resReq.primQueueNum);
63 9 : HCCL_INFO(
64 : "CollServiceAiCpuImpl::OpBasedCollProcess req.resReq.queueNotifys size =%zu", req.resReq.queueNotifys.size());
65 9 : HCCL_INFO(
66 : "CollServiceAiCpuImpl::OpBasedCollProcess req.resReq.localBcastPostCntNotify size =%zu",
67 : req.resReq.localBcastPostCntNotify.size());
68 9 : HCCL_INFO(
69 : "CollServiceAiCpuImpl::OpBasedCollProcess req.resReq.localWaitGroupCntNotify size =%zu",
70 : req.resReq.localWaitGroupCntNotify.size());
71 :
72 3 : AllocWorkStream(req.resReq.primQueueNum);
73 3 : AllocQueueNotify(req.resReq.queueNotifys);
74 3 : AllocBcastPostCntNotify(req.resReq.localBcastPostCntNotify);
75 3 : AllocWaitGroupCntNotify(req.resReq.localWaitGroupCntNotify);
76 :
77 9 : HCCL_INFO("CollServiceAiCpuImpl::OpBasedCollProcess req.resReq.links size =%zu", req.resReq.links.size());
78 9 : HCCL_INFO("CollServiceAiCpuImpl::OpBasedCollProcess op.opTag %s", op.opTag.c_str());
79 :
80 3 : RegisterCclBuffer(req.resReq.links);
81 :
82 : // Socket建链
83 3 : comm->GetSocketManager().BatchCreateSockets(req.resReq.links);
84 : // 建立RmaConnection并建链
85 3 : auto connBuilderPair = connectionsBuilders.emplace(comm->GetId(), make_unique<ConnectionsBuilder>(*comm));
86 3 : connBuilderPair.first->second->BatchBuild(comm->GetId(), req.resReq.links);
87 :
88 3 : AllocNotifies(req.resReq.links);
89 :
90 3 : if (op.opMode == OpMode::OPBASE) {
91 0 : comm->GetMemTransportManager()->BatchBuildOpbasedTransports(req.resReq.links);
92 0 : WaitOpbasedTransportReady();
93 3 : } else if (op.opMode == OpMode::OFFLOAD) {
94 3 : comm->GetMemTransportManager()->BatchBuildOffloadTransports(op.opTag, req.resReq.links);
95 3 : WaitOffloadTransportReady(op.opTag);
96 : }
97 :
98 3 : u32 bsrRemoteRanksHashValue = 0;
99 3 : if (op.opType == OpType::BATCHSENDRECV) {
100 0 : bsrRemoteRanksHashValue = GetRemoteRankIdsHashValue(op);
101 : }
102 :
103 3 : curTagKey = GetTagKey(op, req.algName, bsrRemoteRanksHashValue);
104 :
105 3 : auto it = collOpLoadedMap.find(curTagKey);
106 3 : if (it != collOpLoadedMap.end()) { // 已经向Device Mem写过资源
107 3 : HCCL_INFO("[OpBasedCollProcess] tag[%s] devMem has been allocated, reuse it", curTagKey.c_str());
108 1 : return it->second.get();
109 : }
110 :
111 2 : auto buffer = PackOpData(op.opTag, req);
112 2 : shared_ptr<DevBuffer> devMem = make_shared<DevBuffer>(buffer.size()); // 申请device内存
113 2 : TRY_CATCH_THROW(
114 : InternalException, StringFormat("HrtMemcpy to device memory failed, size[%zu]", buffer.size()),
115 : HrtMemcpy(
116 : reinterpret_cast<void*>(devMem->GetAddr()), devMem->GetSize(), buffer.data(), buffer.size(),
117 : RT_MEMCPY_HOST_TO_DEVICE)); // H2D拷贝，将资源拷贝到device内存
118 :
119 2 : collOpLoadedMap[curTagKey] = devMem;
120 2 : return devMem.get();
121 3 : }
122 :
123 2 : void CollServiceAiCpuImpl::LoadWithOpBasedModeNoRegister(CollOperator& op)
124 : {
125 2 : RegisterOpbasedLocalRmaBuf(op.opTag);
126 :
127 2 : comm->GetAicpuStreamManager().AllocFreeStream();
128 2 : Stream* lanchStream = comm->GetAicpuStreamManager().GetFreeStream();
129 2 : comm->GetAicpuStreamManager().AclGraphCaptureFreeStream(comm->GetStreamManager().opbase->GetMaster());
130 2 : DevBuffer* mem = nullptr;
131 2 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
132 2 : mem = OpBasedCollProcess(op, comm->GetCurAlgName());
133 2 : std::vector<Stream*>& stream_pointers = comm->GetAicpuStreamManager().GetStreams();
134 2 : comm->ReportHcclMC2Info(
135 2 : *lanchStream, *comm->GetStreamManager().opbase->GetMaster(), stream_pointers); // 上报MC2信息
136 2 : AllocOpMem(op);
137 :
138 2 : SaveMirrorDfxOpInfo();
139 2 : AicpuKernelEntranceLaunch(*comm->GetStreamManager().opbase->GetMaster(), op, comm->GetCurAlgName(), mem);
140 2 : }
141 :
142 2 : void CollServiceAiCpuImpl::LoadWithOpBasedMode(CollOperator& op, unique_ptr<Stream> stream)
143 : {
144 2 : RegisterOpBufToBufMgr(op);
145 2 : RegisterOpbasedStream(std::move(stream));
146 :
147 2 : LoadWithOpBasedModeNoRegister(op);
148 2 : }
149 :
150 1 : void CollServiceAiCpuImpl::LoadWithOffloadModeNoRegister(CollOperator& op)
151 : {
152 1 : RegisterOffloadLocalRmaBuf(op.opTag);
153 :
154 : // 将通讯域设置为transport建链中状态
155 1 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
156 :
157 1 : DevBuffer* mem = nullptr;
158 1 : mem = OpBasedCollProcess(op, comm->GetCurAlgName());
159 1 : AllocOpMem(op);
160 :
161 1 : SaveMirrorDfxOpInfo();
162 1 : AicpuKernelEntranceLaunch(*comm->GetStreamManager().offload->GetMaster(op.opTag), op, comm->GetCurAlgName(), mem);
163 1 : }
164 :
165 2 : void CollServiceAiCpuImpl::LoadWithOffloadMode(CollOperator& op, std::unique_ptr<Stream> stream)
166 : {
167 2 : RegisterOpBufToBufMgr(op);
168 2 : RegisterOffloadMasterStream(op.opTag, std::move(stream));
169 :
170 1 : LoadWithOffloadModeNoRegister(op);
171 1 : }
172 :
173 : HcclResult
174 2 : CollServiceAiCpuImpl::AllocCollOpResourceNoRegister(CollOperator& op, const std::string& opAlgTag, void** addr)
175 : {
176 2 : RegisterOpbasedLocalRmaBuf(op.opTag);
177 2 : comm->GetAicpuStreamManager().AllocFreeStream();
178 2 : DevBuffer* mem = nullptr;
179 2 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
180 2 : mem = OpBasedCollProcess(op, comm->GetCurAlgName());
181 2 : CHK_RET(AicpuMc2CommResourcePrepare(op, comm->GetCurAlgName(), mem, opAlgTag, addr));
182 2 : return HCCL_SUCCESS;
183 : }
184 :
185 4 : HcclResult CollServiceAiCpuImpl::AllocCollOpResource(CollOperator& op, const std::string& opAlgTag, void** addr)
186 : {
187 4 : auto iter = aicpuMc2CommResourceMap_.find(opAlgTag);
188 4 : if (iter != aicpuMc2CommResourceMap_.end()) {
189 3 : HCCL_INFO("[AllocCollOpResource] has existed comm %s, return for %p.", opAlgTag.c_str(), addr);
190 1 : shared_ptr<DevBuffer> oldDevBuf = iter->second;
191 1 : *addr = reinterpret_cast<void*>(oldDevBuf->GetAddr());
192 1 : return HCCL_SUCCESS;
193 1 : }
194 3 : RegisterOpBufToBufMgr(op);
195 3 : CHK_RET(AllocCollOpResourceNoRegister(op, opAlgTag, addr));
196 3 : return HCCL_SUCCESS;
197 : }
198 :
199 2 : HcclResult CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare(
200 : const CollOperator& op, const string& algName, const DevBuffer* mem, const std::string& opAlgTag, void** addr)
201 : {
202 6 : HCCL_INFO(
203 : "CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare entry, algName: %s, opAlgTag: %s", algName.c_str(),
204 : opAlgTag.c_str());
205 2 : HcclKernelLaunchParam param{};
206 2 : s32 ret = strcpy_s(param.kernel.algName, sizeof(param.kernel.algName), algName.data());
207 2 : if (ret != EOK) {
208 0 : HCCL_ERROR(
209 : "CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare, strcpy_s algName failed! ret: %d, algName: %s", ret,
210 : algName.c_str());
211 0 : return HCCL_E_INTERNAL;
212 : }
213 :
214 2 : ret = strcpy_s(param.kernel.opTag, sizeof(param.kernel.opTag), op.opTag.data());
215 2 : if (ret != EOK) {
216 0 : HCCL_ERROR(
217 : "CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare, strcpy_s opTag failed! ret: %d, op.opTag: %s", ret,
218 : op.opTag.c_str());
219 0 : return HCCL_E_INTERNAL;
220 : }
221 :
222 6 : HCCL_INFO(
223 : "CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare param.kernel.algName: %s, op.opTag: %s",
224 : param.kernel.algName, op.opTag.c_str());
225 2 : param.kernel.binaryResAddr = mem->GetAddr();
226 2 : param.kernel.binaryResSize = mem->GetSize();
227 :
228 2 : SetHcclKernelLaunchParam(param, comm, false);
229 :
230 2 : comm->SetAicpuKernelLaunched(true);
231 2 : comm->GetStreamManager().ResetSlaveIndex(0);
232 :
233 2 : shared_ptr<DevBuffer> newDevMem = make_shared<DevBuffer>(sizeof(HcclKernelParamLite));
234 2 : HrtMemcpy(
235 2 : reinterpret_cast<void*>(newDevMem->GetAddr()), sizeof(HcclKernelParamLite),
236 : reinterpret_cast<void*>(¶m.kernel), sizeof(HcclKernelParamLite), RT_MEMCPY_HOST_TO_DEVICE);
237 2 : aicpuMc2CommResourceMap_.insert(make_pair(opAlgTag, newDevMem));
238 2 : *addr = reinterpret_cast<void*>(newDevMem->GetAddr());
239 6 : HCCL_INFO(
240 : "CollServiceAiCpuImpl::AicpuMc2CommResourcePrepare alloc %s kernel param success, set addr %p value is %p",
241 : opAlgTag.c_str(), addr, *addr);
242 2 : return HCCL_SUCCESS;
243 2 : }
244 :
245 14 : void InitAicpuLocBufLite(HcclAicpuLocBufLite& lite, u64 addr, u64 size, const string& desc)
246 : {
247 14 : auto tokenPair = HrtUbDevQueryToken(addr, size);
248 14 : lite.addr = addr;
249 14 : lite.size = size;
250 14 : lite.tokenId = tokenPair.first;
251 14 : lite.tokenValue = tokenPair.second;
252 42 : HCCL_INFO("InitAicpuLocBufLite %s, addr=0x%llx, size=0x%llx", desc.c_str(), addr, size);
253 14 : }
254 :
255 3 : void CollServiceAiCpuImpl::SetOpbaseBufferParam(
256 : HcclKernelLaunchParam& param, CommunicatorImpl* comm, CollOperator& op) const
257 : {
258 3 : auto buffer = comm->GetCclBuffer();
259 3 : param.kernel.comm.opBaseScratch.addr = buffer->GetAddr();
260 3 : param.kernel.comm.opBaseScratch.size = buffer->GetSize();
261 3 : InitAicpuLocBufLite(param.kernel.comm.opBaseScratch, buffer->GetAddr(), buffer->GetSize(), "opBaseScratch");
262 3 : if (op.inputMem != nullptr) {
263 6 : InitAicpuLocBufLite(param.kernel.op.input, op.inputMem->GetAddr(), op.inputMem->GetSize(), "inputMem");
264 : }
265 :
266 3 : if (op.outputMem != nullptr) {
267 6 : InitAicpuLocBufLite(param.kernel.op.output, op.outputMem->GetAddr(), op.outputMem->GetSize(), "outputMem");
268 : }
269 9 : HCCL_INFO(
270 : "SetOpbaseBufferParam param.kernel.comm.opBaseScratch.addr %llu, param.kernel.comm.opBaseScratch.size %llu",
271 : param.kernel.comm.opBaseScratch.addr, param.kernel.comm.opBaseScratch.size);
272 3 : }
273 :
274 2 : void CollServiceAiCpuImpl::SetOffloadBufferParam(
275 : HcclKernelLaunchParam& param, CommunicatorImpl* comm, CollOperator& op) const
276 : {
277 6 : HCCL_INFO("SetOffloadBufferParam");
278 2 : auto offloadInput = comm->GetDataBufferManager().Get(op.opTag, BufferType::INPUT);
279 2 : if (offloadInput != nullptr) {
280 0 : InitAicpuLocBufLite(param.kernel.op.input, op.inputMem->GetAddr(), op.inputMem->GetSize(), "inputMem");
281 : }
282 :
283 2 : auto offloadOuput = comm->GetDataBufferManager().Get(op.opTag, BufferType::OUTPUT);
284 2 : if (offloadOuput != nullptr) {
285 0 : InitAicpuLocBufLite(param.kernel.op.output, op.outputMem->GetAddr(), op.outputMem->GetSize(), "outputMem");
286 : }
287 :
288 2 : auto offloadScartch = comm->GetDataBufferManager().Get(op.opTag, BufferType::SCRATCH);
289 2 : if (offloadScartch != nullptr) {
290 2 : InitAicpuLocBufLite(param.kernel.op.scratch, op.scratchMem->GetAddr(), op.scratchMem->GetSize(), "scratchMem");
291 : }
292 2 : }
293 :
294 5 : void CollServiceAiCpuImpl::SetHcclKernelLaunchParam(HcclKernelLaunchParam& param, CommunicatorImpl* comm, bool isLaunch)
295 : {
296 5 : CollOperator op = *comm->GetCurrentCollOperator();
297 :
298 5 : param.kernel.comm.idIndex = comm->GetIdIndex();
299 5 : param.kernel.comm.myRank = comm->GetMyRank();
300 5 : param.kernel.comm.rankSize = comm->GetRankSize();
301 5 : param.kernel.comm.devType = comm->GetDevType();
302 5 : param.kernel.comm.devPhyId = comm->GetDevicePhyId();
303 5 : param.kernel.comm.opIndex_ = comm->GetOpIndex();
304 5 : param.kernel.comm.opCounterAddr = static_cast<u64>(counterBuf->GetAddr());
305 5 : auto ret = strcpy_s(param.kernel.comm.commId, sizeof(param.kernel.comm.commId), comm->GetId().data());
306 5 : if (ret != EOK) {
307 0 : THROW<InternalException>(
308 0 : StringFormat("CollServiceAiCpuImpl::SetHcclKernelLaunchParam, strcpy_s commId failed! ret[%d]", ret));
309 : }
310 5 : if (op.opMode == OpMode::OPBASE) {
311 3 : SetOpbaseBufferParam(param, comm, op);
312 : } else {
313 2 : SetOffloadBufferParam(param, comm, op);
314 : }
315 :
316 5 : param.kernel.op.algOperator.opMode = op.opMode;
317 5 : param.kernel.op.algOperator.opType = op.opType;
318 5 : param.kernel.op.algOperator.reduceOp = op.reduceOp;
319 5 : param.kernel.op.algOperator.dataType = op.dataType;
320 5 : param.kernel.op.algOperator.dataCount = op.dataCount;
321 5 : param.kernel.op.algOperator.root = op.root;
322 5 : HcclResult hcclRet = HCCL_SUCCESS;
323 5 : if (op.opType == OpType::ALLTOALL && isLaunch) {
324 0 : param.kernel.op.algOperator.all2AllDataDes = op.all2AllDataDes;
325 5 : } else if (op.opType == OpType::ALLTOALLV && isLaunch) {
326 1 : hcclRet = FillAllToAllvData(op);
327 4 : } else if (op.opType == OpType::ALLTOALLVC) {
328 1 : hcclRet = FillAllToAllvcData(op);
329 3 : } else if (op.opType == OpType::BATCHSENDRECV) {
330 1 : hcclRet = FillBatchSendRecvData(op);
331 2 : } else if (op.opType == OpType::SEND || op.opType == OpType::RECV) {
332 0 : param.kernel.op.algOperator.sendRecvRemoteRank = op.sendRecvRemoteRank;
333 : }
334 5 : if (hcclRet != HCCL_SUCCESS) {
335 0 : HCCL_ERROR("[CollServiceAiCpuImpl][SetHcclKernelLaunchParam] fill op data failed!");
336 0 : THROW<InternalException>(
337 0 : StringFormat("CollServiceAiCpuImpl::SetHcclKernelLaunchParam, fill op data failed! ret [%d]", hcclRet));
338 : }
339 5 : param.kernel.op.sendRecvRemoteRank = op.sendRecvRemoteRank;
340 5 : Stream* streamPtr = nullptr;
341 5 : if (op.opMode == OpMode::OPBASE) {
342 3 : streamPtr = comm->GetStreamManager().opbase->GetMaster();
343 : } else {
344 2 : streamPtr = comm->GetStreamManager().offload->GetMaster(op.opTag);
345 : }
346 5 : if (streamPtr != nullptr) {
347 1 : param.kernel.op.userStreamId = streamPtr->GetId();
348 : } else {
349 12 : HCCL_WARNING(
350 : "CollServiceAiCpuImpl::%s userStream is nullptr, userStreamId id in kernel param is invalid.", __func__);
351 : }
352 5 : param.kernel.kfcControlTransferH2DParams = comm->GetKfcControlTransferH2D().GetCommunicateParams();
353 5 : param.kernel.kfcControlTransferD2HParams = comm->GetKfcStatusTransferD2H().GetCommunicateParams();
354 :
355 5 : SetDeviceEnvConfigParam(param);
356 5 : }
357 :
358 5 : void CollServiceAiCpuImpl::SetDeviceEnvConfigParam(HcclKernelLaunchParam& param) const
359 : {
360 5 : param.kernel.envConfig.hcclExecTimeout = EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
361 : param.kernel.envConfig.taskExceptionEnable
362 5 : = EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().taskExceptionEnable;
363 5 : }
364 :
365 3 : void CollServiceAiCpuImpl::AicpuKernelEntranceLaunch(
366 : Stream& stream, const CollOperator& op, const string& algName, const DevBuffer* mem)
367 : {
368 3 : HcclKernelLaunchParam param;
369 :
370 3 : s32 ret = strcpy_s(param.kernel.algName, sizeof(param.kernel.algName), algName.data());
371 3 : if (ret != EOK) {
372 0 : THROW<InternalException>(
373 0 : StringFormat("CollServiceAiCpuImpl::AicpuKernelEntranceLaunch, strcpy_s algName failed! ret[%d]", ret));
374 : }
375 :
376 3 : ret = strcpy_s(param.kernel.opTag, sizeof(param.kernel.opTag), op.opTag.data());
377 3 : if (ret != EOK) {
378 0 : THROW<InternalException>(
379 0 : StringFormat("CollServiceAiCpuImpl::AicpuKernelEntranceLaunch, strcpy_s opTag failed! ret[%d]", ret));
380 : }
381 :
382 3 : ret = strcpy_s(param.kernel.tagKey, sizeof(param.kernel.tagKey), curTagKey.data());
383 3 : if (ret != EOK) {
384 0 : THROW<InternalException>(
385 0 : StringFormat("CollServiceAiCpuImpl::AicpuKernelEntranceLaunch, strcpy_s tagKey failed! ret[%d]", ret));
386 : }
387 :
388 9 : HCCL_INFO(
389 : "CollServiceAiCpuImpl::AicpuKernelEntranceLaunch param.kernel.algName: %s, op.opTag %s", param.kernel.algName,
390 : op.opTag.c_str());
391 :
392 3 : param.kernel.binaryResAddr = mem->GetAddr();
393 3 : param.kernel.binaryResSize = mem->GetSize();
394 :
395 3 : SetHcclKernelLaunchParam(param, comm);
396 3 : AicpuKernelLaunch(param, stream, op.opMode);
397 3 : comm->SetAicpuKernelLaunched(true);
398 :
399 3 : comm->GetStreamManager().ResetSlaveIndex(0);
400 3 : }
401 :
402 4 : void CollServiceAiCpuImpl::AicpuKernelLaunch(HcclKernelLaunchParam& param, Stream& stream, OpMode opMode)
403 : {
404 4 : param.kernel.op.userStreamId = stream.GetId();
405 : aclrtLaunchKernelCfg cfg;
406 : aclrtLaunchKernelAttr attr;
407 4 : attr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
408 4 : auto timeoutCheck = EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
409 : attr.value.timeout
410 4 : = static_cast<u16>((timeoutCheck == 0) ? timeoutCheck : (timeoutCheck + 30)); // aicpu kernal超时时间: X+30s
411 4 : cfg.numAttrs = 1;
412 4 : cfg.attrs = &attr;
413 4 : TRY_CATCH_THROW(
414 : InternalException, std::string("HrtMemcpy kernelParamBuf failed"),
415 : HrtMemcpy(
416 : reinterpret_cast<void*>(kernelParamBuf_.get()->GetAddr()), sizeof(HcclKernelParamLite), ¶m,
417 : sizeof(HcclKernelParamLite), RT_MEMCPY_HOST_TO_HOST));
418 :
419 12 : HCCL_INFO("[CollServiceAiCpuImpl][%s] args timeout[%u]s", __func__, attr.value.timeout);
420 :
421 4 : AddPostToUserStream(stream);
422 4 : TaskParam taskParam{};
423 4 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
424 :
425 12 : HCCL_INFO(
426 : "[CollServiceAiCpuImpl::AicpuKernelLaunch] RegisterGetAicpuTaskExceptionCallBack streamId[%u], devLogicId[%u]",
427 : stream.GetId(), comm->GetDeviceLogicId());
428 0 : auto getAicpuTaskExceptionCallBack = [this]() {
429 0 : return this->comm->GetAicpuTaskException();
430 4 : };
431 8 : Hccl::RegisterGetAicpuTaskExceptionCallBackV2(
432 4 : stream.GetId(), comm->GetDeviceLogicId(), getAicpuTaskExceptionCallBack);
433 :
434 12 : HCCL_INFO(
435 : "[CollServiceAiCpuImpl][%s] param.soName: %s, param.kernelName: %s", __func__, param.soName, param.kernelName);
436 4 : const aclrtFuncHandle funcHandle = comm->GetAicpuKernelFuncHandle(param.kernelName);
437 :
438 4 : bool isCapture = false;
439 4 : rtModel_t rtModel = nullptr;
440 4 : CHK_RET_THROW(
441 : RuntimeApiException,
442 : StringFormat("[CollServiceAiCpuImpl][%s] GetStreamCaptureInfo fail, streamId[%u]", __func__, stream.GetId()),
443 : GetStreamCaptureInfo(stream.GetPtr(), rtModel, isCapture));
444 4 : Stream* mStreamPtr = nullptr;
445 4 : if (opMode == OpMode::OPBASE || isCapture) {
446 1 : comm->GetAicpuStreamManager().AllocFreeStream();
447 1 : mStreamPtr = comm->GetAicpuStreamManager().GetFreeStream();
448 1 : comm->GetAicpuStreamManager().AclGraphCaptureFreeStream(&stream);
449 : } else {
450 3 : mStreamPtr = &stream;
451 : }
452 4 : auto& mStream = *mStreamPtr;
453 :
454 4 : std::string mode = (opMode == OpMode::OPBASE) ? "OPBASE" : "OFFLOAD";
455 4 : constexpr u32 numBlocks = 1;
456 4 : HrtAicpuLaunchKernelWithHostArgs(
457 4 : funcHandle, numBlocks, mStream.GetPtr(), &cfg, reinterpret_cast<void*>(kernelParamBuf_.get()->GetAddr()),
458 4 : sizeof(HcclKernelParamLite) + dynamicDataSize);
459 12 : HCCL_INFO(
460 : "[AicpuKernelLauncher][AicpuKernelLaunch] param.kernel.algName: %s, %s mode, %s"
461 : "HrtAicpuLaunchKernelWithHostArgs end!",
462 : param.kernel.algName, mode.c_str(), mStream.Describe().c_str());
463 4 : taskParam.taskType = TaskParamType::TASK_AICPU_KERNEL;
464 4 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
465 :
466 4 : SaveDfxTaskInfo(taskParam, -1, mStream.IsMaster());
467 4 : AddWaitToUserStream(stream);
468 4 : }
469 :
470 : constexpr u8 QUEUE_NOTIFY_POST_QID_POS = 0;
471 : constexpr u8 QUEUE_NOTIFY_WAIT_QID_POS = 1;
472 : constexpr u8 QUEUE_NOTIFY_TOPIC_ID_POS = 2;
473 :
474 3 : void CollServiceAiCpuImpl::AllocQueueNotify(std::vector<std::tuple<QId, QId, u32>>& queueNotifyReq) const
475 : {
476 3 : QueueNotifyManager& queueNotifyMgr = comm->GetAicpuQueueNotifyManager();
477 :
478 3 : std::for_each(queueNotifyReq.begin(), queueNotifyReq.end(), [&queueNotifyMgr](auto item) {
479 0 : queueNotifyMgr.ApplyFor(
480 0 : std::get<QUEUE_NOTIFY_POST_QID_POS>(item), std::get<QUEUE_NOTIFY_WAIT_QID_POS>(item),
481 0 : std::get<QUEUE_NOTIFY_TOPIC_ID_POS>(item));
482 0 : });
483 3 : }
484 :
485 4 : void CollServiceAiCpuImpl::AllocBcastPostCntNotify(std::vector<std::pair<QId, u32>>& bcastPostCntNotifyReq) const
486 : {
487 4 : QueueBcastPostCntNotifyManager& bcastPostCntNotifyMgr = comm->GetBcastPostCntNotifyManager();
488 :
489 4 : std::for_each(bcastPostCntNotifyReq.begin(), bcastPostCntNotifyReq.end(), [&bcastPostCntNotifyMgr](auto item) {
490 0 : bcastPostCntNotifyMgr.ApplyFor(item.first, item.second);
491 0 : HCCL_INFO("[CollServiceAiCpuImpl][%s] qid[%u] topicId[%u]", __func__, item.first, item.second);
492 0 : });
493 4 : }
494 :
495 4 : void CollServiceAiCpuImpl::AllocWaitGroupCntNotify(std::vector<std::pair<QId, u32>>& waitGroupCntNotifyReq) const
496 : {
497 4 : QueueWaitGroupCntNotifyManager& waitGroupCntNotifyMgr = comm->GetQueueWaitGroupCntNotifyManager();
498 :
499 4 : std::for_each(waitGroupCntNotifyReq.begin(), waitGroupCntNotifyReq.end(), [&waitGroupCntNotifyMgr](auto item) {
500 0 : waitGroupCntNotifyMgr.ApplyFor(item.first, item.second);
501 0 : HCCL_INFO("[CollServiceAiCpuImpl][%s] qid[%u] topicId[%u]", __func__, item.first, item.second);
502 0 : });
503 4 : }
504 :
505 4 : void CollServiceAiCpuImpl::AddPostToUserStream(const Stream& stream)
506 : {
507 4 : TaskParam taskParam{};
508 4 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
509 :
510 4 : auto postNotify = comm->GetHostDeviceSyncNotifyManager().GetDeviceWaitNotify();
511 12 : HCCL_INFO(
512 : "[CollServiceAiCpuImpl][AddPostToUserStream] notify id[%u] DevPhyId[%u], streamId[%u]", postNotify->GetId(),
513 : postNotify->GetDevPhyId(), stream.GetId());
514 4 : postNotify->Post(stream);
515 :
516 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
517 4 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
518 4 : taskParam.taskPara.Notify.notifyID = postNotify->GetId();
519 4 : taskParam.taskPara.Notify.value = 1;
520 :
521 4 : SaveDfxTaskInfo(taskParam, -1, stream.IsMaster());
522 4 : }
523 :
524 4 : void CollServiceAiCpuImpl::AddWaitToUserStream(const Stream& stream)
525 : {
526 4 : TaskParam taskParam{};
527 4 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
528 :
529 4 : auto waitNotify = comm->GetHostDeviceSyncNotifyManager().GetHostWaitNotify();
530 12 : HCCL_INFO(
531 : "[CollServiceAiCpuImpl][AddWaitToUserStream] notify id[%u] DevPhyId[%u], streamId[%u]", waitNotify->GetId(),
532 : waitNotify->GetDevPhyId(), stream.GetId());
533 :
534 4 : auto timeoutCheck = EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
535 4 : u32 timeout = (timeoutCheck == 0) ? timeoutCheck : (timeoutCheck + 50); // 主流notifywait超时时间: X+50s
536 12 : HCCL_INFO("[CollServiceAiCpuImpl][%s] notify wait timeout[%u]s", __func__, timeout);
537 4 : waitNotify->Wait(stream, timeout);
538 :
539 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
540 4 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
541 4 : taskParam.taskPara.Notify.notifyID = waitNotify->GetId();
542 4 : taskParam.taskPara.Notify.value = 1;
543 4 : SaveDfxTaskInfo(taskParam, -1, stream.IsMaster());
544 4 : }
545 :
546 3 : void CollServiceAiCpuImpl::AllocWorkStream(u32 primQueueNum) const
547 : {
548 3 : comm->GetAicpuStreamManager().AllocStreams(primQueueNum);
549 3 : }
550 :
551 224 : CollServiceAiCpuImpl::CollServiceAiCpuImpl(CommunicatorImpl* comm) : CollServiceBase(comm) {}
552 :
553 3 : void CollServiceAiCpuImpl::AllocNotifies(const vector<LinkData>& links)
554 : {
555 3 : vector<LinkData> pendingLinks;
556 3 : for (auto& link : links) {
557 0 : if (Contain(availableLinks, link)) {
558 0 : continue;
559 : }
560 0 : pendingLinks.emplace_back(link);
561 : }
562 9 : HCCL_INFO(
563 : "CollServiceAiCpuImpl::AllocNotifies links size %zu, pendingLinks size %zu", links.size(), pendingLinks.size());
564 3 : if (pendingLinks.empty()) {
565 3 : return;
566 : }
567 :
568 0 : for (auto& link : pendingLinks) {
569 : // 待修改: 申请数量
570 0 : comm->GetConnLocalNotifyManager().ApplyFor(link.GetRemoteRankId(), link);
571 : }
572 0 : HCCL_INFO("[CollServiceAiCpuImpl][AllocNotifies] end");
573 :
574 0 : availableLinks.insert(pendingLinks.begin(), pendingLinks.end());
575 3 : }
576 :
577 7 : void CollServiceAiCpuImpl::AllocOpMem(const CollOperator& op)
578 : {
579 7 : dynamicDataSize = CalcOpDynamicDataSize(op, op.opType, comm->GetRankSize());
580 7 : u64 kernelParamSize = sizeof(HcclKernelParamLite) + dynamicDataSize;
581 7 : if (kernelParamBuf_ == nullptr) {
582 7 : kernelParamBuf_ = make_shared<HostBuffer>(KERNEL_PARAM_BUF_SIZE);
583 7 : if (kernelParamBuf_ == nullptr) {
584 0 : HCCL_ERROR("[CollServiceAiCpuImpl][AllocOpMem] Alloc kernelParamBuf failed !");
585 0 : THROW<InternalException>(
586 0 : StringFormat("[CollServiceAiCpuImpl][AllocOpMem] Alloc kernelParamBuf failed len"));
587 : }
588 : }
589 :
590 7 : if (kernelParamBuf_ != nullptr && kernelParamSize > kernelParamBuf_.get()->GetSize()) {
591 0 : kernelParamBuf_ = make_shared<HostBuffer>(kernelParamSize);
592 0 : if (kernelParamBuf_ == nullptr) {
593 0 : HCCL_ERROR("[CollServiceAiCpuImpl][AllocOpMem] Alloc kernelParamBuf len[%llu] failed !", kernelParamSize);
594 0 : THROW<InternalException>(StringFormat(
595 : "[CollServiceAiCpuImpl][AllocOpMem] Alloc kernelParamBuf failed len[%llu]", kernelParamSize));
596 : }
597 : }
598 21 : HCCL_INFO("[AllocOpMem] op.opType[%d]", op.opType);
599 7 : }
600 :
601 7 : u64 CollServiceAiCpuImpl::CalcOpDynamicDataSize(const CollOperator& op, const OpType& opType, const u32& rankSize) const
602 : {
603 7 : u64 dynamicDataSize = 0ULL;
604 7 : switch (opType) {
605 2 : case OpType::BATCHSENDRECV:
606 : dynamicDataSize
607 2 : = sizeof(struct BatchSendRecvDataDes) + op.batchSendRecvDataDes.itemNum * sizeof(HcclSendRecvItem);
608 2 : break;
609 1 : case OpType::ALLTOALLV:
610 1 : dynamicDataSize = sizeof(struct AllToAllvDataDes) + rankSize * ALLTOALL_INFO_SIZE * sizeof(u64);
611 1 : break;
612 1 : case OpType::ALLTOALLVC:
613 1 : dynamicDataSize = sizeof(struct AllToAllvcDataDes) + rankSize * rankSize * sizeof(u64);
614 1 : break;
615 3 : default:
616 3 : break;
617 : }
618 21 : HCCL_INFO("HcclCommunicator::CalcOpDynamicDataSize dynamicDataSize[%llu]", dynamicDataSize);
619 7 : return dynamicDataSize;
620 : }
621 :
622 : // 功能说明：根据输入的LinkData信息，恢复Tansport对象
623 : // 输入说明：vector<LinkData> &links：linkData数据
624 1 : void CollServiceAiCpuImpl::RecoverTransport(
625 : vector<LinkData>& links, vector<std::pair<LinkGroup, u32>> linkGroupPair) // communicatorLinkData
626 : {
627 3 : HCCL_INFO("[CollServiceAiCpuImpl][RecoverTransport] start");
628 1 : RegisterCclBuffer(links);
629 : // 创建TransPort所需的Socket
630 1 : comm->GetSocketManager().BatchCreateSockets(links);
631 :
632 : // 创建TransPort所需的RmaConnection
633 1 : auto connBuilderPair = connectionsBuilders.emplace(comm->GetId(), make_unique<ConnectionsBuilder>(*comm));
634 1 : connBuilderPair.first->second->BatchBuild(comm->GetId(), links);
635 : // 创建TransPort所需的Notify资源
636 1 : AllocNotifies(links);
637 : // 重新构造TransPort
638 1 : auto op = comm->GetCurrentCollOperator();
639 1 : if (op->opMode == OpMode::OPBASE) {
640 1 : comm->GetMemTransportManager()->BatchRecoverOpbasedTransports(links);
641 0 : } else if (op->opMode == OpMode::OFFLOAD) {
642 0 : comm->GetMemTransportManager()->BatchRecoverOffloadTransports(op->opTag, links);
643 : }
644 :
645 3 : HCCL_INFO("[CollServiceAiCpuImpl][RecoverTransport] RecoverTransport success!");
646 2 : return;
647 : }
648 :
649 1 : HcclResult CollServiceAiCpuImpl::GetSnapShotDynamicBuf(CollOperator& op, BinaryStream& buf)
650 : {
651 1 : auto req = comm->GetCollAlgComponent()->GetCollAlgOpReq(op, comm->GetCurAlgName());
652 3 : HCCL_INFO(
653 : "CollServiceAiCpuImpl::GetSnapShotDynamicBuf GetCollAlgOpReq OrchestMode::INSTRUCTION, algName %s",
654 : req.algName.c_str());
655 1 : buf << req.resReq.levelRankPairs.size();
656 3 : HCCL_INFO("LeveRankPairs size is %zu", req.resReq.levelRankPairs.size());
657 2 : for (auto levelRankPair : req.resReq.levelRankPairs) {
658 1 : buf << levelRankPair.first << levelRankPair.second;
659 3 : HCCL_INFO("levelRankPair.first is %u, levelRankPair.second is %d", levelRankPair.first, levelRankPair.second);
660 : }
661 : // 保证快照一致性，使用0占位
662 1 : size_t linkGroupPairCount{0};
663 1 : buf << linkGroupPairCount;
664 3 : HCCL_DEBUG("[%s], linkGroupPairCount[%u]", __func__, linkGroupPairCount);
665 1 : return HcclResult::HCCL_SUCCESS;
666 1 : }
667 :
668 17 : static void SetModuleDataName(ModuleData& module, const std::string& name)
669 : {
670 17 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
671 17 : if (ret != 0) {
672 0 : THROW<InternalException>(StringFormat("strcpy_s name %s failed. ret[%d]", name.c_str(), ret));
673 : }
674 17 : }
675 :
676 2 : std::vector<char> CollServiceAiCpuImpl::PackOpData(const std::string& opTag, const CollAlgOpReq& req) const
677 : {
678 2 : std::vector<ModuleData> dataVec;
679 2 : dataVec.resize(AicpuResMgrType::__COUNT__);
680 :
681 2 : AicpuResMgrType resType = AicpuResMgrType::ALG_COMP_INFO;
682 2 : dataVec[resType].data = comm->GetCollAlgComponent()->GetPackedData();
683 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
684 :
685 2 : resType = AicpuResMgrType::STREAM;
686 2 : SetModuleDataName(dataVec[resType], "StreamManager");
687 2 : dataVec[resType].data = comm->GetAicpuStreamManager().GetPackedData();
688 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
689 :
690 2 : resType = AicpuResMgrType::QUEUE_NOTIFY;
691 2 : SetModuleDataName(dataVec[resType], "QueueNotifyManager");
692 2 : dataVec[resType].data = comm->GetAicpuQueueNotifyManager().GetPackedData();
693 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
694 :
695 2 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
696 2 : SetModuleDataName(dataVec[resType], "QueueWaitGroupCntNotifyManager");
697 2 : dataVec[resType].data = comm->GetQueueWaitGroupCntNotifyManager().GetPackedData();
698 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
699 :
700 2 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
701 2 : SetModuleDataName(dataVec[resType], "GetBcastPostCntNotifyManager");
702 2 : dataVec[resType].data = comm->GetBcastPostCntNotifyManager().GetPackedData();
703 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
704 :
705 2 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
706 2 : SetModuleDataName(dataVec[resType], "HostDeviceSyncNotifyManager");
707 2 : dataVec[resType].data = comm->GetHostDeviceSyncNotifyManager().GetPackedData();
708 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
709 :
710 2 : resType = AicpuResMgrType::TRANSPORT;
711 2 : SetModuleDataName(dataVec[resType], "MemTransportManager");
712 2 : auto op = comm->GetCurrentCollOperator();
713 2 : if (op->opMode == OpMode::OPBASE) { // 单算子模式
714 2 : dataVec[resType].data = comm->GetMemTransportManager()->GetOpbasedPackedData();
715 0 : } else if (op->opMode == OpMode::OFFLOAD) { // 图下沉模式
716 0 : dataVec[resType].data = comm->GetMemTransportManager()->GetOffloadPackedData(opTag);
717 : } else {
718 0 : THROW<InternalException>(StringFormat("opMode=%s failed", op->opMode.Describe().c_str()));
719 : }
720 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
721 :
722 2 : resType = AicpuResMgrType::ALG_TOPO;
723 2 : SetModuleDataName(dataVec[resType], req.algName);
724 : AlgTopoPackageHelper algTopoHelper;
725 2 : dataVec[resType].data = algTopoHelper.GetPackedData(req.resReq.topoInfo);
726 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
727 :
728 2 : resType = AicpuResMgrType::CONNECTD_MGR;
729 2 : SetModuleDataName(dataVec[resType], "ConnectedManager");
730 2 : dataVec[resType].data = comm->GetRankGraph()->GetPackedData(req.resReq.levelRankPairs);
731 6 : HCCL_INFO("CollServiceAiCpuImpl::PackOpData: GetResMgr %s Data", resType.Describe().c_str());
732 :
733 : AicpuResPackageHelper helper;
734 4 : return helper.GetPackedData(dataVec);
735 2 : }
736 :
737 12 : void CollServiceAiCpuImpl::SaveDfxTaskInfo(
738 : const TaskParam& taskParam, const RankId remoteRankId, const bool isMaster) const
739 : {
740 : u32 taskId;
741 : u32 streamId;
742 12 : HrtGetTaskIdAndStreamID(taskId, streamId);
743 :
744 : std::unique_ptr<TaskInfo> taskInfo = std::make_unique<TaskInfo>(
745 12 : streamId, taskId, remoteRankId, taskParam, comm->GetMirrorTaskManager().GetCurrDfxOpInfo(), isMaster);
746 :
747 12 : comm->GetMirrorTaskManager().AddTaskInfo(std::move(taskInfo));
748 12 : }
749 :
750 2 : void CollServiceAiCpuImpl::Resume()
751 : {
752 : // 从connectionsBuilder中获取linkData vector
753 2 : auto it = connectionsBuilders.find(comm->GetId());
754 2 : if (it == connectionsBuilders.end()) {
755 3 : HCCL_WARNING(
756 : "[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, cannot find connectionsBuilder, maybe no op has been "
757 : "loaded before, commId[%s].",
758 : comm->GetId().c_str());
759 1 : return;
760 : }
761 1 : vector<LinkData> links = it->second->GetAvailableLinksVec();
762 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, links size is [%zu].", links.size());
763 :
764 : // 基于当前已使用的LinkData创建connection
765 1 : comm->GetRmaConnManager().BatchCreate(links);
766 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, create connections end.");
767 :
768 : // 创建单算子transport, transport恢复建链，并等待完成
769 1 : comm->GetMemTransportManager()->BatchBuildOpbasedTransports(links);
770 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, BatchBuildOpbasedTransports end.");
771 1 : WaitOpbasedTransportReady();
772 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, WaitOpbasedTransportReady end.");
773 :
774 : // 基于opTag将connection注入到图模式的transport中
775 1 : comm->GetMemTransportManager()->UpdateOffloadTransports();
776 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, UpdateOffloadTransports end.");
777 :
778 : // 将通信域打包
779 1 : auto buffer = PackAllTransportData();
780 3 : HCCL_INFO("[NsRecovery][Resume] CollServiceAiCpuImpl::Resume, PackAllTransportData end.");
781 :
782 1 : shared_ptr<DevBuffer> devMem = make_shared<DevBuffer>(buffer.size()); // 申请device内存
783 3 : HCCL_INFO(
784 : "[NsRecovery][Resume] devMem->GetAddr(): 0x%llx, devMem->GetSize(): %llu", devMem->GetAddr(),
785 : devMem->GetSize());
786 2 : HrtMemcpy(
787 2 : reinterpret_cast<void*>(devMem->GetAddr()), devMem->GetSize(), buffer.data(), buffer.size(),
788 : RT_MEMCPY_HOST_TO_DEVICE); // H2D拷贝，将资源拷贝到device内存
789 :
790 : // 组新增的kernelLaunch命令、将打包数据下发到AICPU侧
791 1 : auto op = comm->GetCurrentCollOperator();
792 1 : if (op->opMode == OpMode::OPBASE) {
793 1 : AicpuUpdateCommLaunch(*comm->GetStreamManager().opbase->GetMaster(), devMem.get());
794 0 : } else if (op->opMode == OpMode::OFFLOAD) {
795 0 : comm->GetAicpuStreamManager().AllocFreeStream();
796 0 : AicpuUpdateCommLaunch(*comm->GetAicpuStreamManager().GetFreeStream(), devMem.get());
797 0 : HcclStreamSynchronize(comm->GetAicpuStreamManager().GetFreeStream()->GetPtr());
798 0 : HCCL_INFO("[NsRecovery][CollServiceAiCpuImpl] HcclUpdateCommKernelEntrance Stream Synchronize finished.");
799 : } else {
800 0 : THROW<InternalException>(StringFormat("[NsRecovery][Resume] opMode=%s failed", op->opMode.Describe().c_str()));
801 : }
802 1 : }
803 :
804 1 : void CollServiceAiCpuImpl::AicpuUpdateCommLaunch(Stream& stream, const DevBuffer* mem)
805 : {
806 1 : HcclKernelLaunchParam param;
807 :
808 1 : param.kernel.binaryResAddr = mem->GetAddr();
809 1 : param.kernel.binaryResSize = mem->GetSize();
810 :
811 1 : SetHcclKernelLaunchParam(param, comm);
812 :
813 1 : s32 ret = strcpy_s(param.kernelName, sizeof(param.kernelName), "HcclUpdateCommKernelEntrance");
814 1 : if (ret != EOK) {
815 0 : THROW<InternalException>(
816 0 : StringFormat("CollServiceAiCpuImpl::AicpuUpdateCommLaunch, strcpy_s kernelName failed! ret[%d]", ret));
817 : }
818 1 : auto op = comm->GetCurrentCollOperator();
819 1 : AicpuKernelLaunch(param, stream, op->opMode);
820 3 : HCCL_INFO("[NsRecovery][CollServiceAiCpuImpl] HcclUpdateCommKernelEntrance launched.");
821 1 : }
822 :
823 1 : std::vector<char> CollServiceAiCpuImpl::PackAllTransportData() const
824 : {
825 1 : std::vector<ModuleData> dataVec;
826 1 : dataVec.resize(AicpuResMgrType::__COUNT__);
827 :
828 1 : AicpuResMgrType resType = AicpuResMgrType::TRANSPORT;
829 1 : SetModuleDataName(dataVec[resType], "MemTransportManager");
830 1 : dataVec[resType].data = comm->GetMemTransportManager()->GetPackedAllTransportData();
831 3 : HCCL_INFO("CollServiceAiCpuImpl::PackTransportData: GetResMgr %s Data", resType.Describe().c_str());
832 :
833 : AicpuResPackageHelper helper;
834 2 : return helper.GetPackedData(dataVec);
835 1 : }
836 :
837 1 : void CollServiceAiCpuImpl::ReLoadWithOpBasedMode(CollOperator& op)
838 : {
839 3 : HCCL_INFO("[CollServiceAiCpuImpl::%s] start.", __func__);
840 1 : LoadWithOpBasedModeNoRegister(op);
841 3 : HCCL_INFO("[CollServiceAiCpuImpl::%s] end.", __func__);
842 1 : }
843 :
844 1 : void CollServiceAiCpuImpl::ReLoadWithOffloadMode(CollOperator& op)
845 : {
846 3 : HCCL_INFO("[CollServiceAiCpuImpl::%s] start.", __func__);
847 1 : LoadWithOffloadModeNoRegister(op);
848 3 : HCCL_INFO("[CollServiceAiCpuImpl::%s] end.", __func__);
849 1 : }
850 :
851 1 : void CollServiceAiCpuImpl::AllocQueueNotify(const InsQueue& insQueue)
852 : {
853 : // 重写基类接口，AICPU下不支持InsQueue传参
854 1 : THROW<InternalException>(
855 : "Should never use this method in AiCpu, use AllocQueueNotify(std::vector<std::tuple<QId, QId, "
856 : "u32>> &queueNotifyReq) Instead.");
857 : }
858 :
859 1 : void CollServiceAiCpuImpl::AllocQNotifyForSingleQ(const InsQueue& insQueue) const
860 : {
861 1 : THROW<InternalException>("AllocQNotifyForSingleQ is not support in AiCpu mode");
862 : }
863 :
864 1 : HcclResult CollServiceAiCpuImpl::ClearOpLoadedInfo(const std::string& opTag)
865 : {
866 1 : if (collOpLoadedMap.find(opTag) == collOpLoadedMap.end()) {
867 3 : HCCL_WARNING(
868 : "[LocalRmaBufManager::%s] opTag[%s] Cannot find Transport in collOpLoadedMap.", __func__, opTag.c_str());
869 1 : return HCCL_SUCCESS;
870 : }
871 0 : collOpLoadedMap.erase(opTag);
872 0 : return HCCL_SUCCESS;
873 : }
874 :
875 0 : u32 CollServiceAiCpuImpl::GetRemoteRankIdsHashValue(const CollOperator& op) const
876 : {
877 0 : vector<RankId> tempRankIds;
878 0 : HcclSendRecvItem* itemPtr = reinterpret_cast<HcclSendRecvItem*>(op.batchSendRecvDataDes.sendRecvItemsPtr);
879 0 : u32 itemNum = op.batchSendRecvDataDes.itemNum;
880 0 : CHK_PTR_NULL(itemPtr);
881 0 : for (u32 i = 0; i < itemNum; i++) {
882 0 : u32 remoteRankId = (itemPtr + i)->remoteRank;
883 0 : tempRankIds.push_back(remoteRankId);
884 0 : HCCL_INFO(
885 : "[CollServiceAiCpuImpl][GetRemoteRankIdsHashValue] insert remoteUserRank[%u] to vector", remoteRankId);
886 : }
887 0 : std::sort(tempRankIds.begin(), tempRankIds.end());
888 :
889 0 : u32 seed = tempRankIds.size();
890 0 : const u32 goldRatio = 0x9e3779b9;
891 0 : for (u32 rankId : tempRankIds) {
892 0 : seed ^= std::hash<uint32_t>()(rankId) + goldRatio + (seed << 6) + (seed >> 2);
893 : }
894 0 : return seed;
895 0 : }
896 :
897 1 : HcclResult CollServiceAiCpuImpl::FillBatchSendRecvData(const CollOperator& op)
898 : {
899 1 : if (dynamicDataSize == 0) {
900 0 : HCCL_ERROR("CollServiceAiCpuImpl::FillBatchSendRecvData dynamicDataSize is 0");
901 0 : THROW<InternalException>(StringFormat("CollServiceAiCpuImpl::FillBatchSendRecvData dynamicDataSize is 0"));
902 : return HCCL_E_PARA;
903 : }
904 1 : Buffer dynamicDataMem = kernelParamBuf_.get()->Range(sizeof(struct HcclKernelParamLite), dynamicDataSize);
905 : struct BatchSendRecvDataDes* batchSendRecvDataPtr
906 1 : = reinterpret_cast<struct BatchSendRecvDataDes*>(dynamicDataMem.GetAddr());
907 1 : batchSendRecvDataPtr->itemNum = op.batchSendRecvDataDes.itemNum;
908 3 : for (u32 i = 0; i < op.batchSendRecvDataDes.itemNum; i++) {
909 2 : CHK_PTR_NULL(static_cast<HcclSendRecvItem*>(op.batchSendRecvDataDes.sendRecvItemsPtr) + i);
910 : batchSendRecvDataPtr->batchSendRecvItem[i]
911 2 : = *(static_cast<HcclSendRecvItem*>(op.batchSendRecvDataDes.sendRecvItemsPtr) + i);
912 6 : HCCL_INFO(
913 : "[CollServiceAiCpuImpl][FillBatchSendRecvData] alloc batchSendRecvItem remoteRank[%u]",
914 : batchSendRecvDataPtr->batchSendRecvItem[i].remoteRank);
915 : }
916 :
917 1 : return HCCL_SUCCESS;
918 1 : }
919 :
920 1 : HcclResult CollServiceAiCpuImpl::FillAllToAllvData(const CollOperator& op)
921 : {
922 1 : if (dynamicDataSize == 0) {
923 0 : HCCL_ERROR("CollServiceAiCpuImpl::FillAllToAllvData dynamicDataSize is 0");
924 0 : THROW<InternalException>(StringFormat("CollServiceAiCpuImpl::FillAllToAllvData dynamicDataSize is 0"));
925 : return HCCL_E_PARA;
926 : }
927 1 : Buffer dynamicDataMem = kernelParamBuf_.get()->Range(sizeof(struct HcclKernelParamLite), dynamicDataSize);
928 1 : struct AllToAllvDataDes* alltoallvDataPtr = reinterpret_cast<struct AllToAllvDataDes*>(dynamicDataMem.GetAddr());
929 1 : alltoallvDataPtr->sendType = static_cast<u8>(op.all2AllVDataDes.sendType);
930 1 : alltoallvDataPtr->recvType = static_cast<u8>(op.all2AllVDataDes.recvType);
931 1 : u32 rankSize = comm->GetRankSize();
932 1 : u64* sendCountsPtr = static_cast<u64*>(alltoallvDataPtr->sendRecvInfos);
933 1 : u64* recvCountsPtr = sendCountsPtr + rankSize;
934 1 : u64* sdisplsPtr = recvCountsPtr + rankSize;
935 1 : u64* rdisplsPtr = sdisplsPtr + rankSize;
936 5 : for (u32 i = 0; i < rankSize; i++) {
937 4 : CHK_PTR_NULL(static_cast<const u64*>(op.all2AllVDataDes.sendCounts) + i);
938 4 : sendCountsPtr[i] = *(static_cast<const u64*>(op.all2AllVDataDes.sendCounts) + i);
939 4 : CHK_PTR_NULL(static_cast<const u64*>(op.all2AllVDataDes.recvCounts) + i);
940 4 : recvCountsPtr[i] = *(static_cast<const u64*>(op.all2AllVDataDes.recvCounts) + i);
941 4 : CHK_PTR_NULL(static_cast<const u64*>(op.all2AllVDataDes.sdispls) + i);
942 4 : sdisplsPtr[i] = *(static_cast<const u64*>(op.all2AllVDataDes.sdispls) + i);
943 4 : CHK_PTR_NULL(static_cast<const u64*>(op.all2AllVDataDes.rdispls) + i);
944 4 : rdisplsPtr[i] = *(static_cast<const u64*>(op.all2AllVDataDes.rdispls) + i);
945 12 : HCCL_INFO(
946 : "[CollServiceAiCpuImpl][FillAllToAllvData] sendCounts[%llu], recvCounts[%llu], sdispls[%llu], "
947 : "rdispls[%llu]",
948 : sendCountsPtr[i], recvCountsPtr[i], sdisplsPtr[i], rdisplsPtr[i]);
949 : }
950 3 : HCCL_INFO(
951 : "[CollServiceAiCpuImpl][FillAllToAllvData] fill alltoallv data success, alltoallvDataPtr[%p]",
952 : alltoallvDataPtr);
953 1 : return HCCL_SUCCESS;
954 1 : }
955 :
956 1 : HcclResult CollServiceAiCpuImpl::FillAllToAllvcData(const CollOperator& op)
957 : {
958 1 : if (dynamicDataSize == 0) {
959 0 : HCCL_ERROR("CollServiceAiCpuImpl::FillAllToAllvcData dynamicDataSize is 0");
960 0 : THROW<InternalException>(StringFormat("CollServiceAiCpuImpl::FillAllToAllvcData dynamicDataSize is 0"));
961 : return HCCL_E_PARA;
962 : }
963 1 : Buffer dynamicDataMem = kernelParamBuf_.get()->Range(sizeof(struct HcclKernelParamLite), dynamicDataSize);
964 1 : struct AllToAllvcDataDes* alltoallvcDataPtr = reinterpret_cast<struct AllToAllvcDataDes*>(dynamicDataMem.GetAddr());
965 1 : alltoallvcDataPtr->sendType = static_cast<u8>(op.all2AllVCDataDes.sendType);
966 1 : alltoallvcDataPtr->recvType = static_cast<u8>(op.all2AllVCDataDes.recvType);
967 1 : u32 rankSize = comm->GetRankSize();
968 17 : for (u64 i = 0; i < rankSize * rankSize; i++) {
969 16 : CHK_PTR_NULL(static_cast<const u64*>(op.all2AllVCDataDes.sendCountMatrix) + i);
970 16 : alltoallvcDataPtr->sendCountMatrix[i] = *(static_cast<const u64*>(op.all2AllVCDataDes.sendCountMatrix) + i);
971 : }
972 1 : return HCCL_SUCCESS;
973 1 : }
974 :
975 : } // namespace Hccl
|
