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 HCCLV2_DEV_UB_CONNECTION_H
12 : #define HCCLV2_DEV_UB_CONNECTION_H
13 :
14 : #include "rma_connection.h"
15 : #include "op_mode.h"
16 : #include "orion_adapter_hccp.h"
17 : #include "../../../framework/env_config/env_config_v2.h"
18 : #include "tp_manager.h"
19 : #include "local_ub_rma_buffer.h"
20 : #include "stream.h"
21 : #include "task.h"
22 : #include "mc2_type.h"
23 : #include "hcomm/hcomm_res_entity_defs.h"
24 : #include <functional>
25 :
26 : namespace Hccl {
27 :
28 : class DevUbConnection : public RmaConnection {
29 : public:
30 : DevUbConnection(
31 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
32 121 : const bool devUsed = false, const HrtUbJfcMode jfcMode = HrtUbJfcMode::STARS_POLL,
33 : const IpAddress& locIpv4Addr = IpAddress(), const IpAddress& rmtIpv4Addr = IpAddress(),
34 : u8 qos = static_cast<u8>(UB_QOS_DEFAULT), CommEngine engine = COMM_ENGINE_RESERVED,
35 : u32 sqDepth = UB_SQ_DEPTH_NOT_SET);
36 : void Connect() override;
37 : RmaConnStatus GetStatus() override;
38 : bool Suspend() override;
39 :
40 : std::unique_ptr<Serializable> GetExchangeDto() override;
41 : void ParseRmtExchangeDto(const Serializable& rmtDto) override;
42 : void ImportRmtDto() override;
43 :
44 : std::vector<char> GetUniqueId() const override;
45 :
46 : void SetCqInfo(HcclAiRMACQ& cq) const;
47 :
48 : void SetWqInfo(HcclAiRMAWQ& wq) const;
49 :
50 : void SetCqContextInfo(CqContext& cq) const;
51 : void SetSqContextInfo(SqContext& sq) const;
52 :
53 : unique_ptr<BaseTask>
54 : PrepareRead(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config) override;
55 :
56 : unique_ptr<BaseTask> PrepareReadReduce(
57 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp,
58 : const SqeConfig& config) override;
59 :
60 : unique_ptr<BaseTask>
61 : PrepareWrite(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config) override;
62 :
63 : unique_ptr<BaseTask> PrepareWriteReduce(
64 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp,
65 : const SqeConfig& config) override;
66 :
67 : unique_ptr<BaseTask>
68 : PrepareInlineWrite(const MemoryBuffer& remoteMemBuf, u64 data, const SqeConfig& config) override;
69 :
70 : unique_ptr<BaseTask> PrepareWriteWithNotify(
71 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, u64 data,
72 : const MemoryBuffer& remoteNotifyMemBuf, const SqeConfig& config) override;
73 :
74 : unique_ptr<BaseTask> PrepareWriteReduceWithNotify(
75 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp,
76 : u64 data, const MemoryBuffer& remoteNotifyMemBuf, const SqeConfig& config) override;
77 :
78 : class UbCiUpdater;
79 :
80 : void AddNop(const Stream& stream) override;
81 :
82 : /**
83 : * @brief 注入共享 jetty 模式：复用外部已创建的 jetty 句柄，connection 不再自建/自销毁 jetty。
84 : * 必须在 connection 构造后、Connect/GetStatus 推进状态机前调用。
85 : * 调用后状态机跳过 JETTY_CREATING，直接进入 JETTY_CREATED。
86 : * @note 架构说明：本组方法属 base_comm 共享 jetty 特性（IS_SHARED_QUEUE）的实现细节，
87 : * 因 DevUbConnection 当前仍位于 legacy/ 而暂置于此。base_comm 侧通过
88 : * shared_jetty_connection_adapter 适配层调用，不直接依赖本类。
89 : * @param[in] jettyHdl 共享 jetty 句柄
90 : * @param[in] jettyHdlPtr 底层 jetty 指针（用于 HrtRaUbPostSend 等）
91 : * @param[in] jId jetty id
92 : * @param[in] sqVa SQ 缓冲 VA
93 : * @param[in] db doorbell 地址
94 : * @param[in] qpKey 本地 QP key
95 : * @param[in] kSize key 长度
96 : * @param[in] sDepth SQ 深度
97 : * @param[in] tpHdl 创建共享 jetty 时使用的 TP handle（注入后主 connection 复用此 tpHandle，
98 : * 避免重新向管控面申请得到不同 tpHandle 导致对端 import 路由不匹配）
99 : * @param[in] epTag Endpoint 不透明标签（透传给 releaseCb 供回调定位 Endpoint）
100 : * @param[in] releaseCb connection 销毁时调用的释放回调（由 base_comm 层注入 Endpoint::ReleaseSharedJetty）
101 : */
102 : HcclResult InjectSharedJetty(
103 : JettyHandle jettyHdl, void* jettyHdlPtr, uint32_t jId, uint64_t sqVa, uint64_t db, const uint8_t* qpKey,
104 : uint32_t kSize, uint32_t sDepth, uint64_t tpHdl, void* epTag, std::function<void(void*)> releaseCb);
105 :
106 : /**
107 : * @brief 将已自建 jetty 的 connection 标记为共享所有权移交：之后析构不再销毁 jetty，
108 : * jetty 生命周期交由 Endpoint::sharedJettyCtx_ 管理。仅当 connection 已完成 SetJettyInfo 后调用。
109 : */
110 : void TransferJettyOwnership();
111 :
112 : /**
113 : * @brief jetty 衍生字段集合，供共享模式下提取注入给其他 connection
114 : */
115 : struct JettyInfo {
116 : JettyHandle handle{0};
117 : void* handlePtr{nullptr};
118 : uint32_t jettyId{0};
119 : uint64_t sqBuffVa{0};
120 : uint64_t dbAddr{0};
121 : uint8_t localQpKey[HRT_UB_QP_KEY_MAX_LEN]{0};
122 : uint32_t keySize{0};
123 : uint32_t sqDepth{0};
124 : uint64_t tpHandle{0};
125 : RdmaHandle rdmaHandle{nullptr}; // 销毁 JFC 所需的 RDMA 句柄
126 : JfcHandle jfcHandle{0}; // 临时 connection 创建的 JFC, 由 Endpoint 统一销毁
127 : };
128 :
129 : /** 获取当前 connection 的 jetty 衍生字段（共享模式下用于 Adopt 到 Holder） */
130 : HcclResult GetJettyInfo(JettyInfo& info) const;
131 :
132 : void ReleaseTp();
133 : ~DevUbConnection() override;
134 :
135 : string Describe() const override;
136 : HcclResult Describe(std::string& dfxMsg) override;
137 :
138 : HrtUbJfcMode GetUbJfcMode() const;
139 : JettyHandle& GetJettyHandle();
140 : JettyHandle& GetRemoteJettyHandle();
141 : RdmaHandle& GetRdmaHandle();
142 : u32 GetPiVal() const;
143 : u32 GetCiVal() const;
144 : u32 GetSqDepth() const;
145 :
146 : void SetMaxReadSize(u32 value);
147 : void SetMaxWriteSize(u32 value);
148 :
149 : protected:
150 : TpProtocol tpProtocol{TpProtocol::INVALID};
151 : void GetTimeOut();
152 : u8 jettyTimeOut{8};
153 :
154 : private:
155 388 : MAKE_ENUM(UbConnStatus, INIT, TP_INFO_GETTING, JETTY_CREATING, JETTY_CREATED, JETTY_IMPORTING, READY, CONN_INVALID);
156 :
157 : UbConnStatus ubConnStatus{UbConnStatus::INIT};
158 :
159 : RdmaHandle rdmaHandle{nullptr};
160 : IpAddress locAddr{};
161 : IpAddress rmtAddr{};
162 : OpMode opMode{OpMode::OPBASE};
163 : HrtUbJfcMode jfcMode{HrtUbJfcMode::STARS_POLL};
164 : CommEngine engine_{COMM_ENGINE_RESERVED};
165 : IpAddress locIpv4Addr{};
166 : IpAddress rmtIpv4Addr{};
167 : u32 tokenValue{GetUbToken()};
168 : Eid rmtEid{};
169 : Eid locEid{};
170 : u8 qos_{static_cast<u8>(UB_QOS_DEFAULT)}; // 业务 QoS，GetTpInfo / ReleaseTpInfo 缓存键
171 :
172 : bool devUsed_{false};
173 :
174 : int32_t devLogicId{0};
175 : u32 dieId{0};
176 : u32 funcId{0};
177 : JfcHandle jfcHandle{0};
178 : u32 sqDepth{0};
179 : uint64_t sqBuffVa{0};
180 :
181 : RequestHandle reqHandle{0};
182 : vector<char_t> reqDataBuffer;
183 :
184 : u8 remoteQpKey[HRT_UB_QP_KEY_MAX_LEN] = {0};
185 : u32 keySize{0};
186 : u32 remoteTokenValue{0};
187 : JettyImportCfg jettyImportCfg{};
188 : void* remoteJettyHandlePtr{nullptr};
189 :
190 : JettyHandle jettyHandle{0};
191 : void* jettyHandlePtr{nullptr};
192 : JettyHandle remoteJettyHandle{0};
193 : u8 localQpKey[HRT_UB_QP_KEY_MAX_LEN]{0};
194 :
195 : u32 jettyId{0};
196 : u64 dbAddr{0};
197 : u32 tpn{0};
198 :
199 : u32 localTpnStart{0};
200 : u32 localTpNum{0};
201 : TpInfo tpInfo{};
202 :
203 : u32 piVal{0};
204 : u32 ciVal{0};
205 :
206 : CqCreateInfo cqInfo_{};
207 :
208 : // 最大传输size，切片使用
209 : u32 maxReadSize{0};
210 : u32 maxWriteSize{0};
211 :
212 : // 共享 jetty 注入模式标记：true 表示复用外部 jetty，不自建/自销毁
213 : bool isSharedJetty_{false};
214 : void* endpointTag_{nullptr}; // 共享模式下透传给 releaseCb_ 的 Endpoint 标签
215 : std::function<void(void*)> releaseCb_{nullptr}; // 共享 jetty 释放回调（调 Endpoint::ReleaseSharedJetty）
216 :
217 : bool CheckRequestResult();
218 : void ThrowAbnormalStatus(std::string funcName);
219 :
220 : void ProcessInit();
221 : void ProcessCreateJetty();
222 : void GenerateLocalPsn();
223 : void CreateJetty(const bool devUsed);
224 : void CreateAivUrmaJfc();
225 : void SetJettyInfo();
226 : bool GetTpInfo();
227 : void UpdateLocTpInfo();
228 : TpInfo SelectTpInfo();
229 : void ImportJetty();
230 : void SetImportInfo();
231 : void UnImportJetty();
232 : void DestroyJetty();
233 : void ReleaseResource();
234 : void ReleaseRemoteJettyIfImported(bool ctxValid);
235 : void ReleaseSharedJettyModeResources(bool ctxValid);
236 : void ReleaseOwnedJettyAndJfc(bool ctxValid);
237 :
238 : void ProcessSlices(
239 : const MemoryBuffer& loc, const MemoryBuffer& rmt,
240 : std::function<void(const MemoryBuffer&, const MemoryBuffer&, u32)> processOneSlice,
241 4 : DataType dataType = DataType::INVALID) const;
242 :
243 : void ProcessSlicesWithNotify(
244 : const MemoryBuffer& loc, const MemoryBuffer& rmt,
245 : std::function<void(const MemoryBuffer&, const MemoryBuffer&, u32)> processOneSlice,
246 : std::function<void(const MemoryBuffer&, const MemoryBuffer&)> processOneSliceWithNotify,
247 2 : DataType dataType = DataType::INVALID) const;
248 :
249 : std::unique_ptr<BaseTask> ConstructTaskUbSend(const HrtRaUbSendWrRespParam& sendWrResp, const SqeConfig& config);
250 : void UpdateCiVal(u32 ci);
251 : HcclResult CalcTotalTimeout(uint32_t& outTotalTimeoutMs);
252 : };
253 :
254 : class DevUbTpConnection : public DevUbConnection {
255 : public:
256 : DevUbTpConnection(
257 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
258 2 : const bool devUsed = false, const HrtUbJfcMode jfcMode = HrtUbJfcMode::STARS_POLL,
259 : const IpAddress& locIpv4Addr = IpAddress(), const IpAddress& rmtIpv4Addr = IpAddress(),
260 : u8 qos = static_cast<u8>(UB_QOS_DEFAULT), CommEngine engine = COMM_ENGINE_RESERVED,
261 : u32 sqDepth = UB_SQ_DEPTH_NOT_SET);
262 : };
263 :
264 : class DevUbCtpConnection : public DevUbConnection {
265 : public:
266 : DevUbCtpConnection(
267 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
268 35 : const bool devUsed = false, const HrtUbJfcMode jfcMode = HrtUbJfcMode::STARS_POLL,
269 : const IpAddress& locIpv4Addr = IpAddress(), const IpAddress& rmtIpv4Addr = IpAddress(),
270 : u8 qos = static_cast<u8>(UB_QOS_DEFAULT), CommEngine engine = COMM_ENGINE_RESERVED,
271 : u32 sqDepth = UB_SQ_DEPTH_NOT_SET);
272 : };
273 :
274 : class DevUbUboeConnection : public DevUbConnection {
275 : public:
276 : DevUbUboeConnection(
277 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
278 : const bool devUsed = false, const HrtUbJfcMode jfcMode = HrtUbJfcMode::STARS_POLL,
279 : const IpAddress& locIpv4Addr = IpAddress(), const IpAddress& rmtIpv4Addr = IpAddress(),
280 : u8 qos = static_cast<u8>(UB_QOS_DEFAULT), CommEngine engine = COMM_ENGINE_RESERVED);
281 : };
282 :
283 : class DevUbRtpConnection : public DevUbConnection {
284 : public:
285 : DevUbRtpConnection(
286 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
287 : const bool devUsed = false, const HrtUbJfcMode jfcMode = HrtUbJfcMode::STARS_POLL,
288 : const IpAddress& locAddrEid = IpAddress(), const IpAddress& rmtAddrEid = IpAddress(),
289 : u8 qos = static_cast<u8>(UB_QOS_DEFAULT), CommEngine engine = COMM_ENGINE_RESERVED,
290 : u32 sqDepth = UB_SQ_DEPTH_NOT_SET);
291 : };
292 :
293 : std::vector<DevUbConnection*> GetStarsPollUbConns(const std::vector<RmaConnection*>& rmaConns);
294 :
295 : bool IfNeedUpdatingUbCi(const std::vector<DevUbConnection*>& ubConns);
296 :
297 : } // namespace Hccl
298 :
299 : #endif // HCCLV2_DEV_UB_CONNECTION_H
|
