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 : #ifndef HOST_CPU_ROCE_CHANNEL_H
11 : #define HOST_CPU_ROCE_CHANNEL_H
12 :
13 : #include <mutex>
14 :
15 : #include "../channel.h"
16 : #include "enum_factory.h"
17 : #include "hccl_common.h"
18 : #include "../../sockets/socket_mgr.h"
19 : #include "infiniband/verbs.h"
20 :
21 : // Orion
22 : #include "../../../../../../legacy/ascend950/unified_platform/resource/socket/socket.h"
23 : #include "../../../../../../legacy/ascend950/unified_platform/resource/buffer/local_rdma_rma_buffer.h"
24 : #include "remote_rma_buffer.h"
25 : #include "host_rdma_connection.h"
26 : #include "task_param.h"
27 :
28 : #include "exchange_data_format.h"
29 : #include "private_types.h"
30 :
31 : namespace hcomm {
32 :
33 : class HostCpuRoceChannel final : public Channel {
34 : public:
35 841 : MAKE_ENUM(RdmaStatus, INIT, SOCKET_OK, CAP_EXCHANGED, QP_CREATED, DATA_EXCHANGE, QP_MODIFIED, CONN_OK)
36 :
37 : HostCpuRoceChannel(EndpointHandle endpointHandle, HcommChannelDesc channelDesc);
38 : ~HostCpuRoceChannel();
39 :
40 : HcclResult Init() override;
41 : HcclResult GetNotifyNum(uint32_t *notifyNum) const override;
42 : HcclResult GetRemoteMems(uint32_t *memNum, CommMem **remoteMem, char ***memInfos) override;
43 : ChannelStatus GetStatus() override;
44 : HcclResult GetStatus(ChannelStatus &status);
45 : HcclResult ProcessStatus();
46 0 : const HcommChannelDesc& GetChannelDesc() const override { return channelDesc_; }
47 :
48 : std::string Describe() const;
49 :
50 : HcclResult SetDfxCallback(std::function<HcclResult(const Hccl::TaskParam&, u64)> callback);
51 :
52 : // 数据面调用verbs接口
53 : HcclResult NotifyRecord(const uint32_t remoteNotifyIdx) override;
54 : HcclResult NotifyWait(const uint32_t localNotifyIdx, const uint32_t timeout) override;
55 : HcclResult WriteWithNotify(void *dst, const void *src, const uint64_t len, uint32_t remoteNotifyIdx) override;
56 : HcclResult Write(void *dst, const void *src, uint64_t len) override;
57 : HcclResult Read(void *dst, const void *src, uint64_t len) override;
58 : HcclResult ChannelFence() override;
59 : HcclResult GetHcclBuffer(void*& addr, uint64_t& size);
60 :
61 : private:
62 : HcclResult WaitForFenceCompletion();
63 :
64 : virtual HcclResult Clean() override;
65 : HcclResult Resume() override;
66 : HcclResult ExchangeCapability();
67 : HcclResult ExchangeDataHybird();
68 : HcclResult GetRemoteAddrHybird(hccl::MemType memType, u8 *&data, u64 &size);
69 : HcclResult ParseRecvExchangeDataHybird();
70 : HcclResult BuildExchangeDataHybird();
71 : HcclResult BuildExchangeDataLengthHybird();
72 :
73 : HcclResult RegisterUserMemHybird();
74 : HcclResult BuildNotifyWrHybird(const uint32_t remoteNotifyIdx, struct ibv_send_wr ¬ifRecordWr);
75 : HcclResult WriteWithNotifyHybrid(void *dst, const void *src, uint64_t len, uint32_t remoteNotifyIdx);
76 : HcclResult NotifyWaitHybrid(uint32_t localNotifyIdx, uint32_t timeout);
77 :
78 : HcclResult CreateNotifyHybird(hccl::MemType notifyType, uint32_t notifyId);
79 : HcclResult CreateNotifyValueBufferHybird();
80 : HcclResult CreateNotifyBufferHybird(hccl::MemType notifyType, uint32_t notifyId, u8 *&data, u64 &size);
81 : hccl::MemType NotifyIdToMemtypeHybird(uint32_t remoteNotifyIdx);
82 : HcclResult ConnectSingleQpHybrid(std::function<bool()> needStop);
83 :
84 : private:
85 : HcclResult ParseInputParam();
86 : HcclResult StartListen();
87 : HcclResult BuildSocket();
88 : HcclResult BuildConnection();
89 : HcclResult BuildNotify();
90 : HcclResult BuildBuffer();
91 :
92 : HcclResult CheckSocketStatus();
93 : HcclResult CreateQp();
94 : HcclResult ExchangeData();
95 : HcclResult ModifyQp();
96 :
97 : void NotifyVecPack(Hccl::BinaryStream &binaryStream);
98 : HcclResult BufferVecPack(Hccl::BinaryStream &binaryStream);
99 : HcclResult ConnVecPack(Hccl::BinaryStream &binaryStream);
100 : // void HandshakeMsgPack(Hccl::BinaryStream &binaryStream);
101 :
102 : // HcclResult HandshakeMsgUnpack(Hccl::BinaryStream &binaryStream);
103 : HcclResult NotifyVecUnpack(Hccl::BinaryStream &binaryStream);
104 : HcclResult RmtBufferVecUnpackProc(Hccl::BinaryStream &binaryStream);
105 : HcclResult ConnVecUnpackProc(Hccl::BinaryStream &binaryStream);
106 :
107 : std::vector<Hccl::QpInfo> GetQpInfos() const; // in Connection
108 :
109 : HcclResult IbvPostRecv() const;
110 : HcclResult PrepareNotifyWrResource(uint32_t qpIdx, const uint64_t len, const uint32_t remoteNotifyIdx, struct ibv_send_wr ¬ifyRecordWr,
111 : Hccl::TaskParam &taskParam) const;
112 : HcclResult PrepareWriteWrResource(const void *dst, const void *src, const uint64_t len, const uint32_t remoteNotifyIdx,
113 : struct ibv_send_wr &writeWithNotifyWr, Hccl::TaskParam &taskParam) const;
114 :
115 : HcclResult PostRdmaOp(const char *caller, ibv_wr_opcode opcode, void *localAddr, const void *remoteAddr, const uint64_t len);
116 : void BuildRdmaWr(const char *caller, ibv_wr_opcode opcode, void *localAddr, const void *remoteAddr, uint64_t len,
117 : size_t localIdx, size_t rmtIdx, struct ibv_send_wr &wr, struct ibv_sge &sg) const;
118 : HcclResult PostAndCheckSend(struct ibv_qp *qp, const uint32_t qpIdx, const char *caller, struct ibv_send_wr &wr);
119 : HcclResult FindLocalBuffer(const uint64_t addr, const uint64_t len, size_t &targetIdx) const;
120 : HcclResult FindRemoteBuffer(const uint64_t addr, const uint64_t len, size_t &targetIdx) const;
121 : HcclResult ReportWcStatusError(enum ibv_wc_status status);
122 :
123 : // Wrapper for stub
124 0 : int IbvPollCq(ibv_cq *sendCq, uint32_t numEntries, ibv_wc *wc) const
125 : {
126 0 : return ibv_poll_cq(sendCq, numEntries, wc);
127 : }
128 :
129 : // 入参
130 : EndpointHandle endpointHandle_;
131 : HcommChannelDesc channelDesc_;
132 :
133 : // 转换参数
134 : EndpointDesc localEp_;
135 : EndpointDesc remoteEp_;
136 : uint32_t notifyNum_{0};
137 : Hccl::Socket *socket_{nullptr};
138 : const Hccl::SocketConfig* socketConfig_{nullptr};
139 : RdmaHandle rdmaHandle_{nullptr};
140 :
141 : std::vector<std::unique_ptr<HostRdmaConnection>> connections_{};
142 : std::vector<Hccl::LocalRdmaRmaBuffer *> localRmaBuffers_{};
143 : std::vector<uint32_t> localDpuNotifyIds_{};
144 : uint32_t bufferNum_{0};
145 : uint32_t connNum_{0};
146 : // Hccl::BaseMemTransport::Attribution attr_;
147 : ChannelStatus channelStatus_{ChannelStatus::INIT};
148 : RdmaStatus rdmaStatus_{RdmaStatus::INIT};
149 : std::vector<uint32_t> remoteDpuNotifyIds_;
150 : std::vector<std::unique_ptr<Hccl::RemoteRdmaRmaBuffer>> rmtRmaBuffers_{};
151 : std::vector<ExchangeRdmaConnDto> rmtConnDtos_;
152 : std::vector<int> wqeNums_;
153 : bool fenceFlag_{false};
154 : std::mutex remoteMemsMutex_; // 远端内存列表互斥锁
155 :
156 : // GetRemoteMems 缓存
157 : bool cacheValid_{false};
158 : std::vector<CommMem> userRemoteMems_;
159 : std::vector<std::string> memInfoCopies_;
160 : std::vector<char*> memInfoPointers_;
161 :
162 : uint64_t maxMsgSize_{0};
163 : uint32_t lbMax_{0}; // 多QP负载均衡
164 :
165 : std::function<HcclResult(const Hccl::TaskParam&, u64)> dfxCallback_;
166 :
167 : std::mutex cq_mutex;
168 : std::mutex sendCq_mutex;
169 :
170 : // ========== 混合模式（RoCE Cross-Mode）成员变量 ==========
171 : // 1. 能力协商结果
172 : RoCECapability remoteCap_; // 对端能力
173 : bool isHybridMode_ = false; // 是否为混合模式
174 :
175 : uint32_t localNotifySize_;
176 : uint32_t localNotifyAccess_;
177 :
178 : std::array<hccl::MemMsg, static_cast<u32>(hccl::MemType::MEM_TYPE_RESERVED)> localMemMsg_;
179 : std::array<hccl::MemMsg, static_cast<u32>(hccl::MemType::MEM_TYPE_RESERVED)> remoteMemMsg_;
180 : uint64_t exchangeDataTotalSize_;
181 : std::vector<uint8_t> exchangeDataForSend_;
182 : std::vector<uint8_t> exchangeDataForRecv_;
183 :
184 : uint32_t devicePhyId_{};
185 : };
186 :
187 : } // namespace hcomm
188 :
189 : #endif // HOST_CPU_ROCE_CHANNEL_H
|
