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