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.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 10 : ThreadCreateParams(CommEngine engine,
51 : uint32_t tNum,
52 : uint32_t nNum,
53 : NotifyLoadType nType,
54 : StreamType sType)
55 10 : : engine(engine)
56 10 : , threadNum(tNum)
57 10 : , notifyNumPerThread(nNum)
58 10 : , notifyLoadType(nType)
59 10 : , streamType(sType) {
60 10 : }
61 : };
62 :
63 : constexpr u32 HCOMM_NOTIFY_MAX_NUM = 64;
64 : constexpr u32 HCOMM_THREADNUM_MAX_NUM = 1000;
65 : constexpr u32 HCCL_THREAD_NOTIFY_MAX_NUM = 65536;
66 : /**
67 : * @note 职责：通信引擎的Thread的C++抽象接口类，表达并行资源，内部包含thread间的同步Notify。
68 : */
69 : class Thread {
70 : public:
71 297 : virtual ~Thread() = default;
72 : virtual HcclResult Init() = 0;
73 : virtual HcclResult DeInit() = 0;
74 : virtual std::string &GetUniqueId() = 0;
75 : virtual uint32_t GetNotifyNum() const = 0;
76 : virtual LocalNotify *GetNotify(uint32_t index) const = 0;
77 : virtual HcclResult SupplementNotify(uint32_t notifyNum) = 0;
78 :
79 : // A3 Stream & A5 Stream
80 : virtual bool IsDeviceA5() const = 0;
81 : virtual Stream *GetStream() const = 0;
82 : virtual void *GetStreamLitePtr() const = 0;
83 : virtual void LaunchTask() const = 0;
84 : virtual void TryLaunchTask() const = 0;
85 :
86 : // Local Data Plane Functions
87 : virtual HcclResult LocalNotifyRecord(uint32_t notifyId) const = 0;
88 : virtual HcclResult LocalNotifyWait(uint32_t notifyId) const = 0;
89 :
90 : virtual HcclResult LocalNotifyRecord(ThreadHandle dstThread, uint32_t dstNotifyIdx) const = 0;
91 : virtual HcclResult LocalNotifyWait(uint32_t notifyIdx, uint32_t timeOut) const = 0;
92 :
93 : virtual HcclResult LocalCopy(void *dst, const void *src, uint64_t sizeByte) const = 0;
94 : virtual HcclResult LocalReduce(
95 : void *dst, const void *src, uint64_t sizeByte, HcommDataType dataType, HcommReduceOp reduceOp) const = 0;
96 : virtual bool GetMaster() const = 0;
97 : virtual void SetIsMaster(bool isMaster) = 0;
98 :
99 : HcclResult AddThreadHandleToMap(CommEngine commEngine, ThreadHandle threadHandle);
100 : Thread *FindThreadByCommEngine(CommEngine commEngine);
101 112 : HcclResult SetAddTaskInfoCallback(std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback) {
102 112 : CHK_PTR_NULL(callback);
103 112 : callback_ = callback;
104 112 : return HCCL_SUCCESS;
105 : }
106 5 : std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> GetCallback() { return callback_; }
107 0 : virtual HcclResult SetCheckExecStatusCallback(std::function<HcclResult(bool)> callback) { return HCCL_SUCCESS; }
108 : protected:
109 : HcclResult ReportAicpuNotifyWaitTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const;
110 : HcclResult ReportHostNotifyWaitTask(u64 notifyId, u64 beginTime, bool isMaster) const;
111 : HcclResult ReportAicpuNotifyRecordTask(u64 notifyId, u64 beginTime, u32 taskId, u32 sqId) const;
112 : HcclResult ReportHostNotifyRecordTask(u64 notifyId, u64 beginTime, bool isMaster) const;
113 : HcclResult ReportAicpuLocalCopyTask(void *dst, const void *src, uint64_t sizeByte, u64 beginTime, u32 taskId,u32 sqId) const;
114 : HcclResult ReportHostLocalCopyTask(void *dst, const void *src, uint64_t sizeByte, u64 beginTime, bool isMaster) const;
115 : HcclResult ReportAicpuLocalReduceTask(void *dst, const void *src, uint64_t sizeByte, HcommDataType dataType,
116 : HcommReduceOp reduceOp, u64 beginTime, u32 taskId, u32 sqId) const;
117 : HcclResult ReportHostLocalReduceTask(void *dst, const void *src, uint64_t sizeByte, HcommDataType dataType,
118 : HcommReduceOp reduceOp, u64 beginTime, bool isMaster) const;
119 : bool IsReportTask() const;
120 : private:
121 : std::unordered_map<CommEngine, ThreadHandle> threadHandleMap_; // CPU_TS上的ThreadHandle与其他引擎上的ThreadHandle的映射
122 : std::function<HcclResult(u32, u32, const Hccl::TaskParam&, u64)> callback_; // 上报task信息的回调函数
123 : };
124 :
125 9 : inline Stream *GetStream(uint64_t thread)
126 : {
127 9 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
128 9 : if (UNLIKELY(threadPtr == nullptr)) {
129 0 : HCCL_ERROR("[Thread][GetStream]thread is nullptr");
130 0 : return nullptr;
131 : }
132 9 : return threadPtr->GetStream();
133 : }
134 :
135 2 : inline LocalNotify *GetNotify(uint64_t thread, uint32_t index)
136 : {
137 2 : Thread *threadPtr = reinterpret_cast<Thread *>(thread);
138 2 : if (UNLIKELY(threadPtr == nullptr)) {
139 0 : HCCL_ERROR("[Thread][GetStream]thread is nullptr");
140 0 : return nullptr;
141 : }
142 2 : return threadPtr->GetNotify(index);
143 : }
144 :
145 : HcclResult CreateThread(CommEngine engine, StreamType streamType, uint32_t notifyNum,
146 : NotifyLoadType loadType, std::shared_ptr<Thread>& out_thread);
147 : HcclResult CommEngineToNotifyLoadType(CommEngine engine, NotifyLoadType &type);
148 : HcclResult CommHostEngineToNotifyLoadType(CommEngine engine, NotifyLoadType &type);
149 : HcclResult CommEngineToStreamType(CommEngine engine, StreamType &type);
150 : HcclResult GetNotifyLoadType(CommEngine engine, ThreadType threadType, NotifyLoadType &type);
151 : HcclResult GetStreamType(CommEngine engine, ThreadType threadType, StreamType &type);
152 : HcclResult ValidateThreadParams(uint32_t threadNum, uint32_t notifyNumPerThread);
153 : HcclResult SaveThreads(const std::vector<std::shared_ptr<hccl::Thread>> &newThreads);
154 : HcclResult CreateAndInitThreads(const ThreadCreateParams& params,
155 : std::vector<std::shared_ptr<hccl::Thread>>& outThreads);
156 : HcclResult StoreThreadHandles(std::vector<std::shared_ptr<hccl::Thread>>& newThreads,
157 : ThreadHandle* threads, CommEngine engine, aclrtBinHandle binHandle);
158 : HcclResult FreeThreads(const ThreadHandle *threads, uint32_t threadNum, aclrtBinHandle binHandle);
159 : HcclResult SupplementThreadNotify(ThreadHandle handle, uint32_t notifyNum);
160 : } // namespace hccl
161 : #endif // THREAD_H
|
