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 "aicpu_thread_process.h"
12 : #include <iomanip>
13 : #include "exception_handle.h"
14 : #include "stream_lite.h"
15 :
16 : using namespace hccl;
17 :
18 : std::shared_mutex AicpuThreadProcess::mutex_;
19 : std::vector<std::shared_ptr<hccl::Thread>> AicpuThreadProcess::threads_;
20 : std::mutex AicpuThreadProcess::bgThreadMutex_;
21 : bool AicpuThreadProcess::daemonFuncRegistered_ = false;
22 : Hccl::CommandToBackGroud AicpuThreadProcess::commandToBackGroud_ = Hccl::CommandToBackGroud::Default;
23 : std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> AicpuThreadProcess::defaultDfxCallback_;
24 :
25 0 : HcclResult AicpuThreadProcess::InitThreads(ThreadMgrAicpuParam* param)
26 : {
27 0 : CHK_PTR_NULL(param);
28 0 : u32 threadNum = param->threadNum;
29 0 : std::vector<std::shared_ptr<Thread>> outThreads;
30 0 : outThreads.reserve(threadNum);
31 0 : std::string hcomId(param->hcomId);
32 0 : CHK_RET(AicpuThreadProcess::ResumeThread(param, outThreads, false));
33 :
34 : // 由调用方 AicpuThreadInit 持有 mutex_ 写锁保护，此处 check-then-set 无并发风险
35 0 : if (!defaultDfxCallback_) { // HcommThreadAlloc接口暂未适配profiling和上报task的能力
36 0 : defaultDfxCallback_ = [](u32 streamId, u32 taskId, const Hccl::TaskParam& taskParam, u64 handle) {
37 0 : HCCL_DEBUG("[AicpuThreadProcess] order launch dfx callback, streamId[%u], taskId[%u]", streamId, taskId);
38 0 : return HCCL_SUCCESS;
39 0 : };
40 : }
41 :
42 0 : ThreadHandle* threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
43 : // 空指针校验
44 0 : CHK_PTR_NULL(threadArray);
45 0 : for (size_t i = 0; i < threadNum; ++i) {
46 0 : threadArray[i] = reinterpret_cast<ThreadHandle>(outThreads[i].get()); // 拷贝裸指针
47 0 : HCCL_INFO("[AicpuThreadProcess][%s] threadArray[%zu] = [%lu]", __func__, i, threadArray[i]);
48 0 : int32_t ret = HcommThreadRegisterDfx(threadArray[i], defaultDfxCallback_);
49 0 : if (ret != 0) {
50 0 : HCCL_WARNING(
51 : "[AicpuThreadProcess][%s] HcommThreadRegisterDfx failed, ret[%d], threadArray[%zu]", __func__, ret, i);
52 : }
53 : }
54 0 : threads_.insert(
55 0 : threads_.end(), std::make_move_iterator(outThreads.begin()), std::make_move_iterator(outThreads.end()));
56 0 : HCCL_INFO(
57 : "[AicpuThreadProcess][%s] comm identifier[%s], init threads num[%u] success", __func__, hcomId.c_str(),
58 : threadNum);
59 0 : return HCCL_SUCCESS;
60 0 : }
61 :
62 1 : const std::vector<std::shared_ptr<hccl::Thread>>& AicpuThreadProcess::GetThreads() { return threads_; }
63 :
64 1 : std::shared_mutex& AicpuThreadProcess::GetMutex() { return mutex_; }
65 :
66 0 : HcclResult AicpuThreadProcess::AicpuThreadInit(ThreadMgrAicpuParam* param)
67 : {
68 0 : CHK_RET(hrtSetWorkModeAicpu(true));
69 0 : CHK_RET(hrtSetlocalDevice(param->deviceLogicId));
70 0 : CHK_RET(hrtSetlocalDeviceType(static_cast<DevType>(param->deviceType)));
71 : {
72 0 : std::unique_lock<std::shared_mutex> rwlock(mutex_);
73 0 : HcclResult ret = InitThreads(param);
74 0 : CHK_PRT_RET(
75 : ret != HCCL_SUCCESS,
76 : HCCL_ERROR(
77 : "[AicpuThreadProcess][AicpuThreadInit]errNo[0x%016llx] Failed to init threads", HCCL_ERROR_CODE(ret)),
78 : ret);
79 0 : }
80 :
81 0 : if (static_cast<DevType>(param->deviceType) == DevType::DEV_TYPE_950
82 0 : || static_cast<DevType>(param->deviceType) == DevType::DEV_TYPE_960) {
83 0 : InitBackGroundThread();
84 : }
85 0 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult AicpuThreadProcess::AicpuThreadDestroy(ThreadMgrAicpuParam* param)
89 : {
90 0 : HCCL_INFO("[AicpuThreadProcess][%s] threadNum[%u]", __func__, param->threadNum);
91 :
92 0 : bool needStopBgThread = false;
93 : {
94 0 : std::unique_lock<std::shared_mutex> rwlock(mutex_);
95 0 : ThreadHandle* threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
96 0 : if (threadArray == nullptr) {
97 0 : HCCL_ERROR("[AicpuThreadProcess][%s] threadArray is nullptr", __func__);
98 0 : return HCCL_E_PTR;
99 : }
100 :
101 0 : for (u32 i = 0; i < param->threadNum; ++i) {
102 0 : ThreadHandle handle = threadArray[i];
103 0 : auto it = std::find_if(threads_.begin(), threads_.end(), [handle](const std::shared_ptr<Thread>& ptr) {
104 0 : return reinterpret_cast<ThreadHandle>(ptr.get()) == handle;
105 : });
106 0 : if (it == threads_.end()) {
107 0 : HCCL_WARNING("[AicpuThreadProcess][%s] thread handle[0x%llx] not found in threads_", __func__, handle);
108 0 : continue;
109 : }
110 0 : Hccl::StreamLite* streamLite = static_cast<Hccl::StreamLite*>((*it)->GetStreamLitePtr());
111 0 : if (streamLite != nullptr) {
112 0 : hcomm::ExceptionHandle::GetInstance().ClearStreamState(streamLite->GetSqId());
113 : }
114 0 : threads_.erase(it);
115 0 : HCCL_DEBUG("[AicpuThreadProcess][%s] destroyed thread handle[0x%llx]", __func__, handle);
116 : }
117 :
118 0 : if (threads_.empty()) {
119 0 : needStopBgThread = true;
120 : }
121 0 : }
122 :
123 0 : if (needStopBgThread) {
124 0 : StopBackGroundThread();
125 : }
126 :
127 0 : HCCL_INFO("[AicpuThreadProcess][%s] success", __func__);
128 0 : return HCCL_SUCCESS;
129 : }
130 :
131 0 : void AicpuThreadProcess::InitBackGroundThread()
132 : {
133 0 : std::lock_guard<std::mutex> lock(bgThreadMutex_);
134 0 : if (daemonFuncRegistered_) {
135 0 : HCCL_INFO("[AicpuThreadProcess][%s] background thread already started, skip.", __func__);
136 0 : return;
137 : }
138 : // 注册守护进程函数
139 0 : Hccl::AicpuDaemonService::GetInstance().Register(&hcomm::ExceptionHandle::GetInstance());
140 0 : daemonFuncRegistered_ = true;
141 :
142 0 : static auto daemonServiceRun = [](void* info) {
143 0 : Hccl::AicpuDaemonService::GetInstance().ServiceRun(info);
144 0 : };
145 0 : static auto daemonServiceStop = [](void* info) {
146 0 : Hccl::AicpuDaemonService::GetInstance().ServiceStop(info);
147 0 : };
148 :
149 0 : commandToBackGroud_ = Hccl::CommandToBackGroud::Default;
150 :
151 : // 启动背景线程，背景线程在runtime实现有保护，背景线程已经启动后会直接返回。
152 0 : if (Hccl::StartMC2MaintenanceThread != nullptr) {
153 0 : Hccl::StartMC2MaintenanceThread(
154 : daemonServiceRun, &commandToBackGroud_, daemonServiceStop, &commandToBackGroud_);
155 0 : HCCL_RUN_INFO("[%s]start BackGround thread success.", __func__);
156 : } else {
157 0 : HCCL_WARNING("[%s]StartMC2MaintenanceThread func is nullptr", __func__);
158 : }
159 0 : }
160 :
161 0 : void AicpuThreadProcess::StopBackGroundThread()
162 : {
163 0 : std::lock_guard<std::mutex> lock(bgThreadMutex_);
164 : // 背景线程是同集合通信共用，这里不停止背景线程，只是将守护函数注销
165 0 : Hccl::AicpuDaemonService::GetInstance().Unregister(&hcomm::ExceptionHandle::GetInstance());
166 0 : daemonFuncRegistered_ = false;
167 0 : HCCL_INFO("[AicpuThreadProcess][%s] success", __func__);
168 0 : }
169 :
170 1 : HcclResult AicpuThreadProcess::ResumeThread(
171 : ThreadMgrAicpuParam* param, std::vector<std::shared_ptr<Thread>>& outThreads, bool isSupplementNotify)
172 : {
173 1 : CHK_PTR_NULL(param);
174 1 : u32 threadNum = param->threadNum;
175 1 : std::string hcomId(param->hcomId);
176 1 : ThreadHandle* threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
177 2 : for (u32 i = 0; i < threadNum; ++i) {
178 1 : std::string thdUniqueId(param->threadParam[i], THREAD_UNIQUE_ID_MAX_SIZE);
179 1 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
180 1 : std::ostringstream oss;
181 1 : oss << "threadParam[" << i << "] raw bytes: ";
182 1 : constexpr u32 HEX_WIDTH = 2;
183 6001 : for (u32 j = 0; j < THREAD_UNIQUE_ID_MAX_SIZE; ++j) {
184 6000 : oss << std::hex << std::setw(HEX_WIDTH) << std::setfill('0')
185 6000 : << static_cast<unsigned int>(static_cast<unsigned char>(param->threadParam[i][j])) << " ";
186 : }
187 1 : HCCL_INFO("[AicpuThreadProcess][%s] %s", __func__, oss.str().c_str());
188 1 : }
189 1 : std::shared_ptr<AicpuTsThread> thread;
190 1 : EXCEPTION_CATCH((thread = std::make_shared<AicpuTsThread>(thdUniqueId)), return HCCL_E_PTR);
191 1 : u32 notifyNum = 0;
192 1 : std::string notifyDesc;
193 1 : CHK_RET(thread->GetNotifyByUniqueId(notifyNum, notifyDesc));
194 1 : if (isSupplementNotify) {
195 1 : AicpuTsThread* threadPtr = reinterpret_cast<AicpuTsThread*>(threadArray[i]);
196 1 : CHK_PTR_NULL(threadPtr);
197 1 : HCCL_INFO(
198 : "[%s]threadIdx[%u], threadHandle[%llu], notifyNum[%u], newNotifyNum[%u]", __func__, i, threadArray[i],
199 : threadPtr->GetNotifyNum(), notifyNum);
200 1 : CHK_RET(threadPtr->SupplementNotify(notifyNum, notifyDesc));
201 : } else {
202 0 : HcclResult ret = thread->Init();
203 0 : if (ret != HCCL_SUCCESS) {
204 0 : HCCL_ERROR(
205 : "[AicpuThreadProcess][%s] comm identifier[%s], init threads num[%u] failed at index %u", __func__,
206 : hcomId.c_str(), param->threadNum, i);
207 0 : return ret;
208 : }
209 0 : outThreads.emplace_back(thread);
210 : }
211 1 : }
212 1 : return HCCL_SUCCESS;
213 1 : }
214 :
215 1 : HcclResult AicpuThreadProcess::AicpuThreadSupplementNotify(ThreadMgrAicpuParam* param)
216 : {
217 1 : CHK_PTR_NULL(param);
218 1 : u32 threadNum = param->threadNum;
219 1 : std::string hcomId(param->hcomId);
220 1 : std::vector<std::shared_ptr<Thread>> outThreads;
221 1 : CHK_RET(AicpuThreadProcess::ResumeThread(param, outThreads, true));
222 :
223 1 : HCCL_INFO(
224 : "[AicpuThreadProcess][%s] comm identifier[%s], init threads num[%u] success", __func__, hcomId.c_str(),
225 : threadNum);
226 1 : return HCCL_SUCCESS;
227 1 : }
|
