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 HCOMM_TRANSPORT_BASE_PUB_H
12 : #define HCOMM_TRANSPORT_BASE_PUB_H
13 :
14 : #include <hccl/base.h>
15 : #include <hccl/hccl_types.h>
16 : #include "stream_pub.h"
17 : #include "sal.h"
18 : #include "dispatcher_pub.h"
19 : #include "mem_name_repository_pub.h"
20 : #include "task_logic_info_pub.h"
21 : #include "transport_pub.h"
22 :
23 : #include "hccl_socket.h"
24 : #include "notify_pool.h"
25 : #include "local_ipc_notify.h"
26 : #include "remote_notify.h"
27 : #include "hccl_mem_defs.h"
28 :
29 : namespace hccl {
30 :
31 : const std::map<LinkType, std::string> LINK_TYPE_STR_MAP{
32 : {LinkType::LINK_ONCHIP, "ONCHIP"},
33 : {LinkType::LINK_HCCS, "HCCS"},
34 : {LinkType::LINK_PCIE, "PCIE"},
35 : {LinkType::LINK_ROCE, "ROCE"},
36 : {LinkType::LINK_SIO, "SIO"},
37 : {LinkType::LINK_HCCS_SW, "HCCS_SW"},
38 : {LinkType::LINK_STANDARD_ROCE, "STANDARD_ROCE"},
39 : {LinkType::LINK_RESERVED, "RESERVED"}
40 : };
41 :
42 9 : inline std::string GetLinkTypeEnumStr(LinkType linkType)
43 : {
44 9 : auto iter = LINK_TYPE_STR_MAP.find(linkType);
45 9 : if (iter == LINK_TYPE_STR_MAP.end()) {
46 0 : return "Invalid LinkType";
47 : } else {
48 9 : return iter->second;
49 : }
50 : }
51 :
52 : class TransportBase {
53 : public:
54 : explicit TransportBase(DispatcherPub *dispatcher,
55 : const std::unique_ptr<NotifyPool> ¬ifyPool,
56 : MachinePara &machinePara, std::chrono::milliseconds timeout);
57 : virtual ~TransportBase();
58 :
59 : virtual HcclResult Init();
60 : virtual HcclResult DeInit();
61 :
62 : virtual HcclResult TxDataSignal(Stream &stream);
63 : virtual HcclResult RxDataSignal(Stream &stream);
64 :
65 : virtual HcclResult Stop();
66 : virtual HcclResult Resume();
67 : virtual HcclResult TxAsync(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len, Stream &stream);
68 : virtual HcclResult TxAsync(std::vector<TxMemoryInfo>& txMems, Stream &stream);
69 :
70 : virtual HcclResult TxWithReduce(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len,
71 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream);
72 : virtual HcclResult TxWithReduce(const std::vector<TxMemoryInfo> &txWithReduceMems, const HcclDataType datatype,
73 : HcclReduceOp redOp, Stream &stream);
74 :
75 : virtual HcclResult RxWithReduce(UserMemType recvSrcMemType, u64 recvSrcOffset, void *recvDst, u64 recvLen,
76 : void *reduceSrc, void *reduceDst, u64 reduceDataCount, HcclDataType reduceDatatype,
77 : HcclReduceOp reduceOp, Stream &stream, const u64 reduceAttr);
78 : virtual HcclResult RxWithReduce(const std::vector<RxWithReduceMemoryInfo> &rxWithReduceMems,
79 : HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream &stream, const u64 reduceAttr);
80 :
81 : virtual bool IsSupportTransportWithReduce();
82 :
83 : virtual HcclResult RxAsync(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream);
84 : virtual HcclResult RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream &stream);
85 :
86 : virtual HcclResult DataReceivedAck(Stream &stream);
87 :
88 : virtual HcclResult TxAck(Stream &stream);
89 : virtual HcclResult RxAck(Stream &stream);
90 :
91 : virtual HcclResult TxData(UserMemType dstMemType, u64 dstOffset, const void *src, u64 len, Stream &stream);
92 :
93 : virtual HcclResult RxData(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream);
94 :
95 : virtual HcclResult TxPrepare(Stream &stream);
96 : virtual HcclResult RxPrepare(Stream &stream);
97 :
98 : virtual HcclResult TxDone(Stream &stream);
99 : virtual HcclResult RxDone(Stream &stream);
100 :
101 : // 保证send语义完成
102 : virtual HcclResult TxWaitDone(Stream &stream);
103 : // 保证recv语义完成
104 : virtual HcclResult RxWaitDone(Stream &stream);
105 : // TxWaitDone、RxWaitDone共同出现保证sendrecv语义完成
106 :
107 : virtual HcclResult Post(u32 notifyIdx, Stream &stream);
108 : virtual HcclResult Wait(u32 notifyIdx, Stream &stream, const u32 timeOut = NOTIFY_INVALID_WAIT_TIME);
109 :
110 : virtual HcclResult GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t *memNum, HcclMemType memType);
111 : virtual HcclResult GetIndOpRemoteMem(HcclMem **remoteMem, uint32_t *memNum);
112 : virtual HcclResult GetRemoteMem(UserMemType memType, void **remotePtr);
113 : virtual HcclResult GetRemoteMem(std::vector<void *> *remotePtrVec);
114 : virtual HcclResult GetRemoteMemKey(UserMemType memType, uint32_t *remoteMemKey);
115 : virtual HcclResult GetRemoteMemSize(UserMemType memType, u64 &size);
116 : virtual HcclResult GetLocalRdmaNotify(std::vector<HcclSignalInfo> &rdmaNotify);
117 : virtual HcclResult GetDrainLocalDataNotify(void* &localAddr, uint32_t& lkey, HcclSignalInfo &dataNotify);
118 : virtual HcclResult GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey> &rdmaNotifyAddr);
119 : virtual HcclResult GetLocalNotifyValueAddrKey(std::vector<AddrKey> ¬ifyValue);
120 : virtual HcclResult GetLocalMemDetails(UserMemType memType, MemDetails &memDetails);
121 : virtual HcclResult GetLocalNotify(std::vector<HcclSignalInfo> &localNotify);
122 : virtual HcclResult GetRemoteNotify(std::vector<HcclSignalInfo> &localNotify);
123 :
124 : virtual HcclResult GetAiQpInfo(std::vector<HcclQpInfoV2> &aiQpInfo);
125 : virtual HcclResult GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo> &aiRMAQueueInfo);
126 : virtual HcclResult GetTransportId(u32 &id);
127 : HcclResult GetChipId(s64 &chipId);
128 : HcclResult GetTxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo);
129 : HcclResult GetRxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo);
130 : HcclResult GetTxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo);
131 : HcclResult GetRxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo);
132 9 : inline hccl::LinkType GetLinkType() const
133 : {
134 9 : return transportAttr_.linkType;
135 : }
136 :
137 9 : inline bool GetSupportDataReceivedAck() const
138 : {
139 9 : return machinePara_.supportDataReceivedAck;
140 : }
141 :
142 0 : inline void SetSupportDataReceivedAck(bool supportDataReceivedAck)
143 : {
144 0 : machinePara_.supportDataReceivedAck = supportDataReceivedAck;
145 0 : }
146 :
147 27 : inline bool IsSpInlineReduce() const
148 : {
149 81 : bool isSpInlineReduce = transportAttr_.linkType == LinkType::LINK_HCCS ||
150 27 : transportAttr_.linkType == LinkType::LINK_PCIE ||
151 81 : transportAttr_.linkType == LinkType::LINK_SIO ||
152 27 : transportAttr_.linkType == LinkType::LINK_HCCS_SW;
153 27 : return isSpInlineReduce;
154 : }
155 :
156 0 : inline u32 GetRemoteRank() const
157 : {
158 0 : return machinePara_.remoteWorldRank;
159 : }
160 0 : virtual HcclResult ConnectAsync(u32& status)
161 : {
162 0 : return HCCL_SUCCESS;
163 : };
164 0 : virtual HcclResult ConnectQuerry(u32& status)
165 : {
166 0 : return HCCL_SUCCESS;
167 : };
168 :
169 0 : virtual void Break()
170 : {
171 0 : return;
172 : }
173 :
174 0 : inline void EnableUseOneDoorbell()
175 : {
176 0 : useOneDoorbell_ = true;
177 0 : }
178 :
179 0 : inline bool GetUseOneDoorbellValue()
180 : {
181 0 : return useOneDoorbell_;
182 : }
183 :
184 0 : inline u32 GetNotifyNum()
185 : {
186 0 : return notifyNum_;
187 : }
188 : HcclResult OpenRemoteNotify(const std::vector<u8>& byteVector, std::shared_ptr<RemoteNotify> &remoteNotify);
189 :
190 : virtual HcclResult TxEnv(const void *ptr, const u64 len, Stream &stream);
191 : virtual HcclResult RxEnv(Stream &stream);
192 :
193 : virtual HcclResult WriteAsync(
194 : struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream);
195 : virtual HcclResult WriteSync(
196 : struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf, Stream &stream);
197 :
198 : virtual HcclResult WriteReduceAsync(struct Transport::Buffer &remoteBuf, struct Transport::Buffer &localBuf,
199 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream);
200 :
201 : virtual HcclResult ReadAsync(
202 : struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream);
203 : virtual HcclResult ReadSync(
204 : struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf, Stream &stream);
205 : virtual HcclResult ReadReduceSync(struct Transport::Buffer &localBuf, struct Transport::Buffer &remoteBuf,
206 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream);
207 :
208 : virtual HcclResult BatchTransferAsync(
209 : const HcommBatchTransferDesc *transferDescs, uint32_t descNum, Stream &stream);
210 :
211 : virtual HcclResult PostReady(Stream &stream);
212 : virtual HcclResult WaitReady(Stream &stream);
213 :
214 : virtual HcclResult PostFin(Stream &stream);
215 : virtual HcclResult WaitFin(Stream &stream);
216 :
217 : virtual HcclResult PostFinAck(Stream &stream);
218 : virtual HcclResult WaitFinAck(Stream &stream);
219 : virtual HcclResult Drain(Stream &stream);
220 :
221 : virtual HcclResult GetDrainRemSrcMem(void* &remoteAddr, uint32_t &remoteKey, uint32_t &size);
222 : TransportAttr GetTransportAttr();
223 :
224 : HcclResult SetStopFlag(bool value);
225 : bool GetStopFlag();
226 : virtual HcclResult Fence();
227 : virtual HcclResult UpdateRemoteAddr(void *remoteIn, void *remoteOut);
228 :
229 0 : std::vector<u8> &GetExchangeInfo()
230 : {
231 0 : return exchangeMsg_;
232 : }
233 :
234 0 : virtual bool GetIsUseAtomicWrite() { return useAtomicWrite_; }
235 :
236 0 : inline HcclResult GetSpecificNotify(HcclSignalInfo& notifyInfo, bool& isValid, const std::string& notifyName) {
237 : // 针对alltoallv算子aicpu cache, 提供Tx/RxAck和Tx/RxDataSignal的相关notify信息
238 0 : if (notifyName == "localSendReady") { // For RxDataSignal
239 0 : if (!localSendReadyNotify_) {
240 0 : isValid = false;
241 : } else {
242 0 : CHK_RET(localSendReadyNotify_->GetNotifyData(notifyInfo));
243 0 : isValid = true;
244 : }
245 0 : } else if (notifyName == "localSendDone") { // For RxAck
246 0 : if (!localSendDoneNotify_) {
247 0 : isValid = false;
248 : } else {
249 0 : CHK_RET(localSendDoneNotify_->GetNotifyData(notifyInfo));
250 0 : isValid = true;
251 : }
252 0 : } else if (notifyName == "remoteSendReady") { // For TxDataSignal
253 0 : if (!remoteSendReadyNotify_) {
254 0 : isValid = false;
255 : } else {
256 0 : CHK_RET(remoteSendReadyNotify_->GetNotifyData(notifyInfo));
257 0 : isValid = true;
258 : }
259 0 : } else if (notifyName == "remoteSendDone") { // For TxAck
260 0 : if (!remoteSendDoneNotify_) {
261 0 : isValid = false;
262 : } else {
263 0 : CHK_RET(remoteSendDoneNotify_->GetNotifyData(notifyInfo));
264 0 : isValid = true;
265 : }
266 : } else {
267 0 : HCCL_ERROR("[TransportBase][GetSpecificNotify] unsupported notifyName[%s]", notifyName.c_str());
268 0 : return HCCL_E_NOT_SUPPORT;
269 : }
270 :
271 0 : return HCCL_SUCCESS;
272 : }
273 :
274 : virtual HcclResult InitDrainNotifyInfo();
275 : protected:
276 : virtual HcclResult FillExchangeDataTotalSize();
277 : virtual HcclResult ConstructExchangeForSend();
278 : virtual HcclResult ParseReceivedExchangeData();
279 : HcclResult ConstructExchangeDataForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
280 : HcclResult ParseExchangeData(u8*& exchangeDataPtr, u64& exchangeDataBlankSize);
281 : HcclResult ExchangeTgidMesg();
282 : // 以下两个接口用于ibv、tcp进行信息交换、校验
283 : HcclResult RecvAndCheckExchangeData(void);
284 : HcclResult SendExchangeData(void);
285 :
286 : // 以下接口仅用于P2P和host shm中转的子类
287 : HcclResult SendNotifyReadyMesg();
288 : HcclResult SendNotifyDoneMesg();
289 : HcclResult SendDeviceIpcNotifyReadyMesg();
290 : HcclResult SendDeviceIpcNotifyDoneMesg();
291 : HcclResult RecvNotifyReadyMesg();
292 : HcclResult RecvNotifyDoneMesg();
293 : HcclResult RecvDeviceIpcNotifyReadyMesg();
294 : HcclResult RecvDeviceIpcNotifyDoneMesg();
295 : HcclResult CheckLinkStatus();
296 : HcclResult CheckLinkMode();
297 : HcclResult LinkSendNotifyMesg();
298 : HcclResult LinkRecvNotifyMesg();
299 :
300 : // 以下接口用于aicpu侧的transport子类
301 : HcclResult SetNotify();
302 : HcclResult SetNotifyPtr(const TransportDeviceP2pData &transDevP2pData);
303 : HcclResult SignalInit(const std::shared_ptr<LocalNotify> ¬ify, std::shared_ptr<LocalIpcNotify> &ipcNotify);
304 :
305 : void SignalDestroy(); // TransportP2P & TranshportShm 公有信号销毁函数
306 : void DestroyDeviceSignal();
307 : void DestroyHostSignal();
308 : HcclResult CheckDeviceId();
309 :
310 85 : inline HcclResult CheckExchangeData()
311 : {
312 85 : CHK_PRT_RET(machinePara_.exchangeInfo.size() > MAX_EXCHANGE_DATA_LEN,
313 : HCCL_ERROR("[[Check][ExchangeData]errNo[0x%016llx]custom exchange data size[%zu]is too large, "
314 : "Expected to less than[%llu]", HCCL_ERROR_CODE(HCCL_E_PARA), machinePara_.exchangeInfo.size(),
315 : MAX_EXCHANGE_DATA_LEN), HCCL_E_PARA);
316 85 : return HCCL_SUCCESS;
317 : }
318 : u64 exchangeDataTotalSize_;
319 : std::vector<u8> exchangeDataForSend_;
320 : std::vector<u8> exchangeDataForRecv_;
321 : DispatcherPub *dispatcher_;
322 : const std::unique_ptr<NotifyPool> ¬ifyPool_;
323 : std::shared_ptr<HcclSocket> defaultSocket_;
324 : MachinePara machinePara_;
325 : const std::chrono::milliseconds timeout_;
326 : std::shared_ptr<LocalIpcNotify> localSendReadyNotify_ = nullptr;
327 : std::shared_ptr<LocalIpcNotify> localSendDoneNotify_ = nullptr;
328 : std::shared_ptr<LocalIpcNotify> localSendReadyDeviceNotify_ = nullptr;
329 : std::shared_ptr<LocalIpcNotify> localSendDoneDeviceNotify_ = nullptr;
330 : std::vector<std::shared_ptr<LocalIpcNotify>> userLocalNotify_;
331 :
332 : std::shared_ptr<RemoteNotify> remoteSendReadyNotify_ = nullptr;
333 : std::shared_ptr<RemoteNotify> remoteSendDoneNotify_ = nullptr;
334 :
335 : std::shared_ptr<RemoteNotify> remoteSendReadyDeviceNotify_ = nullptr;
336 : std::shared_ptr<RemoteNotify> remoteSendDoneDeviceNotify_ = nullptr;
337 : std::vector<std::shared_ptr<RemoteNotify>> userRemoteNotify_;
338 :
339 : u64 remoteSendReadyAddress_;
340 : u64 remoteSendReadyOffset_;
341 : u64 remoteSendDoneOffset_;
342 : u64 remoteSendDoneAddress_;
343 : std::vector<u64> userRemoteNotifyAddr_;
344 : std::vector<u64> userRemoteNotifyOffset_;
345 :
346 : s32 recvPid_;
347 : s32 recvSdid_; // 超节点上device唯一标识, super pod device id
348 : NICDeployment nicDeploy_;
349 :
350 : bool useOneDoorbell_;
351 : TransportAttr transportAttr_;
352 : u32 notifyNum_;
353 :
354 : std::atomic<bool> stopFlag_{false};
355 : std::vector<u8> exchangeMsg_;
356 : bool useAtomicWrite_{false}; // 本端和对端同时使能atomic write时，才会使用atomic write，否则退化回普通模式
357 : };
358 :
359 : } // namespace hccl
360 :
361 : #endif /* TRANSPORT_BASE_PUB_H */
|
