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 JETTY_CONTEXT_H
12 : #define JETTY_CONTEXT_H
13 :
14 : #include <cstdint>
15 : #include <functional>
16 : #include <mutex>
17 : #include <condition_variable>
18 : #include <vector>
19 : #include "hccl/hccl_types.h"
20 : #include "hcomm_res_defs.h"
21 : #include "rdma_handle_manager.h"
22 :
23 : namespace hcomm {
24 :
25 : /**
26 : * @note 职责：共享 Jetty 数据面资源的内聚集合（jetty 句柄 / QP key / PI·CI device 内存 / JFC / 远端 jetty 缓存）。
27 : * Endpoint（控制面）持 unique_ptr<JettyContext> 延迟创建，控制面职责不因共享 jetty 特性膨胀。
28 : * 引用计数 + condition_variable 管理同 endpoint 下多 channel 复用，替代 sleep 轮询。
29 : */
30 : class JettyContext {
31 : public:
32 : struct SharedRemoteJettyCtx {
33 : std::vector<uint8_t> remoteQpKey{};
34 : uint64_t handle{0};
35 : void* handlePtr{nullptr};
36 : uint32_t tpn{0};
37 : bool ready{false};
38 : };
39 :
40 : /**
41 : * @brief jetty 资源字段集合。channel 通过 Acquire 取得 Ctx 视图，
42 : * 读取 PI/CI 指针绑给 transport，不持有资源 ownership。
43 : * @note tpHandle 不在此共享：一对多场景下各主 connection 到不同对端需各自申请 tpHandle，
44 : * 否则对端 import 时 peerTpHandle 路由不匹配。仅 jetty/SQ/JFC/CQ/psn 共享。
45 : * @note 同步约束：本结构与 Inner 的共享字段一一对应，新增/修改字段须同步四处：
46 : * Ctx、Inner、InnerToCtx（jetty_context.cc）、Acquire 第三段逐字段拷贝，漏改会静默丢字段。
47 : */
48 : struct Ctx {
49 : Hccl::JettyHandle handle{0};
50 : void* handlePtr{nullptr};
51 : uint32_t jettyId{0};
52 : uint64_t sqBuffVa{0};
53 : uint64_t dbAddr{0};
54 : uint8_t localQpKey[Hccl::HRT_UB_QP_KEY_MAX_LEN]{0};
55 : uint32_t keySize{0};
56 : uint32_t sqDepth{0};
57 : void* sqPiPtr{nullptr};
58 : void* sqCiPtr{nullptr};
59 : void* cqPiPtr{nullptr};
60 : void* cqCiPtr{nullptr};
61 : uint64_t queueIndexMemSize{0};
62 : void* rdmaHandle{nullptr};
63 : uint64_t jfcHandle{0};
64 : Hccl::CqCreateInfo cqInfo{};
65 : uint32_t localPsn{0}; // 共享 jetty 统一 psn：临时 connection 生成后存入，主 connection 复用，避免多 connection
66 : // 各自 GenerateLocalPsn 导致 import 同一 TP 对时 psn 互相覆盖
67 : };
68 :
69 26 : JettyContext() = default;
70 : ~JettyContext();
71 :
72 : JettyContext(const JettyContext&) = delete;
73 : JettyContext& operator=(const JettyContext&) = delete;
74 :
75 : /**
76 : * @brief 获取或创建共享 jetty。命中复用 refCount++；未命中调 provideCtx 创建并缓存。
77 : * 用 condition_variable 等待并发创建者，替代 sleep 轮询。
78 : * @param[in] provideCtx 创建回调（首次时调用，回调内创建 jetty 并填入 ctx）
79 : * @param[out] outCtx 输出的 jetty 上下文视图
80 : */
81 : HcclResult Acquire(const std::function<HcclResult(Ctx&)>& provideCtx, Ctx& outCtx);
82 :
83 : /** 释放共享 jetty 引用，refCount-- 归 0 时销毁 jetty 并清空 ctx */
84 : HcclResult Release();
85 :
86 : /**
87 : * 查询或预约远端共享 jetty。三种返回场景：
88 : * - 已发布(ready=true)：needImport=false, handle!=0，调用方直接复用 handle
89 : * - 已预约未发布(ready=false)：needImport=false, handle=0，调用方应进入等待，不使用 handle
90 : * - 首次预约：needImport=true, handle=0，调用方应执行 import 并随后调 PublishSharedRemoteJetty
91 : */
92 : HcclResult AcquireSharedRemoteJetty(
93 : const uint8_t* remoteQpKey, uint32_t keySize, bool& needImport, uint64_t& handle, void*& handlePtr,
94 : uint32_t& tpn);
95 :
96 : HcclResult PublishSharedRemoteJetty(
97 : const uint8_t* remoteQpKey, uint32_t keySize, uint64_t handle, void* handlePtr, uint32_t tpn);
98 :
99 : private:
100 : // @note 同步约束：共享字段与 Ctx 一一对应（见 Ctx 的同步注释），多出的 refCount/valid/creating/remoteJettys
101 : // 为内部状态
102 : struct Inner {
103 : Hccl::JettyHandle handle{0};
104 : void* handlePtr{nullptr};
105 : uint32_t jettyId{0};
106 : uint64_t sqBuffVa{0};
107 : uint64_t dbAddr{0};
108 : uint8_t localQpKey[Hccl::HRT_UB_QP_KEY_MAX_LEN]{0};
109 : uint32_t keySize{0};
110 : uint32_t sqDepth{0};
111 : uint32_t refCount{0};
112 : bool valid{false};
113 : bool creating{false};
114 : void* sqPiPtr{nullptr};
115 : void* sqCiPtr{nullptr};
116 : void* cqPiPtr{nullptr};
117 : void* cqCiPtr{nullptr};
118 : uint64_t queueIndexMemSize{0};
119 : void* rdmaHandle{nullptr};
120 : uint64_t jfcHandle{0};
121 : Hccl::CqCreateInfo cqInfo{};
122 : uint32_t localPsn{0}; // 共享 jetty 统一 psn（与 Ctx::localPsn 对应）
123 : std::vector<SharedRemoteJettyCtx> remoteJettys{};
124 : };
125 :
126 : void UnimportSharedRemoteJettys(Inner& inner);
127 : void DestroyJettyResources(Inner& inner);
128 : Ctx InnerToCtx(const Inner& inner);
129 :
130 : mutable std::mutex mtx_;
131 : std::condition_variable cv_;
132 : Inner inner_{};
133 : };
134 :
135 : } // namespace hcomm
136 :
137 : #endif // JETTY_CONTEXT_H
|
