Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "aicpu_channel_process.h"
12 : #include "dev_aicpu_ts_channel_mgr.h"
13 : #include "aicpu_res_package_helper.h"
14 : #include "../channel.h"
15 : #include "aicpu_ts_channel_helper.h"
16 : #include "ub_transport_lite_impl.h"
17 : #include "roce_transport_lite_impl.h"
18 : #include "p2p_transport_lite_impl.h"
19 : #include "aicpu_task_cache_manager.h"
20 :
21 : #include "adapter_rts_common.h"
22 : #include "log.h"
23 :
24 : #include <securec.h>
25 :
26 : #include <cstdint>
27 : #include <vector>
28 :
29 : std::mutex AicpuChannelProcess::mutex_;
30 : std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::BaseTransportLiteImpl>> AicpuChannelProcess::transportMap_;
31 :
32 : namespace {
33 : // 创建 lite transport 并登记进 transportMap_（handle 即 transport 指针，primitives 层按指针解引用）。
34 : // ROCE 无额外配置走这里；UB 需在入表前设 cache callback，故单独内联。
35 : // 注意：本函数不加锁，调用方必须已持有 AicpuChannelProcess::mutex_。
36 : template <typename T>
37 4 : HcclResult CreateAndInsertTransport(
38 : std::vector<char>& uniqueId, ChannelHandle& handle,
39 : std::unordered_map<ChannelHandle, std::unique_ptr<Hccl::BaseTransportLiteImpl>>& transportMap)
40 : {
41 4 : std::unique_ptr<T> impl;
42 4 : EXCEPTION_CATCH(impl = std::make_unique<T>(uniqueId), return HCCL_E_PTR);
43 4 : CHK_SMART_PTR_NULL(impl);
44 4 : handle = reinterpret_cast<uint64_t>(impl.get());
45 4 : transportMap.insert({handle, std::move(impl)});
46 4 : return HCCL_SUCCESS;
47 4 : }
48 : } // namespace
49 :
50 6 : HcclResult AicpuChannelProcess::ParsePackData(std::vector<char>& data, ChannelHandle& handle)
51 : {
52 6 : HCCL_DEBUG("[HcclCommAicpu][%s] data: ptr[%p], size[%u]", __func__, data.data(), data.size());
53 6 : Hccl::BinaryStream binaryStream(data);
54 :
55 6 : std::vector<char> transpUniqueId;
56 6 : binaryStream >> transpUniqueId;
57 :
58 6 : Hccl::BinaryStream binaryStreamForType(transpUniqueId);
59 : u32 transType;
60 6 : binaryStreamForType >> transType;
61 6 : HCCL_INFO("[CollCommAicpu][ParsePackData] transType[%u]", transType);
62 6 : if (transType == Hccl::TransportType::UB || transType == Hccl::TransportType::UBoE) {
63 1 : std::unique_ptr<Hccl::UbTransportLiteImpl> ubTransportLiteImpl;
64 1 : EXCEPTION_CATCH(
65 : (ubTransportLiteImpl = std::make_unique<Hccl::UbTransportLiteImpl>(transpUniqueId)), return HCCL_E_PTR);
66 1 : CHK_SMART_PTR_NULL(ubTransportLiteImpl);
67 1 : CHK_RET(ubTransportLiteImpl->SetNeedCacheTaskCallback(hcomm::AicpuTaskCacheManager::NeedCacheTask));
68 1 : CHK_RET(ubTransportLiteImpl->SetAddWqeArrayCallback(hcomm::AicpuTaskCacheManager::AddWqeArray));
69 1 : handle = reinterpret_cast<uint64_t>(ubTransportLiteImpl.get());
70 1 : transportMap_.insert({handle, std::move(ubTransportLiteImpl)});
71 7 : } else if (transType == Hccl::TransportType::ROCE) {
72 3 : CHK_RET(CreateAndInsertTransport<Hccl::RoceTransportLiteImpl>(transpUniqueId, handle, transportMap_));
73 2 : } else if (transType == Hccl::TransportType::P2P) {
74 1 : CHK_RET(CreateAndInsertTransport<Hccl::P2PTransportLiteImpl>(transpUniqueId, handle, transportMap_));
75 : } else {
76 1 : HCCL_ERROR("[AicpuChannelProcess][%s] transType[%u] is invalid", __func__, transType);
77 1 : return HCCL_E_PARA;
78 : }
79 :
80 5 : return HCCL_SUCCESS;
81 6 : }
82 :
83 0 : HcclResult AicpuChannelProcess::InitUrmaChannel(HcclChannelUrmaRes* commParam)
84 : {
85 0 : HCCL_INFO(
86 : "[HcclCommAicpu][%s] commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]", __func__,
87 : commParam->uniqueIdAddr, commParam->uniqueIdSize);
88 :
89 0 : u8* currentSrcAddr = reinterpret_cast<u8*>(commParam->uniqueIdAddr);
90 0 : u32* addSize = reinterpret_cast<u32*>(commParam->channelSizeAddr);
91 0 : for (u32 index = 0; index < commParam->listNum; index++) {
92 0 : std::vector<char> data(*addSize);
93 :
94 0 : CHK_SAFETY_FUNC_RET(memcpy_s(data.data(), data.size(), currentSrcAddr, *addSize));
95 0 : currentSrcAddr += *addSize;
96 0 : addSize++;
97 : Hccl::AicpuResPackageHelper helper;
98 0 : auto dataVec = helper.ParsePackedData(data);
99 :
100 0 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
101 0 : if (static_cast<u32>(resType) >= dataVec.size()) {
102 0 : HCCL_ERROR("[HcclCommAicpu][%s] fail, resType[%d], dataVec size[%u]", __func__, resType, dataVec.size());
103 0 : return HCCL_E_PARA;
104 : }
105 : ChannelHandle channelHandle;
106 0 : CHK_RET(ParsePackData(dataVec[resType].data, channelHandle));
107 :
108 0 : if (commParam->ctxList != nullptr) {
109 : // ctx模式：device侧填充abiHeader + deviceChannel
110 0 : auto** ctxList = reinterpret_cast<HcommAicpuChannelCtx**>(commParam->ctxList);
111 0 : ctxList[index]->abiHeader.version = HCOMM_AICPU_CHANNEL_CTX_VERSION;
112 0 : ctxList[index]->abiHeader.magicWord = HCOMM_AICPU_CHANNEL_CTX_MAGIC_WORD;
113 0 : ctxList[index]->abiHeader.size = sizeof(HcommAicpuChannelCtx);
114 0 : ctxList[index]->deviceChannel = reinterpret_cast<void*>(channelHandle);
115 : } else {
116 0 : ChannelHandle* channelList = reinterpret_cast<ChannelHandle*>(commParam->channelList);
117 0 : channelList[index] = channelHandle;
118 : }
119 0 : HCCL_INFO(
120 : "[HcclCommAicpu][%s] index[%u], currentSrcAddr[%p], channelSizeAddr[%p], channelHandle[0x%llx]", __func__,
121 : index, currentSrcAddr, commParam->channelSizeAddr, channelHandle);
122 0 : }
123 :
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 0 : HcclResult AicpuChannelProcess::AicpuChannelInit(HcclChannelUrmaRes* commParam)
128 : {
129 0 : HCCL_INFO(
130 : "[AicpuChannelProcess][%s] commParam->channelList[%p], commParam->listNum[%u], commParam->uniqueIdAddr[%p], "
131 : "commParam->uniqueIdSize[%u]",
132 : __func__, commParam->channelList, commParam->listNum, commParam->uniqueIdAddr, commParam->uniqueIdSize);
133 :
134 0 : CHK_RET(hrtSetWorkModeAicpu(true));
135 0 : CHK_RET(hrtSetlocalDevice(commParam->deviceLogicId));
136 0 : CHK_RET(hrtSetlocalDeviceType(static_cast<DevType>(commParam->deviceType)));
137 :
138 0 : std::lock_guard<std::mutex> addLock(mutex_);
139 :
140 0 : HcclResult ret = InitUrmaChannel(commParam);
141 0 : CHK_PRT_RET(
142 : ret != HCCL_SUCCESS,
143 : HCCL_ERROR(
144 : "[AicpuChannelProcess][AicpuChannelInit]errNo[0x%016llx] Failed to init channels", HCCL_ERROR_CODE(ret)),
145 : ret);
146 :
147 0 : HCCL_INFO("[AicpuChannelProcess][%s] aicpuTask End.", __func__);
148 0 : return HCCL_SUCCESS;
149 0 : }
150 :
151 : namespace {
152 :
153 1 : void RollbackDestroy(DevAicpuTsChannelMgr& mgr, const std::vector<ChannelHandle>& rollback)
154 : {
155 1 : for (const auto& h : rollback) {
156 0 : if (mgr.DestroyChannel(h)) {
157 0 : HCCL_DEBUG("[AicpuChannelProcess][%s] rollback destroyed handle[0x%llx]", __func__, h);
158 : } else {
159 0 : HCCL_WARNING("[AicpuChannelProcess][%s] rollback failed to destroy handle[0x%llx]", __func__, h);
160 : }
161 : }
162 1 : }
163 :
164 2 : HcclResult CreateSingleHcommChannel(
165 : DevAicpuTsChannelMgr& mgr, void* dp, u64 sz, const HcommDeviceInfo& deviceInfo, hcomm::HcommChannelKind kind,
166 : HcommChannelRes* commParam, u32 index, ChannelHandle* channelList, std::vector<ChannelHandle>& rollback)
167 : {
168 2 : DevAicpuTsChannel* channel = mgr.GetOrCreateAicpuTsChannel(kind);
169 2 : if (channel == nullptr) {
170 1 : HCCL_ERROR(
171 : "[AicpuChannelProcess][%s] index[%u] unsupported kind[%u]", __func__, index, static_cast<uint32_t>(kind));
172 1 : RollbackDestroy(mgr, rollback);
173 1 : return HCCL_E_NOT_SUPPORT;
174 : }
175 1 : CHK_PTR_NULL(dp);
176 1 : ChannelHandle h{};
177 1 : HcclResult pret = channel->Create(dp, sz, deviceInfo, h);
178 1 : if (pret != HCCL_SUCCESS) {
179 0 : HCCL_ERROR("[AicpuChannelProcess][%s] parse fail at index[%u]", __func__, index);
180 0 : RollbackDestroy(mgr, rollback);
181 0 : return pret;
182 : }
183 1 : if (commParam->ctxList != nullptr) {
184 0 : auto** ctxList = reinterpret_cast<HcommAicpuChannelCtx**>(commParam->ctxList);
185 0 : ctxList[index]->abiHeader.version = HCOMM_AICPU_CHANNEL_CTX_VERSION;
186 0 : ctxList[index]->abiHeader.magicWord = HCOMM_AICPU_CHANNEL_CTX_MAGIC_WORD;
187 0 : ctxList[index]->abiHeader.size = sizeof(HcommAicpuChannelCtx);
188 0 : ctxList[index]->deviceChannel = reinterpret_cast<void*>(h);
189 : } else {
190 1 : channelList[index] = h;
191 : }
192 1 : rollback.push_back(h);
193 1 : return HCCL_SUCCESS;
194 : }
195 :
196 : } // namespace
197 :
198 4 : HcclResult AicpuChannelProcess::InitHcommChannelRes(HcommChannelRes* commParam)
199 : {
200 4 : CHK_PTR_NULL(commParam);
201 3 : HCCL_INFO(
202 : "[AicpuChannelProcess][%s] channelList[%p], listNum[%u]", __func__, commParam->channelList, commParam->listNum);
203 :
204 3 : CHK_PTR_NULL(commParam->channelList);
205 2 : CHK_PTR_NULL(commParam->channelDataListAddr);
206 2 : CHK_PTR_NULL(commParam->channelDataSizeListAddr);
207 2 : CHK_PTR_NULL(commParam->channelTypeListAddr);
208 :
209 2 : CHK_RET(hrtSetWorkModeAicpu(true));
210 2 : CHK_RET(hrtSetlocalDevice(commParam->deviceInfo.deviceLogicId));
211 2 : CHK_RET(hrtSetlocalDeviceType(static_cast<DevType>(commParam->deviceInfo.deviceType)));
212 :
213 2 : void** dataList = reinterpret_cast<void**>(commParam->channelDataListAddr);
214 2 : auto* sizeList = reinterpret_cast<u64*>(commParam->channelDataSizeListAddr);
215 2 : auto* typeList = reinterpret_cast<u32*>(commParam->channelTypeListAddr);
216 2 : auto* channelList = reinterpret_cast<ChannelHandle*>(commParam->channelList);
217 :
218 2 : auto& mgr = DevAicpuTsChannelMgr::Instance();
219 2 : std::vector<ChannelHandle> rollback;
220 2 : rollback.reserve(commParam->listNum);
221 :
222 2 : std::lock_guard<std::mutex> addLock(mutex_);
223 3 : for (u32 index = 0; index < commParam->listNum; ++index) {
224 2 : hcomm::HcommChannelKind kind = static_cast<hcomm::HcommChannelKind>(typeList[index]);
225 2 : CHK_RET(CreateSingleHcommChannel(
226 : mgr, dataList[index], sizeList[index], commParam->deviceInfo, kind, commParam, index, channelList,
227 : rollback));
228 1 : HCCL_INFO(
229 : "[AicpuChannelProcess][%s] index[%u] channelHandle[0x%llx]", __func__, index,
230 : commParam->ctxList != nullptr ? 0 : channelList[index]);
231 : }
232 :
233 1 : HCCL_INFO("[AicpuChannelProcess][%s] aicpu_task End.", __func__);
234 1 : return HCCL_SUCCESS;
235 2 : }
236 :
237 5 : HcclResult AicpuChannelProcess::AicpuChannelDestroy(HcclChannelUrmaRes* commParam)
238 : {
239 5 : HCCL_INFO(
240 : "[AicpuChannelProcess][%s] commParam->channelList[%p], commParam->listNum[%u]", __func__,
241 : commParam->channelList, commParam->listNum);
242 :
243 5 : auto& mgr = DevAicpuTsChannelMgr::Instance();
244 5 : std::lock_guard<std::mutex> addLock(mutex_);
245 :
246 5 : ChannelHandle* channelList = reinterpret_cast<ChannelHandle*>(commParam->channelList);
247 10 : for (u32 index = 0; index < commParam->listNum; ++index) {
248 5 : ChannelHandle handle = channelList[index];
249 :
250 5 : auto it = transportMap_.find(handle);
251 5 : if (it != transportMap_.end()) {
252 5 : transportMap_.erase(it);
253 5 : HCCL_DEBUG("[AicpuChannelProcess][%s] destroyed lite transport handle[0x%llx]", __func__, handle);
254 5 : continue;
255 : }
256 :
257 0 : if (mgr.DestroyChannel(handle)) {
258 0 : HCCL_DEBUG("[AicpuChannelProcess][%s] destroyed hcomm res handle[0x%llx]", __func__, handle);
259 0 : continue;
260 : }
261 :
262 0 : HCCL_WARNING(
263 : "[AicpuChannelProcess][%s] handle[0x%llx] not found in ub/hcomm maps, maybe already destroyed?", __func__,
264 : handle);
265 : }
266 :
267 5 : HCCL_INFO("[AicpuChannelProcess][%s] aicpu_task End.", __func__);
268 5 : return HCCL_SUCCESS;
269 5 : }
|
