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