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_INC_DISPATCHER_PUB_H
12 : #define HCCL_INC_DISPATCHER_PUB_H
13 :
14 : #include "dispatcher.h"
15 : #include "adapter_hccp.h"
16 : #include "mem_host_pub.h"
17 : #include "mem_device_pub.h"
18 : #include "stream_pub.h"
19 : #include "../platform/common/dlprof_func.h"
20 : #include "externalinput_pub.h"
21 : #include "hccl_common.h"
22 :
23 : #ifdef CCL_LLT
24 : constexpr s64 HCCL_SDMA_MAX_COUNT_4GB = 0xC800000; // llt模块编译时设置SDMA最大数据量为200M
25 : #else
26 : constexpr s64 HCCL_SDMA_MAX_COUNT_4GB = 0x100000000; // SDMA任务最大数据量4GB
27 : #endif
28 :
29 : #if T_DESC("DispatcherPub", true)
30 : namespace hccl {
31 : struct HostNicTaskInfo {
32 : u32 streamId = 0;
33 : u32 taskId = 0;
34 : u64 notifyID = 0;
35 : std::string tag;
36 : };
37 : struct RaSendWrParams {
38 : QpHandle qpHandle;
39 : SendWrlistDataExt wr;
40 : SendWrRsp opRsp;
41 : HostNicTaskInfo taskInfo;
42 : void *dispatcherPtr = nullptr;
43 : HcclWorkflowMode workMode = HcclWorkflowMode::HCCL_WORKFLOW_MODE_RESERVED;
44 : LoadTaskCallBack callback = nullptr;
45 : void *callBackUserPtr = nullptr;
46 :
47 0 : RaSendWrParams(QpHandle &qpHandle, SendWrlistDataExt &wr, void *dispatcherPtr, u32 &streamId, u32 &taskId,
48 : u64 ¬ifyID, HcclWorkflowMode &workMode, LoadTaskCallBack callback, void *callBackUserPtr)
49 0 : : qpHandle(qpHandle), wr(wr), dispatcherPtr(dispatcherPtr), workMode(workMode),
50 0 : callback(callback), callBackUserPtr(callBackUserPtr)
51 : {
52 0 : opRsp = {0};
53 0 : taskInfo.streamId = streamId;
54 0 : taskInfo.taskId = taskId;
55 0 : taskInfo.notifyID = notifyID;
56 0 : }
57 : };
58 : struct RaSocketParams {
59 : FdHandle socketFdHandle;
60 : void *socketBufferPtr;
61 : u64 socketBufferLen;
62 : void *ptr;
63 : u64 len;
64 : HostNicTaskInfo taskInfo;
65 : void *dispatcherPtr = nullptr;
66 : HcclWorkflowMode workMode = HcclWorkflowMode::HCCL_WORKFLOW_MODE_RESERVED;
67 : s32 deviceLogicId; // 当前设备的device id
68 : NICDeployment nicDeploy = NICDeployment::NIC_DEPLOYMENT_RESERVED;
69 : LoadTaskCallBack callback = nullptr;
70 : void *callBackUserPtr = nullptr;
71 :
72 0 : RaSocketParams(FdHandle &socketFdHandle, const void *constSocketBufferPtr, u64 socketBufferLen,
73 : const void *constPtr, u64 len, void *dispatcherPtr, u32 &streamId, u32 &taskId, HcclWorkflowMode &workMode,
74 : s32 deviceLogicId, NICDeployment nicDeploy, LoadTaskCallBack callback, void *callBackUserPtr)
75 0 : : socketFdHandle(socketFdHandle), socketBufferLen(socketBufferLen), len(len), dispatcherPtr(dispatcherPtr),
76 0 : workMode(workMode), deviceLogicId(deviceLogicId), nicDeploy(nicDeploy),
77 0 : callback(callback), callBackUserPtr(callBackUserPtr)
78 : {
79 0 : ptr = const_cast<void *>(constPtr);
80 0 : socketBufferPtr = const_cast<void *>(constSocketBufferPtr);
81 0 : taskInfo.streamId = streamId;
82 0 : taskInfo.taskId = taskId;
83 0 : }
84 0 : RaSocketParams(const RaSocketParams& that) : socketFdHandle(that.socketFdHandle),
85 0 : socketBufferPtr(that.socketBufferPtr), socketBufferLen(that.socketBufferLen), ptr(that.ptr), len(that.len),
86 0 : taskInfo(that.taskInfo), dispatcherPtr(that.dispatcherPtr), workMode(that.workMode),
87 0 : deviceLogicId(that.deviceLogicId), nicDeploy(that.nicDeploy),
88 0 : callback(that.callback), callBackUserPtr(that.callBackUserPtr)
89 : {
90 0 : }
91 : RaSocketParams(const RaSocketParams&& that) : socketFdHandle(that.socketFdHandle),
92 : socketBufferPtr(that.socketBufferPtr), socketBufferLen(that.socketBufferLen), ptr(that.ptr), len(that.len),
93 : taskInfo(that.taskInfo), dispatcherPtr(that.dispatcherPtr), workMode(that.workMode),
94 : deviceLogicId(that.deviceLogicId), nicDeploy(that.nicDeploy),
95 : callback(that.callback), callBackUserPtr(that.callBackUserPtr)
96 : {
97 : }
98 : };
99 :
100 : using WrInformation = struct TagWrInfo {
101 : struct WrInfo wrData{};
102 : u64 type; // 默认 WqeType::WQE_TYPE_DATA
103 : u64 wrDataAddr;
104 : u32 notifyId;
105 1 : TagWrInfo() : type(0), wrDataAddr(0), notifyId(INVALID_UINT) {
106 1 : wrData = {0};
107 1 : }
108 : };
109 :
110 : struct RdmaTaskInfo {
111 : u32 remoteRank = INVALID_UINT;
112 : RdmaType rdmaType = RdmaType::RDMA_TYPE_RESERVED;
113 : std::vector<WrInformation> wrInfos;
114 : };
115 :
116 : class DispatcherPub {
117 : public:
118 : explicit DispatcherPub(const s32 deviceLogicId);
119 : virtual ~DispatcherPub();
120 :
121 : virtual HcclResult Init(); // 初始化必要信息
122 : virtual HcclResult AddRetryPreamble(Stream &stream);
123 : virtual HcclResult StreamSync(Stream &stream);
124 : HcclResult SetNotifyWaitMode(SyncMode notifyWaitMode);
125 : SyncMode GetNotifyWaitMode();
126 :
127 : // 算法下发task时，不要使用HcclRtStream参数类型接口，需要改为hccl::Stream参数类型的接口
128 : HcclResult MemcpySync(void *dst, uint64_t destMax, const void *src, uint64_t count,
129 : HcclRtMemcpyKind kind);
130 : HcclResult MemcpyAsync(void *dst, uint64_t destMax, const void *src, u64 count,
131 : HcclRtMemcpyKind kind, hccl::Stream &stream, u32 remoteUserRank = INVALID_VALUE_RANKID,
132 : hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP);
133 : HcclResult MemcpyAsync(hccl::HostMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream);
134 : HcclResult MemcpyAsync(hccl::HostMem &dst, const hccl::HostMem &src, hccl::Stream &stream);
135 : HcclResult MemcpyAsync(hccl::DeviceMem &dst, const hccl::HostMem &src, hccl::Stream &stream);
136 : HcclResult MemcpyAsyncWithoutCheckKind(void *dst, uint64_t destMax, const void *src, u64 count,
137 : HcclRtMemcpyKind kind, hccl::Stream &stream, u32 remoteUserRank = INVALID_VALUE_RANKID,
138 : hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP);
139 : virtual HcclResult WaitValue(hccl::Stream &stream, u64 waitAddr, u64 valueAddr, bool reset);
140 : virtual HcclResult WriteValue(hccl::Stream &stream, u64 writeAddr, u64 valueAddr);
141 : virtual HcclResult MemcpyAsync(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream,
142 : u32 remoteUserRank = INVALID_VALUE_RANKID, hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP);
143 : virtual HcclResult InlineReduceAsync(const void *src, u64 count, const HcclDataType datatype, HcclReduceOp redOp,
144 : Stream& stream, void *dst, u32 remoteUserRank = INVALID_VALUE_RANKID,
145 : hccl::LinkType inLinkType = hccl::LinkType::LINK_ONCHIP);
146 : virtual HcclResult ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
147 : HcclReduceOp redOp, Stream& stream, HcclReduceType reduceType = HcclReduceType::HCCL_TBE_REDUCE);
148 9 : HcclResult ReduceAsync(const void *src, u64 dataCount, const HcclDataType datatype,
149 : HcclReduceOp redOp, Stream& stream, void *dst, const u32 remoteUserRank, const hccl::LinkType linkType,
150 : const u64 reduceAttr)
151 : {
152 9 : return (INLINE_REDUCE_BITMASK & reduceAttr) ?
153 9 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst, remoteUserRank, linkType) :
154 9 : ReduceAsync(src, dst, dataCount, datatype, redOp, stream);
155 : }
156 :
157 : virtual HcclResult SignalRecord(hccl::DeviceMem &dst, hccl::DeviceMem &src, hccl::Stream &stream,
158 : u32 remoteUserRank, hccl::LinkType inLinkType, u32 notifyId);
159 : virtual HcclResult RdmaRecord(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
160 : RdmaType rdmaType, u32 userRank, u64 offset, u32 notifyId);
161 :
162 : // 下沉模式下的发送接口
163 : HcclResult RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, hccl::Stream &stream,
164 : u32 userRank = INVALID_VALUE_RANKID);
165 : HcclResult RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, hccl::Stream &stream,
166 : u32 userRank, u64 offset);
167 :
168 : // op base 模式下的发送接口
169 : virtual HcclResult RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
170 : u32 remoteUserRank = INVALID_VALUE_RANKID, bool isCapture = false);
171 : virtual HcclResult RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
172 : u32 userRank, u64 offset, bool isCapture = false);
173 :
174 : virtual HcclResult RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream &stream, RdmaTaskInfo &taskInfo);
175 :
176 : // host网卡模式下的rdma send
177 : HcclResult HostNicRdmaSend(QpHandle qpHandle, SendWrlistDataExt &wr, SendWrRsp &opRsp,
178 : hccl::Stream &stream, u32 userRank = INVALID_VALUE_RANKID, u64 offset = 0xFFFFFFFFFFFFFFFF);
179 : // host网卡模式下的tcp send
180 : HcclResult HostNicTcpSend(SocketHandle socketFdHandle, const void *socketBufferPtr, u64 socketBufferLen,
181 : const void *src, u64 len, hccl::Stream &stream, const NICDeployment nicDeploy);
182 : // host网卡模式下的tcp recv
183 : HcclResult HostNicTcpRecv(SocketHandle socketFdHandle, const void *socketBufferPtr, u64 socketBufferLen,
184 : const void *src, u64 len, hccl::Stream &stream, const NICDeployment nicDeploy);
185 :
186 : // host网卡模式下的tcp send处理线程
187 : void HostNicTcpSendThreadTask();
188 : // 下callback task：阻塞入队列，等待send线程将当前队列中send task执行完毕
189 : HcclResult HostNicTcpWaitSendCompletion(hccl::Stream &stream);
190 : // host网卡模式下的tcp send参数入队列
191 : HcclResult SetHostNicTcpSendThreadPara(void *fnData);
192 : void JudgeOpBaseTcpSendComplete(bool &closeSendThreadFlag);
193 : void WaitHostNicTcpSendThreadComplete();
194 : void WaitHostNicTcpSendTaskDone();
195 : void ClearHostNicRdmaParamsVec();
196 : void ClearHostNicTcpSendParamsVec();
197 : void ClearHostNicTcpRecvParamsVec();
198 : HcclResult DelHostNICRdmaTask(u32 streamID, u32 taskID);
199 : HcclResult DelHostNICTcpSendTask(u32 streamID, u32 taskID);
200 : HcclResult DelHostNICTcpRecvTask(u32 streamID, u32 taskID);
201 : HcclResult GetCallbackResult();
202 : HcclResult SetGlobalWorkSpace(std::vector<void *> &globalWorkSpaceAddr);
203 : HcclResult GetNotifyMaxWaitTime();
204 : HcclResult SetHcclExecTimeOut(s32 execTimeOut = NOTIFY_DEFAULT_WAIT_TIME);
205 : s32 GetExecTimeOut();
206 : bool GetExecTimeOutSet();
207 : virtual HcclResult SignalRecord(HcclRtNotify signal, hccl::Stream &stream, u32 userRank, u64 offset = INVALID_U64,
208 : s32 stage = INVALID_VALUE_STAGE, bool inchip = false, u64 signalAddr = INVALID_U64,
209 : u32 notifyId = INVALID_UINT);
210 : virtual HcclResult SignalWait(HcclRtNotify signal, hccl::Stream &stream, u32 userRank, u32 remoteUserRank,
211 : s32 stage = INVALID_VALUE_STAGE, bool inchip = false, u32 notifyId = INVALID_UINT,
212 : u32 timeOut = NOTIFY_INVALID_WAIT_TIME);
213 :
214 0 : virtual HcclResult SignalRecord(Stream &stream, u64 notifyId)
215 : {
216 0 : return SignalRecord(reinterpret_cast<HcclRtNotify>(notifyId), stream, INVALID_VALUE_RANKID, INVALID_U64,
217 0 : INVALID_VALUE_STAGE, true, INVALID_U64, INVALID_UINT);
218 : }
219 0 : virtual HcclResult SignalWait(Stream &stream, u32 notifyId, u32 timeOut)
220 : {
221 0 : return SignalWait(reinterpret_cast<HcclRtNotify>(notifyId), stream, INVALID_VALUE_RANKID, INVALID_VALUE_RANKID,
222 0 : INVALID_VALUE_STAGE, true, INVALID_UINT, timeOut);
223 : }
224 78 : virtual HcclResult LaunchTasksEx(Stream &stream, std::vector<Stream> &subStreams)
225 : {
226 78 : return HCCL_SUCCESS;
227 : }
228 0 : virtual HcclResult LaunchAllTasks()
229 : {
230 0 : return HCCL_SUCCESS;
231 : }
232 57 : virtual HcclResult ResetGraphCtx(bool enableCache, const std::string &key, bool useGraphConstructorV2)
233 : {
234 57 : return HCCL_SUCCESS;
235 : }
236 0 : virtual void SetNormalMode()
237 : {
238 0 : return;
239 : }
240 :
241 1053 : virtual void RegLoadTaskCallBack(void *userPtr, LoadTaskCallBack callback)
242 : {
243 1053 : callback_ = callback;
244 1053 : callBackUserPtr_ = userPtr;
245 1053 : }
246 :
247 100 : uint64_t GetMsprofSysCycleTime(void) {
248 100 : if (!GetIfProfile()) {
249 0 : return 0;
250 : }
251 100 : u64 ret = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
252 100 : return ret;
253 : }
254 :
255 76 : static void ForceProf(bool isForce) {
256 76 : isForce_ = isForce;
257 76 : }
258 :
259 : static bool IsProfSubscribeAdditionInfo();
260 :
261 0 : virtual HcclResult SetMultiQpMode(bool multiQpMode)
262 : {
263 0 : return HCCL_SUCCESS;
264 : }
265 :
266 : void SetHcclQos(u32 hcclQos);
267 : void SetMpamid(u32 mPamid);
268 :
269 11 : uint32_t GetHcclQos()
270 : {
271 11 : return hcclQos_;
272 : }
273 :
274 : inline bool IsPlaceholder() const
275 : {
276 : return isPlaceholder_;
277 : }
278 :
279 0 : inline void SetPlaceholder(const bool isPlaceholder)
280 : {
281 0 : isPlaceholder_ = isPlaceholder;
282 0 : return;
283 : }
284 :
285 : protected:
286 : HcclResult RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, HcclRtStream stream, hccl::RdmaType rdmaType,
287 : u64 notifyID = INVALID_U64, bool isMainStream = false);
288 : HcclResult RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, HcclRtStream stream, hccl::RdmaType rdmaType,
289 : u64 notifyID = INVALID_U64, u64 offset = 0, bool isMainStream = false);
290 : HcclResult SignalRecord(HcclRtNotify signal, HcclRtStream stream, u32 userRank, u64 offset = INVALID_U64,
291 : s32 stage = INVALID_VALUE_STAGE, bool isMainStream = false);
292 : HcclResult SignalWait(HcclRtNotify signal, HcclRtStream stream, u32 userRank, u32 remoteUserRank,
293 : s32 stage = INVALID_VALUE_STAGE, u32 timeOut = NOTIFY_INVALID_WAIT_TIME, bool isMainStream = false);
294 : HcclResult TbeReduceAsync(const void *src1, const void *src2, u64 count, const HcclDataType datatype,
295 : HcclReduceOp redOp, Stream &stream, const void *dst);
296 : u32 GetNotifyWaitTime(u32 timeOut);
297 : HcclResult DevMemMalloc(void *stream, void *&devMem1, void *&devMem2);
298 : HcclResult JudgeIsTail(const void *src1, const void *src2, const void *dst, u64 count, const HcclDataType dataType,
299 : u64 &headCount, u64 &tailCount, void *&tailSrc1, void *&tailSrc2, void *&tailDst);
300 :
301 : s32 deviceLogicId_; // 当前设备的device id
302 : std::mutex mutex_;
303 :
304 : SyncMode notifyWaitMode_;
305 : std::map<u32, std::queue<std::unique_ptr<RaSendWrParams>>> hostNicRdmaParamsVec_;
306 : std::map<u32, std::queue<std::unique_ptr<RaSocketParams>>> hostNicTcpSendParamsVec_; // host网卡tcp模式下存放发task
307 : std::map<u32, std::queue<std::unique_ptr<RaSocketParams>>> hostNicTcpRecvParamsVec_; // host网卡tcp模式下存放收task
308 : std::unique_ptr<RaSocketParams> hostNicTcpSendThreadParam_;
309 : std::unique_ptr<std::thread> hostNicTcpSendThread_;
310 : bool hostNicTcpSendThreadState_;
311 : std::mutex hostNicMutex_;
312 : void* overflowAddr_;
313 : void *fftsPubInfo_{nullptr};
314 : bool setDeviceFlag_;
315 : uint32_t notifyMaxWaitTime_;
316 : LoadTaskCallBack callback_{nullptr};
317 : void *callBackUserPtr_{nullptr};
318 : std::map<int32_t, void *> devMemMap_; // streamId和device内存的map
319 : std::mutex devMemMutex_;
320 : static bool isForce_; // 强制profiling上报或缓存
321 : s32 execTimeOut_;
322 : bool execTimeOutByConfig_;
323 : uint32_t hcclQos_;
324 : uint32_t mPamid_;
325 : bool isPlaceholder_ = false; // 用于区分是否生成placeholder SQE还是正常SQE
326 :
327 : private:
328 : void SetupTaskParaDma(hccl::TaskPara& taskPara, hccl::TaskParaDMA& para, TaskType taskType,
329 : ProfilerType profilerType, hccl::Stream &stream, u64 beginTime, bool isMainStream) const;
330 : void SetupTaskParaDma(hccl::TaskPara& taskPara, hccl::TaskParaDMA& para, TaskType taskType,
331 : HcclRtStream stream, u64 beginTime, bool isMainStream) const;
332 : HcclResult DealTbeReduce(const void *src1, const void *src2, u64 count,
333 : const HcclDataType datatype, HcclReduceOp redOp, Stream& stream, const void *dst);
334 : };
335 : } // namespace hccl
336 : #endif
337 : #endif // HCCL_INC_DISPATCHER_PUB_H
|
