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