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