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