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 <unordered_map>
12 : #include <vector>
13 :
14 : #include "hcomm_c_adpt.h"
15 : #include "hcomm_c_adpt_common.h"
16 : #include "hcomm_res.h"
17 : #include "hcomm_res_defs.h"
18 : #include "log.h"
19 : #include "thread.h"
20 : #include "cpu_ts_thread.h"
21 : #include "param_check_pub.h"
22 : #include "comm_engine_utils.h"
23 : #include "exception_handler.h"
24 : #include "adapter_rts_common.h"
25 : #include "aicpu_ts_channel_helper.h"
26 :
27 : namespace hcomm {
28 : static std::unordered_map<ThreadHandle, std::shared_ptr<hccl::Thread>> g_ThreadMap;
29 : } // namespace hcomm
30 :
31 : using namespace hcomm;
32 :
33 15 : HcommResult HcommThreadAlloc(
34 : CommEngine engine, uint32_t threadNum, const uint32_t *notifyNumPerThread, ThreadHandle *threads)
35 : {
36 15 : CHK_PTR_NULL(threads);
37 14 : CHK_PTR_NULL(notifyNumPerThread);
38 14 : (void)HcommResMgrInit();
39 14 : const uint32_t notifyNum = notifyNumPerThread[0];
40 14 : if (threadNum > 1U) {
41 7 : HCCL_RUN_WARNING("[%s] only notifyNumPerThread[0] is used currently, threadNum[%u], notifyNum[0][%u].",
42 : __func__, threadNum, notifyNum);
43 : }
44 14 : HCCL_INFO("[%s] ThreadAcquire begin. engine[%s], threadNum[%u], notifyPerThread[%u], threads[%p]", __func__,
45 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum, threads);
46 14 : CHK_RET(RefreshCommEngineContext(engine));
47 :
48 : // 1. 参数校验
49 14 : CHK_RET(hccl::ValidateThreadParams(threadNum, notifyNum));
50 :
51 : // 2. 获取引擎对应的类型
52 : hccl::NotifyLoadType notifyLoadType;
53 : hccl::StreamType streamType;
54 11 : CHK_RET(hccl::CommEngineToNotifyLoadType(engine, notifyLoadType));
55 10 : CHK_RET(hccl::CommEngineToStreamType(engine, streamType));
56 :
57 : // 3. 创建线程
58 10 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
59 10 : hccl::ThreadCreateParams params(engine, threadNum, notifyNum, notifyLoadType, streamType);
60 10 : CHK_RET(hccl::CreateAndInitThreads(params, newThreads));
61 :
62 : // 4. 插入全局映射表
63 8 : CHK_RET(hccl::SaveThreads(newThreads));
64 :
65 : // 5. 储存线程句柄
66 8 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
67 8 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
68 :
69 8 : HCCL_INFO("[HcommThreadAlloc] ThreadAcquire done: engine[%s] threadNum[%u], notifyPerThread[%u]",
70 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), threadNum, notifyNum);
71 8 : return HCCL_SUCCESS;
72 10 : }
73 :
74 10 : HcommResult HcommThreadAlloc(CommEngine engine, uint32_t threadNum, uint32_t notifyNumPerThread, ThreadHandle *threads)
75 : {
76 10 : return ::HcommThreadAlloc(engine, threadNum, ¬ifyNumPerThread, threads);
77 : }
78 :
79 0 : HcommResult HcommThreadAllocWithConfig(
80 : CommEngine engine, uint32_t threadNum, ThreadType type, const ThreadConfig *config, ThreadHandle *threads)
81 : {
82 0 : CHK_PTR_NULL(threads);
83 0 : CHK_PTR_NULL(config);
84 0 : CHK_PRT_RET(type == THREAD_TYPE_INVALID,
85 : HCCL_ERROR("[%s] thread type[%d] is invalid", __func__, static_cast<int32_t>(type)), (HcommResult)HCCL_E_PARA);
86 0 : CHK_PRT_RET(engine == COMM_ENGINE_AICPU_TS || engine == COMM_ENGINE_CPU_TS,
87 : HCCL_ERROR("[%s] commEngine[%d] CPU_TS/AICPU_TS not supported, use engine with ThreadType instead", __func__,
88 : static_cast<int32_t>(engine)),
89 : (HcommResult)HCCL_E_PARA);
90 0 : CHK_PRT_RET(engine == COMM_ENGINE_AIV || engine == COMM_ENGINE_CCU,
91 : HCCL_ERROR("[%s] commEngine[%d] AIV/CCU not supported, supported engines: CPU/AICPU", __func__,
92 : static_cast<int32_t>(engine)),
93 : (HcommResult)HCCL_E_PARA);
94 0 : CHK_PRT_RET(
95 : threadNum == 0, HCCL_ERROR("[%s] threadNum[%u] is invalid", __func__, threadNum), (HcommResult)HCCL_E_PARA);
96 0 : HcommResult hcommRet = HcommResMgrInit();
97 0 : CHK_PRT_RET(hcommRet != HCCL_SUCCESS,
98 : HCCL_ERROR("[%s] HcommResMgrInit failed, ret[%d]", __func__, static_cast<int32_t>(hcommRet)), hcommRet);
99 0 : CHK_RET(RefreshCommEngineContext(engine));
100 :
101 0 : HCCL_INFO("[%s] begin. engine[%d], threadType[%d], threadNum[%u], threads[%p]", __func__, engine,
102 : static_cast<int32_t>(type), threadNum, threads);
103 :
104 : hccl::NotifyLoadType notifyLoadType;
105 : hccl::StreamType streamType;
106 0 : CHK_RET(hccl::GetNotifyLoadType(engine, type, notifyLoadType));
107 0 : CHK_RET(hccl::GetStreamType(engine, type, streamType));
108 :
109 0 : std::vector<std::shared_ptr<hccl::Thread>> newThreads;
110 0 : newThreads.reserve(threadNum);
111 0 : for (uint32_t i = 0; i < threadNum; ++i) {
112 0 : CHK_PRT_RET(config[i].header.magicWord != HCOMM_THREAD_CONFIG_MAGIC_WORD,
113 : HCCL_ERROR("[%s] config[%u] magicWord[0x%x] mismatch, expected[0x%x], call ThreadConfigInit first",
114 : __func__, i, config[i].header.magicWord, HCOMM_THREAD_CONFIG_MAGIC_WORD),
115 : (HcommResult)HCCL_E_PARA);
116 0 : CHK_RET(hccl::ValidateThreadParams(1, config[i].notifyNumPerThread));
117 0 : std::shared_ptr<hccl::Thread> threadPtr;
118 : HcclResult ret
119 0 : = hccl::CreateThread(engine, streamType, config[i].notifyNumPerThread, notifyLoadType, threadPtr);
120 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
121 : HCCL_ERROR("[%s] Failed to create thread at index[%u], ret[%d]", __func__, i, ret), (HcommResult)ret);
122 0 : ret = threadPtr->Init();
123 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
124 : HCCL_ERROR("[%s] Failed to init thread at index[%u], ret[%d]", __func__, i, ret), (HcommResult)ret);
125 0 : newThreads.emplace_back(std::move(threadPtr));
126 0 : }
127 :
128 0 : CHK_RET(hccl::SaveThreads(newThreads));
129 0 : CHK_RET(AicpuTsChannelHelper::EnsureKernelBinLoaded(engine));
130 0 : CHK_RET(hccl::StoreThreadHandles(newThreads, threads, engine, AicpuTsChannelHelper::GetBinHandle()));
131 :
132 0 : HCCL_INFO(
133 : "[%s] done: engine[%d] threadType[%d] threadNum[%u]", __func__, engine, static_cast<int32_t>(type), threadNum);
134 0 : return HCCL_SUCCESS;
135 0 : }
136 :
137 9 : HcommResult HcommThreadFree(const ThreadHandle *threads, uint32_t threadNum)
138 : {
139 9 : CHK_PTR_NULL(threads);
140 8 : (void)HcommResMgrInit();
141 8 : return hccl::FreeThreads(threads, threadNum, AicpuTsChannelHelper::GetBinHandle());
142 : }
143 :
144 9 : HcommResult HcommThreadAllocWithStream(CommEngine engine, rtStream_t stream, uint32_t notifyNum, ThreadHandle *thread)
145 : {
146 9 : CHK_PTR_NULL(thread);
147 : hccl::NotifyLoadType notifyLoadType;
148 8 : CHK_RET(CommHostEngineToNotifyLoadType(engine, notifyLoadType));
149 7 : std::shared_ptr<hccl::Thread> handle;
150 7 : EXCEPTION_CATCH(handle = std::make_shared<hccl::CpuTsThread>(stream, notifyNum, notifyLoadType), return HCCL_E_PTR);
151 7 : CHK_RET(handle->Init());
152 :
153 : // 返回第一个句柄
154 6 : *thread = reinterpret_cast<ThreadHandle>(handle.get());
155 6 : hcomm::g_ThreadMap.emplace(*thread, handle);
156 :
157 6 : HCCL_INFO("[ThreadMgr] ThreadAcquireWithStream done: engine[%s] stream[%p],"
158 : "notifyNum[%u]",
159 : GetEnumToString(GetCommEngineStatusStrMap(), engine).c_str(), stream, notifyNum);
160 6 : return HCCL_SUCCESS;
161 7 : }
|
