Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "channel_aicpu_mgr.h"
12 : #include "ub_transport_lite_impl.h"
13 : #include "p2p_transport_lite_impl.h"
14 : #include "roce_transport_lite_impl.h"
15 : #include "aicpu_res_package_helper.h"
16 : #include "aicpu_indop_env.h"
17 : #include "aicpu_task_cache_manager.h"
18 : #include "adapter_hal_pub.h"
19 : #include "log.h"
20 :
21 10 : ChannelAicpuMgr::ChannelAicpuMgr(HcclCommDfxLite& dfx, const HcclTopoInfo& topoInfo) : dfx_(dfx), topoInfo_(topoInfo) {}
22 :
23 2 : HcclResult ChannelAicpuMgr::AllocChannelResource(HcclChannelUrmaRes* commParam)
24 : {
25 2 : CHK_PTR_NULL(commParam);
26 1 : HCCL_INFO(
27 : "[ChannelAicpuMgr][%s] deviceLogicId[%d], devicePhyId[%u], deviceType[%d], commParam->channelList[%p], "
28 : "commParam->listNum[%u], commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]",
29 : __func__, topoInfo_.deviceLogicId, topoInfo_.devicePhyId, topoInfo_.deviceType, commParam->channelList,
30 : commParam->listNum, commParam->uniqueIdAddr, commParam->uniqueIdSize);
31 1 : CHK_RET(InitUrmaChannel(commParam));
32 0 : return HCCL_SUCCESS;
33 : }
34 :
35 2 : HcclResult ChannelAicpuMgr::InitUrmaChannel(HcclChannelUrmaRes* commParam) { return ProcessUrmaRes(commParam, true); }
36 :
37 3 : HcclResult ChannelAicpuMgr::ProcessUrmaRes(HcclChannelUrmaRes* commParam, bool isInit)
38 : {
39 3 : HCCL_INFO(
40 : "[ChannelAicpuMgr][%s] commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]", __func__,
41 : commParam->uniqueIdAddr, commParam->uniqueIdSize);
42 3 : ChannelHandle* channelList = reinterpret_cast<ChannelHandle*>(commParam->channelList);
43 3 : u8* currentSrcAddr = reinterpret_cast<u8*>(commParam->uniqueIdAddr);
44 3 : u32* addSize = reinterpret_cast<u32*>(commParam->channelSizeAddr);
45 3 : CHK_PTR_NULL(channelList);
46 0 : CHK_PTR_NULL(currentSrcAddr);
47 0 : CHK_PTR_NULL(addSize);
48 :
49 0 : for (u32 index = 0; index < commParam->listNum; index++) {
50 0 : std::vector<char> data(*addSize);
51 :
52 0 : CHK_SAFETY_FUNC_RET(memcpy_s(data.data(), data.size(), currentSrcAddr, *addSize));
53 0 : currentSrcAddr += *addSize;
54 0 : addSize++;
55 : Hccl::AicpuResPackageHelper helper;
56 0 : auto dataVec = helper.ParsePackedData(data);
57 :
58 0 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
59 0 : if (static_cast<u32>(resType) >= dataVec.size()) {
60 0 : HCCL_ERROR("[ChannelAicpuMgr][%s] fail, resType[%d], dataVec size[%zu]", __func__, resType, dataVec.size());
61 0 : return HCCL_E_PARA;
62 : }
63 :
64 0 : ChannelHandle channelHandle{0};
65 0 : if (isInit) {
66 0 : CHK_RET(ParsePackData(dataVec[resType].data, channelHandle));
67 0 : channelList[index] = channelHandle;
68 0 : CHK_RET(RegisterChannelCacheCallback(channelHandle));
69 0 : dfx_.AddChannelRemoteRankId(channelHandle, commParam->remoteRankList[index]);
70 : } else {
71 0 : channelHandle = channelList[index];
72 0 : if (!transportMap_.count(channelHandle)) {
73 0 : HCCL_ERROR("[ChannelAicpuMgr][%s] fail, resType[%d], current ChannelHandle nullptr", __func__, resType);
74 0 : return HCCL_E_PARA;
75 : }
76 0 : CHK_RET(ResumePackData(dataVec[resType].data, channelHandle));
77 : }
78 :
79 0 : HCCL_INFO(
80 : "[ChannelAicpuMgr][%s] index[%u], currentSrcAddr[%p], channelSizeAddr[%p], channelHandle[0x%llx]", __func__,
81 : index, currentSrcAddr, commParam->channelSizeAddr, static_cast<unsigned long long>(channelHandle));
82 0 : }
83 :
84 0 : return HCCL_SUCCESS;
85 : }
86 :
87 : namespace {
88 : template <typename T>
89 0 : HcclResult CreateAndInsertTransport(
90 : std::vector<char>& uniqueId, ChannelHandle& handle, T*& outPtr,
91 : std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::BaseTransportLiteImpl>>& transportMap)
92 : {
93 0 : std::unique_ptr<T> impl;
94 0 : EXCEPTION_CATCH(impl = std::make_unique<T>(uniqueId), return HCCL_E_PTR);
95 0 : CHK_SMART_PTR_NULL(impl);
96 0 : outPtr = impl.get();
97 0 : handle = reinterpret_cast<uint64_t>(impl.get());
98 0 : transportMap.insert({handle, std::move(impl)});
99 0 : return HCCL_SUCCESS;
100 0 : }
101 : } // namespace
102 :
103 0 : HcclResult ChannelAicpuMgr::ParsePackData(std::vector<char>& data, ChannelHandle& handle)
104 : {
105 0 : HCCL_DEBUG("[ChannelAicpuMgr][%s] data: ptr[%p], size[%zu]", __func__, data.data(), data.size());
106 0 : Hccl::BinaryStream binaryStream(data);
107 :
108 0 : std::vector<char> transpUniqueId;
109 0 : binaryStream >> transpUniqueId;
110 :
111 0 : Hccl::BinaryStream binaryStreamForType(transpUniqueId);
112 : u32 transType;
113 0 : binaryStreamForType >> transType;
114 0 : HCCL_INFO("[ChannelAicpuMgr][ParsePackData] transType[%u]", transType);
115 :
116 0 : if (transType == Hccl::TransportType::UB) {
117 0 : Hccl::UbTransportLiteImpl* ubPtr = nullptr;
118 0 : CHK_RET(CreateAndInsertTransport<Hccl::UbTransportLiteImpl>(transpUniqueId, handle, ubPtr, transportMap_));
119 0 : ubPtr->SetTaskExceptionEnable(hcomm::GetTaskExceptionEnable());
120 0 : } else if (transType == Hccl::TransportType::P2P) {
121 0 : Hccl::P2PTransportLiteImpl* p2pPtr = nullptr;
122 0 : CHK_RET(CreateAndInsertTransport<Hccl::P2PTransportLiteImpl>(transpUniqueId, handle, p2pPtr, transportMap_));
123 0 : } else if (transType == Hccl::TransportType::ROCE) {
124 0 : Hccl::RoceTransportLiteImpl* rocePtr = nullptr;
125 0 : CHK_RET(CreateAndInsertTransport<Hccl::RoceTransportLiteImpl>(transpUniqueId, handle, rocePtr, transportMap_));
126 : } else {
127 0 : HCCL_ERROR("[ChannelAicpuMgr][ParsePackData] unsupported transportType[%u]", transType);
128 0 : return HCCL_E_INTERNAL;
129 : }
130 :
131 0 : return HCCL_SUCCESS;
132 0 : }
133 :
134 1 : HcclResult ChannelAicpuMgr::ResumePackData(std::vector<char>& data, ChannelHandle& handle)
135 : {
136 1 : Hccl::BinaryStream binaryStream(data);
137 1 : std::vector<char> transpUniqueId;
138 1 : binaryStream >> transpUniqueId;
139 :
140 1 : auto it = transportMap_.find(handle);
141 1 : CHK_PRT_RET(
142 : it == transportMap_.end(),
143 : HCCL_ERROR("[ChannelAicpuMgr][ResumePackData] channel handle[0x%llx] not found", handle), HCCL_E_PARA);
144 :
145 0 : auto* ub = dynamic_cast<Hccl::UbTransportLiteImpl*>(it->second.get());
146 0 : CHK_PRT_RET(
147 : ub == nullptr,
148 : HCCL_ERROR("[ChannelAicpuMgr][ResumePackData] transport is not UB type for handle[0x%llx]", handle),
149 : HCCL_E_INTERNAL);
150 0 : return ub->Resume(transpUniqueId);
151 1 : }
152 :
153 1 : HcclResult ChannelAicpuMgr::Resume(HcclChannelUrmaRes* commParam)
154 : {
155 1 : CHK_PTR_NULL(commParam);
156 0 : CHK_RET(ProcessUrmaRes(commParam, false));
157 0 : return HCCL_SUCCESS;
158 : }
159 :
160 0 : HcclResult ChannelAicpuMgr::RegisterChannelCacheCallback(ChannelHandle channel)
161 : {
162 : // 目前aicpu task cache只支持UB.URMA协议，从合并后的transportMap_中查找UB条目
163 0 : auto it = transportMap_.find(channel);
164 0 : if (it != transportMap_.end()) {
165 0 : auto* ub = dynamic_cast<Hccl::UbTransportLiteImpl*>(it->second.get());
166 0 : if (ub != nullptr) {
167 0 : HCCL_INFO(
168 : "[ChannelAicpuMgr][RegisterChannelCacheCallback] register cache callback for channel[0x%016llx]",
169 : channel);
170 0 : CHK_RET(ub->SetNeedCacheTaskCallback(hcomm::AicpuTaskCacheManager::NeedCacheTask));
171 0 : CHK_RET(ub->SetAddWqeArrayCallback(hcomm::AicpuTaskCacheManager::AddWqeArray));
172 0 : HCCL_INFO(
173 : "[ChannelAicpuMgr][RegisterChannelCacheCallback] register cache callback for channel[0x%016llx] "
174 : "success",
175 : channel);
176 : }
177 : }
178 0 : return HCCL_SUCCESS;
179 : }
180 :
181 1 : HcclResult ChannelAicpuMgr::Clean()
182 : {
183 1 : for (auto& impl_pair : transportMap_) {
184 0 : auto& impl = impl_pair.second;
185 0 : auto* ub = dynamic_cast<Hccl::UbTransportLiteImpl*>(impl.get());
186 0 : if (ub != nullptr) {
187 0 : CHK_RET(ub->Clean());
188 : }
189 : }
190 1 : HCCL_INFO("[%s][ChannelAicpuMgr]Clean() finished", __func__);
191 1 : return HCCL_SUCCESS;
192 : }
|
