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(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{0};
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(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 0 : HcclResult AicpuTsThread::LocalNotifyRecord(ThreadHandle dstThread, uint32_t dstNotifyIdx) const
247 : {
248 0 : HCCL_ERROR("[AicpuTsThread][%s]not support", __func__);
249 0 : return HCCL_E_NOT_SUPPORT;
250 : }
251 :
252 4 : HcclResult AicpuTsThread::LocalNotifyWait(uint32_t notifyId, uint32_t timeout) const
253 : {
254 4 : void* streamLitePtr = GetStreamLitePtr();
255 4 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(streamLitePtr);
256 4 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
257 4 : u32 taskId = rtsq->GetTaskId();
258 :
259 4 : CHK_RET(pImpl_->NotifyWait(notifyId, timeout));
260 :
261 4 : auto* slot = streamLite->NextTaskSlot();
262 4 : slot->taskType = Hccl::TaskParamTypeVal::TASK_NOTIFY_WAIT;
263 4 : slot->sqId = streamLite->GetSqId();
264 4 : slot->taskId = taskId;
265 4 : const void* notifyWaitOpInfo = streamLite->GetLatestDfxOpInfo();
266 4 : slot->dfxOpInfo = (notifyWaitOpInfo != nullptr) ? reinterpret_cast<u64>(notifyWaitOpInfo) : DFX_INVALID_U64;
267 4 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
268 4 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
269 4 : slot->channelHandle = DFX_INVALID_U64;
270 4 : slot->taskPara.Notify.sqeAddr = rtsq->GetSqeAddr();
271 :
272 4 : return HCCL_SUCCESS;
273 : }
274 :
275 : template <typename Operation, typename ReportOp>
276 : HcclResult
277 6 : AicpuTsThread::LocalProcess(void* dst, const void* src, uint64_t size, Operation&& op, ReportOp&& reportOp) const
278 : {
279 6 : void* streamLitePtr = GetStreamLitePtr();
280 6 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(streamLitePtr);
281 6 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
282 :
283 6 : uint64_t dstAddr = reinterpret_cast<uint64_t>(dst);
284 6 : uint64_t srcAddr = reinterpret_cast<uint64_t>(src);
285 6 : uint8_t* dstByte = static_cast<uint8_t*>(dst);
286 6 : const uint8_t* srcByte = static_cast<const uint8_t*>(src);
287 :
288 6 : uint64_t remainSize = size;
289 6 : uint64_t doneSize = 0;
290 :
291 16 : while (remainSize > 0) {
292 10 : uint64_t realSize = remainSize > SDMA_SEND_MAX_SIZE ? SDMA_SEND_MAX_SIZE : remainSize;
293 10 : u64 beginTime = ProfGetCurCpuTimestamp();
294 10 : u32 taskId = rtsq->GetTaskId();
295 :
296 10 : CHK_RET(op(dstAddr + doneSize, srcAddr + doneSize, realSize));
297 10 : CHK_RET(
298 : reportOp(dstByte + doneSize, srcByte + doneSize, realSize, beginTime, taskId, streamLite->GetSqId(), rtsq));
299 :
300 10 : doneSize += realSize;
301 10 : remainSize -= realSize;
302 : }
303 6 : return HCCL_SUCCESS;
304 : }
305 :
306 3 : HcclResult AicpuTsThread::LocalCopy(void* dst, const void* src, uint64_t size) const
307 : {
308 6 : return LocalProcess(
309 : dst, src, size,
310 0 : [this](uint64_t dst, uint64_t src, uint64_t size) {
311 4 : return pImpl_->SdmaCopy(dst, src, size);
312 : },
313 6 : [this](
314 : void* dst, const void* src, uint64_t size, uint64_t beginTime, uint32_t taskId, uint32_t sqId,
315 : Hccl::RtsqBase* rtsq) {
316 4 : Hccl::StreamLite* sl = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
317 4 : auto* slot = sl->NextTaskSlot();
318 4 : slot->taskType = Hccl::TaskParamTypeVal::TASK_SDMA;
319 4 : slot->sqId = sqId;
320 4 : slot->taskId = taskId;
321 4 : const void* copyOpInfo = sl->GetLatestDfxOpInfo();
322 4 : slot->dfxOpInfo = (copyOpInfo != nullptr) ? reinterpret_cast<u64>(copyOpInfo) : DFX_INVALID_U64;
323 4 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
324 4 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
325 4 : slot->channelHandle = DFX_INVALID_U64;
326 4 : slot->taskPara.Dma.sqeAddr = rtsq->GetSqeAddr();
327 4 : return HCCL_SUCCESS;
328 6 : });
329 : }
330 :
331 3 : HcclResult AicpuTsThread::LocalReduce(
332 : void* dst, const void* src, uint64_t size, HcommDataType dataType, HcommReduceOp reduceOp) const
333 : {
334 3 : uint32_t dataTypeRaw = static_cast<uint32_t>(dataType);
335 3 : uint32_t reduceOpRaw = static_cast<uint32_t>(reduceOp);
336 6 : return LocalProcess(
337 : dst, src, size,
338 0 : [this, &dataTypeRaw, &reduceOpRaw](uint64_t d, uint64_t s, uint64_t size) {
339 6 : return pImpl_->SdmaReduce(d, s, size, dataTypeRaw, reduceOpRaw);
340 : },
341 6 : [this, &dataType, &reduceOp](
342 : void* dst, const void* src, uint64_t size, uint64_t beginTime, uint32_t taskId, uint32_t sqId,
343 : Hccl::RtsqBase* rtsq) {
344 6 : Hccl::StreamLite* sl = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
345 6 : auto* slot = sl->NextTaskSlot();
346 6 : slot->taskType = Hccl::TaskParamTypeVal::TASK_REDUCE_INLINE;
347 6 : slot->sqId = sqId;
348 6 : slot->taskId = taskId;
349 6 : const void* reduceOpInfo = sl->GetLatestDfxOpInfo();
350 6 : slot->dfxOpInfo = (reduceOpInfo != nullptr) ? reinterpret_cast<u64>(reduceOpInfo) : DFX_INVALID_U64;
351 6 : slot->linkType = Hccl::DfxLinkTypeVal::LINK_ONCHIP;
352 6 : slot->transportType = static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL);
353 6 : slot->channelHandle = DFX_INVALID_U64;
354 6 : slot->taskPara.Reduce.sqeAddr = rtsq->GetSqeAddr();
355 6 : slot->taskPara.Reduce.srcAddr = reinterpret_cast<u64>(src);
356 6 : slot->taskPara.Reduce.dstAddr = reinterpret_cast<u64>(dst);
357 6 : slot->taskPara.Reduce.size = size;
358 6 : slot->taskPara.Reduce.notifyId = INVALID_U32;
359 6 : slot->taskPara.Reduce.reduceOp = static_cast<u8>(reduceOp);
360 6 : return HCCL_SUCCESS;
361 6 : });
362 : }
363 :
364 : // Private functions
365 109 : HcclResult AicpuTsThread::HostInit()
366 : {
367 109 : CHK_PRT_RET(
368 : !uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][Init]not support init with uniqueId on host"),
369 : HCCL_E_NOT_SUPPORT);
370 : s32 deviceLogicId;
371 109 : CHK_RET(hrtGetDevice(&deviceLogicId));
372 109 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devId_));
373 109 : CHK_RET(hrtGetDeviceType(devType_));
374 109 : if (rtStream_ == nullptr) {
375 109 : stream_.reset(new (std::nothrow) Stream(streamType_));
376 109 : CHK_SMART_PTR_NULL(stream_);
377 109 : rtStream_ = stream_->ptr();
378 : }
379 :
380 326 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
381 219 : notifys_.emplace_back(nullptr);
382 219 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
383 219 : CHK_SMART_PTR_NULL(notifys_[idx]);
384 219 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
385 217 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
386 22 : CHK_RET(notifys_[idx]->SetIpc());
387 : }
388 : }
389 :
390 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
391 107 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
392 96 : notifys_.emplace_back(nullptr);
393 96 : notifys_[notifyNum_].reset(new (std::nothrow) LocalNotify());
394 96 : CHK_SMART_PTR_NULL(notifys_[notifyNum_]);
395 96 : CHK_RET(notifys_[notifyNum_]->Init(NotifyLoadType::HOST_NOTIFY));
396 96 : notifyNum_ += 1;
397 : }
398 :
399 107 : if (streamType_ == StreamType::STREAM_TYPE_DEVICE && devType_ != DevType::DEV_TYPE_950
400 10 : && devType_ != DevType::DEV_TYPE_960) {
401 10 : uint64_t size = sizeof(SqCqeContext);
402 10 : sqCqeContext_ = DeviceMem::alloc(size);
403 10 : CHK_PTR_NULL(sqCqeContext_.ptr());
404 10 : CHK_RET(hrtMemSet(sqCqeContext_.ptr(), size, size));
405 : }
406 107 : return HCCL_SUCCESS;
407 : }
408 :
409 98 : HcclResult AicpuTsThread::DeviceInit()
410 : {
411 98 : CHK_PRT_RET(uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][Init]uniqueIdStr is empty"), HCCL_E_INTERNAL);
412 97 : std::istringstream iss(uniqueIdStr_);
413 97 : CHK_RET(hrtGetDeviceType(devType_));
414 97 : uint32_t hostPhyId = 0;
415 97 : iss.read(reinterpret_cast<char_t*>(&streamType_), sizeof(streamType_));
416 97 : iss.read(reinterpret_cast<char_t*>(¬ifyLoadType_), sizeof(notifyLoadType_));
417 97 : HCCL_INFO("[AicpuTsThread][Init]streamType[%d], notifyLoadType[%d].", streamType_, notifyLoadType_);
418 97 : iss.read(reinterpret_cast<char_t*>(&hostPhyId), sizeof(hostPhyId));
419 97 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(hostPhyId, &devId_));
420 96 : iss.read(reinterpret_cast<char_t*>(¬ifyNum_), sizeof(notifyNum_));
421 :
422 96 : HcclStreamParam streamParam;
423 96 : iss.read(reinterpret_cast<char_t*>(&streamParam), sizeof(streamParam));
424 : // 91095初始化streamlite，初始化rtsq接口
425 96 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
426 94 : CHK_RET(InitStreamLite(streamParam.streamInfo, hostPhyId));
427 94 : } else {
428 2 : CHK_RET(InitStream(streamParam));
429 : }
430 :
431 96 : notifys_.reserve(notifyNum_);
432 :
433 375 : for (uint32_t idx = 0; idx < notifyNum_; idx++) {
434 279 : notifys_.emplace_back(nullptr);
435 : HcclSignalInfo notifyInfo;
436 279 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
437 279 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
438 279 : CHK_SMART_PTR_NULL(notifys_[idx]);
439 279 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
440 275 : CHK_RET(notifys_[idx]->InitNotifyLite(notifyInfo));
441 275 : HCCL_INFO(
442 : "[AicpuTsThread][Init]local notifyLite init success, resId[%u], devId[%u]", notifyInfo.resId,
443 : notifyInfo.devId);
444 275 : } else {
445 4 : CHK_RET(notifys_[idx]->Init(notifyInfo, notifyLoadType_));
446 4 : HCCL_INFO(
447 : "[AicpuTsThread][Init]local notifyLite init success, resId[%u], tsId:%d, devId[%u]", notifyInfo.resId,
448 : notifyInfo.tsId, notifyInfo.devId);
449 : }
450 : }
451 :
452 96 : return HCCL_SUCCESS;
453 97 : }
454 :
455 0 : HcclResult AicpuTsThread::GetSqHeadAndTail(uint32_t& sqHead, uint32_t& sqTail)
456 : {
457 : #ifdef CCL_KERNEL_AICPU
458 :
459 : uint32_t sqIds = pImpl_->GetSqId();
460 :
461 : CHK_RET(QuerySqStatusByType(devId_, sqIds, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
462 : CHK_RET(QuerySqStatusByType(devId_, sqIds, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
463 : #endif
464 0 : return HCCL_SUCCESS;
465 : }
466 :
467 0 : bool AicpuTsThread::GetMaster() const { return isMaster_; }
468 :
469 0 : void AicpuTsThread::SetIsMaster(bool isMaster) { isMaster_ = isMaster; }
470 :
471 7 : HcclResult AicpuTsThread::SupplementNotify(uint32_t notifyNum)
472 : {
473 7 : HCCL_INFO("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
474 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
475 7 : u32 beginIdx = notifyNum_;
476 7 : u32 allNotifyNum = notifyNum_ + notifyNum;
477 7 : u32 endIdx = allNotifyNum - 1;
478 7 : notifys_.resize(allNotifyNum);
479 7 : if ((devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) && notifyNum_ > 0) {
480 7 : beginIdx--;
481 7 : CHK_SMART_PTR_NULL(notifys_[beginIdx]);
482 7 : notifys_[endIdx] = std::move(notifys_[beginIdx]);
483 : }
484 :
485 17 : for (uint32_t idx = beginIdx; idx < endIdx; idx++) {
486 10 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
487 10 : CHK_SMART_PTR_NULL(notifys_[idx]);
488 10 : CHK_RET(notifys_[idx]->Init(notifyLoadType_));
489 10 : if (devType_ != DevType::DEV_TYPE_950 && devType_ != DevType::DEV_TYPE_960) {
490 0 : CHK_RET(notifys_[idx]->SetIpc());
491 : }
492 10 : notifyNum_++;
493 : }
494 :
495 7 : uniqueIdStr_.clear();
496 7 : UpdateUniqueId();
497 7 : return HCCL_SUCCESS;
498 : }
499 :
500 2 : HcclResult AicpuTsThread::GetNotifyByUniqueId(u32& notifyNum, std::string& notifyDesc)
501 : {
502 2 : CHK_PRT_RET(
503 : uniqueIdStr_.empty(), HCCL_ERROR("[AicpuTsThread][GetNotifyByUniqueId]uniqueIdStr is empty"), HCCL_E_INTERNAL);
504 2 : std::istringstream iss(uniqueIdStr_);
505 2 : StreamType streamType = StreamType::STREAM_TYPE_RESERVED;
506 2 : NotifyLoadType notifyLoadType = NotifyLoadType::HOST_NOTIFY;
507 2 : uint32_t hostPhyId = 0;
508 2 : HcclStreamParam streamParam;
509 2 : iss.read(reinterpret_cast<char_t*>(&streamType), sizeof(streamType));
510 2 : iss.read(reinterpret_cast<char_t*>(¬ifyLoadType), sizeof(notifyLoadType));
511 2 : iss.read(reinterpret_cast<char_t*>(&hostPhyId), sizeof(hostPhyId));
512 2 : iss.read(reinterpret_cast<char_t*>(¬ifyNum), sizeof(notifyNum));
513 2 : iss.read(reinterpret_cast<char_t*>(&streamParam), sizeof(streamParam));
514 :
515 : // 序列化信息
516 2 : std::ostringstream oss;
517 11 : for (uint32_t idx = 0; idx < notifyNum; idx++) {
518 : HcclSignalInfo notifyInfo;
519 9 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
520 9 : HCCL_INFO(
521 : "[AicpuTsThread][%s]get local notify data success, resId[%u], tsId:%d, devId[%u]", __func__,
522 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
523 9 : oss.write(reinterpret_cast<const char_t*>(¬ifyInfo), sizeof(notifyInfo));
524 : }
525 :
526 2 : notifyDesc = oss.str();
527 2 : return HCCL_SUCCESS;
528 2 : }
529 :
530 2 : HcclResult AicpuTsThread::SupplementNotify(u32 notifyNum, const std::string& notifyDesc)
531 : {
532 2 : if (notifyNum <= notifyNum_) {
533 0 : HCCL_WARNING("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
534 0 : return HCCL_SUCCESS;
535 : }
536 2 : HCCL_INFO("[%s]supplement notifyNum[%u], notifyNum_[%u]", __func__, notifyNum, notifyNum_);
537 :
538 2 : std::istringstream iss(notifyDesc);
539 : // A5 aicpu场景thread多申请一个host类型notify，用于host&device同步
540 2 : u32 beginIdx = notifyNum_;
541 2 : u32 endIdx = notifyNum - 1;
542 2 : notifys_.resize(notifyNum);
543 2 : if ((devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) && notifyNum_ > 0) {
544 1 : beginIdx--;
545 1 : CHK_SMART_PTR_NULL(notifys_[beginIdx]);
546 1 : notifys_[endIdx] = std::move(notifys_[beginIdx]);
547 1 : HCCL_INFO(
548 : "[AicpuTsThread][SupplementNotify]notifyId[%u] beginIdx[%u], endIdx[%u]", notifys_[endIdx]->notifyId_,
549 : beginIdx, endIdx);
550 : }
551 4 : for (uint32_t idx = 0; idx < beginIdx; idx++) {
552 : HcclSignalInfo notifyInfo;
553 2 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
554 2 : HCCL_INFO(
555 : "[AicpuTsThread][SupplementNotify]skip init, resId[%u], tsId:%d, devId[%u]", notifyInfo.resId,
556 : notifyInfo.tsId, notifyInfo.devId);
557 : }
558 :
559 7 : for (uint32_t idx = beginIdx; idx < endIdx; idx++) {
560 : HcclSignalInfo notifyInfo;
561 5 : iss.read(reinterpret_cast<char_t*>(¬ifyInfo), sizeof(notifyInfo));
562 5 : notifys_[idx].reset(new (std::nothrow) LocalNotify());
563 5 : CHK_SMART_PTR_NULL(notifys_[idx]);
564 5 : if (devType_ == DevType::DEV_TYPE_950 || devType_ == DevType::DEV_TYPE_960) {
565 4 : CHK_RET(notifys_[idx]->InitNotifyLite(notifyInfo));
566 4 : HCCL_INFO(
567 : "[AicpuTsThread][SupplementNotify]local notifyLite init success, resId[%u], devId[%u]",
568 : notifyInfo.resId, notifyInfo.devId);
569 4 : } else {
570 1 : CHK_RET(notifys_[idx]->Init(notifyInfo, notifyLoadType_));
571 1 : HCCL_INFO(
572 : "[AicpuTsThread][SupplementNotify]local notifyLite init success, resId[%u], tsId:%d, devId[%u]",
573 : notifyInfo.resId, notifyInfo.tsId, notifyInfo.devId);
574 : }
575 5 : notifyNum_++;
576 : }
577 2 : return HCCL_SUCCESS;
578 2 : }
579 :
580 2 : HcclResult AicpuTsThread::SetCheckExecStatusCallback(std::function<HcclResult(bool)> callback)
581 : {
582 2 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
583 2 : CHK_PTR_NULL(streamLite);
584 2 : Hccl::RtsqBase* rtsq = streamLite->GetRtsq();
585 2 : CHK_PTR_NULL(rtsq);
586 2 : rtsq->SetCheckExecStatusCallback(callback);
587 2 : return HCCL_SUCCESS;
588 : }
589 :
590 1 : Hccl::TaskInfoCircularQueue* AicpuTsThread::GetTaskInfos() const
591 : {
592 1 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
593 1 : if (streamLite == nullptr) {
594 0 : return nullptr;
595 : }
596 1 : return streamLite->GetTaskInfos();
597 : }
598 :
599 1 : HcclResult AicpuTsThread::GetTaskInfoCount(u32& count) const
600 : {
601 1 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
602 1 : CHK_PTR_NULL(streamLite);
603 1 : Hccl::TaskInfoCircularQueue* taskInfos = streamLite->GetTaskInfos();
604 1 : count = static_cast<u32>(taskInfos->GetCount());
605 1 : return HCCL_SUCCESS;
606 : }
607 :
608 2 : void AicpuTsThread::SetReportStreamTaskCallback(std::function<void(Hccl::TaskInfoCircularQueue*)> callback)
609 : {
610 2 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
611 2 : if (streamLite != nullptr) {
612 2 : streamLite->SetReportStreamTaskCallback(std::move(callback));
613 : }
614 2 : }
615 :
616 0 : void AicpuTsThread::SetGetLatestDfxOpInfoCallback(std::function<const void*()> callback)
617 : {
618 0 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>(GetStreamLitePtr());
619 0 : if (streamLite != nullptr) {
620 0 : streamLite->SetGetLatestDfxOpInfoCallback(std::move(callback));
621 : }
622 0 : }
623 : } // namespace hccl
|
