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 "aicpu_ts_thread.h"
12 : #include "hccl_common.h"
13 : #include "aicpu/aicpu_hccl_sqcq.h"
14 : #include "device_capacity.h"
15 :
16 : namespace hccl {
17 110 : AicpuTsThread::AicpuTsThread(StreamType streamType, uint32_t notifyNum, const NotifyLoadType notifyLoadType)
18 110 : : streamType_(streamType),
19 110 : notifyNum_(notifyNum),
20 110 : notifyLoadType_(notifyLoadType)
21 110 : {}
22 :
23 202 : AicpuTsThread::AicpuTsThread(const std::string& uniqueIdStr) : uniqueIdStr_(uniqueIdStr) {}
24 :
25 589 : AicpuTsThread::~AicpuTsThread() { DeInit(); }
26 :
27 207 : HcclResult AicpuTsThread::Init()
28 : {
29 207 : CHK_RET(GetRunSideIsDevice(isDeviceSide_));
30 207 : if (!isDeviceSide_) {
31 : // host侧申请资源
32 109 : HCCL_INFO("HcclThread::%s, is hostside", __func__);
33 109 : return HostInit();
34 : } else {
35 : // device侧反序列化，恢复资源
36 98 : HCCL_INFO("HcclThread::%s, is DeviceSide", __func__);
37 98 : return DeviceInit();
38 : }
39 : }
40 :
41 312 : HcclResult AicpuTsThread::DeInit()
42 : {
43 312 : streamType_ = StreamType::STREAM_TYPE_RESERVED;
44 312 : notifyNum_ = 0;
45 312 : if (stream_) {
46 109 : stream_->SetInvalidFlag();
47 : }
48 312 : stream_ = nullptr;
49 312 : notifys_.clear();
50 312 : uniqueIdStr_ = std::string();
51 312 : devType_ = DevType::DEV_TYPE_COUNT;
52 312 : return HCCL_SUCCESS;
53 : }
54 :
55 99 : std::string& AicpuTsThread::GetUniqueId()
56 : {
57 99 : if (!uniqueIdStr_.empty()) {
58 1 : return uniqueIdStr_;
59 : }
60 :
61 98 : return UpdateUniqueId();
62 : }
63 :
64 105 : std::string& AicpuTsThread::UpdateUniqueId()
65 : {
66 : // 序列化信息
67 105 : std::ostringstream oss;
68 105 : oss.write(reinterpret_cast<const char_t*>(&streamType_), sizeof(streamType_));
69 105 : oss.write(reinterpret_cast<const char_t*>(¬ifyLoadType_), sizeof(notifyLoadType_));
70 105 : oss.write(reinterpret_cast<const char_t*>(&devId_), sizeof(devId_));
71 105 : oss.write(reinterpret_cast<const char_t*>(¬ifyNum_), sizeof(notifyNum_));
72 :
73 105 : HcclStreamParam streamParam;
74 105 : streamParam.streamInfo.streamIds = stream_->id();
75 105 : streamParam.streamInfo.sqIds = stream_->sqId();
76 105 : streamParam.streamInfo.cqIds = stream_->cqId();
77 105 : streamParam.streamInfo.logicCqids = stream_->logicCqId();
78 105 : streamParam.sqCqContextAddr = reinterpret_cast<uint64_t>(sqCqeContext_.ptr());
79 105 : streamParam.sqCqContextSize = sqCqeContext_.size();
80 105 : oss.write(reinterpret_cast<const char_t*>(&streamParam), sizeof(streamParam));
81 :
82 105 : HcclResult ret = HCCL_SUCCESS;
83 428 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
84 : HcclSignalInfo notifyInfo;
85 323 : ret = notifys_[idx]->GetNotifyData(notifyInfo);
86 323 : if (ret != HCCL_SUCCESS) {
87 0 : HCCL_ERROR("[AicpuTsThread][UpdateUniqueId]GetNotifyData failed, ret[%d]", ret);
88 0 : uniqueIdStr_ = std::string();
89 0 : return uniqueIdStr_;
90 : }
91 323 : HCCL_INFO(
92 : "[AicpuTsThread][UpdateUniqueId]get local notify data success, resId[%u], tsId[%d], devId[%u]",
93 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
94 323 : oss.write(reinterpret_cast<const char_t*>(¬ifyInfo), sizeof(notifyInfo));
95 : }
96 105 : HCCL_DEBUG("[AicpuTsThread][UpdateUniqueId] stream[%p], notifyNum[%u]", stream_->ptr(), notifyNum_);
97 :
98 105 : uniqueIdStr_ = oss.str();
99 105 : return uniqueIdStr_;
100 105 : }
101 :
102 : #ifdef CCL_KERNEL_AICPU
103 : HcclResult AicpuTsThread::BuildComStreamInfo(const HcclStreamInfo& streamInfo, HcclComStreamInfo& comStreamInfo) const
104 : {
105 : comStreamInfo.sqId = streamInfo.sqIds;
106 : comStreamInfo.actualStreamId = streamInfo.streamIds;
107 : comStreamInfo.logicCqId = streamInfo.logicCqids;
108 : u64 sqAddr = 0;
109 : CHK_RET(QuerySqBaseAddr(devId_, streamInfo.sqIds, sqAddr));
110 : comStreamInfo.sqBaseAddr = reinterpret_cast<void*>(sqAddr);
111 : if (comStreamInfo.sqBaseAddr == nullptr) {
112 : HCCL_ERROR("[AicpuTsThread::InitStream] sqe base addr ptr is null.");
113 : return HCCL_E_PARA;
114 : }
115 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_DEPTH, comStreamInfo.sqDepth));
116 : HCCL_DEBUG(
117 : "[AicpuTsThread::InitStream] get stream data success, "
118 : "streamId[%d], sqId[%d], logicCqId[%u], sqDepth[%u]",
119 : comStreamInfo.actualStreamId, comStreamInfo.sqId, comStreamInfo.logicCqId, comStreamInfo.sqDepth);
120 : return HCCL_SUCCESS;
121 : }
122 : #endif
123 :
124 2 : HcclResult AicpuTsThread::InitStream([[maybe_unused]] HcclStreamParam& streamParam)
125 : {
126 : #ifdef CCL_KERNEL_AICPU
127 : HcclStreamInfo& streamInfo = streamParam.streamInfo;
128 :
129 : static bool isCustom = false;
130 : static bool init = false;
131 :
132 : if (UNLIKELY(!init)) {
133 : uint32_t cpType = DEVDRV_PROCESS_CPTYPE_MAX;
134 : unsigned int hostpid = 0;
135 : CHK_RET(HrtHalDrvQueryProcessHostPid(getpid(), nullptr, nullptr, &hostpid, &cpType));
136 : isCustom = cpType == static_cast<uint32_t>(DEVDRV_PROCESS_CP2) ? true : false;
137 : init = true;
138 : }
139 : HcclResult ret = hrtHalResourceIdRestore(devId_, 0, DRV_STREAM_ID, streamInfo.streamIds, 0);
140 : // custom进程需要恢复stream资源, custom进程调用失败直接报错，aicpu进程调用失败做兼容性处理
141 : if (ret == HCCL_E_NOT_SUPPORT) {
142 : CHK_PRT_RET(
143 : isCustom,
144 : HCCL_ERROR(
145 : "%s hrtHalResourceIdRestore fail, drv not support, custom[%d], ret[%d]", __func__, isCustom, ret),
146 : HCCL_E_DRV);
147 : } else if (ret != HCCL_SUCCESS) {
148 : HCCL_ERROR("%s hrtHalResourceIdRestore fail, ret[%d]", __func__, ret);
149 : return HCCL_E_DRV;
150 : }
151 :
152 : HcclComStreamInfo comStreamInfo{};
153 : CHK_RET(BuildComStreamInfo(streamInfo, comStreamInfo));
154 :
155 : stream_.reset(new (std::nothrow) Stream(comStreamInfo));
156 : CHK_SMART_PTR_NULL(stream_);
157 :
158 : // 初始化stream的sqeContext
159 : SqCqeContext* sqCqeContext = reinterpret_cast<SqCqeContext*>(streamParam.sqCqContextAddr);
160 : uint64_t sqCqContextSize = streamParam.sqCqContextSize;
161 : if (sqCqeContext == nullptr || sqCqContextSize != sizeof(SqCqeContext)) {
162 : HCCL_ERROR(
163 : "%s fail, sqCqeContext[%p] is null or size[%llu] is not equal to SqCqeContext size[%llu]", __func__,
164 : sqCqeContext, sqCqContextSize, sizeof(SqCqeContext));
165 : return HCCL_E_PARA;
166 : }
167 : sqCqeContext_ = DeviceMem::create(reinterpret_cast<void*>(sqCqeContext), sqCqContextSize);
168 :
169 : uint32_t sqTail = 0;
170 : uint32_t sqHead = 0;
171 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
172 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
173 : HCCL_DEBUG("[AicpuTsThread::InitStream] sqHead[%u], sqTail[%u]", sqHead, sqTail);
174 :
175 : ret = stream_->InitSqAndCqeContext(sqHead, sqTail, sqCqeContext);
176 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("%s InitSqAndCqeContext failed", __func__), ret);
177 : HCCL_INFO("%s success, streamId[%d]", __func__, stream_->id());
178 : #endif
179 2 : return HCCL_SUCCESS;
180 : }
181 :
182 95 : HcclResult AicpuTsThread::InitStreamLite(HcclStreamInfo& streamParam, uint32_t hostPhyId)
183 : {
184 : // 在aicpu侧查询cqe时，需要使用logicCqids，而不是cqIds
185 95 : EXCEPTION_CATCH(
186 : pImpl_ = std::make_unique<Hccl::IAicpuTsThread>(
187 : streamParam.streamIds, streamParam.sqIds, hostPhyId, streamParam.logicCqids),
188 : return HCCL_E_PTR);
189 95 : return HCCL_SUCCESS;
190 : }
191 :
192 19 : uint32_t AicpuTsThread::GetNotifyNum() const { return notifyNum_; }
193 :
194 9 : LocalNotify* AicpuTsThread::GetNotify(uint32_t index) const
195 : {
196 9 : if (UNLIKELY(index >= notifyNum_)) {
197 1 : HCCL_ERROR(
198 : "[AicpuTsThread][GetNotify] notifyNum[%u], index[%u] out of range[0, %u)", notifyNum_, index, notifyNum_);
199 1 : return nullptr;
200 : }
201 8 : return notifys_[index].get();
202 : }
203 :
204 : // A3 Stream
205 16 : Stream* AicpuTsThread::GetStream() const { return stream_.get(); }
206 :
207 0 : void AicpuTsThread::LaunchTask() const
208 : {
209 0 : pImpl_->LaunchTask();
210 0 : return;
211 : }
212 :
213 0 : void AicpuTsThread::TryLaunchTask() const
214 : {
215 0 : pImpl_->TryLaunchTask();
216 0 : return;
217 : }
218 :
219 : // Local Data Plane Functions
220 0 : HcclResult AicpuTsThread::LocalNotifyWait([[maybe_unused]] uint32_t notifyId) const
221 : {
222 0 : HCCL_ERROR("[AicpuTsThread][%s] without timeout not support", __func__);
223 0 : return HCCL_E_NOT_SUPPORT;
224 : }
225 :
226 1 : HcclResult AicpuTsThread::LocalNotifyRecord(uint32_t notifyId) const
227 : {
228 1 : void* streamLitePtr = GetStreamLitePtr();
229 1 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(streamLitePtr);
230 1 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
231 1 : u32 taskId = rtsq->GetTaskId();
232 :
233 1 : CHK_RET(pImpl_->NotifyRecordLoc(notifyId));
234 :
235 1 : auto* slot = streamLite->NextTaskSlot();
236 1 : slot->taskType = Hccl::TaskParamTypeVal::TASK_NOTIFY_RECORD;
237 1 : slot->sqId = streamLite->GetSqId();
238 1 : slot->taskId = taskId;
239 1 : const void* notifyRecordOpInfo = streamLite->GetLatestDfxOpInfo();
240 1 : slot->dfxOpInfo = (notifyRecordOpInfo != nullptr) ? reinterpret_cast<u64>(notifyRecordOpInfo) : DFX_INVALID_U64;
241 1 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
242 1 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
243 1 : slot->channelHandle = DFX_INVALID_U64;
244 1 : slot->taskPara.Notify.sqeAddr = rtsq->GetSqeAddr();
245 :
246 1 : return HCCL_SUCCESS;
247 : }
248 :
249 : HcclResult
250 0 : AicpuTsThread::LocalNotifyRecord([[maybe_unused]] ThreadHandle dstThread, [[maybe_unused]] uint32_t dstNotifyIdx) const
251 : {
252 0 : HCCL_ERROR("[AicpuTsThread][%s]not support", __func__);
253 0 : return HCCL_E_NOT_SUPPORT;
254 : }
255 :
256 4 : HcclResult AicpuTsThread::LocalNotifyWait(uint32_t notifyId, uint32_t timeout) const
257 : {
258 4 : void* streamLitePtr = GetStreamLitePtr();
259 4 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(streamLitePtr);
260 4 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
261 4 : u32 taskId = rtsq->GetTaskId();
262 :
263 4 : CHK_RET(pImpl_->NotifyWait(notifyId, timeout));
264 :
265 4 : auto* slot = streamLite->NextTaskSlot();
266 4 : slot->taskType = Hccl::TaskParamTypeVal::TASK_NOTIFY_WAIT;
267 4 : slot->sqId = streamLite->GetSqId();
268 4 : slot->taskId = taskId;
269 4 : const void* notifyWaitOpInfo = streamLite->GetLatestDfxOpInfo();
270 4 : slot->dfxOpInfo = (notifyWaitOpInfo != nullptr) ? reinterpret_cast<u64>(notifyWaitOpInfo) : DFX_INVALID_U64;
271 4 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
272 4 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
273 4 : slot->channelHandle = DFX_INVALID_U64;
274 4 : slot->taskPara.Notify.sqeAddr = rtsq->GetSqeAddr();
275 :
276 4 : return HCCL_SUCCESS;
277 : }
278 :
279 : template <typename Operation, typename ReportOp>
280 : HcclResult
281 6 : AicpuTsThread::LocalProcess(void* dst, const void* src, uint64_t size, Operation&& op, ReportOp&& reportOp) const
282 : {
283 6 : void* streamLitePtr = GetStreamLitePtr();
284 6 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(streamLitePtr);
285 6 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
286 :
287 6 : uint64_t dstAddr = reinterpret_cast<uint64_t>(dst);
288 6 : uint64_t srcAddr = reinterpret_cast<uint64_t>(src);
289 6 : uint8_t* dstByte = static_cast<uint8_t*>(dst);
290 6 : const uint8_t* srcByte = static_cast<const uint8_t*>(src);
291 :
292 6 : uint64_t remainSize = size;
293 6 : uint64_t doneSize = 0;
294 :
295 16 : while (remainSize > 0) {
296 10 : uint64_t realSize = remainSize > SDMA_SEND_MAX_SIZE ? SDMA_SEND_MAX_SIZE : remainSize;
297 10 : u64 beginTime = ProfGetCurCpuTimestamp();
298 10 : u32 taskId = rtsq->GetTaskId();
299 :
300 10 : CHK_RET(op(dstAddr + doneSize, srcAddr + doneSize, realSize));
301 10 : CHK_RET(
302 : reportOp(dstByte + doneSize, srcByte + doneSize, realSize, beginTime, taskId, streamLite->GetSqId(), rtsq));
303 :
304 10 : doneSize += realSize;
305 10 : remainSize -= realSize;
306 : }
307 6 : return HCCL_SUCCESS;
308 : }
309 :
310 3 : HcclResult AicpuTsThread::LocalCopy(void* dst, const void* src, uint64_t size) const
311 : {
312 6 : return LocalProcess(
313 : dst, src, size,
314 0 : [this](uint64_t dst, uint64_t src, uint64_t size) {
315 4 : return pImpl_->SdmaCopy(dst, src, size);
316 : },
317 6 : [this](
318 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t size,
319 : [[maybe_unused]] uint64_t beginTime, uint32_t taskId, uint32_t sqId, Hccl::RtsqBase* rtsq) {
320 4 : Hccl::StreamLite* sl = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
321 4 : auto* slot = sl->NextTaskSlot();
322 4 : slot->taskType = Hccl::TaskParamTypeVal::TASK_SDMA;
323 4 : slot->sqId = sqId;
324 4 : slot->taskId = taskId;
325 4 : const void* copyOpInfo = sl->GetLatestDfxOpInfo();
326 4 : slot->dfxOpInfo = (copyOpInfo != nullptr) ? reinterpret_cast<u64>(copyOpInfo) : DFX_INVALID_U64;
327 4 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
328 4 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
329 4 : slot->channelHandle = DFX_INVALID_U64;
330 4 : slot->taskPara.Dma.sqeAddr = rtsq->GetSqeAddr();
331 4 : return HCCL_SUCCESS;
332 6 : });
333 : }
334 :
335 3 : HcclResult AicpuTsThread::LocalReduce(
336 : void* dst, const void* src, uint64_t size, HcommDataType dataType, HcommReduceOp reduceOp) const
337 : {
338 3 : uint32_t dataTypeRaw = static_cast<uint32_t>(dataType);
339 3 : uint32_t reduceOpRaw = static_cast<uint32_t>(reduceOp);
340 6 : return LocalProcess(
341 : dst, src, size,
342 0 : [this, &dataTypeRaw, &reduceOpRaw](uint64_t d, uint64_t s, uint64_t size) {
343 6 : return pImpl_->SdmaReduce(d, s, size, dataTypeRaw, reduceOpRaw);
344 : },
345 6 : [this, &dataType, &reduceOp](
346 : void* dst, const void* src, uint64_t size, [[maybe_unused]] uint64_t beginTime, uint32_t taskId,
347 : uint32_t sqId, Hccl::RtsqBase* rtsq) {
348 6 : Hccl::StreamLite* sl = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
349 6 : auto* slot = sl->NextTaskSlot();
350 6 : slot->taskType = Hccl::TaskParamTypeVal::TASK_REDUCE_INLINE;
351 6 : slot->sqId = sqId;
352 6 : slot->taskId = taskId;
353 6 : const void* reduceOpInfo = sl->GetLatestDfxOpInfo();
354 6 : slot->dfxOpInfo = (reduceOpInfo != nullptr) ? reinterpret_cast<u64>(reduceOpInfo) : DFX_INVALID_U64;
355 6 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
356 6 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
357 6 : slot->channelHandle = DFX_INVALID_U64;
358 6 : slot->taskPara.Reduce.sqeAddr = rtsq->GetSqeAddr();
359 6 : slot->taskPara.Reduce.srcAddr = reinterpret_cast<u64>(src);
360 6 : slot->taskPara.Reduce.dstAddr = reinterpret_cast<u64>(dst);
361 6 : slot->taskPara.Reduce.size = size;
362 6 : slot->taskPara.Reduce.notifyId = INVALID_U32;
363 6 : slot->taskPara.Reduce.reduceOp = static_cast<u8>(reduceOp);
364 6 : return HCCL_SUCCESS;
365 6 : });
366 : }
367 :
368 : // Private functions
369 109 : HcclResult AicpuTsThread::HostInit()
370 : {
371 109 : CHK_PRT_RET(
372 : !uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][Init]not support init with uniqueId on host"),
373 : HCCL_E_NOT_SUPPORT);
374 : s32 deviceLogicId;
375 109 : CHK_RET(hrtGetDevice(&deviceLogicId));
376 109 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devId_));
377 109 : CHK_RET(hrtGetDeviceType(devType_));
378 109 : if (rtStream_ == nullptr) {
379 109 : stream_.reset(new (std::nothrow) Stream(streamType_));
380 109 : CHK_SMART_PTR_NULL(stream_);
381 109 : rtStream_ = stream_->ptr();
382 : }
383 :
384 326 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
385 219 : notifys_.emplace_back(nullptr);
386 219 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
387 219 : CHK_SMART_PTR_NULL(notifys_[idx]);
388 219 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
389 217 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
390 22 : CHK_RET(notifys_[idx]->SetIpc());
391 : }
392 : }
393 :
394 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
395 107 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
396 96 : notifys_.emplace_back(nullptr);
397 96 : notifys_[notifyNum_].reset(new (std::nothrow) LocalNotify());
398 96 : CHK_SMART_PTR_NULL(notifys_[notifyNum_]);
399 96 : CHK_RET(notifys_[notifyNum_]->Init(NotifyLoadType::HOST_NOTIFY));
400 96 : notifyNum_ += 1;
401 : }
402 :
403 107 : if (streamType_ == StreamType::STREAM_TYPE_DEVICE && devType_ != DevType::DEV_TYPE_950
404 10 : && devType_ != DevType::DEV_TYPE_960) {
405 10 : uint64_t size = sizeof(SqCqeContext);
406 10 : sqCqeContext_ = DeviceMem::alloc(size);
407 10 : CHK_PTR_NULL(sqCqeContext_.ptr());
408 10 : CHK_RET(hrtMemSet(sqCqeContext_.ptr(), size, size));
409 : }
410 107 : return HCCL_SUCCESS;
411 : }
412 :
413 98 : HcclResult AicpuTsThread::DeviceInit()
414 : {
415 98 : CHK_PRT_RET(uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][Init]uniqueIdStr is empty"), HCCL_E_INTERNAL);
416 97 : std::istringstream iss(uniqueIdStr_);
417 97 : CHK_RET(hrtGetDeviceType(devType_));
418 97 : uint32_t hostPhyId = 0;
419 97 : iss.read(reinterpret_cast<char_t*>(&streamType_), sizeof(streamType_));
420 97 : iss.read(reinterpret_cast<char_t*>(¬ifyLoadType_), sizeof(notifyLoadType_));
421 97 : HCCL_INFO("[AicpuTsThread][Init]streamType[%d], notifyLoadType[%d].", streamType_, notifyLoadType_);
422 97 : iss.read(reinterpret_cast<char_t*>(&hostPhyId), sizeof(hostPhyId));
423 97 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(hostPhyId, &devId_));
424 96 : iss.read(reinterpret_cast<char_t*>(¬ifyNum_), sizeof(notifyNum_));
425 :
426 96 : HcclStreamParam streamParam;
427 96 : iss.read(reinterpret_cast<char_t*>(&streamParam), sizeof(streamParam));
428 : // 91095初始化streamlite，初始化rtsq接口
429 96 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
430 94 : CHK_RET(InitStreamLite(streamParam.streamInfo, hostPhyId));
431 94 : } else {
432 2 : CHK_RET(InitStream(streamParam));
433 : }
434 :
435 96 : notifys_.reserve(notifyNum_);
436 :
437 375 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
438 279 : notifys_.emplace_back(nullptr);
439 : HcclSignalInfo notifyInfo;
440 279 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
441 279 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
442 279 : CHK_SMART_PTR_NULL(notifys_[idx]);
443 279 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
444 275 : CHK_RET(notifys_[idx]->InitNotifyLite(notifyInfo));
445 275 : HCCL_INFO(
446 : "[AicpuTsThread][Init]local notifyLite init success, resId[%u], devId[%u]", notifyInfo.resId,
447 : notifyInfo.devId);
448 275 : } else {
449 4 : CHK_RET(notifys_[idx]->Init(notifyInfo, notifyLoadType_));
450 4 : HCCL_INFO(
451 : "[AicpuTsThread][Init]local notifyLite init success, resId[%u], tsId:%d, devId[%u]", notifyInfo.resId,
452 : notifyInfo.tsId, notifyInfo.devId);
453 : }
454 : }
455 :
456 96 : return HCCL_SUCCESS;
457 97 : }
458 :
459 0 : HcclResult AicpuTsThread::GetSqHeadAndTail([[maybe_unused]] uint32_t& sqHead, [[maybe_unused]] uint32_t& sqTail)
460 : {
461 : #ifdef CCL_KERNEL_AICPU
462 :
463 : uint32_t sqIds = pImpl_->GetSqId();
464 :
465 : CHK_RET(QuerySqStatusByType(devId_, sqIds, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
466 : CHK_RET(QuerySqStatusByType(devId_, sqIds, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
467 : #endif
468 0 : return HCCL_SUCCESS;
469 : }
470 :
471 0 : bool AicpuTsThread::GetMaster() const { return isMaster_; }
472 :
473 0 : void AicpuTsThread::SetIsMaster(bool isMaster) { isMaster_ = isMaster; }
474 :
475 7 : HcclResult AicpuTsThread::SupplementNotify(uint32_t notifyNum)
476 : {
477 7 : HCCL_INFO("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
478 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
479 7 : u32 beginIdx = notifyNum_;
480 7 : u32 allNotifyNum = notifyNum_ + notifyNum;
481 7 : u32 endIdx = allNotifyNum - 1;
482 7 : notifys_.resize(allNotifyNum);
483 7 : if ((devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) && notifyNum_ > 0) {
484 7 : beginIdx--;
485 7 : CHK_SMART_PTR_NULL(notifys_[beginIdx]);
486 7 : notifys_[endIdx] = std::move(notifys_[beginIdx]);
487 : }
488 :
489 17 : for (uint32_t idx = beginIdx; idx < endIdx; idx++) {
490 10 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
491 10 : CHK_SMART_PTR_NULL(notifys_[idx]);
492 10 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
493 10 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
494 0 : CHK_RET(notifys_[idx]->SetIpc());
495 : }
496 10 : notifyNum_++;
497 : }
498 :
499 7 : uniqueIdStr_.clear();
500 7 : UpdateUniqueId();
501 7 : return HCCL_SUCCESS;
502 : }
503 :
504 2 : HcclResult AicpuTsThread::GetNotifyByUniqueId(u32& notifyNum, std::string& notifyDesc)
505 : {
506 2 : CHK_PRT_RET(
507 : uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][GetNotifyByUniqueId]uniqueIdStr is empty"), HCCL_E_INTERNAL);
508 2 : std::istringstream iss(uniqueIdStr_);
509 2 : StreamType streamType = StreamType::STREAM_TYPE_RESERVED;
510 2 : NotifyLoadType notifyLoadType = NotifyLoadType::HOST_NOTIFY;
511 2 : uint32_t hostPhyId = 0;
512 2 : HcclStreamParam streamParam;
513 2 : iss.read(reinterpret_cast<char_t*>(&streamType), sizeof(streamType));
514 2 : iss.read(reinterpret_cast<char_t*>(¬ifyLoadType), sizeof(notifyLoadType));
515 2 : iss.read(reinterpret_cast<char_t*>(&hostPhyId), sizeof(hostPhyId));
516 2 : iss.read(reinterpret_cast<char_t*>(¬ifyNum), sizeof(notifyNum));
517 2 : iss.read(reinterpret_cast<char_t*>(&streamParam), sizeof(streamParam));
518 :
519 : // 序列化信息
520 2 : std::ostringstream oss;
521 11 : for (uint32_t idx = 0; idx < notifyNum; idx++) {
522 : HcclSignalInfo notifyInfo;
523 9 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
524 9 : HCCL_INFO(
525 : "[AicpuTsThread][%s]get local notify data success, resId[%u], tsId:%d, devId[%u]", __func__,
526 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
527 9 : oss.write(reinterpret_cast<const char_t*>(¬ifyInfo), sizeof(notifyInfo));
528 : }
529 :
530 2 : notifyDesc = oss.str();
531 2 : return HCCL_SUCCESS;
532 2 : }
533 :
534 2 : HcclResult AicpuTsThread::SupplementNotify(u32 notifyNum, const std::string& notifyDesc)
535 : {
536 2 : if (notifyNum <= notifyNum_) {
537 0 : HCCL_WARNING("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
538 0 : return HCCL_SUCCESS;
539 : }
540 2 : HCCL_INFO("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
541 :
542 2 : std::istringstream iss(notifyDesc);
543 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
544 2 : u32 beginIdx = notifyNum_;
545 2 : u32 endIdx = notifyNum - 1;
546 2 : notifys_.resize(notifyNum);
547 2 : if ((devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) && notifyNum_ > 0) {
548 1 : beginIdx--;
549 1 : CHK_SMART_PTR_NULL(notifys_[beginIdx]);
550 1 : notifys_[endIdx] = std::move(notifys_[beginIdx]);
551 1 : HCCL_INFO(
552 : "[AicpuTsThread][SupplementNotify]notifyId[%u] beginIdx[%u], endIdx[%u]", notifys_[endIdx]->notifyId_,
553 : beginIdx, endIdx);
554 : }
555 4 : for (uint32_t idx = 0; idx < beginIdx; idx++) {
556 : HcclSignalInfo notifyInfo;
557 2 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
558 2 : HCCL_INFO(
559 : "[AicpuTsThread][SupplementNotify]skip init, resId[%u], tsId:%d, devId[%u]", notifyInfo.resId,
560 : notifyInfo.tsId, notifyInfo.devId);
561 : }
562 :
563 7 : for (uint32_t idx = beginIdx; idx < endIdx; idx++) {
564 : HcclSignalInfo notifyInfo;
565 5 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
566 5 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
567 5 : CHK_SMART_PTR_NULL(notifys_[idx]);
568 5 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
569 4 : CHK_RET(notifys_[idx]->InitNotifyLite(notifyInfo));
570 4 : HCCL_INFO(
571 : "[AicpuTsThread][SupplementNotify]local notifyLite init success, resId[%u], devId[%u]",
572 : notifyInfo.resId, notifyInfo.devId);
573 4 : } else {
574 1 : CHK_RET(notifys_[idx]->Init(notifyInfo, notifyLoadType_));
575 1 : HCCL_INFO(
576 : "[AicpuTsThread][SupplementNotify]local notifyLite init success, resId[%u], tsId:%d, devId[%u]",
577 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
578 : }
579 5 : notifyNum_++;
580 : }
581 2 : return HCCL_SUCCESS;
582 2 : }
583 :
584 2 : HcclResult AicpuTsThread::SetCheckExecStatusCallback(std::function<HcclResult(bool)> callback)
585 : {
586 2 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
587 2 : CHK_PTR_NULL(streamLite);
588 2 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
589 2 : CHK_PTR_NULL(rtsq);
590 2 : rtsq->SetCheckExecStatusCallback(callback);
591 2 : return HCCL_SUCCESS;
592 : }
593 :
594 1 : Hccl::TaskInfoCircularQueue* AicpuTsThread::GetTaskInfos() const
595 : {
596 1 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
597 1 : if (streamLite == nullptr) {
598 0 : return nullptr;
599 : }
600 1 : return streamLite->GetTaskInfos();
601 : }
602 :
603 1 : HcclResult AicpuTsThread::GetTaskInfoCount(u32& count) const
604 : {
605 1 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
606 1 : CHK_PTR_NULL(streamLite);
607 1 : Hccl::TaskInfoCircularQueue* taskInfos = streamLite->GetTaskInfos();
608 1 : count = static_cast<u32>(taskInfos->GetCount());
609 1 : return HCCL_SUCCESS;
610 : }
611 :
612 2 : void AicpuTsThread::SetReportStreamTaskCallback(std::function<void(Hccl::TaskInfoCircularQueue*)> callback)
613 : {
614 2 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
615 2 : if (streamLite != nullptr) {
616 2 : streamLite->SetReportStreamTaskCallback(std::move(callback));
617 : }
618 2 : }
619 :
620 0 : void AicpuTsThread::SetGetLatestDfxOpInfoCallback(std::function<const void*()> callback)
621 : {
622 0 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
623 0 : if (streamLite != nullptr) {
624 0 : streamLite->SetGetLatestDfxOpInfoCallback(std::move(callback));
625 : }
626 0 : }
627 : } // namespace hccl
|
