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