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 "thread_manager.h"
12 : #include <cstring>
13 : #include "aicpu_launch_manager.h"
14 : #include "independent_op.h"
15 : #include "comm_engine_utils.h"
16 : #include "hcomm_res.h"
17 :
18 : namespace hccl {
19 :
20 :
21 334 : ThreadMgr::ThreadMgr(uint32_t threadNum, uint32_t notifyNumPerThread, std::string commId,
22 334 : aclrtBinHandle binHandle, const ManagerCallbacks& callbacks) : threadNum_(threadNum), notifyNumPerThread_(notifyNumPerThread),
23 334 : commId_(commId), binHandle_(binHandle), callbacks_(callbacks){}
24 :
25 334 : ThreadMgr::~ThreadMgr()
26 : {
27 334 : auto it = dedicatedThreadMap_.find(HCCL_DED_THREAD_TYPE_AICPU_LAUNCH);
28 334 : if (it != dedicatedThreadMap_.end()) {
29 5 : ThreadHandle thread = it->second;
30 5 : HcommThreadFree(&thread, 1);
31 : }
32 334 : }
33 :
34 64 : uint64_t ThreadMgr::GetMaxNotifyTotal()
35 : {
36 : // 如果没设定最大值，设置一下
37 64 : uint64_t maxNotifyTotal = 0;
38 64 : if (threadNum_ == HCCL_COMM_THREADNUM_CONFIG_NOT_SET &&
39 13 : notifyNumPerThread_ == HCCL_COMM_NOTIFY_NUM_PER_THREAD_CONFIG_NOT_SET) {
40 13 : maxNotifyTotal = HCCL_THREAD_NOTIFY_MAX_NUM;
41 13 : threadNum_ = SIGNAL_DEV_STREAM_MAX_NUM;
42 13 : notifyNumPerThread_ = HCCL_THREAD_NOTIFY_MAX_NUM;
43 : } else {
44 51 : maxNotifyTotal = static_cast<uint64_t>(threadNum_) * static_cast<uint64_t>(notifyNumPerThread_);
45 51 : maxNotifyTotal = maxNotifyTotal > HCCL_THREAD_NOTIFY_MAX_NUM ? HCCL_THREAD_NOTIFY_MAX_NUM : maxNotifyTotal;
46 : }
47 64 : return maxNotifyTotal;
48 : }
49 :
50 38 : HcclResult ThreadMgr::CheckNotifyNum(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread)
51 : {
52 38 : uint64_t maxNotifyTotal = GetMaxNotifyTotal();
53 38 : const uint64_t used = usedNotifyNum_;
54 38 : uint64_t remainNotifyQuota = (maxNotifyTotal > used) ? (maxNotifyTotal - used) : 0;
55 38 : uint64_t needNotifyTotal = static_cast<uint64_t>(threadNum) * static_cast<uint64_t>(notifyNumPerThread);
56 38 : if (remainNotifyQuota < needNotifyTotal || notifyNumPerThread > notifyNumPerThread_ ||
57 : maxNotifyTotal > HCCL_THREAD_NOTIFY_MAX_NUM) {
58 1 : HCCL_ERROR("[ThreadMgr][%s] Notify quota exhausted: remainQuota[%llu], total[%llu], used[%llu], need[%llu], "
59 : "setPreNum[%u], allocPreNum[%u]", __func__, remainNotifyQuota, maxNotifyTotal, used, needNotifyTotal,
60 : notifyNumPerThread_, notifyNumPerThread);
61 1 : return HCCL_E_UNAVAIL;
62 : }
63 :
64 37 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] HcclThreadAcquire quota: engine[%s], "
65 : "remainNotifyQuota[%llu]", __func__, commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), remainNotifyQuota);
66 37 : return HCCL_SUCCESS;
67 : }
68 :
69 26 : HcclResult ThreadMgr::CheckThreadNum(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread)
70 : {
71 26 : GetMaxNotifyTotal();
72 26 : uint32_t remainQuota = (threadNum_ > threads_.size()) ? (threadNum_ - threads_.size()) : 0;
73 26 : if (remainQuota == 0 || threadNum > remainQuota) {
74 0 : HCCL_ERROR("[ThreadMgr][%s] Threads quota exhausted: remainQuota[%u], need[%u].",
75 : __func__, remainQuota, threadNum);
76 0 : return HCCL_E_UNAVAIL;
77 : }
78 :
79 26 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] HcclThreadAcquire quota: engine[%s] threadNum[%u].",
80 : __func__, commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), remainQuota);
81 26 : return CheckNotifyNum(engine, threadNum, notifyNumPerThread);
82 : }
83 :
84 6 : HcclResult ThreadMgr::SupplementNotify(CommEngine engine, std::vector<std::shared_ptr<Thread>> &needSupplementThread)
85 : {
86 6 : HcclResult ret = HCCL_E_INTERNAL;
87 : // thread资源 AICPU侧展开
88 6 : if ((engine == COMM_ENGINE_AICPU) && needSupplementThread.size() > 0) {
89 2 : if (!callbacks_.getAicpuCommState()) {
90 0 : HCCL_INFO("ThreadMgr::SupplementNotify kernelLaunchAicpuCommInit start");
91 0 : ret = callbacks_.kernelLaunchAicpuCommInit();
92 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
93 : HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
94 0 : callbacks_.setAicpuCommState(true);
95 : }
96 :
97 2 : std::unique_ptr<ThreadHandle[]> threadHandle;
98 2 : EXCEPTION_CATCH(threadHandle = std::make_unique<ThreadHandle[]>(needSupplementThread.size()),
99 : return HCCL_E_PTR);
100 8 : for (size_t i = 0; i < needSupplementThread.size(); ++i) {
101 6 : threadHandle[i] = hostToDeviceThreadHandle_[reinterpret_cast<ThreadHandle>(needSupplementThread[i].get())];
102 : }
103 2 : HCCL_INFO("ThreadMgr::SupplementNotify ThreadKernelLaunchForComm start needSupplementThread size[%zu]",
104 : needSupplementThread.size());
105 2 : ret = AicpuLaunchMgr::SupplementNotifyKernelLaunch(needSupplementThread, commId_, threadHandle, binHandle_);
106 2 : HCCL_INFO("ThreadMgr::SupplementNotify ThreadKernelLaunchForComm end");
107 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
108 : HCCL_ERROR("[ThreadMgr][SupplementNotify] AiCpuKernelLaunch failed, return [%d].", ret), ret);
109 2 : }
110 6 : return HCCL_SUCCESS;
111 : }
112 :
113 6 : HcclResult ThreadMgr::SupplementNotify(CommEngine engine, uint32_t threadNum, ThreadType type,
114 : const ThreadConfig *config)
115 : {
116 : NotifyLoadType notifyLoadType;
117 6 : CHK_RET(GetNotifyLoadType(engine, type, notifyLoadType));
118 6 : std::vector<std::shared_ptr<Thread>> &threads = engineToThreadsMap_[std::make_pair(engine, type)];
119 6 : std::vector<std::shared_ptr<Thread>> needSupplementThread;
120 6 : HcclResult ret = HCCL_E_INTERNAL;
121 6 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] threadNum[%zu] notifyLoadType[%d]",
122 : __func__, commId_.c_str(), threads.size(), static_cast<int32_t>(notifyLoadType));
123 :
124 6 : DevType devType = DevType::DEV_TYPE_COUNT;
125 6 : CHK_RET(hrtGetDeviceType(devType));
126 0 : u32 aicpuHostNotify = ((devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960) &&
127 6 : (engine == COMM_ENGINE_AICPU)) ? 1 : 0;
128 :
129 22 : for (size_t i = 0; i < threads.size() && i < threadNum; ++i) {
130 16 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] AicpuTsThread[%u] notifyLoadType[%d] notifyNum[%u]",
131 : __func__, commId_.c_str(), static_cast<uint32_t>(i), static_cast<int32_t>(notifyLoadType), config[i].notifyNumPerThread);
132 16 : u32 notifyNum = threads[i]->GetNotifyNum() >= aicpuHostNotify ?
133 16 : (threads[i]->GetNotifyNum() - aicpuHostNotify) : 0;
134 16 : if (config[i].notifyNumPerThread > notifyNum) {
135 12 : u32 supplementNotifyNum = config[i].notifyNumPerThread - notifyNum;
136 12 : CHK_RET(CheckNotifyNum(engine, 1, supplementNotifyNum));
137 12 : ret = threads[i]->SupplementNotify(supplementNotifyNum);
138 12 : if (ret != HCCL_SUCCESS) {
139 0 : HCCL_ERROR("[ThreadMgr][%s] Failed to supplement notify[%u] on thread index[%u], ret[%d]",
140 : __func__, supplementNotifyNum, i, ret);
141 0 : return ret;
142 : }
143 12 : needSupplementThread.push_back(threads[i]);
144 12 : usedNotifyNum_ += supplementNotifyNum;
145 : }
146 : }
147 6 : CHK_RET(SupplementNotify(engine, needSupplementThread));
148 6 : return HCCL_SUCCESS;
149 6 : }
150 :
151 14 : HcclResult ThreadMgr::SupplementThread(CommEngine engine, std::vector<std::shared_ptr<Thread>> &newThreads,
152 : std::unique_ptr<ThreadHandle[]> &hostHandle)
153 : {
154 : // thread资源 AICPU侧展开
155 14 : if (engine == COMM_ENGINE_AICPU) {
156 4 : HcclResult ret = HCCL_E_INTERNAL;
157 4 : if (!callbacks_.getAicpuCommState()) {
158 0 : HCCL_INFO("ThreadMgr::HcclAllocThreadRes kernelLaunchAicpuCommInit start");
159 0 : ret = callbacks_.kernelLaunchAicpuCommInit();
160 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
161 : HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
162 0 : callbacks_.setAicpuCommState(true);
163 : }
164 :
165 4 : EXCEPTION_CATCH(hostHandle = std::make_unique<ThreadHandle[]>(newThreads.size()),
166 : return HCCL_E_PTR);
167 4 : HCCL_INFO("ThreadMgr::HcclAllocThreadRes ThreadKernelLaunchForComm start");
168 4 : ret = AicpuLaunchMgr::ThreadKernelLaunchForComm(newThreads, commId_, hostHandle, binHandle_);
169 4 : HCCL_INFO("ThreadMgr::HcclAllocThreadRes ThreadKernelLaunchForComm end");
170 4 : CHK_PRT_RET(ret != HCCL_SUCCESS,
171 : HCCL_ERROR("[ThreadMgr][HcclThreadAcquire] AiCpuKernelLaunch failed, return [%d].", ret), ret);
172 : }
173 14 : return HCCL_SUCCESS;
174 : }
175 :
176 14 : HcclResult ThreadMgr::SupplementThread(CommEngine engine, uint32_t supplementThreadNum, ThreadType type,
177 : const ThreadConfig *config)
178 : {
179 : NotifyLoadType notifyLoadType;
180 : StreamType streamType;
181 14 : CHK_RET(GetNotifyLoadType(engine, type, notifyLoadType));
182 14 : CHK_RET(GetStreamType(engine, type, streamType));
183 14 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] supplementThreadNum[%u] notifyLoadType[%d], streamType[%d]", __func__,
184 : commId_.c_str(), supplementThreadNum, static_cast<int32_t>(notifyLoadType), static_cast<int32_t>(streamType));
185 14 : std::vector<std::shared_ptr<Thread>> newThreads;
186 14 : newThreads.reserve(supplementThreadNum);
187 14 : HcclResult ret = HCCL_E_INTERNAL;
188 :
189 33 : for (uint32_t i = 0; i < supplementThreadNum; ++i) {
190 19 : std::shared_ptr<Thread> handle;
191 19 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] [%u]notifyLoadType[%d], streamType[%d] notifyNum[%u]", __func__,
192 : commId_.c_str(), i, static_cast<int32_t>(notifyLoadType),
193 : static_cast<int32_t>(streamType), config[i].notifyNumPerThread);
194 19 : CHK_RET(CheckThreadNum(engine, 1, config[i].notifyNumPerThread));
195 19 : CHK_RET(CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, handle));
196 19 : ret = handle->Init();
197 19 : if (ret != HCCL_SUCCESS) {
198 0 : HCCL_ERROR("[ThreadMgr][HcclThreadAcquire] Failed to init thread index[%u], ret[%d]", i, ret);
199 0 : return ret;
200 : }
201 19 : usedNotifyNum_ += config[i].notifyNumPerThread;
202 19 : newThreads.emplace_back(std::move(handle));
203 19 : }
204 :
205 14 : std::unique_ptr<ThreadHandle[]> hostHandle;
206 14 : CHK_RET(SupplementThread(engine, newThreads, hostHandle));
207 14 : threads_.reserve(threads_.size() + newThreads.size());
208 14 : auto iter = engineToThreadsMap_.find(std::make_pair(engine, type));
209 14 : if (iter == engineToThreadsMap_.end()) {
210 0 : HCCL_ERROR("[SupplementThread] engine+type pair not found in engineToThreadsMap_ "
211 : "engine[%d], type[%d].", engine, type);
212 0 : return HCCL_E_INTERNAL;
213 : }
214 14 : auto &threadVec = iter->second;
215 14 : threadVec.reserve(threadVec.size() + newThreads.size());
216 14 : threadVec.insert(threadVec.end(), newThreads.begin(), newThreads.end());
217 14 : threads_.insert(threads_.end(), newThreads.begin(), newThreads.end());
218 :
219 14 : if (engine == COMM_ENGINE_AICPU) {
220 9 : for (size_t i = 0; i < newThreads.size(); ++i) {
221 5 : ThreadHandle cpuTsHandle = reinterpret_cast<ThreadHandle>(newThreads[i].get());
222 5 : newThreads[i]->AddThreadHandleToMap(engine, hostHandle[i]);
223 5 : hostToDeviceThreadHandle_[cpuTsHandle] = hostHandle[i];
224 5 : threadHandleOthersToCpu_[hostHandle[i]] = cpuTsHandle;
225 : }
226 : }
227 14 : return HCCL_SUCCESS;
228 14 : }
229 :
230 16 : HcclResult ThreadMgr::HcclThreadAcquireV2(CommEngine engine, uint32_t threadNum, ThreadType type,
231 : const ThreadConfig *config, ThreadHandle *threads, std::vector<uint32_t> &threadId)
232 : {
233 16 : CHK_PTR_NULL(threads);
234 16 : CHK_PTR_NULL(config);
235 16 : if (threadNum == 0) {
236 0 : HCCL_ERROR("[ThreadMgr][HcclThreadAcquire] threadNum is 0");
237 0 : return HCCL_E_PARA;
238 : }
239 :
240 51 : for (u32 i = 0; i < threadNum; ++i) {
241 35 : CHK_PRT_RET(config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
242 : HCCL_ERROR("[ThreadMgr][%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x]",
243 : __func__, i, config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD), HCCL_E_PARA);
244 : }
245 :
246 16 : std::lock_guard<std::mutex> lock(threadMutex_);
247 16 : std::lock_guard<std::mutex> lockMap(threadMapMutex_);
248 16 : std::lock_guard<std::mutex> engineToThreadMtx(engineToThreadMutex_);
249 16 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] HcclThreadAcquire begin, max: engine[%s] threadNum[%u],"
250 : "notifyPerThread[%u], need: threadNum[%u], threadType[%d]",
251 : __func__, commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum_, notifyNumPerThread_, threadNum, static_cast<int32_t>(type));
252 :
253 : // 1、thread上的notify数量不够，需要给thread补充notify
254 16 : auto it = engineToThreadsMap_.find(std::make_pair(engine, type));
255 16 : if (it == engineToThreadsMap_.end()) {
256 10 : it = engineToThreadsMap_.emplace(std::make_pair(engine, type), std::vector<std::shared_ptr<Thread>>{}).first;
257 : }
258 16 : auto &threadVec = it->second;
259 16 : if (threadVec.size() > 0) {
260 : // 调用补充函数，如果engine是COMM_ENGINE_AICPU，需要去device恢复
261 6 : CHK_RET(SupplementNotify(engine, threadNum, type, config));
262 : }
263 :
264 : // 2、补充thread
265 16 : if (threadVec.size() < threadNum) {
266 14 : u32 supplementThreadNum = threadNum - threadVec.size();
267 : // 调用补充函数，如果engine是COMM_ENGINE_AICPU，需要去device恢复
268 14 : CHK_RET(SupplementThread(engine, supplementThreadNum, type, &config[threadVec.size()]));
269 : }
270 : // 3、返回threadHandle和id
271 51 : for (u32 idx = 0; idx < threadNum; idx++) {
272 35 : ThreadHandle handle = reinterpret_cast<ThreadHandle>(threadVec[idx].get());
273 35 : threads[idx] = (engine == COMM_ENGINE_AICPU) ?
274 13 : hostToDeviceThreadHandle_[handle] : handle;
275 35 : uint32_t id = threadVec[idx]->GetStream()->sqId();
276 35 : HCCL_DEBUG("[%s]idx[%u] threadHandle[%llu] thread id = [%u]", __func__, idx, threads[idx], id);
277 35 : threadId.push_back(id);
278 :
279 35 : std::lock_guard<std::mutex> threadhandleToThreadMtx(threadhandleToThreadMutex_);
280 35 : threadMap_[threads[idx]] = threadVec[idx];
281 35 : }
282 :
283 16 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] HcclThreadAcquire done: engine[%s] threadNum[%u]%s",
284 : __func__, commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum,
285 : (engine == COMM_ENGINE_AICPU) ? " (AICPU token ready)" : "");
286 16 : return HCCL_SUCCESS;
287 16 : }
288 :
289 5 : HcclResult ThreadMgr::CreateAndInitThreads(CommEngine engine, StreamType streamType,
290 : NotifyLoadType notifyLoadType, uint32_t threadNum, const ThreadConfig *config,
291 : std::vector<std::shared_ptr<Thread>> &newThreads)
292 : {
293 11 : for (uint32_t i = 0; i < threadNum; ++i) {
294 7 : std::shared_ptr<Thread> handle;
295 7 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] AicpuTsThread notifyLoadType[%d], streamType[%d], notifyPerThread[%u]",
296 : __func__, commId_.c_str(), static_cast<int32_t>(notifyLoadType), static_cast<int32_t>(streamType),
297 : config[i].notifyNumPerThread);
298 7 : CHK_RET(CheckThreadNum(engine, 1, config[i].notifyNumPerThread));
299 6 : CHK_RET(CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, handle));
300 6 : HcclResult ret = handle->Init();
301 6 : if (ret != HCCL_SUCCESS) {
302 0 : HCCL_ERROR("[ThreadMgr][HcclThreadAcquire] Failed to init thread index %u", i);
303 0 : return ret;
304 : }
305 6 : usedNotifyNum_ += config[i].notifyNumPerThread;
306 6 : newThreads.emplace_back(std::move(handle));
307 7 : }
308 4 : return HCCL_SUCCESS;
309 : }
310 :
311 4 : HcclResult ThreadMgr::AssignThreadHandles(CommEngine engine,
312 : std::vector<std::shared_ptr<Thread>> &newThreads, ThreadHandle *threads,
313 : std::unique_ptr<ThreadHandle[]> &hostHandle)
314 : {
315 4 : if (engine == COMM_ENGINE_AICPU) {
316 0 : CHK_RET(SupplementThread(engine, newThreads, hostHandle));
317 0 : for (size_t i = 0; i < newThreads.size(); ++i) {
318 0 : threads[i] = hostHandle[i];
319 0 : HCCL_INFO("[ThreadMgr][%s] aicpu threadArray[%u] = [%llu]", __func__, i, threads[i]);
320 : }
321 : } else {
322 10 : for (size_t i = 0; i < newThreads.size(); ++i) {
323 6 : threads[i] = reinterpret_cast<ThreadHandle>(newThreads[i].get());
324 6 : HCCL_INFO("[ThreadMgr][%s] host threadArray[%u] = [%llu]", __func__, i, threads[i]);
325 : }
326 : }
327 4 : return HCCL_SUCCESS;
328 : }
329 :
330 4 : HcclResult ThreadMgr::StoreThreadsAndBuildHandleMap(CommEngine engine,
331 : std::vector<std::shared_ptr<Thread>> &newThreads,
332 : std::unique_ptr<ThreadHandle[]> &hostHandle)
333 : {
334 4 : auto threadsIt = threads_.insert(threads_.end(),
335 : std::make_move_iterator(newThreads.begin()),
336 : std::make_move_iterator(newThreads.end()));
337 :
338 4 : if (engine == COMM_ENGINE_AICPU) {
339 0 : for (size_t i = 0; i < newThreads.size(); ++i, ++threadsIt) {
340 0 : ThreadHandle cpuTsHandle = reinterpret_cast<ThreadHandle>((*threadsIt).get());
341 0 : (*threadsIt)->AddThreadHandleToMap(engine, hostHandle[i]);
342 0 : hostToDeviceThreadHandle_[cpuTsHandle] = hostHandle[i];
343 0 : threadHandleOthersToCpu_[hostHandle[i]] = cpuTsHandle;
344 : }
345 : }
346 4 : return HCCL_SUCCESS;
347 : }
348 :
349 5 : HcclResult ThreadMgr::HcclThreadAcquire(CommEngine engine, uint32_t threadNum, ThreadType type,
350 : const ThreadConfig *config, ThreadHandle *threads, std::vector<uint32_t> &threadId)
351 : {
352 5 : CHK_PTR_NULL(threads);
353 5 : CHK_PTR_NULL(config);
354 5 : std::lock_guard<std::mutex> lock(threadMutex_);
355 5 : std::lock_guard<std::mutex> lockMap(threadMapMutex_);
356 5 : HCCL_INFO("[ThreadMgr][%s] Hcom[%s] HcclThreadAcquire begin, max: engine[%d] threadNum[%u],"
357 : "notifyPerThread[%u], need: threadNum[%u], threadType[%d]",
358 : __func__, commId_.c_str(), engine, threadNum_, notifyNumPerThread_, threadNum, static_cast<int32_t>(type));
359 :
360 5 : if (threadNum == 0) {
361 0 : HCCL_ERROR("[ThreadMgr][HcclThreadAcquire] threadNum is 0");
362 0 : return HCCL_E_PARA;
363 : }
364 :
365 12 : for (u32 i = 0; i < threadNum; ++i) {
366 7 : CHK_PRT_RET(config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
367 : HCCL_ERROR("[ThreadMgr][%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x]",
368 : __func__, i, config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD), HCCL_E_PARA);
369 : }
370 :
371 : NotifyLoadType notifyLoadType;
372 : StreamType streamType;
373 5 : CHK_RET(GetNotifyLoadType(engine, type, notifyLoadType));
374 5 : CHK_RET(GetStreamType(engine, type, streamType));
375 :
376 5 : std::vector<std::shared_ptr<Thread>> newThreads;
377 5 : newThreads.reserve(threadNum);
378 5 : CHK_RET(CreateAndInitThreads(engine, streamType, notifyLoadType, threadNum, config, newThreads));
379 :
380 4 : std::unique_ptr<ThreadHandle[]> hostHandle;
381 4 : CHK_RET(AssignThreadHandles(engine, newThreads, threads, hostHandle));
382 :
383 10 : for (size_t i = 0; i < newThreads.size(); ++i) {
384 6 : uint32_t id = newThreads[i]->GetStream()->id();
385 6 : HCCL_DEBUG("[%s] thread id = [%u]", __func__, id);
386 6 : threadId.push_back(id);
387 : }
388 4 : threads_.reserve(threads_.size() + newThreads.size());
389 4 : CHK_RET(StoreThreadsAndBuildHandleMap(engine, newThreads, hostHandle));
390 :
391 4 : HCCL_INFO("[ThreadMgr][HcclThreadAcquire] Hcom[%s] HcclThreadAcquire done: engine[%s] threadNum[%u]%s",
392 : commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, (engine == COMM_ENGINE_AICPU) ? " (AICPU token ready)" : "");
393 4 : return HCCL_SUCCESS;
394 5 : }
395 :
396 4 : HcclResult ThreadMgr::HcclGetNotifyNumInThread(ThreadHandle thread, uint32_t *notifyNum)
397 : {
398 4 : CHK_PTR_NULL(notifyNum);
399 4 : Thread* hcclThread = reinterpret_cast<Thread*>(thread);
400 4 : CHK_PTR_NULL(hcclThread);
401 3 : *notifyNum = hcclThread->GetNotifyNum();
402 3 : HCCL_INFO("[ThreadMgr] Hcom[%s] HcclGetNotifyNumInThread done: notifyPerThread[%u]",
403 : commId_.c_str(), *notifyNum);
404 3 : return HCCL_SUCCESS;
405 : }
406 :
407 3 : HcclResult ThreadMgr::HcclThreadAcquireWithStream(CommEngine engine,
408 : rtStream_t stream, uint32_t notifyNum, ThreadHandle *thread)
409 : {
410 3 : CHK_PTR_NULL(thread);
411 :
412 3 : if (mainThread_.find(stream) != mainThread_.end()) {
413 0 : if (mainThread_[stream]->GetNotifyNum() < notifyNum) {
414 0 : u32 supplementNotifyNum = notifyNum - mainThread_[stream]->GetNotifyNum();
415 0 : CHK_RET(mainThread_[stream]->SupplementNotify(supplementNotifyNum));
416 : }
417 0 : *thread = reinterpret_cast<ThreadHandle>(mainThread_[stream].get());
418 0 : return HCCL_SUCCESS;
419 : }
420 :
421 : NotifyLoadType notifyLoadType;
422 3 : CHK_RET(CommHostEngineToNotifyLoadType(engine, notifyLoadType));
423 3 : std::shared_ptr<CpuTsThread> handle;
424 3 : EXCEPTION_CATCH(handle = std::make_shared<CpuTsThread>(stream, notifyNum, notifyLoadType), return HCCL_E_PTR);
425 3 : CHK_RET(handle->Init());
426 3 : handle->SetIsMaster(true);
427 :
428 : // 返回第一个句柄
429 3 : std::lock_guard<std::mutex> lock(mainThreadMutex_);
430 3 : mainThread_.emplace(stream, std::move(handle));
431 3 : *thread = reinterpret_cast<ThreadHandle>(mainThread_[stream].get());
432 3 : std::lock_guard<std::mutex> threadhandleToThreadMtx(threadhandleToThreadMutex_);
433 3 : threadMap_[*thread] = mainThread_[stream];
434 3 : HCCL_INFO("[ThreadMgr] Hcom[%s] HcclThreadAcquireWithStream done: engine[%s] stream[%p],"
435 : "notifyNum[%u]", commId_.c_str(), GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), stream, notifyNum);
436 3 : return HCCL_SUCCESS;
437 3 : }
438 :
439 2 : HcclResult ThreadMgr::ThreadExportToCommEngineCpu(uint32_t threadNum, const ThreadHandle *threads, ThreadHandle *exportedThreads)
440 : {
441 2 : std::lock_guard<std::mutex> lock(threadMapMutex_);
442 3 : for (u32 i = 0; i < threadNum; i++) {
443 2 : if (threadHandleOthersToCpu_.find(threads[i]) == threadHandleOthersToCpu_.end()) {
444 1 : HCCL_ERROR("[CommEngineResMgr]%s Unknown ThreadHandle[%llu]", __func__, threads[i]);
445 1 : return HCCL_E_PARA;
446 : }
447 1 : exportedThreads[i] = threadHandleOthersToCpu_[threads[i]];
448 : }
449 1 : return HCCL_SUCCESS;
450 2 : }
451 :
452 4 : HcclResult ThreadMgr::GetExportedThread(const ThreadHandle threadHandle, CommEngine commEngine, Thread *&exportedThread, std::shared_ptr<Thread> &threadOut)
453 : {
454 4 : Thread *threadPtr = reinterpret_cast<Thread *>(threadHandle);
455 4 : for (auto &thread : threads_) {
456 0 : if (thread.get() == threadPtr) {
457 0 : exportedThread = thread->FindThreadByCommEngine(commEngine);
458 0 : threadOut = thread;
459 0 : return HCCL_SUCCESS;
460 : }
461 : }
462 :
463 4 : for (auto &pair : mainThread_) {
464 2 : if (pair.second.get() == threadPtr) {
465 2 : exportedThread = pair.second->FindThreadByCommEngine(commEngine);
466 2 : threadOut = pair.second;
467 2 : return HCCL_SUCCESS;
468 : }
469 : }
470 :
471 2 : HCCL_ERROR("[ThreadMgr][%s]Unknown ThreadHandle[%llu]", __func__, threadHandle);
472 2 : return HCCL_E_PARA;
473 : }
474 :
475 4 : HcclResult ThreadMgr::ThreadExportToCommEngineAicpu(uint32_t threadNum, const ThreadHandle *threads, CommEngine dstCommEngine, ThreadHandle *exportedThreads)
476 : {
477 4 : std::vector<std::shared_ptr<Thread>> hostThreads;
478 4 : std::vector<u32> index;
479 : Thread *exportedThread;
480 6 : for (u32 i = 0; i < threadNum; i++) {
481 4 : std::shared_ptr<Thread> handle;
482 4 : CHK_RET(GetExportedThread(threads[i], dstCommEngine, exportedThread, handle));
483 2 : if (exportedThread != nullptr) {
484 0 : exportedThreads[i] = reinterpret_cast<ThreadHandle>(exportedThread);
485 0 : continue;
486 : } else {
487 2 : hostThreads.push_back(handle);
488 2 : index.push_back(i);
489 : }
490 4 : }
491 2 : if (!hostThreads.empty()) {
492 2 : std::lock_guard<std::mutex> lock(threadMapMutex_);
493 2 : if (!callbacks_.getAicpuCommState()) {
494 1 : HcclResult ret = callbacks_.kernelLaunchAicpuCommInit();
495 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
496 : HCCL_ERROR("[%s] kernelLaunchAicpuCommInit failed, return [%d].", __func__, ret), ret);
497 1 : callbacks_.setAicpuCommState(true);
498 : }
499 2 : std::unique_ptr<ThreadHandle[]> aicpuHandle;
500 2 : EXCEPTION_CATCH(aicpuHandle = std::make_unique<ThreadHandle[]>(hostThreads.size()),
501 : return HCCL_E_PTR);
502 2 : uint64_t beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
503 2 : HcclResult ret = AicpuLaunchMgr::ThreadKernelLaunchForComm(hostThreads, commId_, aicpuHandle, binHandle_);
504 2 : CHK_PRT_RET(ret != HCCL_SUCCESS,
505 : HCCL_ERROR("[ThreadMgr][HcclThreadExportToCommEngine] AiCpuKernelLaunch failed, return [%d].", ret), ret);
506 2 : if (callbacks_.reportProfilingKernel != nullptr) {
507 2 : ret = callbacks_.reportProfilingKernel(beginTime, "RunAicpuIndOpThreadInit");
508 1 : CHK_PRT_RET(ret != HCCL_SUCCESS,
509 : HCCL_ERROR("[ThreadMgr][HcclThreadExportToCommEngine] ReportProfilingAiCpuKernelLaunch failed, return [%d].", ret), ret);
510 : }
511 4 : for (size_t i = 0; i < hostThreads.size(); ++i) {
512 2 : exportedThreads[index[i]] = aicpuHandle[i];
513 2 : CHK_RET(hostThreads[i]->AddThreadHandleToMap(dstCommEngine, aicpuHandle[i]));
514 2 : threadHandleOthersToCpu_[aicpuHandle[i]] = threads[index[i]];
515 2 : HCCL_INFO("[ThreadMgr][%s] aicpu threadArray[%u] = [%llu]", __func__, i, aicpuHandle[i]);
516 : }
517 2 : }
518 2 : return HCCL_SUCCESS;
519 4 : }
520 :
521 6 : HcclResult ThreadMgr::HcclThreadExportToCommEngine(uint32_t threadNum, const ThreadHandle *threads, CommEngine dstCommEngine, ThreadHandle *exportedThreads)
522 : {
523 6 : switch (dstCommEngine) {
524 2 : case COMM_ENGINE_CPU_TS:
525 : case COMM_ENGINE_CPU:
526 : case COMM_ENGINE_CCU:
527 2 : CHK_RET(ThreadExportToCommEngineCpu(threadNum, threads, exportedThreads));
528 1 : break;
529 4 : case COMM_ENGINE_AICPU:
530 : case COMM_ENGINE_AICPU_TS:
531 4 : CHK_RET(ThreadExportToCommEngineAicpu(threadNum, threads, dstCommEngine, exportedThreads));
532 2 : break;
533 0 : case COMM_ENGINE_AIV:
534 : default:
535 0 : HCCL_ERROR("[ThreadMgr] Unknown comm engine type: %s", GetEnumToString(GetCommEngineStatusStrMap(), dstCommEngine).c_str());
536 0 : return HCCL_E_PARA;
537 : }
538 3 : return HCCL_SUCCESS;
539 : }
540 :
541 4 : HcclResult ThreadMgr::HcclThreadResGetInfo(ThreadHandle thread, ThreadResType resType, uint32_t infoLen, void **info)
542 : {
543 4 : CHK_PRT_RET(resType != ThreadResType::THREAD_RES_TYPE_STREAM, HCCL_ERROR("[%s] failed. resType[%d] is not supported.",
544 : __func__, static_cast<int32_t>(resType)), HCCL_E_NOT_SUPPORT);
545 :
546 3 : std::lock_guard<std::mutex> threadhandleToThreadMtx(threadhandleToThreadMutex_);
547 3 : auto it = threadMap_.find(thread);
548 3 : CHK_PRT_RET(it == threadMap_.end(),
549 : HCCL_ERROR("[%s] failed to find handle mapping in threadMap_, thread[0x%llx].", __func__, thread), HCCL_E_NOT_FOUND);
550 2 : std::shared_ptr<Thread> threadPtr = it->second;
551 2 : CHK_PTR_NULL(threadPtr);
552 2 : if (resType == ThreadResType::THREAD_RES_TYPE_STREAM) {
553 2 : CHK_PRT_RET(infoLen != sizeof(ThreadResTypeStream), HCCL_ERROR("[%s] failed. infoLen[%u] is mismatch sizeof(ThreadResTypeStream)[%zu]",
554 : __func__, infoLen, sizeof(ThreadResTypeStream)), HCCL_E_PARA);
555 1 : CHK_PTR_NULL(threadPtr->GetStream());
556 1 : ThreadResTypeStream stream = threadPtr->GetStream()->ptr();
557 1 : CHK_PTR_NULL(stream);
558 1 : *info = stream;
559 : } else {
560 0 : HCCL_ERROR("[%s] unsupported resType[%d]", __func__, static_cast<int32_t>(resType));
561 0 : return HCCL_E_NOT_SUPPORT;
562 : }
563 1 : HCCL_INFO("[%s] success. thread[0x%llx] resType[%d] info[%p]", __func__, thread, static_cast<int32_t>(resType), *info);
564 1 : return HCCL_SUCCESS;
565 3 : }
566 :
567 10 : HcclResult ThreadMgr::HcclUnfoldThreadAcquire(HcclDedicatedThreadType useType, uint32_t notifyNumPerThread, ThreadHandle *thread)
568 : {
569 10 : CHK_PRT_RET(thread == nullptr, HCCL_ERROR("[%s] thread is null", __func__), HCCL_E_PTR);
570 10 : auto it = dedicatedThreadMap_.find(useType);
571 10 : if (it != dedicatedThreadMap_.end()) {
572 4 : *thread = it->second;
573 4 : HCCL_INFO("[%s] reuse dedicated thread, dedThreadType[%u], thread[0x%llx]", __func__, useType, *thread);
574 4 : CHK_RET(SupplementThreadNotify(*thread, notifyNumPerThread));
575 : } else {
576 6 : if (useType == HCCL_DED_THREAD_TYPE_AICPU_LAUNCH_GE) {
577 1 : *thread = 0;
578 1 : HCCL_WARNING("[%s] dedicated thread not found, dedThreadType[%u], return threadHandle[0]", __func__, useType);
579 1 : return HCCL_SUCCESS;
580 : }
581 5 : CommEngine engine = CommEngine::COMM_ENGINE_CPU;
582 5 : uint32_t notifyNumPerThreadVec[1] = { notifyNumPerThread };
583 5 : HcclResult ret = static_cast<HcclResult>(HcommThreadAlloc(engine, 1, notifyNumPerThreadVec, thread));
584 5 : if (ret != HCCL_SUCCESS) {
585 0 : HCCL_ERROR("[%s] Failed to cache dedicated thread, dedThreadType[%u], ret[%d]", __func__, useType, ret);
586 0 : return ret;
587 : }
588 5 : dedicatedThreadMap_[useType] = *thread;
589 : }
590 9 : return HCCL_SUCCESS;
591 : }
592 :
593 10 : HcclResult ThreadMgr::HcclDedicatedThreadAcquire(HcclDedicatedThreadType useType, uint32_t notifyNumPerThread, ThreadHandle *thread)
594 : {
595 10 : CHK_PRT_RET(thread == nullptr, HCCL_ERROR("[%s] thread is null", __func__), HCCL_E_PTR);
596 10 : CHK_PRT_RET(useType == HCCL_DED_THREAD_TYPE_INVALID, HCCL_ERROR("[%s] dedThreadType is invalid", __func__), HCCL_E_PARA);
597 10 : HCCL_INFO("Entry-%s: dedThreadType[%u] notifyNumPerThread[%u]", __func__, useType, notifyNumPerThread);
598 :
599 10 : std::lock_guard<std::mutex> lock(dedicatedThreadMutex_);
600 10 : if (useType == HCCL_DED_THREAD_TYPE_AICPU_LAUNCH || useType == HCCL_DED_THREAD_TYPE_AICPU_LAUNCH_GE) {
601 10 : CHK_RET(HcclUnfoldThreadAcquire(useType, notifyNumPerThread, thread));
602 10 : } else {
603 0 : HCCL_ERROR("[%s] unsupport dedThreadType[%u]", __func__, useType);
604 0 : return HCCL_E_NOT_SUPPORT;
605 : }
606 :
607 10 : HCCL_INFO("[%s] success, useType[%u], thread[0x%llx], notifyNumPerThread[%u]", __func__, useType, *thread,
608 : notifyNumPerThread);
609 10 : return HCCL_SUCCESS;
610 10 : }
611 : }
|
