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 <memory>
12 : #include <mutex>
13 : #include <unordered_map>
14 : #include <vector>
15 :
16 : #include "hcomm_c_adpt.h"
17 : #include "hcomm_c_adpt_common.h"
18 : #include "hcomm_thread_c_adpt.h"
19 : #include "hcomm_res.h"
20 : #include "hcomm_res_defs.h"
21 : #include "../hcomm_res_mgr.h"
22 : #include "log.h"
23 : #include "thread.h"
24 : #include "cpu_ts_thread.h"
25 : #include "param_check_pub.h"
26 : #include "comm_engine_utils.h"
27 : #include "exception_handler.h"
28 : #include "adapter_rts_common.h"
29 : #include "aicpu_ts_channel_helper.h"
30 : #include "aicpu_launch_manager.h"
31 :
32 : namespace hcomm {
33 : static std::unordered_map<ThreadHandle, std::shared_ptr<hccl::Thread>> g_ThreadMap;
34 : static std::mutex g_ThreadMapMtx;
35 : } // namespace hcomm
36 :
37 : using namespace hcomm;
38 :
39 : HcommResult
40 22 : HcommThreadAlloc(CommEngine engine, uint32_t threadNum, const uint32_t* notifyNumPerThread, ThreadHandle* threads)
41 : {
42 22 : CHK_PTR_NULL(threads);
43 21 : CHK_PTR_NULL(notifyNumPerThread);
44 21 : (void)HcommResMgrInit();
45 21 : const uint32_t notifyNum = notifyNumPerThread[0];
46 21 : if (threadNum > 1U) {
47 7 : HCCL_RUN_WARNING(
48 : "[%s] only notifyNumPerThread[0] is used currently, threadNum[%u], notifyNum[0][%u].", __func__, threadNum,
49 : notifyNum);
50 : }
51 21 : HCCL_INFO(
52 : "[%s] ThreadAcquire begin. engine[%s], threadNum[%u], notifyPerThread[%u], threads[%p]", __func__,
53 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum, threads);
54 21 : CHK_RET(RefreshCommEngineContext(engine));
55 :
56 : // 1. 参数校验
57 21 : CHK_RET(hccl::ValidateThreadParams(threadNum, notifyNum));
58 :
59 : // 2. 获取引擎对应的类型
60 : hccl::NotifyLoadType notifyLoadType;
61 : hccl::StreamType streamType;
62 18 : CHK_RET(hccl::CommEngineToNotifyLoadType(engine, notifyLoadType));
63 17 : CHK_RET(hccl::CommEngineToStreamType(engine, streamType));
64 :
65 : // 3. 创建线程
66 17 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
67 17 : hccl::ThreadCreateParams params(engine, threadNum, notifyNum, notifyLoadType, streamType);
68 17 : CHK_RET(hccl::CreateAndInitThreads(params, newThreads));
69 :
70 : // 4. 插入全局映射表
71 15 : CHK_RET(hccl::SaveThreads(newThreads));
72 :
73 : // 5. 储存线程句柄
74 15 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
75 15 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
76 :
77 15 : HCCL_INFO(
78 : "[HcommThreadAlloc] ThreadAcquire done: engine[%s] threadNum[%u], notifyPerThread[%u]",
79 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum);
80 15 : return HCCL_SUCCESS;
81 17 : }
82 :
83 14 : HcommResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread, ThreadHandle* threads)
84 : {
85 14 : return ::HcommThreadAlloc(engine, threadNum, ¬ifyNumPerThread, threads);
86 : }
87 :
88 0 : HcommResult HcommThreadAllocWithConfig(
89 : CommEngine engine, uint32_t threadNum, ThreadType type, const ThreadConfig* config, ThreadHandle* threads)
90 : {
91 0 : CHK_PTR_NULL(threads);
92 0 : CHK_PTR_NULL(config);
93 0 : CHK_PRT_RET(
94 : type == THREAD_TYPE_INVALID,
95 : HCCL_ERROR("[%s] thread type[%d] is invalid", __func__, static_cast<int32_t>(type)), (HcommResult)HCCL_E_PARA);
96 0 : CHK_PRT_RET(
97 : engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_CPU_TS,
98 : HCCL_ERROR(
99 : "[%s] commEngine[%d] CPU_TS/AICPU_TS not supported, use engine with ThreadType instead", __func__,
100 : static_cast<int32_t>(engine)),
101 : (HcommResult)HCCL_E_PARA);
102 0 : CHK_PRT_RET(
103 : engine == COMM_ENGINE_AIV || engine == COMM_ENGINE_CCU,
104 : HCCL_ERROR(
105 : "[%s] commEngine[%d] AIV/CCU not supported, supported engines: CPU/AICPU", __func__,
106 : static_cast<int32_t>(engine)),
107 : (HcommResult)HCCL_E_PARA);
108 0 : CHK_PRT_RET(
109 : threadNum == 0, HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), (HcommResult)HCCL_E_PARA);
110 0 : HcommResult hcommRet = HcommResMgrInit();
111 0 : CHK_PRT_RET(
112 : hcommRet != HCCL_SUCCESS,
113 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
114 0 : CHK_RET(RefreshCommEngineContext(engine));
115 :
116 0 : HCCL_INFO(
117 : "[%s] begin. engine[%d], threadType[%d], threadNum[%u], threads[%p]", __func__, engine,
118 : static_cast<int32_t>(type), threadNum, threads);
119 :
120 : hccl::NotifyLoadType notifyLoadType;
121 : hccl::StreamType streamType;
122 0 : CHK_RET(hccl::GetNotifyLoadType(engine, type, notifyLoadType));
123 0 : CHK_RET(hccl::GetStreamType(engine, type, streamType));
124 :
125 0 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
126 0 : newThreads.reserve(threadNum);
127 0 : for (uint32_t i = 0; i < threadNum; ++i) {
128 0 : CHK_PRT_RET(
129 : config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
130 : HCCL_ERROR(
131 : "[%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x], call ThreadConfigInit first", __func__, i,
132 : config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD),
133 : (HcommResult)HCCL_E_PARA);
134 0 : CHK_RET(hccl::ValidateThreadParams(1, config[i].notifyNumPerThread));
135 0 : std::shared_ptr<hccl::Thread> threadPtr;
136 : HcclResult ret
137 0 : = hccl::CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, threadPtr);
138 0 : CHK_PRT_RET(
139 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Failed to create thread at index[%u], ret[%d]", __func__, i, ret),
140 : (HcommResult)ret);
141 0 : ret = threadPtr->Init();
142 0 : CHK_PRT_RET(
143 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Failed to init thread at index[%u], ret[%d]", __func__, i, ret),
144 : (HcommResult)ret);
145 0 : newThreads.emplace_back(std::move(threadPtr));
146 0 : }
147 :
148 0 : CHK_RET(hccl::SaveThreads(newThreads));
149 0 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
150 0 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
151 :
152 0 : HCCL_INFO(
153 : "[%s] done: engine[%d] threadType[%d] threadNum[%u]", __func__, engine, static_cast<int32_t>(type), threadNum);
154 0 : return HCCL_SUCCESS;
155 0 : }
156 :
157 24 : HcommResult HcommThreadFree(const ThreadHandle* threads, uint32_t threadNum)
158 : {
159 24 : CHK_PTR_NULL(threads);
160 23 : (void)HcommResMgrInit();
161 23 : return hccl::FreeThreads(threads, threadNum, AicpuTsChannelHelper::GetBinHandle());
162 : }
163 :
164 10 : HcommResult HcommThreadAllocWithStream(CommEngine engine, rtStream_t stream, uint32_t notifyNum, ThreadHandle* thread)
165 : {
166 10 : CHK_PTR_NULL(thread);
167 : hccl::NotifyLoadType notifyLoadType;
168 9 : CHK_RET(CommHostEngineToNotifyLoadType(engine, notifyLoadType));
169 8 : std::shared_ptr<hccl::Thread> handle;
170 8 : EXCEPTION_CATCH(handle = std::make_shared<hccl::CpuTsThread>(stream, notifyNum, notifyLoadType), return HCCL_E_PTR);
171 8 : CHK_RET(handle->Init());
172 :
173 : // 返回第一个句柄
174 7 : *thread = reinterpret_cast<ThreadHandle>(handle.get());
175 : {
176 7 : std::lock_guard<std::mutex> lock(hcomm::g_ThreadMapMtx);
177 7 : hcomm::g_ThreadMap.emplace(*thread, handle);
178 7 : }
179 :
180 7 : HCCL_INFO(
181 : "[ThreadMgr] ThreadAcquireWithStream done: engine[%s] stream[%p], "
182 : "notifyNum[%u]",
183 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), stream, notifyNum);
184 7 : return HCCL_SUCCESS;
185 8 : }
186 :
187 5 : HcommResult HcommThreadFreeWithStream(const ThreadHandle* threads, uint32_t threadNum)
188 : {
189 5 : CHK_PTR_NULL(threads);
190 4 : if (threadNum == 0U) {
191 1 : HCCL_ERROR("[%s] threadNum is 0", __func__);
192 1 : return HCCL_E_PARA;
193 : }
194 3 : HcommResult ret = HCCL_SUCCESS;
195 3 : std::lock_guard<std::mutex> lock(hcomm::g_ThreadMapMtx);
196 6 : for (uint32_t i = 0; i < threadNum; ++i) {
197 3 : ThreadHandle handle = threads[i];
198 3 : auto it = hcomm::g_ThreadMap.find(handle);
199 3 : if (it == hcomm::g_ThreadMap.end()) {
200 1 : HCCL_WARNING("[%s] thread handle[0x%llx] not found in g_ThreadMap, skip", __func__, handle);
201 1 : continue;
202 : }
203 2 : HcclResult deInitRet = it->second->DeInit();
204 2 : if (deInitRet != HCCL_SUCCESS) {
205 0 : HCCL_WARNING("[%s] thread DeInit failed, ret[%d], handle[0x%llx]", __func__, deInitRet, handle);
206 0 : ret = static_cast<HcommResult>(deInitRet);
207 : }
208 2 : hcomm::g_ThreadMap.erase(it);
209 2 : HCCL_INFO("[%s] thread freed, handle[0x%llx]", __func__, handle);
210 : }
211 3 : return ret;
212 3 : }
213 :
214 2 : HcommResult HcommThreadSupplementNotify(
215 : CommEngine engine, ThreadHandle* handles, uint32_t threadNum, uint32_t* supplementNotifyNums)
216 : {
217 2 : CHK_PTR_NULL(handles);
218 1 : CHK_PTR_NULL(supplementNotifyNums);
219 1 : HcommResult hcommRet = HcommResMgrInit();
220 1 : CHK_PRT_RET(
221 : hcommRet != HCCL_SUCCESS,
222 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
223 :
224 1 : std::vector<std::shared_ptr<hccl::Thread>> needSupplementThread;
225 1 : std::unique_ptr<ThreadHandle[]> threadHandle;
226 1 : EXCEPTION_CATCH(threadHandle = std::make_unique<ThreadHandle[]>(threadNum), return (HcommResult)HCCL_E_PTR);
227 :
228 2 : for (uint32_t i = 0; i < threadNum; ++i) {
229 1 : std::shared_ptr<hccl::Thread> threadPtr;
230 1 : CHK_RET(hccl::LookupThreadByHandle(handles[i], threadPtr));
231 1 : CHK_RET(threadPtr->SupplementNotify(supplementNotifyNums[i]));
232 1 : needSupplementThread.push_back(std::move(threadPtr));
233 1 : threadHandle[i] = handles[i];
234 1 : }
235 :
236 : // 设备侧 kernel launch（仅 AICPU 引擎触发）
237 1 : if (engine == COMM_ENGINE_AICPU && !needSupplementThread.empty()) {
238 0 : CHK_RET(HcommResMgr::EnsureKernelBinLoaded(engine));
239 0 : HcclResult ret = hccl::AicpuLaunchMgr::SupplementNotifyKernelLaunch(
240 0 : needSupplementThread, std::string(""), threadHandle, HcommResMgr::GetBinHandle());
241 0 : CHK_PRT_RET(
242 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] SupplementNotifyKernelLaunch failed, ret[%d]", __func__, ret),
243 : (HcommResult)ret);
244 : }
245 1 : return HCCL_SUCCESS;
246 1 : }
247 :
248 1 : HcommResult HcommThreadGetNotifyNum(ThreadHandle thread, uint32_t* notifyNum)
249 : {
250 1 : CHK_PTR_NULL(notifyNum);
251 1 : (void)HcommResMgrInit();
252 1 : hccl::Thread* threadPtr = reinterpret_cast<hccl::Thread*>(thread);
253 1 : CHK_PTR_NULL(threadPtr);
254 1 : *notifyNum = threadPtr->GetNotifyNum();
255 1 : HCCL_INFO("[%s] thread[0x%llx] notifyNum[%u]", __func__, thread, *notifyNum);
256 1 : return HCCL_SUCCESS;
257 : }
258 :
259 1 : HcommResult HcommThreadExportToCommEngineAiCpu(
260 : ThreadHandle* handles, const std::string& commIdStr, uint32_t threadNum, CommEngine dstEngine,
261 : ThreadHandle* outHandles)
262 : {
263 : // AICPU 方向：正查 FindThreadByCommEngine + miss 批量建 + 入表 + 映射
264 1 : std::vector<std::shared_ptr<hccl::Thread>> hostThreads;
265 1 : std::vector<uint32_t> missIdx;
266 2 : for (uint32_t i = 0; i < threadNum; ++i) {
267 1 : std::shared_ptr<hccl::Thread> threadPtr;
268 1 : CHK_RET(hccl::LookupThreadByHandle(handles[i], threadPtr));
269 1 : hccl::Thread* exported = threadPtr->FindThreadByCommEngine(dstEngine);
270 1 : if (exported != nullptr) {
271 0 : outHandles[i] = reinterpret_cast<ThreadHandle>(exported);
272 : } else {
273 1 : hostThreads.push_back(std::move(threadPtr));
274 1 : missIdx.push_back(i);
275 : }
276 1 : }
277 1 : if (!hostThreads.empty()) {
278 1 : CHK_RET(HcommResMgr::EnsureKernelBinLoaded(dstEngine));
279 1 : std::unique_ptr<ThreadHandle[]> aicpuHandle;
280 1 : EXCEPTION_CATCH(
281 : aicpuHandle = std::make_unique<ThreadHandle[]>(hostThreads.size()), return (HcommResult)HCCL_E_PTR);
282 1 : HcclResult ret = hccl::AicpuLaunchMgr::ThreadKernelLaunchForComm(
283 : hostThreads, commIdStr, aicpuHandle, HcommResMgr::GetBinHandle());
284 1 : CHK_PRT_RET(
285 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] ThreadKernelLaunchForComm failed, ret[%d]", __func__, ret),
286 : (HcommResult)ret);
287 2 : for (size_t i = 0; i < hostThreads.size(); ++i) {
288 1 : outHandles[missIdx[i]] = aicpuHandle[i];
289 1 : CHK_RET(hostThreads[i]->AddThreadHandleToMap(dstEngine, aicpuHandle[i]));
290 : // 入 g_ThreadD2HMap（device->host）
291 1 : ThreadHandle hostHandle = reinterpret_cast<ThreadHandle>(hostThreads[i].get());
292 1 : CHK_RET(hccl::FillThreadD2HMap(&aicpuHandle[i], &hostHandle, 1));
293 : }
294 1 : }
295 :
296 1 : return HCCL_SUCCESS;
297 1 : }
298 :
299 2 : HcommResult HcommThreadExportToCommEngine(
300 : ThreadHandle* handles, const char* commId, uint32_t threadNum, CommEngine dstEngine, ThreadHandle* outHandles)
301 : {
302 2 : CHK_PTR_NULL(handles);
303 1 : CHK_PTR_NULL(outHandles);
304 1 : HcommResult hcommRet = HcommResMgrInit();
305 1 : CHK_PRT_RET(
306 : hcommRet != HCCL_SUCCESS,
307 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
308 1 : CHK_RET(RefreshCommEngineContext(dstEngine));
309 2 : const std::string commIdStr = (commId != nullptr) ? std::string(commId) : std::string();
310 1 : switch (dstEngine) {
311 0 : case COMM_ENGINE_CPU:
312 : case COMM_ENGINE_CPU_TS:
313 : case COMM_ENGINE_CCU: {
314 : // CPU 方向：反向查询 g_ThreadD2HMap（device 到 host 映射）
315 0 : for (uint32_t i = 0; i < threadNum; ++i) {
316 0 : CHK_RET(hccl::LookupD2HHandle(handles[i], outHandles[i]));
317 : }
318 0 : return HCCL_SUCCESS;
319 : }
320 1 : case COMM_ENGINE_AICPU:
321 : case COMM_ENGINE_AICPU_TS: {
322 1 : CHK_RET(
323 : (HcclResult)HcommThreadExportToCommEngineAiCpu(handles, commIdStr, threadNum, dstEngine, outHandles));
324 1 : break;
325 : }
326 0 : default:
327 0 : HCCL_ERROR("[%s] unsupported dstEngine[%d]", __func__, static_cast<int32_t>(dstEngine));
328 0 : return (HcommResult)HCCL_E_PARA;
329 : }
330 1 : return HCCL_SUCCESS;
331 1 : }
332 :
333 5 : HcommResult HcommThreadResGetInfo(ThreadHandle thread, ThreadResType resType, uint32_t infoLen, void** info)
334 : {
335 5 : CHK_PTR_NULL(info);
336 4 : CHK_PRT_RET(thread == 0, HCCL_ERROR("[%s] thread is 0", __func__), HCCL_E_PTR);
337 :
338 3 : HCCL_INFO(
339 : "[%s] begin, thread[0x%llx], resType[%d], infoLen[%u]", __func__, thread, static_cast<int32_t>(resType),
340 : infoLen);
341 :
342 : /* ThreadHandle 是 Thread* 的 reinterpret_cast，可直接转换 */
343 3 : auto* threadPtr = reinterpret_cast<hccl::Thread*>(thread);
344 :
345 3 : if (resType != ThreadResType::THREAD_RES_TYPE_STREAM) {
346 1 : HCCL_ERROR("[%s] resType[%d] is not supported", __func__, static_cast<int32_t>(resType));
347 1 : return HCCL_E_NOT_SUPPORT;
348 : }
349 :
350 2 : CHK_PRT_RET(
351 : infoLen != sizeof(ThreadResTypeStream),
352 : HCCL_ERROR(
353 : "[%s] infoLen[%u] mismatch sizeof(ThreadResTypeStream)[%zu]", __func__, infoLen,
354 : sizeof(ThreadResTypeStream)),
355 : HCCL_E_PARA);
356 :
357 1 : hccl::Stream* streamPtr = threadPtr->GetStream();
358 1 : CHK_PTR_NULL(streamPtr);
359 1 : ThreadResTypeStream stream = streamPtr->ptr();
360 1 : CHK_PTR_NULL(stream);
361 :
362 1 : *info = stream;
363 :
364 1 : HCCL_INFO(
365 : "[%s] success, thread[0x%llx] resType[%d] stream[%p]", __func__, thread, static_cast<int32_t>(resType), *info);
366 1 : return HCCL_SUCCESS;
367 : }
|
