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