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