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