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