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