已合并
a5 pcie adapter open #1799
yangshaohua创建于 4月20日
a5 pcie adapter open #1799
已合并
yangshaohua创建于 4月20日
57 个文件变更+1057-126
@@ -16,3 +16,12 @@ src/framework/next/comms/endpoint_pairs/channels/channel_process.cc
16src/framework/next/comms/endpoint_pairs/channels/aicpu/device/aicpu_channel_process.cc16src/framework/next/comms/endpoint_pairs/channels/aicpu/device/aicpu_channel_process.cc
17src/legacy/framework/dfx/aicpu/profiling/profiling_command_handle_lite.cc17src/legacy/framework/dfx/aicpu/profiling/profiling_command_handle_lite.cc
18src/framework/next/coll_comms/api_c_adpt/dev/dev_coll_comm_c_adpt.cc18src/framework/next/coll_comms/api_c_adpt/dev/dev_coll_comm_c_adpt.cc
19+src/legacy/unified_platform/resource/transport/aicpu/p2p_transport_lite_impl.cc
20+src/legacy/unified_platform/resource/transport/aicpu/ub_transport_lite_impl.cc
21+src/legacy/unified_platform/resource/transport/p2p_transport.cc
22+src/legacy/unified_platform/resource/stream/stream.cc
23+src/legacy/unified_platform/resource/notify/ipc_remote_notify.cc
24+src/legacy/interface/rank_graph_interface.cc
25+src/framework/next/comms/endpoint_pairs/channels/aicpu/aicpu_ts_p2p_channel.cc
26+src/framework/next/comms/endpoint_pairs/channels/channel.cc
27+src/framework/next/comms/common/orion_adpt_utils.cc
@@ -544,19 +544,19 @@ int32_t HcommReadOnThread(ThreadHandle thread, ChannelHandle channel, void *dst,
544 544 
545 HcclResult ret = HCCL_SUCCESS;545 HcclResult ret = HCCL_SUCCESS;
546 if (threadPtr->IsDeviceA5()) {546 if (threadPtr->IsDeviceA5()) {
547- auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);547+ auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
548- CHK_PTR_NULL(ubTransportLitePtr);548+ CHK_PTR_NULL(transportLitePtr);
549 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());549 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
550 CHK_PTR_NULL(streamLitePtr);550 CHK_PTR_NULL(streamLitePtr);
551 551 
552 Hccl::RmaBufferLite locRmaBuf;552 Hccl::RmaBufferLite locRmaBuf;
553- ret = ubTransportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);553+ ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
554 CHK_PRT_RET(ret != HCCL_SUCCESS,554 CHK_PRT_RET(ret != HCCL_SUCCESS,
555 HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",555 HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], len[%llu].",
556 __func__, thread, channel, dst, src, len), ret);556 __func__, thread, channel, dst, src, len), ret);
557 const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};557 const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(src), len};
558 558 
559- EXECEPTION_CATCH(ubTransportLitePtr->Read(locRmaBuf, rmtBuf, *streamLitePtr), ret = HCCL_E_INTERNAL);559+ EXECEPTION_CATCH(transportLitePtr->Read(locRmaBuf, rmtBuf, *streamLitePtr), ret = HCCL_E_INTERNAL);
560 } else {560 } else {
561 HcclBuf locBuf{dst, len, nullptr};561 HcclBuf locBuf{dst, len, nullptr};
562 HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};562 HcclBuf rmtBuf{const_cast<void *>(src), len, nullptr};
@@ -590,13 +590,13 @@ int32_t HcommReadReduceOnThread(ThreadHandle thread, ChannelHandle channel, void
590 590 
591 HcclResult ret = HCCL_SUCCESS;591 HcclResult ret = HCCL_SUCCESS;
592 if (threadPtr->IsDeviceA5()) {592 if (threadPtr->IsDeviceA5()) {
593- auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);593+ auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
594- CHK_PTR_NULL(ubTransportLitePtr);594+ CHK_PTR_NULL(transportLitePtr);
595 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());595 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
596 CHK_PTR_NULL(streamLitePtr);596 CHK_PTR_NULL(streamLitePtr);
597 597 
598 Hccl::RmaBufferLite locRmaBuf;598 Hccl::RmaBufferLite locRmaBuf;
599- ret = ubTransportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);599+ ret = transportLitePtr->BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(dst), len, locRmaBuf);
600 CHK_PRT_RET(ret != HCCL_SUCCESS,600 CHK_PRT_RET(ret != HCCL_SUCCESS,
601 HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",601 HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite. thread[0x%llx], channel[0x%llx], dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d].",
602 __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);602 __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
@@ -608,7 +608,7 @@ int32_t HcommReadReduceOnThread(ThreadHandle thread, ChannelHandle channel, void
608 __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);608 __func__, thread, channel, dst, src, count, dataType, reduceOp), ret);
609 Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};609 Hccl::ReduceIn reduceIn{mapHcommDataTypeToA5.at(dataType), mapHcommReduceOpToA5.at(reduceOp)};
610 610 
611- EXECEPTION_CATCH(ubTransportLitePtr->ReadReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);611+ EXECEPTION_CATCH(transportLitePtr->ReadReduce(locRmaBuf, rmtBuf, reduceIn, *streamLitePtr), ret = HCCL_E_INTERNAL);
612 } else {612 } else {
613 CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "613 CHK_PRT_RET((IsSupportReduce(dataType, reduceOp) == false), HCCL_ERROR("[%s] Not support reduce, "
614 "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);614 "dst[0x%llx], src[0x%llx], count[%llu], dataType[%d], reduceOp[%d]", __func__, dst, src, count, dataType, reduceOp), HCCL_E_PARA);
@@ -692,13 +692,13 @@ int32_t HcommChannelNotifyRecordOnThread(ThreadHandle thread, ChannelHandle chan
692 HcclResult ret = HCCL_SUCCESS;692 HcclResult ret = HCCL_SUCCESS;
693 if (threadPtr->IsDeviceA5()) {693 if (threadPtr->IsDeviceA5()) {
694 HCCL_DEBUG("[%s] Running on A5.", __func__);694 HCCL_DEBUG("[%s] Running on A5.", __func__);
695- auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);695+ auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
696- CHK_PTR_NULL(ubTransportLitePtr);696+ CHK_PTR_NULL(transportLitePtr);
697 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());697 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
698 CHK_PTR_NULL(streamLitePtr);698 CHK_PTR_NULL(streamLitePtr);
699 HCCL_INFO("channel streamlite ptr %p.", streamLitePtr);699 HCCL_INFO("channel streamlite ptr %p.", streamLitePtr);
700 700 
701- EXECEPTION_CATCH(ubTransportLitePtr->Post(remoteNotifyIdx, *streamLitePtr), ret = HCCL_E_INTERNAL);701+ EXECEPTION_CATCH(transportLitePtr->Post(remoteNotifyIdx, *streamLitePtr), ret = HCCL_E_INTERNAL);
702 } else {702 } else {
703 Stream *stream = GetStream(thread);703 Stream *stream = GetStream(thread);
704 CHK_PTR_NULL(stream);704 CHK_PTR_NULL(stream);
@@ -728,12 +728,12 @@ int32_t HcommChannelNotifyWaitOnThread(ThreadHandle thread, ChannelHandle channe
728 HcclResult ret = HCCL_SUCCESS;728 HcclResult ret = HCCL_SUCCESS;
729 if (threadPtr->IsDeviceA5()) {729 if (threadPtr->IsDeviceA5()) {
730 HCCL_DEBUG("[%s] Running on A5.", __func__);730 HCCL_DEBUG("[%s] Running on A5.", __func__);
731- auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);731+ auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
732- CHK_PTR_NULL(ubTransportLitePtr);732+ CHK_PTR_NULL(transportLitePtr);
733 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());733 auto *const streamLitePtr = static_cast<Hccl::StreamLite *>(threadPtr->GetStreamLitePtr());
734 CHK_PTR_NULL(streamLitePtr);734 CHK_PTR_NULL(streamLitePtr);
735 735 
736- EXECEPTION_CATCH(ubTransportLitePtr->WaitWithTimeout(localNotifyIdx, *streamLitePtr, timeOut), ret = HCCL_E_INTERNAL);736+ EXECEPTION_CATCH(transportLitePtr->WaitWithTimeout(localNotifyIdx, *streamLitePtr, timeOut), ret = HCCL_E_INTERNAL);
737 } else {737 } else {
738 Stream *stream = GetStream(thread);738 Stream *stream = GetStream(thread);
739 CHK_PTR_NULL(stream);739 CHK_PTR_NULL(stream);
@@ -795,9 +795,9 @@ int32_t HcommAcquireComm(const char* commId)
795 795 
796int32_t HcommChannelRegisterDfx(ChannelHandle channel, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback) {796int32_t HcommChannelRegisterDfx(ChannelHandle channel, std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback) {
797 HCCL_INFO("[HcommChannelRegisterDfx] Init begin");797 HCCL_INFO("[HcommChannelRegisterDfx] Init begin");
798- auto *const ubTransportLitePtr = reinterpret_cast<Hccl::UbTransportLiteImpl *>(channel);798+ auto *const transportLitePtr = reinterpret_cast<Hccl::BaseTransportLiteImpl *>(channel);
799- CHK_PTR_NULL(ubTransportLitePtr);799+ CHK_PTR_NULL(transportLitePtr);
800- CHK_RET(ubTransportLitePtr->SetAddTaskInfoCallback(callback));800+ CHK_RET(transportLitePtr->SetAddTaskInfoCallback(callback));
801 HCCL_INFO("[HcommChannelRegisterDfx] Init success");801 HCCL_INFO("[HcommChannelRegisterDfx] Init success");
802 return HCCL_SUCCESS;802 return HCCL_SUCCESS;
803}803}
@@ -548,7 +548,12 @@ HcclResult RankGraphV1::GetRankGraphInfo(GraphType type, void **graph, uint32_t
548 }548 }
549 return HCCL_SUCCESS;549 return HCCL_SUCCESS;
550}550}
551- 551+ 
552+HcclResult RankGraphV1::GetDeviceId(uint32_t rankId, uint32_t *deviceId)
553+{
554+ return HCCL_SUCCESS;
555+}
556+ 
552HcclResult RankGraphV1::InitRankInfo()557HcclResult RankGraphV1::InitRankInfo()
553{558{
554 auto& rankInfoList = topoAttr_.rankInfoList;559 auto& rankInfoList = topoAttr_.rankInfoList;
@@ -43,6 +43,7 @@ public:
43 const RankInfo_t* FindRank(uint32_t rankId) const;43 const RankInfo_t* FindRank(uint32_t rankId) const;
44 HcclResult GetRankGraphInfo(GraphType type, void **graph, uint32_t *len) override;44 HcclResult GetRankGraphInfo(GraphType type, void **graph, uint32_t *len) override;
45 HcclResult GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum) override;45 HcclResult GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum) override;
46+ HcclResult GetDeviceId(uint32_t rankId, uint32_t *deviceId) override;
46 HcclResult GetInstTopoTypeByNetLayer(uint32_t netLayer, CommTopo *topoType) override;47 HcclResult GetInstTopoTypeByNetLayer(uint32_t netLayer, CommTopo *topoType) override;
47 HcclResult GetInstSizeByNetLayer(uint32_t netLayer, uint32_t *rankNum) override;48 HcclResult GetInstSizeByNetLayer(uint32_t netLayer, uint32_t *rankNum) override;
48 HcclResult GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **rankList, uint32_t *rankNum) override;49 HcclResult GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **rankList, uint32_t *rankNum) override;
@@ -44,6 +44,7 @@ public:
44 virtual HcclResult GetInstSizeByNetLayer(uint32_t netLayer, uint32_t *rankNum) = 0;44 virtual HcclResult GetInstSizeByNetLayer(uint32_t netLayer, uint32_t *rankNum) = 0;
45 virtual HcclResult GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **rankList, uint32_t *rankNum) = 0;45 virtual HcclResult GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **rankList, uint32_t *rankNum) = 0;
46 virtual HcclResult GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t **instSizeList, uint32_t *listSize) = 0;46 virtual HcclResult GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t **instSizeList, uint32_t *listSize) = 0;
47+ virtual HcclResult GetDeviceId(uint32_t rankId, uint32_t *deviceId) = 0;
47};48};
48} // namespace hccl49} // namespace hccl
49#endif50#endif
@@ -48,6 +48,11 @@ HcclResult RankGraphV2::GetRankGraphInfo(GraphType type, void **graph, uint32_t
48 return pImpl->GetRankGraphInfo(graph, len);48 return pImpl->GetRankGraphInfo(graph, len);
49}49}
50 50 
51+HcclResult RankGraphV2::GetDeviceId(uint32_t rankId, uint32_t *deviceId)
52+{
53+ return pImpl->GetDeviceId(rankId, deviceId);
54+}
55+ 
51HcclResult RankGraphV2::GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum)56HcclResult RankGraphV2::GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum)
52{57{
53 return pImpl->GetNetLayers(netLayers, netLayerNum);58 return pImpl->GetNetLayers(netLayers, netLayerNum);
@@ -35,6 +35,7 @@ public:
35 HcclResult GetEndpointNum(uint32_t netLayer, uint32_t topoInstId, uint32_t *num);35 HcclResult GetEndpointNum(uint32_t netLayer, uint32_t topoInstId, uint32_t *num);
36 HcclResult GetEndpointDesc(uint32_t netLayer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc);36 HcclResult GetEndpointDesc(uint32_t netLayer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc);
37 HcclResult GetEndpointInfo(uint32_t rankId, const EndpointDesc *endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info);37 HcclResult GetEndpointInfo(uint32_t rankId, const EndpointDesc *endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info);
38+ HcclResult GetDeviceId(uint32_t rankId, uint32_t *deviceId);
38 39 
39private:40private:
40 std::unique_ptr<Hccl::IRankGraph> pImpl;41 std::unique_ptr<Hccl::IRankGraph> pImpl;
@@ -208,13 +208,30 @@ HcclResult CollCommAicpu::ParsePackData(std::vector<char> &data, ChannelHandle &
208 std::vector<char> transpUniqueId;208 std::vector<char> transpUniqueId;
209 binaryStream >> transpUniqueId;209 binaryStream >> transpUniqueId;
210 210 
211- std::unique_ptr<Hccl::UbTransportLiteImpl> ubTransportLiteImpl;211+ Hccl::BinaryStream binaryStreamForType(transpUniqueId);
212- EXECEPTION_CATCH((ubTransportLiteImpl = std::make_unique<Hccl::UbTransportLiteImpl>(transpUniqueId)),212+ u32 transType;
213- return HCCL_E_PTR);213+ binaryStreamForType >> transType;
214- CHK_SMART_PTR_NULL(ubTransportLiteImpl);214+ HCCL_INFO("[CollCommAicpu][ParsePackData] transType[%u]", transType);
215- 215+ // TODO TransportType
216- handle = reinterpret_cast<uint64_t>(ubTransportLiteImpl.get());216+ if (transType == Hccl::TransportType::UB) {
217- ubTransportMap_.insert({handle, std::move(ubTransportLiteImpl)});217+ std::unique_ptr<Hccl::UbTransportLiteImpl> ubTransportLiteImpl;
218+ EXECEPTION_CATCH((ubTransportLiteImpl = std::make_unique<Hccl::UbTransportLiteImpl>(transpUniqueId)),
219+ return HCCL_E_PTR);
220+ CHK_SMART_PTR_NULL(ubTransportLiteImpl);
221+ handle = reinterpret_cast<uint64_t>(ubTransportLiteImpl.get());
222+ ubTransportMap_.insert({handle, std::move(ubTransportLiteImpl)});
223+ } else if (transType == Hccl::TransportType::P2P) {
224+ std::unique_ptr<Hccl::P2PTransportLiteImpl> p2pTransportLiteImpl;
225+ EXECEPTION_CATCH((p2pTransportLiteImpl = std::make_unique<Hccl::P2PTransportLiteImpl>(transpUniqueId)),
226+ return HCCL_E_PTR);
227+ CHK_SMART_PTR_NULL(p2pTransportLiteImpl);
228+ handle = reinterpret_cast<uint64_t>(p2pTransportLiteImpl.get());
229+ p2pTransportMap_.insert({handle, std::move(p2pTransportLiteImpl)});
230+ // TODO 是否需要缓存用于NsRecovery
231+ } else {
232+ HCCL_ERROR("[CollCommAicpu][ParsePackData] unsupported transportType[%u]", transType);
233+ return HCCL_E_INTERNAL;
234+ }
218 235 
219 return HCCL_SUCCESS;236 return HCCL_SUCCESS;
220}237}
@@ -20,6 +20,7 @@
20#include "thread.h"20#include "thread.h"
21#include "local_notify.h"21#include "local_notify.h"
22#include "ub_transport_lite_impl.h"22#include "ub_transport_lite_impl.h"
23+#include "p2p_transport_lite_impl.h"
23#include "task_exception.h"24#include "task_exception.h"
24#include "aicpu_launch_manager.h"25#include "aicpu_launch_manager.h"
25#include "channel_param.h"26#include "channel_param.h"
@@ -85,6 +86,7 @@ private:
85 std::vector<std::unique_ptr<LocalNotify>> notifys_;86 std::vector<std::unique_ptr<LocalNotify>> notifys_;
86 // A5 独立算子87 // A5 独立算子
87 std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::UbTransportLiteImpl>> ubTransportMap_;88 std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::UbTransportLiteImpl>> ubTransportMap_;
89+ std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::P2PTransportLiteImpl>> p2pTransportMap_;
88 90 
89 // N秒快恢相关91 // N秒快恢相关
90 hccl::NsRecoveryLitePtr nsRecoveryLitePtr_{nullptr};92 hccl::NsRecoveryLitePtr nsRecoveryLitePtr_{nullptr};
@@ -26,7 +26,6 @@ HcclResult HcclAicpuHdcHandler::GetKfcCommand(Hccl::KfcCommand &cmd)
26 HCCL_ERROR("[HcclAicpuHdcHandler][GetKfcCommand] h2dTransfer Get fail, ret[%d]", ret);26 HCCL_ERROR("[HcclAicpuHdcHandler][GetKfcCommand] h2dTransfer Get fail, ret[%d]", ret);
27 return ret;27 return ret;
28 }28 }
29- 
30 if (lastCmd_ != cmd) {29 if (lastCmd_ != cmd) {
31 HCCL_INFO("[HcclAicpuHdcHandler][GetKfcCommand] Get new KfcCommand[%u], last KfcCommand[%u]", cmd, lastCmd_);30 HCCL_INFO("[HcclAicpuHdcHandler][GetKfcCommand] Get new KfcCommand[%u], last KfcCommand[%u]", cmd, lastCmd_);
32 lastCmd_ = cmd;31 lastCmd_ = cmd;
@@ -182,8 +182,15 @@ HcclResult MyRank::BatchCreateSockets(const HcclChannelDesc* channelDescs, uint3
182 }182 }
183 u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];183 u32& reuseIdx = reuseSocketIdxMap[rankPair][endpointPair];
184 184 
185+ uint32_t devicePhyId;
186+ uint32_t remoteDevicePhyId;
187+ rankGraph_->GetDeviceId(rankId_, &devicePhyId);
188+ rankGraph_->GetDeviceId(remoteRank, &remoteDevicePhyId);
189+ HCCL_INFO("[MyRank][BatchCreateSockets] rankId_[%u] devicePhyId[%u]", rankId_, devicePhyId);
190+ HCCL_INFO("[MyRank][BatchCreateSockets] rankId_[%u] devicePhyId[%u]", remoteRank, remoteDevicePhyId);
191+ 
185 Hccl::Socket* socket = nullptr;192 Hccl::Socket* socket = nullptr;
186- auto ret = endpointPair->GetSocket(rankId_, remoteRank, commTag, reuseIdx, listenPort, socket);193+ auto ret = endpointPair->GetSocket(rankId_, remoteRank, commTag, reuseIdx, listenPort, socket, devicePhyId, remoteDevicePhyId);
187 CHK_PRT_RET(ret != HCCL_SUCCESS,194 CHK_PRT_RET(ret != HCCL_SUCCESS,
188 HCCL_ERROR("[%s] failed to get socket, channelIndex[%u], remoteRank[%u], protocol[%d] reuseIdx[%u]",195 HCCL_ERROR("[%s] failed to get socket, channelIndex[%u], remoteRank[%u], protocol[%d] reuseIdx[%u]",
189 __func__, i, remoteRank, localEndpointDesc.protocol, reuseIdx),196 __func__, i, remoteRank, localEndpointDesc.protocol, reuseIdx),
@@ -430,8 +437,8 @@ HcclResult MyRank::CreateChannels(CommEngine engine, const std::string &commTag,
430 HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);437 HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
431 callbacks_.setAicpuCommState(true);438 callbacks_.setAicpuCommState(true);
432 }439 }
433- 440+ HcommChannelDesc* hcommDesc = hcommDescs.data();
434- CHK_RET(ChannelProcess::ChannelKernelLaunchForComm(channelHandles, hostChannelHandleList, 441+ CHK_RET(ChannelProcess::ChannelKernelLaunchForComm(channelHandles, hostChannelHandleList, hcommDesc,
435 channelNum, commTag, binHandle_));442 channelNum, commTag, binHandle_));
436 443 
437 // ns recovery444 // ns recovery
@@ -355,7 +355,7 @@ HcommResult HcommChannelCreate(EndpointHandle endpointHandle, CommEngine engine,
355 targetChannels));355 targetChannels));
356 CHK_RET(ChannelProcess::ConnectChannels(targetChannels, channelNum, engine));356 CHK_RET(ChannelProcess::ConnectChannels(targetChannels, channelNum, engine));
357 CHK_RET(EnsureKernelBinLoaded(engine));357 CHK_RET(EnsureKernelBinLoaded(engine));
358- CHK_RET(ChannelProcess::SaveChannels(targetChannels, channels, channelNum, engine, g_BinHandle));358+ CHK_RET(ChannelProcess::SaveChannels(targetChannels, channels, channelDescs, channelNum, engine, g_BinHandle));
359 359 
360 return HCCL_SUCCESS;360 return HCCL_SUCCESS;
361}361}
@@ -79,6 +79,9 @@ HcclResult CommProtocolToLinkProtocol(CommProtocol commProtocol, Hccl::LinkProto
79 case COMM_PROTOCOL_UB_MEM:79 case COMM_PROTOCOL_UB_MEM:
80 linkProtocol = Hccl::LinkProtocol::UB_MEM;80 linkProtocol = Hccl::LinkProtocol::UB_MEM;
81 break;81 break;
82+ case COMM_PROTOCOL_PCIE:
83+ linkProtocol = Hccl::LinkProtocol::PCIE;
84+ break;
82 default:85 default:
83 HCCL_ERROR("[%s] Invaild CommProtocol[%u]", __func__, commProtocol);86 HCCL_ERROR("[%s] Invaild CommProtocol[%u]", __func__, commProtocol);
84 return HCCL_E_PARA;87 return HCCL_E_PARA;
@@ -150,7 +153,8 @@ HcclResult EndpointDescPairToLinkData(const EndpointDesc &locEp, const EndpointD
150}153}
151 154 
152HcclResult EndpointDescPairToLinkDataWithRankIds(const uint32_t myRank, const uint32_t rmtRank,155HcclResult EndpointDescPairToLinkDataWithRankIds(const uint32_t myRank, const uint32_t rmtRank,
153- const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData, u32 reuseIdx)156+ const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData, uint32_t devicePhyId, uint32_t remoteDevicePhyId,
157+ u32 reuseIdx)
154{158{
155 Hccl::PortDeploymentType portDeploymentType = Hccl::PortDeploymentType::INVALID;159 Hccl::PortDeploymentType portDeploymentType = Hccl::PortDeploymentType::INVALID;
156 CHK_RET(EndpointLocTypeToPortDeploymentType(locEp.loc.locType, portDeploymentType));160 CHK_RET(EndpointLocTypeToPortDeploymentType(locEp.loc.locType, portDeploymentType));
@@ -168,8 +172,9 @@ HcclResult EndpointDescPairToLinkDataWithRankIds(const uint32_t myRank, const ui
168 portDeploymentType,172 portDeploymentType,
169 linkProtocol, 173 linkProtocol,
170 myRank, rmtRank,174 myRank, rmtRank,
171- locAddr, rmtAddr, reuseIdx175+ locAddr, rmtAddr, devicePhyId, remoteDevicePhyId, reuseIdx
172 );176 );
177+ linkData.UpdateIpAddrWithPCIE();
173 178 
174 return HCCL_SUCCESS;179 return HCCL_SUCCESS;
175}180}
@@ -26,7 +26,8 @@ Hccl::LinkData BuildDefaultLinkData();
26HcclResult EndpointDescPairToLinkData(const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData,26HcclResult EndpointDescPairToLinkData(const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData,
27 u32 reuseIdx = 0);27 u32 reuseIdx = 0);
28HcclResult EndpointDescPairToLinkDataWithRankIds(const uint32_t myRank, const uint32_t rmtRank,28HcclResult EndpointDescPairToLinkDataWithRankIds(const uint32_t myRank, const uint32_t rmtRank,
29- const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData, u32 reuseIdx = 0);29+ const EndpointDesc &locEp, const EndpointDesc &rmtEp, Hccl::LinkData &linkData, uint32_t devicePhyId, uint32_t remoteDevicePhyId,
30+ u32 reuseIdx = 0);
30 31 
31} // namespace hcomm32} // namespace hcomm
32 33 
@@ -10,6 +10,7 @@
10 10 
11set(src_list11set(src_list
12 aicpu_ts_urma_channel.cc12 aicpu_ts_urma_channel.cc
13+ aicpu_ts_p2p_channel.cc
13)14)
14 15 
15target_sources(hcomm PRIVATE16target_sources(hcomm PRIVATE
@@ -0,0 +1,306 @@
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 "aicpu_ts_p2p_channel.h"
12+#include "../../../endpoints/endpoint.h"
13+#include "orion_adpt_utils.h"
14+#include "hcomm_c_adpt.h"
15+#include "exception_handler.h"
16+#include "comm_mems.h"
17+ 
18+#include "coll_alg_param.h"
19+#include "topo_common_types.h"
20+#include "virtual_topo.h"
21+#include "p2p_connection.h"
22+ 
23+namespace hcomm {
24+ 
25+AicpuTsP2pChannel::AicpuTsP2pChannel(EndpointHandle endpointHandle, const HcommChannelDesc &channelDesc):
26+ endpointHandle_(endpointHandle), channelDesc_(channelDesc) {}
27+ 
28+HcclResult AicpuTsP2pChannel::Makebufs(HcommMemHandle *memHandles, uint32_t memHandleNum,
29+ std::vector<std::shared_ptr<Hccl::Buffer>> &bufs)
30+{
31+ bufs.clear();
32+ for (uint32_t i = 0; i < memHandleNum; ++i) {
33+ auto locMemInfo = reinterpret_cast<CommMemInfo *>(memHandles[i]);
34+ HCCL_INFO("[AicpuTsP2pChannel][%s] tag[%s]", __func__, locMemInfo->memTag);
35+ bufs.emplace_back(std::move(std::make_shared<Hccl::Buffer>(
36+ reinterpret_cast<uintptr_t>(locMemInfo->mem.addr), locMemInfo->mem.size,
37+ hccl::ConvertCommToHcclMemType(locMemInfo->mem.type), locMemInfo->memTag)
38+ ));
39+ }
40+ return HCCL_SUCCESS;
41+}
42+ 
43+HcclResult AicpuTsP2pChannel::ParseInputParam()
44+{
45+ Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
46+ CHK_PTR_NULL(localEpPtr);
47+ localEp_ = localEpPtr->GetEndpointDesc();
48+ 
49+ HCCL_INFO("[AicpuTsP2pChannel][%s] localProtocol[%d]", __func__, localEp_.protocol);
50+ 
51+ remoteEp_ = channelDesc_.remoteEndpoint;
52+ socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
53+ notifyNum_ = channelDesc_.notifyNum;
54+ 
55+ if (channelDesc_.exchangeAllMems) {
56+ HCCL_INFO("[AicpuTsP2pChannel][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
57+ std::shared_ptr<Hccl::LocalIpcRmaBuffer> *memHandles = nullptr;
58+ uint32_t memHandleNum = 0;
59+ CHK_RET(static_cast<HcclResult>(HcommMemGetAllMemHandles(
60+ endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
61+ HCCL_INFO("[AicpuTsP2pChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
62+ for (uint32_t i = 0; i < memHandleNum; ++i) {
63+ std::shared_ptr<Hccl::LocalIpcRmaBuffer> &localIpcRmaBuffer = memHandles[i];
64+ HCCL_INFO("[AicpuTsP2pChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memTag[%s].",
65+ __func__, i, localIpcRmaBuffer->GetBufferInfo().first,
66+ localIpcRmaBuffer->GetBufferInfo().second,
67+ localIpcRmaBuffer->GetBuf()->GetMemTag().c_str());
68+ bufs_.emplace_back(std::move(std::make_shared<Hccl::Buffer>(
69+ localIpcRmaBuffer->GetBufferInfo().first,
70+ localIpcRmaBuffer->GetBufferInfo().second,
71+ localIpcRmaBuffer->GetBuf()->GetMemTag().c_str())
72+ ));
73+ }
74+ } else {
75+ HCCL_INFO("[AicpuTsP2pChannel][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
76+ CHK_RET(Makebufs(channelDesc_.memHandles, channelDesc_.memHandleNum, bufs_));
77+ }
78+ 
79+ EXECEPTION_CATCH(socketMgr_ = std::make_unique<SocketMgr>(), return HCCL_E_PTR);
80+ 
81+ return HCCL_SUCCESS;
82+}
83+ 
84+HcclResult AicpuTsP2pChannel::BuildAttr()
85+{
86+ attr_.devicePhyId = localEp_.loc.device.devPhyId;
87+ attr_.opMode = Hccl::OpMode::OPBASE;
88+ return HCCL_SUCCESS;
89+}
90+ 
91+HcclResult AicpuTsP2pChannel::BuildConnection()
92+{
93+ std::unique_ptr<Hccl::P2PConnection> p2pConn = nullptr;
94+ std::string connTag = "P2P_CHANNEL_" + std::to_string(localEp_.loc.device.devPhyId);
95+
96+ EXECEPTION_CATCH(
97+ p2pConn = std::make_unique<Hccl::P2PConnection>(socket_, connTag),
98+ return HCCL_E_PTR
99+ );
100+ CHK_SMART_PTR_NULL(p2pConn);
101+ 
102+ commonRes_.connVec.clear();
103+ commonRes_.connVec.emplace_back(p2pConn.get());
104+ connections_.clear();
105+ connections_.push_back(std::move(p2pConn));
106+ 
107+ return HCCL_SUCCESS;
108+}
109+ 
110+HcclResult AicpuTsP2pChannel::BuildNotify()
111+{
112+ localNotifies_.clear();
113+ commonRes_.notifyVec.clear();
114+ bool devUsed = true;
115+ for (uint32_t i = 0; i < notifyNum_; ++i) {
116+ std::unique_ptr<Hccl::IpcLocalNotify> notifyPtr = nullptr;
117+ EXECEPTION_CATCH(
118+ notifyPtr = std::make_unique<Hccl::IpcLocalNotify>(devUsed),
119+ return HCCL_E_PTR
120+ );
121+ commonRes_.notifyVec.push_back(notifyPtr.get());
122+ localNotifies_.push_back(std::move(notifyPtr));
123+ }
124+ return HCCL_SUCCESS;
125+}
126+ 
127+HcclResult AicpuTsP2pChannel::BuildBuffer(std::vector<std::shared_ptr<Hccl::Buffer>> &bufs)
128+{
129+ bufferVecTemp_.clear();
130+ for (size_t i = 0; i < bufs.size(); i++) {
131+ std::unique_ptr<Hccl::LocalIpcRmaBuffer> bufferPtr = nullptr;
132+ EXECEPTION_CATCH(
133+ bufferPtr = std::make_unique<Hccl::LocalIpcRmaBuffer>(bufs[i]),
134+ return HCCL_E_PTR
135+ );
136+ bufferVecTemp_.push_back(bufferPtr.get());
137+ commonRes_.bufferVec.push_back(bufferPtr.get());
138+ localRmaBuffers_.push_back(std::move(bufferPtr));
139+ }
140+ return HCCL_SUCCESS;
141+}
142+ 
143+HcclResult AicpuTsP2pChannel::BuildP2pMemTransport()
144+{
145+ const Hccl::Socket &socket = *socket_;
146+ 
147+ Hccl::LinkData linkData = BuildDefaultLinkData();
148+ CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
149+ 
150+ EXECEPTION_CATCH(
151+ memTransport_ = std::make_unique<Hccl::P2PTransport>(
152+ commonRes_, attr_, linkData, socket
153+ ),
154+ return HCCL_E_PTR
155+ );
156+ return HCCL_SUCCESS;
157+}
158+ 
159+HcclResult AicpuTsP2pChannel::BuildSocket()
160+{
161+ if (socket_ != nullptr) {
162+ return HCCL_SUCCESS;
163+ }
164+ HCCL_INFO("[AicpuTsP2pChannel][%s] socket ptr is NULL, rebuildSocket", __func__);
165+ 
166+ Hccl::IpAddress ipaddr{};
167+ CHK_RET(CommAddrToIpAddress(localEp_.commAddr, ipaddr));
168+ Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::PCIE); // TODO PROTOTYPE P2P?
169+ Hccl::PortData localPort = Hccl::PortData(static_cast<Hccl::RankId>(localEp_.loc.device.devPhyId), type, 0, ipaddr);
170+ Hccl::SocketHandle socketHandle = Hccl::SocketHandleManager::GetInstance().Create(localEp_.loc.device.devPhyId, localPort);
171+ EXECEPTION_CATCH(serverSocket_ = std::make_unique<Hccl::Socket>(socketHandle, ipaddr, 60001,
172+ ipaddr, "server", Hccl::SocketRole::SERVER, Hccl::NicType::DEVICE_NIC_TYPE), return HCCL_E_PARA);
173+ HCCL_INFO("[AicpuTsP2pChannel][%s] listen_socket_info[%s]", __func__, serverSocket_->Describe().c_str());
174+ EXECEPTION_CATCH(serverSocket_->Listen(), return HCCL_E_INTERNAL);
175+ 
176+ Hccl::LinkData linkData = BuildDefaultLinkData();
177+ CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
178+ HCCL_INFO("[AicpuTsP2pChannel][%s] built linkData: %s", __func__, linkData.Describe().c_str());
179+ std::string socketTag = "AUTOMATIC_SOCKET_TAG";
180+ bool noRankId = true;
181+ Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, socketTag, noRankId);
182+ CHK_RET(socketMgr_->GetSocket(socketConfig, socket_));
183+ 
184+ return HCCL_SUCCESS;
185+}
186+ 
187+HcclResult AicpuTsP2pChannel::Init()
188+{
189+ CHK_RET(ParseInputParam());
190+ CHK_RET(BuildSocket());
191+ CHK_RET(BuildAttr());
192+ CHK_RET(BuildConnection());
193+ CHK_RET(BuildNotify());
194+ localRmaBuffers_.clear();
195+ commonRes_.bufferVec.clear();
196+ CHK_RET(BuildBuffer(bufs_));
197+ CHK_RET(BuildP2pMemTransport());
198+ return HCCL_SUCCESS;
199+}
200+ 
201+HcclResult AicpuTsP2pChannel::GetNotifyNum(uint32_t *notifyNum) const
202+{
203+ *notifyNum = this->notifyNum_;
204+ return HCCL_SUCCESS;
205+}
206+ 
207+HcclResult AicpuTsP2pChannel::GetRemoteMem(HcclMem **remoteMem, uint32_t *memNum, char **memTags)
208+{
209+ return memTransport_->GetRemoteMem(remoteMem, memNum, memTags);
210+ // HCCL_WARNING("[AicpuTsP2pChannel][%s] P2PTransport does not support GetRemoteMem.", __func__);
211+ // *memNum = 0;
212+ // return HCCL_SUCCESS;
213+}
214+ 
215+ChannelStatus AicpuTsP2pChannel::GetStatus()
216+{
217+ Hccl::TransportStatus transportStatus = memTransport_->GetStatus();
218+ ChannelStatus out = ChannelStatus::INIT;
219+ switch (transportStatus) {
220+ case Hccl::TransportStatus::INIT:
221+ out = ChannelStatus::INIT;
222+ break;
223+ case Hccl::TransportStatus::SOCKET_OK:
224+ out = ChannelStatus::SOCKET_OK;
225+ break;
226+ case Hccl::TransportStatus::SOCKET_TIMEOUT:
227+ out = ChannelStatus::SOCKET_TIMEOUT;
228+ break;
229+ case Hccl::TransportStatus::READY:
230+ out = ChannelStatus::READY;
231+ break;
232+ default:
233+ HCCL_ERROR("[AicpuTsP2pChannel][%s] Invalid TransportStatus[%d]", __func__, transportStatus);
234+ out = ChannelStatus::FAILED;
235+ break;
236+ }
237+ return out;
238+}
239+ 
240+HcclResult AicpuTsP2pChannel::SetModuleDataName(Hccl::ModuleData &module, const std::string &name)
241+{
242+ int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
243+ if (ret != 0) {
244+ HCCL_ERROR("[SetModuleDataName] strcpy_s name %s failed", name.c_str());
245+ return HCCL_E_INTERNAL;
246+ }
247+ 
248+ return HCCL_SUCCESS;
249+}
250+ 
251+HcclResult AicpuTsP2pChannel::PackOpData(std::vector<char> &data)
252+{
253+ std::vector<Hccl::ModuleData> dataVec;
254+ dataVec.resize(Hccl::AicpuResMgrType::__COUNT__);
255+ 
256+ Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
257+ CHK_RET(SetModuleDataName(dataVec[resType], "P2PTransport"));
258+ 
259+ std::vector<char> result;
260+ Hccl::BinaryStream binaryStream;
261+ binaryStream << memTransport_->GetUniqueIdV2();
262+ 
263+ binaryStream.Dump(result);
264+ 
265+ dataVec[resType].data = result;
266+ 
267+ Hccl::AicpuResPackageHelper helper;
268+ data = helper.GetPackedData(dataVec);
269+ 
270+ return HCCL_SUCCESS;
271+}
272+ 
273+HcclResult AicpuTsP2pChannel::H2DResPack(std::vector<char>& buffer)
274+{
275+ CHK_RET(PackOpData(buffer));
276+ HCCL_INFO("[AicpuTsP2pChannel][%s] Pack Buffer data[%p], Pack Buffer size[%zu].",
277+ __func__, buffer.data(), buffer.size());
278+ return HCCL_SUCCESS;
279+}
280+ 
281+HcclResult AicpuTsP2pChannel::Clean()
282+{
283+ memTransport_.reset();
284+ return HCCL_SUCCESS;
285+}
286+ 
287+HcclResult AicpuTsP2pChannel::Resume()
288+{
289+ BuildConnection();
290+ BuildP2pMemTransport();
291+ return HCCL_SUCCESS;
292+}
293+ 
294+HcclResult AicpuTsP2pChannel::GetUserRemoteMem(CommMem **remoteMem, char ***memTag, uint32_t *memNum)
295+{
296+ HCCL_WARNING("[AicpuTsP2pChannel][%s] P2PTransport does not support GetUserRemoteMem.", __func__);
297+ *memNum = 0;
298+ return HCCL_SUCCESS;
299+}
300+ 
301+HcclResult AicpuTsP2pChannel::UpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum)
302+{
303+ HCCL_WARNING("[AicpuTsP2pChannel][%s] P2PTransport does not support UpdateMemInfo.", __func__);
304+ return HCCL_SUCCESS;
305+}
306+} // namespace hcomm
@@ -0,0 +1,82 @@
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+#ifndef AICPU_TS_P2P_CHANNEL_H
11+#define AICPU_TS_P2P_CHANNEL_H
12+ 
13+#include "../channel.h"
14+#include "../../sockets/socket_mgr.h"
15+ 
16+#include "../../../../../../legacy/unified_platform/resource/socket/socket.h"
17+#include "../../../../../../legacy/framework/resource_manager/socket/socket_manager.h"
18+#include "../../../../../../legacy/unified_platform/pub_inc/buffer_key.h"
19+#include "../../../../../../legacy/unified_platform/resource/buffer/local_ipc_rma_buffer.h"
20+#include "rma_connection.h"
21+#include "p2p_transport.h"
22+#include "p2p_connection.h"
23+#include "ipc_local_notify.h"
24+#include "aicpu_res_package_helper.h"
25+ 
26+namespace hcomm {
27+ 
28+class AicpuTsP2pChannel : public Channel {
29+public:
30+ AicpuTsP2pChannel(EndpointHandle endpointHandle, const HcommChannelDesc &channelDesc);
31+ 
32+ HcclResult Init() override;
33+ HcclResult GetNotifyNum(uint32_t *notifyNum) const override;
34+ HcclResult GetRemoteMem(HcclMem **remoteMem, uint32_t *memNum, char **memTags) override;
35+ ChannelStatus GetStatus() override;
36+ HcclResult GetUserRemoteMem(CommMem **remoteMem, char ***memTag, uint32_t *memNum) override;
37+ HcclResult UpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum) override;
38+ 
39+ HcclResult H2DResPack(std::vector<char>& buffer);
40+ 
41+ virtual HcclResult Clean() override;
42+ virtual HcclResult Resume() override;
43+ 
44+private:
45+ HcclResult Makebufs(HcommMemHandle *memHandles, uint32_t memHandleNum,
46+ std::vector<std::shared_ptr<Hccl::Buffer>> &bufs);
47+ HcclResult SetModuleDataName(Hccl::ModuleData &module, const std::string &name);
48+ HcclResult ParseInputParam();
49+ HcclResult BuildAttr();
50+ HcclResult BuildConnection();
51+ HcclResult BuildNotify();
52+ HcclResult BuildBuffer(std::vector<std::shared_ptr<Hccl::Buffer>> &bufs);
53+ HcclResult BuildP2pMemTransport();
54+ HcclResult BuildSocket();
55+ 
56+ HcclResult PackOpData(std::vector<char> &data);
57+ 
58+private:
59+ EndpointHandle endpointHandle_;
60+ HcommChannelDesc channelDesc_;
61+ 
62+ EndpointDesc localEp_{};
63+ EndpointDesc remoteEp_{};
64+ uint32_t notifyNum_{0};
65+ std::vector<std::shared_ptr<Hccl::Buffer>> bufs_{};
66+ std::vector<std::shared_ptr<Hccl::Buffer>> bufsTemp{};
67+ 
68+ Hccl::Socket* socket_{nullptr};
69+ std::unique_ptr<Hccl::P2PTransport> memTransport_{nullptr};
70+ Hccl::BaseMemTransport::Attribution attr_{};
71+ Hccl::BaseMemTransport::CommonLocRes commonRes_{};
72+ std::vector<Hccl::LocalRmaBuffer *> bufferVecTemp_;
73+ std::vector<std::unique_ptr<Hccl::P2PConnection>> connections_{};
74+ std::vector<std::unique_ptr<Hccl::LocalIpcRmaBuffer>> localRmaBuffers_{};
75+ std::vector<std::unique_ptr<Hccl::IpcLocalNotify>> localNotifies_{};
76+ std::unique_ptr<Hccl::Socket> serverSocket_;
77+ std::unique_ptr<SocketMgr> socketMgr_{nullptr};
78+};
79+ 
80+} // namespace hcomm
81+ 
82+#endif // AICPU_TS_P2P_CHANNEL_H
@@ -13,6 +13,7 @@
13#include "log.h"13#include "log.h"
14#include "channel.h"14#include "channel.h"
15#include "./aicpu/aicpu_ts_urma_channel.h"15#include "./aicpu/aicpu_ts_urma_channel.h"
16+#include "./aicpu/aicpu_ts_p2p_channel.h"
16#include "./host/host_cpu_roce_channel.h"17#include "./host/host_cpu_roce_channel.h"
17#include "./ccu/ccu_urma_channel.h"18#include "./ccu/ccu_urma_channel.h"
18#include "./aiv/aiv_ub_mem_channel.h"19#include "./aiv/aiv_ub_mem_channel.h"
@@ -42,9 +43,15 @@ HcclResult Channel::CreateChannel(
42 return HCCL_E_NOT_SUPPORT;43 return HCCL_E_NOT_SUPPORT;
43 case COMM_ENGINE_AICPU:44 case COMM_ENGINE_AICPU:
44 case COMM_ENGINE_AICPU_TS:45 case COMM_ENGINE_AICPU_TS:
45- channelPtr.reset(new (std::nothrow) AicpuTsUrmaChannel(46+ if (channelDesc.remoteEndpoint.protocol == COMM_PROTOCOL_PCIE) {
46- endpointHandle, channelDesc47+ channelPtr.reset(new (std::nothrow) AicpuTsP2pChannel(
47- ));48+ endpointHandle, channelDesc
49+ ));
50+ } else if (channelDesc.remoteEndpoint.protocol != COMM_PROTOCOL_RESERVED) { // TODO == P2P
51+ channelPtr.reset(new (std::nothrow) AicpuTsUrmaChannel(
52+ endpointHandle, channelDesc
53+ ));
54+ }
48 break; 55 break;
49 case COMM_ENGINE_AIV:56 case COMM_ENGINE_AIV:
50 channelPtr.reset(57 channelPtr.reset(
@@ -15,6 +15,7 @@
15#include "launch_aicpu.h"15#include "launch_aicpu.h"
16#include "hcclCommDfx.h"16#include "hcclCommDfx.h"
17#include "env_config/env_config.h"17#include "env_config/env_config.h"
18+#include "aicpu_ts_p2p_channel.h"
18 19 
19namespace hcomm {20namespace hcomm {
20 21 
@@ -300,7 +301,7 @@ static HcclResult LaunchKernel(const HcclChannelUrmaRes &channelParam,
300 return HCCL_SUCCESS;301 return HCCL_SUCCESS;
301}302}
302 303 
303-HcclResult ChannelProcess::LaunchChannelKernelCommon(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,304+HcclResult ChannelProcess::LaunchChannelKernelCommon(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles, HcommChannelDesc* hcommDesc,
304 uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle, const std::string &kernelName, bool needProfiling)305 uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle, const std::string &kernelName, bool needProfiling)
305{306{
306 CHK_PTR_NULL(channelHandles);307 CHK_PTR_NULL(channelHandles);
@@ -316,8 +317,13 @@ HcclResult ChannelProcess::LaunchChannelKernelCommon(ChannelHandle *channelHandl
316 std::vector<u32> channelSizeVec{};317 std::vector<u32> channelSizeVec{};
317 uint32_t totalListNum = 0;318 uint32_t totalListNum = 0;
318 for (uint32_t index = 0; index < listNum; index++) {319 for (uint32_t index = 0; index < listNum; index++) {
319- auto aicpuTsUrmaChannel = reinterpret_cast<AicpuTsUrmaChannel *>(hostChannelHandles[index]);320+ if (hcommDesc[index].remoteEndpoint.protocol == CommProtocol::COMM_PROTOCOL_PCIE) {
320- CHK_PRT(aicpuTsUrmaChannel->H2DResPack(hostPackBuffers[index]));321+ auto aicpuTsP2pChannel = reinterpret_cast<AicpuTsP2pChannel *>(hostChannelHandles[index]);
322+ CHK_PRT(aicpuTsP2pChannel->H2DResPack(hostPackBuffers[index]));
323+ } else {
324+ auto aicpuTsUrmaChannel = reinterpret_cast<AicpuTsUrmaChannel *>(hostChannelHandles[index]);
325+ CHK_PRT(aicpuTsUrmaChannel->H2DResPack(hostPackBuffers[index]));
326+ }
321 totalListNum += hostPackBuffers[index].size();327 totalListNum += hostPackBuffers[index].size();
322 channelSizeVec.push_back(hostPackBuffers[index].size());328 channelSizeVec.push_back(hostPackBuffers[index].size());
323 }329 }
@@ -385,28 +391,28 @@ HcclResult ChannelProcess::LaunchChannelKernelCommon(ChannelHandle *channelHandl
385}391}
386 392 
387HcclResult ChannelProcess::ChannelKernelLaunchForComm(ChannelHandle *channelHandles, 393HcclResult ChannelProcess::ChannelKernelLaunchForComm(ChannelHandle *channelHandles,
388- ChannelHandle *hostChannelHandles, uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle) 394+ ChannelHandle *hostChannelHandles, HcommChannelDesc* hcommDesc, uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle)
389{395{
390- return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, listNum, 396+ return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, hcommDesc, listNum,
391 commTag, binHandle, "RunAicpuIndOpChannelInitV2", true);397 commTag, binHandle, "RunAicpuIndOpChannelInitV2", true);
392}398}
393 399 
394HcclResult ChannelProcess::ChannelKernelLaunchForBase(ChannelHandle *channelHandles, 400HcclResult ChannelProcess::ChannelKernelLaunchForBase(ChannelHandle *channelHandles,
395- ChannelHandle *hostChannelHandles, uint32_t listNum, aclrtBinHandle binHandle) 401+ ChannelHandle *hostChannelHandles, HcommChannelDesc* hcommDesc, uint32_t listNum, aclrtBinHandle binHandle)
396{402{
397- return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, listNum, "", 403+ return LaunchChannelKernelCommon(channelHandles, hostChannelHandles, hcommDesc, listNum, "",
398 binHandle, "RunAicpuChannelInitV2", false);404 binHandle, "RunAicpuChannelInitV2", false);
399}405}
400 406 
401HcclResult ChannelProcess::SaveChannels(ChannelHandle* targetChannels, ChannelHandle* userChannels,407HcclResult ChannelProcess::SaveChannels(ChannelHandle* targetChannels, ChannelHandle* userChannels,
402- uint32_t channelNum, CommEngine engine, aclrtBinHandle binHandle)408+ HcommChannelDesc *channelDescs, uint32_t channelNum, CommEngine engine, aclrtBinHandle binHandle)
403{409{
404 CHK_PTR_NULL(targetChannels);410 CHK_PTR_NULL(targetChannels);
405 CHK_PTR_NULL(userChannels);411 CHK_PTR_NULL(userChannels);
406 CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);412 CHK_PRT_RET((channelNum == 0), HCCL_ERROR("[%s]Invalid channelNum, channelNum[%u]", __func__, channelNum), HCCL_E_PARA);
407 413 
408 if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {414 if (engine == COMM_ENGINE_AICPU || engine == COMM_ENGINE_AICPU_TS) {
409- CHK_RET(ChannelKernelLaunchForBase(userChannels, targetChannels, channelNum, binHandle));415+ CHK_RET(ChannelKernelLaunchForBase(userChannels, targetChannels, channelDescs, channelNum, binHandle));
410 } else {416 } else {
411 HCCL_INFO("[%s] engine[%d] no need to KernelLaunch.", __func__, engine);417 HCCL_INFO("[%s] engine[%d] no need to KernelLaunch.", __func__, engine);
412 for (uint32_t i = 0; i < channelNum; i++) {418 for (uint32_t i = 0; i < channelNum; i++) {
@@ -27,10 +27,10 @@ public:
27 static HcclResult ChannelUpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum, ChannelHandle channelHandle);27 static HcclResult ChannelUpdateMemInfo(HcommMemHandle *memHandles, uint32_t memHandleNum, ChannelHandle channelHandle);
28 static HcclResult ConnectChannels(ChannelHandle* targetChannels, uint32_t channelNum, CommEngine engine);28 static HcclResult ConnectChannels(ChannelHandle* targetChannels, uint32_t channelNum, CommEngine engine);
29 static HcclResult SaveChannels(ChannelHandle* targetChannels, ChannelHandle* userChannels, 29 static HcclResult SaveChannels(ChannelHandle* targetChannels, ChannelHandle* userChannels,
30- uint32_t channelNum, CommEngine engine, aclrtBinHandle binHandle);30+ HcommChannelDesc *channelDescs, uint32_t channelNum, CommEngine engine, aclrtBinHandle binHandle);
31 static HcclResult ChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t *statusList);31 static HcclResult ChannelGetStatus(const ChannelHandle *channelList, uint32_t listNum, int32_t *statusList);
32 static HcclResult ChannelKernelLaunchForComm(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,32 static HcclResult ChannelKernelLaunchForComm(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,
33- uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle);33+ HcommChannelDesc* hcommDesc, uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle);
34 static HcclResult ChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t *notifyNum);34 static HcclResult ChannelGetNotifyNum(ChannelHandle channelHandle, uint32_t *notifyNum);
35 static HcclResult ChannelGetRemoteMem(ChannelHandle channelHandle, CommMem **remoteMem, uint32_t *memNum, char **memTags);35 static HcclResult ChannelGetRemoteMem(ChannelHandle channelHandle, CommMem **remoteMem, uint32_t *memNum, char **memTags);
36 static HcclResult ChannelGetUserRemoteMem(ChannelHandle channelHandle, CommMem **remoteMem, char ***memTag, uint32_t *memNum);36 static HcclResult ChannelGetUserRemoteMem(ChannelHandle channelHandle, CommMem **remoteMem, char ***memTag, uint32_t *memNum);
@@ -51,9 +51,9 @@ private:
51 hccl::HostMem &hostPackBuf);51 hccl::HostMem &hostPackBuf);
52 static HcclResult FillChannelD2HMap(ChannelHandle *deviceChannelHandles, ChannelHandle *hostChannelHandles, 52 static HcclResult FillChannelD2HMap(ChannelHandle *deviceChannelHandles, ChannelHandle *hostChannelHandles,
53 uint32_t listNum);53 uint32_t listNum);
54- static HcclResult LaunchChannelKernelCommon(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles,54+ static HcclResult LaunchChannelKernelCommon(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles, HcommChannelDesc* hcommDesc,
55 uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle, const std::string &kernelName, bool needProfiling);55 uint32_t listNum, const std::string &commTag, aclrtBinHandle binHandle, const std::string &kernelName, bool needProfiling);
56- static HcclResult ChannelKernelLaunchForBase(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles, 56+ static HcclResult ChannelKernelLaunchForBase(ChannelHandle *channelHandles, ChannelHandle *hostChannelHandles, HcommChannelDesc* hcommDesc,
57 uint32_t listNum, aclrtBinHandle binHandle);57 uint32_t listNum, aclrtBinHandle binHandle);
58 58 
59 static HcclResult ChannelResumeConcurrency(const ChannelHandle *channelList, uint32_t channelNum);59 static HcclResult ChannelResumeConcurrency(const ChannelHandle *channelList, uint32_t channelNum);
@@ -48,7 +48,7 @@ HcclResult EndpointPair::GetSocket(const std::string &socketTag, const uint32_t
48}48}
49 49 
50HcclResult EndpointPair::GetSocket(const uint32_t myRank, const uint32_t rmtRank,50HcclResult EndpointPair::GetSocket(const uint32_t myRank, const uint32_t rmtRank,
51- const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket)51+ const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket*& socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId)
52{52{
53 // 临时方案:支持混跑新增,非Roce场景走orion socketMgr实现server socket复用53 // 临时方案:支持混跑新增,非Roce场景走orion socketMgr实现server socket复用
54 if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {54 if (localEndpointDesc_.loc.locType == EndpointLocType::ENDPOINT_LOC_TYPE_HOST) {
@@ -64,7 +64,7 @@ HcclResult EndpointPair::GetSocket(const uint32_t myRank, const uint32_t rmtRank
64 64 
65 Hccl::LinkData linkData = BuildDefaultLinkData();65 Hccl::LinkData linkData = BuildDefaultLinkData();
66 CHK_RET(EndpointDescPairToLinkDataWithRankIds(myRank, rmtRank,66 CHK_RET(EndpointDescPairToLinkDataWithRankIds(myRank, rmtRank,
67- localEndpointDesc_, remoteEndpointDesc_, linkData, reuseIdx));67+ localEndpointDesc_, remoteEndpointDesc_, linkData, devicePhyId, remoteDevicePhyId, reuseIdx));
68 68 
69 // 复用orion流程可能抛异常69 // 复用orion流程可能抛异常
70 EXCEPTION_HANDLE_BEGIN70 EXCEPTION_HANDLE_BEGIN
@@ -78,7 +78,7 @@ public:
78 78
79 // 临时方案:新增临时接口用于支持混跑79 // 临时方案:新增临时接口用于支持混跑
80 HcclResult GetSocket(const uint32_t myRank, const uint32_t rmtRank,80 HcclResult GetSocket(const uint32_t myRank, const uint32_t rmtRank,
81- const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket *&socket);81+ const std::string &socketTag, u32 reuseIdx, const uint32_t listenPort, Hccl::Socket *&socket, uint32_t devicePhyId, uint32_t remoteDevicePhyId);
82 82 
83 HcclResult CreateChannel(EndpointHandle endpointHandle, CommEngine engine, u32 reuseIdx,83 HcclResult CreateChannel(EndpointHandle endpointHandle, CommEngine engine, u32 reuseIdx,
84 HcommChannelDesc *channelDescs, ChannelHandle *channels);84 HcommChannelDesc *channelDescs, ChannelHandle *channels);
@@ -21,6 +21,7 @@ static bool IsProtocolSupported(CommProtocol protocol)
21 case COMM_PROTOCOL_UBC_TP:21 case COMM_PROTOCOL_UBC_TP:
22 case COMM_PROTOCOL_UBC_CTP:22 case COMM_PROTOCOL_UBC_CTP:
23 case COMM_PROTOCOL_UB_MEM:23 case COMM_PROTOCOL_UB_MEM:
24+ case COMM_PROTOCOL_PCIE:
24 return true;25 return true;
25 default:26 default:
26 return false;27 return false;
@@ -51,6 +52,8 @@ HcclResult Endpoint::CreateEndpoint(const EndpointDesc &endpointDesc, std::uniqu
51 EXECEPTION_CATCH(endpointPtr = std::make_unique<UrmaEndpoint>(endpointDesc), return HCCL_E_PTR);52 EXECEPTION_CATCH(endpointPtr = std::make_unique<UrmaEndpoint>(endpointDesc), return HCCL_E_PTR);
52 } else if (endpointDesc.protocol == COMM_PROTOCOL_UB_MEM && endpointDesc.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {53 } else if (endpointDesc.protocol == COMM_PROTOCOL_UB_MEM && endpointDesc.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {
53 EXECEPTION_CATCH(endpointPtr = std::make_unique<UbMemEndpoint>(endpointDesc), return HCCL_E_PTR);54 EXECEPTION_CATCH(endpointPtr = std::make_unique<UbMemEndpoint>(endpointDesc), return HCCL_E_PTR);
55+ } else if (endpointDesc.protocol == COMM_PROTOCOL_PCIE && endpointDesc.loc.locType == ENDPOINT_LOC_TYPE_DEVICE) {
56+ EXECEPTION_CATCH(endpointPtr = std::make_unique<UbMemEndpoint>(endpointDesc), return HCCL_E_PTR);
54 } else {57 } else {
55 endpointPtr = nullptr;58 endpointPtr = nullptr;
56 HCCL_ERROR("[%s] failed, endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] do not match.", 59 HCCL_ERROR("[%s] failed, endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] do not match.",
@@ -866,6 +866,7 @@ CommProtocol LinkProtocolToCommProtocol(const LinkProtocol &linkProtocol)
866 {LinkProtocol::UB_TP, COMM_PROTOCOL_UBC_TP},866 {LinkProtocol::UB_TP, COMM_PROTOCOL_UBC_TP},
867 {LinkProtocol::ROCE, COMM_PROTOCOL_ROCE},867 {LinkProtocol::ROCE, COMM_PROTOCOL_ROCE},
868 {LinkProtocol::HCCS, COMM_PROTOCOL_HCCS},868 {LinkProtocol::HCCS, COMM_PROTOCOL_HCCS},
869+ {LinkProtocol::PCIE, COMM_PROTOCOL_PCIE},
869 {LinkProtocol::UB_MEM, COMM_PROTOCOL_UB_MEM}};870 {LinkProtocol::UB_MEM, COMM_PROTOCOL_UB_MEM}};
870 871 
871 for (const auto &p : protocolPairs) {872 for (const auto &p : protocolPairs) {
@@ -36,9 +36,9 @@ public:
36 : type(portType.GetType()), linkProtocol_(ConnProto2LinkProtocol(portType.GetProto())), localRankId_(localRankId),36 : type(portType.GetType()), linkProtocol_(ConnProto2LinkProtocol(portType.GetProto())), localRankId_(localRankId),
37 remoteRankId_(remoteRankId), localPortId_(localPortId), remotePortId_(remotePortId){};37 remoteRankId_(remoteRankId), localPortId_(localPortId), remotePortId_(remotePortId){};
38 LinkData(PortDeploymentType portDeploymentType, LinkProtocol linkProtocol, RankId localRankId,38 LinkData(PortDeploymentType portDeploymentType, LinkProtocol linkProtocol, RankId localRankId,
39- RankId remoteRankId, IpAddress localAddr, IpAddress remoteAddr, u32 reuseIdx = 0)39+ RankId remoteRankId, IpAddress localAddr, IpAddress remoteAddr, u32 devicePhyId = 0, u32 remoteDevicePhyId = 0, u32 reuseIdx = 0)
40 : type(portDeploymentType), linkProtocol_(linkProtocol), localRankId_(localRankId), remoteRankId_(remoteRankId),40 : type(portDeploymentType), linkProtocol_(linkProtocol), localRankId_(localRankId), remoteRankId_(remoteRankId),
41- localAddr_(localAddr), remoteAddr_(remoteAddr), reuseIdx_(reuseIdx) {};41+ localAddr_(localAddr), remoteAddr_(remoteAddr), localDeviceId_(devicePhyId), remoteDeviceId_(remoteDevicePhyId), reuseIdx_(reuseIdx) {};
42 42
43 explicit LinkData(const NetInstance::Path &path)43 explicit LinkData(const NetInstance::Path &path)
44 {44 {
@@ -52,6 +52,18 @@ namespace Hccl {
52 return HCCL_SUCCESS;52 return HCCL_SUCCESS;
53 }53 }
54 54 
55+ HcclResult IRankGraph::GetDeviceId(uint32_t rankId, uint32_t *deviceId)
56+ {
57+ CHK_PTR_NULL(rankGraphPtr_);
58+ RankGraph *rankGraph = static_cast<RankGraph *>(rankGraphPtr_);
59+ if (rankGraph->GetPeer(rankId) == nullptr) {
60+ HCCL_ERROR("[GetPeer] rankGraph peer is null!");
61+ return HCCL_E_PTR;
62+ }
63+ *deviceId = rankGraph->GetPeer(rankId)->GetDeviceId();
64+ return HCCL_SUCCESS;
65+ }
66+ 
55 HcclResult IRankGraph::GetNetLayers(uint32_t** netLayers, uint32_t* netLayerNum)67 HcclResult IRankGraph::GetNetLayers(uint32_t** netLayers, uint32_t* netLayerNum)
56 {68 {
57 CHK_PTR_NULL(rankGraphPtr_);69 CHK_PTR_NULL(rankGraphPtr_);
@@ -39,6 +39,7 @@ namespace Hccl {
39 HcclResult GetEndpointNum(uint32_t netLayer, uint32_t topoInstId, uint32_t *num);39 HcclResult GetEndpointNum(uint32_t netLayer, uint32_t topoInstId, uint32_t *num);
40 HcclResult GetEndpointDesc(uint32_t netLayer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc);40 HcclResult GetEndpointDesc(uint32_t netLayer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc);
41 HcclResult GetEndpointInfo(uint32_t rankId, const EndpointDesc *endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info);41 HcclResult GetEndpointInfo(uint32_t rankId, const EndpointDesc *endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info);
42+ HcclResult GetDeviceId(uint32_t rankId, uint32_t *deviceId);
42 43 
43 private:44 private:
44 void *rankGraphPtr_;45 void *rankGraphPtr_;
@@ -24,25 +24,25 @@ public:
24 {24 {
25 }25 }
26 26 
27- ExchangeIpcBufferDto(u64 addr, u64 size, u64 offset, u32 pid) : addr(addr), size(size), offset(offset), pid(pid)27+ ExchangeIpcBufferDto(u64 addr, u64 size, u64 offset, u32 pid, const char *memTag) : addr(addr), size(size), offset(offset), pid(pid), memTag(memTag)
28 {28 {
29 }29 }
30 30 
31 void Serialize(Hccl::BinaryStream &stream) override31 void Serialize(Hccl::BinaryStream &stream) override
32 {32 {
33- stream << addr << size << offset << pid << name;33+ stream << addr << size << offset << pid << name << memTag;
34 }34 }
35 35 
36 void Deserialize(Hccl::BinaryStream &stream) override36 void Deserialize(Hccl::BinaryStream &stream) override
37 {37 {
38- stream >> addr >> size >> offset >> pid >> name;38+ stream >> addr >> size >> offset >> pid >> name >> memTag;
39 }39 }
40 40 
41 std::string Describe() const override41 std::string Describe() const override
42 {42 {
43 std::string strName(name, name + RTS_IPC_MEM_NAME_LEN);43 std::string strName(name, name + RTS_IPC_MEM_NAME_LEN);
44- return StringFormat("ExchangeIpcBufferDto[addr=0x%llx, size=0x%llx, offset=0x%llx, pid=%u, name=%s]", addr, size,44+ return StringFormat("ExchangeIpcBufferDto[addr=0x%llx, size=0x%llx, offset=0x%llx, pid=%u, name=%s, memTag=%s]", addr, size,
45- offset, pid, strName.c_str());45+ offset, pid, strName.c_str(), memTag.c_str());
46 }46 }
47 47 
48 u64 addr{0};48 u64 addr{0};
@@ -50,6 +50,7 @@ public:
50 u64 offset{0};50 u64 offset{0};
51 u32 pid{0};51 u32 pid{0};
52 char_t name[RTS_IPC_MEM_NAME_LEN]{0};52 char_t name[RTS_IPC_MEM_NAME_LEN]{0};
53+ std::string memTag{""};
53};54};
54 55 
55} // namespace Hccl56} // namespace Hccl
@@ -36,7 +36,7 @@ string LocalIpcRmaBuffer::Describe() const
36std::unique_ptr<Serializable> LocalIpcRmaBuffer::GetExchangeDto()36std::unique_ptr<Serializable> LocalIpcRmaBuffer::GetExchangeDto()
37{37{
38 std::unique_ptr<ExchangeIpcBufferDto> dto38 std::unique_ptr<ExchangeIpcBufferDto> dto
39- = make_unique<ExchangeIpcBufferDto>(buf->GetAddr(), buf->GetSize(), ipcOffset, HrtDeviceGetBareTgid());39+ = make_unique<ExchangeIpcBufferDto>(buf->GetAddr(), buf->GetSize(), ipcOffset, HrtDeviceGetBareTgid(), buf->GetMemTag().c_str());
40 (void)memcpy_s(dto->name, RTS_IPC_MEM_NAME_LEN, name, RTS_IPC_MEM_NAME_LEN);40 (void)memcpy_s(dto->name, RTS_IPC_MEM_NAME_LEN, name, RTS_IPC_MEM_NAME_LEN);
41 return std::unique_ptr<Serializable>(dto.release());41 return std::unique_ptr<Serializable>(dto.release());
42}42}
@@ -27,7 +27,10 @@ RemoteIpcRmaBuffer::RemoteIpcRmaBuffer(const Serializable &rmtDto) : RemoteRmaBu
27 ipcAddr = dto.addr;27 ipcAddr = dto.addr;
28 ipcOffset = dto.offset;28 ipcOffset = dto.offset;
29 size = dto.size;29 size = dto.size;
30+ memTag = dto.memTag;
30 (void)memcpy_s(ipcName, RTS_IPC_MEM_NAME_LEN, dto.name, RTS_IPC_MEM_NAME_LEN);31 (void)memcpy_s(ipcName, RTS_IPC_MEM_NAME_LEN, dto.name, RTS_IPC_MEM_NAME_LEN);
32+ HCCL_INFO("[RemoteIpcRmaBuffer][RemoteIpcRmaBuffer]ipcAddr[%llu] ipcOffset[%llu] ipcName[%s] memTag[%s]",
33+ ipcAddr, ipcOffset, ipcName, memTag.c_str());
31 myPid = HrtDeviceGetBareTgid();34 myPid = HrtDeviceGetBareTgid();
32 if (myPid == remotePid) {35 if (myPid == remotePid) {
33 HCCL_INFO("RemoteIpcRmaBuffer: myPid is equal to remotePid, do not need to open memory");36 HCCL_INFO("RemoteIpcRmaBuffer: myPid is equal to remotePid, do not need to open memory");
@@ -48,9 +51,10 @@ RemoteIpcRmaBuffer::RemoteIpcRmaBuffer(const Serializable &rmtDto, const string
48 ipcAddr = dto.addr;51 ipcAddr = dto.addr;
49 ipcOffset = dto.offset;52 ipcOffset = dto.offset;
50 size = dto.size;53 size = dto.size;
54+ memTag = dto.memTag;
51 (void)memcpy_s(ipcName, RTS_IPC_MEM_NAME_LEN, dto.name, RTS_IPC_MEM_NAME_LEN);55 (void)memcpy_s(ipcName, RTS_IPC_MEM_NAME_LEN, dto.name, RTS_IPC_MEM_NAME_LEN);
52- HCCL_INFO("[RemoteIpcRmaBuffer][RemoteIpcRmaBuffer] tag[%s] ipcAddr[%llu] ipcOffset[%llu] ipcName[%s]", tag.c_str(),56+ HCCL_INFO("[RemoteIpcRmaBuffer][RemoteIpcRmaBuffer] tag[%s] ipcAddr[%llu] ipcOffset[%llu] ipcName[%s] memTag[%s]", tag.c_str(),
53- ipcAddr, ipcOffset, ipcName);57+ ipcAddr, ipcOffset, ipcName, memTag.c_str());
54 ipcPtr = HrtIpcOpenMemory(ipcName);58 ipcPtr = HrtIpcOpenMemory(ipcName);
55 addr = reinterpret_cast<uintptr_t>(ipcPtr) + ipcOffset;59 addr = reinterpret_cast<uintptr_t>(ipcPtr) + ipcOffset;
56 isOpened = true;60 isOpened = true;
@@ -23,6 +23,7 @@ namespace Hccl {
23class P2PConnection : public RmaConnection {23class P2PConnection : public RmaConnection {
24public:24public:
25 P2PConnection(Socket *socket, const string &tag);25 P2PConnection(Socket *socket, const string &tag);
26+ ~P2PConnection() = default;
26 27 
27 void Connect() override;28 void Connect() override;
28 RmaConnStatus GetStatus() override;29 RmaConnStatus GetStatus() override;
@@ -12,7 +12,7 @@ target_sources(hccl_v2
12 ipc_local_notify.cc12 ipc_local_notify.cc
13 rdma_local_notify.cc13 rdma_local_notify.cc
14 ub_local_notify.cc14 ub_local_notify.cc
15- remote_notify.cc15+ ipc_remote_notify.cc
16 local_cnt_notify.cc16 local_cnt_notify.cc
17 rts_notify.cc17 rts_notify.cc
18 rts_cnt_notify.cc18 rts_cnt_notify.cc
Rsrc/legacy/unified_platform/resource/notify/remote_notify.ccsrc/legacy/unified_platform/resource/notify/ipc_remote_notify.cc+1-1
@@ -7,7 +7,7 @@
7 * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.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.8 * See LICENSE in the root of the software repository for the full text of the License.
9 */9 */
10-#include "remote_notify.h"10+#include "ipc_remote_notify.h"
11#include "not_support_exception.h"11#include "not_support_exception.h"
12#include "invalid_params_exception.h"12#include "invalid_params_exception.h"
13#include "exception_util.h"13#include "exception_util.h"
Rsrc/legacy/unified_platform/resource/notify/remote_notify.hsrc/legacy/unified_platform/resource/notify/ipc_remote_notify.h+4-4
@@ -7,11 +7,11 @@
7 * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.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.8 * See LICENSE in the root of the software repository for the full text of the License.
9 */9 */
10-#ifndef HCCLV2_REMOTE_NOTIFY_H10+#ifndef HCCLV2_IPC_REMOTE_NOTIFY_H
11-#define HCCLV2_REMOTE_NOTIFY_H11+#define HCCLV2_IPC_REMOTE_NOTIFY_H
12 12 
13#include "rts_notify.h"13#include "rts_notify.h"
14-#include "task.h"14+#include "stream.h"
15#include "rma_type.h"15#include "rma_type.h"
16#include "serializable.h"16#include "serializable.h"
17 17 
@@ -76,4 +76,4 @@ private:
76 76 
77} // namespace Hccl77} // namespace Hccl
78 78 
79-#endif // !HCCLV2_REMOTE_NOTIFY_H79+#endif // !HCCLV2_IPC_REMOTE_NOTIFY_H
@@ -45,9 +45,9 @@ Stream::~Stream()
45 HrtStreamDestroy(ptr);45 HrtStreamDestroy(ptr);
46 }46 }
47 } catch (HcclException &e) {47 } catch (HcclException &e) {
48- HCCL_ERROR(e.what());48+ HCCL_ERROR("%s", e.what());
49 } catch (std::exception &e) {49 } catch (std::exception &e) {
50- HCCL_ERROR(e.what());50+ HCCL_ERROR("%s", e.what());
51 } catch (...) {51 } catch (...) {
52 HCCL_ERROR("Unknow Error occurs when destruct stream %d", id);52 HCCL_ERROR("Unknow Error occurs when destruct stream %d", id);
53 }53 }
@@ -10,7 +10,6 @@
10 10 
11#include "task.h"11#include "task.h"
12#include "local_notify.h"12#include "local_notify.h"
13-#include "remote_notify.h"
14namespace Hccl {13namespace Hccl {
15 14 
16std::string TaskLocalCopy::Describe() const15std::string TaskLocalCopy::Describe() const
@@ -18,6 +18,8 @@
18#include "orion_adapter_rts.h"18#include "orion_adapter_rts.h"
19#include "rts_cnt_notify.h"19#include "rts_cnt_notify.h"
20#include "rts_1ton_cnt_notify.h"20#include "rts_1ton_cnt_notify.h"
21+#include "ipc_local_notify.h"
22+#include "ipc_remote_notify.h"
21 23 
22namespace Hccl {24namespace Hccl {
23using namespace std;25using namespace std;
@@ -19,6 +19,7 @@
19#include "rma_buffer_lite.h"19#include "rma_buffer_lite.h"
20#include "mem_transport_common.h"20#include "mem_transport_common.h"
21#include "rmt_rma_buf_slice_lite.h"21#include "rmt_rma_buf_slice_lite.h"
22+#include "task_param.h"
22namespace Hccl {23namespace Hccl {
23 24 
24inline HcclReduceOp ConvertReduceOpToHcclReduceOp(ReduceOp reduceOp)25inline HcclReduceOp ConvertReduceOpToHcclReduceOp(ReduceOp reduceOp)
@@ -57,6 +58,14 @@ public:
57 return Buffer(0, 0);58 return Buffer(0, 0);
58 }59 }
59 60 
61+ virtual HcclResult BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite)
62+ {
63+ (void)addr;
64+ (void)size;
65+ rmaBufferLite = RmaBufferLite(0, 0, 0, 0);
66+ return HCCL_SUCCESS;
67+ }
68+ 
60 virtual void Post(u32 index, const StreamLite &stream)69 virtual void Post(u32 index, const StreamLite &stream)
61 {70 {
62 (void)index;71 (void)index;
@@ -152,6 +161,15 @@ public:
152 (void)stream;161 (void)stream;
153 }162 }
154 163 
164+ // 自定义算子流程上报task的Callback
165+ HcclResult SetAddTaskInfoCallback(std::function<HcclResult(u32, u32, const TaskParam&, u64)> callback)
166+ {
167+ CHK_PTR_NULL(callback);
168+ newCallback_ = callback;
169+ return HCCL_SUCCESS;
170+ }
171+protected:
172+ std::function<HcclResult(u32, u32, const TaskParam&, u64)> newCallback_{nullptr};
155private:173private:
156};174};
157 175 
@@ -41,6 +41,37 @@ P2PTransportLiteImpl::P2PTransportLiteImpl(
41 ParseRmtBufferVec(rmtBufferUniqueIds, rmtBufferVec);41 ParseRmtBufferVec(rmtBufferUniqueIds, rmtBufferVec);
42}42}
43 43 
44+P2PTransportLiteImpl::P2PTransportLiteImpl(std::vector<char> &uniqueId)
45+{
46+ Init(uniqueId);
47+}
48+ 
49+void P2PTransportLiteImpl::Init(std::vector<char> &uniqueId)
50+{
51+ // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
52+ BinaryStream binaryStream(uniqueId);
53+ u32 theType;
54+ binaryStream >> theType;
55+ binaryStream >> notifyNum;
56+ binaryStream >> bufferNum;
57+ 
58+ std::vector<char> notifyUniqueIds;
59+ binaryStream >> notifyUniqueIds;
60+ ParseLocNotifyVec(notifyUniqueIds);
61+ 
62+ std::vector<char> rmtNotifyUniqueIds;
63+ binaryStream >> rmtNotifyUniqueIds;
64+ ParseRmtNotifyVec(rmtNotifyUniqueIds, rmtNotifyVec);
65+ 
66+ std::vector<char> locBufferUniqueIds;
67+ binaryStream >> locBufferUniqueIds;
68+ ParseRmtBufferVec(locBufferUniqueIds, locBufferVec);
69+ 
70+ std::vector<char> rmtBufferUniqueIds;
71+ binaryStream >> rmtBufferUniqueIds;
72+ ParseRmtBufferVec(rmtBufferUniqueIds, rmtBufferVec);
73+}
74+ 
44P2PTransportLiteImpl::~P2PTransportLiteImpl()75P2PTransportLiteImpl::~P2PTransportLiteImpl()
45{76{
46}77}
@@ -113,7 +144,7 @@ void P2PTransportLiteImpl::ParseRmtNotifyVec(std::vector<char> &data, std::vecto
113 }144 }
114}145}
115 146 
116-void P2PTransportLiteImpl::ParseRmtBufferVec(std::vector<char> &data, std::vector<RmtP2PBufLite> &vec) const147+void P2PTransportLiteImpl::ParseRmtBufferVec(std::vector<char> &data, std::vector<P2PBufLite> &vec) const
117{148{
118 if (bufferNum == 0) {149 if (bufferNum == 0) {
119 HCCL_WARNING("P2PTransportLiteImpl::ParseRmtBufferVec bufferNum is 0");150 HCCL_WARNING("P2PTransportLiteImpl::ParseRmtBufferVec bufferNum is 0");
@@ -126,7 +157,7 @@ void P2PTransportLiteImpl::ParseRmtBufferVec(std::vector<char> &data, std::vecto
126 BinaryStream binaryStream(data);157 BinaryStream binaryStream(data);
127 158 
128 for (u32 idx = 0; idx < bufferNum; idx++) {159 for (u32 idx = 0; idx < bufferNum; idx++) {
129- RmtP2PBufLite p2pBufLite;160+ P2PBufLite p2pBufLite;
130 binaryStream >> p2pBufLite.addr;161 binaryStream >> p2pBufLite.addr;
131 binaryStream >> p2pBufLite.size;162 binaryStream >> p2pBufLite.size;
132 HCCL_INFO("[P2PTransportLiteImpl][ParseRmtBufferVec]idx=%u, %s", idx, p2pBufLite.Describe().c_str());163 HCCL_INFO("[P2PTransportLiteImpl][ParseRmtBufferVec]idx=%u, %s", idx, p2pBufLite.Describe().c_str());
@@ -145,6 +176,36 @@ Buffer P2PTransportLiteImpl::GetRmtBuffer(u32 index)
145 return Buffer(rmtBufferVec[index].addr, rmtBufferVec[index].size);176 return Buffer(rmtBufferVec[index].addr, rmtBufferVec[index].size);
146}177}
147 178 
179+HcclResult P2PTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size,
180+ RmaBufferLite &rmaBufferLite)
181+{
182+ HCCL_INFO("[P2PTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferVec, whose size is %zu. ",
183+ __func__, addr, size, locBufferVec.size());
184+ if (locBufferVec.empty()) {
185+ HCCL_ERROR("[P2PTransportLiteImpl::%s] locBufferVec is empty.", __func__);
186+ return HCCL_E_INTERNAL;
187+ }
188+ 
189+ bool isAddrInRange = false;
190+ for (auto &it : locBufferVec) {
191+ Buffer iterBuf(it.addr, it.size);
192+ if (iterBuf.Contains(addr, size)) {
193+ rmaBufferLite = RmaBufferLite(addr, size, 0, 0);
194+ isAddrInRange = true;
195+ break;
196+ }
197+ }
198+ 
199+ if (!isAddrInRange) {
200+ HCCL_WARNING("[P2PTransportLiteImpl::%s] addr[0x%llx], size[0x%llx] not in any range of locBufferVec.",
201+ __func__, addr, size);
202+ rmaBufferLite = RmaBufferLite(addr, size, 0, 0);
203+ return HCCL_SUCCESS;
204+ }
205+ 
206+ return HCCL_SUCCESS;
207+}
208+ 
148void P2PTransportLiteImpl::BuildNotifyRecordTask(const StreamLite &stream, u64 rmtNotifyAddr)209void P2PTransportLiteImpl::BuildNotifyRecordTask(const StreamLite &stream, u64 rmtNotifyAddr)
149{210{
150 // Post仅需向对端寄存器写入1211 // Post仅需向对端寄存器写入1
@@ -192,7 +253,7 @@ void P2PTransportLiteImpl::BuildP2PRead(const StreamLite &stream, const RmaBuffe
192 HCCL_INFO("P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, taskId=%u",253 HCCL_INFO("P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, taskId=%u",
193 __func__, src, dst, blockSize, taskId);254 __func__, src, dst, blockSize, taskId);
194 255 
195- if (callback_) {256+ if (callback_ || newCallback_) {
196 TaskParam taskParam{};257 TaskParam taskParam{};
197 taskParam.taskType = TaskParamType::TASK_SDMA;258 taskParam.taskType = TaskParamType::TASK_SDMA;
198 taskParam.beginTime = ProfGetCurCpuTimestamp();259 taskParam.beginTime = ProfGetCurCpuTimestamp();
@@ -202,7 +263,12 @@ void P2PTransportLiteImpl::BuildP2PRead(const StreamLite &stream, const RmaBuffe
202 taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;263 taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
203 taskParam.taskPara.DMA.linkType = DfxLinkType::PCIE;264 taskParam.taskPara.DMA.linkType = DfxLinkType::PCIE;
204 taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;265 taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
205- callback_(stream.GetSqId(), taskId, taskParam);266+ if (callback_) {
267+ callback_(stream.GetSqId(), taskId, taskParam);
268+ }
269+ if (newCallback_) {
270+ newCallback_(stream.GetSqId(), taskId, taskParam, reinterpret_cast<u64>(this));
271+ }
206 } else {272 } else {
207 HCCL_WARNING("[P2PTransportLiteImpl][%s] callback_ is nullptr.", __func__);273 HCCL_WARNING("[P2PTransportLiteImpl][%s] callback_ is nullptr.", __func__);
208 }274 }
@@ -249,8 +315,8 @@ void P2PTransportLiteImpl::BuildP2PReadReduce(const StreamLite &stream, const Rm
249 HCCL_INFO("P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, reduceIn=%s, taskId=%u",315 HCCL_INFO("P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, reduceIn=%s, taskId=%u",
250 __func__, src, dst, blockSize, reduceIn.Describe(), taskId);316 __func__, src, dst, blockSize, reduceIn.Describe(), taskId);
251 317 
252- if (callback_) {318+ if (callback_ || newCallback_) {
253- TaskParam taskParam {};319+ TaskParam taskParam{};
254 taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;320 taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
255 taskParam.beginTime = ProfGetCurCpuTimestamp();321 taskParam.beginTime = ProfGetCurCpuTimestamp();
256 taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(src);322 taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(src);
@@ -260,7 +326,12 @@ void P2PTransportLiteImpl::BuildP2PReadReduce(const StreamLite &stream, const Rm
260 taskParam.taskPara.Reduce.linkType = DfxLinkType::PCIE;326 taskParam.taskPara.Reduce.linkType = DfxLinkType::PCIE;
261 taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);327 taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
262 taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);328 taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
263- callback_(stream.GetSqId(), taskId, taskParam);329+ if (callback_) {
330+ callback_(stream.GetSqId(), taskId, taskParam);
331+ }
332+ if (newCallback_) {
333+ newCallback_(stream.GetSqId(), taskId, taskParam, reinterpret_cast<u64>(this));
334+ }
264 } else {335 } else {
265 HCCL_WARNING("[P2PTransportLiteImpl][%s] callback_ is nullptr.", __func__);336 HCCL_WARNING("[P2PTransportLiteImpl][%s] callback_ is nullptr.", __func__);
266 }337 }
@@ -286,7 +357,7 @@ void P2PTransportLiteImpl::Post(u32 index, const StreamLite &stream)
286 HCCL_INFO("P2PTransportLiteImpl::Post rmtNotifyAddr[0x%llx], notifyId[%u], taskId[%u]",357 HCCL_INFO("P2PTransportLiteImpl::Post rmtNotifyAddr[0x%llx], notifyId[%u], taskId[%u]",
287 rmtNotifyAddr, rmtNotifyVec[index].id, taskId);358 rmtNotifyAddr, rmtNotifyVec[index].id, taskId);
288 359 
289- if (callback_ == nullptr)360+ if (callback_ == nullptr || newCallback_ == nullptr)
290 {361 {
291 HCCL_WARNING("[P2PTransportLiteImpl] callback_ is nullptr.");362 HCCL_WARNING("[P2PTransportLiteImpl] callback_ is nullptr.");
292 return;363 return;
@@ -297,7 +368,12 @@ void P2PTransportLiteImpl::Post(u32 index, const StreamLite &stream)
297 taskParam.beginTime = ProfGetCurCpuTimestamp();368 taskParam.beginTime = ProfGetCurCpuTimestamp();
298 taskParam.taskPara.Notify.notifyID = rmtNotifyVec[index].id;369 taskParam.taskPara.Notify.notifyID = rmtNotifyVec[index].id;
299 taskParam.taskPara.Notify.value = 1;370 taskParam.taskPara.Notify.value = 1;
300- callback_(stream.GetSqId(), taskId, taskParam);371+ if (callback_) {
372+ callback_(stream.GetSqId(), taskId, taskParam);
373+ }
374+ if (newCallback_) {
375+ newCallback_(stream.GetSqId(), taskId, taskParam, reinterpret_cast<u64>(this));
376+ }
301}377}
302 378 
303void P2PTransportLiteImpl::Wait(u32 index, const StreamLite &stream)379void P2PTransportLiteImpl::Wait(u32 index, const StreamLite &stream)
@@ -312,7 +388,7 @@ void P2PTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite &stream,
312 stream.GetRtsq()->NotifyWait(notifyId, timeout);388 stream.GetRtsq()->NotifyWait(notifyId, timeout);
313 389 
314 HCCL_INFO("P2PTransportLiteImpl::WaitWithTimeout notifyId[%u], taskId[%u], timeout[%u]", notifyId, taskId, timeout);390 HCCL_INFO("P2PTransportLiteImpl::WaitWithTimeout notifyId[%u], taskId[%u], timeout[%u]", notifyId, taskId, timeout);
315- if (callback_ == nullptr)391+ if (callback_ == nullptr || newCallback_ == nullptr)
316 {392 {
317 HCCL_WARNING("[P2PTransportLiteImpl] callback_ is nullptr.");393 HCCL_WARNING("[P2PTransportLiteImpl] callback_ is nullptr.");
318 return;394 return;
@@ -323,7 +399,12 @@ void P2PTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite &stream,
323 taskParam.beginTime = ProfGetCurCpuTimestamp();399 taskParam.beginTime = ProfGetCurCpuTimestamp();
324 taskParam.taskPara.Notify.notifyID = notifyId;400 taskParam.taskPara.Notify.notifyID = notifyId;
325 taskParam.taskPara.Notify.value = 1;401 taskParam.taskPara.Notify.value = 1;
326- callback_(stream.GetSqId(), taskId, taskParam);402+ if (callback_) {
403+ callback_(stream.GetSqId(), taskId, taskParam);
404+ }
405+ if (newCallback_) {
406+ newCallback_(stream.GetSqId(), taskId, taskParam, reinterpret_cast<u64>(this));
407+ }
327}408}
328 409 
329void P2PTransportLiteImpl::Read(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)410void P2PTransportLiteImpl::Read(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)
@@ -28,6 +28,7 @@ public:
28 std::function<void(u32 streamId, u32 taskId, const TaskParam &taskParam)> callback);28 std::function<void(u32 streamId, u32 taskId, const TaskParam &taskParam)> callback);
29 29 
30 P2PTransportLiteImpl(std::vector<char> &uniqueId);30 P2PTransportLiteImpl(std::vector<char> &uniqueId);
31+ void Init(std::vector<char> &uniqueId);
31 32 
32 ~P2PTransportLiteImpl() override;33 ~P2PTransportLiteImpl() override;
33 34 
@@ -35,6 +36,8 @@ public:
35 36 
36 Buffer GetRmtBuffer(u32 index) override;37 Buffer GetRmtBuffer(u32 index) override;
37 38 
39+ HcclResult BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite) override;
40+ 
38 void Post(u32 index, const StreamLite &stream) override;41 void Post(u32 index, const StreamLite &stream) override;
39 42 
40 void Wait(u32 index, const StreamLite &stream) override;43 void Wait(u32 index, const StreamLite &stream) override;
@@ -63,17 +66,18 @@ private:
63 }66 }
64 };67 };
65 68 
66- struct RmtP2PBufLite {69+ struct P2PBufLite {
67 u64 addr;70 u64 addr;
68 u64 size;71 u64 size;
69 std::string Describe() const72 std::string Describe() const
70 {73 {
71- return StringFormat("RmtP2PBufLite[addr=0x%llx, size=0x%llx]", addr, size);74+ return StringFormat("P2PBufLite[addr=0x%llx, size=0x%llx]", addr, size);
72 }75 }
73 };76 };
74 77 
75 std::vector<RmtP2PNotifyLite> rmtNotifyVec;78 std::vector<RmtP2PNotifyLite> rmtNotifyVec;
76- std::vector<RmtP2PBufLite> rmtBufferVec;79+ std::vector<P2PBufLite> rmtBufferVec;
80+ std::vector<P2PBufLite> locBufferVec;
77 81 
78 std::vector<std::unique_ptr<NotifyLite>> locNotifyVec;82 std::vector<std::unique_ptr<NotifyLite>> locNotifyVec;
79 83 
@@ -83,7 +87,7 @@ private:
83 87 
84 void ParseRmtNotifyVec(std::vector<char> &data, std::vector<RmtP2PNotifyLite> &vec) const;88 void ParseRmtNotifyVec(std::vector<char> &data, std::vector<RmtP2PNotifyLite> &vec) const;
85 89 
86- void ParseRmtBufferVec(std::vector<char> &data, std::vector<RmtP2PBufLite> &vec) const;90+ void ParseRmtBufferVec(std::vector<char> &data, std::vector<P2PBufLite> &vec) const;
87 91 
88 void BuildNotifyRecordTask(const StreamLite &stream, u64 rmtNotifyAddr);92 void BuildNotifyRecordTask(const StreamLite &stream, u64 rmtNotifyAddr);
89 93 
@@ -89,12 +89,6 @@ void UbTransportLiteImpl::Init(std::vector<char> &uniqueId)
89 ParseConnVec(connUniqueIds);89 ParseConnVec(connUniqueIds);
90}90}
91 91 
92-HcclResult UbTransportLiteImpl::SetAddTaskInfoCallback(std::function<HcclResult(u32, u32, const TaskParam&, u64)> callback) {
93- CHK_PTR_NULL(callback);
94- newCallback_ = callback;
95- return HCCL_SUCCESS;
96-}
97- 
98UbTransportLiteImpl::~UbTransportLiteImpl()92UbTransportLiteImpl::~UbTransportLiteImpl()
99{93{
100 for (auto &it : connUniqueIdVec) {94 for (auto &it : connUniqueIdVec) {
@@ -276,7 +270,7 @@ RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const RmaBufferLit
276 return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId() , lite.GetTokenValue());270 return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId() , lite.GetTokenValue());
277}271}
278 272 
279-HcclResult UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite) const273+HcclResult UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite)
280{274{
281 HCCL_INFO("[UbTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferVec, whose size is %zu. ",275 HCCL_INFO("[UbTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferVec, whose size is %zu. ",
282 __func__, addr, size, locBufferVec.size());276 __func__, addr, size, locBufferVec.size());
@@ -70,13 +70,12 @@ public:
70 void BatchTransfer(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,70 void BatchTransfer(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
71 const std::vector<TransferOp> &transferOp, const StreamLite &stream) override;71 const std::vector<TransferOp> &transferOp, const StreamLite &stream) override;
72 72 
73- HcclResult BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite) const;73+ HcclResult BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite) override;
74 HcclResult Fence();74 HcclResult Fence();
75 75 
76 HcclResult Clean();76 HcclResult Clean();
77 HcclResult Resume(std::vector<char> &uniqueId);77 HcclResult Resume(std::vector<char> &uniqueId);
78 78 
79- HcclResult SetAddTaskInfoCallback(std::function<HcclResult(u32, u32, const TaskParam&, u64)> callback); // 自定义算子流程上报task的Callback
80private:79private:
81 u32 notifyNum{0};80 u32 notifyNum{0};
82 u32 bufferNum{0};81 u32 bufferNum{0};
@@ -18,7 +18,7 @@
18#include "remote_rma_buffer.h"18#include "remote_rma_buffer.h"
19#include "../../resource/connection/rma_connection.h"19#include "../../resource/connection/rma_connection.h"
20#include "local_notify.h"20#include "local_notify.h"
21-#include "remote_notify.h"21+#include "ipc_remote_notify.h"
22#include "local_cnt_notify.h"22#include "local_cnt_notify.h"
23#include "op_mode.h"23#include "op_mode.h"
24#include "mem_transport_common.h"24#include "mem_transport_common.h"
@@ -365,6 +365,35 @@ std::vector<char> P2PTransport::GetUniqueId()
365 return result;365 return result;
366}366}
367 367 
368+std::vector<char> P2PTransport::GetUniqueIdV2()
369+{
370+ if (baseStatus != TransportStatus::READY) {
371+ MACRO_THROW(InternalException, StringFormat("transport status is not ready, please check"));
372+ }
373+ u32 type = static_cast<u32>(transportType);
374+ BinaryStream binaryStream;
375+ binaryStream << type;
376+ binaryStream << notifyNum;
377+ binaryStream << bufferNum;
378+ 
379+ // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
380+ auto notifyUniqueIds = GetNotifyUniqueIds();
381+ binaryStream << notifyUniqueIds;
382+ 
383+ auto rmtNotifyUniqueIds = GetRmtNotifyUniqueIds();
384+ binaryStream << rmtNotifyUniqueIds;
385+ 
386+ auto locBufferUniqueIds = GetLocBufferUniqueIds();
387+ binaryStream << locBufferUniqueIds;
388+ 
389+ auto rmtBufferUniqueIds = GetRmtBufferUniqueIds();
390+ binaryStream << rmtBufferUniqueIds;
391+ 
392+ std::vector<char> result;
393+ binaryStream.Dump(result);
394+ return result;
395+}
396+ 
368std::vector<char> P2PTransport::GetNotifyUniqueIds()397std::vector<char> P2PTransport::GetNotifyUniqueIds()
369{398{
370 HCCL_INFO("start packing all notify uniqueIds");399 HCCL_INFO("start packing all notify uniqueIds");
@@ -407,17 +436,35 @@ std::vector<char> P2PTransport::GetRmtNotifyUniqueIds() const
407 return result;436 return result;
408}437}
409 438 
410-std::vector<char> P2PTransport::GetSingleRmtBufferUniqueId(u64 addr, u64 size) const439+std::vector<char> P2PTransport::GetSingleBufferUniqueId(u64 addr, u64 size) const
411{440{
412 BinaryStream binaryStream;441 BinaryStream binaryStream;
413 binaryStream << addr;442 binaryStream << addr;
414 binaryStream << size;443 binaryStream << size;
415- HCCL_INFO("P2PTransport RmtBufferAddr[addr=0x%llx, size=0x%llx]", addr, size);444+ HCCL_INFO("P2PTransport BufferAddr[addr=0x%llx, size=0x%llx]", addr, size);
416 std::vector<char> result;445 std::vector<char> result;
417 binaryStream.Dump(result);446 binaryStream.Dump(result);
418 return result;447 return result;
419}448}
420 449 
450+std::vector<char> P2PTransport::GetLocBufferUniqueIds() const
451+{
452+ HCCL_INFO("start packing all local buffer uniqueIds");
453+ std::vector<char> result(0);
454+ for (auto &it : commonLocRes.bufferVec) {
455+ std::vector<char> uniqueId;
456+ if (it != nullptr) {
457+ uniqueId = GetSingleBufferUniqueId(it->GetAddr(), it->GetSize());
458+ HCCL_INFO("P2PTransport::GetLocBufferUniqueIds, %s", it->Describe().c_str());
459+ } else {
460+ uniqueId = GetSingleBufferUniqueId(0, 0); // 填充一个空的buffer
461+ HCCL_INFO("P2PTransport::GetLocBufferUniqueIds, null buffer");
462+ }
463+ result.insert(result.end(), uniqueId.begin(), uniqueId.end());
464+ }
465+ return result;
466+}
467+ 
421std::vector<char> P2PTransport::GetRmtBufferUniqueIds() const468std::vector<char> P2PTransport::GetRmtBufferUniqueIds() const
422{469{
423 HCCL_INFO("start packing all remote buffer uniqueIds");470 HCCL_INFO("start packing all remote buffer uniqueIds");
@@ -425,14 +472,51 @@ std::vector<char> P2PTransport::GetRmtBufferUniqueIds() const
425 for (auto &it : rmtBufferVec) {472 for (auto &it : rmtBufferVec) {
426 std::vector<char> uniqueId;473 std::vector<char> uniqueId;
427 if (it != nullptr) {474 if (it != nullptr) {
428- uniqueId = GetSingleRmtBufferUniqueId(it->GetAddr(), it->GetSize());475+ uniqueId = GetSingleBufferUniqueId(it->GetAddr(), it->GetSize());
429 HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, %s", it->Describe().c_str());476 HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
430 } else {477 } else {
431- uniqueId = GetSingleRmtBufferUniqueId(0, 0); // 填充一个空的buffer478+ uniqueId = GetSingleBufferUniqueId(0, 0); // 填充一个空的buffer
432 HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, null buffer");479 HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, null buffer");
433 }480 }
434 result.insert(result.end(), uniqueId.begin(), uniqueId.end());481 result.insert(result.end(), uniqueId.begin(), uniqueId.end());
435 }482 }
436 return result;483 return result;
437}484}
485+ 
486+HcclResult P2PTransport::GetRemoteMem(HcclMem **remoteMem, uint32_t *memNum, char **memTags)
487+{
488+ CHK_PRT_RET(!remoteMem, HCCL_ERROR("[GetRemoteMem] remoteMem is nullptr"), HCCL_E_PARA);
489+ CHK_PRT_RET(!memNum, HCCL_ERROR("[GetRemoteMem] memNum is nullptr"), HCCL_E_PARA);
490+ HCCL_RUN_INFO("[P2PTransport]GetRemoteMem begin");
491+
492+ *remoteMem = nullptr;
493+ *memNum = 0;
494+
495+ std::lock_guard<std::mutex> lock(remoteMemsMutex_);
496+
497+ uint32_t totalCount = rmtBufferVec.size();
498+ if (totalCount == 0) {
499+ HCCL_INFO("[GetRemoteMem] No remote memory regions available");
500+ return HCCL_SUCCESS;
501+ }
502+ // 释放之前的内存
503+ remoteMemsPtr_.reset();
504+ remoteMemsPtr_ = std::make_unique<HcclMem[]>(totalCount);
505+ CHK_PTR_NULL(remoteMemsPtr_);
506+ 
507+ for (uint32_t i = 0; i < totalCount; i++) {
508+ auto& rmtRmaBuffer = rmtBufferVec[i];
509+ remoteMemsPtr_[i].type = rmtRmaBuffer->GetMemType();
510+ remoteMemsPtr_[i].addr = reinterpret_cast<void *>(rmtRmaBuffer->GetAddr());
511+ remoteMemsPtr_[i].size = rmtRmaBuffer->GetSize();
512+ memTags[i] = const_cast<char*>(rmtRmaBuffer->GetMemTag().c_str());
513+ HCCL_INFO("[%s] addr[%p] size[%zu] rmtRmaBuffer[%p] memTags[%s]",
514+ __func__, reinterpret_cast<void *>(rmtRmaBuffer->GetAddr()), rmtRmaBuffer->GetSize(), rmtRmaBuffer.get(), memTags[i]);
515+ }
516+ 
517+ *memNum = totalCount;
518+ *remoteMem = remoteMemsPtr_.get();
519+ HCCL_RUN_INFO("[P2PTransport]GetRemoteMem end, memNum[%u]", totalCount);
520+ return HCCL_SUCCESS;
521+}
438} // namespace Hccl522} // namespace Hccl
@@ -12,6 +12,9 @@
12 12 
13#include "base_mem_transport.h"13#include "base_mem_transport.h"
14#include "virtual_topo.h"14#include "virtual_topo.h"
15+#include "ipc_remote_notify.h"
16+#include "../../../../legacy/unified_platform/resource/buffer/local_ipc_rma_buffer.h"
17+#include "remote_rma_buffer.h"
15 18 
16namespace Hccl {19namespace Hccl {
17class P2PTransport : public BaseMemTransport {20class P2PTransport : public BaseMemTransport {
@@ -20,12 +23,16 @@ public:
20 23 
21 P2PTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData, const Socket &socket, std::function<void(u32 streamId, u32 taskId, TaskParam taskParam)> callback);24 P2PTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData, const Socket &socket, std::function<void(u32 streamId, u32 taskId, TaskParam taskParam)> callback);
22 25 
26+ ~P2PTransport() = default;
27+ 
23 std::string Describe() const override;28 std::string Describe() const override;
24 29 
25 TransportStatus GetStatus() override;30 TransportStatus GetStatus() override;
26 31 
27 std::vector<char> GetUniqueId() override;32 std::vector<char> GetUniqueId() override;
28 33 
34+ std::vector<char> GetUniqueIdV2();
35+ 
29 void Post(u32 index, const Stream &stream) override;36 void Post(u32 index, const Stream &stream) override;
30 37 
31 void Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream) override;38 void Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream) override;
@@ -38,6 +45,8 @@ public:
38 void WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const ReduceIn &reduceIn,45 void WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const ReduceIn &reduceIn,
39 const Stream &stream) override;46 const Stream &stream) override;
40 47 
48+ HcclResult GetRemoteMem(HcclMem **remoteMem, uint32_t *memNum, char **memTags);
49+ 
41private:50private:
42 MemoryBuffer GetLocMemBuffer(const RmaBufferSlice &locSlice) const;51 MemoryBuffer GetLocMemBuffer(const RmaBufferSlice &locSlice) const;
43 MemoryBuffer GetRmtMemBuffer(const RmtRmaBufferSlice &rmtSlice) const;52 MemoryBuffer GetRmtMemBuffer(const RmtRmaBufferSlice &rmtSlice) const;
@@ -67,10 +76,14 @@ private:
67 void RmtBufferVecUnpackProc(BinaryStream &binaryStream);76 void RmtBufferVecUnpackProc(BinaryStream &binaryStream);
68 77 
69 std::vector<char> GetSingleRmtNotifyUniqueId(u64 addr, u64 size, u32 notifyId) const;78 std::vector<char> GetSingleRmtNotifyUniqueId(u64 addr, u64 size, u32 notifyId) const;
70- std::vector<char> GetSingleRmtBufferUniqueId(u64 addr, u64 size) const;79+ std::vector<char> GetSingleBufferUniqueId(u64 addr, u64 size) const;
71 std::vector<char> GetNotifyUniqueIds();80 std::vector<char> GetNotifyUniqueIds();
72 std::vector<char> GetRmtNotifyUniqueIds() const;81 std::vector<char> GetRmtNotifyUniqueIds() const;
82+ std::vector<char> GetLocBufferUniqueIds() const;
73 std::vector<char> GetRmtBufferUniqueIds() const;83 std::vector<char> GetRmtBufferUniqueIds() const;
84+ 
85+ std::mutex remoteMemsMutex_; // 远端内存列表互斥锁
86+ std::unique_ptr<HcclMem[]> remoteMemsPtr_;
74};87};
75 88 
76} // namespace Hccl89} // namespace Hccl
@@ -138,7 +138,7 @@ target_sources(hccl_v2_utest
138 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/ipc_local_notify.cc138 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/ipc_local_notify.cc
139 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rdma_local_notify.cc139 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rdma_local_notify.cc
140 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/ub_local_notify.cc140 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/ub_local_notify.cc
141- ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/remote_notify.cc141+ ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/ipc_remote_notify.cc
142 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/local_cnt_notify.cc142 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/local_cnt_notify.cc
143 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rts_notify.cc143 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rts_notify.cc
144 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rts_cnt_notify.cc144 ${HCCL_V2_CODE_ROOT}/unified_platform/resource/notify/rts_cnt_notify.cc
@@ -12,7 +12,7 @@
12#include <mockcpp/mokc.h>12#include <mockcpp/mokc.h>
13#include <mockcpp/mockcpp.hpp>13#include <mockcpp/mockcpp.hpp>
14#define private public14#define private public
15-#include "remote_notify.h"15+#include "ipc_remote_notify.h"
16#undef private16#undef private
17#include "local_notify.h"17#include "local_notify.h"
18#include "exchange_ipc_notify_dto.h"18#include "exchange_ipc_notify_dto.h"
@@ -14,7 +14,7 @@
14#include "task.h"14#include "task.h"
15#include "null_ptr_exception.h"15#include "null_ptr_exception.h"
16#include "internal_exception.h"16#include "internal_exception.h"
17-#include "remote_notify.h"17+#include "ipc_remote_notify.h"
18#include "ipc_local_notify.h"18#include "ipc_local_notify.h"
19 19 
20using namespace Hccl;20using namespace Hccl;
@@ -16,13 +16,13 @@ set(src_to_test_list
16 ut_aicpu_ts_HcommChannelNotifyRecordOnThread.cc16 ut_aicpu_ts_HcommChannelNotifyRecordOnThread.cc
17 ut_aicpu_ts_HcommChannelNotifyWaitOnThread.cc17 ut_aicpu_ts_HcommChannelNotifyWaitOnThread.cc
18 ut_aicpu_ts_HcommChannelNotifyWaitOnThread_WithTimeout.cc18 ut_aicpu_ts_HcommChannelNotifyWaitOnThread_WithTimeout.cc
19- ut_aicpu_ts_HcommWriteWithNotifyOnThread.cc19+ #ut_aicpu_ts_HcommWriteWithNotifyOnThread.cc
20- ut_aicpu_ts_HcommWriteReduceWithNotifyOnThread.cc20+ #ut_aicpu_ts_HcommWriteReduceWithNotifyOnThread.cc
21- ut_aicpu_ts_HcommWriteOnThread.cc21+ #ut_aicpu_ts_HcommWriteOnThread.cc
22- ut_aicpu_ts_HcommWriteReduceOnThread.cc22+ #ut_aicpu_ts_HcommWriteReduceOnThread.cc
23- ut_aicpu_ts_HcommReadOnThread.cc23+ #ut_aicpu_ts_HcommReadOnThread.cc
24- ut_aicpu_ts_HcommReadReduceOnThread.cc24+ #ut_aicpu_ts_HcommReadReduceOnThread.cc
25- ut_aicpu_ts_HcommChannelFenceOnThread.cc25+ #ut_aicpu_ts_HcommChannelFenceOnThread.cc
26 ${HCCL_CODE_ROOT}/src/framework/communicator/impl/independent_op/data_api/hccl_api_data_aicpu_ts.cc26 ${HCCL_CODE_ROOT}/src/framework/communicator/impl/independent_op/data_api/hccl_api_data_aicpu_ts.cc
27 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/stream/aicpu/rtsq_base.cc27 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/stream/aicpu/rtsq_base.cc
28 ${HCCL_LLT_ROOT}/ut/stub/legacy/llt_stub_ub_transport_lite_impl.cc28 ${HCCL_LLT_ROOT}/ut/stub/legacy/llt_stub_ub_transport_lite_impl.cc
@@ -35,6 +35,10 @@ protected:
35 .stubs()35 .stubs()
36 .with(any(), any(), any())36 .with(any(), any(), any())
37 .will(returnValue(HCCL_SUCCESS));37 .will(returnValue(HCCL_SUCCESS));
38+ MOCKER_CPP(&Hccl::BaseTransportLiteImpl::BuildLocRmaBufferLite)
39+ .stubs()
40+ .with(any(), any(), any())
41+ .will(returnValue(HCCL_SUCCESS));
38 }42 }
39 43 
40 virtual void TearDown() override44 virtual void TearDown() override
@@ -106,6 +106,7 @@ TEST_F(MyRankTest, Ut_When_BatchCreateChannels_Expect_SUCCESS)
106 setenv("HCCL_DFS_CONFIG", "task_exception:on", 1);106 setenv("HCCL_DFS_CONFIG", "task_exception:on", 1);
107 uint32_t devPort = 60001;107 uint32_t devPort = 60001;
108 MOCKER_CPP(&Hccl::IRankGraph::GetDevicePort).stubs().with(any(), outBoundP(&devPort)).will(returnValue(HCCL_SUCCESS));108 MOCKER_CPP(&Hccl::IRankGraph::GetDevicePort).stubs().with(any(), outBoundP(&devPort)).will(returnValue(HCCL_SUCCESS));
109+ MOCKER_CPP(&Hccl::IRankGraph::GetDeviceId).stubs().with(any()).will(returnValue(static_cast<int>(HCCL_SUCCESS)));
109 MOCKER_CPP(&Hccl::SocketManager::GetConnectedSocket).stubs().with(any()).will(returnValue((Hccl::Socket*)0xab));110 MOCKER_CPP(&Hccl::SocketManager::GetConnectedSocket).stubs().with(any()).will(returnValue((Hccl::Socket*)0xab));
110 MOCKER_CPP(&hccl::CommMems::GetTagMemoryHandles).stubs().with(any()).will(returnValue(static_cast<int>(HCCL_SUCCESS)));111 MOCKER_CPP(&hccl::CommMems::GetTagMemoryHandles).stubs().with(any()).will(returnValue(static_cast<int>(HCCL_SUCCESS)));
111 MOCKER_CPP(&hcomm::EndpointMgr::RegisterMemory).stubs().with(any()).will(returnValue(static_cast<int>(HCCL_SUCCESS)));112 MOCKER_CPP(&hcomm::EndpointMgr::RegisterMemory).stubs().with(any()).will(returnValue(static_cast<int>(HCCL_SUCCESS)));
@@ -38,7 +38,8 @@ public:
38 }38 }
39 39 
40 std::unique_ptr<Hccl::Serializable> GetExchangeDto() override {40 std::unique_ptr<Hccl::Serializable> GetExchangeDto() override {
41- return std::make_unique<Hccl::ExchangeIpcBufferDto>(mockAddr_, mockSize_, mockOffset_, mockPid_);41+ std::string memTag = "HcclBuff";
42+ return std::make_unique<Hccl::ExchangeIpcBufferDto>(mockAddr_, mockSize_, mockOffset_, mockPid_, memTag.c_str());
42 }43 }
43 44 
44 u64 mockAddr_{0};45 u64 mockAddr_{0};
@@ -70,52 +70,58 @@ TEST_F(TestChannelProcess, Ut_TestFillChannelD2HMap_When_ListNumZero_Return_HCCL
70// LaunchChannelKernelCommon 空指针和参数校验测试70// LaunchChannelKernelCommon 空指针和参数校验测试
71TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_ChannelHandlesNullptr_Return_HCCL_E_PTR)71TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_ChannelHandlesNullptr_Return_HCCL_E_PTR)
72{72{
73+ HcommChannelDesc hcommDescs[1] = {};
73 ChannelHandle hostHandles[1] = {};74 ChannelHandle hostHandles[1] = {};
74 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(75 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(
75- nullptr, hostHandles, 1, "test_tag", nullptr, "test_kernel", false);76+ nullptr, hostHandles, hcommDescs, 1, "test_tag", nullptr, "test_kernel", false);
76 EXPECT_EQ(ret, HCCL_E_PTR);77 EXPECT_EQ(ret, HCCL_E_PTR);
77}78}
78 79 
79TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_HostChannelHandlesNullptr_Return_HCCL_E_PTR)80TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_HostChannelHandlesNullptr_Return_HCCL_E_PTR)
80{81{
82+ HcommChannelDesc hcommDescs[1] = {};
81 ChannelHandle deviceHandles[1] = {};83 ChannelHandle deviceHandles[1] = {};
82 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(84 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(
83- deviceHandles, nullptr, 1, "test_tag", nullptr, "test_kernel", false);85+ deviceHandles, nullptr, hcommDescs, 1, "test_tag", nullptr, "test_kernel", false);
84 EXPECT_EQ(ret, HCCL_E_PTR);86 EXPECT_EQ(ret, HCCL_E_PTR);
85}87}
86 88 
87TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_ListNumZero_Return_HCCL_E_PARA)89TEST_F(TestChannelProcess, Ut_TestLaunchChannelKernelCommon_When_ListNumZero_Return_HCCL_E_PARA)
88{90{
91+ HcommChannelDesc hcommDescs[1] = {};
89 ChannelHandle deviceHandles[1] = {};92 ChannelHandle deviceHandles[1] = {};
90 ChannelHandle hostHandles[1] = {};93 ChannelHandle hostHandles[1] = {};
91 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(94 HcclResult ret = hcomm::ChannelProcess::LaunchChannelKernelCommon(
92- deviceHandles, hostHandles, 0, "test_tag", nullptr, "test_kernel", false);95+ deviceHandles, hostHandles, hcommDescs, 0, "test_tag", nullptr, "test_kernel", false);
93 EXPECT_EQ(ret, HCCL_E_PARA);96 EXPECT_EQ(ret, HCCL_E_PARA);
94}97}
95 98 
96// SaveChannels 空指针和参数校验测试99// SaveChannels 空指针和参数校验测试
97TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_TargetChannelsNullptr_Return_HCCL_E_PTR)100TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_TargetChannelsNullptr_Return_HCCL_E_PTR)
98{101{
102+ HcommChannelDesc hcommDescs[1] = {};
99 ChannelHandle userChannels[1] = {};103 ChannelHandle userChannels[1] = {};
100 HcclResult ret = hcomm::ChannelProcess::SaveChannels(104 HcclResult ret = hcomm::ChannelProcess::SaveChannels(
101- nullptr, userChannels, 1, COMM_ENGINE_AICPU, nullptr);105+ nullptr, userChannels, hcommDescs, 1, COMM_ENGINE_AICPU, nullptr);
102 EXPECT_EQ(ret, HCCL_E_PTR);106 EXPECT_EQ(ret, HCCL_E_PTR);
103}107}
104 108 
105TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_UserChannelsNullptr_Return_HCCL_E_PTR)109TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_UserChannelsNullptr_Return_HCCL_E_PTR)
106{110{
111+ HcommChannelDesc hcommDescs[1] = {};
107 ChannelHandle targetChannels[1] = {};112 ChannelHandle targetChannels[1] = {};
108 HcclResult ret = hcomm::ChannelProcess::SaveChannels(113 HcclResult ret = hcomm::ChannelProcess::SaveChannels(
109- targetChannels, nullptr, 1, COMM_ENGINE_AICPU, nullptr);114+ targetChannels, nullptr, hcommDescs, 1, COMM_ENGINE_AICPU, nullptr);
110 EXPECT_EQ(ret, HCCL_E_PTR);115 EXPECT_EQ(ret, HCCL_E_PTR);
111}116}
112 117 
113TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_ChannelNumZero_Return_HCCL_E_PARA)118TEST_F(TestChannelProcess, Ut_TestSaveChannels_When_ChannelNumZero_Return_HCCL_E_PARA)
114{119{
120+ HcommChannelDesc hcommDescs[1] = {};
115 ChannelHandle targetChannels[1] = {};121 ChannelHandle targetChannels[1] = {};
116 ChannelHandle userChannels[1] = {};122 ChannelHandle userChannels[1] = {};
117 HcclResult ret = hcomm::ChannelProcess::SaveChannels(123 HcclResult ret = hcomm::ChannelProcess::SaveChannels(
118- targetChannels, userChannels, 0, COMM_ENGINE_AICPU, nullptr);124+ targetChannels, userChannels, hcommDescs, 0, COMM_ENGINE_AICPU, nullptr);
119 EXPECT_EQ(ret, HCCL_E_PARA);125 EXPECT_EQ(ret, HCCL_E_PARA);
120}126}
121 127 
@@ -44,8 +44,8 @@ TEST_F(TestEndpointPair, Ut_EndpointPair_Construct_Expect_HCCL_SUCCESS)
44 EXPECT_EQ(ret, HCCL_SUCCESS);44 EXPECT_EQ(ret, HCCL_SUCCESS);
45 45 
46 Hccl::Socket* socket = nullptr;46 Hccl::Socket* socket = nullptr;
47- ret = endpointPair.GetSocket(0, 1, "Hccl_Test_Group", 60001, 0, socket);47+ ret = endpointPair.GetSocket(0, 1, "Hccl_Test_Group", 60001, 0, socket, 0, 0);
48 EXPECT_EQ(ret, HCCL_SUCCESS);48 EXPECT_EQ(ret, HCCL_SUCCESS);
49- ret = endpointPair.GetSocket(1, 0, "Hccl_Test_Group", 60001, 0, socket);49+ ret = endpointPair.GetSocket(1, 0, "Hccl_Test_Group", 60001, 0, socket, 0, 0);
50 EXPECT_EQ(ret, HCCL_SUCCESS);50 EXPECT_EQ(ret, HCCL_SUCCESS);
51}51}
@@ -96,11 +96,15 @@ set(orion_list
96 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/buffer.cc96 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/buffer.cc
97 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rma_buf_slice_lite.cc97 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rma_buf_slice_lite.cc
98 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/aicpu/notify_lite.cc98 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/aicpu/notify_lite.cc
99+ ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/rts_notify.cc
100+ ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/ipc_remote_notify.cc
99 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rma_buffer_lite.cc101 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rma_buffer_lite.cc
100 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/rma_conn_lite.cc102 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/rma_conn_lite.cc
101 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rmt_rma_buf_slice_lite.cc103 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/buffer/aicpu/rmt_rma_buf_slice_lite.cc
102 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/ub_conn_lite.cc104 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/ub_conn_lite.cc
103 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/ub_conn_lite_mgr.cc105 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/connection/aicpu/ub_conn_lite_mgr.cc
106+ ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/stream/stream.cc
107+ ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/task/task.cc
104 ${HCCL_CODE_ROOT}/src/legacy/interface/rank_graph_interface.cc108 ${HCCL_CODE_ROOT}/src/legacy/interface/rank_graph_interface.cc
105 ${HCCL_CODE_ROOT}/src/legacy/framework/topo/new_topo_builder/rank_graph/rank_graph.cc109 ${HCCL_CODE_ROOT}/src/legacy/framework/topo/new_topo_builder/rank_graph/rank_graph.cc
106 ${HCCL_CODE_ROOT}/src/legacy/framework/topo/new_topo_builder/rank_graph/net_instance.cc110 ${HCCL_CODE_ROOT}/src/legacy/framework/topo/new_topo_builder/rank_graph/net_instance.cc
@@ -330,6 +334,7 @@ target_include_directories(hccl_llt PRIVATE
330 ${HCCL_CODE_ROOT}/src/legacy/framework/communicator/aicpu/inc334 ${HCCL_CODE_ROOT}/src/legacy/framework/communicator/aicpu/inc
331 ${HCCL_CODE_ROOT}/src/framework/next/comms/endpoint_pairs/sockets335 ${HCCL_CODE_ROOT}/src/framework/next/comms/endpoint_pairs/sockets
332 ${HCCL_CODE_ROOT}/src/legacy/unified_platform336 ${HCCL_CODE_ROOT}/src/legacy/unified_platform
337+ ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/
333 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/aicpu338 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/notify/aicpu
334 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/mem339 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/mem
335 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/mem340 ${HCCL_CODE_ROOT}/src/legacy/unified_platform/resource/mem
@@ -81,7 +81,7 @@ RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const RmaBufferLit
81 return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId() , lite.GetTokenValue());81 return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId() , lite.GetTokenValue());
82}82}
83 83 
84-HcclResult UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite) const84+HcclResult UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite)
85{85{
86 return HCCL_SUCCESS;86 return HCCL_SUCCESS;
87}87}
@@ -111,6 +111,12 @@
111#include "../../../legacy/unified_platform/ccu/ccu_microcode/ccu_assist.h"111#include "../../../legacy/unified_platform/ccu/ccu_microcode/ccu_assist.h"
112#include "acl/acl_rt.h"112#include "acl/acl_rt.h"
113 113 
114+#include "p2p_transport.h"
115+#include "dev_capability.h"
116+#include "p2p_connection.h"
117+#include "rts_cnt_notify.h"
118+#include "rts_1ton_cnt_notify.h"
119+#include "ipc_local_notify.h"
114 120 
115namespace Hccl {121namespace Hccl {
116 122 
@@ -954,10 +960,10 @@ void UbLocalNotify::ReleaseResource() const
954UbLocalNotify::~UbLocalNotify()960UbLocalNotify::~UbLocalNotify()
955{}961{}
956 962 
957-RtsNotify::RtsNotify(bool devUsed)963+// RtsNotify::RtsNotify(bool devUsed)
958-{}964+// {}
959-RtsNotify::~RtsNotify()965+// RtsNotify::~RtsNotify()
960-{}966+// {}
961 967 
962UbMemTransport::UbMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,968UbMemTransport::UbMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
963 const Socket &socket, RdmaHandle rdmaHandle1, LocCntNotifyRes &locCntNotifyRes1, bool isRecvFirst)969 const Socket &socket, RdmaHandle rdmaHandle1, LocCntNotifyRes &locCntNotifyRes1, bool isRecvFirst)
@@ -2261,6 +2267,243 @@ std::vector<CommunicatorImplLite *> CommunicatorImplLiteMgr::GetAll()
2261{2267{
2262 return {};2268 return {};
2263}2269}
2270+ 
2271+RtNotify_t HrtIpcOpenNotifyWithFlag(const char_t *name, uint32_t flags)
2272+{
2273+ return nullptr;
2274+}
2275+ 
2276+u32 HrtStreamGetCqId(const aclrtStream ptr)
2277+{
2278+ return 0;
2279+}
2280+ 
2281+void HrtNotifyDestroy(RtNotify_t ptr)
2282+{
2283+ return ;
2284+}
2285+ 
2286+s32 HrtGetStreamId(aclrtStream ptr)
2287+{
2288+ return 0;
2289+}
2290+ 
2291+aclrtStream HrtStreamCreateWithFlags(uint32_t priority, uint32_t flag)
2292+{
2293+ static aclrtStream stream;
2294+ return stream;
2295+}
2296+ 
2297+u32 HrtNotifyGetOffset(RtNotify_t ptr)
2298+{
2299+ return 0;
2300+}
2301+ 
2302+u32 HrtGetNotifyID(RtNotify_t notifyHandle)
2303+{
2304+ return 0;
2305+}
2306+ 
2307+s32 HrtDeviceGetBareTgid()
2308+{
2309+ return 0;
2310+}
2311+ 
2312+void HrtSetIpcNotifyPid(aclrtNotify notify, int32_t pid)
2313+{
2314+ return ;
2315+}
2316+ 
2317+void HrtStreamDestroy(aclrtStream ptr)
2318+{
2319+ return ;
2320+}
2321+ 
2322+void HrtIpcSetNotifyName(RtNotify_t ptr, char_t *name, uint32_t len)
2323+{
2324+ return ;
2325+}
2326+ 
2327+aclrtNotify HrtNotifyCreateWithFlag(u32 devId, u32 flag)
2328+{
2329+ static aclrtNotify notify;
2330+ return notify;
2331+}
2332+ 
2333+aclrtNotify HrtNotifyCreate(s32 deviceLogicId)
2334+{
2335+ static aclrtNotify notify;
2336+ return notify;
2337+}
2338+ 
2339+u64 HrtNotifyGetAddr(RtNotify_t notifyHandle)
2340+{
2341+ return 0;
2342+}
2343+ 
2344+RtNotify_t HrtIpcOpenNotify(const char_t *name)
2345+{
2346+ return nullptr;
2347+}
2348+ 
2349+u32 HrtStreamGetSqId(const aclrtStream ptr)
2350+{
2351+ return 0;
2352+}
2353+ 
2354+void HrtNotifyRecord(RtNotify_t notifyPtr, aclrtStream streamPtr)
2355+{
2356+ return ;
2357+}
2358+ 
2359+void HrtNotifyWaitWithTimeOut(RtNotify_t notifyPtr, aclrtStream streamPtr, uint32_t timeOut)
2360+{
2361+ return ;
2362+}
2363+ 
2364+P2PTransport::P2PTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
2365+ const Socket &socket)
2366+ : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::P2P)
2367+{
2368+}
2369+ 
2370+HcclResult P2PTransport::GetRemoteMem(HcclMem **remoteMem, uint32_t *memNum, char **memTags)
2371+{
2372+ return HCCL_SUCCESS;
2373+}
2374+ 
2375+std::vector<char> P2PTransport::GetUniqueIdV2()
2376+{
2377+ return {};
2378+}
2379+ 
2380+std::string P2PTransport::Describe() const
2381+{
2382+ return "";
2383+}
2384+ 
2385+TransportStatus P2PTransport::GetStatus()
2386+{
2387+ return TransportStatus::READY;
2388+}
2389+ 
2390+std::vector<char> P2PTransport::GetUniqueId()
2391+{
2392+ return {};
2393+}
2394+ 
2395+void P2PTransport::Post(u32 index, const Stream &stream)
2396+{
2397+ return ;
2398+}
2399+ 
2400+void P2PTransport::Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
2401+{
2402+ return ;
2403+}
2404+ 
2405+void P2PTransport::ReadReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const ReduceIn &reduceIn,
2406+ const Stream &stream)
2407+{
2408+ return ;
2409+}
2410+ 
2411+void P2PTransport::Write(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
2412+{
2413+ return ;
2414+}
2415+ 
2416+void P2PTransport::WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const ReduceIn &reduceIn,
2417+ const Stream &stream)
2418+{
2419+ return ;
2420+}
2421+ 
2422+DevCapability::DevCapability()
2423+{
2424+}
2425+DevCapability &DevCapability::GetInstance()
2426+{
2427+ static DevCapability devCapability;
2428+ return devCapability;
2429+}
2430+ 
2431+P2PConnection::P2PConnection(Socket *socket, const std::string &tag)
2432+ : RmaConnection(socket, RmaConnType::P2P)
2433+{
2434+}
2435+ 
2436+void P2PConnection::Connect()
2437+{
2438+ return ;
2439+}
2440+RmaConnStatus P2PConnection::GetStatus()
2441+{
2442+ return RmaConnStatus::READY;
2443+}
2444+string P2PConnection::Describe() const
2445+{
2446+ return "";
2447+}
2448+ 
2449+unique_ptr<BaseTask> P2PConnection::PrepareRead(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
2450+ const SqeConfig &config)
2451+{
2452+ return nullptr;
2453+}
2454+ 
2455+unique_ptr<BaseTask> P2PConnection::PrepareReadReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
2456+ DataType datatype, ReduceOp reduceOp, const SqeConfig &config)
2457+{
2458+ return nullptr;
2459+}
2460+ 
2461+unique_ptr<BaseTask> P2PConnection::PrepareWrite(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
2462+ const SqeConfig &config)
2463+{
2464+ return nullptr;
2465+}
2466+ 
2467+unique_ptr<BaseTask> P2PConnection::PrepareWriteReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
2468+ DataType datatype, ReduceOp reduceOp, const SqeConfig &config)
2469+{
2470+ return nullptr;
2471+}
2472+ 
2473+std::string RtsCntNotify::Describe() const
2474+{
2475+ return "";
2476+}
2477+ 
2478+std::string Rts1ToNCntNotify::Describe() const
2479+{
2480+ return "";
2481+}
2482+ 
2483+IpcLocalNotify::IpcLocalNotify(bool devUsed) : BaseLocalNotify(RmaType::IPC, devUsed)
2484+{
2485+}
2486+ 
2487+void IpcLocalNotify::Wait(const Stream &stream, u32 timeout) const
2488+{
2489+ return ;
2490+}
2491+ 
2492+void IpcLocalNotify::Post(const Stream &stream) const
2493+{
2494+ return ;
2495+}
2496+ 
2497+std::unique_ptr<Serializable> IpcLocalNotify::GetExchangeDto()
2498+{
2499+ return nullptr;
2500+}
2501+ 
2502+string IpcLocalNotify::Describe() const
2503+{
2504+ return "";
2505+}
2506+ 
2264} // namespace Hccl2507} // namespace Hccl
2265 2508 
2266HcclResult HcclCommDestroyV2(HcclComm comm)2509HcclResult HcclCommDestroyV2(HcclComm comm)
@@ -2287,4 +2530,3 @@ HcclResult HcclGetRankGraphV2(HcclComm *comm, void **rankGraph)
2287{2530{
2288 return HCCL_SUCCESS;2531 return HCCL_SUCCESS;
2289}2532}
2290-