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 SHARED_JETTY_CHANNEL_POOL_H
12 : #define SHARED_JETTY_CHANNEL_POOL_H
13 :
14 : #include <cstdint>
15 : #include <cstring>
16 : #include <functional>
17 : #include <memory>
18 : #include <mutex>
19 : #include <string>
20 : #include <unordered_map>
21 : #include <vector>
22 : #include "hcomm_res_defs.h"
23 : #include "hcomm_channel.h"
24 : #include "hcomm_res.h"
25 : #include "hccl/hccl_types.h"
26 :
27 : namespace hccl {
28 :
29 : class MyRank;
30 :
31 : using EndpointDescPair = std::pair<EndpointDesc, EndpointDesc>;
32 :
33 : // 字段级 hash，规避 EndpointDesc padding 字节未初始化导致的误判。
34 : // 将参与区分的有效字段序列化到 std::string，复用标准库 std::hash<string> 完成组合，
35 : // 避免手写魔数/位运算 combine（如 0x9e3779b9）带来的可读性与稳健性问题。
36 : struct EndpointDescPairHash {
37 90 : static void AppendEndpointDesc(std::string& s, const EndpointDesc& d)
38 : {
39 90 : s.append(reinterpret_cast<const char*>(&d.protocol), sizeof(d.protocol));
40 90 : s.append(reinterpret_cast<const char*>(&d.commAddr.type), sizeof(d.commAddr.type));
41 : // commAddr.union 内 raws[36] 覆盖全部 union 存储，可安全用于 hash
42 90 : s.append(reinterpret_cast<const char*>(d.commAddr.raws), sizeof(d.commAddr.raws));
43 90 : s.append(reinterpret_cast<const char*>(&d.loc.locType), sizeof(d.loc.locType));
44 90 : s.append(reinterpret_cast<const char*>(d.loc.raws), sizeof(d.loc.raws));
45 90 : }
46 45 : std::size_t operator()(const EndpointDescPair& p) const noexcept
47 : {
48 45 : std::string buf;
49 45 : buf.reserve(sizeof(EndpointDesc) * 2);
50 45 : AppendEndpointDesc(buf, p.first);
51 45 : AppendEndpointDesc(buf, p.second);
52 45 : return std::hash<std::string>{}(buf);
53 45 : }
54 : };
55 :
56 : // 字段级比较，规避 EndpointDesc padding 字段未初始化导致的误判
57 : struct EndpointDescPairEqual {
58 29 : bool operator()(const EndpointDescPair& a, const EndpointDescPair& b) const noexcept
59 : {
60 29 : return a.first.protocol == b.first.protocol && a.first.commAddr.type == b.first.commAddr.type
61 29 : && std::memcmp(a.first.commAddr.raws, b.first.commAddr.raws, sizeof(a.first.commAddr.raws)) == 0
62 29 : && a.first.loc.locType == b.first.loc.locType
63 29 : && std::memcmp(a.first.loc.raws, b.first.loc.raws, sizeof(a.first.loc.raws)) == 0
64 29 : && a.second.protocol == b.second.protocol && a.second.commAddr.type == b.second.commAddr.type
65 29 : && std::memcmp(a.second.commAddr.raws, b.second.commAddr.raws, sizeof(a.second.commAddr.raws)) == 0
66 29 : && a.second.loc.locType == b.second.loc.locType
67 58 : && std::memcmp(a.second.loc.raws, b.second.loc.raws, sizeof(a.second.loc.raws)) == 0;
68 : }
69 : };
70 :
71 : /**
72 : * @note 职责：MyRank 级别的共享 Jetty Channel 池。
73 : * 按 myRank -> tag -> (localEndpoint, remoteEndpoint) -> [ChannelHandle] 管理复用的 Channel。
74 : * 重复调用时，若 tag->endpointPair 下 channel 数量不足则创建后再返回；若足够直接按序返回。
75 : * MyRank 析构时通过 DestroyAllByMyRank 统一清理，避免悬挂引用。
76 : * 约束：池内 Channel 生命周期由池统一管理，调用方不可单独对其调用
77 : * HcommChannelDestroy/HcclChannelRelease，否则池内会残留悬挂句柄。
78 : */
79 : class SharedJettyChannelPool {
80 : public:
81 : struct EpPairChannels {
82 : std::vector<ChannelHandle> channels;
83 : uint32_t nextReturnIdx{0};
84 : };
85 :
86 : static SharedJettyChannelPool& GetInstance();
87 :
88 : /**
89 : * @brief 获取或创建共享 Jetty Channel
90 : * @param[in] myRank 归属的 MyRank 指针（用于销毁时索引）
91 : * @param[in] tag 共享队列 tag
92 : * @param[in] epPair 源目的 endpointPair
93 : * @param[in] requestedNum 请求的 channel 数量
94 : * @param[in] createFunc 创建新 channel 的回调
95 : * @param[out] outChannels 输出的 channel 句柄数组（前 outReusedCount 个为池中复用，其余为新建）
96 : * @param[out] outReusedCount 输出从池中复用的 channel 数量（可选，nullptr 时不输出）
97 : * @return HcclResult 执行结果
98 : */
99 : HcclResult AcquireChannels(
100 : MyRank* myRank, const std::string& tag, const EndpointDescPair& epPair, uint32_t requestedNum,
101 : const std::function<HcclResult(uint32_t, ChannelHandle*)>& createFunc, ChannelHandle* outChannels,
102 : uint32_t* outReusedCount = nullptr);
103 :
104 : /**
105 : * @brief 销毁 MyRank 下所有共享 Jetty Channel（MyRank 析构时调用）
106 : * @param[in] myRank MyRank 指针
107 : */
108 : HcclResult DestroyAllByMyRank(MyRank* myRank);
109 :
110 : /**
111 : * @brief 检查 MyRank 是否有共享 Jetty Channel 未销毁
112 : * @param[in] myRank MyRank 指针
113 : * @return HcclResult HCCL_SUCCESS 表示可以销毁，HCCL_E_UNAVAIL 表示仍有 Channel
114 : */
115 : HcclResult CheckMyRankDestroy(MyRank* myRank);
116 :
117 : /**
118 : * @brief 从池中移除指定 channel 句柄（建链失败时清理已销毁的句柄）
119 : * @param[in] myRank 归属的 MyRank 指针
120 : * @param[in] tag 共享队列 tag
121 : * @param[in] epPair 源目的 endpointPair
122 : * @param[in] channels 要移除的 channel 句柄数组
123 : * @param[in] channelNum 数量
124 : */
125 : void RemoveChannels(
126 : MyRank* myRank, const std::string& tag, const EndpointDescPair& epPair, const ChannelHandle* channels,
127 : uint32_t channelNum);
128 :
129 : private:
130 5 : SharedJettyChannelPool() = default;
131 5 : ~SharedJettyChannelPool() = default;
132 : SharedJettyChannelPool(const SharedJettyChannelPool&) = delete;
133 : SharedJettyChannelPool& operator=(const SharedJettyChannelPool&) = delete;
134 :
135 : HcclResult ReturnExistingChannels(
136 : MyRank* myRank, const std::string& tag, const EndpointDescPair& epPair, uint32_t requestedNum,
137 : ChannelHandle* outChannels, uint32_t& returnFromExisting, uint32_t& needCreate);
138 :
139 : using EpPairMap = std::unordered_map<EndpointDescPair, EpPairChannels, EndpointDescPairHash, EndpointDescPairEqual>;
140 : using TagMap = std::unordered_map<std::string, EpPairMap>;
141 : using RankPoolIter = std::unordered_map<MyRank*, TagMap>::iterator;
142 :
143 : // 调用者须持有 mtx_。收集 myRank 下全部共享 Jetty channel 句柄。
144 : // 返回值：rankPools_ 中 myRank 对应的迭代器；未找到返回 rankPools_.end()。
145 : // allChannels 输出收集到的句柄（已 reserve 预分配）。
146 : RankPoolIter CollectMyRankChannelsLocked(MyRank* myRank, std::vector<ChannelHandle>& allChannels);
147 :
148 : std::mutex mtx_;
149 : std::unordered_map<MyRank*, TagMap> rankPools_;
150 : };
151 :
152 : } // namespace hccl
153 :
154 : #endif // SHARED_JETTY_CHANNEL_POOL_H
|
