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 : #include <memory>
11 : #include <mutex>
12 : #include <unordered_map>
13 : #include <vector>
14 :
15 : #include "hcomm_c_adpt.h"
16 : #include "hcomm_c_adpt_common.h"
17 : #include "hcomm_res.h"
18 : #include "hcomm_res_defs.h"
19 : #include "log.h"
20 : #include "thread.h"
21 : #include "cpu_ts_thread.h"
22 : #include "param_check_pub.h"
23 : #include "comm_engine_utils.h"
24 : #include "exception_handler.h"
25 : #include "adapter_rts_common.h"
26 : #include "aicpu_ts_channel_helper.h"
27 :
28 : namespace hcomm {
29 : static std::unordered_map<ThreadHandle, std::shared_ptr<hccl::Thread>> g_ThreadMap;
30 : static std::mutex g_ThreadMapMtx;
31 : } // namespace hcomm
32 :
33 : using namespace hcomm;
34 :
35 : HcommResult
36 19 : HcommThreadAlloc(CommEngine engine, uint32_t threadNum, const uint32_t* notifyNumPerThread, ThreadHandle* threads)
37 : {
38 19 : CHK_PTR_NULL(threads);
39 18 : CHK_PTR_NULL(notifyNumPerThread);
40 18 : (void)HcommResMgrInit();
41 18 : const uint32_t notifyNum = notifyNumPerThread[0];
42 18 : if (threadNum > 1U) {
43 7 : HCCL_RUN_WARNING(
44 : "[%s] only notifyNumPerThread[0] is used currently, threadNum[%u], notifyNum[0][%u].", __func__, threadNum,
45 : notifyNum);
46 : }
47 18 : HCCL_INFO(
48 : "[%s] ThreadAcquire begin. engine[%s], threadNum[%u], notifyPerThread[%u], threads[%p]", __func__,
49 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum, threads);
50 18 : CHK_RET(RefreshCommEngineContext(engine));
51 :
52 : // 1. 参数校验
53 18 : CHK_RET(hccl::ValidateThreadParams(threadNum, notifyNum));
54 :
55 : // 2. 获取引擎对应的类型
56 : hccl::NotifyLoadType notifyLoadType;
57 : hccl::StreamType streamType;
58 15 : CHK_RET(hccl::CommEngineToNotifyLoadType(engine, notifyLoadType));
59 14 : CHK_RET(hccl::CommEngineToStreamType(engine, streamType));
60 :
61 : // 3. 创建线程
62 14 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
63 14 : hccl::ThreadCreateParams params(engine, threadNum, notifyNum, notifyLoadType, streamType);
64 14 : CHK_RET(hccl::CreateAndInitThreads(params, newThreads));
65 :
66 : // 4. 插入全局映射表
67 12 : CHK_RET(hccl::SaveThreads(newThreads));
68 :
69 : // 5. 储存线程句柄
70 12 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
71 12 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
72 :
73 12 : HCCL_INFO(
74 : "[HcommThreadAlloc] ThreadAcquire done: engine[%s] threadNum[%u], notifyPerThread[%u]",
75 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum);
76 12 : return HCCL_SUCCESS;
77 14 : }
78 :
79 14 : HcommResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread, ThreadHandle* threads)
80 : {
81 14 : return ::HcommThreadAlloc(engine, threadNum, ¬ifyNumPerThread, threads);
82 : }
83 :
84 0 : HcommResult HcommThreadAllocWithConfig(
85 : CommEngine engine, uint32_t threadNum, ThreadType type, const ThreadConfig* config, ThreadHandle* threads)
86 : {
87 0 : CHK_PTR_NULL(threads);
88 0 : CHK_PTR_NULL(config);
89 0 : CHK_PRT_RET(
90 : type == THREAD_TYPE_INVALID,
91 : HCCL_ERROR("[%s] thread type[%d] is invalid", __func__, static_cast<int32_t>(type)), (HcommResult)HCCL_E_PARA);
92 0 : CHK_PRT_RET(
93 : engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_CPU_TS,
94 : HCCL_ERROR(
95 : "[%s] commEngine[%d] CPU_TS/AICPU_TS not supported, use engine with ThreadType instead", __func__,
96 : static_cast<int32_t>(engine)),
97 : (HcommResult)HCCL_E_PARA);
98 0 : CHK_PRT_RET(
99 : engine == COMM_ENGINE_AIV || engine == COMM_ENGINE_CCU,
100 : HCCL_ERROR(
101 : "[%s] commEngine[%d] AIV/CCU not supported, supported engines: CPU/AICPU", __func__,
102 : static_cast<int32_t>(engine)),
103 : (HcommResult)HCCL_E_PARA);
104 0 : CHK_PRT_RET(
105 : threadNum == 0, HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), (HcommResult)HCCL_E_PARA);
106 0 : HcommResult hcommRet = HcommResMgrInit();
107 0 : CHK_PRT_RET(
108 : hcommRet != HCCL_SUCCESS,
109 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
110 0 : CHK_RET(RefreshCommEngineContext(engine));
111 :
112 0 : HCCL_INFO(
113 : "[%s] begin. engine[%d], threadType[%d], threadNum[%u], threads[%p]", __func__, engine,
114 : static_cast<int32_t>(type), threadNum, threads);
115 :
116 : hccl::NotifyLoadType notifyLoadType;
117 : hccl::StreamType streamType;
118 0 : CHK_RET(hccl::GetNotifyLoadType(engine, type, notifyLoadType));
119 0 : CHK_RET(hccl::GetStreamType(engine, type, streamType));
120 :
121 0 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
122 0 : newThreads.reserve(threadNum);
123 0 : for (uint32_t i = 0; i < threadNum; ++i) {
124 0 : CHK_PRT_RET(
125 : config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
126 : HCCL_ERROR(
127 : "[%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x], call ThreadConfigInit first", __func__, i,
128 : config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD),
129 : (HcommResult)HCCL_E_PARA);
130 0 : CHK_RET(hccl::ValidateThreadParams(1, config[i].notifyNumPerThread));
131 0 : std::shared_ptr<hccl::Thread> threadPtr;
132 : HcclResult ret
133 0 : = hccl::CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, threadPtr);
134 0 : CHK_PRT_RET(
135 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Failed to create thread at index[%u], ret[%d]", __func__, i, ret),
136 : (HcommResult)ret);
137 0 : ret = threadPtr->Init();
138 0 : CHK_PRT_RET(
139 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] Failed to init thread at index[%u], ret[%d]", __func__, i, ret),
140 : (HcommResult)ret);
141 0 : newThreads.emplace_back(std::move(threadPtr));
142 0 : }
143 :
144 0 : CHK_RET(hccl::SaveThreads(newThreads));
145 0 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
146 0 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
147 :
148 0 : HCCL_INFO(
149 : "[%s] done: engine[%d] threadType[%d] threadNum[%u]", __func__, engine, static_cast<int32_t>(type), threadNum);
150 0 : return HCCL_SUCCESS;
151 0 : }
152 :
153 21 : HcommResult HcommThreadFree(const ThreadHandle* threads, uint32_t threadNum)
154 : {
155 21 : CHK_PTR_NULL(threads);
156 20 : (void)HcommResMgrInit();
157 20 : return hccl::FreeThreads(threads, threadNum, AicpuTsChannelHelper::GetBinHandle());
158 : }
159 :
160 9 : HcommResult HcommThreadAllocWithStream(CommEngine engine, rtStream_t stream, uint32_t notifyNum, ThreadHandle* thread)
161 : {
162 9 : CHK_PTR_NULL(thread);
163 : hccl::NotifyLoadType notifyLoadType;
164 8 : CHK_RET(CommHostEngineToNotifyLoadType(engine, notifyLoadType));
165 7 : std::shared_ptr<hccl::Thread> handle;
166 7 : EXCEPTION_CATCH(handle = std::make_shared<hccl::CpuTsThread>(stream, notifyNum, notifyLoadType), return HCCL_E_PTR);
167 7 : CHK_RET(handle->Init());
168 :
169 : // 返回第一个句柄
170 6 : *thread = reinterpret_cast<ThreadHandle>(handle.get());
171 : {
172 6 : std::lock_guard<std::mutex> lock(hcomm::g_ThreadMapMtx);
173 6 : hcomm::g_ThreadMap.emplace(*thread, handle);
174 6 : }
175 :
176 6 : HCCL_INFO(
177 : "[ThreadMgr] ThreadAcquireWithStream done: engine[%s] stream[%p], "
178 : "notifyNum[%u]",
179 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), stream, notifyNum);
180 6 : return HCCL_SUCCESS;
181 7 : }
182 :
183 4 : HcommResult HcommThreadFreeWithStream(const ThreadHandle* threads, uint32_t threadNum)
184 : {
185 4 : CHK_PTR_NULL(threads);
186 3 : if (threadNum == 0U) {
187 1 : HCCL_ERROR("[%s] threadNum is 0", __func__);
188 1 : return HCCL_E_PARA;
189 : }
190 2 : HcommResult ret = HCCL_SUCCESS;
191 2 : std::lock_guard<std::mutex> lock(hcomm::g_ThreadMapMtx);
192 4 : for (uint32_t i = 0; i < threadNum; ++i) {
193 2 : ThreadHandle handle = threads[i];
194 2 : auto it = hcomm::g_ThreadMap.find(handle);
195 2 : if (it == hcomm::g_ThreadMap.end()) {
196 1 : HCCL_WARNING("[%s] thread handle[0x%llx] not found in g_ThreadMap, skip", __func__, handle);
197 1 : continue;
198 : }
199 1 : HcclResult deInitRet = it->second->DeInit();
200 1 : if (deInitRet != HCCL_SUCCESS) {
201 0 : HCCL_WARNING("[%s] thread DeInit failed, ret[%d], handle[0x%llx]", __func__, deInitRet, handle);
202 0 : ret = static_cast<HcommResult>(deInitRet);
203 : }
204 1 : hcomm::g_ThreadMap.erase(it);
205 1 : HCCL_INFO("[%s] thread freed, handle[0x%llx]", __func__, handle);
206 : }
207 2 : return ret;
208 2 : }
209 :
210 5 : HcommResult HcommThreadResGetInfo(ThreadHandle thread, ThreadResType resType, uint32_t infoLen, void** info)
211 : {
212 5 : CHK_PTR_NULL(info);
213 4 : CHK_PRT_RET(thread == 0, HCCL_ERROR("[%s] thread is 0", __func__), HCCL_E_PTR);
214 :
215 3 : HCCL_INFO(
216 : "[%s] begin, thread[0x%llx], resType[%d], infoLen[%u]", __func__, thread, static_cast<int32_t>(resType),
217 : infoLen);
218 :
219 : /* ThreadHandle 是 Thread* 的 reinterpret_cast，可直接转换 */
220 3 : auto* threadPtr = reinterpret_cast<hccl::Thread*>(thread);
221 :
222 3 : if (resType != ThreadResType::THREAD_RES_TYPE_STREAM) {
223 1 : HCCL_ERROR("[%s] resType[%d] is not supported", __func__, static_cast<int32_t>(resType));
224 1 : return HCCL_E_NOT_SUPPORT;
225 : }
226 :
227 2 : CHK_PRT_RET(
228 : infoLen != sizeof(ThreadResTypeStream),
229 : HCCL_ERROR(
230 : "[%s] infoLen[%u] mismatch sizeof(ThreadResTypeStream)[%zu]", __func__, infoLen,
231 : sizeof(ThreadResTypeStream)),
232 : HCCL_E_PARA);
233 :
234 1 : hccl::Stream* streamPtr = threadPtr->GetStream();
235 1 : CHK_PTR_NULL(streamPtr);
236 1 : ThreadResTypeStream stream = streamPtr->ptr();
237 1 : CHK_PTR_NULL(stream);
238 :
239 1 : *info = stream;
240 :
241 1 : HCCL_INFO(
242 : "[%s] success, thread[0x%llx] resType[%d] stream[%p]", __func__, thread, static_cast<int32_t>(resType), *info);
243 1 : return HCCL_SUCCESS;
244 : }
|
