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),
72 0 : taskId(taskId),
73 0 : errorCode(errorCode),
74 0 : sqeType(sqeType)
75 0 : {}
76 : };
77 :
78 12 : struct SqCqeContext {
79 : HcclSqeContext sqContext;
80 : ErrCqeContext cqeContext;
81 : };
82 :
83 : /*
84 : * NOTE : hccl中, 节点内device间的link都有自己的event. 当前约定:
85 : * link对象作为发送方时record自己的event
86 : * link对象作为接收方时wait发送方的event
87 : */
88 : class Stream {
89 : public:
90 : explicit Stream();
91 : Stream(const Stream& that);
92 : Stream(Stream&& that);
93 : // 基于类型构造Stream，是stream owner
94 : explicit Stream(const StreamType streamType, bool isMainStream = false);
95 : // 使用rtStream构造Stream，不是stream owner
96 : explicit Stream(const rtStream_t rtStream, bool isMainStream = true);
97 : // 基于HcclComStreamInfo信息构造stream，不是stream owner
98 : explicit Stream(const HcclComStreamInfo& streamInfo, bool isMainStream = false);
99 :
100 : virtual ~Stream();
101 :
102 : // 初始化sq和cq资源
103 : HcclResult InitSqAndCqeContext(uint32_t sqHead, uint32_t sqTail, SqCqeContext* context);
104 :
105 : // 保存一个逻辑task信息
106 : void PushTaskLogicInfo(TaskLogicInfo& taskLogicInfo);
107 : // 获取一个逻辑task信息
108 : HcclResult PopTaskLogicInfo(TaskLogicInfo& taskLogicInfo);
109 : // 设置stream模式
110 : HcclResult SetMode(const uint64_t stmMode);
111 : // 获取stream模式
112 : HcclResult GetMode(uint64_t* const stmMode);
113 : // 获取sqebuffer
114 : HcclResult
115 : GetNextSqeBufferAddr(uint8_t*& sqeBufferAddr, uint8_t*& sqeTypeAddr, uint8_t*& sqeDfxInfoAddr, uint16_t& taskId);
116 : HcclResult GetStreamInfo(const HcclComStreamInfo*& streamInfo); // deprecated
117 66 : inline const HcclComStreamInfo& GetHcclStreamInfo() { return streamInfo_; }
118 :
119 : HcclResult SetCqeContext(const ErrCqeContext& cqeContext);
120 : HcclResult GetCqeContext(ErrCqeContext& cqeContext);
121 :
122 : // 提供获取裸指针的接口，接口调用耗时优于获取智能指针的接口
123 56 : inline HcclSqeContext* GetSqeContextPtr() { return sqeContext_; }
124 :
125 : HcclResult ClearLocalBuff();
126 :
127 : // 设置流的主、从流属性信息
128 47 : inline bool IsMainStream() { return isMainStream_; }
129 :
130 : Stream& operator=(const Stream& that);
131 : Stream operator=(Stream&& that);
132 :
133 : // "bool"运算符(可执行if(object){...}的操作判断该Stream对象是否有效)
134 149 : operator bool() const { return stream_ != nullptr; }
135 :
136 : // 判断stream是否已销毁 (streamMap_中的副本通过此接口感知原stream销毁, 避免访问悬空的sqeContext_)
137 : // 原stream销毁时将invalidFlag_置true, 所有共享同一invalidFlag_的拷贝副本均能感知
138 10 : bool IsInvalid() const { return invalidFlag_ != nullptr && invalidFlag_->load(std::memory_order_relaxed); }
139 :
140 : // 取地址
141 4641 : void* ptr() const { return stream_; }
142 1291 : s32 id() const { return streamId_; }
143 189 : u32 sqId() const { return sqId_; }
144 : void* stream_;
145 :
146 119 : u32 logicCqId() const { return logicCqid_; }
147 :
148 119 : u32 cqId() const { return cqId_; }
149 :
150 : protected:
151 : private:
152 : /* 非无效的构造函数 */
153 : void DestroyStream();
154 : void SetEmpty();
155 : HcclResult InitStream();
156 :
157 : /* stream所属的device, 当前由用户来操作device, 代码中不再指定device */
158 : s32 device_id_;
159 :
160 : /* 标记stream_是否是本对象是否申请，如果有stream不是用户传入, 而是代码申请的, 在析构时需要销毁 */
161 : bool stream_owner_;
162 :
163 : /* stram所编排的逻辑task信息 */
164 : std::queue<TaskLogicInfo> taskLogicInfo_;
165 :
166 : s32 streamId_;
167 : /* stream的type主流：true， 从流：false */
168 : bool isMainStream_;
169 :
170 : bool modeGotFlag_;
171 : uint64_t streamMode_;
172 :
173 : u32 sqId_;
174 : HcclRtContext ctx_;
175 : u32 cqId_;
176 : u32 logicCqid_;
177 :
178 : HcclSqeContext* sqeContext_ = nullptr; // device侧写sqe时使用的信息
179 : ErrCqeContext* cqeContext_ = nullptr; // device侧写cqe时使用的信息
180 : HcclComStreamInfo streamInfo_; // device侧写stream时使用的信息
181 :
182 : // stream销毁标志, 通过shared_ptr在所有拷贝副本间共享
183 : // 仅stream_owner_=true的owner析构时置true, 副本(IsInvalid调用方)只读
184 : // 配合std::atomic保证多线程下dispatcher遍历streamMap_与原stream销毁的并发安全
185 : std::shared_ptr<std::atomic<bool>> invalidFlag_ = std::make_shared<std::atomic<bool>>(false);
186 :
187 500 : void SetStreamInfo(const HcclComStreamInfo& streamInfo)
188 : {
189 500 : streamInfo_.actualStreamId = streamInfo.actualStreamId;
190 500 : streamInfo_.sqId = streamInfo.sqId;
191 500 : streamInfo_.sqDepth = streamInfo.sqDepth;
192 500 : streamInfo_.sqBaseAddr = streamInfo.sqBaseAddr;
193 500 : streamInfo_.logicCqId = streamInfo.logicCqId;
194 500 : }
195 : };
196 : } // namespace hccl
197 :
198 : #endif /* STREAM_PUB_H */
|
