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 <shared_mutex>
12 : #include "inc/aicpu_utils.h"
13 : #include "log.h"
14 : #include "aicpu_comm_destroy_func.h"
15 : #include "communicator_impl_lite_manager.h"
16 : #include "ub_conn_lite_mgr.h"
17 : #include "aicpu_daemon_service.h"
18 : #include "task_exception_func.h"
19 : #include "task_exception_handler_lite.h"
20 : #include "coll_operator.h"
21 : #include "profiling_handler_lite.h"
22 : using namespace Hccl;
23 :
24 1 : AicpuUtils::AicpuUtils() {}
25 :
26 134 : AicpuUtils& AicpuUtils::GetInstance()
27 : {
28 134 : static AicpuUtils instance_;
29 134 : return instance_;
30 : }
31 :
32 6 : void AicpuUtils::CreateSingleInstance(void* args) const
33 : {
34 6 : auto* kernelParam = reinterpret_cast<HcclKernelParamLite*>(args);
35 6 : UbConnLiteMgr::GetInstance();
36 6 : AicpuDaemonService::GetInstance();
37 6 : TaskExceptionFunc::GetInstance().SetEnable(
38 6 : kernelParam->envConfig.taskExceptionEnable); // 根据环境变量使能TaskException
39 6 : AicpuCommDestroyFunc::GetInstance();
40 6 : TaskExceptionHandlerLite::GetInstance();
41 6 : ProfilingHandlerLite::GetInstance();
42 6 : DevCapability::GetInstance();
43 6 : CommunicatorImplLiteMgr::GetInstance().SetEnvConfig(kernelParam->envConfig); // 初始化并设置Device侧环境变量
44 6 : }
45 :
46 0 : HcclResult AicpuUtils::Init() const
47 : {
48 0 : CHK_RET(ProfilingHandlerLite::GetInstance().Init());
49 0 : return HCCL_SUCCESS;
50 : }
51 :
52 6 : HcclResult AicpuUtils::WaitCommFree(CommunicatorImplLite* communicatorImplLite, const char* funcName) const
53 : {
54 6 : auto startTime = std::chrono::steady_clock::now();
55 6 : constexpr uint32_t pollIntervalUs = 10; // 轮询间隔10us
56 6 : constexpr uint32_t pollTimeoutMs = 10; // 轮询超时时间10ms
57 6 : auto waitPollTimeOutMs = std::chrono::milliseconds(pollTimeoutMs);
58 6 : unique_lock<std::mutex> aicpuLock(communicatorImplLite->GetAicpuMc2Mutex());
59 : while (true) {
60 167 : if (communicatorImplLite->IsUsed()) {
61 162 : if ((std::chrono::steady_clock::now() - startTime) >= waitPollTimeOutMs) {
62 3 : HCCL_ERROR(
63 : "%s poll timeout, comm id [%u] has been used", funcName, communicatorImplLite->GetCommIdIndex());
64 1 : return HCCL_E_TIMEOUT;
65 : }
66 161 : aicpuLock.unlock();
67 161 : usleep(pollIntervalUs);
68 161 : aicpuLock.lock();
69 : } else {
70 5 : communicatorImplLite->SetIsUsed(true);
71 5 : aicpuLock.unlock();
72 5 : break;
73 : }
74 161 : }
75 5 : return HCCL_SUCCESS;
76 6 : }
77 :
78 4 : HcclResult AicpuUtils::GetCommHandle(CommunicatorImplLite* communicatorImplLite, void** opHandle) const
79 : {
80 : // 启动计时，一直获取不到comm.isUsed会退出
81 7 : CHK_RET(WaitCommFree(communicatorImplLite, __func__));
82 :
83 : // 默认执行反序列化
84 3 : auto reporter = communicatorImplLite->GetProfilingReporterLite();
85 3 : CHK_PTR_NULL(reporter);
86 3 : reporter->UpdateProfStat();
87 3 : if (kernelParam_->op.algOperator.opMode == OpMode::OPBASE) {
88 2 : communicatorImplLite->SetCurrentOpMode(kernelParam_->op.algOperator.opMode);
89 2 : communicatorImplLite->UpdateCommParam(kernelParam_);
90 2 : EXCEPTION_CATCH(communicatorImplLite->UpdateRes(kernelParam_), return HCCL_E_INTERNAL);
91 : } else {
92 3 : HCCL_ERROR(
93 : "[%s]%s only support opbase, but get opMode %s.", __func__, __func__,
94 : kernelParam_->op.algOperator.opMode.Describe().c_str());
95 1 : return HCCL_E_PARA;
96 : }
97 :
98 2 : *opHandle = reinterpret_cast<void*>(communicatorImplLite);
99 2 : return HCCL_SUCCESS;
100 : }
101 :
102 2 : int AicpuUtils::GetException(
103 : StreamLite* curStream, uint32_t flag, CommunicatorImplLite* communicatorImplLite, string additionInfo) const
104 : {
105 : // 遍历主从流的状态
106 2 : auto recvInfo = make_shared<halReportRecvInfo>();
107 2 : constexpr uint32_t cqeSize = MAX_REPORT_CNT * sizeof(rtLogicCqReport_t);
108 2 : uint8_t tmpAddr[cqeSize] = {}; // cqe byte size
109 2 : recvInfo->cqe_addr = tmpAddr; // 外部保证是有效的地址
110 :
111 2 : const char* typeStr = (flag == GET_TASK_STATUS) ? "HcclGetTaskStatus" : "HcclPrintTaskExceptionAllComm";
112 :
113 2 : if (TaskExceptionFunc::GetInstance().GetReporterInfo(curStream, recvInfo) == 1) {
114 0 : HCCL_WARNING("[%s]GetReporterInfo execute failed", typeStr);
115 0 : return 1;
116 : }
117 2 : uint32_t reportNum = recvInfo->report_cqe_num;
118 2 : if (reportNum > MAX_REPORT_CNT) {
119 0 : HCCL_WARNING("[%s]report cqe num %u should not big than %u", typeStr, reportNum, MAX_REPORT_CNT);
120 0 : return 1;
121 : }
122 :
123 2 : if (flag == GET_TASK_STATUS) {
124 6 : HCCL_INFO(
125 : "[%s]Status info:stream %u, head %u, tail %u", __func__, curStream->GetId(),
126 : curStream->GetRtsq()->GetHead(), curStream->GetRtsq()->GetTail());
127 3 : for (uint32_t idx = 0U; idx < reportNum; ++idx) {
128 : auto& reportOfOne
129 2 : = *((reinterpret_cast<rtLogicCqReport_t*>(recvInfo->cqe_addr)) + idx); // 外部保证是有效的地址
130 2 : if (TaskExceptionFunc::GetInstance().IsExceptionCqe(reportOfOne)) {
131 1 : return 1;
132 : }
133 : }
134 : } else {
135 0 : for (uint32_t idx = 0U; idx < reportNum; ++idx) {
136 : auto& reportOfOne
137 0 : = *((reinterpret_cast<rtLogicCqReport_t*>(recvInfo->cqe_addr)) + idx); // 外部保证是有效的地址
138 0 : if (TaskExceptionFunc::GetInstance().IsExceptionCqe(reportOfOne)) {
139 0 : if (additionInfo != "") {
140 0 : HCCL_ERROR("%s", additionInfo.c_str());
141 : }
142 0 : TaskExceptionHandlerLite::Process(communicatorImplLite, &reportOfOne);
143 : }
144 : }
145 : }
146 1 : return 0;
147 2 : }
148 :
149 3 : void AicpuUtils::GetStreamException(
150 : StreamLite* curStream, string nullInfo, CommunicatorImplLite* communicatorImplLite, string additionInfo) const
151 : {
152 3 : if (curStream == nullptr) {
153 3 : HCCL_WARNING("[%s]%s", __func__, nullInfo.c_str());
154 1 : return;
155 : }
156 2 : if (communicatorImplLite == nullptr) {
157 0 : HCCL_WARNING("[%s]communicatorImplLite is nullptr", __func__);
158 0 : return;
159 : }
160 2 : auto* curRtsq = curStream->GetRtsq();
161 2 : if (curRtsq == nullptr) {
162 0 : HCCL_WARNING("[%s]Stream[%u] rtsq is nullptr.", __func__, curStream->GetId());
163 0 : return;
164 : }
165 2 : auto curSqHead = curRtsq->QuerySqHead();
166 2 : auto curSqTail = curRtsq->QuerySqTail();
167 :
168 2 : string finishInfo = "finished";
169 2 : if (curSqHead != curSqTail) {
170 0 : finishInfo = "unfinished";
171 0 : GetException(curStream, GET_EXCEPTION_INFO, communicatorImplLite, additionInfo);
172 : }
173 6 : HCCL_INFO(
174 : "[%s]Stream %u %s, sq id %u, head %u, tail %u.", __func__, curStream->GetId(), finishInfo.c_str(),
175 : curStream->GetSqId(), curSqHead, curSqTail);
176 2 : return;
177 2 : }
178 :
179 5 : HcclResult AicpuUtils::HcclLaunchCcore(
180 : void* opHandle, uint64_t dstAddr, uint32_t turnNum, uint64_t turnNumAddr, bool isLast, int ccoreType) const
181 : {
182 5 : const char* typeStr = (ccoreType == CCORE_NOTIFY_TYPE) ? "HcclLaunchCcoreWait" : "HcclLaunchCcorePost";
183 15 : HCCL_INFO(
184 : "[%s]opHandle %p, dstAddr %llu, turnNum %u, turnNumAddr %llu, isLast %u, type %s.", __func__, opHandle, dstAddr,
185 : turnNum, turnNumAddr, isLast, typeStr);
186 5 : if (ccoreType != CCORE_WAIT_TYPE && ccoreType != CCORE_NOTIFY_TYPE) {
187 0 : HCCL_ERROR("[%s]Args type %d is not in CCORE_WAIT_TYPE(0) or CCORE_NOTIFY_TYPE(1).", __func__, ccoreType);
188 0 : return HCCL_E_PARA;
189 : }
190 :
191 5 : CommunicatorImplLite* communicatorImplLite = reinterpret_cast<CommunicatorImplLite*>(opHandle);
192 5 : auto* streamLiteMgr = communicatorImplLite->GetStreamLiteMgr();
193 5 : CHK_PTR_NULL(streamLiteMgr);
194 :
195 5 : auto* master = streamLiteMgr->GetMaster();
196 5 : CHK_PTR_NULL(master);
197 :
198 5 : auto* rtsq = master->GetRtsq();
199 5 : CHK_PTR_NULL(rtsq);
200 :
201 5 : if (ccoreType == CCORE_NOTIFY_TYPE) {
202 3 : rtsq->CCoreNotifyRecord(dstAddr, turnNumAddr + turnNum * sizeof(uint32_t));
203 : } else {
204 2 : rtsq->CCoreNotifyWait(dstAddr, turnNumAddr + turnNum * sizeof(uint32_t), isLast);
205 : }
206 4 : rtsq->LaunchTask();
207 4 : return HCCL_SUCCESS;
208 : }
209 :
210 3 : void AicpuUtils::CalcA2ASendRecvMem(const CollAlgOperator& algOperator, uint64_t& sendSize, uint64_t& recvSize) const
211 : {
212 3 : uint64_t sendCount = 0;
213 3 : uint64_t recvCount = 0;
214 3 : uint32_t sendTypeSize = 0;
215 3 : uint32_t recvTypeSize = 0;
216 :
217 3 : if (algOperator.opType == OpType::ALLTOALLV) {
218 1 : for (uint32_t i = 0; i < rankSize_; i++) {
219 0 : uint64_t curSendCount = *(static_cast<const uint64_t*>(algOperator.all2AllVDataDes.sendCounts) + i)
220 0 : + *(static_cast<const uint64_t*>(algOperator.all2AllVDataDes.sdispls) + i);
221 0 : sendCount = std::max(sendCount, curSendCount);
222 0 : uint64_t curRecvCount = *(static_cast<const uint64_t*>(algOperator.all2AllVDataDes.recvCounts) + i)
223 0 : + *(static_cast<const uint64_t*>(algOperator.all2AllVDataDes.rdispls) + i);
224 0 : recvCount = std::max(recvCount, curRecvCount);
225 : }
226 1 : sendTypeSize = DataTypeSizeGet(algOperator.all2AllVDataDes.sendType);
227 1 : recvTypeSize = DataTypeSizeGet(algOperator.all2AllVDataDes.recvType);
228 2 : } else if (algOperator.opType == OpType::ALLTOALLVC) {
229 1 : for (uint32_t i = 0; i < rankSize_; i++) {
230 0 : sendCount += *(
231 0 : static_cast<const uint64_t*>(algOperator.all2AllVCDataDes.sendCountMatrix) + myRank_ * rankSize_ + i);
232 0 : recvCount += *(
233 0 : static_cast<const uint64_t*>(algOperator.all2AllVCDataDes.sendCountMatrix) + myRank_ + rankSize_ * i);
234 : }
235 1 : sendTypeSize = DataTypeSizeGet(algOperator.all2AllVCDataDes.sendType);
236 1 : recvTypeSize = DataTypeSizeGet(algOperator.all2AllVCDataDes.recvType);
237 : } else {
238 1 : sendCount = algOperator.all2AllDataDes.sendCount * rankSize_;
239 1 : recvCount = algOperator.all2AllDataDes.recvCount * rankSize_;
240 1 : sendTypeSize = DataTypeSizeGet(algOperator.all2AllDataDes.sendType);
241 1 : recvTypeSize = DataTypeSizeGet(algOperator.all2AllDataDes.recvType);
242 : }
243 3 : sendSize = sendCount * sendTypeSize;
244 3 : recvSize = recvCount * recvTypeSize;
245 9 : HCCL_INFO(
246 : "[%s]CalcA2ASendRecvMem finish, algOperator %s, sendCount %llu, sendTypeSize %u, "
247 : "recvCount %llu, recvTypeSize %u, sendSize %llu, recvSize %llu",
248 : __func__, algOperator.opType.Describe().c_str(), sendCount, sendTypeSize, recvCount, recvTypeSize, sendSize,
249 : recvSize);
250 3 : }
251 : HcclResult
252 1 : AicpuUtils::ConvertCollOperatorMemV(CollAlgOperator& algOperator, HcclAicpuOpLite& op, const HcclOpData* data) const
253 : {
254 1 : auto dataType = HcclDataTypeToDataType(data->dataType);
255 1 : CHECK_DATA_TYPE(dataType);
256 1 : uint64_t size = DataTypeSizeGet(dataType) * data->dataCount;
257 1 : uint64_t* counts = static_cast<uint64_t*>(data->vDataDes.counts);
258 1 : uint64_t totalCount = 0;
259 1 : for (size_t index = 0; index < rankSize_; index++) {
260 0 : totalCount += counts[index];
261 : }
262 1 : uint64_t totalSize = DataTypeSizeGet(dataType) * totalCount;
263 :
264 1 : if (algOperator.opType == OpType::REDUCESCATTERV) {
265 1 : algOperator.inputMem = make_shared<Buffer>(data->input, totalSize);
266 1 : op.input.size = totalSize;
267 : } else {
268 0 : algOperator.inputMem = make_shared<Buffer>(data->input, size);
269 0 : op.input.size = size;
270 : }
271 1 : if (algOperator.opType == OpType::ALLGATHERV) {
272 0 : algOperator.outputMem = make_shared<Buffer>(data->output, totalSize);
273 0 : op.output.size = totalSize;
274 : } else {
275 1 : algOperator.outputMem = make_shared<Buffer>(data->output, size);
276 1 : op.output.size = size;
277 : }
278 :
279 3 : HCCL_INFO(
280 : "[%s] finish, opType[%s], inputSize[%llu], outputSize[%llu]", __func__, algOperator.opType.Describe().c_str(),
281 : op.input.size, op.output.size);
282 1 : return HCCL_SUCCESS;
283 : }
284 :
285 1 : void AicpuUtils::ConvertCollOperatorMem(
286 : CollAlgOperator& algOperator, HcclAicpuOpLite& op, const HcclOpData* data, const uint64_t& size) const
287 : {
288 1 : if (algOperator.opType == OpType::REDUCESCATTER || algOperator.opType == OpType::SCATTER) {
289 0 : algOperator.inputMem = make_shared<Buffer>(data->input, size * rankSize_);
290 0 : op.input.size = size * rankSize_;
291 : } else {
292 1 : algOperator.inputMem = make_shared<Buffer>(data->input, size);
293 1 : op.input.size = size;
294 : }
295 1 : if (algOperator.opType == OpType::ALLGATHER || algOperator.opType == OpType::GATHER) {
296 0 : algOperator.outputMem = make_shared<Buffer>(data->output, size * rankSize_);
297 0 : op.output.size = size * rankSize_;
298 : } else {
299 1 : algOperator.outputMem = make_shared<Buffer>(data->output, size);
300 1 : op.output.size = size;
301 : }
302 :
303 3 : HCCL_INFO(
304 : "[%s] finish, opType[%s], inputSize[%llu], outputSize[%llu]", __func__, algOperator.opType.Describe().c_str(),
305 : op.input.size, op.output.size);
306 1 : }
307 :
308 : HcclResult
309 5 : AicpuUtils::FillCollOperatorMemInfo(CollAlgOperator& algOperator, HcclAicpuOpLite& op, const HcclOpData* data) const
310 : {
311 5 : op.input.addr = data->input;
312 5 : op.input.tokenId = 0;
313 5 : op.input.tokenValue = 0;
314 5 : op.output.addr = data->output;
315 5 : op.output.tokenId = 0;
316 5 : op.output.tokenValue = 0;
317 9 : if (algOperator.opType == OpType::ALLTOALL || algOperator.opType == OpType::ALLTOALLV
318 9 : || algOperator.opType == OpType::ALLTOALLVC) {
319 3 : uint64_t sendSize = 0, recvSize = 0;
320 3 : CalcA2ASendRecvMem(algOperator, sendSize, recvSize);
321 3 : algOperator.inputMem = make_shared<Buffer>(data->input, sendSize);
322 3 : algOperator.outputMem = make_shared<Buffer>(data->output, recvSize);
323 3 : op.input.size = sendSize;
324 3 : op.output.size = recvSize;
325 2 : } else if (algOperator.opType == OpType::BATCHSENDRECV) {
326 0 : HCCL_INFO("[%s] OpType::BATCHSENDRECV item = %llu", __func__, algOperator.batchSendRecvDataDes.itemNum);
327 : } else {
328 2 : if (algOperator.opType == OpType::REDUCESCATTERV || algOperator.opType == OpType::ALLGATHERV) {
329 1 : return ConvertCollOperatorMemV(algOperator, op, data);
330 : } else {
331 1 : auto tmp = HcclDataTypeToDataType(data->dataType);
332 1 : CHECK_DATA_TYPE(tmp);
333 1 : uint64_t size = DataTypeSizeGet(tmp) * data->dataCount;
334 1 : if (size != 0) {
335 3 : HCCL_INFO("[%s] size is %llu", __func__, size);
336 1 : ConvertCollOperatorMem(algOperator, op, data, size);
337 : } else {
338 0 : HCCL_WARNING("[%s] data size is 0", __func__);
339 : }
340 : }
341 : }
342 12 : HCCL_INFO(
343 : "[%s]opType %s, op.input.addr %llu, op.input.size %llu, op.output.addr %llu, "
344 : "op.output.size %llu",
345 : __func__, algOperator.opType.Describe().c_str(), op.input.addr, op.input.size, op.output.addr, op.output.size);
346 4 : return HCCL_SUCCESS;
347 : }
348 :
349 6 : HcclResult AicpuUtils::FillKernelParam(HcclOpData* data) const
350 : {
351 6 : kernelParam_->op.algOperator.reduceOp = HcclReduceOpToReduceOp(HCCL_REDUCE_RESERVED);
352 6 : if (data->opType == HCCL_CMD_ALLREDUCE || data->opType == HCCL_CMD_REDUCE || data->opType == HCCL_CMD_REDUCE_SCATTER
353 4 : || data->opType == HCCL_CMD_REDUCE_SCATTER_V) {
354 3 : kernelParam_->op.algOperator.reduceOp = HcclReduceOpToReduceOp(data->reduceOp);
355 : }
356 6 : kernelParam_->op.algOperator.dataType = HcclDataTypeToDataType(data->dataType);
357 6 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.dataType);
358 6 : kernelParam_->op.algOperator.outputDataType = HcclDataTypeToDataType(data->outputDataType);
359 6 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.outputDataType);
360 6 : kernelParam_->op.algOperator.dataCount = data->dataCount;
361 6 : kernelParam_->op.algOperator.root = data->root;
362 6 : kernelParam_->op.algOperator.sendRecvRemoteRank = data->sendRecvRemoteRank;
363 18 : HCCL_INFO(
364 : "[%s]opType=%s, reduceOp=%u, dataType=%u, outputDataType=%u, dataCount=%llu, root=%u, sendRecvRemoteRank=%u",
365 : __func__, kernelParam_->op.algOperator.opType.Describe().c_str(), data->reduceOp, data->dataType,
366 : data->outputDataType, data->dataCount, data->root, data->sendRecvRemoteRank);
367 6 : if (kernelParam_->op.algOperator.opType == OpType::ALLTOALL) {
368 1 : kernelParam_->op.algOperator.all2AllDataDes.recvType = HcclDataTypeToDataType(data->all2AllDataDes.recvType);
369 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllDataDes.recvType);
370 1 : kernelParam_->op.algOperator.all2AllDataDes.sendType = HcclDataTypeToDataType(data->all2AllDataDes.sendType);
371 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllDataDes.sendType);
372 1 : kernelParam_->op.algOperator.all2AllDataDes.sendCount = data->all2AllDataDes.sendCount;
373 1 : kernelParam_->op.algOperator.all2AllDataDes.recvCount = data->all2AllDataDes.recvCount;
374 5 : } else if (kernelParam_->op.algOperator.opType == OpType::ALLTOALLV) {
375 1 : kernelParam_->op.algOperator.all2AllVDataDes.sendType = HcclDataTypeToDataType(data->all2AllVDataDes.sendType);
376 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllVDataDes.sendType);
377 1 : kernelParam_->op.algOperator.all2AllVDataDes.recvType = HcclDataTypeToDataType(data->all2AllVDataDes.recvType);
378 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllVDataDes.recvType);
379 1 : CHK_PTR_NULL(data->all2AllVDataDes.sendCounts);
380 1 : kernelParam_->op.algOperator.all2AllVDataDes.sendCounts = data->all2AllVDataDes.sendCounts;
381 1 : CHK_PTR_NULL(data->all2AllVDataDes.recvCounts);
382 1 : kernelParam_->op.algOperator.all2AllVDataDes.recvCounts = data->all2AllVDataDes.recvCounts;
383 1 : CHK_PTR_NULL(data->all2AllVDataDes.sdispls);
384 1 : kernelParam_->op.algOperator.all2AllVDataDes.sdispls = data->all2AllVDataDes.sdispls;
385 1 : CHK_PTR_NULL(data->all2AllVDataDes.rdispls);
386 1 : kernelParam_->op.algOperator.all2AllVDataDes.rdispls = data->all2AllVDataDes.rdispls;
387 4 : } else if (kernelParam_->op.algOperator.opType == OpType::ALLTOALLVC) {
388 1 : kernelParam_->op.algOperator.all2AllVCDataDes.sendType
389 1 : = HcclDataTypeToDataType(data->all2AllVCDataDes.sendType);
390 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllVCDataDes.sendType);
391 1 : kernelParam_->op.algOperator.all2AllVCDataDes.recvType
392 1 : = HcclDataTypeToDataType(data->all2AllVCDataDes.recvType);
393 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.all2AllVCDataDes.recvType);
394 1 : CHK_PTR_NULL(data->all2AllVCDataDes.sendCountMatrix);
395 1 : kernelParam_->op.algOperator.all2AllVCDataDes.sendCountMatrix = data->all2AllVCDataDes.sendCountMatrix;
396 3 : } else if (
397 3 : kernelParam_->op.algOperator.opType == OpType::ALLGATHERV
398 3 : || kernelParam_->op.algOperator.opType == OpType::REDUCESCATTERV) {
399 1 : CHK_PTR_NULL(data->vDataDes.counts);
400 1 : kernelParam_->op.algOperator.vDataDes.counts = data->vDataDes.counts;
401 1 : CHK_PTR_NULL(data->vDataDes.displs);
402 1 : kernelParam_->op.algOperator.vDataDes.displs = data->vDataDes.displs;
403 1 : kernelParam_->op.algOperator.vDataDes.dataType = HcclDataTypeToDataType(data->vDataDes.dataType);
404 1 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.vDataDes.dataType);
405 2 : } else if (kernelParam_->op.algOperator.opType == OpType::BATCHSENDRECV) {
406 0 : CHK_PTR_NULL(data->batchSendRecvDataDes.sendRecvItemsPtr);
407 0 : kernelParam_->op.algOperator.batchSendRecvDataDes.sendRecvItemsPtr
408 0 : = data->batchSendRecvDataDes.sendRecvItemsPtr;
409 0 : kernelParam_->op.algOperator.dataType = HcclDataTypeToDataType(
410 0 : static_cast<HcclSendRecvItem*>(data->batchSendRecvDataDes.sendRecvItemsPtr)->dataType);
411 0 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.dataType);
412 0 : kernelParam_->op.algOperator.batchSendRecvDataDes.itemNum = data->batchSendRecvDataDes.itemNum;
413 : } else {
414 2 : kernelParam_->op.algOperator.dataDes.dataType = HcclDataTypeToDataType(data->dataDes.dataType);
415 5 : CHECK_DATA_TYPE(kernelParam_->op.algOperator.dataDes.dataType);
416 1 : kernelParam_->op.algOperator.dataDes.dataCount = data->dataDes.dataCount;
417 1 : kernelParam_->op.algOperator.dataDes.strideCount = data->dataDes.strideCount;
418 : }
419 5 : return HCCL_SUCCESS;
420 : }
421 :
422 7 : HcclResult AicpuUtils::RecoverKernelParam(CommunicatorImplLite* communicatorImplLite, HcclOpData* data)
423 : {
424 7 : unique_lock<std::shared_timed_mutex> handlerLock(handlerMutex_);
425 7 : uint32_t commIdIndex = communicatorImplLite->GetCommIdIndex();
426 7 : auto kernelParamIter = kernelParamMap_.find(commIdIndex);
427 7 : if (kernelParamIter == kernelParamMap_.end()) {
428 3 : HCCL_ERROR(
429 : "[%s]KernelParam is not found, commId %u, please execute HcclGetCommHandleByCtx first.", __func__,
430 : commIdIndex);
431 1 : return HCCL_E_PTR;
432 : }
433 6 : kernelParam_ = kernelParamIter->second;
434 6 : rankSize_ = communicatorImplLite->GetRankSize();
435 6 : myRank_ = communicatorImplLite->GetMyRank();
436 :
437 : // 恢复op算子信息，buffer
438 6 : if (OP_TYPE_MAP.find(data->opType) == OP_TYPE_MAP.end()) {
439 0 : HCCL_ERROR(
440 : "[%s]Args OP_TYPE_MAP not find data->opType %u, commId %u.", __func__, data->opType,
441 : communicatorImplLite->GetCommIdIndex());
442 0 : return HCCL_E_PARA;
443 : }
444 6 : if (kernelParam_->op.algOperator.opType != OP_TYPE_MAP.at(data->opType)) {
445 0 : HCCL_ERROR(
446 : "[%s]Args kernelParam_->op.algOperator.opType %s is not equal to data->opType %s, commId %u.", __func__,
447 : kernelParam_->op.algOperator.opType.Describe().c_str(), OP_TYPE_MAP.at(data->opType).Describe().c_str(),
448 : communicatorImplLite->GetCommIdIndex());
449 0 : return HCCL_E_PARA;
450 : }
451 18 : HCCL_INFO(
452 : "[%s]opHandle %p, commId %u, rankSize_ %u, myRank_ %u, opType %u", __func__, communicatorImplLite,
453 : communicatorImplLite->GetCommIdIndex(), rankSize_, myRank_, data->opType);
454 6 : auto ret = FillKernelParam(data);
455 6 : if (ret != HCCL_SUCCESS) {
456 3 : HCCL_ERROR(
457 : "[%s]FillKernelParam execute failed, ret %u, commId %u", __func__, ret,
458 : communicatorImplLite->GetCommIdIndex());
459 1 : return ret;
460 : }
461 5 : ret = FillCollOperatorMemInfo(kernelParam_->op.algOperator, kernelParam_->op, data);
462 5 : if (ret != HCCL_SUCCESS) {
463 0 : HCCL_ERROR(
464 : "[%s]FillCollOperatorMemInfo execute failed, ret %u, commId %u", __func__, ret,
465 : communicatorImplLite->GetCommIdIndex());
466 0 : return ret;
467 : }
468 5 : return HCCL_SUCCESS;
469 7 : }
470 :
471 5 : HcclResult AicpuUtils::RestoreOpRes(CommunicatorImplLite* communicatorImplLite)
472 : {
473 5 : std::shared_lock<std::shared_timed_mutex> sharedLock(handlerMutex_);
474 5 : communicatorImplLite->UpdateLocBuffer(kernelParam_);
475 :
476 5 : uint64_t beginTime = ProfGetCurCpuTimestamp();
477 5 : communicatorImplLite->SetDfxOpInfo(beginTime);
478 :
479 : // 使用op信息分配input，output
480 5 : communicatorImplLite->UpdateHDCommnicate(kernelParam_);
481 5 : communicatorImplLite->RegisterRtsqCallback();
482 5 : communicatorImplLite->SetIsCommReady(true);
483 5 : return HCCL_SUCCESS;
484 5 : }
485 :
486 5 : HcclResult AicpuUtils::ExecuteOp(CommunicatorImplLite* communicatorImplLite)
487 : {
488 5 : std::shared_lock<std::shared_timed_mutex> sharedLock(handlerMutex_);
489 : // 修改Orchestrate编排入参
490 5 : std::shared_ptr<InsQueue> insQueue = communicatorImplLite->GetInsQueue(kernelParam_);
491 5 : sharedLock.unlock();
492 5 : CHK_PTR_NULL(insQueue);
493 :
494 : // 执行算子指令队列&&报告任务信息&&报告算子信息
495 15 : HCCL_INFO("[%s]DevType is DEV_TYPE_950 or DEV_TYPE_960.", __func__);
496 5 : auto* executor = communicatorImplLite->GetInsExecutor();
497 5 : CHK_PTR_NULL(executor);
498 5 : executor->ExecuteV82(*insQueue, true);
499 :
500 5 : auto* reporter = communicatorImplLite->GetProfilingReporterLite();
501 5 : CHK_PTR_NULL(reporter);
502 5 : reporter->ReportAllTasks();
503 :
504 5 : auto* taskMgr = communicatorImplLite->GetMirrorTaskMgrLite();
505 5 : CHK_PTR_NULL(taskMgr);
506 5 : ProfilingHandlerLite::GetInstance().ReportHcclOpInfo(*(taskMgr->GetCurrDfxOpInfo()));
507 5 : return HCCL_SUCCESS;
508 5 : }
|
