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 BASE_MEM_TRANSPORT_H
11 : #define BASE_MEM_TRANSPORT_H
12 :
13 : #include <memory>
14 : #include <unordered_map>
15 :
16 : #include "task.h"
17 : #include "local_rma_buffer.h"
18 : #include "remote_rma_buffer.h"
19 : #include "../../resource/connection/rma_connection.h"
20 : #include "local_notify.h"
21 : #include "ipc_remote_notify.h"
22 : #include "local_cnt_notify.h"
23 : #include "op_mode.h"
24 : #include "mem_transport_common.h"
25 : #include "task_param.h"
26 : #include "transport_status.h"
27 : #include "socket.h"
28 : #include "virtual_topo.h"
29 :
30 : namespace Hccl {
31 :
32 : struct RmaBufferSlice {
33 : u64 addr;
34 : u64 size;
35 : LocalRmaBuffer *buf;
36 : std::string Describe() const
37 : {
38 : if (buf == nullptr) {
39 : return StringFormat("RmaBufferSlice[addr=0x%llx, size=0x%llx, buf is null]", addr, size);
40 : } else {
41 : return StringFormat("RmaBufferSlice[addr=0x%llx, size=0x%llx, buf=%s]", addr, size,
42 : buf->Describe().c_str());
43 : }
44 : }
45 : };
46 :
47 : struct RmtRmaBufferSlice {
48 : u64 addr;
49 : u64 size;
50 : RemoteRmaBuffer *buf;
51 : std::string Describe() const
52 : {
53 : if (buf == nullptr) {
54 : return StringFormat("RmtRmaBufferSlice=[addr=0x%llx, size=0x%llx, buf is null]", addr, size);
55 : } else {
56 : return StringFormat("RmtRmaBufferSlice=[addr=0x%llx, size=0x%llx, buf=%s]", addr, size,
57 : buf->Describe().c_str());
58 : }
59 : }
60 : };
61 :
62 : class BaseMemTransport {
63 : public:
64 : struct CommonLocRes {
65 : std::vector<BaseLocalNotify *> notifyVec;
66 : std::vector<LocalRmaBuffer *> bufferVec;
67 : std::vector<RmaConnection *> connVec;
68 16 : string Describe() const
69 : {
70 : string msg = StringFormat("MemTransportCommonLocRes=[notifyNum=%zu, bufferNum=%zu, connNum=%zu]",
71 16 : notifyVec.size(), bufferVec.size(), connVec.size());
72 16 : return msg;
73 : }
74 : };
75 :
76 : struct LocCntNotifyRes {
77 : vector<LocalCntNotify *> vec{};
78 : vector<char> desc{}; // 将 topicId + index 映射到 index的关系交换对端
79 :
80 128 : std::string Describe() const
81 : {
82 : string msg = StringFormat("LocCntNotifyRes[cntNotifyNum=%zu], desc=%s",
83 128 : vec.size(), Bytes2hex(desc.data(), desc.size()).c_str());
84 128 : return msg;
85 : }
86 : };
87 :
88 : struct Attribution {
89 : OpMode opMode;
90 : u32 devicePhyId{0};
91 : vector<char> handshakeMsg{};
92 : AcceleratorState opAcceState{AcceleratorState::AICPU_TS};
93 : string Describe() const
94 :
95 : {
96 : return StringFormat("MemTransportAttribution[opMode=%s, devicePhyId=%u, handleshakeMsg=%s]",
97 : opMode.Describe().c_str(), devicePhyId,
98 : Bytes2hex(handshakeMsg.data(), handshakeMsg.size()).c_str());
99 : }
100 : };
101 : BaseMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData, const Socket &socket,
102 : TransportType type);
103 :
104 : BaseMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData, const Socket &socket,
105 : TransportType type, std::function<void(u32 streamId, u32 taskId, TaskParam taskParam)> callback);
106 :
107 143 : virtual ~BaseMemTransport() = default;
108 :
109 0 : virtual vector<char> &GetRmtHandshakeMsg() // 返回握手消息
110 : {
111 0 : return rmtHandshakeMsg;
112 : }
113 :
114 : AcceleratorState &GetRmtOpAcceState()
115 : {
116 : return rmtOpAcceState;
117 : }
118 :
119 : virtual std::string Describe() const = 0;
120 :
121 : virtual void Establish();
122 :
123 0 : virtual TransportStatus GetStatus()
124 : {
125 0 : return TransportStatus::READY;
126 : }
127 :
128 0 : virtual std::vector<char> GetUniqueId()
129 : {
130 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
131 : }
132 :
133 0 : virtual RemoteRmaBuffer *GetRmtRmaBuffer(u32 index)
134 : {
135 0 : if (index >= rmtRmaBufferVec.size()) {
136 0 : MACRO_THROW(InvalidParamsException,
137 : StringFormat("Get remote rmaBuffer fail, index[%u] is not in range.", index));
138 : }
139 0 : return rmtRmaBufferVec[index];
140 : }
141 :
142 0 : virtual void SetConnVec(std::vector<RmaConnection *> &connVec)
143 : {
144 : (void)connVec;
145 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
146 : }
147 :
148 0 : virtual vector<char> &GetRmtCntNotifyDesc()
149 : {
150 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
151 : }
152 :
153 0 : virtual void Post(u32 index, const Stream &stream)
154 : {
155 : (void)index;
156 : (void)stream;
157 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
158 : }
159 :
160 0 : virtual void Wait(u32 index, const Stream &stream, u32 timeout)
161 : {
162 : (void)index;
163 : (void)stream;
164 : (void)timeout;
165 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
166 : }
167 :
168 0 : virtual void Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
169 : {
170 : (void)locSlice;
171 : (void)rmtSlice;
172 : (void)stream;
173 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
174 : }
175 :
176 0 : virtual void ReadReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const ReduceIn &reduceIn,
177 : const Stream &stream)
178 : {
179 : (void)locSlice;
180 : (void)rmtSlice;
181 : (void)reduceIn;
182 : (void)stream;
183 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
184 : }
185 :
186 0 : virtual void Write(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
187 : {
188 : (void)locSlice;
189 : (void)rmtSlice;
190 : (void)stream;
191 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
192 : }
193 :
194 0 : virtual void WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
195 : const ReduceIn &reduceIn, const Stream &stream)
196 : {
197 : (void)locSlice;
198 : (void)rmtSlice;
199 : (void)reduceIn;
200 : (void)stream;
201 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
202 : }
203 :
204 0 : virtual void WriteWithNotify(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
205 : const WithNotifyIn &withNotify, const Stream &stream)
206 : {
207 : (void)locSlice;
208 : (void)rmtSlice;
209 : (void)withNotify;
210 : (void)stream;
211 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
212 : }
213 :
214 0 : virtual void WriteReduceWithNotify(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
215 : const ReduceIn &reduceIn, const WithNotifyIn &withNotify, const Stream &stream)
216 : {
217 : (void)locSlice;
218 : (void)rmtSlice;
219 : (void)reduceIn;
220 : (void)withNotify;
221 : (void)stream;
222 0 : MACRO_THROW(NotSupportException, StringFormat("not supported."));
223 : }
224 :
225 0 : virtual vector<char> &GetLocalHandshakeMsg() // 返回本端握手消息
226 : {
227 0 : return attr.handshakeMsg;
228 : }
229 :
230 : AcceleratorState &GetLocalOpAcceState()
231 : {
232 : return attr.opAcceState;
233 : }
234 :
235 : void SetLocalOpAcceState(const AcceleratorState &opAcceState)
236 : {
237 : attr.opAcceState = opAcceState;
238 : }
239 :
240 0 : void SetIsHost()
241 : {
242 0 : isHost_ = true;
243 0 : }
244 :
245 : string GetLinkDescInfo();
246 : string DescribeSocket() const;
247 : protected:
248 : CommonLocRes commonLocRes{};
249 : Attribution attr;
250 : LinkData linkData;
251 : Socket *socket{};
252 : TransportType transportType;
253 : std::function<void(u32 streamId, u32 taskId, const TaskParam &taskParam)> callback;
254 :
255 : std::vector<RemoteRmaBuffer *> rmtRmaBufferVec;
256 :
257 : TransportStatus baseStatus{TransportStatus::INIT};
258 :
259 : vector<char> rmtHandshakeMsg{0}; // 远端握手消息
260 : AcceleratorState rmtOpAcceState{AcceleratorState::AICPU_TS};
261 :
262 : u32 notifyNum{0};
263 : u32 bufferNum{0};
264 : u32 connNum{0};
265 : u32 exchangeDataSize{0}; // 交换的消息大小
266 : bool isHost_{false};
267 :
268 : void SetBaseStatusReady();
269 :
270 : bool IsSocketReady();
271 :
272 : void NotifyVecPack(BinaryStream &binaryStream);
273 :
274 : void ConnVecPack(BinaryStream &binaryStream);
275 :
276 : void HandshakeMsgPack(BinaryStream &binaryStream);
277 :
278 : HcclResult HandshakeMsgUnpack(BinaryStream &binaryStream);
279 :
280 : private:
281 : HcclResult CheckLocNotify(CommonLocRes &res);
282 :
283 : void CheckLocBuffer(CommonLocRes &res);
284 :
285 : HcclResult CheckLocConn(CommonLocRes &res);
286 :
287 : HcclResult CheckCommonLocRes(CommonLocRes &res);
288 : };
289 :
290 : } // namespace Hccl
291 : #endif
|
