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