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 : #include "transport.h"
12 : #include "transport_base.h"
13 : #include "transport_ibverbs.h"
14 : #include "transport_direct_npu.h"
15 : #ifdef CCL_KERNEL
16 : #include "transport_device_p2p.h"
17 : #include "transport_device_ibverbs.h"
18 : #endif
19 : #include "transport_p2p.h"
20 : #include "transport_virtural.h"
21 : namespace hccl {
22 : std::mutex Transport::mapMutex_;
23 : std::unordered_map<TransportBase*, Transport*> Transport::transportMap_;
24 21 : Transport::Transport(
25 : TransportType type, TransportPara& para, const HcclDispatcher dispatcherPtr,
26 : const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
27 : [[maybe_unused]] const TransportDeviceP2pData& transDevP2pData,
28 21 : [[maybe_unused]] const TransportDeviceIbverbsData& transDevIbverbsData)
29 21 : : type_(type)
30 : {
31 21 : DispatcherPub* dispatcher = reinterpret_cast<DispatcherPub*>(const_cast<HcclDispatcher>(dispatcherPtr));
32 21 : if (type == TransportType::TRANS_TYPE_IBV_EXP) {
33 0 : pimpl_ = new (std::nothrow) TransportIbverbs(dispatcher, notifyPool, machinePara, para.timeout);
34 0 : if (pimpl_ != nullptr) {
35 0 : std::lock_guard<std::mutex> maplock(mapMutex_);
36 0 : transportMap_.insert({pimpl_, this});
37 0 : }
38 21 : } else if (type == TransportType::TRANS_TYPE_DEVICE_DIRECT) {
39 0 : pimpl_ = new (std::nothrow) TransportDirectNpu(dispatcher, notifyPool, machinePara, para.timeout);
40 0 : if (pimpl_ != nullptr) {
41 0 : std::lock_guard<std::mutex> maplock(mapMutex_);
42 0 : transportMap_.insert({pimpl_, this});
43 0 : }
44 21 : } else if (type == TransportType::TRANS_TYPE_P2P) {
45 2 : pimpl_ = new (std::nothrow) TransportP2p(dispatcher, notifyPool, machinePara, para.timeout);
46 19 : } else if (type == TransportType::TRANS_TYPE_DEVICE_P2P) {
47 : #ifdef CCL_KERNEL
48 : pimpl_
49 0 : = new (std::nothrow) TransportDeviceP2p(dispatcher, notifyPool, machinePara, para.timeout, transDevP2pData);
50 : // 创建设备间P2P传输
51 : #else
52 : HCCL_ERROR("TRANS_TYPE_DEVICE_P2P Only running on the AICPU");
53 : #endif
54 19 : } else if (type == TransportType::TRANS_TYPE_DEVICE_IBVERBS) {
55 : #ifdef CCL_KERNEL
56 2 : pimpl_ = new (std::nothrow)
57 2 : TransportDeviceIbverbs(dispatcher, notifyPool, machinePara, para.timeout, transDevIbverbsData);
58 : #else
59 : HCCL_ERROR("TRANS_TYPE_DEVICE_IBVERBS Only running on the AICPU");
60 : #endif
61 18 : } else if (para.virtualFlag) {
62 18 : pimpl_ = new (std::nothrow) TransportVirtural(dispatcher, notifyPool, machinePara, para.timeout, para.index);
63 : } else {
64 0 : pimpl_ = new (std::nothrow) TransportBase(dispatcher, notifyPool, machinePara, para.timeout);
65 : }
66 21 : CHK_PRT_CONT(
67 : pimpl_ == nullptr,
68 : HCCL_ERROR("[Transport][Transport] create pimpl_ failed, type[%d].", static_cast<int>(type)));
69 21 : HCCL_DEBUG("Transport::Transport, type = %d", static_cast<int>(type));
70 21 : }
71 :
72 116 : Transport::~Transport()
73 : {
74 116 : std::unique_lock<std::mutex> maplock(mapMutex_);
75 116 : if (transportMap_.find(pimpl_) != transportMap_.end()) {
76 0 : transportMap_.erase(pimpl_);
77 : }
78 116 : maplock.unlock();
79 :
80 116 : delete pimpl_;
81 116 : pimpl_ = nullptr;
82 116 : }
83 :
84 86 : HcclResult Transport::Init()
85 : {
86 86 : CHK_PTR_NULL(pimpl_);
87 86 : return pimpl_->Init();
88 : }
89 :
90 1 : HcclResult Transport::DeInit()
91 : {
92 1 : CHK_PTR_NULL(pimpl_);
93 1 : return pimpl_->DeInit();
94 : }
95 :
96 0 : HcclResult Transport::TxDataSignal(Stream& stream)
97 : {
98 0 : CHK_PTR_NULL(pimpl_);
99 0 : return pimpl_->TxDataSignal(stream);
100 : }
101 :
102 9 : HcclResult Transport::RxDataSignal(Stream& stream)
103 : {
104 9 : CHK_PTR_NULL(pimpl_);
105 9 : return pimpl_->RxDataSignal(stream);
106 : }
107 :
108 9 : HcclResult Transport::TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
109 : {
110 9 : CHK_PTR_NULL(pimpl_);
111 : // src在transport内部校验
112 9 : return pimpl_->TxAsync(dstMemType, dstOffset, src, len, stream);
113 : }
114 :
115 0 : HcclResult Transport::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream)
116 : {
117 0 : CHK_PTR_NULL(pimpl_);
118 0 : return pimpl_->TxAsync(txMems, stream);
119 : }
120 :
121 5 : HcclResult Transport::TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
122 : {
123 5 : CHK_PTR_NULL(pimpl_);
124 : // src在transport内部校验
125 5 : return pimpl_->TxData(dstMemType, dstOffset, src, len, stream);
126 : }
127 :
128 2 : HcclResult Transport::RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
129 : {
130 2 : CHK_PTR_NULL(pimpl_);
131 : // dst在transport内部校验
132 2 : return pimpl_->RxData(srcMemType, srcOffset, dst, len, stream);
133 : }
134 :
135 5 : HcclResult Transport::TxPrepare(Stream& stream)
136 : {
137 5 : CHK_PTR_NULL(pimpl_);
138 5 : return pimpl_->TxPrepare(stream);
139 : }
140 :
141 2 : HcclResult Transport::RxPrepare(Stream& stream)
142 : {
143 2 : CHK_PTR_NULL(pimpl_);
144 2 : return pimpl_->RxPrepare(stream);
145 : }
146 :
147 5 : HcclResult Transport::TxDone(Stream& stream)
148 : {
149 5 : CHK_PTR_NULL(pimpl_);
150 5 : return pimpl_->TxDone(stream);
151 : }
152 :
153 2 : HcclResult Transport::RxDone(Stream& stream)
154 : {
155 2 : CHK_PTR_NULL(pimpl_);
156 2 : return pimpl_->RxDone(stream);
157 : }
158 :
159 0 : HcclResult Transport::Stop()
160 : {
161 0 : CHK_PTR_NULL(pimpl_);
162 0 : return pimpl_->Stop();
163 : }
164 :
165 0 : HcclResult Transport::Resume()
166 : {
167 0 : CHK_PTR_NULL(pimpl_);
168 0 : return pimpl_->Resume();
169 : }
170 :
171 0 : HcclResult Transport::TxWithReduce(
172 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, const HcclDataType datatype, HcclReduceOp redOp,
173 : Stream& stream)
174 : {
175 0 : CHK_PTR_NULL(pimpl_);
176 : // src在transport内部校验
177 0 : return pimpl_->TxWithReduce(dstMemType, dstOffset, src, len, datatype, redOp, stream);
178 : }
179 :
180 0 : HcclResult Transport::TxWithReduce(
181 : const std::vector<TxMemoryInfo>& txWithReduceMems, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream)
182 : {
183 0 : CHK_PTR_NULL(pimpl_);
184 0 : return pimpl_->TxWithReduce(txWithReduceMems, datatype, redOp, stream);
185 : }
186 :
187 0 : HcclResult Transport::RxWithReduce(
188 : UserMemType recvSrcMemType, u64 recvSrcOffset, void* recvDst, u64 recvLen, void* reduceSrc, void* reduceDst,
189 : u64 reduceDataCount, HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream& stream, const u64 reduceAttr)
190 : {
191 0 : CHK_PTR_NULL(pimpl_);
192 0 : CHK_PTR_NULL(recvDst);
193 0 : CHK_PTR_NULL(reduceSrc);
194 0 : CHK_PTR_NULL(reduceDst);
195 0 : return pimpl_->RxWithReduce(
196 : recvSrcMemType, recvSrcOffset, recvDst, recvLen, reduceSrc, reduceDst, reduceDataCount, reduceDatatype,
197 0 : reduceOp, stream, reduceAttr);
198 : }
199 :
200 0 : HcclResult Transport::RxWithReduce(
201 : const std::vector<RxWithReduceMemoryInfo>& rxWithReduceMems, HcclDataType reduceDatatype, HcclReduceOp reduceOp,
202 : Stream& stream, const u64 reduceAttr)
203 : {
204 0 : CHK_PTR_NULL(pimpl_);
205 0 : return pimpl_->RxWithReduce(rxWithReduceMems, reduceDatatype, reduceOp, stream, reduceAttr);
206 : }
207 :
208 18 : bool Transport::IsSupportTransportWithReduce()
209 : {
210 18 : if (pimpl_ == nullptr) {
211 0 : return false;
212 : }
213 18 : return pimpl_->IsSupportTransportWithReduce();
214 : }
215 :
216 9 : HcclResult Transport::RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
217 : {
218 9 : CHK_PTR_NULL(pimpl_);
219 : // dst在transport内部校验
220 9 : return pimpl_->RxAsync(srcMemType, srcOffset, dst, len, stream);
221 : }
222 :
223 0 : HcclResult Transport::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream)
224 : {
225 0 : CHK_PTR_NULL(pimpl_);
226 0 : return pimpl_->RxAsync(rxMems, stream);
227 : }
228 :
229 0 : HcclResult Transport::DataReceivedAck(Stream& stream)
230 : {
231 0 : CHK_PTR_NULL(pimpl_);
232 0 : return pimpl_->DataReceivedAck(stream);
233 : }
234 :
235 18 : HcclResult Transport::TxAck(Stream& stream)
236 : {
237 18 : CHK_PTR_NULL(pimpl_);
238 18 : return pimpl_->TxAck(stream);
239 : }
240 :
241 9 : HcclResult Transport::RxAck(Stream& stream)
242 : {
243 9 : CHK_PTR_NULL(pimpl_);
244 9 : return pimpl_->RxAck(stream);
245 : }
246 :
247 0 : HcclResult Transport::TxWaitDone(Stream& stream)
248 : {
249 0 : CHK_PTR_NULL(pimpl_);
250 0 : return pimpl_->TxWaitDone(stream);
251 : }
252 :
253 0 : HcclResult Transport::RxWaitDone(Stream& stream)
254 : {
255 0 : CHK_PTR_NULL(pimpl_);
256 0 : return pimpl_->RxWaitDone(stream);
257 : }
258 :
259 0 : HcclResult Transport::Post(u32 notifyIdx, Stream& stream)
260 : {
261 0 : CHK_PTR_NULL(pimpl_);
262 0 : return pimpl_->Post(notifyIdx, stream);
263 : }
264 :
265 0 : HcclResult Transport::Wait(u32 notifyIdx, Stream& stream, const u32 timeOut)
266 : {
267 0 : CHK_PTR_NULL(pimpl_);
268 0 : return pimpl_->Wait(notifyIdx, stream, timeOut);
269 : }
270 :
271 0 : u32 Transport::GetNotifyNum()
272 : {
273 0 : CHK_PTR_NULL(pimpl_);
274 0 : return pimpl_->GetNotifyNum();
275 : }
276 :
277 0 : HcclResult Transport::GetLocalNotify(std::vector<HcclSignalInfo>& localNotify)
278 : {
279 0 : CHK_PTR_NULL(pimpl_);
280 0 : return pimpl_->GetLocalNotify(localNotify);
281 : }
282 :
283 0 : HcclResult Transport::GetRemoteNotify(std::vector<HcclSignalInfo>& localNotify)
284 : {
285 0 : CHK_PTR_NULL(pimpl_);
286 0 : return pimpl_->GetRemoteNotify(localNotify);
287 : }
288 :
289 0 : HcclResult Transport::GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t* memNum, HcclMemType memType)
290 : {
291 0 : CHK_PTR_NULL(pimpl_);
292 0 : return pimpl_->GetIndOpRemoteMemDetails(remoteMem, memNum, memType);
293 : }
294 :
295 0 : HcclResult Transport::GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum)
296 : {
297 0 : CHK_PTR_NULL(pimpl_);
298 0 : return pimpl_->GetIndOpRemoteMem(remoteMem, memNum);
299 : }
300 :
301 0 : HcclResult Transport::GetRemoteMem(UserMemType memType, void** remotePtr)
302 : {
303 0 : CHK_PTR_NULL(pimpl_);
304 0 : CHK_PTR_NULL(remotePtr);
305 0 : return pimpl_->GetRemoteMem(memType, remotePtr);
306 : }
307 :
308 0 : HcclResult Transport::GetRemoteMem(std::vector<void*>* remotePtrVec)
309 : {
310 0 : CHK_PTR_NULL(pimpl_);
311 0 : CHK_PTR_NULL(remotePtrVec);
312 0 : return pimpl_->GetRemoteMem(remotePtrVec);
313 : }
314 :
315 0 : HcclResult Transport::GetRemoteMemKey(UserMemType memType, uint32_t* remoteMemKey)
316 : {
317 0 : CHK_PTR_NULL(pimpl_);
318 0 : return pimpl_->GetRemoteMemKey(memType, remoteMemKey);
319 : }
320 :
321 0 : HcclResult Transport::GetLocalRdmaNotify(std::vector<HcclSignalInfo>& rdmaNotify)
322 : {
323 0 : CHK_PTR_NULL(pimpl_);
324 0 : return pimpl_->GetLocalRdmaNotify(rdmaNotify);
325 : }
326 :
327 0 : HcclResult Transport::GetDrainLocalDataNotify(void*& localAddr, uint32_t& lkey, HcclSignalInfo& dataNotify)
328 : {
329 0 : CHK_PTR_NULL(pimpl_);
330 0 : return pimpl_->GetDrainLocalDataNotify(localAddr, lkey, dataNotify);
331 : }
332 :
333 0 : HcclResult Transport::GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey>& rdmaNotifyAddr)
334 : {
335 0 : CHK_PTR_NULL(pimpl_);
336 0 : return pimpl_->GetRemoteRdmaNotifyAddrKey(rdmaNotifyAddr);
337 : }
338 :
339 0 : HcclResult Transport::GetLocalNotifyValueAddrKey(std::vector<AddrKey>& notifyValue)
340 : {
341 0 : CHK_PTR_NULL(pimpl_);
342 0 : return pimpl_->GetLocalNotifyValueAddrKey(notifyValue);
343 : }
344 :
345 0 : HcclResult Transport::GetLocalMemDetails(UserMemType memType, MemDetails& memDetails)
346 : {
347 0 : CHK_PTR_NULL(pimpl_);
348 0 : return pimpl_->GetLocalMemDetails(memType, memDetails);
349 : }
350 :
351 0 : HcclResult Transport::GetChipId(s64& chipId)
352 : {
353 0 : CHK_PTR_NULL(pimpl_);
354 0 : return pimpl_->GetChipId(chipId);
355 : }
356 :
357 0 : HcclResult Transport::GetAiQpInfo(std::vector<HcclQpInfoV2>& aiQpInfo)
358 : {
359 0 : CHK_PTR_NULL(pimpl_);
360 0 : return pimpl_->GetAiQpInfo(aiQpInfo);
361 : }
362 0 : HcclResult Transport::GetTransportId(u32& id)
363 : {
364 0 : CHK_PTR_NULL(pimpl_);
365 0 : return pimpl_->GetTransportId(id);
366 : }
367 :
368 0 : HcclResult Transport::GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo>& aiRMAQueueInfo)
369 : {
370 0 : CHK_PTR_NULL(pimpl_);
371 0 : return pimpl_->GetAiRMAQueueInfo(aiRMAQueueInfo);
372 : }
373 :
374 0 : HcclResult Transport::GetRemoteMemSize(UserMemType memType, u64& size)
375 : {
376 0 : CHK_PTR_NULL(pimpl_);
377 0 : return pimpl_->GetRemoteMemSize(memType, size);
378 : }
379 :
380 1 : HcclResult Transport::GetTxAckDevNotifyInfo(HcclSignalInfo& notifyInfo)
381 : {
382 1 : CHK_PTR_NULL(pimpl_);
383 1 : return pimpl_->GetTxAckDevNotifyInfo(notifyInfo);
384 : }
385 :
386 1 : HcclResult Transport::GetRxAckDevNotifyInfo(HcclSignalInfo& notifyInfo)
387 : {
388 1 : CHK_PTR_NULL(pimpl_);
389 1 : return pimpl_->GetRxAckDevNotifyInfo(notifyInfo);
390 : }
391 :
392 1 : HcclResult Transport::GetTxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo)
393 : {
394 1 : CHK_PTR_NULL(pimpl_);
395 1 : return pimpl_->GetTxDataSigleDevNotifyInfo(notifyInfo);
396 : }
397 :
398 1 : HcclResult Transport::GetRxDataSigleDevNotifyInfo(HcclSignalInfo& notifyInfo)
399 : {
400 1 : CHK_PTR_NULL(pimpl_);
401 1 : return pimpl_->GetRxDataSigleDevNotifyInfo(notifyInfo);
402 : }
403 :
404 0 : hccl::LinkType Transport::GetLinkType() const
405 : {
406 0 : if (pimpl_ == nullptr) {
407 0 : return hccl::LinkType::LINK_RESERVED;
408 : }
409 0 : return pimpl_->GetLinkType();
410 : }
411 :
412 9 : bool Transport::GetSupportDataReceivedAck() const
413 : {
414 9 : if (pimpl_ == nullptr) {
415 0 : return false;
416 : }
417 9 : return pimpl_->GetSupportDataReceivedAck();
418 : }
419 :
420 0 : void Transport::SetSupportDataReceivedAck(bool supportDataReceivedAck)
421 : {
422 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
423 0 : pimpl_->SetSupportDataReceivedAck(supportDataReceivedAck);
424 : }
425 :
426 27 : bool Transport::IsSpInlineReduce() const
427 : {
428 27 : if (pimpl_ == nullptr) {
429 0 : return false;
430 : }
431 27 : return pimpl_->IsSpInlineReduce();
432 : }
433 :
434 0 : u32 Transport::GetRemoteRank()
435 : {
436 0 : if (pimpl_ == nullptr) {
437 0 : return INVALID_VALUE_RANKID;
438 : }
439 0 : return pimpl_->GetRemoteRank();
440 : }
441 :
442 0 : HcclResult Transport::ConnectAsync(u32& status)
443 : {
444 0 : CHK_PTR_NULL(pimpl_);
445 0 : return pimpl_->ConnectAsync(status);
446 : }
447 :
448 0 : HcclResult Transport::ConnectQuerry(u32& status)
449 : {
450 0 : CHK_PTR_NULL(pimpl_);
451 0 : return pimpl_->ConnectQuerry(status);
452 : }
453 :
454 0 : void Transport::Break()
455 : {
456 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
457 0 : pimpl_->Break();
458 : }
459 :
460 0 : void Transport::EnableUseOneDoorbell()
461 : {
462 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
463 0 : pimpl_->EnableUseOneDoorbell();
464 : }
465 :
466 0 : bool Transport::GetUseOneDoorbellValue()
467 : {
468 0 : if (pimpl_ == nullptr) {
469 0 : return false;
470 : }
471 0 : return pimpl_->GetUseOneDoorbellValue();
472 : }
473 :
474 0 : HcclResult Transport::GetTransportAttr(TransportAttr& attr)
475 : {
476 0 : CHK_PTR_NULL(pimpl_);
477 0 : attr = pimpl_->GetTransportAttr();
478 0 : return HCCL_SUCCESS;
479 : }
480 :
481 0 : HcclResult Transport::TxEnv(const void* ptr, const u64 len, Stream& stream)
482 : {
483 0 : CHK_PTR_NULL(pimpl_);
484 0 : CHK_PTR_NULL(ptr);
485 0 : return pimpl_->TxEnv(ptr, len, stream);
486 : }
487 :
488 0 : HcclResult Transport::RxEnv(Stream& stream)
489 : {
490 0 : CHK_PTR_NULL(pimpl_);
491 0 : return pimpl_->RxEnv(stream);
492 : }
493 :
494 0 : bool Transport::IsTransportRoce() { return false; }
495 :
496 0 : HcclResult Transport::WriteAsync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream)
497 : {
498 0 : CHK_PTR_NULL(pimpl_);
499 0 : CHK_PTR_NULL(remoteBuf.addr);
500 0 : CHK_PTR_NULL(localBuf.addr);
501 : // localAddr在transport内部校验
502 0 : return pimpl_->WriteAsync(remoteBuf, localBuf, stream);
503 : }
504 :
505 0 : HcclResult Transport::WriteSync(struct Buffer& remoteBuf, struct Buffer& localBuf, Stream& stream)
506 : {
507 0 : CHK_PTR_NULL(pimpl_);
508 0 : CHK_PTR_NULL(remoteBuf.addr);
509 0 : CHK_PTR_NULL(localBuf.addr);
510 : // localAddr在transport内部校验
511 0 : return pimpl_->WriteSync(remoteBuf, localBuf, stream);
512 : }
513 :
514 0 : HcclResult Transport::WriteReduceAsync(
515 : struct Buffer& remoteBuf, struct Buffer& localBuf, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream)
516 : {
517 0 : CHK_PTR_NULL(pimpl_);
518 0 : CHK_PTR_NULL(remoteBuf.addr);
519 0 : CHK_PTR_NULL(localBuf.addr);
520 : // localAddr在transport内部校验
521 0 : return pimpl_->WriteReduceAsync(remoteBuf, localBuf, datatype, redOp, stream);
522 : }
523 :
524 0 : HcclResult Transport::ReadAsync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream)
525 : {
526 0 : CHK_PTR_NULL(pimpl_);
527 0 : CHK_PTR_NULL(remoteBuf.addr);
528 0 : CHK_PTR_NULL(localBuf.addr);
529 : // localAddr在transport内部校验
530 0 : return pimpl_->ReadAsync(localBuf, remoteBuf, stream);
531 : }
532 :
533 0 : HcclResult Transport::ReadSync(struct Buffer& localBuf, struct Buffer& remoteBuf, Stream& stream)
534 : {
535 0 : CHK_PTR_NULL(pimpl_);
536 0 : CHK_PTR_NULL(remoteBuf.addr);
537 0 : CHK_PTR_NULL(localBuf.addr);
538 : // localAddr在transport内部校验
539 0 : return pimpl_->ReadSync(localBuf, remoteBuf, stream);
540 : }
541 :
542 0 : HcclResult Transport::ReadReduceSync(
543 : struct Buffer& localBuf, struct Buffer& remoteBuf, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream)
544 : {
545 0 : CHK_PTR_NULL(pimpl_);
546 0 : CHK_PTR_NULL(localBuf.addr);
547 0 : CHK_PTR_NULL(remoteBuf.addr);
548 0 : return pimpl_->ReadReduceSync(localBuf, remoteBuf, datatype, redOp, stream);
549 : }
550 :
551 0 : HcclResult Transport::BatchTransferAsync(const HcommBatchTransferDesc* transferDescs, uint32_t descNum, Stream& stream)
552 : {
553 0 : CHK_PTR_NULL(pimpl_);
554 0 : return pimpl_->BatchTransferAsync(transferDescs, descNum, stream);
555 : }
556 :
557 0 : HcclResult Transport::PostReady(Stream& stream)
558 : {
559 0 : CHK_PTR_NULL(pimpl_);
560 0 : return pimpl_->PostReady(stream);
561 : }
562 :
563 0 : HcclResult Transport::WaitReady(Stream& stream)
564 : {
565 0 : CHK_PTR_NULL(pimpl_);
566 0 : return pimpl_->WaitReady(stream);
567 : }
568 :
569 0 : HcclResult Transport::PostFin(Stream& stream)
570 : {
571 0 : CHK_PTR_NULL(pimpl_);
572 0 : return pimpl_->PostFin(stream);
573 : }
574 :
575 0 : HcclResult Transport::WaitFin(Stream& stream)
576 : {
577 0 : CHK_PTR_NULL(pimpl_);
578 0 : return pimpl_->WaitFin(stream);
579 : }
580 :
581 9 : HcclResult Transport::PostFinAck(Stream& stream)
582 : {
583 9 : CHK_PTR_NULL(pimpl_);
584 9 : return pimpl_->PostFinAck(stream);
585 : }
586 :
587 9 : HcclResult Transport::WaitFinAck(Stream& stream)
588 : {
589 9 : CHK_PTR_NULL(pimpl_);
590 9 : return pimpl_->WaitFinAck(stream);
591 : }
592 :
593 0 : HcclResult Transport::SetStopFlag(bool value)
594 : {
595 0 : if (pimpl_ != nullptr) {
596 0 : return pimpl_->SetStopFlag(value);
597 : }
598 0 : return HCCL_SUCCESS;
599 : }
600 :
601 0 : HcclResult Transport::UpdateRemoteAddr(void* remoteIn, void* remoteOut)
602 : {
603 0 : CHK_PTR_NULL(pimpl_);
604 0 : CHK_PTR_NULL(remoteIn);
605 0 : CHK_PTR_NULL(remoteOut);
606 0 : return pimpl_->UpdateRemoteAddr(remoteIn, remoteOut);
607 : }
608 :
609 0 : std::vector<u8> Transport::GetExchangeInfo()
610 : {
611 0 : if (UNLIKELY(pimpl_ == nullptr)) {
612 0 : return std::vector<u8>();
613 : }
614 0 : return pimpl_->GetExchangeInfo();
615 : }
616 :
617 14 : HcclResult Transport::GetTransportErrorCqe(
618 : const HcclNetDevCtx netDevCtx, std::vector<std::pair<Transport*, CqeInfo>>& infos, u32& num)
619 : {
620 14 : CHK_PTR_NULL(netDevCtx);
621 14 : HcclIpAddress localIp;
622 14 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
623 :
624 14 : std::vector<std::pair<TransportBase*, CqeInfo>> infolist;
625 14 : CHK_RET(TransportIbverbs::GetTransportErrorCqe(netDevCtx, infolist, num));
626 :
627 14 : std::lock_guard<std::mutex> maplock(mapMutex_);
628 14 : for (auto info : infolist) {
629 0 : auto iter = transportMap_.find(info.first);
630 0 : if (iter != transportMap_.end()) {
631 0 : infos.push_back(std::make_pair(iter->second, info.second));
632 : } else {
633 0 : HCCL_RUN_WARNING(
634 : "[GetTransportErrorCqe]get err failed, transport is not find, localIp[%s], remoteIp[%s]",
635 : localIp.GetReadableAddress(), info.second.remoteIp.GetReadableAddress());
636 : }
637 0 : }
638 14 : num = infos.size();
639 :
640 14 : return HCCL_SUCCESS;
641 14 : }
642 :
643 0 : HcclResult Transport::Fence()
644 : {
645 0 : CHK_PTR_NULL(pimpl_);
646 0 : return pimpl_->Fence();
647 : }
648 :
649 0 : bool Transport::GetIsUseAtomicWrite()
650 : {
651 0 : if (pimpl_ == nullptr) {
652 0 : return false;
653 : }
654 0 : return pimpl_->GetIsUseAtomicWrite();
655 : }
656 :
657 0 : HcclResult Transport::GetSpecificNotify(HcclSignalInfo& notifyInfo, bool& isValid, const std::string& notifyName)
658 : {
659 0 : CHK_PTR_NULL(pimpl_);
660 0 : return pimpl_->GetSpecificNotify(notifyInfo, isValid, notifyName);
661 : }
662 :
663 0 : HcclResult Transport::HcclBatchRead(
664 : [[maybe_unused]] const TransportDeviceNormalData& ibvData, [[maybe_unused]] struct MemDetails* localMems,
665 : [[maybe_unused]] struct MemDetails* remoteMems, [[maybe_unused]] u32 memNum, [[maybe_unused]] u64& dbInfo)
666 : {
667 : #ifdef CCL_KERNEL
668 0 : return TransportDeviceIbverbs::HnsPostSend(
669 0 : ibvData, localMems, remoteMems, memNum, HcclWrOpCode::HCCL_WR_RDMA_READ, dbInfo);
670 : #else
671 : HCCL_ERROR("[Transport][HcclBatchRead]Does not support this interface.");
672 : return HCCL_E_NOT_SUPPORT;
673 : #endif
674 : }
675 :
676 1 : HcclResult Transport::SetDeviceUnavailable(u32 deviceId)
677 : {
678 1 : return MemNameRepository::GetInstance(deviceId)->SetDeviceUnavailable(true);
679 : }
680 :
681 0 : HcclResult Transport::HcclBatchWrite(
682 : [[maybe_unused]] const TransportDeviceNormalData& ibvData, [[maybe_unused]] struct MemDetails* localMems,
683 : [[maybe_unused]] struct MemDetails* remoteMems, [[maybe_unused]] u32 memNum, [[maybe_unused]] u64& dbInfo)
684 : {
685 : #ifdef CCL_KERNEL
686 0 : return TransportDeviceIbverbs::HnsPostSend(
687 0 : ibvData, localMems, remoteMems, memNum, HcclWrOpCode::HCCL_WR_RDMA_WRITE, dbInfo);
688 : #else
689 : HCCL_ERROR("[Transport][HcclBatchWrite]Does not support this interface.");
690 : return HCCL_E_NOT_SUPPORT;
691 : #endif
692 : }
693 :
694 0 : HcclResult Transport::Drain(Stream& stream)
695 : {
696 0 : CHK_PTR_NULL(pimpl_);
697 0 : return pimpl_->Drain(stream);
698 : }
699 :
700 1 : HcclResult Transport::InitDrainNotifyInfo()
701 : {
702 1 : CHK_PTR_NULL(pimpl_);
703 1 : return pimpl_->InitDrainNotifyInfo();
704 : }
705 :
706 0 : HcclResult Transport::GetDrainRemSrcMem(void*& remoteAddr, uint32_t& remoteKey, uint32_t& size)
707 : {
708 0 : CHK_PTR_NULL(pimpl_);
709 0 : return pimpl_->GetDrainRemSrcMem(remoteAddr, remoteKey, size);
710 : }
711 : } // namespace hccl
|
