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