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