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