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 : #ifndef THREAD_H
12 : #define THREAD_H
13 :
14 : #include <string>
15 : #include <vector>
16 : #include <memory>
17 : #include "hccl_types.h"
18 : #include "hccl_common.h"
19 : #include "hcomm_primitives.h"
20 : #include "local_notify.h"
21 : #include "stream_pub.h"
22 : #include "acl/acl_rt.h"
23 : #include "adapter_hal_pub.h"
24 : #include "device_capacity.h"
25 : #include "task_param.h"
26 : #include "sal_pub.h"
27 : #include "stream_lite.h"
28 : #include "task_info.h"
29 : #include "adapter_prof.h"
30 : #include "../../../../../legacy/ascend950/framework/dfx/profiling/dlprof_function_v2.h"
31 :
32 : namespace hccl {
33 :
34 : struct ThreadCreateParams {
35 : CommEngine engine; // 通信引擎类型
36 : uint32_t threadNum; // 线程数量
37 : uint32_t notifyNumPerThread; // 每个线程的通知量数量
38 : NotifyLoadType notifyLoadType; // 通知量加载类型
39 : StreamType streamType; // 流类型
40 :
41 : // 默认构造函数
42 : ThreadCreateParams()
43 : : engine(COMM_ENGINE_RESERVED),
44 : threadNum(0),
45 : notifyNumPerThread(0),
46 : notifyLoadType(NotifyLoadType::HOST_NOTIFY),
47 : streamType(StreamType::STREAM_TYPE_RESERVED)
48 : {}
49 :
50 : // 带参数的构造函数
51 17 : ThreadCreateParams(CommEngine engine, uint32_t tNum, uint32_t nNum, NotifyLoadType nType, StreamType sType)
52 17 : : engine(engine),
53 17 : threadNum(tNum),
54 17 : notifyNumPerThread(nNum),
55 17 : notifyLoadType(nType),
56 17 : streamType(sType)
57 17 : {}
58 : };
59 :
60 : constexpr u32 HCOMM_NOTIFY_MAX_NUM = 64;
61 : constexpr u32 HCOMM_THREADNUM_MAX_NUM = 1000;
62 : constexpr u32 HCCL_THREAD_NOTIFY_MAX_NUM = 65536;
63 : /**
64 : * @note 职责：通信引擎的Thread的C++抽象接口类，表达并行资源，内部包含thread间的同步Notify。
65 : */
66 : class Thread {
67 : public:
68 436 : virtual ~Thread() = default;
69 : virtual HcclResult Init() = 0;
70 : virtual HcclResult DeInit() = 0;
71 : virtual std::string& GetUniqueId() = 0;
72 : virtual uint32_t GetNotifyNum() const = 0;
73 : virtual LocalNotify* GetNotify(uint32_t index) const = 0;
74 : virtual HcclResult SupplementNotify(uint32_t notifyNum) = 0;
75 :
76 : // A3 Stream & A5 Stream
77 : virtual bool IsDeviceA5() const = 0;
78 : virtual Stream* GetStream() const = 0;
79 : virtual void* GetStreamLitePtr() const = 0;
80 : virtual void LaunchTask() const = 0;
81 : virtual void TryLaunchTask() const = 0;
82 :
83 : // Local Data Plane Functions
84 : virtual HcclResult LocalNotifyRecord(uint32_t notifyId) const = 0;
85 : virtual HcclResult LocalNotifyWait(uint32_t notifyId) const = 0;
86 :
87 : virtual HcclResult LocalNotifyRecord(ThreadHandle dstThread, uint32_t dstNotifyIdx) const = 0;
88 : virtual HcclResult LocalNotifyWait(uint32_t notifyIdx, uint32_t timeOut) const = 0;
89 :
90 : virtual HcclResult LocalCopy(void* dst, const void* src, uint64_t sizeByte) const = 0;
91 : virtual HcclResult
92 : LocalReduce(void* dst, const void* src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp) const
93 : = 0;
94 : virtual bool GetMaster() const = 0;
95 : virtual void SetIsMaster(bool isMaster) = 0;
96 :
97 : HcclResult AddThreadHandleToMap(CommEngine commEngine, ThreadHandle threadHandle);
98 : Thread* FindThreadByCommEngine(CommEngine commEngine);
99 112 : HcclResult SetAddTaskInfoCallback(std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback)
100 : {
101 112 : CHK_PTR_NULL(callback);
102 112 : callback_ = callback;
103 112 : return HCCL_SUCCESS;
104 : }
105 8 : std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> GetCallback() { return callback_; }
106 2 : virtual HcclResult SetCheckExecStatusCallback([[maybe_unused]] std::function<HcclResult(bool)> callback)
107 : {
108 2 : return HCCL_SUCCESS;
109 : }
110 :
111 1 : void SetCommEngine(CommEngine engine) { engine_ = engine; }
112 1 : CommEngine GetCommEngine() const { return engine_; }
113 :
114 : protected:
115 : HcclResult ReportAicpuNotifyWaitTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const;
116 : HcclResult ReportHostNotifyWaitTask(u64 notifyId, u64 beginTime, bool isMaster) const;
117 : HcclResult ReportAicpuNotifyRecordTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const;
118 : HcclResult ReportHostNotifyRecordTask(u64 notifyId, u64 beginTime, bool isMaster) const;
119 : HcclResult
120 : ReportAicpuLocalCopyTask(void* dst, const void* src, uint64_t sizeByte, u64 beginTime, u32 taskId, u32 sqId) const;
121 : HcclResult
122 : ReportHostLocalCopyTask(void* dst, const void* src, uint64_t sizeByte, u64 beginTime, bool isMaster) const;
123 : HcclResult ReportAicpuLocalReduceTask(
124 : void* dst, const void* src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp, u64 beginTime,
125 : u32 taskId, u32 sqId) const;
126 : HcclResult ReportHostLocalReduceTask(
127 : void* dst, const void* src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp, u64 beginTime,
128 : bool isMaster) const;
129 : bool IsReportTask() const;
130 :
131 : private:
132 : std::unordered_map<CommEngine, ThreadHandle>
133 : threadHandleMap_; // CPU_TS上的ThreadHandle与其他引擎上的ThreadHandle的映射
134 : std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback_; // 上报task信息的回调函数
135 : CommEngine engine_ = COMM_ENGINE_RESERVED; // 创建时设置的引擎类型
136 : };
137 :
138 9 : inline Stream* GetStream(uint64_t thread)
139 : {
140 9 : Thread* threadPtr = reinterpret_cast<Thread*>(thread);
141 9 : if (UNLIKELY(threadPtr == nullptr)) {
142 0 : HCCL_ERROR("[Thread][GetStream]thread is nullptr");
143 0 : return nullptr;
144 : }
145 9 : return threadPtr->GetStream();
146 : }
147 :
148 2 : inline LocalNotify* GetNotify(uint64_t thread, uint32_t index)
149 : {
150 2 : Thread* threadPtr = reinterpret_cast<Thread*>(thread);
151 2 : if (UNLIKELY(threadPtr == nullptr)) {
152 0 : HCCL_ERROR("[Thread][GetNotify]thread is nullptr");
153 0 : return nullptr;
154 : }
155 2 : return threadPtr->GetNotify(index);
156 : }
157 :
158 : HcclResult CreateThread(
159 : CommEngine engine, StreamType streamType, uint32_t notifyNum, NotifyLoadType loadType,
160 : std::shared_ptr<Thread>& out_thread);
161 : HcclResult CommEngineToNotifyLoadType(CommEngine engine, NotifyLoadType& type);
162 : HcclResult CommHostEngineToNotifyLoadType(CommEngine engine, NotifyLoadType& type);
163 : HcclResult CommEngineToStreamType(CommEngine engine, StreamType& type);
164 : HcclResult GetNotifyLoadType(CommEngine engine, ThreadType threadType, NotifyLoadType& type);
165 : HcclResult GetStreamType(CommEngine engine, ThreadType threadType, StreamType& type);
166 : HcclResult ValidateThreadParams(uint32_t threadNum, uint32_t notifyNumPerThread);
167 : HcclResult SaveThreads(const std::vector<std::shared_ptr<hccl::Thread>>& newThreads);
168 : HcclResult
169 : CreateAndInitThreads(const ThreadCreateParams& params, std::vector<std::shared_ptr<hccl::Thread>>& outThreads);
170 : HcclResult StoreThreadHandles(
171 : std::vector<std::shared_ptr<hccl::Thread>>& newThreads, ThreadHandle* threads, CommEngine engine,
172 : aclrtBinHandle binHandle);
173 : HcclResult FreeThreads(const ThreadHandle* threads, uint32_t threadNum, aclrtBinHandle binHandle);
174 : HcclResult SupplementThreadNotify(ThreadHandle handle, uint32_t notifyNum);
175 : HcclResult FillThreadD2HMap(ThreadHandle* deviceThreadHandles, ThreadHandle* hostThreadHandles, uint32_t listNum);
176 : /**
177 : * @brief 按句柄查全局线程表，获取对应的 shared_ptr<Thread>
178 : * @param[in] handle 线程句柄
179 : * @param[out] outThread 输出的线程共享指针
180 : * @return HcclResult 成功返回 HCCL_SUCCESS，未找到返回 HCCL_E_NOT_FOUND
181 : */
182 : HcclResult LookupThreadByHandle(ThreadHandle handle, std::shared_ptr<hccl::Thread>& outThread);
183 :
184 : /**
185 : * @brief 按 device 句柄查 device→host 映射表，返回对应的 host 句柄
186 : * @param[in] deviceHandle device 侧线程句柄
187 : * @param[out] outHostHandle 输出的 host 侧线程句柄
188 : * @return HcclResult 成功返回 HCCL_SUCCESS，未找到返回 HCCL_E_NOT_FOUND
189 : */
190 : HcclResult LookupD2HHandle(ThreadHandle deviceHandle, ThreadHandle& outHostHandle);
191 : } // namespace hccl
192 : #endif // THREAD_H
|
