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 : #ifndef STREAM_PUB_H
12 : #define STREAM_PUB_H
13 :
14 : #include <atomic>
15 : #include <memory>
16 : #include <mutex>
17 : #include <queue>
18 : #include <hccl/hccl_types.h>
19 :
20 : #include "hccl/base.h"
21 : #include "task_logic_info_pub.h"
22 : #include "hccl_common.h"
23 :
24 : namespace hccl {
25 : constexpr int32_t HCCL_STREAM_PRIORITY_LOW = 0;
26 : constexpr int32_t HCCL_STREAM_PRIORITY_HIGH = 0;
27 : constexpr uint32_t HCCL_SQE_MAX_CNT = 2048U; // 一次mc2算子一个流上最大下发sqe数量
28 : constexpr u32 HCCL_SQE_SIZE = 64U;
29 : constexpr uint32_t STREAM_MODE_STOP_ON_FAILURE = 1; // 配置流失败模式为遇错即停
30 :
31 : enum class StreamType {
32 : STREAM_TYPE_OFFLINE = 0,
33 : STREAM_TYPE_ONLINE = 1,
34 : STREAM_TYPE_DEVICE = 2,
35 : STREAM_TYPE_RESERVED = 3
36 : };
37 : struct AicpuDfxInfo {
38 : uint32_t remoteRank = INVALID_VALUE_RANKID; // 记录算子Remote RANKID
39 : uint32_t opRingBufferIdx = 0; // index, 0~OpInfoRingMax
40 : uint32_t notifyId = INVALID_VALUE_RANKID;
41 : };
42 : struct SqeRingBuffer {
43 : uint8_t localBuff[HCCL_SQE_SIZE * HCCL_SQE_MAX_CNT] {0}; // local buffer
44 : uint8_t rtsMirrorBuffer[HCCL_SQE_SIZE * HCCL_SQE_MAX_CNT] {0}; // launch buffer
45 : uint8_t rtsqSqeType[HCCL_SQE_MAX_CNT] {0}; // 记录SQE类型,用于后续解析
46 : uint8_t sqeType[HCCL_SQE_MAX_CNT] {0}; // 记录SQE类型,用于后续解析
47 : AicpuDfxInfo dfxInfo[HCCL_SQE_MAX_CNT]; //
48 : AicpuDfxInfo rtsDfxInfo[HCCL_SQE_MAX_CNT];
49 : uint32_t addInfo[HCCL_SQE_MAX_CNT] {0}; // 记录额外信息
50 : uint64_t profTimestap[HCCL_SQE_MAX_CNT] {0}; // profiling上报
51 : uint16_t tailSqeTaskId = 0; // 最后一个sqe对应的taskId
52 : uint16_t tailSqeIdx = 0; // 最后一个sqe对应的数组idx
53 : uint16_t sqeCnt = 0; // 当前轮保存的sqe数量(下发后重置)
54 : uint32_t sqHead = 0;
55 : uint32_t sqTail = 0;
56 : uint16_t filpNum = 0;
57 : };
58 :
59 24 : struct HcclSqeContext {
60 : SqeRingBuffer buffer; // SqeRingBuffer[AC_MAX_RANK_NUM]
61 : bool inited = false;
62 : };
63 :
64 : struct ErrCqeContext {
65 : u32 cqeStatus;
66 : u16 taskId;
67 : u8 errorCode;
68 : u8 sqeType;
69 46 : ErrCqeContext() : cqeStatus(0), taskId(0), errorCode(0), sqeType(0) {}
70 0 : ErrCqeContext(u32 cqeStatus, u16 taskId, u8 errorCode, u8 sqeType) :
71 0 : cqeStatus(cqeStatus), taskId(taskId), errorCode(errorCode), sqeType(sqeType) {}
72 : };
73 :
74 12 : struct SqCqeContext {
75 : HcclSqeContext sqContext;
76 : ErrCqeContext cqeContext;
77 : };
78 :
79 : /*
80 : * NOTE : hccl中, 节点内device间的link都有自己的event. 当前约定:
81 : * link对象作为发送方时record自己的event
82 : * link对象作为接收方时wait发送方的event
83 : */
84 : class Stream {
85 : public:
86 : explicit Stream();
87 : Stream(const Stream &that);
88 : Stream(Stream &&that);
89 : // 基于类型构造Stream，是stream owner
90 : explicit Stream(const StreamType streamType, bool isMainStream = false);
91 : // 使用rtStream构造Stream，不是stream owner
92 : explicit Stream(const rtStream_t rtStream, bool isMainStream = true);
93 : // 基于HcclComStreamInfo信息构造stream，不是stream owner
94 : explicit Stream(const HcclComStreamInfo &streamInfo, bool isMainStream = false);
95 :
96 : virtual ~Stream();
97 :
98 : // 初始化sq和cq资源
99 : HcclResult InitSqAndCqeContext(uint32_t sqHead, uint32_t sqTail, SqCqeContext* context);
100 :
101 : // 保存一个逻辑task信息
102 : void PushTaskLogicInfo(TaskLogicInfo &taskLogicInfo);
103 : // 获取一个逻辑task信息
104 : HcclResult PopTaskLogicInfo(TaskLogicInfo &taskLogicInfo);
105 : // 设置stream模式
106 : HcclResult SetMode(const uint64_t stmMode);
107 : // 获取stream模式
108 : HcclResult GetMode(uint64_t *const stmMode);
109 : // 获取sqebuffer
110 : HcclResult GetNextSqeBufferAddr(uint8_t *&sqeBufferAddr, uint8_t *&sqeTypeAddr, uint8_t *&sqeDfxInfoAddr,
111 : uint16_t &taskId);
112 : HcclResult GetStreamInfo(const HcclComStreamInfo *&streamInfo); // deprecated
113 66 : inline const HcclComStreamInfo &GetHcclStreamInfo()
114 : {
115 66 : return streamInfo_;
116 : }
117 :
118 : HcclResult SetCqeContext(const ErrCqeContext &cqeContext);
119 : HcclResult GetCqeContext(ErrCqeContext &cqeContext);
120 :
121 : // 提供获取裸指针的接口，接口调用耗时优于获取智能指针的接口
122 56 : inline HcclSqeContext* GetSqeContextPtr()
123 : {
124 56 : return sqeContext_;
125 : }
126 :
127 : HcclResult ClearLocalBuff();
128 :
129 : // 设置流的主、从流属性信息
130 47 : inline bool IsMainStream()
131 : {
132 47 : return isMainStream_;
133 : }
134 :
135 : Stream &operator=(const Stream &that);
136 : Stream operator=(Stream &&that);
137 :
138 : // "bool"运算符(可执行if(object){...}的操作判断该Stream对象是否有效)
139 149 : operator bool() const
140 : {
141 149 : return stream_ != nullptr;
142 : }
143 :
144 : // 判断stream是否已销毁 (streamMap_中的副本通过此接口感知原stream销毁, 避免访问悬空的sqeContext_)
145 : // 原stream销毁时将invalidFlag_置true, 所有共享同一invalidFlag_的拷贝副本均能感知
146 10 : bool IsInvalid() const
147 : {
148 10 : return invalidFlag_ != nullptr && invalidFlag_->load(std::memory_order_relaxed);
149 : }
150 :
151 : // 取地址
152 4627 : void *ptr() const
153 : {
154 4627 : return stream_;
155 : }
156 1287 : s32 id() const
157 : {
158 1287 : return streamId_;
159 : }
160 185 : u32 sqId() const
161 : {
162 185 : return sqId_;
163 : }
164 : void *stream_;
165 :
166 115 : u32 logicCqId() const
167 : {
168 115 : return logicCqid_;
169 : }
170 :
171 115 : u32 cqId() const
172 : {
173 115 : return cqId_;
174 : }
175 :
176 : protected:
177 : private:
178 : /* 非无效的构造函数 */
179 : void DestroyStream();
180 : void SetEmpty();
181 : HcclResult InitStream();
182 :
183 : /* stream所属的device, 当前由用户来操作device, 代码中不再指定device */
184 : s32 device_id_;
185 :
186 : /* 标记stream_是否是本对象是否申请，如果有stream不是用户传入, 而是代码申请的, 在析构时需要销毁 */
187 : bool stream_owner_;
188 :
189 : /* stram所编排的逻辑task信息 */
190 : std::queue<TaskLogicInfo> taskLogicInfo_;
191 :
192 : s32 streamId_;
193 : /* stream的type主流：true， 从流：false */
194 : bool isMainStream_;
195 :
196 : bool modeGotFlag_;
197 : uint64_t streamMode_;
198 :
199 : u32 sqId_;
200 : HcclRtContext ctx_;
201 : u32 cqId_;
202 : u32 logicCqid_;
203 :
204 : HcclSqeContext* sqeContext_ = nullptr; // device侧写sqe时使用的信息
205 : ErrCqeContext* cqeContext_ = nullptr; // device侧写cqe时使用的信息
206 : HcclComStreamInfo streamInfo_; // device侧写stream时使用的信息
207 :
208 : // stream销毁标志, 通过shared_ptr在所有拷贝副本间共享
209 : // 仅stream_owner_=true的owner析构时置true, 副本(IsInvalid调用方)只读
210 : // 配合std::atomic保证多线程下dispatcher遍历streamMap_与原stream销毁的并发安全
211 : std::shared_ptr<std::atomic<bool>> invalidFlag_ = std::make_shared<std::atomic<bool>>(false);
212 :
213 500 : void SetStreamInfo(const HcclComStreamInfo &streamInfo)
214 : {
215 500 : streamInfo_.actualStreamId = streamInfo.actualStreamId;
216 500 : streamInfo_.sqId = streamInfo.sqId;
217 500 : streamInfo_.sqDepth = streamInfo.sqDepth;
218 500 : streamInfo_.sqBaseAddr = streamInfo.sqBaseAddr;
219 500 : streamInfo_.logicCqId = streamInfo.logicCqId;
220 500 : }
221 : };
222 : } // namespace hccl
223 :
224 : #endif /* STREAM_PUB_H */
|
