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 "adapter_rts.h"
12 : #include "adapter_error_manager.h"
13 : #include "sal.h"
14 : #include "stream_pub.h"
15 :
16 : namespace hccl {
17 : // 默认构造函数只产生无效的Stream对象
18 12025 : Stream::Stream() : stream_(nullptr), device_id_(HCCL_DEVICE_NOT_SET), stream_owner_(false), streamId_(0),
19 6009 : isMainStream_(true), modeGotFlag_(false), streamMode_(0), sqId_(0), ctx_(nullptr), cqId_(0), logicCqid_(0) {}
20 :
21 2323 : Stream::Stream(const Stream &that)
22 6967 : : stream_(that.ptr()), device_id_(that.device_id_), stream_owner_(false), streamId_(that.streamId_),
23 2322 : isMainStream_(that.isMainStream_), modeGotFlag_(that.modeGotFlag_), streamMode_(that.streamMode_),
24 2322 : sqId_(that.sqId_), ctx_(that.ctx_), cqId_(that.cqId_), logicCqid_(that.logicCqid_),
25 2322 : sqeContext_(that.sqeContext_), cqeContext_(that.cqeContext_), streamInfo_(that.streamInfo_),
26 2323 : invalidFlag_(that.invalidFlag_) {} // 共享销毁标志, 副本与owner持有同一份
27 :
28 593 : Stream::Stream(Stream &&that)
29 1779 : : stream_(that.ptr()), device_id_(that.device_id_), stream_owner_(that.stream_owner_), streamId_(that.streamId_),
30 593 : isMainStream_(that.isMainStream_), modeGotFlag_(that.modeGotFlag_), streamMode_(that.streamMode_),
31 593 : sqId_(that.sqId_), ctx_(that.ctx_), cqId_(that.cqId_), logicCqid_(that.logicCqid_),
32 593 : sqeContext_(that.sqeContext_), cqeContext_(that.cqeContext_), streamInfo_(that.streamInfo_),
33 593 : invalidFlag_(std::move(that.invalidFlag_)) // 转移销毁标志所有权, 源置空避免源析构影响
34 : {
35 593 : that.stream_ = nullptr;
36 593 : that.device_id_ = HCCL_DEVICE_NOT_SET;
37 593 : that.stream_owner_ = false;
38 593 : that.streamId_ = 0;
39 593 : that.isMainStream_ = true;
40 593 : that.modeGotFlag_ = false;
41 593 : that.streamMode_ = 0;
42 593 : that.sqId_ = 0;
43 593 : that.cqId_ = 0;
44 593 : that.logicCqid_ = 0;
45 593 : that.sqeContext_ = nullptr;
46 593 : that.cqeContext_ = nullptr;
47 593 : that.streamInfo_.actualStreamId = 0;
48 593 : that.streamInfo_.logicCqId = 0;
49 593 : that.streamInfo_.sqBaseAddr = nullptr;
50 593 : that.streamInfo_.sqDepth = 0;
51 593 : that.streamInfo_.sqId = 0;
52 593 : }
53 :
54 686 : Stream::Stream(const StreamType streamType, bool isMainStream)
55 1370 : : stream_(nullptr), device_id_(HCCL_DEVICE_NOT_SET), stream_owner_(true), streamId_(0),
56 686 : isMainStream_(isMainStream), modeGotFlag_(false), streamMode_(0), sqId_(0), ctx_(nullptr), cqId_(0), logicCqid_(0)
57 : {
58 : HcclResult ret;
59 683 : aclrtStream rtStream = nullptr;
60 :
61 : // 申请rtStream
62 683 : if (streamType == StreamType::STREAM_TYPE_ONLINE) {
63 287 : ret = hrtStreamCreateWithFlags(&rtStream, HCCL_STREAM_PRIORITY_HIGH,
64 : ACL_STREAM_FAST_LAUNCH | ACL_STREAM_FAST_SYNC);
65 396 : } else if (streamType == StreamType::STREAM_TYPE_DEVICE) {
66 273 : ret = hrtStreamCreateWithFlags(&rtStream, HCCL_STREAM_PRIORITY_HIGH,
67 : ACL_STREAM_DEVICE_USE_ONLY);
68 : } else {
69 123 : ret = hrtStreamCreateWithFlags(&rtStream, HCCL_STREAM_PRIORITY_LOW,
70 : ACL_STREAM_PERSISTENT);
71 : }
72 :
73 688 : if (ret == HCCL_SUCCESS) {
74 683 : HCCL_DEBUG("rtStreamCreate ok, streamType[%d]", streamType);
75 683 : stream_ = const_cast<void *>(rtStream);
76 683 : InitStream();
77 683 : HCCL_INFO("Construct stream by stream type success, ptr[%p] ctx[%p], stream id[%d], cqId[%d], logicCqid[%d]",
78 : rtStream, ctx_, streamId_, cqId_, logicCqid_);
79 : } else {
80 55 : RPT_ENV_ERR(true, "EI0007", std::vector<std::string>({"resource_type", "resource_info"}), \
81 : std::vector<std::string>({"stream", std::string("StreamCreateWithFlags, streamType:") + std::to_string(uint32_t(streamType))}));
82 5 : HCCL_ERROR("[%s][%s]Construct stream by stream type failed, errNo[0x%016llx] rtStreamCreate error, ret[%d]",
83 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_RESOURCE.c_str(), HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
84 : }
85 688 : bool isSupportV2 = false;
86 688 : CHK_PRT_CONT(hrtGetHcclV2Support(&isSupportV2), HCCL_WARNING("[Stream] Can not check hccl version"));
87 688 : if (isSupportV2 && streamType == StreamType::STREAM_TYPE_ONLINE) {
88 73 : HcclResult setModeRet = hrtStreamSetMode(stream_, STREAM_MODE_STOP_ON_FAILURE);
89 73 : if (setModeRet != HCCL_SUCCESS) {
90 0 : HCCL_ERROR("[Stream][SetMode]Failed to set stream mode, errNo[0x%016llx], ret[%d], stream id[%d]",
91 : HCCL_ERROR_CODE(setModeRet), setModeRet, streamId_);
92 : }
93 : }
94 703 : }
95 :
96 469 : Stream::Stream(const rtStream_t rtStream, bool isMainStream)
97 888 : : stream_(const_cast<void *>(rtStream)), device_id_(HCCL_DEVICE_NOT_SET), stream_owner_(false), streamId_(0),
98 469 : isMainStream_(isMainStream), modeGotFlag_(false), streamMode_(0), sqId_(0), ctx_(nullptr), cqId_(0), logicCqid_(0)
99 : {
100 425 : InitStream();
101 476 : }
102 :
103 500 : Stream::Stream(const HcclComStreamInfo &streamInfo, bool isMainStream)
104 500 : : stream_(static_cast<void *>(streamInfo.sqBaseAddr)), device_id_(HCCL_DEVICE_NOT_SET), stream_owner_(false),
105 500 : streamId_(streamInfo.actualStreamId), isMainStream_(isMainStream), modeGotFlag_(false), streamMode_(0),
106 500 : sqId_(streamInfo.sqId), ctx_(nullptr)
107 : {
108 500 : SetStreamInfo(streamInfo);
109 500 : }
110 :
111 10786 : Stream::~Stream()
112 : {
113 10575 : DestroyStream();
114 10812 : }
115 :
116 430 : Stream &Stream::operator=(const Stream &that)
117 : {
118 430 : if (&that != this) {
119 431 : stream_ = that.ptr();
120 428 : device_id_ = that.device_id_;
121 428 : stream_owner_ = false;
122 428 : taskLogicInfo_ = that.taskLogicInfo_;
123 392 : streamId_ = that.streamId_;
124 392 : isMainStream_ = that.isMainStream_;
125 392 : modeGotFlag_ = that.modeGotFlag_;
126 392 : streamMode_ = that.streamMode_;
127 392 : sqId_ = that.sqId_;
128 392 : ctx_ = that.ctx_;
129 392 : cqId_ = that.cqId_;
130 392 : logicCqid_ = that.logicCqid_;
131 392 : sqeContext_ = that.sqeContext_;
132 392 : cqeContext_ = that.cqeContext_;
133 392 : streamInfo_.actualStreamId = that.streamInfo_.actualStreamId;
134 392 : streamInfo_.logicCqId = that.streamInfo_.logicCqId;
135 392 : streamInfo_.sqBaseAddr = that.streamInfo_.sqBaseAddr;
136 392 : streamInfo_.sqDepth = that.streamInfo_.sqDepth;
137 392 : streamInfo_.sqId = that.streamInfo_.sqId;
138 392 : invalidFlag_ = that.invalidFlag_; // 共享销毁标志, 副本与owner持有同一份
139 : }
140 410 : return *this;
141 : }
142 :
143 582 : Stream Stream::operator=(Stream &&that)
144 : {
145 582 : if (&that != this) {
146 582 : stream_ = that.stream_;
147 582 : device_id_ = that.device_id_;
148 582 : stream_owner_ = that.stream_owner_;
149 582 : taskLogicInfo_ = that.taskLogicInfo_;
150 582 : streamId_ = that.streamId_;
151 582 : isMainStream_ = that.isMainStream_;
152 582 : modeGotFlag_ = that.modeGotFlag_;
153 582 : streamMode_ = that.streamMode_;
154 582 : sqId_ = that.sqId_;
155 582 : ctx_ = that.ctx_;
156 582 : cqId_ = that.cqId_;
157 582 : logicCqid_ = that.logicCqid_;
158 582 : sqeContext_ = that.sqeContext_;
159 582 : cqeContext_ = that.cqeContext_;
160 582 : streamInfo_.actualStreamId = that.streamInfo_.actualStreamId;
161 582 : streamInfo_.logicCqId = that.streamInfo_.logicCqId;
162 582 : streamInfo_.sqBaseAddr = that.streamInfo_.sqBaseAddr;
163 582 : streamInfo_.sqDepth = that.streamInfo_.sqDepth;
164 582 : streamInfo_.sqId = that.streamInfo_.sqId;
165 582 : invalidFlag_ = std::move(that.invalidFlag_); // 转移销毁标志所有权
166 : }
167 :
168 582 : that.stream_ = nullptr;
169 582 : that.device_id_ = HCCL_DEVICE_NOT_SET;
170 582 : that.stream_owner_ = false;
171 582 : that.taskLogicInfo_ = taskLogicInfo_;
172 582 : that.streamId_ = 0;
173 582 : that.isMainStream_ = isMainStream_;
174 582 : that.modeGotFlag_ = modeGotFlag_;
175 582 : that.streamMode_ = streamMode_;
176 582 : that.sqId_ = sqId_;
177 582 : that.ctx_ = ctx_;
178 582 : that.cqId_ = cqId_;
179 582 : that.logicCqid_ = logicCqid_;
180 582 : that.sqeContext_ = nullptr;
181 582 : that.cqeContext_ = nullptr;
182 582 : that.streamInfo_.actualStreamId = streamInfo_.actualStreamId;
183 582 : that.streamInfo_.logicCqId = streamInfo_.logicCqId;
184 582 : that.streamInfo_.sqBaseAddr = nullptr;
185 582 : that.streamInfo_.sqDepth = streamInfo_.sqDepth;
186 582 : that.streamInfo_.sqId = streamInfo_.sqId;
187 582 : return *this;
188 : }
189 :
190 10572 : void Stream::DestroyStream()
191 : {
192 : // owner销毁stream前先标记invalid, 让持有副本的dispatcher能感知并跳过, 避免访问悬空的sqeContext_
193 : // 必须在hrtStreamDestroy之前置位, 确保并发遍历streamMap_的线程在此期间读到true
194 10572 : if (invalidFlag_ != nullptr) {
195 9390 : invalidFlag_->store(true, std::memory_order_relaxed);
196 : }
197 : // 销毁stream
198 10607 : if (stream_owner_ && stream_ != nullptr) {
199 : // stream需要在原ctx上销毁
200 673 : aclrtContext ctxTmp = nullptr;
201 673 : HcclResult ret = hrtCtxGetCurrent(&ctxTmp);
202 673 : bool needChangeCtx = (ret == HCCL_SUCCESS && ctx_ != nullptr);
203 673 : if (needChangeCtx) {
204 670 : ret = hrtCtxSetCurrent(ctx_);
205 670 : HCCL_INFO("Switch Ctx ret[%d], curCtx[%p], setCtx[%p], stream id[%d]",
206 : ret, ctxTmp, ctx_, streamId_);
207 : }
208 673 : ret = hrtStreamDestroy(stream_);
209 673 : HCCL_RUN_INFO("[HCCL_TRACE]StreamDestroy, streamPtr[%p], stream id[%d]", stream_, streamId_);
210 673 : if (ret != HCCL_SUCCESS) {
211 0 : HCCL_WARNING("errNo[0x%016llx] hrtStreamDestroy error, ret[%d]",
212 : HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
213 : }
214 673 : if (needChangeCtx) {
215 670 : ret = hrtCtxSetCurrent(ctxTmp);
216 670 : HCCL_INFO("Restore Ctx ret[%d], setCtx[%p], stream id[%d]",
217 : ret, ctxTmp, streamId_);
218 : }
219 : }
220 10607 : }
221 :
222 2 : void Stream::SetEmpty()
223 : {
224 2 : DestroyStream();
225 2 : stream_ = nullptr;
226 2 : device_id_ = HCCL_DEVICE_NOT_SET;
227 2 : stream_owner_ = false;
228 2 : streamId_ = 0;
229 2 : isMainStream_ = true;
230 2 : sqId_ = 0;
231 2 : ctx_ = nullptr;
232 2 : cqId_ = 0;
233 2 : logicCqid_ = 0;
234 2 : }
235 :
236 1108 : HcclResult Stream::InitStream()
237 : {
238 1108 : if (stream_ != nullptr) {
239 1108 : HcclResult ret = hrtGetStreamId(stream_, streamId_);
240 1154 : if (ret != HCCL_SUCCESS) {
241 0 : SetEmpty();
242 0 : HCCL_ERROR("[InitStream]Failed to get the streamId through the rtstream, errNo[0x%016llx]" \
243 : "hrtGetStreamId error, ret[%d]", HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
244 0 : return HCCL_E_INTERNAL;
245 : }
246 :
247 1154 : ret = hrtStreamGetSqid(stream_, &(sqId_));
248 1156 : if (ret != HCCL_SUCCESS) {
249 2 : SetEmpty();
250 2 : HCCL_ERROR("[InitStream]Failed to get the sqId through the rtstream, errNo[0x%016llx]" \
251 : "hrtStreamGetSqid error, ret[%d]", HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
252 2 : return HCCL_E_INTERNAL;
253 : }
254 1154 : (void)hrtCtxGetCurrent(&ctx_);
255 :
256 1153 : ret = hrtStreamGetCqid(stream_, &(cqId_), &(logicCqid_));
257 1157 : if (ret != HCCL_SUCCESS) {
258 0 : SetEmpty();
259 0 : HCCL_ERROR("[InitStream]Failed to get the cqId through the rtstream, errNo[0x%016llx]" \
260 : "hrtStreamGetCqid error, ret[%d]", HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
261 0 : return HCCL_E_INTERNAL;
262 : }
263 : }
264 1157 : return HCCL_SUCCESS;
265 : }
266 :
267 106 : HcclResult Stream::SetMode(const uint64_t stmMode)
268 : {
269 106 : HcclResult ret = hrtStreamSetMode(stream_, stmMode);
270 106 : if (ret != HCCL_SUCCESS) {
271 1 : HCCL_ERROR("[Stream][SetMode]errNo[0x%016llx] hrtStreamSetMode error, ret[%d]",
272 : HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
273 1 : return HCCL_E_INTERNAL;
274 : }
275 105 : return HCCL_SUCCESS;
276 : }
277 :
278 99 : HcclResult Stream::GetMode(uint64_t *const stmMode)
279 : {
280 99 : if (modeGotFlag_ == false) {
281 99 : HcclResult ret = hrtStreamGetMode(stream_, &streamMode_);
282 99 : if (ret != HCCL_SUCCESS) {
283 0 : HCCL_ERROR("[Stream][GetMode]errNo[0x%016llx] hrtStreamGetMode error, ret[%d]",
284 : HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret);
285 0 : return HCCL_E_INTERNAL;
286 : }
287 : }
288 99 : *stmMode = streamMode_;
289 99 : return HCCL_SUCCESS;
290 : }
291 :
292 0 : void Stream::PushTaskLogicInfo(TaskLogicInfo &taskLogicInfo)
293 : {
294 0 : taskLogicInfo_.push(taskLogicInfo);
295 0 : HCCL_INFO("[PushTaskLogicInfo] stream[%p], taskLogicType[%d], taskLogicFuncType[%d], taskLogicInfo size[%d]",
296 : stream_, taskLogicInfo.taskLogicCmd.taskLogicType, taskLogicInfo.taskFuncType, taskLogicInfo_.size());
297 0 : }
298 :
299 0 : HcclResult Stream::PopTaskLogicInfo(TaskLogicInfo &taskLogicInfo)
300 : {
301 0 : if (taskLogicInfo_.size() > 0) {
302 0 : taskLogicInfo = taskLogicInfo_.front();
303 0 : HCCL_INFO("[PopTaskLogicInfo] stream[%p], taskLogicType[%d], taskLogicFuncType[%d], taskLogicInfo size[%d]",
304 : stream_, taskLogicInfo.taskLogicCmd.taskLogicType, taskLogicInfo.taskFuncType, taskLogicInfo_.size());
305 0 : taskLogicInfo_.pop();
306 0 : return HCCL_SUCCESS;
307 : }
308 0 : return HCCL_E_NOT_FOUND;
309 : }
310 :
311 20 : HcclResult Stream::GetNextSqeBufferAddr(uint8_t *&sqeBufferAddr, uint8_t *&sqeTypeAddr, uint8_t *&sqeDfxInfoAddr,
312 : uint16_t &taskId)
313 : {
314 20 : if (UNLIKELY(sqeContext_ == nullptr)) {
315 0 : HCCL_ERROR("[Stream][GetNextSqeBufferAddr] Sqe context is null");
316 0 : return HCCL_E_INTERNAL;
317 : }
318 20 : auto &buff = sqeContext_->buffer;
319 20 : if (UNLIKELY(buff.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
320 0 : HCCL_INFO("[Stream][GetNextSqeBufferAddr] Sqe index to 2048, need clear");
321 0 : if (buff.sqeCnt != 0) {
322 0 : HCCL_ERROR("[Stream][GetNextSqeBufferAddr] Sqe index to 2048, but sqeCnt is not 0");
323 0 : return HCCL_E_INTERNAL;
324 : }
325 0 : CHK_RET(ClearLocalBuff());
326 : }
327 20 : sqeBufferAddr = buff.localBuff + buff.tailSqeIdx * HCCL_SQE_SIZE;
328 20 : sqeTypeAddr = &buff.sqeType[buff.tailSqeIdx];
329 20 : sqeDfxInfoAddr = reinterpret_cast<uint8_t*>(&buff.dfxInfo[buff.tailSqeIdx]);
330 :
331 20 : buff.profTimestap[buff.tailSqeIdx] = ProfGetCurCpuTimestamp();
332 20 : taskId = buff.tailSqeTaskId;
333 :
334 20 : HCCL_DEBUG("[Stream][GetNextSqeBufferAddr] streamId: %u Get next idx:%u, taskId:%u, flipNum:%u",
335 : streamInfo_.actualStreamId, buff.tailSqeIdx, taskId, buff.filpNum);
336 20 : if (UNLIKELY(buff.tailSqeTaskId == UINT16_MAX)) {
337 0 : buff.filpNum++;
338 0 : HCCL_WARNING("[Stream][GetNextSqeBufferAddr] Sqe context cur taskId is uint16_max");
339 : }
340 20 : buff.tailSqeTaskId++;
341 20 : buff.sqeCnt++;
342 20 : buff.tailSqeIdx++;
343 20 : return HCCL_SUCCESS;
344 : }
345 :
346 496 : HcclResult Stream::InitSqAndCqeContext(uint32_t sqHead, uint32_t sqTail, SqCqeContext* context)
347 : {
348 496 : CHK_PTR_NULL(context);
349 496 : sqeContext_ = &context->sqContext;
350 496 : CHK_PTR_NULL(sqeContext_);
351 496 : cqeContext_ = &context->cqeContext;
352 496 : CHK_PTR_NULL(cqeContext_);
353 :
354 496 : auto &buff = sqeContext_->buffer;
355 496 : buff.sqHead = sqHead;
356 496 : buff.sqTail = sqTail;
357 496 : cqeContext_->cqeStatus = 0;
358 496 : HCCL_INFO("%s success, streamId:%u, sqHead:%u, sqTail:%u, context:%p", __func__, streamId_, sqHead, sqTail, context);
359 496 : return HCCL_SUCCESS;
360 : }
361 :
362 6 : HcclResult Stream::ClearLocalBuff()
363 : {
364 6 : CHK_PTR_NULL(sqeContext_);
365 1 : auto &buff = sqeContext_->buffer;
366 1 : if (memset_s(buff.localBuff, sizeof(buff.localBuff), 0, buff.tailSqeIdx * HCCL_SQE_SIZE) != EOK) {
367 0 : HCCL_ERROR("[Stream][ClearLocalBuff] clear local buff failed");
368 0 : return HCCL_E_MEMORY;
369 : }
370 1 : if (memset_s(buff.sqeType, sizeof(buff.sqeType), 0, buff.tailSqeIdx) != EOK) {
371 0 : HCCL_ERROR("[Stream][ClearLocalBuff] clear sqe type failed");
372 0 : return HCCL_E_MEMORY;
373 : }
374 1 : if (memset_s(buff.addInfo, sizeof(buff.addInfo), 0, buff.tailSqeIdx) != EOK) {
375 0 : HCCL_ERROR("[Stream][ClearLocalBuff] clear add info failed");
376 0 : return HCCL_E_MEMORY;
377 : }
378 1 : buff.sqeCnt = 0;
379 1 : buff.tailSqeIdx = 0;
380 :
381 1 : if (cqeContext_ != nullptr && memset_s(cqeContext_, sizeof(ErrCqeContext), 0, sizeof(ErrCqeContext)) != EOK) {
382 0 : HCCL_ERROR("[Stream][ClearLocalBuff] clear cqe context failed");
383 0 : return HCCL_E_MEMORY;
384 : }
385 1 : return HCCL_SUCCESS;
386 : }
387 :
388 0 : HcclResult Stream::SetCqeContext(const ErrCqeContext &cqeContext)
389 : {
390 0 : CHK_PTR_NULL(cqeContext_);
391 0 : *cqeContext_ = cqeContext;
392 0 : return HCCL_SUCCESS;
393 : }
394 :
395 0 : HcclResult Stream::GetCqeContext(ErrCqeContext &cqeContext)
396 : {
397 0 : CHK_PTR_NULL(cqeContext_);
398 0 : cqeContext = *cqeContext_;
399 0 : return HCCL_SUCCESS;
400 : }
401 :
402 5 : HcclResult Stream::GetStreamInfo(const HcclComStreamInfo *&streamInfo)
403 : {
404 5 : streamInfo = &streamInfo_;
405 5 : return HCCL_SUCCESS;
406 : }
407 : } // namespace hccl
|
