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 HCCL_DISPATCHER_AICPU_PUB_H
12 : #define HCCL_DISPATCHER_AICPU_PUB_H
13 :
14 : #include <vector>
15 : #include <functional>
16 : #include <mutex>
17 : #include "sal_pub.h"
18 : #include "dispatcher_pub.h"
19 :
20 : #include "aicpu/aicpu_hccl_sqcq.h"
21 : #include "aicpu/aicpu_hccl_sqcqv1.h"
22 : #include "aicpu/aicpu_hccl_sqcqv2.h"
23 :
24 : #include "op_unfold_cache.h"
25 :
26 : namespace hccl {
27 : using AddOneNotifyWaitSqe = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*, const dfx::DfxTimeOutConfig&);
28 : using AddOneRecordSqe = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*);
29 : using AddOneWriteValueRecordSqe = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*);
30 : using AddOneMemcpySqe = void (*)(
31 : uint16_t, uint16_t, const void*, uint32_t, const aclDataType, aclrtReduceKind, const void*, uint32_t, uint32_t,
32 : uint32_t, u64, uint8_t, const uint8_t*, uint8_t*, uint32_t);
33 : using AddOneEventResetSqe = void (*)(uint16_t, int32_t, uint16_t, int64_t, int64_t, u64, const uint8_t*, uint8_t*);
34 : using AddOneEventRecordSqe = void (*)(uint16_t, int32_t, uint16_t, const uint8_t*, uint8_t*);
35 : using AddOneEventWaitSqe = void (*)(uint16_t, int32_t, uint16_t, const uint8_t*, uint8_t*);
36 : using AddOneRdmaDbSendSqe
37 : = void (*)(uint16_t, uint16_t, uint64_t, uint64_t, uint32_t, uint8_t, const uint8_t*, uint8_t*);
38 : using AddOnePlaceHolderSqe = void (*)(uint16_t, uint16_t, uint16_t, const uint8_t*, uint8_t*);
39 : using AddOneCacheMemcpyPlaceHolderSqe
40 : = void (*)(uint16_t, uint16_t, const void*, const void*, uint8_t, const uint8_t*, uint8_t*, uint32_t);
41 : using AddOneCacheNotifyWaitPlaceholderSqe
42 : = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*, const dfx::DfxTimeOutConfig&);
43 : using AddOneCacheNotifyRecordPlaceholderSqe = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*);
44 : using AddOneCacheWriteValuePlaceholderSqe = void (*)(uint16_t, uint16_t, u64, const uint8_t*, uint8_t*);
45 : using AddOneCacheMemcpyRecordPlaceholderSqe = void (*)(
46 : uint16_t, uint16_t, const void*, uint32_t, const aclDataType, aclrtReduceKind, const void*, uint32_t, uint32_t,
47 : uint32_t, u64, uint8_t, const uint8_t*, uint8_t*, uint32_t);
48 :
49 : class DispatcherAiCpu : public DispatcherPub {
50 : public:
51 : explicit DispatcherAiCpu(const u32 devPhyId);
52 : ~DispatcherAiCpu() override;
53 : HcclResult Init() override;
54 : HcclResult WaitValue(hccl::Stream& stream, u64 waitAddr, u64 valueAddr, bool reset) override;
55 : HcclResult WriteValue(hccl::Stream& stream, u64 writeAddr, u64 valueAddr) override;
56 : HcclResult SignalRecord(
57 : HcclRtNotify signal, hccl::Stream& stream, u32 userRank, u64 offset = INVALID_U64,
58 : s32 stage = INVALID_VALUE_STAGE, bool inchip = false, u64 signalAddr = INVALID_U64,
59 : u32 notifyId = INVALID_UINT) override;
60 : HcclResult SignalRecord(
61 : hccl::DeviceMem& dst, hccl::DeviceMem& src, hccl::Stream& stream, u32 remoteUserRank, hccl::LinkType inLinkType,
62 : u32 notifyId) override;
63 : HcclResult SignalWait(
64 : HcclRtNotify signal, hccl::Stream& stream, u32 userRank, u32 remoteUserRank, s32 stage = INVALID_VALUE_STAGE,
65 : bool inchip = false, u32 notifyId = INVALID_UINT, u32 timeOut = NOTIFY_INVALID_WAIT_TIME) override;
66 : HcclResult MemcpyAsync(
67 : hccl::DeviceMem& dst, const hccl::DeviceMem& src, hccl::Stream& stream,
68 : u32 remoteUserRank = INVALID_VALUE_RANKID, hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP) override;
69 : HcclResult ReduceAsync(
70 : const void* src, void* dst, u64 dataCount, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
71 : HcclReduceType reduceType = HcclReduceType::HCCL_TBE_REDUCE) override;
72 : HcclResult InlineReduceAsync(
73 : const void* src, u64 dataCount, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream, void* dst,
74 : u32 remoteUserRank = INVALID_VALUE_RANKID, hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP) override;
75 : HcclResult RdmaRecord(
76 : u32 dbindex, u64 dbinfo, const struct SendWr& wr, hccl::Stream& stream, RdmaType rdmaType, u32 userRank,
77 : u64 offset, u32 notifyId) override;
78 :
79 : HcclResult LaunchTasksEx(Stream& stream, std::vector<Stream>& subStreams) override;
80 : HcclResult LaunchAllTasks() override;
81 :
82 : HcclResult RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream& stream, RdmaTaskInfo& taskInfo) override;
83 : // 新增接口用于算子展开的动态缓存
84 : HcclResult ClearLaunchContext(); // 当前算子展开不需要使用动态缓存
85 : // 设置launch context, 在LaunchTask时用于算子展开动态缓存的admission
86 : // (因为需要在DispatcherAicpu中暂存AlltoallvMetadata, 所以传入指针而不是引用)
87 : HcclResult SetLaunchContext(
88 : const OpUnfoldKey& key, OpUnfoldCache* cachePtr, const std::vector<OpUnfoldMemRange>& userInputMemRanges,
89 : const std::vector<OpUnfoldMemRange>& userOutputMemRanges, const bool isAlltoallv,
90 : const AlltoallvMetadata* alltoallvMetadataPtr);
91 : // 缓存命中时, 使用缓存中的SQE信息下发给RTSQ
92 : HcclResult LaunchNewTask(
93 : OpUnfoldCacheEntry* entryPtr, const std::vector<OpUnfoldMemRange>& userInputMemRanges,
94 : const std::vector<OpUnfoldMemRange>& userOutputMemRanges, Stream& mainStream, std::vector<Stream>& slaveStreams,
95 : const bool profL1Enable, const bool isAlltoallv, const AlltoallvMetadata& alltoallvMetadata,
96 : const AlltoallvSendRecvInfo& alltoallvSendRecvInfo);
97 :
98 : HcclResult LaunchTask(Stream& stream, bool isBlockLaunch);
99 : HcclResult TbeReduceAsync(
100 : const void* src1, const void* src2, u64 count, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
101 : const void* dst);
102 : HcclResult AddRetryPreamble(Stream& stream) override;
103 : HcclResult StreamSync(Stream& stream) override;
104 :
105 11 : void SetOpExecStatusCallback(std::function<HcclResult()> checkOpExecStatusCallback)
106 : {
107 11 : checkOpExecStatusCallback_ = checkOpExecStatusCallback;
108 11 : return;
109 : }
110 :
111 0 : void SetOpRingBufferIdx(const u32 opRingBufferIdx)
112 : {
113 0 : opRingBufferIdx_ = opRingBufferIdx;
114 0 : HCCL_INFO("[DispatcherAiCpu][SetOpRingBufferIdx]DFX opRingBufferIdx: [%u]", opRingBufferIdx);
115 0 : return;
116 : }
117 :
118 13 : void SetSqeTimeOut(const u64 timeOut)
119 : {
120 13 : if (timeOut > notifyMaxWaitTime_) {
121 1 : dfxTimeOutConfig_.sqeTimeOutTimeOut = notifyMaxWaitTime_;
122 1 : HCCL_WARNING(
123 : "[SetSqeTimeOut] timeOut[%llu s] exceeds the maximum allowed value "
124 : "for notifyMaxWaitTime[%u s].",
125 : timeOut, notifyMaxWaitTime_);
126 : } else {
127 12 : dfxTimeOutConfig_.sqeTimeOutTimeOut = timeOut;
128 : }
129 13 : HCCL_INFO(
130 : "[DispatcherAiCpu][SetSqeTimeOut]DFX timeout config init successfully with details: [%s]",
131 : dfxTimeOutConfig_.ToString().c_str());
132 13 : return;
133 : }
134 :
135 : void GetSqeTimeOut(u64& timeOut)
136 : {
137 : timeOut = dfxTimeOutConfig_.sqeWaitTimeOut;
138 : return;
139 : }
140 :
141 0 : HcclResult SetSqFullWaitTimeOut(u64 notifyWaitTime)
142 : {
143 : dfxTimeOutConfig_.sqFullWaitTimeOut
144 0 : = (notifyWaitTime == 0) ? notifyWaitTime : (notifyWaitTime + AICPU_RTSQ_TIMEOUT_INC);
145 0 : HCCL_INFO(
146 : "[DispatcherAiCpu][SetSqFullWaitTimeOut]DFX timeout config with details: [%s]",
147 : dfxTimeOutConfig_.ToString().c_str());
148 0 : return HCCL_SUCCESS;
149 : }
150 0 : HcclResult SignalRecord(Stream& stream, u64 notifyId)
151 : {
152 0 : return SignalRecord(
153 : nullptr, stream, INVALID_VALUE_RANKID, INVALID_U64, INVALID_VALUE_STAGE, true, INVALID_U64,
154 0 : static_cast<u32>(notifyId));
155 : }
156 0 : HcclResult SignalWait(Stream& stream, u32 notifyId, u32 timeOut)
157 : {
158 0 : return SignalWait(
159 : nullptr, stream, INVALID_VALUE_RANKID, INVALID_VALUE_RANKID, INVALID_VALUE_STAGE, true,
160 0 : static_cast<u32>(notifyId), timeOut);
161 : }
162 :
163 : public:
164 : dfx::DfxTimeOutConfig dfxTimeOutConfig_ = {};
165 : uint32_t opRingBufferIdx_ = 0;
166 :
167 : private:
168 : // 新增接口用于算子展开的动态缓存
169 : HcclResult WaitRtsq(
170 : Stream& stream, const size_t& sqeCount,
171 : const bool isBlockLaunch); // 等待RTSQ直到有sqeCount的SQE的空间 (与LaunchTask中相同的逻辑)
172 : HcclResult MemcpyRtsq(
173 : Stream& stream, const size_t sqeCount, const uint8_t* sqeArray, const uint8_t* sqeTypeArray,
174 : const AicpuDfxInfo* sqeDfxInfoArray, const bool profL1Enable, const std::vector<uint64_t>& profTimestamps,
175 : const size_t profTimestampStartIdx); // 将动态缓存中更新后的SQE的相关信息下发到RTSQ中
176 :
177 : HcclResult AddFlipTask(Stream& stream);
178 : HcclResult GetStreamSqeBufferAddr(
179 : hccl::Stream& stream, uint8_t*& sqeBufferAddr, uint8_t*& sqeTypeAddr, uint8_t*& sqeDfxInfoAddr,
180 : uint16_t& taskId);
181 : void SaveStreamInfo(hccl::Stream& stream);
182 : u64 CalcDbAddr(u32 dbindex);
183 : void InitTimeOutConfig();
184 450 : u32 GetMaxNotifyWaitTime() { return notifyMaxWaitTime_; }
185 :
186 : AddOneNotifyWaitSqe addOneNotifyWaitSqe_ = nullptr;
187 : AddOneRecordSqe addOneRecordSqe_ = nullptr;
188 : AddOneWriteValueRecordSqe addOneWriteValueRecordSqe_ = nullptr;
189 : AddOneMemcpySqe addOneMemcpySqe_ = nullptr;
190 : AddOneEventResetSqe addOneEventResetSqe_ = nullptr;
191 : AddOneEventRecordSqe addOneEventRecordSqe_ = nullptr;
192 : AddOneEventWaitSqe addOneEventWaitSqe_ = nullptr;
193 : AddOneRdmaDbSendSqe addOneRdmaDbSendSqe_ = nullptr;
194 : AddOnePlaceHolderSqe addOneFlipPlaceHolderSqe_ = nullptr;
195 : AddOneCacheMemcpyPlaceHolderSqe addOneCacheMemcpyPlaceHolderSqe_ = nullptr;
196 : AddOneCacheNotifyWaitPlaceholderSqe addOneCacheNotifyWaitPlaceholderSqe_ = nullptr;
197 : AddOneCacheNotifyRecordPlaceholderSqe addOneCacheNotifyRecordPlaceholderSqe_ = nullptr;
198 : AddOneCacheWriteValuePlaceholderSqe addOneCacheWriteValuePlaceholderSqe_ = nullptr;
199 : AddOneCacheMemcpyRecordPlaceholderSqe addOneCacheMemcpyRecordPlaceholderSqe_ = nullptr;
200 : std::function<HcclResult()> checkOpExecStatusCallback_ = nullptr;
201 :
202 : HcclAicpuDispatcherInfo aicpuInfo_;
203 :
204 : std::recursive_mutex streamMapMutex_; // 保护streamMap_并发遍历/插入/删除
205 : std::unordered_map<s32, Stream> streamMap_; // 保存下过task的stream
206 : u64 notifySize_ = 0;
207 :
208 : // Launch context用于算子展开的动态缓存
209 : // 注意: cachePtr_初始化为空, needAddSqe_初始化为false, 即暂无算子展开的动态缓存
210 : OpUnfoldKey key_; // 当前展开算子的标识符
211 : OpUnfoldCache* cachePtr_ = nullptr; // 算子展开的动态缓存
212 : std::vector<OpUnfoldMemRange>
213 : userInputMemRanges_; // 当前算子展开执行时, 通信域内各rank分配的user input memory range
214 : std::vector<OpUnfoldMemRange>
215 : userOutputMemRanges_; // 当前算子展开执行时, 通信域内各rank分配的user output memory range
216 : bool isAlltoallv_ = false;
217 : const AlltoallvMetadata* alltoallvMetadataPtr_ = nullptr; // alltoallv算子对应的metadata (与通信域绑定)
218 : bool needAddSqe_ = false;
219 : };
220 : } // namespace hccl
221 : #endif // HCCL_DISPATCHER_AICPU_PUB_H
|
