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 : #include "externalinput.h"
12 : #include "adapter_rts.h"
13 : #include "log.h"
14 : #include "dtype_common.h"
15 : #include "dispatcher_graph.h"
16 : #include "hccl_tbe_task.h"
17 : #include "graph_ctx_mgr_common.h"
18 : #include "config_plf_log.h"
19 :
20 : constexpr u32 UB_BLOCK_SIZE = 32;
21 : constexpr u64 TBE_REDUCE_MAX_COUNT = INT32_MAX;
22 :
23 : __attribute__((weak)) HcclResult GraphAddRecordTaskWithSignalAddr(void *fftsPubInfo, void *ctx, uint32_t streamId,
24 : void *signal, bool inchip, u64 signalAddr, uint32_t *ctxIdx);
25 : __attribute__((weak)) HcclResult LaunchGraphAndGetGraphInfo(void *fftsPubInfo, void *streamPtr, void *ctx,
26 : uint32_t timeout, uint32_t *ctxNum, void **descBuf, size_t *descBufLen);
27 :
28 : namespace hccl {
29 119 : DispatcherGraph::DispatcherGraph(const s32 deviceLogicId)
30 119 : : DispatcherPub(deviceLogicId), fftsCtxsPtr(nullptr), disableFfts_(true), multiQpMode_(false)
31 119 : {}
32 :
33 234 : DispatcherGraph::~DispatcherGraph()
34 234 : {}
35 :
36 0 : void DispatcherGraph::SetNormalMode()
37 : {
38 0 : disableFfts_ = true;
39 0 : }
40 :
41 0 : HcclResult DispatcherGraph::SetMultiQpMode(bool multiQpMode)
42 : {
43 0 : multiQpMode_ = multiQpMode;
44 0 : HCCL_DEBUG("[MultiQp][DispatcherGraph::SetMultiQpMode] [%d]", multiQpMode);
45 0 : return HcclResult::HCCL_SUCCESS;
46 : }
47 :
48 0 : HcclResult DispatcherGraph::ResetGraphCtx(bool enableCache, const std::string &key, bool useGraphConstructorV2)
49 : {
50 0 : disableFfts_ = false;
51 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
52 0 : HCCL_DEBUG("ffts task is disabled.");
53 0 : disableFfts_ = true;
54 : } else {
55 0 : if (multiQpMode_) {
56 0 : enableCache = false;
57 : }
58 :
59 0 : std::string sKey = key;
60 0 : if (UNLIKELY(!enableCache)) {
61 : // 当enableCache使能时,key传空值，用来区分当enableCache
62 0 : sKey = "";
63 : }
64 0 : HCCL_INFO("useGraphConstructorV2[%d] sKey[%s] length[%u] key[%s]",
65 : useGraphConstructorV2, sKey.c_str(), sKey.length(), key.c_str());
66 0 : if (useGraphConstructorV2) {
67 0 : fftsCtxsPtr = GetGraphCtxV2(fftsPubInfo_, sKey.c_str(), sKey.length());
68 : } else {
69 0 : fftsCtxsPtr = GetGraphCtx(fftsPubInfo_, sKey.c_str(), sKey.length());
70 : }
71 0 : CHK_PTR_NULL(fftsCtxsPtr);
72 0 : disableFfts_ = false;
73 0 : }
74 0 : return HCCL_SUCCESS;
75 : }
76 :
77 19 : HcclResult DispatcherGraph::LaunchTasksEx(Stream &stream, std::vector<Stream> &subStreams)
78 : {
79 19 : if (UNLIKELY(disableFfts_)) {
80 18 : return HCCL_SUCCESS;
81 : }
82 :
83 1 : uint64_t beginTime = GetMsprofSysCycleTime();
84 1 : CHK_PTR_NULL(fftsCtxsPtr); // 检查Context是否进行过Reset
85 :
86 1 : u32 timeout = 0;
87 : // 配置notify wait 超时时间
88 : // 因为老版本用户设置HCCL_EXEC_TIMEOUT为0，hccl将0传递给rts,rts将0转换为1770s传递给硬件, 未达到永不超时效果，不符合预期；
89 : // 所以现版用户配置为0时，hccl转换成65535，rts识别到65535后会又会转换成0，去硬件设置永不超时
90 1 : if (execTimeOut_ == 0) {
91 0 : timeout = FFTS_TIMEOUT_MAX;
92 : // 因为65535被当作永不超时处理，所以当用户配置65535时需要改变他的值，防止误错做成永不超时
93 1 : } else if (execTimeOut_ == FFTS_TIMEOUT_MAX) {
94 0 : timeout = FFTS_TIMEOUT_MAX - 1;
95 : } else {
96 1 : timeout = execTimeOut_ ;
97 : }
98 : u32 ctxNum;
99 1 : void *descBuf = nullptr;
100 1 : size_t descBufLen = 0;
101 1 : if (LaunchGraphAndGetGraphInfo != nullptr) {
102 1 : CHK_RET(LaunchGraphAndGetGraphInfo(fftsPubInfo_, stream.ptr(), fftsCtxsPtr, timeout, &ctxNum, &descBuf, &descBufLen));
103 : } else {
104 0 : CHK_RET(LaunchGraph(fftsPubInfo_, stream.ptr(), fftsCtxsPtr, timeout, &ctxNum));
105 : }
106 1 : disableFfts_ = true;
107 : // 调用回调来保存task信息
108 1 : if (callback_ != nullptr) {
109 1 : struct TaskPara taskPara;
110 1 : taskPara.type = TaskType::TASK_GRAPH_LAUNCH;
111 1 : taskPara.stream = stream.ptr();
112 1 : taskPara.isMainStream = stream.IsMainStream();
113 1 : taskPara.beginTime = beginTime;
114 1 : taskPara.graphLaunch.ctxNum = ctxNum;
115 1 : taskPara.graphLaunch.descBuf = descBuf;
116 1 : taskPara.graphLaunch.descBufLen = descBufLen;
117 1 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
118 1 : }
119 :
120 1 : return HCCL_SUCCESS;
121 : }
122 :
123 0 : HcclResult DispatcherGraph::GetNotifyDfxInfo(HcclRtNotify signal, u32 userRank, u64 &offset, u32 &remoteUserRank,
124 : u64 ¬ifyID)
125 : {
126 0 : if (offset == INVALID_U64) {
127 0 : CHK_RET(hrtNotifyGetOffset(static_cast<HcclRtNotify>(signal), offset));
128 : }
129 0 : notifyID = userRank;
130 0 : notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
131 0 : remoteUserRank = (remoteUserRank == INVALID_UINT) ? static_cast<u32>(notifyID >> 32) : remoteUserRank;
132 0 : return HCCL_SUCCESS;
133 : }
134 :
135 : HcclResult DispatcherGraph::SignalTaskParaSave(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank,
136 : u64 offset, s32 stage, TaskType taskType, uint64_t beginTime, u32 ctxIdx)
137 : {
138 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
139 : u64 notifyID;
140 0 : CHK_RET(GetNotifyDfxInfo(signal, userRank, offset, remoteUserRank, notifyID));
141 : // 调用回调来保存task信息
142 0 : hccl::TaskParaNotify para(notifyID, stage, remoteUserRank, (ctxIdx - 1));
143 0 : struct TaskPara taskPara;
144 0 : taskPara.stream = stream.ptr();
145 0 : taskPara.isMainStream = stream.IsMainStream();
146 0 : taskPara.beginTime = beginTime;
147 0 : taskPara.notify = para;
148 0 : taskPara.type = taskType;
149 0 : taskPara.isFftsDispatcher = true;
150 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
151 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
152 0 : }
153 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
154 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
155 : u64 notifyID;
156 0 : CHK_RET(GetNotifyDfxInfo(signal, userRank, offset, remoteUserRank, notifyID));
157 0 : hccl::TaskParaNotify para(notifyID, stage, remoteUserRank, (ctxIdx - 1));
158 0 : struct TaskPara taskPara;
159 0 : taskPara.stream = stream.ptr();
160 0 : taskPara.isMainStream = stream.IsMainStream();
161 0 : taskPara.beginTime = beginTime;
162 0 : taskPara.notify = para;
163 0 : taskPara.type = taskType;
164 0 : taskPara.isFftsDispatcher = true;
165 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
166 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
167 0 : }
168 :
169 0 : return HCCL_SUCCESS;
170 : }
171 :
172 0 : HcclResult DispatcherGraph::SignalRecord(HcclRtNotify signal, Stream &stream, u32 userRank, u64 offset, s32 stage,
173 : bool inchip, u64 signalAddr, u32 notifyId)
174 : {
175 0 : uint64_t beginTime = GetMsprofSysCycleTime();
176 : (void)notifyId;
177 0 : if (UNLIKELY(disableFfts_)) {
178 0 : return DispatcherPub::SignalRecord(signal, stream, userRank, offset, stage, inchip);
179 : }
180 : u32 ctxIdx;
181 0 : if (GraphAddRecordTaskWithSignalAddr != nullptr) {
182 0 : CHK_RET(GraphAddRecordTaskWithSignalAddr(fftsPubInfo_, fftsCtxsPtr, stream.id(), signal, inchip, signalAddr, &ctxIdx));
183 : } else {
184 0 : CHK_RET(GraphAddRecordTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), signal, inchip, &ctxIdx));
185 : }
186 0 : if (!inchip && ctxIdx > 0) {
187 0 : CHK_RET(SignalTaskParaSave(signal, stream, userRank, INVALID_UINT,
188 : offset, stage, TaskType::TASK_NOTIFY_RECORD, beginTime, ctxIdx));
189 : }
190 :
191 0 : if (HcclCheckLogLevel(HCCL_LOG_INFO) || (GetExternalInputDebugConfig() & PLF_TASK)) {
192 0 : u64 notifyID = userRank;
193 0 : u32 remoteUserRank = INVALID_UINT;
194 0 : CHK_RET(GetNotifyDfxInfo(signal, userRank, offset, remoteUserRank, notifyID));
195 0 : PLF_CONFIG_INFO(PLF_TASK,
196 : "%s para: notifyId[0x%016llx] streamId[%u] userRank[%u] remoteUserRank[%u] offset[%llu] stage[%d] inchip[%d]",
197 : __func__, notifyID, stream.id(), userRank, remoteUserRank, offset, stage, inchip);
198 : }
199 0 : return HCCL_SUCCESS;
200 : }
201 :
202 0 : HcclResult DispatcherGraph::SignalWait(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank, s32 stage,
203 : bool inchip, u32 notifyId, u32 timeOut)
204 : {
205 0 : uint64_t beginTime = GetMsprofSysCycleTime();
206 0 : if (UNLIKELY(disableFfts_)) {
207 0 : return DispatcherPub::SignalWait(signal, stream, userRank, remoteUserRank, stage, inchip, notifyId, timeOut);
208 : }
209 : u32 ctxIdx;
210 0 : CHK_RET(GraphAddWaitTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), signal, inchip, &ctxIdx));
211 0 : if (!inchip && ctxIdx > 0) {
212 0 : CHK_RET(SignalTaskParaSave(signal, stream, userRank, remoteUserRank,
213 : INVALID_U64, stage, TaskType::TASK_NOTIFY_WAIT, beginTime, ctxIdx));
214 : }
215 :
216 0 : if (HcclCheckLogLevel(HCCL_LOG_INFO) || (GetExternalInputDebugConfig() & PLF_TASK)) {
217 0 : u64 notifyID = userRank;
218 0 : u64 offset = INVALID_U64;
219 0 : CHK_RET(GetNotifyDfxInfo(signal, userRank, offset, remoteUserRank, notifyID));
220 0 : PLF_CONFIG_INFO(PLF_TASK,
221 : "%s para: notifyId[0x%016llx] streamId[%u] userRank[%u] remoteUserRank[%u] offset[%llu] stage[%d] inchip[%d]",
222 : __func__, notifyID, stream.id(), userRank, remoteUserRank, offset, stage, inchip);
223 : }
224 0 : return HCCL_SUCCESS;
225 : }
226 :
227 5 : HcclResult DispatcherGraph::MemcpyAsync(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream,
228 : u32 remoteUserRank, hccl::LinkType inLinkType)
229 : {
230 5 : uint64_t beginTime = GetMsprofSysCycleTime();
231 5 : if (dst.size() < src.size()) {
232 0 : HCCL_ERROR(
233 : "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
234 : dst.ptr(), dst.size(), src.ptr(), src.size());
235 0 : return HCCL_E_PTR;
236 : }
237 :
238 5 : if (UNLIKELY(disableFfts_)) {
239 5 : return DispatcherPub::MemcpyAsync(dst, src, stream, remoteUserRank, inLinkType);
240 : }
241 0 : PLF_CONFIG_INFO(PLF_TASK,
242 : "%s para: dst[%p] destMax[%llu] src[%p] count[%llu] rtMemcpyKind[%d] inLinkType[%d] remoteUserRank[%u] streamId[%u]",
243 : __func__, dst.ptr(), dst.size(), src.ptr(), src.size(), HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE,
244 : inLinkType, remoteUserRank, stream.id());
245 : u32 ctxIdx;
246 0 : CHK_RET(GraphAddMemcpyTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), dst.ptr(), src.ptr(), src.size(), &ctxIdx));
247 : // 调用回调来保存task信息
248 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
249 0 : hccl::TaskParaDMA para(src.ptr(), dst.ptr(), src.size(), inLinkType, remoteUserRank,
250 0 : hccl::RdmaType::RDMA_TYPE_RESERVED, (ctxIdx - 1));
251 0 : struct TaskPara taskPara;
252 0 : taskPara.stream = stream.ptr();
253 0 : taskPara.isMainStream = stream.IsMainStream();
254 0 : taskPara.beginTime = beginTime;
255 0 : taskPara.dma = para;
256 0 : taskPara.type = TaskType::TASK_SDMA;
257 0 : taskPara.isFftsDispatcher = true;
258 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
259 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
260 0 : }
261 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
262 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
263 0 : hccl::TaskParaDMA para(src.ptr(), dst.ptr(), src.size(), inLinkType, remoteUserRank,
264 0 : hccl::RdmaType::RDMA_TYPE_RESERVED, (ctxIdx - 1));
265 0 : struct TaskPara taskPara;
266 0 : taskPara.stream = stream.ptr();
267 0 : taskPara.isMainStream = stream.IsMainStream();
268 0 : taskPara.beginTime = beginTime;
269 0 : taskPara.dma = para;
270 0 : taskPara.type = TaskType::TASK_SDMA;
271 0 : taskPara.isFftsDispatcher = true;
272 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
273 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
274 0 : }
275 0 : return HCCL_SUCCESS;
276 : }
277 :
278 0 : HcclResult DispatcherGraph::ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
279 : HcclReduceOp redOp, Stream &stream, HcclReduceType reduceType)
280 : {
281 0 : uint64_t beginTime = GetMsprofSysCycleTime();
282 0 : if (UNLIKELY(disableFfts_)) {
283 0 : return DispatcherPub::ReduceAsync(src, dst, dataCount, datatype, redOp, stream, reduceType);
284 : }
285 0 : PLF_CONFIG_INFO(PLF_TASK,
286 : "%s para: src[%p] dst[%p] dataCount[%llu] datatype[%s] redOp[%s] reduceType[%d] streamID[%u]",
287 : __func__, src, dst, dataCount, GetDataTypeEnumStr(datatype).c_str(),
288 : GetReduceOpEnumStr(redOp).c_str(), reduceType, stream.id());
289 :
290 0 : if (reduceType == HcclReduceType::HCCL_TBE_REDUCE) {
291 : // dtype=int64 或者 redOp=prod 都会走TbeReduce, 算法层控制的
292 0 : return TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
293 : }
294 :
295 0 : u32 ctxIdx = 0;
296 0 : CHK_RET(GraphAddReduceTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), dst, src, dataCount, datatype,
297 : redOp, &ctxIdx));
298 : // 调用回调来保存 task 信息
299 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
300 0 : hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype,
301 0 : LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (ctxIdx - 1));
302 0 : struct TaskPara taskPara;
303 0 : taskPara.stream = stream.ptr();
304 0 : taskPara.isMainStream = stream.IsMainStream();
305 0 : taskPara.beginTime = beginTime;
306 0 : taskPara.reduce = para;
307 0 : taskPara.type = TaskType::TASK_REDUCE_INLINE;
308 0 : taskPara.isFftsDispatcher = true;
309 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
310 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
311 0 : }
312 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
313 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
314 0 : hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype,
315 0 : LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (ctxIdx - 1));
316 0 : struct TaskPara taskPara;
317 0 : taskPara.stream = stream.ptr();
318 0 : taskPara.isMainStream = stream.IsMainStream();
319 0 : taskPara.beginTime = beginTime;
320 0 : taskPara.reduce = para;
321 0 : taskPara.type = TaskType::TASK_REDUCE_INLINE;
322 0 : taskPara.isFftsDispatcher = true;
323 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
324 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
325 0 : }
326 :
327 0 : return HCCL_SUCCESS;
328 : }
329 :
330 0 : HcclResult DispatcherGraph::InlineReduceAsync(const void *src, u64 dataCount, const HcclDataType datatype,
331 : HcclReduceOp redOp, Stream &stream, void *dst, u32 remoteUserRank, hccl::LinkType inLinkType)
332 : {
333 0 : uint64_t beginTime = GetMsprofSysCycleTime();
334 0 : if (UNLIKELY(disableFfts_)) {
335 0 : return DispatcherPub::InlineReduceAsync(
336 0 : src, dataCount, datatype, redOp, stream, dst, remoteUserRank, inLinkType);
337 : }
338 0 : PLF_CONFIG_INFO(PLF_TASK,
339 : "%s para: src[%p] dst[%p] dataCount[%llu] datatype[%s] redOp[%s] inLinkType[%d] remoteUserRank[%u] streamID[%u]",
340 : __func__, src, dst, dataCount, GetDataTypeEnumStr(datatype).c_str(), GetReduceOpEnumStr(redOp).c_str(),
341 : inLinkType, remoteUserRank, stream.id());
342 0 : u32 ctxIdx = 0;
343 0 : CHK_RET(GraphAddInlineReduceTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), dst, src, dataCount, datatype,
344 : redOp, &ctxIdx));
345 :
346 : // 调用回调来保存 task 信息
347 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
348 0 : hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype, inLinkType,
349 0 : remoteUserRank, (ctxIdx - 1));
350 0 : struct TaskPara taskPara;
351 0 : taskPara.stream = stream.ptr();
352 0 : taskPara.isMainStream = stream.IsMainStream();
353 0 : taskPara.beginTime = beginTime;
354 0 : taskPara.reduce = para;
355 0 : taskPara.type = TaskType::TASK_REDUCE_INLINE;
356 0 : taskPara.isFftsDispatcher = true;
357 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
358 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
359 0 : }
360 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
361 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
362 0 : hccl::TaskParaReduce para(src, dst, dataCount * SIZE_TABLE[datatype], redOp, datatype, inLinkType,
363 0 : remoteUserRank, (ctxIdx - 1));
364 0 : struct TaskPara taskPara;
365 0 : taskPara.stream = stream.ptr();
366 0 : taskPara.isMainStream = stream.IsMainStream();
367 0 : taskPara.beginTime = beginTime;
368 0 : taskPara.reduce = para;
369 0 : taskPara.type = TaskType::TASK_REDUCE_INLINE;
370 0 : taskPara.isFftsDispatcher = true;
371 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
372 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
373 0 : }
374 0 : return HCCL_SUCCESS;
375 : }
376 :
377 1 : HcclResult DispatcherGraph::RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
378 : u32 remoteUserRank, bool isCapture)
379 : {
380 1 : uint64_t beginTime = GetMsprofSysCycleTime();
381 1 : if (UNLIKELY(disableFfts_)) {
382 0 : return DispatcherPub::RdmaSend(dbindex, dbinfo, wr, stream, remoteUserRank);
383 : }
384 :
385 1 : u64 notifyID = (static_cast<u64>(remoteUserRank) << 32) | (0x00000000FFFFFFFF);
386 1 : PLF_CONFIG_INFO(PLF_TASK,
387 : "%s para: dbindex[%u], dbinfo[%llu], notifyId[0x%016llx], remoteUserRank[%u], isCapture[%d], streamID[%u]",
388 : __func__, dbindex, dbinfo, notifyID, remoteUserRank, stream.id(), isCapture);
389 :
390 1 : u32 ctxIdx = 0;
391 1 : CHK_RET(GraphAddRdmaSendTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), dbindex, dbinfo, isCapture, &ctxIdx));
392 : // 调用回调来保存task信息
393 1 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
394 : // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
395 1 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
396 1 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
397 1 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_PAYLOAD,
398 1 : (ctxIdx - 1));
399 1 : struct TaskPara taskPara;
400 1 : taskPara.stream = stream.ptr();
401 1 : taskPara.isMainStream = stream.IsMainStream();
402 1 : taskPara.beginTime = beginTime;
403 1 : taskPara.dma = para;
404 1 : taskPara.type = TaskType::TASK_RDMA;
405 1 : taskPara.isFftsDispatcher = true;
406 1 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
407 1 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
408 1 : }
409 2 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
410 2 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
411 1 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
412 1 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
413 1 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_PAYLOAD,
414 1 : (ctxIdx - 1));
415 1 : struct TaskPara taskPara;
416 1 : taskPara.stream = stream.ptr();
417 1 : taskPara.isMainStream = stream.IsMainStream();
418 1 : taskPara.beginTime = beginTime;
419 1 : taskPara.dma = para;
420 1 : taskPara.type = TaskType::TASK_RDMA;
421 1 : taskPara.isFftsDispatcher = true;
422 1 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
423 1 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
424 1 : }
425 1 : return HCCL_SUCCESS;
426 : }
427 :
428 1 : HcclResult DispatcherGraph::RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
429 : u32 userRank, u64 offset, bool isCapture)
430 : {
431 1 : uint64_t beginTime = GetMsprofSysCycleTime();
432 1 : if (UNLIKELY(disableFfts_)) {
433 0 : return DispatcherPub::RdmaSend(dbindex, dbinfo, wr, stream, userRank, offset);
434 : }
435 :
436 1 : u64 notifyID = (static_cast<u64>(userRank) << 32) | (offset & 0x00000000FFFFFFFF);
437 1 : PLF_CONFIG_INFO(PLF_TASK,
438 : "%s para: dbindex[%u], dbinfo[%llu], notifyId[0x%016llx], userRank[%u], offset[%llu], streamID[%u]",
439 : __func__, dbindex, dbinfo, notifyID, userRank, offset, stream.id());
440 :
441 1 : u32 ctxIdx = 0;
442 1 : CHK_RET(GraphAddRdmaSendTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), dbindex, dbinfo, isCapture, &ctxIdx));
443 : // 调用回调来保存task信息
444 1 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
445 : // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
446 1 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
447 1 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
448 1 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_NOTIFY,
449 1 : (ctxIdx - 1));
450 1 : struct TaskPara taskPara;
451 1 : taskPara.stream = stream.ptr();
452 1 : taskPara.isMainStream = stream.IsMainStream();
453 1 : taskPara.beginTime = beginTime;
454 1 : taskPara.dma = para;
455 1 : taskPara.type = TaskType::TASK_RDMA;
456 1 : taskPara.isFftsDispatcher = true;
457 1 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
458 1 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
459 1 : }
460 2 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
461 2 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
462 1 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
463 1 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
464 1 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_SEND_NOTIFY,
465 1 : (ctxIdx - 1));
466 1 : struct TaskPara taskPara;
467 1 : taskPara.stream = stream.ptr();
468 1 : taskPara.isMainStream = stream.IsMainStream();
469 1 : taskPara.beginTime = beginTime;
470 1 : taskPara.dma = para;
471 1 : taskPara.type = TaskType::TASK_RDMA;
472 1 : taskPara.isFftsDispatcher = true;
473 1 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
474 1 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
475 1 : }
476 1 : return HCCL_SUCCESS;
477 : }
478 :
479 0 : HcclResult DispatcherGraph::VectorReduce(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
480 : HcclReduceOp redOp, Stream &stream, const void *dst)
481 : {
482 0 : TbeReduceArg args{};
483 0 : if (count != 0) {
484 : #ifndef HCCD
485 0 : TbeReduceParam param;
486 0 : std::vector<void *> overflowAddrs;
487 0 : overflowAddrs.push_back(overflowAddr_);
488 0 : param.src1 = const_cast<void*>(src1);
489 0 : param.src2 = const_cast<void*>(src2);
490 0 : param.dst = const_cast<void*>(dst);
491 0 : param.count = count;
492 0 : param.dataType = dataType;
493 0 : param.redOp = redOp;
494 0 : CHK_RET(HcclTbeReduceGenArgs(¶m, stream.ptr(), overflowAddrs.data(), overflowAddrs.size(),
495 : &args, deviceLogicId_));
496 : #else
497 : HCCL_ERROR("[DispatcherGraph][VectorReduce] does not support this interface.");
498 : return HCCL_E_PARA;
499 : #endif
500 0 : GraphAddVectorReduceArgs(fftsPubInfo_, args.argsHandle);
501 0 : }
502 0 : CHK_RET(SetGraphDescVectorReduce(src1, dst, count, args.addrListDevMem, args.funcAddr,
503 : args.blockDim, dataType, redOp, stream));
504 0 : return HCCL_SUCCESS;
505 : }
506 :
507 0 : HcclResult DispatcherGraph::VectorReduceLoop(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
508 : HcclReduceOp redOp, Stream &stream, const void *dst)
509 : {
510 0 : const u32 unitSize = SIZE_TABLE[dataType];
511 0 : void *currentSrc1 = const_cast<void *>(src1);
512 0 : void *currentSrc2 = const_cast<void *>(src2);
513 0 : void *currentDst = const_cast<void *>(dst);
514 :
515 : // 计算出字节数为32字节整倍数的最大count
516 0 : const u64 maxCountPerLoop = ((TBE_REDUCE_MAX_COUNT * unitSize) / UB_BLOCK_SIZE * UB_BLOCK_SIZE) / unitSize;
517 :
518 0 : u64 countLeft = count;
519 :
520 : // 使用do while循环，是为了保证count为0时也进入一次VectorReduce，避免子图复用出错
521 : do {
522 0 : u64 currentCount = countLeft > maxCountPerLoop ? maxCountPerLoop : countLeft;
523 0 : HCCL_DEBUG(
524 : "[VectorReduceLoop] currentCount[%llu], countLeft[%llu], currentSrc1[%p], currentSrc2[%p], currentDst[%p]",
525 : currentCount, countLeft, currentSrc1, currentSrc2, currentDst);
526 :
527 0 : CHK_RET(VectorReduce(currentSrc1, currentSrc2, currentCount, dataType, redOp, stream, currentDst));
528 :
529 0 : currentSrc1 = static_cast<void *>(static_cast<s8 *>(currentSrc1) + currentCount * unitSize);
530 0 : currentSrc2 = static_cast<void *>(static_cast<s8 *>(currentSrc2) + currentCount * unitSize);
531 0 : currentDst = static_cast<void *>(static_cast<s8 *>(currentDst) + currentCount * unitSize);
532 0 : countLeft -= currentCount;
533 0 : } while (countLeft > 0);
534 :
535 0 : return HCCL_SUCCESS;
536 : }
537 :
538 0 : HcclResult DispatcherGraph::SetGraphTailVectorReduceDescSdma(void *devMem, const void *tailSrc, u64 count,
539 : const HcclDataType dataType, HcclReduceOp redOp, Stream &stream)
540 : {
541 0 : uint64_t beginTime = GetMsprofSysCycleTime();
542 0 : u32 ctxIdx = 0;
543 0 : CHK_RET(GraphAddTailVectorReduceTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), devMem, tailSrc, count, &ctxIdx));
544 :
545 : // 调用回调来保存 task 信息
546 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
547 : hccl::TaskParaDMA para(tailSrc, devMem, count, LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID,
548 0 : hccl::RdmaType::RDMA_TYPE_RESERVED, (ctxIdx - 1));
549 0 : struct TaskPara taskPara;
550 0 : taskPara.stream = stream.ptr();
551 0 : taskPara.isMainStream = stream.IsMainStream();
552 0 : taskPara.beginTime = beginTime;
553 0 : taskPara.dma = para;
554 0 : taskPara.type = TaskType::TASK_SDMA;
555 0 : taskPara.isFftsDispatcher = true;
556 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
557 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
558 0 : }
559 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
560 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
561 : hccl::TaskParaDMA para(tailSrc, devMem, count, LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID,
562 0 : hccl::RdmaType::RDMA_TYPE_RESERVED, (ctxIdx - 1));
563 0 : struct TaskPara taskPara;
564 0 : taskPara.stream = stream.ptr();
565 0 : taskPara.isMainStream = stream.IsMainStream();
566 0 : taskPara.beginTime = beginTime;
567 0 : taskPara.dma = para;
568 0 : taskPara.type = TaskType::TASK_SDMA;
569 0 : taskPara.isFftsDispatcher = true;
570 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
571 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
572 0 : }
573 0 : return HCCL_SUCCESS;
574 : }
575 :
576 0 : HcclResult DispatcherGraph::SetGraphDescVectorReduce(const void *src, const void *dst, int count, void *addrListDevMemPtr,
577 : void *funcAddr, uint32_t numBlocks, const HcclDataType dataType, HcclReduceOp redOp, Stream &stream)
578 : {
579 0 : uint64_t beginTime = GetMsprofSysCycleTime();
580 0 : u32 ctxIdx = 0;
581 0 : CHK_RET(GraphAddVectorReduceTask(fftsPubInfo_, fftsCtxsPtr, stream.id(), count, addrListDevMemPtr,
582 : funcAddr, numBlocks, &ctxIdx));
583 :
584 : // 调用回调来保存 task 信息
585 0 : if (DispatcherPub::IsProfSubscribeAdditionInfo() && callback_ != nullptr) {
586 0 : hccl::TaskParaReduce para(src, dst, count * SIZE_TABLE[dataType], redOp, dataType,
587 0 : LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (ctxIdx - 1));
588 0 : struct TaskPara taskPara;
589 0 : taskPara.stream = stream.ptr();
590 0 : taskPara.isMainStream = stream.IsMainStream();
591 0 : taskPara.beginTime = beginTime;
592 0 : taskPara.reduce = para;
593 0 : taskPara.type = TaskType::TASK_REDUCE_TBE;
594 0 : taskPara.isFftsDispatcher = true;
595 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
596 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
597 0 : }
598 0 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE &&
599 0 : GetExternalInputTaskExceptionSwitch() == 1 && callback_ != nullptr) {
600 0 : hccl::TaskParaReduce para(src, dst, count * SIZE_TABLE[dataType], redOp, dataType,
601 0 : LinkType::LINK_ONCHIP, INVALID_VALUE_RANKID, (ctxIdx - 1));
602 0 : struct TaskPara taskPara;
603 0 : taskPara.stream = stream.ptr();
604 0 : taskPara.isMainStream = stream.IsMainStream();
605 0 : taskPara.beginTime = beginTime;
606 0 : taskPara.reduce = para;
607 0 : taskPara.type = TaskType::TASK_REDUCE_TBE;
608 0 : taskPara.isFftsDispatcher = true;
609 0 : taskPara.profilerType = ProfilerType::TASK_EXCEPTION;
610 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
611 0 : }
612 0 : return HCCL_SUCCESS;
613 : }
614 :
615 0 : HcclResult DispatcherGraph::TailVectorReduce(const void *tailSrc1, const void *tailSrc2, u64 tailCount,
616 : const HcclDataType dataType, HcclReduceOp redOp, Stream &stream, void *tailDst)
617 : {
618 0 : void *devMem1 = nullptr;
619 0 : void *devMem2 = nullptr;
620 0 : TbeReduceArg args{};
621 0 : if (tailCount != 0) {
622 : #ifndef HCCD
623 0 : CHK_RET(DevMemMalloc(stream.ptr(), devMem1, devMem2));
624 0 : TbeReduceParam param;
625 0 : std::vector<void *> overflowAddrs;
626 0 : overflowAddrs.push_back(overflowAddr_);
627 0 : param.src1 = devMem1;
628 0 : param.src2 = devMem2;
629 0 : param.dst = devMem2;
630 0 : param.count = tailCount;
631 0 : param.dataType = dataType;
632 0 : param.redOp = redOp;
633 0 : CHK_RET(HcclTbeReduceGenArgs(¶m, stream.ptr(), overflowAddrs.data(), overflowAddrs.size(),
634 : &args, deviceLogicId_));
635 : #else
636 : HCCL_ERROR("[DispatcherGraph][VectorReduce] does not support this interface.");
637 : return HCCL_E_PARA;
638 : #endif
639 0 : GraphAddVectorReduceArgs(fftsPubInfo_, args.argsHandle);
640 0 : }
641 0 : u64 dataCount = tailCount * SIZE_TABLE[dataType];
642 :
643 0 : CHK_RET(SetGraphTailVectorReduceDescSdma(devMem1, tailSrc1, dataCount, dataType, redOp, stream));
644 0 : CHK_RET(SetGraphTailVectorReduceDescSdma(devMem2, tailSrc2, dataCount, dataType, redOp, stream));
645 0 : CHK_RET(SetGraphDescVectorReduce(devMem1, tailDst, tailCount, args.addrListDevMem, args.funcAddr,
646 : args.blockDim, dataType, redOp, stream));
647 0 : CHK_RET(SetGraphTailVectorReduceDescSdma(tailDst, devMem2, dataCount, dataType, redOp, stream));
648 0 : return HCCL_SUCCESS;
649 : }
650 :
651 0 : HcclResult DispatcherGraph::TbeReduceAsync(const void *src1, const void *src2, u64 count, const HcclDataType dataType,
652 : HcclReduceOp redOp, Stream &stream, const void *dst)
653 : {
654 0 : void *tailSrc1 = nullptr;
655 0 : void *tailSrc2 = nullptr;
656 0 : void *tailDst = nullptr;
657 0 : u64 headCount = 0;
658 0 : u64 tailCount = 0;
659 : #ifndef HCCD
660 0 : CHK_RET(JudgeIsTail(src1, src2, dst, count, dataType, headCount, tailCount, tailSrc1, tailSrc2,
661 : tailDst));
662 : #else
663 : HCCL_ERROR("[DispatcherGraph][TbeReduceAsync] does not support this interface.");
664 : return HCCL_E_PARA;
665 : #endif
666 0 : CHK_RET(VectorReduceLoop(src1, src2, headCount, dataType, redOp, stream, dst));
667 0 : CHK_RET(TailVectorReduce(tailSrc1, tailSrc2, tailCount, dataType, redOp, stream, tailDst));
668 :
669 0 : return HCCL_SUCCESS;
670 : }
671 :
672 0 : HcclResult DispatcherGraph::SignalRecord(Stream &stream, u64 notifyId)
673 : {
674 0 : if (UNLIKELY(disableFfts_)) {
675 0 : return DispatcherPub::SignalRecord(stream, notifyId);
676 : }
677 :
678 0 : CHK_RET(GraphAddRecordTaskById(fftsPubInfo_, fftsCtxsPtr, static_cast<u32>(notifyId), stream.id()));
679 :
680 0 : if (HcclCheckLogLevel(HCCL_LOG_INFO) || (GetExternalInputDebugConfig() & PLF_TASK)) {
681 0 : PLF_CONFIG_INFO(PLF_TASK,
682 : "%s para: notifyId[0x%016llx] streamId[%u]", __func__, notifyId, stream.id());
683 : }
684 :
685 0 : return HCCL_SUCCESS;
686 : }
687 :
688 0 : HcclResult DispatcherGraph::SignalWait(Stream &stream, u32 notifyId, u32 timeOut)
689 : {
690 0 : if (UNLIKELY(disableFfts_)) {
691 0 : return DispatcherPub::SignalWait(stream, notifyId, timeOut);
692 : }
693 0 : CHK_RET(GraphAddWaitTaskById(fftsPubInfo_, fftsCtxsPtr, static_cast<u32>(notifyId), stream.id()));
694 :
695 0 : if (HcclCheckLogLevel(HCCL_LOG_INFO) || (GetExternalInputDebugConfig() & PLF_TASK)) {
696 0 : PLF_CONFIG_INFO(PLF_TASK,
697 : "%s para: notifyId[0x%016llx] streamId[%u]", __func__, notifyId, stream.id());
698 : }
699 :
700 0 : return HCCL_SUCCESS;
701 : }
702 :
703 : } // namespace hccl
|
