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 :
11 : #include "thread_aicpu_mgr.h"
12 : #include "aicpu_ts_thread.h"
13 : #include "hcclCommOp.h"
14 : #include "aicpu_task_cache_manager.h"
15 : #include "stream_lite.h"
16 : #include "rtsq_a5.h"
17 : #include "log.h"
18 : #include <sstream>
19 : #include <iomanip>
20 :
21 24 : ThreadAicpuMgr::ThreadAicpuMgr(HcclCommDfxLite& dfx, std::function<HcclResult(bool)> checkExecStatusCallback)
22 24 : : dfx_(dfx),
23 24 : checkExecStatusCallback_(std::move(checkExecStatusCallback))
24 24 : {}
25 :
26 24 : ThreadAicpuMgr::~ThreadAicpuMgr()
27 : {
28 24 : std::unique_lock<std::shared_mutex> rwLock(threadMutex_);
29 28 : for (auto& thread : threads_) {
30 4 : HcommThreadRegisterCheckExecStatus(reinterpret_cast<ThreadHandle>(thread.get()), nullptr);
31 : }
32 24 : threads_.clear();
33 24 : }
34 :
35 3 : HcclResult ThreadAicpuMgr::InitThreads(ThreadMgrAicpuParam* param)
36 : {
37 3 : CHK_PTR_NULL(param);
38 1 : u32 threadNum = param->threadNum;
39 1 : std::vector<std::shared_ptr<Thread>> outThreads;
40 1 : outThreads.reserve(threadNum);
41 1 : std::string hcomId(param->hcomId);
42 2 : for (u32 i = 0; i < threadNum; ++i) {
43 1 : std::string thdUniqueId(param->threadParam[i], THREAD_UNIQUE_ID_MAX_SIZE);
44 1 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
45 1 : std::ostringstream oss;
46 1 : oss << "threadParam[" << i << "] raw bytes: ";
47 6001 : for (u32 j = 0; j < THREAD_UNIQUE_ID_MAX_SIZE; ++j) {
48 6000 : oss << std::hex << std::setw(2) << std::setfill('0')
49 6000 : << static_cast<unsigned int>(static_cast<unsigned char>(param->threadParam[i][j])) << " ";
50 : }
51 1 : HCCL_INFO("[ThreadAicpuMgr][%s] %s", __func__, oss.str().c_str());
52 1 : }
53 1 : std::shared_ptr<AicpuTsThread> thread;
54 1 : EXCEPTION_CATCH((thread = std::make_shared<AicpuTsThread>(thdUniqueId)), return HCCL_E_PTR);
55 1 : HcclResult ret = thread->Init();
56 1 : if (ret != HCCL_SUCCESS) {
57 0 : HCCL_ERROR(
58 : "[ThreadAicpuMgr][%s] comm identifier[%s], init threads num[%u] failed at index %u", __func__,
59 : hcomId.c_str(), param->threadNum, i);
60 0 : return ret;
61 : }
62 1 : outThreads.emplace_back(thread);
63 1 : }
64 :
65 1 : ThreadHandle* threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
66 1 : CHK_PTR_NULL(threadArray);
67 2 : for (size_t i = 0; i < outThreads.size(); ++i) {
68 1 : threadArray[i] = reinterpret_cast<ThreadHandle>(outThreads[i].get());
69 1 : HCCL_INFO(
70 : "[ThreadAicpuMgr][%s] threadArray[%zu] = [%llu]", __func__, i,
71 : static_cast<unsigned long long>(threadArray[i]));
72 1 : CHK_RET(RegisterThreadAddDfxTaskInfo(threadArray[i]));
73 1 : CHK_RET(RegisterThreadCacheCallback(static_cast<AicpuTsThread*>(outThreads[i].get())));
74 : }
75 1 : std::unique_lock<std::shared_mutex> rwLock(threadMutex_);
76 2 : threads_.insert(
77 1 : threads_.end(), std::make_move_iterator(outThreads.begin()), std::make_move_iterator(outThreads.end()));
78 1 : HCCL_INFO(
79 : "[ThreadAicpuMgr][%s] comm identifier[%s], init threads num[%u] success", __func__, hcomId.c_str(), threadNum);
80 1 : return HCCL_SUCCESS;
81 1 : }
82 :
83 0 : HcclResult ThreadAicpuMgr::RegisterThreadAddDfxTaskInfo(ThreadHandle thread)
84 : {
85 0 : int32_t ret = HcommThreadRegisterCheckExecStatus(thread, checkExecStatusCallback_);
86 0 : if (ret != 0) {
87 0 : HCCL_ERROR(
88 : "[%s]HcommThreadRegisterCheckExecStatus failed, ret[%d], thread[0x%llx], checkExecStatusCallback[%p]",
89 : __func__, ret, static_cast<unsigned long long>(thread),
90 : static_cast<const void*>(&checkExecStatusCallback_));
91 0 : return HCCL_E_PTR;
92 : }
93 :
94 0 : std::function<void(Hccl::TaskInfoCircularQueue*)> reportCallback = [this](Hccl::TaskInfoCircularQueue* taskQueue) {
95 0 : dfx_.ReportStreamTask(taskQueue);
96 0 : };
97 0 : ret = HcommNewThreadRegisterDfx(thread, reportCallback);
98 0 : if (ret != 0) {
99 0 : HCCL_ERROR("[%s] HcommNewThreadRegisterDfx failed, ret[%d], thread[0x%llx]", __func__, ret, thread);
100 0 : return HCCL_E_PTR;
101 : }
102 :
103 0 : std::function<const void*()> getLatestOpInfoCallback = [this]() -> const void* {
104 0 : return dfx_.GetLatestDfxOpInfo();
105 0 : };
106 0 : ret = HcommNewThreadRegisterGetLatestDfxOpInfo(thread, getLatestOpInfoCallback);
107 0 : if (ret != 0) {
108 0 : HCCL_ERROR(
109 : "[%s] HcommNewThreadRegisterGetLatestDfxOpInfo failed, ret[%d], thread[0x%llx]", __func__, ret, thread);
110 0 : return HCCL_E_PTR;
111 : }
112 :
113 0 : return HCCL_SUCCESS;
114 0 : }
115 :
116 1 : HcclResult ThreadAicpuMgr::RegisterThreadCacheCallback(AicpuTsThread* thread)
117 : {
118 1 : HCCL_INFO("[ThreadAicpuMgr][%s] register cache callback for thread[%p]", __func__, thread);
119 1 : CHK_PTR_NULL(thread);
120 1 : auto* const streamLitePtr = static_cast<Hccl::StreamLite*>(thread->GetStreamLitePtr());
121 1 : CHK_PTR_NULL(streamLitePtr);
122 1 : Hccl::RtsqA5* rtsqA5 = static_cast<Hccl::RtsqA5*>(streamLitePtr->GetRtsq());
123 1 : CHK_PTR_NULL(rtsqA5);
124 1 : CHK_RET(rtsqA5->SetAicpuTsThreadPtr(thread));
125 1 : CHK_RET(rtsqA5->SetNeedCacheTaskCallback(hcomm::AicpuTaskCacheManager::NeedCacheTask));
126 1 : CHK_RET(rtsqA5->SetAddSqeArrayCallback(hcomm::AicpuTaskCacheManager::AddSqeArray));
127 1 : HCCL_INFO("[ThreadAicpuMgr][%s] register cache callback for thread[%p] success", __func__, thread);
128 1 : return HCCL_SUCCESS;
129 : }
|
