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