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