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(TransportType type, TransportPara& para,
25 : const HcclDispatcher dispatcherPtr,
26 : const std::unique_ptr<NotifyPool> ¬ifyPool,
27 : MachinePara &machinePara,
28 : const TransportDeviceP2pData &transDevP2pData,
29 21 : const TransportDeviceIbverbsData &transDevIbverbsData) : 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 0 : 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) TransportDeviceIbverbs(dispatcher, notifyPool,
57 2 : 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 36 : pimpl_ = new (std::nothrow) TransportVirtural(dispatcher, notifyPool, machinePara,
63 36 : para.timeout, para.index);
64 : } else {
65 0 : pimpl_ = new (std::nothrow) TransportBase(dispatcher, notifyPool, machinePara, para.timeout);
66 : }
67 21 : CHK_PRT_CONT(pimpl_ == nullptr, HCCL_ERROR("[Transport][Transport] create pimpl_ failed, type[%d].",
68 : 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,
173 : const HcclDataType datatype, HcclReduceOp redOp, 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(const std::vector<TxMemoryInfo> &txWithReduceMems,
181 : 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,
189 : void *reduceSrc, void *reduceDst, u64 reduceDataCount, HcclDataType reduceDatatype,
190 : HcclReduceOp reduceOp, Stream &stream, const u64 reduceAttr)
191 : {
192 0 : CHK_PTR_NULL(pimpl_);
193 0 : CHK_PTR_NULL(recvDst);
194 0 : CHK_PTR_NULL(reduceSrc);
195 0 : CHK_PTR_NULL(reduceDst);
196 0 : return pimpl_->RxWithReduce(recvSrcMemType, recvSrcOffset, recvDst, recvLen,
197 0 : reduceSrc, reduceDst, reduceDataCount, reduceDatatype, reduceOp, stream, reduceAttr);
198 : }
199 :
200 0 : HcclResult Transport::RxWithReduce(
201 : const std::vector<RxWithReduceMemoryInfo> &rxWithReduceMems,
202 : HcclDataType reduceDatatype, HcclReduceOp reduceOp, Stream &stream,
203 : const u64 reduceAttr)
204 : {
205 0 : CHK_PTR_NULL(pimpl_);
206 0 : return pimpl_->RxWithReduce(rxWithReduceMems, reduceDatatype, reduceOp, stream, reduceAttr);
207 : }
208 :
209 18 : bool Transport::IsSupportTransportWithReduce()
210 : {
211 18 : if (pimpl_ == nullptr) {
212 0 : return false;
213 : }
214 18 : return pimpl_->IsSupportTransportWithReduce();
215 : }
216 :
217 9 : HcclResult Transport::RxAsync(UserMemType srcMemType, u64 srcOffset, void *dst, u64 len, Stream &stream)
218 : {
219 9 : CHK_PTR_NULL(pimpl_);
220 : // dst在transport内部校验
221 9 : return pimpl_->RxAsync(srcMemType, srcOffset, dst, len, stream);
222 : }
223 :
224 0 : HcclResult Transport::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream &stream)
225 : {
226 0 : CHK_PTR_NULL(pimpl_);
227 0 : return pimpl_->RxAsync(rxMems, stream);
228 : }
229 :
230 0 : HcclResult Transport::DataReceivedAck(Stream &stream)
231 : {
232 0 : CHK_PTR_NULL(pimpl_);
233 0 : return pimpl_->DataReceivedAck(stream);
234 : }
235 :
236 18 : HcclResult Transport::TxAck(Stream &stream)
237 : {
238 18 : CHK_PTR_NULL(pimpl_);
239 18 : return pimpl_->TxAck(stream);
240 : }
241 :
242 9 : HcclResult Transport::RxAck(Stream &stream)
243 : {
244 9 : CHK_PTR_NULL(pimpl_);
245 9 : return pimpl_->RxAck(stream);
246 : }
247 :
248 0 : HcclResult Transport::TxWaitDone(Stream &stream)
249 : {
250 0 : CHK_PTR_NULL(pimpl_);
251 0 : return pimpl_->TxWaitDone(stream);
252 : }
253 :
254 0 : HcclResult Transport::RxWaitDone(Stream &stream)
255 : {
256 0 : CHK_PTR_NULL(pimpl_);
257 0 : return pimpl_->RxWaitDone(stream);
258 : }
259 :
260 0 : HcclResult Transport::Post(u32 notifyIdx, Stream &stream)
261 : {
262 0 : CHK_PTR_NULL(pimpl_);
263 0 : return pimpl_->Post(notifyIdx, stream);
264 : }
265 :
266 0 : HcclResult Transport::Wait(u32 notifyIdx, Stream &stream, const u32 timeOut)
267 : {
268 0 : CHK_PTR_NULL(pimpl_);
269 0 : return pimpl_->Wait(notifyIdx, stream, timeOut);
270 : }
271 :
272 0 : u32 Transport::GetNotifyNum()
273 : {
274 0 : CHK_PTR_NULL(pimpl_);
275 0 : return pimpl_->GetNotifyNum();
276 : }
277 :
278 0 : HcclResult Transport::GetLocalNotify(std::vector<HcclSignalInfo> &localNotify)
279 : {
280 0 : CHK_PTR_NULL(pimpl_);
281 0 : return pimpl_->GetLocalNotify(localNotify);
282 : }
283 :
284 0 : HcclResult Transport::GetRemoteNotify(std::vector<HcclSignalInfo> &localNotify)
285 : {
286 0 : CHK_PTR_NULL(pimpl_);
287 0 : return pimpl_->GetRemoteNotify(localNotify);
288 : }
289 :
290 0 : HcclResult Transport::GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t *memNum, HcclMemType memType)
291 : {
292 0 : CHK_PTR_NULL(pimpl_);
293 0 : return pimpl_->GetIndOpRemoteMemDetails(remoteMem, memNum, memType);
294 : }
295 :
296 0 : HcclResult Transport::GetIndOpRemoteMem(HcclMem **remoteMem, uint32_t *memNum)
297 : {
298 0 : CHK_PTR_NULL(pimpl_);
299 0 : return pimpl_->GetIndOpRemoteMem(remoteMem, memNum);
300 : }
301 :
302 0 : HcclResult Transport::GetRemoteMem(UserMemType memType, void **remotePtr)
303 : {
304 0 : CHK_PTR_NULL(pimpl_);
305 0 : CHK_PTR_NULL(remotePtr);
306 0 : return pimpl_->GetRemoteMem(memType, remotePtr);
307 : }
308 :
309 0 : HcclResult Transport::GetRemoteMem(std::vector<void *> *remotePtrVec)
310 : {
311 0 : CHK_PTR_NULL(pimpl_);
312 0 : CHK_PTR_NULL(remotePtrVec);
313 0 : return pimpl_->GetRemoteMem(remotePtrVec);
314 : }
315 :
316 0 : HcclResult Transport::GetRemoteMemKey(UserMemType memType, uint32_t *remoteMemKey)
317 : {
318 0 : CHK_PTR_NULL(pimpl_);
319 0 : return pimpl_->GetRemoteMemKey(memType, remoteMemKey);
320 : }
321 :
322 0 : HcclResult Transport::GetLocalRdmaNotify(std::vector<HcclSignalInfo> &rdmaNotify)
323 : {
324 0 : CHK_PTR_NULL(pimpl_);
325 0 : return pimpl_->GetLocalRdmaNotify(rdmaNotify);
326 : }
327 :
328 0 : HcclResult Transport::GetDrainLocalDataNotify(void* &localAddr, uint32_t& lkey, HcclSignalInfo &dataNotify)
329 : {
330 0 : CHK_PTR_NULL(pimpl_);
331 0 : return pimpl_->GetDrainLocalDataNotify(localAddr, lkey, dataNotify);
332 : }
333 :
334 0 : HcclResult Transport::GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey> &rdmaNotifyAddr)
335 : {
336 0 : CHK_PTR_NULL(pimpl_);
337 0 : return pimpl_->GetRemoteRdmaNotifyAddrKey(rdmaNotifyAddr);
338 : }
339 :
340 0 : HcclResult Transport::GetLocalNotifyValueAddrKey(std::vector<AddrKey> ¬ifyValue)
341 : {
342 0 : CHK_PTR_NULL(pimpl_);
343 0 : return pimpl_->GetLocalNotifyValueAddrKey(notifyValue);
344 : }
345 :
346 0 : HcclResult Transport::GetLocalMemDetails(UserMemType memType, MemDetails &memDetails)
347 : {
348 0 : CHK_PTR_NULL(pimpl_);
349 0 : return pimpl_->GetLocalMemDetails(memType, memDetails);
350 : }
351 :
352 0 : HcclResult Transport::GetChipId(s64 &chipId)
353 : {
354 0 : CHK_PTR_NULL(pimpl_);
355 0 : return pimpl_->GetChipId(chipId);
356 : }
357 :
358 0 : HcclResult Transport::GetAiQpInfo(std::vector<HcclQpInfoV2> &aiQpInfo)
359 : {
360 0 : CHK_PTR_NULL(pimpl_);
361 0 : return pimpl_->GetAiQpInfo(aiQpInfo);
362 : }
363 0 : HcclResult Transport::GetTransportId(u32 &id)
364 : {
365 0 : CHK_PTR_NULL(pimpl_);
366 0 : return pimpl_->GetTransportId(id);
367 : }
368 :
369 0 : HcclResult Transport::GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo> &aiRMAQueueInfo)
370 : {
371 0 : CHK_PTR_NULL(pimpl_);
372 0 : return pimpl_->GetAiRMAQueueInfo(aiRMAQueueInfo);
373 : }
374 :
375 0 : HcclResult Transport::GetRemoteMemSize(UserMemType memType, u64 &size)
376 : {
377 0 : CHK_PTR_NULL(pimpl_);
378 0 : return pimpl_->GetRemoteMemSize(memType, size);
379 : }
380 :
381 1 : HcclResult Transport::GetTxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
382 : {
383 1 : CHK_PTR_NULL(pimpl_);
384 1 : return pimpl_->GetTxAckDevNotifyInfo(notifyInfo);
385 : }
386 :
387 1 : HcclResult Transport::GetRxAckDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
388 : {
389 1 : CHK_PTR_NULL(pimpl_);
390 1 : return pimpl_->GetRxAckDevNotifyInfo(notifyInfo);
391 : }
392 :
393 1 : HcclResult Transport::GetTxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
394 : {
395 1 : CHK_PTR_NULL(pimpl_);
396 1 : return pimpl_->GetTxDataSigleDevNotifyInfo(notifyInfo);
397 : }
398 :
399 1 : HcclResult Transport::GetRxDataSigleDevNotifyInfo(HcclSignalInfo ¬ifyInfo)
400 : {
401 1 : CHK_PTR_NULL(pimpl_);
402 1 : return pimpl_->GetRxDataSigleDevNotifyInfo(notifyInfo);
403 : }
404 :
405 0 : hccl::LinkType Transport::GetLinkType() const
406 : {
407 0 : if (pimpl_ == nullptr) {
408 0 : return hccl::LinkType::LINK_RESERVED;
409 : }
410 0 : return pimpl_->GetLinkType();
411 : }
412 :
413 9 : bool Transport::GetSupportDataReceivedAck() const
414 : {
415 9 : if (pimpl_ == nullptr) {
416 0 : return false;
417 : }
418 9 : return pimpl_->GetSupportDataReceivedAck();
419 : }
420 :
421 0 : void Transport::SetSupportDataReceivedAck(bool supportDataReceivedAck)
422 : {
423 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
424 0 : pimpl_->SetSupportDataReceivedAck(supportDataReceivedAck);
425 : }
426 :
427 27 : bool Transport::IsSpInlineReduce() const
428 : {
429 27 : if (pimpl_ == nullptr) {
430 0 : return false;
431 : }
432 27 : return pimpl_->IsSpInlineReduce();
433 : }
434 :
435 0 : u32 Transport::GetRemoteRank()
436 : {
437 0 : if (pimpl_ == nullptr) {
438 0 : return INVALID_VALUE_RANKID;
439 : }
440 0 : return pimpl_->GetRemoteRank();
441 : }
442 :
443 0 : HcclResult Transport::ConnectAsync(u32& status)
444 : {
445 0 : CHK_PTR_NULL(pimpl_);
446 0 : return pimpl_->ConnectAsync(status);
447 : }
448 :
449 0 : HcclResult Transport::ConnectQuerry(u32& status)
450 : {
451 0 : CHK_PTR_NULL(pimpl_);
452 0 : return pimpl_->ConnectQuerry(status);
453 : }
454 :
455 0 : void Transport::Break()
456 : {
457 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
458 0 : pimpl_->Break();
459 : }
460 :
461 0 : void Transport::EnableUseOneDoorbell()
462 : {
463 0 : CHK_SMART_PTR_RET_NULL(pimpl_);
464 0 : pimpl_->EnableUseOneDoorbell();
465 : }
466 :
467 0 : bool Transport::GetUseOneDoorbellValue()
468 : {
469 0 : if (pimpl_ == nullptr) {
470 0 : return false;
471 : }
472 0 : return pimpl_->GetUseOneDoorbellValue();
473 : }
474 :
475 0 : HcclResult Transport::GetTransportAttr(TransportAttr &attr)
476 : {
477 0 : CHK_PTR_NULL(pimpl_);
478 0 : attr = pimpl_->GetTransportAttr();
479 0 : return HCCL_SUCCESS;
480 : }
481 :
482 0 : HcclResult Transport::TxEnv(const void *ptr, const u64 len, Stream &stream)
483 : {
484 0 : CHK_PTR_NULL(pimpl_);
485 0 : CHK_PTR_NULL(ptr);
486 0 : return pimpl_->TxEnv(ptr, len, stream);
487 : }
488 :
489 0 : HcclResult Transport::RxEnv(Stream &stream)
490 : {
491 0 : CHK_PTR_NULL(pimpl_);
492 0 : return pimpl_->RxEnv(stream);
493 : }
494 :
495 0 : bool Transport::IsTransportRoce()
496 : {
497 0 : return false;
498 : }
499 :
500 0 : HcclResult Transport::WriteAsync(struct Buffer &remoteBuf, struct Buffer &localBuf, Stream &stream)
501 : {
502 0 : CHK_PTR_NULL(pimpl_);
503 0 : CHK_PTR_NULL(remoteBuf.addr);
504 0 : CHK_PTR_NULL(localBuf.addr);
505 : // localAddr在transport内部校验
506 0 : return pimpl_->WriteAsync(remoteBuf, localBuf, stream);
507 : }
508 :
509 0 : HcclResult Transport::WriteSync(struct Buffer &remoteBuf, struct Buffer &localBuf, Stream &stream)
510 : {
511 0 : CHK_PTR_NULL(pimpl_);
512 0 : CHK_PTR_NULL(remoteBuf.addr);
513 0 : CHK_PTR_NULL(localBuf.addr);
514 : // localAddr在transport内部校验
515 0 : return pimpl_->WriteSync(remoteBuf, localBuf, stream);
516 : }
517 :
518 0 : HcclResult Transport::WriteReduceAsync(struct Buffer &remoteBuf, struct Buffer &localBuf,
519 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
520 : {
521 0 : CHK_PTR_NULL(pimpl_);
522 0 : CHK_PTR_NULL(remoteBuf.addr);
523 0 : CHK_PTR_NULL(localBuf.addr);
524 : // localAddr在transport内部校验
525 0 : return pimpl_->WriteReduceAsync(remoteBuf, localBuf, datatype, redOp, stream);
526 : }
527 :
528 0 : HcclResult Transport::ReadAsync(struct Buffer &localBuf, struct Buffer &remoteBuf, Stream &stream)
529 : {
530 0 : CHK_PTR_NULL(pimpl_);
531 0 : CHK_PTR_NULL(remoteBuf.addr);
532 0 : CHK_PTR_NULL(localBuf.addr);
533 : // localAddr在transport内部校验
534 0 : return pimpl_->ReadAsync(localBuf, remoteBuf, stream);
535 : }
536 :
537 0 : HcclResult Transport::ReadSync(struct Buffer &localBuf, struct Buffer &remoteBuf, Stream &stream)
538 : {
539 0 : CHK_PTR_NULL(pimpl_);
540 0 : CHK_PTR_NULL(remoteBuf.addr);
541 0 : CHK_PTR_NULL(localBuf.addr);
542 : // localAddr在transport内部校验
543 0 : return pimpl_->ReadSync(localBuf, remoteBuf, stream);
544 : }
545 :
546 0 : HcclResult Transport::ReadReduceSync(struct Buffer &localBuf, struct Buffer &remoteBuf,
547 : const HcclDataType datatype, HcclReduceOp redOp, Stream &stream)
548 : {
549 0 : CHK_PTR_NULL(pimpl_);
550 0 : CHK_PTR_NULL(localBuf.addr);
551 0 : CHK_PTR_NULL(remoteBuf.addr);
552 0 : return pimpl_->ReadReduceSync(localBuf, remoteBuf, datatype, redOp, stream);
553 : }
554 :
555 0 : HcclResult Transport::BatchTransferAsync(const HcommBatchTransferDesc *transferDescs,
556 : uint32_t descNum, Stream &stream)
557 : {
558 0 : CHK_PTR_NULL(pimpl_);
559 0 : return pimpl_->BatchTransferAsync(transferDescs, descNum, stream);
560 : }
561 :
562 0 : HcclResult Transport::PostReady(Stream &stream)
563 : {
564 0 : CHK_PTR_NULL(pimpl_);
565 0 : return pimpl_->PostReady(stream);
566 : }
567 :
568 0 : HcclResult Transport::WaitReady(Stream &stream)
569 : {
570 0 : CHK_PTR_NULL(pimpl_);
571 0 : return pimpl_->WaitReady(stream);
572 : }
573 :
574 0 : HcclResult Transport::PostFin(Stream &stream)
575 : {
576 0 : CHK_PTR_NULL(pimpl_);
577 0 : return pimpl_->PostFin(stream);
578 : }
579 :
580 0 : HcclResult Transport::WaitFin(Stream &stream)
581 : {
582 0 : CHK_PTR_NULL(pimpl_);
583 0 : return pimpl_->WaitFin(stream);
584 : }
585 :
586 9 : HcclResult Transport::PostFinAck(Stream &stream)
587 : {
588 9 : CHK_PTR_NULL(pimpl_);
589 9 : return pimpl_->PostFinAck(stream);
590 : }
591 :
592 9 : HcclResult Transport::WaitFinAck(Stream &stream)
593 : {
594 9 : CHK_PTR_NULL(pimpl_);
595 9 : return pimpl_->WaitFinAck(stream);
596 : }
597 :
598 0 : HcclResult Transport::SetStopFlag(bool value)
599 : {
600 0 : if (pimpl_ != nullptr) {
601 0 : return pimpl_->SetStopFlag(value);
602 : }
603 0 : return HCCL_SUCCESS;
604 : }
605 :
606 0 : HcclResult Transport::UpdateRemoteAddr(void *remoteIn, void *remoteOut)
607 : {
608 0 : CHK_PTR_NULL(pimpl_);
609 0 : CHK_PTR_NULL(remoteIn);
610 0 : CHK_PTR_NULL(remoteOut);
611 0 : return pimpl_->UpdateRemoteAddr(remoteIn, remoteOut);
612 : }
613 :
614 0 : std::vector<u8> Transport::GetExchangeInfo()
615 : {
616 0 : if (UNLIKELY(pimpl_ == nullptr)) {
617 0 : return std::vector<u8>();
618 : }
619 0 : return pimpl_->GetExchangeInfo();
620 : }
621 :
622 14 : HcclResult Transport::GetTransportErrorCqe(const HcclNetDevCtx netDevCtx,
623 : std::vector<std::pair<Transport*, CqeInfo>> &infos, u32 &num)
624 : {
625 14 : CHK_PTR_NULL(netDevCtx);
626 14 : HcclIpAddress localIp;
627 14 : CHK_RET(HcclNetDevGetLocalIp(netDevCtx, localIp));
628 :
629 14 : std::vector<std::pair<TransportBase*, CqeInfo>> infolist;
630 14 : CHK_RET(TransportIbverbs::GetTransportErrorCqe(netDevCtx, infolist, num));
631 :
632 14 : std::lock_guard<std::mutex> maplock(mapMutex_);
633 14 : for (auto info : infolist) {
634 0 : auto iter = transportMap_.find(info.first);
635 0 : if (iter != transportMap_.end()) {
636 0 : infos.push_back(std::make_pair(iter->second, info.second));
637 : } else {
638 0 : HCCL_RUN_WARNING("[GetTransportErrorCqe]get err failed, transport is not find, localIp[%s], remoteIp[%s]",
639 : localIp.GetReadableAddress(), info.second.remoteIp.GetReadableAddress());
640 : }
641 0 : }
642 14 : num = infos.size();
643 :
644 14 : return HCCL_SUCCESS;
645 14 : }
646 :
647 0 : HcclResult Transport::Fence()
648 : {
649 0 : CHK_PTR_NULL(pimpl_);
650 0 : return pimpl_->Fence();
651 : }
652 :
653 0 : bool Transport::GetIsUseAtomicWrite()
654 : {
655 0 : if (pimpl_ == nullptr) {
656 0 : return false;
657 : }
658 0 : return pimpl_->GetIsUseAtomicWrite();
659 : }
660 :
661 0 : HcclResult Transport::GetSpecificNotify(HcclSignalInfo& notifyInfo, bool& isValid, const std::string& notifyName)
662 : {
663 0 : CHK_PTR_NULL(pimpl_);
664 0 : return pimpl_->GetSpecificNotify(notifyInfo, isValid, notifyName);
665 : }
666 :
667 0 : HcclResult Transport::HcclBatchRead(const TransportDeviceNormalData &ibvData, struct MemDetails *localMems,
668 : struct MemDetails *remoteMems, u32 memNum, u64 &dbInfo)
669 : {
670 : #ifdef CCL_KERNEL
671 0 : return TransportDeviceIbverbs::HnsPostSend(ibvData, localMems, remoteMems, memNum, HcclWrOpCode::HCCL_WR_RDMA_READ,
672 0 : dbInfo);
673 : #else
674 : HCCL_ERROR("[Transport][HcclBatchRead]Does not support this interface.");
675 : return HCCL_E_NOT_SUPPORT;
676 : #endif
677 : }
678 :
679 1 : HcclResult Transport::SetDeviceUnavailable(u32 deviceId)
680 : {
681 1 : return MemNameRepository::GetInstance(deviceId)->SetDeviceUnavailable(true);
682 : }
683 :
684 0 : HcclResult Transport::HcclBatchWrite(const TransportDeviceNormalData &ibvData,
685 : struct MemDetails *localMems, struct MemDetails *remoteMems, u32 memNum, u64 &dbInfo)
686 : {
687 : #ifdef CCL_KERNEL
688 0 : return TransportDeviceIbverbs::HnsPostSend(ibvData, localMems, remoteMems, memNum,
689 0 : HcclWrOpCode::HCCL_WR_RDMA_WRITE, dbInfo);
690 : #else
691 : HCCL_ERROR("[Transport][HcclBatchWrite]Does not support this interface.");
692 : return HCCL_E_NOT_SUPPORT;
693 : #endif
694 : }
695 :
696 0 : HcclResult Transport::Drain(Stream &stream)
697 : {
698 0 : CHK_PTR_NULL(pimpl_);
699 0 : return pimpl_->Drain(stream);
700 : }
701 :
702 1 : HcclResult Transport::InitDrainNotifyInfo()
703 : {
704 1 : CHK_PTR_NULL(pimpl_);
705 1 : return pimpl_->InitDrainNotifyInfo();
706 : }
707 :
708 0 : HcclResult Transport::GetDrainRemSrcMem(void* &remoteAddr, uint32_t &remoteKey, uint32_t &size)
709 : {
710 0 : CHK_PTR_NULL(pimpl_);
711 0 : return pimpl_->GetDrainRemSrcMem(remoteAddr, remoteKey, size);
712 : }
713 : }
|
