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