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 "dispatcher_pub.h"
12 : #include "externalinput_pub.h"
13 : #include "externalinput.h"
14 : #include "adapter_rts.h"
15 : #include "sal_pub.h"
16 : #include "config_plf_log.h"
17 : #include "hccl_tbe_task.h"
18 : #ifndef HCCD
19 : #include "graph_ctx_mgr_common.h"
20 : #endif
21 :
22 : using namespace hccl;
23 :
24 : #if T_DESC("DispatcherPub", true)
25 :
26 : namespace {
27 : HcclResult g_callBackResult = HCCL_SUCCESS;
28 : const std::map<HcclDataType, aclDataType> HCCL_RT_DATA_TYPE_MAP = {
29 : {HCCL_DATA_TYPE_INT8, ACL_INT8},
30 : {HCCL_DATA_TYPE_INT16, ACL_INT16},
31 : {HCCL_DATA_TYPE_INT32, ACL_INT32},
32 : {HCCL_DATA_TYPE_FP16, ACL_FLOAT16},
33 : {HCCL_DATA_TYPE_FP32, ACL_FLOAT},
34 : {HCCL_DATA_TYPE_BFP16, ACL_BF16},
35 : };
36 : const std::map<HcclReduceOp, aclrtReduceKind> HCCL_RT_REDUCE_OP_MAP = {
37 : {HCCL_REDUCE_SUM, ACL_RT_MEMCPY_SDMA_AUTOMATIC_SUM},
38 : {HCCL_REDUCE_MAX, ACL_RT_MEMCPY_SDMA_AUTOMATIC_MAX},
39 : {HCCL_REDUCE_MIN, ACL_RT_MEMCPY_SDMA_AUTOMATIC_MIN},
40 : };
41 : }
42 :
43 : bool DispatcherPub::isForce_ = false;
44 :
45 1593 : DispatcherPub::DispatcherPub(const s32 deviceLogicId)
46 3186 : : deviceLogicId_(deviceLogicId), notifyWaitMode_(SyncMode::DEFAULT_TIMEWAITSYNCMODE),
47 3186 : hostNicTcpSendThreadState_(true), overflowAddr_(nullptr), setDeviceFlag_(false),
48 1593 : execTimeOut_(NOTIFY_DEFAULT_WAIT_TIME), execTimeOutByConfig_(false)
49 : {
50 1593 : }
51 :
52 2098 : DispatcherPub::~DispatcherPub()
53 : {
54 1593 : HcclResult ret = HCCL_SUCCESS;
55 : #ifndef HCCD
56 1593 : std::map<int32_t, void *>::iterator devMemIter;
57 1593 : std::unique_lock<std::mutex> lock(devMemMutex_);
58 1592 : for (devMemIter = devMemMap_.begin(); devMemIter != devMemMap_.end(); devMemIter++) {
59 0 : if (devMemIter->second != nullptr) {
60 0 : if (hrtFree(devMemIter->second) != HCCL_SUCCESS) {
61 0 : HCCL_WARNING("free device memory failed");
62 : }
63 0 : devMemIter->second = nullptr;
64 : }
65 : }
66 1589 : devMemMap_.clear();
67 1591 : if (deviceLogicId_ != HOST_DEVICE_ID) {
68 1146 : ret = HcclTbeTaskDeInit(deviceLogicId_);
69 1145 : if (ret != HCCL_SUCCESS) {
70 0 : HCCL_ERROR("tbe task deinit failed. ret[%d] device id[%d]", ret, deviceLogicId_);
71 : }
72 : }
73 1590 : lock.unlock();
74 :
75 1591 : if (fftsPubInfo_ != nullptr) {
76 552 : GraphMgrDeInit(fftsPubInfo_);
77 553 : fftsPubInfo_ = nullptr;
78 : }
79 : #endif
80 :
81 : // 清空task信息
82 1592 : if (hostNicTcpSendThread_ != nullptr) {
83 0 : WaitHostNicTcpSendThreadComplete();
84 : }
85 1590 : ClearHostNicRdmaParamsVec();
86 1590 : ClearHostNicTcpSendParamsVec();
87 1592 : ClearHostNicTcpRecvParamsVec();
88 :
89 1587 : if (setDeviceFlag_) {
90 0 : ret = hrtResetDevice(deviceLogicId_);
91 0 : if (ret != HCCL_SUCCESS) {
92 0 : HCCL_ERROR("[DispatcherPub][Destroy]In dispathcer enhanced destruct, reset device failed.errno[%d] "\
93 : "device id[%d]", ret, deviceLogicId_);
94 : }
95 : }
96 2092 : }
97 :
98 0 : void DispatcherPub::JudgeOpBaseTcpSendComplete(bool &closeSendThreadFlag)
99 : {
100 0 : bool hostNicTcpSendParamsVecIsEmpty = true;
101 0 : for (auto it = hostNicTcpSendParamsVec_.begin(); it != hostNicTcpSendParamsVec_.end(); it++) {
102 0 : if (it->second.size() != 0) {
103 0 : hostNicTcpSendParamsVecIsEmpty = false;
104 0 : HCCL_WARNING("host nic TCP send task is not completed. streamID[%llu], size[%llu]",
105 : it->first, it->second.size());
106 : }
107 : }
108 0 : closeSendThreadFlag = (hostNicTcpSendThreadParam_ == nullptr) && hostNicTcpSendParamsVecIsEmpty;
109 0 : }
110 :
111 0 : void DispatcherPub::WaitHostNicTcpSendThreadComplete()
112 : {
113 : // 等待tcp send线程join
114 0 : bool closeSendThreadFlag = true;
115 : while (true) {
116 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
117 0 : if (workflowMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
118 0 : JudgeOpBaseTcpSendComplete(closeSendThreadFlag);
119 : } else {
120 0 : closeSendThreadFlag = (hostNicTcpSendThreadParam_ == nullptr);
121 : }
122 0 : if (closeSendThreadFlag) {
123 0 : break;
124 : }
125 0 : HCCL_WARNING("host nic TCP send thread is not finished");
126 0 : SaluSleep(TCP_SEND_THREAD_SLEEP_TWO_HUNDRED_MICROSECOND);
127 0 : }
128 0 : hostNicTcpSendThreadState_ = false;
129 0 : if (hostNicTcpSendThread_ != nullptr && hostNicTcpSendThread_->joinable()) {
130 0 : hostNicTcpSendThread_->join(); // 等待线程执行完毕
131 : }
132 0 : }
133 :
134 1590 : void DispatcherPub::ClearHostNicRdmaParamsVec()
135 : {
136 1590 : for (auto it = hostNicRdmaParamsVec_.begin(); it != hostNicRdmaParamsVec_.end(); it++) {
137 0 : if (it->second.size() != 0) {
138 0 : HCCL_WARNING("host nic RDMA task is not completed. streamID[%llu], size[%llu]",
139 : it->first, it->second.size());
140 0 : while (!it->second.empty()) {
141 0 : it->second.pop();
142 : }
143 : }
144 : }
145 1592 : hostNicRdmaParamsVec_.clear();
146 1589 : }
147 :
148 1590 : void DispatcherPub::ClearHostNicTcpSendParamsVec()
149 : {
150 1590 : for (auto it = hostNicTcpSendParamsVec_.begin(); it != hostNicTcpSendParamsVec_.end(); it++) {
151 0 : if (it->second.size() != 0) {
152 0 : HCCL_WARNING("host nic TCP send task is not completed. streamID[%llu], size[%llu]",
153 : it->first, it->second.size());
154 0 : while (!it->second.empty()) {
155 0 : it->second.pop();
156 : }
157 : }
158 : }
159 1593 : hostNicTcpSendParamsVec_.clear();
160 1592 : }
161 :
162 1592 : void DispatcherPub::ClearHostNicTcpRecvParamsVec()
163 : {
164 1592 : for (auto it = hostNicTcpRecvParamsVec_.begin(); it != hostNicTcpRecvParamsVec_.end(); it++) {
165 0 : if (it->second.size() != 0) {
166 0 : HCCL_WARNING("host nic TCP recv task is not completed. streamID[%llu], size[%llu]",
167 : it->first, it->second.size());
168 0 : while (!it->second.empty()) {
169 0 : it->second.pop();
170 : }
171 : }
172 : }
173 1588 : hostNicTcpRecvParamsVec_.clear();
174 1587 : }
175 :
176 0 : void DispatcherPub::WaitHostNicTcpSendTaskDone()
177 : {
178 0 : while (hostNicTcpSendThreadParam_ != nullptr) {
179 0 : SaluSleep(TCP_SEND_THREAD_SLEEP_TWO_HUNDRED_MICROSECOND);
180 : }
181 0 : }
182 :
183 : // 获取算子最大超时时间
184 998 : HcclResult DispatcherPub::GetNotifyMaxWaitTime()
185 : {
186 : DevType deviceType;
187 998 : CHK_RET(hrtGetDeviceType(deviceType));
188 998 : notifyMaxWaitTime_ = (deviceType == DevType::DEV_TYPE_910_93 || deviceType == DevType::DEV_TYPE_910B) ?\
189 : NOTIFY_MAX_WAIT_TIME_910_93 : NOTIFY_MAX_WAIT_TIME;
190 998 : HCCL_INFO("[GetNotifyMaxWaitTime] notifyMaxWaitTime_ is %us", notifyMaxWaitTime_);
191 998 : return HCCL_SUCCESS;
192 : }
193 :
194 0 : s32 DispatcherPub::GetExecTimeOut()
195 : {
196 0 : return execTimeOut_;
197 : }
198 0 : bool DispatcherPub::GetExecTimeOutSet()
199 : {
200 0 : return execTimeOutByConfig_;
201 : }
202 :
203 554 : HcclResult DispatcherPub::Init()
204 : {
205 : #ifndef HCCD
206 554 : if (deviceLogicId_ == HOST_DEVICE_ID) {
207 0 : return HCCL_SUCCESS;
208 : }
209 :
210 554 : aclrtContext ctx = nullptr;
211 554 : CHK_RET(hrtCtxGetCurrent(&ctx));
212 554 : if (ctx == nullptr) {
213 0 : CHK_RET(hrtSetDevice(deviceLogicId_));
214 0 : setDeviceFlag_ = true;
215 : }
216 :
217 554 : CHK_RET(HcclTbeTaskInit(deviceLogicId_));
218 :
219 554 : fftsPubInfo_ = GraphMgrInit();
220 554 : CHK_PTR_NULL(fftsPubInfo_);
221 :
222 1108 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET ||
223 554 : execTimeOutByConfig_) {
224 0 : notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
225 : }
226 :
227 554 : if (GetExternalInputHcclIsTcpMode()) {
228 0 : hostNicTcpSendThread_.reset(new (std::nothrow) std::thread(&DispatcherPub::HostNicTcpSendThreadTask, this));
229 : }
230 :
231 554 : CHK_RET(GetNotifyMaxWaitTime());
232 : #else
233 : HCCL_ERROR("does not support this interface.");
234 : return HCCL_E_PARA;
235 : #endif
236 :
237 554 : return HCCL_SUCCESS;
238 : }
239 :
240 32 : void DispatcherPub::SetupTaskParaDma(hccl::TaskPara& taskPara, hccl::TaskParaDMA& para, TaskType taskType,
241 : ProfilerType profilerType, hccl::Stream &stream, u64 beginTime, bool isMainStream) const
242 : {
243 32 : taskPara.type = taskType;
244 32 : taskPara.profilerType = profilerType;
245 32 : taskPara.stream = stream.ptr();
246 32 : taskPara.beginTime = beginTime;
247 32 : taskPara.dma = para;
248 32 : taskPara.isMainStream = isMainStream;
249 32 : }
250 :
251 0 : void DispatcherPub::SetupTaskParaDma(hccl::TaskPara& taskPara, hccl::TaskParaDMA& para, TaskType taskType,
252 : HcclRtStream stream, u64 beginTime, bool isMainStream) const
253 : {
254 0 : taskPara.type = taskType;
255 0 : taskPara.stream = stream;
256 0 : taskPara.beginTime = beginTime;
257 0 : taskPara.dma = para;
258 0 : taskPara.isMainStream = isMainStream;
259 0 : }
260 :
261 0 : HcclResult DispatcherPub::SignalRecord(HcclRtNotify signal, HcclRtStream stream, \
262 : u32 userRank, u64 offset, s32 stage, bool isMainStream)
263 : {
264 0 : uint64_t beginTime = GetMsprofSysCycleTime();
265 0 : CHK_RET(hrtNotifyRecord(static_cast<HcclRtNotify>(signal), stream));
266 :
267 : // 若没有输入offset， 则认为record的为本地notify，直接获取其offset
268 0 : u64 NotifyID = userRank;
269 0 : if (offset == INVALID_U64) {
270 0 : CHK_RET(hrtNotifyGetOffset(static_cast<HcclRtNotify>(signal), offset));
271 : }
272 0 : NotifyID = (NotifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
273 : // 调用回调来保存task信息
274 0 : if (callback_ != nullptr) {
275 0 : hccl::TaskParaNotify para(NotifyID, stage);
276 0 : hccl::TaskPara taskPara(TaskType::TASK_NOTIFY_RECORD, para);
277 0 : taskPara.stream = stream;
278 0 : taskPara.beginTime = beginTime;
279 0 : taskPara.isMainStream = isMainStream;
280 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
281 0 : }
282 :
283 0 : u32 taskID = 0;
284 0 : u32 streamID = 0;
285 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
286 0 : PLF_CONFIG_INFO(PLF_TASK,
287 : "%s para: notifyId[0x%016llx] taskId[%u] streamID[%u] userRank[%u] offset[%llu] stage[%d]",
288 : __func__, NotifyID, taskID, streamID, userRank, offset, stage);
289 0 : return HCCL_SUCCESS;
290 : }
291 :
292 0 : u32 DispatcherPub::GetNotifyWaitTime(u32 timeOut)
293 : {
294 0 : u32 notifyWaitTime = 0;
295 0 : if (timeOut > 0 && timeOut <= notifyMaxWaitTime_) {
296 0 : notifyWaitTime = timeOut;
297 0 : } else if (notifyWaitMode_ == SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE) {
298 0 : notifyWaitTime = execTimeOut_;
299 0 : } else if (notifyWaitMode_ == SyncMode::UNLIMITED_TIMEWAITSYNCMODE) {
300 0 : notifyWaitTime = notifyMaxWaitTime_;
301 : } else {
302 0 : notifyWaitTime = NOTIFY_DEFAULT_WAIT_TIME;
303 : }
304 0 : return notifyWaitTime;
305 : }
306 :
307 0 : HcclResult DispatcherPub::SignalWait(HcclRtNotify signal, HcclRtStream stream, u32 userRank,
308 : u32 remoteUserRank, s32 stage, u32 timeOut, bool isMainStream)
309 : {
310 0 : uint64_t beginTime = GetMsprofSysCycleTime();
311 0 : CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream, GetNotifyWaitTime(timeOut)));
312 :
313 : // 调用回调来保存task信息
314 0 : u64 NotifyID = userRank;
315 0 : u64 offset = 0;
316 0 : CHK_RET(hrtNotifyGetOffset(static_cast<HcclRtNotify>(signal), offset));
317 :
318 0 : NotifyID = (NotifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
319 0 : if (callback_ != nullptr) {
320 0 : hccl::TaskParaNotify para(NotifyID, stage, remoteUserRank);
321 0 : hccl::TaskPara taskPara(TaskType::TASK_NOTIFY_WAIT, para);
322 0 : taskPara.stream = stream;
323 0 : taskPara.beginTime = beginTime;
324 0 : taskPara.isMainStream = isMainStream;
325 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
326 0 : }
327 :
328 0 : u32 taskID = 0;
329 0 : u32 streamID = 0;
330 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
331 0 : PLF_CONFIG_INFO(PLF_TASK,
332 : "%s para: notifyId[0x%016llx] taskId[%u] streamID[%u] userRank[%u] remoteUserRank[%u] stage[%d] timeOut[%u s]",
333 : __func__, NotifyID, taskID, streamID, userRank, remoteUserRank, stage, timeOut);
334 0 : return HCCL_SUCCESS;
335 : }
336 :
337 0 : HcclResult DispatcherPub::SetNotifyWaitMode(SyncMode notifyWaitMode)
338 : {
339 0 : notifyWaitMode_ = notifyWaitMode;
340 0 : return HCCL_SUCCESS;
341 : }
342 :
343 0 : SyncMode DispatcherPub::GetNotifyWaitMode()
344 : {
345 0 : return notifyWaitMode_;
346 : }
347 :
348 1042 : HcclResult DispatcherPub::SetHcclExecTimeOut(s32 execTimeOut)
349 : {
350 1042 : execTimeOut_ = execTimeOut;
351 1042 : execTimeOutByConfig_ = true;
352 1042 : return HCCL_SUCCESS;
353 : }
354 :
355 0 : HcclResult DispatcherPub::MemcpySync(void *dst, uint64_t destMax, const void *src, uint64_t count,
356 : HcclRtMemcpyKind kind)
357 : {
358 0 : return hrtMemSyncCopy(dst, destMax, src, count, kind);
359 : }
360 :
361 83 : HcclResult DispatcherPub::MemcpyAsync(void *dst, uint64_t destMax, const void *src, u64 count,
362 : HcclRtMemcpyKind kind, Stream &stream,
363 : u32 remoteUserRank, hccl::LinkType inLinkType)
364 : {
365 83 : uint64_t beginTime = GetMsprofSysCycleTime();
366 :
367 : // 参数有效性检查
368 83 : if (stream.ptr() == nullptr) {
369 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dst, destMax, src, count));
370 0 : return HCCL_SUCCESS;
371 : }
372 :
373 83 : if (count == 0) {
374 43 : HCCL_DEBUG("count is 0, return success.");
375 43 : return HCCL_SUCCESS;
376 : }
377 :
378 40 : if (src == dst) {
379 2 : HCCL_DEBUG("src == dst, return success.");
380 2 : return HCCL_SUCCESS;
381 : }
382 :
383 38 : if (destMax < count) {
384 0 : HCCL_ERROR("The size of destMax is smaller than that of count. destMax[%llu], count[%llu]", destMax, count);
385 0 : return HCCL_E_PARA;
386 : }
387 :
388 38 : uint64_t spiltLoop = 0;
389 38 : uint64_t addrOffset = 0;
390 38 : uint64_t contSplit = 0;
391 38 : if (count > HCCL_SDMA_MAX_COUNT_4GB) {
392 0 : spiltLoop = (count % HCCL_SDMA_MAX_COUNT_4GB) ?
393 0 : (count / HCCL_SDMA_MAX_COUNT_4GB) : ((count / HCCL_SDMA_MAX_COUNT_4GB) - 1);
394 0 : HCCL_INFO("MemcpyAsync SDMA task countSize is bigger than 4GB and do segmentation splitloop[%llu]", spiltLoop);
395 : }
396 : /* SDMA任务拆分 */
397 76 : for (uint64_t index = 0 ; index <= spiltLoop; index++) {
398 38 : addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
399 38 : contSplit = (index == spiltLoop) ? (count - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
400 38 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void*>(src)) + addrOffset);
401 38 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addrOffset);
402 :
403 38 : CHK_RET(hrtMemAsyncCopy(dstSplit, destMax, srcSplit, contSplit, kind,
404 : stream.ptr()));
405 : // 调用回调来保存task信息
406 38 : if (callback_ != nullptr) {
407 32 : hccl::TaskParaDMA para((const void*)srcSplit, dstSplit, contSplit, inLinkType, remoteUserRank);
408 32 : hccl::TaskPara taskPara;
409 32 : SetupTaskParaDma(taskPara, para, TaskType::TASK_SDMA, ProfilerType::TASK_ALL, stream, beginTime,
410 32 : stream.IsMainStream());
411 32 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
412 32 : }
413 38 : u32 taskID = 0;
414 38 : u32 streamID = 0;
415 38 : hrtGetTaskIdAndStreamID(taskID, streamID);
416 38 : PLF_CONFIG_INFO(PLF_TASK,
417 : "%s para: dst[%p] destMax[%llu] src[%p] count[%llu] rtMemcpyKind[%d] taskID[%u] streamID[%u] "\
418 : "remoteUserRank[%u] inLinkType[%d]", __func__, dstSplit, destMax, srcSplit,
419 : contSplit, kind, taskID, streamID, remoteUserRank, inLinkType);
420 : }
421 :
422 38 : return HCCL_SUCCESS;
423 : }
424 :
425 0 : HcclResult DispatcherPub::MemcpyAsync(hccl::HostMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream)
426 : {
427 0 : if (dst.size() < src.size()) {
428 0 : HCCL_ERROR(
429 : "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
430 : dst.ptr(), dst.size(), src.ptr(), src.size());
431 0 : return HCCL_E_PTR;
432 : }
433 :
434 0 : CHK_RET(MemcpyAsync(dst.ptr(), dst.size(), src.ptr(), src.size(),
435 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST, stream));
436 :
437 0 : return HCCL_SUCCESS;
438 : }
439 :
440 0 : HcclResult DispatcherPub::MemcpyAsync(hccl::HostMem &dst, const hccl::HostMem &src, hccl::Stream &stream)
441 : {
442 0 : if (dst.size() < src.size()) {
443 0 : HCCL_ERROR(
444 : "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
445 : dst.ptr(), dst.size(), src.ptr(), src.size());
446 0 : return HCCL_E_PTR;
447 : }
448 :
449 0 : CHK_RET(MemcpyAsync(dst.ptr(), dst.size(), src.ptr(), src.size(),
450 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_HOST, stream));
451 :
452 0 : return HCCL_SUCCESS;
453 : }
454 :
455 84 : HcclResult DispatcherPub::MemcpyAsync(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream,
456 : u32 remoteUserRank, hccl::LinkType inLinkType)
457 : {
458 84 : if (dst.size() < src.size()) {
459 1 : HCCL_ERROR(
460 : "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
461 : dst.ptr(), dst.size(), src.ptr(), src.size());
462 1 : return HCCL_E_PTR;
463 : }
464 :
465 83 : return MemcpyAsync(dst.ptr(), dst.size(), src.ptr(), src.size(),
466 83 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream, remoteUserRank, inLinkType);
467 : }
468 :
469 0 : HcclResult DispatcherPub::MemcpyAsync(hccl::DeviceMem &dst, const hccl::HostMem &src, hccl::Stream &stream)
470 : {
471 0 : if (dst.size() < src.size()) {
472 0 : HCCL_ERROR(
473 : "The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], src addr[%p], src size[%llu]",
474 : dst.ptr(), dst.size(), src.ptr(), src.size());
475 0 : return HCCL_E_PTR;
476 : }
477 :
478 0 : CHK_RET(MemcpyAsync(dst.ptr(), dst.size(), src.ptr(), src.size(),
479 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, stream));
480 :
481 0 : return HCCL_SUCCESS;
482 : }
483 :
484 0 : HcclResult DispatcherPub::MemcpyAsyncWithoutCheckKind(void *dst, uint64_t destMax, const void *src, u64 count,
485 : HcclRtMemcpyKind kind, Stream &stream,
486 : u32 remoteUserRank, hccl::LinkType inLinkType)
487 : {
488 0 : uint64_t beginTime = GetMsprofSysCycleTime();
489 :
490 : // 参数有效性检查
491 0 : if (stream.ptr() == nullptr) {
492 0 : HCCL_DEBUG("stream ptr is null, use memcpy.");
493 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dst, destMax, src, count));
494 0 : return HCCL_SUCCESS;
495 : }
496 :
497 0 : if (count == 0 || src == dst) {
498 0 : HCCL_DEBUG("count[%llu]] is 0 or src is equal to dst, return success.", count);
499 0 : return HCCL_SUCCESS;
500 : }
501 :
502 0 : if (destMax < count) {
503 0 : HCCL_ERROR("The size of destMax is smaller than that of count. destMax[%llu], count[%llu]", destMax, count);
504 0 : return HCCL_E_PARA;
505 : }
506 :
507 0 : uint64_t spiltLoop = 0;
508 0 : uint64_t addrOffset = 0;
509 0 : uint64_t contSplit = 0;
510 0 : if (count > HCCL_SDMA_MAX_COUNT_4GB) {
511 0 : spiltLoop = (count % HCCL_SDMA_MAX_COUNT_4GB) ?
512 0 : (count / HCCL_SDMA_MAX_COUNT_4GB) : ((count / HCCL_SDMA_MAX_COUNT_4GB) - 1);
513 0 : HCCL_INFO("MemcpyAsync SDMA task countSize is bigger than 4GB and do segmentation splitloop[%llu]", spiltLoop);
514 : }
515 : /* SDMA任务拆分 */
516 0 : for (uint64_t index = 0 ; index <= spiltLoop; index++) {
517 0 : addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
518 0 : contSplit = (index == spiltLoop) ? (count - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
519 0 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void*>(src)) + addrOffset);
520 0 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addrOffset);
521 :
522 0 : CHK_RET(hrtMemAsyncCopyWithoutCheckKind(dstSplit, destMax, srcSplit, contSplit, kind,
523 : stream.ptr()));
524 :
525 : // 调用回调来保存task信息
526 0 : if (callback_ != nullptr) {
527 0 : hccl::TaskParaDMA para((const void*)srcSplit, dstSplit, contSplit, inLinkType, remoteUserRank);
528 0 : hccl::TaskPara taskPara;
529 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_SDMA, ProfilerType::TASK_ALL, stream, beginTime,
530 0 : stream.IsMainStream());
531 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
532 0 : }
533 0 : u32 taskID = 0;
534 0 : u32 streamID = 0;
535 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
536 0 : PLF_CONFIG_INFO(PLF_TASK,
537 : "%s para: dst[%p] destMax[%llu] src[%p] count[%llu] rtMemcpyKind[%d] taskID[%u] streamID[%u] "\
538 : "remoteUserRank[%u] inLinkType[%d]", __func__, dstSplit, destMax, srcSplit,
539 : contSplit, kind, taskID, streamID, remoteUserRank, inLinkType);
540 : }
541 :
542 0 : return HCCL_SUCCESS;
543 : }
544 :
545 0 : HcclResult DispatcherPub::DevMemMalloc(void *stream, void *&devMem1, void *&devMem2)
546 : {
547 : #ifndef HCCD
548 : int32_t streamId;
549 : u32 blockSize;
550 0 : CHK_RET(hrtGetStreamId(stream, streamId));
551 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
552 :
553 0 : std::unique_lock<std::mutex> lock(devMemMutex_);
554 0 : if (devMemMap_.find(streamId) == devMemMap_.end()) {
555 0 : u32 devMemSize = blockSize + blockSize;
556 0 : CHK_RET(hrtMalloc(&devMem1, devMemSize));
557 0 : CHK_PTR_NULL(devMem1);
558 0 : CHK_RET(hrtMemSet(devMem1, devMemSize, devMemSize));
559 0 : devMem2 = static_cast<char *>(devMem1) + blockSize;
560 0 : devMemMap_[streamId] = devMem1;
561 : } else {
562 0 : devMem1 = devMemMap_[streamId];
563 0 : devMem2 = static_cast<char *>(devMem1) + blockSize;
564 : }
565 : #endif
566 0 : return HCCL_SUCCESS;
567 0 : }
568 :
569 0 : HcclResult DispatcherPub::JudgeIsTail(const void *src1, const void *src2, const void *dst, u64 count,
570 : const HcclDataType dataType, u64 &headCount, u64 &tailCount, void *&tailSrc1, void *&tailSrc2, void *&tailDst)
571 : {
572 : #ifndef HCCD
573 0 : CHK_PRT_RET(dataType >= HCCL_DATA_TYPE_RESERVED, HCCL_ERROR("dataType is failed."), HCCL_E_PARA);
574 0 : u32 blockSize = 0;
575 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
576 : // 获取总的数据量
577 0 : u64 dataSize = SIZE_TABLE[dataType] * count; // 计算总的字节数
578 0 : headCount = dataSize / blockSize * blockSize / SIZE_TABLE[dataType]; // 计算出32字节整倍数的数据数量
579 0 : tailCount = count - headCount;
580 :
581 0 : if (tailCount != 0) {
582 0 : tailSrc1 = static_cast<char *>(const_cast<void *>(src1)) + (headCount * SIZE_TABLE[dataType]);
583 0 : tailSrc2 = static_cast<char *>(const_cast<void *>(src2)) + (headCount * SIZE_TABLE[dataType]);
584 0 : tailDst = static_cast<char *>(const_cast<void *>(dst)) + (headCount * SIZE_TABLE[dataType]);
585 : }
586 : #endif
587 0 : return HCCL_SUCCESS;
588 : }
589 :
590 0 : HcclResult DispatcherPub::DealTbeReduce(const void *src1, const void *src2, u64 count,
591 : const HcclDataType datatype, HcclReduceOp redOp, Stream& stream, const void *dst)
592 : {
593 : #ifndef HCCD
594 0 : HcclResult ret = HCCL_SUCCESS;
595 0 : void *tailSrc1 = nullptr;
596 0 : void *tailSrc2 = nullptr;
597 0 : void *tailDst = nullptr;
598 0 : u64 headCount = 0;
599 0 : u64 tailCount = 0;
600 0 : TbeReduceParam param;
601 0 : std::vector<void *> overflowAddrs;
602 0 : overflowAddrs.push_back(overflowAddr_);
603 0 : param.dataType = datatype;
604 0 : param.redOp = redOp;
605 0 : CHK_RET(JudgeIsTail(src1, src2, dst, count, datatype, headCount, tailCount, tailSrc1, tailSrc2, tailDst));
606 0 : if (headCount != 0) {
607 0 : param.src1 = const_cast<void*>(src1);
608 0 : param.src2 = const_cast<void*>(src2);
609 0 : param.dst = const_cast<void*>(dst);
610 0 : param.count = headCount;
611 : // 对满足32字节整倍数的数据进行reduce
612 0 : ret = HcclTbeReduce(¶m, stream.ptr(), overflowAddrs.data(),
613 0 : overflowAddrs.size(), deviceLogicId_);
614 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
615 : HCCL_ERROR("[DispatcherPub][ReduceAsync]errNo[0x%016llx] tbe vector Reduce fail,return[%d]. "\
616 : "para: src1[%p] src2[%p] count_reduce[%llu] datatype[%s] op[%s] stream[%p] dst_reduce[%p].",
617 : HCCL_ERROR_CODE(ret), ret, src1, src2, count, GetDataTypeEnumStr(datatype).c_str(),
618 : GetReduceOpEnumStr(redOp).c_str(), stream.ptr(), dst), ret);
619 : }
620 : // 对不满足32字节整倍数的剩余数据进行reduce
621 0 : if (tailCount != 0) {
622 0 : void *devMem1 = nullptr;
623 0 : void *devMem2 = nullptr;
624 0 : u32 blockSize = 0;
625 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
626 0 : CHK_RET(DevMemMalloc(stream.ptr(), devMem1, devMem2));
627 0 : CHK_RET(hrtMemAsyncCopy(devMem1, blockSize, tailSrc1, tailCount * SIZE_TABLE[datatype],
628 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
629 0 : CHK_RET(hrtMemAsyncCopy(devMem2, blockSize, tailSrc2, tailCount * SIZE_TABLE[datatype],
630 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
631 0 : param.src1 = devMem1;
632 0 : param.src2 = devMem2;
633 0 : param.dst = devMem2;
634 0 : param.count = tailCount;
635 0 : ret = HcclTbeReduce(¶m, stream.ptr(), overflowAddrs.data(),
636 0 : overflowAddrs.size(), deviceLogicId_);
637 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
638 : HCCL_ERROR("[DispatcherPub][ReduceAsync]errNo[0x%016llx] tbe vector Reduce fail,return[%d]. "\
639 : "para: src1[%p] src2[%p] count_reduce[%llu] datatype[%s] op[%s] stream[%p] dst_reduce[%p].",
640 : HCCL_ERROR_CODE(ret), ret, src1, src2, count, GetDataTypeEnumStr(datatype).c_str(),
641 : GetReduceOpEnumStr(redOp).c_str(), stream.ptr(), dst), ret);
642 0 : CHK_RET(hrtMemAsyncCopy(tailDst, tailCount * SIZE_TABLE[datatype], devMem2, tailCount * SIZE_TABLE[datatype],
643 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
644 : }
645 : #endif
646 0 : return HCCL_SUCCESS;
647 0 : }
648 :
649 0 : HcclResult DispatcherPub::TbeReduceAsync(const void *src1, const void *src2, u64 count,
650 : const HcclDataType datatype, HcclReduceOp redOp, Stream& stream, const void *dst)
651 : {
652 0 : HCCL_DEBUG("Enter--para: src1[%p], src2[%p], count[%llu], datatype[%s], red_op[%s], dst[%p].",
653 : src1, src2, count, GetDataTypeEnumStr(datatype).c_str(), GetReduceOpEnumStr(redOp).c_str(), dst);
654 0 : uint64_t beginTime = GetMsprofSysCycleTime();
655 :
656 0 : if (count == 0) {
657 0 : HCCL_WARNING("count is 0, return success.");
658 0 : return HCCL_SUCCESS;
659 : }
660 : #ifndef HCCD
661 0 : CHK_RET(DealTbeReduce(src1, src2, count, datatype, redOp, stream, dst));
662 : #else
663 : HCCL_ERROR("[DispatcherPub][ReduceAsync] does not support this interface.");
664 : return HCCL_E_PARA;
665 : #endif
666 : // 调用回调来保存task信息
667 0 : if (callback_ != nullptr) {
668 0 : hccl::TaskParaReduce para(src1, dst, count, redOp, datatype, hccl::LinkType::LINK_ONCHIP);
669 0 : hccl::TaskPara taskPara(TaskType::TASK_REDUCE_TBE, para);
670 0 : taskPara.stream = stream.ptr();
671 0 : taskPara.beginTime = beginTime;
672 0 : taskPara.isMainStream = stream.IsMainStream();
673 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
674 0 : }
675 :
676 0 : return HCCL_SUCCESS;
677 : }
678 :
679 92 : HcclResult DispatcherPub::SetGlobalWorkSpace(std::vector<void *> &globalWorkSpaceAddr)
680 : {
681 : #ifndef HCCD
682 : DevType devType;
683 92 : CHK_RET(hrtGetDeviceType(devType));
684 92 : if (devType != DevType::DEV_TYPE_910 && devType != DevType::DEV_TYPE_310P3) {
685 4 : return HCCL_SUCCESS;
686 : }
687 :
688 88 : void *overflowAddr = nullptr;
689 88 : CHK_RET(hrtCtxGetOverflowAddr(&overflowAddr));
690 88 : globalWorkSpaceAddr.push_back(overflowAddr);
691 88 : if (globalWorkSpaceAddr.size()!=0) {
692 : // 第0位代表溢出检测
693 88 : overflowAddr_ = globalWorkSpaceAddr[static_cast<u32>(GlobalWorkSpaceType::OVERFLOW_DETECT_MODE)];
694 : }
695 : #else
696 : HCCL_ERROR("[DispatcherPub][SetGlobalWorkSpace] does not support this interface.");
697 : return HCCL_E_PARA;
698 : #endif
699 88 : return HCCL_SUCCESS;
700 : }
701 :
702 9 : HcclResult DispatcherPub::InlineReduceAsync(const void *src, u64 count, const HcclDataType datatype,
703 : HcclReduceOp redOp, Stream& stream, void *dst, u32 remoteUserRank, hccl::LinkType inLinkType)
704 : {
705 9 : if (count == 0) {
706 0 : HCCL_WARNING("count is 0, return success.");
707 0 : return HCCL_SUCCESS;
708 : }
709 : /* 注意：profiling数据任务时间仍提供切分前整个任务时间 */
710 9 : uint64_t beginTime = GetMsprofSysCycleTime();
711 :
712 9 : CHK_PTR_NULL(stream.ptr());
713 :
714 9 : aclDataType runtimeDataType = ACL_DT_UNDEFINED;
715 9 : aclrtReduceKind rtReduceOp = ACL_RT_MEMCPY_SDMA_AUTOMATIC_EQUAL;
716 : try {
717 9 : runtimeDataType = HCCL_RT_DATA_TYPE_MAP.at(datatype);
718 9 : rtReduceOp = HCCL_RT_REDUCE_OP_MAP.at(redOp);
719 0 : } catch (...) {
720 0 : HCCL_ERROR("[DispatcherPub][ReduceAsync]data type[%s] or reduceOp[%s] is not support",
721 : GetDataTypeEnumStr(datatype).c_str(), GetReduceOpEnumStr(redOp).c_str());
722 0 : return HCCL_E_PARA;
723 0 : }
724 :
725 : DevType deviceType;
726 9 : CHK_RET(hrtGetDeviceType(deviceType));
727 :
728 9 : uint64_t spiltLoop = 0;
729 9 : uint64_t addr_offset = 0;
730 9 : uint64_t contSplit = 0;
731 9 : uint64_t countSize = count * SIZE_TABLE[datatype];
732 9 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
733 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ?
734 0 : (countSize / HCCL_SDMA_MAX_COUNT_4GB) : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
735 0 : HCCL_INFO("InlineReduceAsync SDMA task countSize is bigger than 4GB and do segmentation splitloop[%llu]",
736 : spiltLoop);
737 : }
738 18 : for (uint64_t index = 0 ; index <= spiltLoop; index++) {
739 9 : addr_offset = index * HCCL_SDMA_MAX_COUNT_4GB;
740 9 : contSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
741 9 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void*>(src)) + addr_offset);
742 9 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addr_offset);
743 :
744 9 : CHK_RET(hrtReduceAsync(dstSplit, contSplit, srcSplit, contSplit, rtReduceOp, runtimeDataType, stream.ptr()));
745 :
746 : // 调用回调来保存 task 信息
747 9 : if (callback_ != nullptr) {
748 9 : hccl::TaskParaReduce para(srcSplit, dstSplit, contSplit, redOp, datatype, inLinkType, remoteUserRank);
749 9 : hccl::TaskPara taskPara(TaskType::TASK_REDUCE_INLINE, para);
750 9 : taskPara.stream = stream.ptr();
751 9 : taskPara.beginTime = beginTime;
752 9 : taskPara.isMainStream = stream.IsMainStream();
753 9 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
754 9 : }
755 :
756 9 : u32 taskID = 0;
757 9 : u32 streamID = 0;
758 9 : hrtGetTaskIdAndStreamID(taskID, streamID);
759 9 : PLF_CONFIG_INFO(PLF_TASK,
760 : "%s para: dst[%p] src[%p] count[%llu] rtReduceOp[%d] runtimeDataType[%d] taskID[%u] streamID[%u] "\
761 : "remoteUserRank[%u] inLinkType[%d]", __func__, dstSplit, srcSplit, contSplit / SIZE_TABLE[datatype],
762 : redOp, runtimeDataType, taskID, streamID, remoteUserRank, inLinkType);
763 : }
764 :
765 9 : return HCCL_SUCCESS;
766 : }
767 :
768 0 : HcclResult DispatcherPub::ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
769 : HcclReduceOp redOp, Stream& stream, HcclReduceType reduceType)
770 : {
771 0 : return (reduceType == HcclReduceType::HCCL_INLINE_REDUCE) ?
772 0 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst) :
773 0 : TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
774 : }
775 :
776 0 : HcclResult DispatcherPub::SignalRecord(hccl::DeviceMem &dst, hccl::DeviceMem &src, hccl::Stream &stream,
777 : u32 remoteUserRank, hccl::LinkType inLinkType, u32 notifyId)
778 : {
779 0 : HCCL_ERROR("does not support this interface.");
780 0 : return HCCL_E_NOT_SUPPORT;
781 : }
782 :
783 0 : HcclResult DispatcherPub::RdmaRecord(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
784 : RdmaType rdmaType, u32 userRank, u64 offset, u32 notifyId)
785 : {
786 0 : HCCL_ERROR("does not support this interface.");
787 0 : return HCCL_E_NOT_SUPPORT;
788 : }
789 :
790 0 : HcclResult DispatcherPub::GetCallbackResult()
791 : {
792 0 : return g_callBackResult;
793 : }
794 :
795 0 : void HostNicTcpCallBackProfiling(RaSocketParams *params, std::chrono::microseconds duration)
796 : {
797 : hccl::TaskParaHost para(params->taskInfo.streamId, params->taskInfo.taskId, params->len, duration,
798 0 : params->taskInfo.tag);
799 0 : hccl::TaskPara taskPara(TaskType::TASK_HOST, para);
800 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
801 0 : params->callback(params->callBackUserPtr, (void *)&taskPara, sizeof(struct TaskPara));
802 0 : }
803 :
804 0 : void HostNicCallbackSendWr(void *fnData)
805 : {
806 0 : RaSendWrParams *params = static_cast<RaSendWrParams *>(fnData);
807 0 : unsigned int completeNum = 0;
808 0 : HcclUs startut = TIME_NOW();
809 0 : HcclResult ret = HrtRaSendWrlistExt(params->qpHandle, ¶ms->wr, ¶ms->opRsp,
810 : 1, &completeNum);
811 0 : HcclUs endtut = TIME_NOW();
812 0 : std::chrono::microseconds duration = DURATION_US(endtut - startut);
813 0 : if (ret != HCCL_SUCCESS) {
814 0 : HCCL_ERROR("[Send][Wr]host nic hrtRaSendWrlist failed");
815 0 : g_callBackResult = ret;
816 : }
817 :
818 : hccl::TaskParaHost para(params->taskInfo.streamId, params->taskInfo.taskId,
819 0 : params->wr.memList.len, duration, params->taskInfo.tag);
820 0 : hccl::TaskPara taskPara(TaskType::TASK_HOST, para);
821 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
822 0 : params->callback(params->callBackUserPtr, (void *)&taskPara, sizeof(struct TaskPara));
823 :
824 : // 单算子场景内存需要及时释放
825 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
826 0 : DispatcherPub *tmpDispatcherPtr = static_cast<DispatcherPub *>(params->dispatcherPtr);
827 0 : ret = tmpDispatcherPtr->DelHostNICRdmaTask(params->taskInfo.streamId, params->taskInfo.taskId);
828 0 : if (ret != HCCL_SUCCESS) {
829 0 : HCCL_ERROR("[Send][Wr]Del Host NIC Task failed");
830 0 : g_callBackResult = ret;
831 : }
832 : }
833 0 : }
834 :
835 : // 一次callback，多次收发
836 0 : void HostNicCallbackTcpSend(void *fnData)
837 : {
838 0 : RaSocketParams *params = static_cast<RaSocketParams *>(fnData);
839 0 : u64 bufferSize = params->socketBufferLen;
840 0 : u64 sendCount = params->len / bufferSize + (params->len % bufferSize != 0); // 要发送buffer的次数
841 0 : u64 totalSentSize = 0; // 已发送大小
842 0 : HcclResult ret = HCCL_SUCCESS;
843 0 : ret = hrtSetDevice(params->deviceLogicId);
844 0 : if (ret != HCCL_SUCCESS) {
845 0 : HCCL_ERROR("[Socket][Send] set deviceId[%d] failed", params->deviceLogicId);
846 0 : g_callBackResult = ret;
847 0 : return;
848 : }
849 0 : HcclUs startut = TIME_NOW();
850 0 : for (u64 i = 0; i < sendCount; ++i) {
851 0 : u64 curSendSize = bufferSize;
852 0 : if (i == sendCount - 1 && totalSentSize + bufferSize > params->len) {
853 0 : curSendSize = params->len - totalSentSize;
854 : }
855 0 : ret = hrtMemSyncCopy(params->socketBufferPtr, curSendSize,
856 0 : static_cast<void *>(reinterpret_cast<char *>(params->ptr) + totalSentSize), curSendSize,
857 0 : (params->nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) ?
858 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST :
859 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE);
860 0 : if (ret != HCCL_SUCCESS) {
861 0 : HCCL_ERROR("[Socket][Send]host nic hrtRaSocketBlockSend memcpy failed, tcp nicDeploy[%d]",
862 : params->nicDeploy);
863 0 : g_callBackResult = ret;
864 : }
865 0 : ret = hrtRaSocketBlockSend(params->socketFdHandle, params->socketBufferPtr, curSendSize);
866 0 : if (ret != HCCL_SUCCESS) {
867 0 : HCCL_ERROR("[Socket][Send]host nic hrtRaSocketBlockSend send failed");
868 0 : g_callBackResult = ret;
869 : }
870 0 : totalSentSize += curSendSize;
871 : }
872 0 : ret = hrtResetDevice(params->deviceLogicId);
873 0 : if (ret != HCCL_SUCCESS) {
874 0 : HCCL_ERROR("[Socket][Send] reset deviceId[%d] failed", params->deviceLogicId);
875 0 : g_callBackResult = ret;
876 : }
877 0 : HostNicTcpCallBackProfiling(params, DURATION_US(TIME_NOW() - startut));
878 : }
879 :
880 0 : void HostNicCallbackTcpRecv(void *fnData)
881 : {
882 0 : RaSocketParams *params = static_cast<RaSocketParams *>(fnData);
883 0 : u64 bufferSize = params->socketBufferLen;
884 0 : u64 recvCount = params->len / bufferSize + (params->len % bufferSize != 0); // 要接收buffer的次数
885 0 : u64 totalRecvSize = 0; // 已接收大小
886 0 : HcclResult ret = hrtSetDevice(params->deviceLogicId);
887 0 : if (ret != HCCL_SUCCESS) {
888 0 : HCCL_ERROR("[Socket][Recv] set deviceId[%d] failed", params->deviceLogicId);
889 0 : g_callBackResult = ret;
890 : }
891 0 : HcclUs startut = TIME_NOW();
892 0 : for (u64 i = 0; i < recvCount; ++i) {
893 0 : u64 curRecvSize = bufferSize;
894 0 : if (i == recvCount - 1 && totalRecvSize + bufferSize > params->len) {
895 0 : curRecvSize = params->len - totalRecvSize;
896 : }
897 0 : ret = hrtRaSocketBlockRecv(params->socketFdHandle, params->socketBufferPtr, curRecvSize);
898 0 : if (ret != HCCL_SUCCESS) {
899 0 : HCCL_ERROR("[Socket][Recv]host nic hrtRaSocketBlockRecv recv failed");
900 0 : g_callBackResult = ret;
901 : }
902 0 : ret = hrtMemSyncCopy(static_cast<void *>(reinterpret_cast<char *>(params->ptr) + totalRecvSize),
903 0 : curRecvSize, params->socketBufferPtr, curRecvSize,
904 0 : (params->nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) ?
905 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST :
906 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE);
907 0 : if (ret != HCCL_SUCCESS) {
908 0 : HCCL_ERROR("[Socket][Recv]host nic hrtRaSocketBlockRecv memcpy failed, tcp nicDeploy[%d]",
909 : params->nicDeploy);
910 0 : g_callBackResult = ret;
911 : }
912 0 : totalRecvSize += curRecvSize;
913 : }
914 0 : ret = hrtResetDevice(params->deviceLogicId);
915 0 : if (ret != HCCL_SUCCESS) {
916 0 : HCCL_ERROR("[Socket][Reset] reset deviceId[%d] failed", params->deviceLogicId);
917 0 : g_callBackResult = ret;
918 : }
919 0 : HostNicTcpCallBackProfiling(params, DURATION_US(TIME_NOW() - startut));
920 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
921 0 : DispatcherPub *tmpDispatcherPtr = static_cast<DispatcherPub *>(params->dispatcherPtr);
922 0 : ret = tmpDispatcherPtr->DelHostNICTcpRecvTask(params->taskInfo.streamId, params->taskInfo.taskId);
923 0 : if (ret != HCCL_SUCCESS) {
924 0 : HCCL_ERROR("[Socket][Send]Del Host NIC Task failed");
925 0 : g_callBackResult = ret;
926 : }
927 : } // 单算子场景内存需要及时释放
928 0 : }
929 :
930 0 : void WaitHostNicTcpSendDone(void *dispatcher)
931 : {
932 0 : static_cast<DispatcherPub *>(dispatcher)->WaitHostNicTcpSendTaskDone();
933 0 : }
934 :
935 0 : void StartHostNicTcpSendThread(void *fnData)
936 : {
937 0 : RaSocketParams *params = static_cast<RaSocketParams *>(fnData);
938 0 : DispatcherPub *tmpDispatcherPtr = static_cast<DispatcherPub *>(params->dispatcherPtr);
939 0 : HcclResult ret = tmpDispatcherPtr->SetHostNicTcpSendThreadPara(fnData);
940 : // 单算子场景内存需要及时释放
941 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
942 0 : tmpDispatcherPtr = static_cast<DispatcherPub *>(params->dispatcherPtr);
943 0 : ret = tmpDispatcherPtr->DelHostNICTcpSendTask(params->taskInfo.streamId, params->taskInfo.taskId);
944 0 : if (ret != HCCL_SUCCESS) {
945 0 : HCCL_ERROR("[Socket][Send]Del Host NIC Task failed");
946 0 : g_callBackResult = ret;
947 : }
948 : }
949 0 : }
950 :
951 0 : HcclResult DispatcherPub::HostNicRdmaSend(QpHandle qpHandle, SendWrlistDataExt &wr, SendWrRsp &opRsp,
952 : hccl::Stream &stream, u32 userRank, u64 offset)
953 : {
954 0 : uint64_t beginTime = GetMsprofSysCycleTime();
955 0 : CHK_PTR_NULL(qpHandle);
956 0 : CHK_PTR_NULL(stream.ptr());
957 : (void)opRsp;
958 :
959 0 : if (wr.memList.len == 0) {
960 : // zero byte message 不需要进行通信
961 0 : return HCCL_SUCCESS;
962 : }
963 :
964 0 : u64 notifyID = userRank;
965 0 : notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
966 0 : u32 taskID = 0;
967 0 : u32 streamID = 0;
968 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
969 :
970 0 : std::unique_ptr<RaSendWrParams> params = nullptr;
971 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
972 0 : params.reset(new (std::nothrow) RaSendWrParams(qpHandle, wr, static_cast<void *>(this),
973 0 : streamID, taskID, notifyID, workflowMode, callback_, callBackUserPtr_));
974 0 : CHK_PTR_NULL(params);
975 :
976 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
977 0 : hostNicRdmaParamsVec_[streamID].push(move(params));
978 0 : lock.unlock();
979 :
980 0 : CHK_RET(hrtCallbackLaunch(HostNicCallbackSendWr, hostNicRdmaParamsVec_[streamID].back().get(), stream.ptr(), true));
981 :
982 0 : RdmaType rdmaType = (offset == 0xFFFFFFFFFFFFFFFF) ? RdmaType::RDMA_SEND_PAYLOAD : RdmaType::RDMA_SEND_NOTIFY;
983 :
984 : // 调用回调来保存task信息
985 0 : if (callback_ != nullptr) {
986 0 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.memList.addr)),
987 0 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
988 0 : wr.memList.len, notifyID, hccl::LinkType::LINK_ROCE, rdmaType);
989 0 : hccl::TaskPara taskPara;
990 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
991 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
992 0 : }
993 :
994 0 : return HCCL_SUCCESS;
995 0 : }
996 0 : HcclResult DispatcherPub::HostNicTcpSend(FdHandle socketFdHandle, const void *socketBufferPtr, u64 socketBufferLen,
997 : const void *src, u64 len, hccl::Stream &stream, const NICDeployment nicDeploy)
998 : {
999 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1000 0 : CHK_PTR_NULL(socketFdHandle);
1001 0 : CHK_PTR_NULL(stream.ptr());
1002 0 : u32 taskID = 0;
1003 0 : u32 streamID = 0;
1004 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
1005 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
1006 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1007 0 : params.reset(new (std::nothrow) RaSocketParams(socketFdHandle, socketBufferPtr, socketBufferLen, src, len,
1008 : static_cast<void *>(this), streamID, taskID, workflowMode, deviceLogicId_, nicDeploy,
1009 0 : callback_, callBackUserPtr_));
1010 0 : std::unique_lock<std::mutex> taskLock(hostNicMutex_);
1011 0 : hostNicTcpSendParamsVec_[streamID].push(move(params));
1012 0 : taskLock.unlock();
1013 :
1014 : // 下发callback task
1015 0 : CHK_RET(hrtCallbackLaunch(StartHostNicTcpSendThread, hostNicTcpSendParamsVec_[streamID].back().get(), stream.ptr(),
1016 : true));
1017 :
1018 : // 回调保存信息供profiling记录
1019 0 : if (callback_ != nullptr) {
1020 : hccl::TaskParaDMA para(src, socketBufferPtr, len, INVALID_U64, hccl::LinkType::LINK_ROCE,
1021 0 : RdmaType::RDMA_TYPE_RESERVED);
1022 0 : hccl::TaskPara taskPara;
1023 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
1024 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1025 0 : }
1026 0 : return HCCL_SUCCESS;
1027 0 : }
1028 0 : HcclResult DispatcherPub::HostNicTcpRecv(FdHandle socketFdHandle, const void *socketBufferPtr, u64 socketBufferLen,
1029 : const void *src, u64 len, hccl::Stream &stream, const NICDeployment nicDeploy)
1030 : {
1031 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1032 0 : CHK_PTR_NULL(socketFdHandle);
1033 0 : CHK_PTR_NULL(stream.ptr());
1034 :
1035 0 : u32 taskID = 0;
1036 0 : u32 streamID = 0;
1037 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
1038 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
1039 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1040 0 : params.reset(new (std::nothrow) RaSocketParams(socketFdHandle, socketBufferPtr, socketBufferLen, src, len,
1041 : static_cast<void *>(this), streamID, taskID, workflowMode, deviceLogicId_, nicDeploy,
1042 0 : callback_, callBackUserPtr_));
1043 0 : CHK_SMART_PTR_NULL(params);
1044 :
1045 0 : std::unique_lock<std::mutex> taskLock(hostNicMutex_);
1046 0 : hostNicTcpRecvParamsVec_[streamID].push(move(params));
1047 0 : taskLock.unlock();
1048 :
1049 : // 下发callback task
1050 0 : CHK_RET(
1051 : hrtCallbackLaunch(HostNicCallbackTcpRecv, hostNicTcpRecvParamsVec_[streamID].back().get(), stream.ptr(), true));
1052 :
1053 : // 回调保存信息供profiling记录
1054 0 : if (callback_ != nullptr) {
1055 : hccl::TaskParaDMA para(src, socketBufferPtr, len, INVALID_U64, hccl::LinkType::LINK_ROCE,
1056 0 : RdmaType::RDMA_TYPE_RESERVED);
1057 0 : hccl::TaskPara taskPara;
1058 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
1059 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1060 0 : }
1061 :
1062 0 : return HCCL_SUCCESS;
1063 0 : }
1064 :
1065 0 : HcclResult DispatcherPub::SetHostNicTcpSendThreadPara(void *fnData)
1066 : {
1067 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1068 0 : auto tmpRaSocketParamsPtr = new (std::nothrow) RaSocketParams(*(static_cast<RaSocketParams *>(fnData)));
1069 0 : CHK_PTR_NULL(tmpRaSocketParamsPtr);
1070 0 : params.reset(tmpRaSocketParamsPtr);
1071 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1072 0 : if (hostNicTcpSendThreadParam_ == nullptr) {
1073 0 : hostNicTcpSendThreadParam_ = move(params);
1074 : } else {
1075 0 : HCCL_ERROR("last send task is not finished! stream[%u] task[%u]",
1076 : hostNicTcpSendThreadParam_->taskInfo.streamId, hostNicTcpSendThreadParam_->taskInfo.taskId);
1077 : }
1078 0 : return HCCL_SUCCESS;
1079 0 : }
1080 :
1081 0 : void DispatcherPub::HostNicTcpSendThreadTask()
1082 : {
1083 : // 给当前线程添加名字
1084 0 : SetThreadName("Hccl_HostNicTcp");
1085 :
1086 0 : while (hostNicTcpSendThreadState_) {
1087 0 : if (hostNicTcpSendThreadParam_ == nullptr) {
1088 0 : SaluSleep(TCP_SEND_THREAD_SLEEP_TWO_HUNDRED_MICROSECOND);
1089 : } else {
1090 0 : void *fnData = hostNicTcpSendThreadParam_.get();
1091 0 : if (fnData != nullptr) {
1092 0 : HostNicCallbackTcpSend(fnData);
1093 : }
1094 0 : hostNicTcpSendThreadParam_ = nullptr;
1095 : }
1096 : }
1097 0 : }
1098 :
1099 0 : HcclResult DispatcherPub::HostNicTcpWaitSendCompletion(hccl::Stream &stream)
1100 : {
1101 0 : CHK_RET(hrtCallbackLaunch(WaitHostNicTcpSendDone, this, stream.ptr(), true));
1102 0 : return HCCL_SUCCESS;
1103 : }
1104 0 : HcclResult DispatcherPub::DelHostNICRdmaTask(u32 streamID, u32 taskID)
1105 : {
1106 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1107 0 : CHK_PRT_RET((hostNicRdmaParamsVec_.find(streamID) == hostNicRdmaParamsVec_.end()),
1108 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicRdmaParamsVec_",
1109 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID), HCCL_E_PARA);
1110 :
1111 0 : CHK_PRT_RET((hostNicRdmaParamsVec_[streamID].size() == 0), HCCL_ERROR("[DispatcherPub][DelHostNICTask]"
1112 : "errNo[0x%016llx] streamID[%u] task num is 0", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID), HCCL_E_INTERNAL);
1113 :
1114 0 : CHK_PRT_RET((hostNicRdmaParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1115 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]" \
1116 : " is not equal to the front taskID[%u]", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID,
1117 : taskID, hostNicRdmaParamsVec_[streamID].front()->taskInfo.taskId), HCCL_E_INTERNAL);
1118 :
1119 0 : hostNicRdmaParamsVec_[streamID].pop();
1120 0 : return HCCL_SUCCESS;
1121 0 : }
1122 0 : HcclResult DispatcherPub::DelHostNICTcpSendTask(u32 streamID, u32 taskID)
1123 : {
1124 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1125 0 : CHK_PRT_RET((hostNicTcpSendParamsVec_.find(streamID) == hostNicTcpSendParamsVec_.end()),
1126 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicTcpSendParamsVec_",
1127 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID), HCCL_E_PARA);
1128 :
1129 0 : CHK_PRT_RET((hostNicTcpSendParamsVec_[streamID].size() == 0), HCCL_ERROR("[DispatcherPub][DelHostNICTask]"
1130 : "errNo[0x%016llx] streamID[%u] task num is 0", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID), HCCL_E_INTERNAL);
1131 :
1132 0 : CHK_PRT_RET((hostNicTcpSendParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1133 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]" \
1134 : " is not equal to the front taskID[%u]", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID,
1135 : taskID, hostNicTcpSendParamsVec_[streamID].front()->taskInfo.taskId), HCCL_E_INTERNAL);
1136 :
1137 0 : hostNicTcpSendParamsVec_[streamID].pop();
1138 0 : return HCCL_SUCCESS;
1139 0 : }
1140 0 : HcclResult DispatcherPub::DelHostNICTcpRecvTask(u32 streamID, u32 taskID)
1141 : {
1142 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1143 0 : CHK_PRT_RET((hostNicTcpRecvParamsVec_.find(streamID) == hostNicTcpRecvParamsVec_.end()),
1144 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicTcpRecvParamsVec_",
1145 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID), HCCL_E_PARA);
1146 :
1147 0 : CHK_PRT_RET((hostNicTcpRecvParamsVec_[streamID].size() == 0), HCCL_ERROR("[DispatcherPub][DelHostNICTask]"
1148 : "errNo[0x%016llx] streamID[%u] task num is 0", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID), HCCL_E_INTERNAL);
1149 :
1150 0 : CHK_PRT_RET((hostNicTcpRecvParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1151 : HCCL_ERROR("[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]" \
1152 : " is not equal to the front taskID[%u]", HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID,
1153 : taskID, hostNicTcpRecvParamsVec_[streamID].front()->taskInfo.taskId), HCCL_E_INTERNAL);
1154 :
1155 0 : hostNicTcpRecvParamsVec_[streamID].pop();
1156 0 : return HCCL_SUCCESS;
1157 0 : }
1158 : // 下沉模式下内部接口
1159 0 : HcclResult DispatcherPub::RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, HcclRtStream stream,
1160 : RdmaType rdmaType, u64 notifyID, bool isMainStream)
1161 : {
1162 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1163 0 : if ((qpn == INVALID_UINT) && (wqeIndex == INVALID_UINT)) {
1164 : // zero byte message 不需要下发rdma send task
1165 0 : return HCCL_SUCCESS;
1166 : }
1167 :
1168 0 : CHK_RET(hrtRDMASend(qpn, wqeIndex, stream));
1169 :
1170 : // 调用回调来保存task信息
1171 0 : if (callback_ != nullptr) {
1172 0 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
1173 0 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
1174 0 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, rdmaType);
1175 0 : hccl::TaskPara taskPara;
1176 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, stream, beginTime, isMainStream);
1177 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1178 0 : }
1179 :
1180 0 : u32 taskID = 0;
1181 0 : u32 streamID = 0;
1182 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
1183 0 : PLF_CONFIG_INFO(PLF_TASK,
1184 : "%s para: qpn[%u] wqeIndex[%u] rdmaType[%d] notifyId[0x%016llx] taskID[%u] streamID[%u]",
1185 : __func__, qpn, wqeIndex, rdmaType, notifyID, taskID, streamID);
1186 0 : return HCCL_SUCCESS;
1187 : }
1188 :
1189 : // 下沉模式下对外接口, 用于发送notify 信息
1190 0 : HcclResult DispatcherPub::RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, hccl::Stream &stream,
1191 : u32 userRank, u64 offset)
1192 : {
1193 0 : u64 NotifyID =
1194 0 : (static_cast<u64>(userRank) << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
1195 0 : return RdmaSend(qpn, wqeIndex, wr, stream.ptr(), RdmaType::RDMA_SEND_NOTIFY, NotifyID, stream.IsMainStream());
1196 : }
1197 :
1198 : // 下沉模式下对外接口, 用于发送payload 信息
1199 0 : HcclResult DispatcherPub::RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr &wr, hccl::Stream &stream,
1200 : u32 userRank)
1201 : {
1202 0 : u64 NotifyID =
1203 0 : (static_cast<u64>(userRank) << 32) | (0x00000000FFFFFFFF); // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
1204 0 : return RdmaSend(qpn, wqeIndex, wr, stream.ptr(), RdmaType::RDMA_SEND_PAYLOAD, NotifyID, stream.IsMainStream());
1205 : }
1206 :
1207 : // opbase 模式下内部接口
1208 0 : HcclResult DispatcherPub::RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, HcclRtStream stream,
1209 : RdmaType rdmaType, u64 notifyID, u64 offset, bool isMainStream)
1210 : {
1211 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1212 0 : if ((dbindex == INVALID_UINT) && (dbinfo == INVALID_U64)) {
1213 : // zero byte message 不需要下发rdma send task
1214 0 : return HCCL_SUCCESS;
1215 : }
1216 :
1217 0 : CHK_RET(hrtRDMADBSend(dbindex, dbinfo, stream));
1218 :
1219 : // 调用回调来保存task信息
1220 0 : if (callback_ != nullptr) {
1221 0 : notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
1222 0 : hccl::TaskParaDMA para(reinterpret_cast<void *>(static_cast<uintptr_t>(wr.bufList[0].addr)),
1223 0 : reinterpret_cast<void *>(static_cast<uintptr_t>(wr.dstAddr)),
1224 0 : wr.bufList[0].len, notifyID, hccl::LinkType::LINK_ROCE, rdmaType);
1225 0 : hccl::TaskPara taskPara;
1226 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, stream, beginTime, isMainStream);
1227 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1228 0 : }
1229 :
1230 0 : u32 taskID = 0;
1231 0 : u32 streamID = 0;
1232 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
1233 0 : PLF_CONFIG_INFO(PLF_TASK,
1234 : "%s para: dbindex[%u] dbinfo[%llu] rdmaType[%d] notifyId[0x%016llx] offset[%llu] taskID[%u] streamID[%u]",
1235 : __func__, dbindex, dbinfo, rdmaType, notifyID, offset, taskID, streamID);
1236 0 : return HCCL_SUCCESS;
1237 : }
1238 :
1239 : // opbase 模式下对外接口，用于发送notify 信息
1240 0 : HcclResult DispatcherPub::RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
1241 : u32 userRank, u64 offset, bool isCapture)
1242 : {
1243 0 : CHK_RET(RdmaSend(dbindex, dbinfo, wr, stream.ptr(), RdmaType::RDMA_SEND_NOTIFY, userRank, offset,
1244 : stream.IsMainStream()));
1245 :
1246 0 : return HCCL_SUCCESS;
1247 : }
1248 :
1249 : // opbase 模式下对外接口，用于发送payload 信息
1250 0 : HcclResult DispatcherPub::RdmaSend(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
1251 : u32 remoteUserRank, bool isCapture)
1252 : {
1253 0 : u64 offset = 0;
1254 0 : CHK_RET(RdmaSend(dbindex, dbinfo, wr, stream.ptr(), RdmaType::RDMA_SEND_PAYLOAD, remoteUserRank, offset,
1255 : stream.IsMainStream()));
1256 :
1257 0 : return HCCL_SUCCESS;
1258 : }
1259 :
1260 0 : HcclResult DispatcherPub::RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream &stream, RdmaTaskInfo &taskInfo)
1261 : {
1262 0 : HCCL_ERROR("does not support this interface."); // host暂不使用此接口，待后续归一
1263 0 : return HCCL_E_NOT_SUPPORT;
1264 : }
1265 :
1266 0 : HcclResult DispatcherPub::SignalRecord(HcclRtNotify signal, Stream &stream, u32 userRank, u64 offset, s32 stage,
1267 : bool inchip, u64 signalAddr, u32 notifyId)
1268 : {
1269 0 : CHK_RET(SignalRecord(signal, stream.ptr(), userRank, offset, stage, stream.IsMainStream()));
1270 :
1271 0 : return HCCL_SUCCESS;
1272 : }
1273 :
1274 1 : HcclResult DispatcherPub::SignalWait(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank, s32 stage,
1275 : bool inchip, u32 notifyId, u32 timeOut)
1276 : {
1277 : (void) notifyId;
1278 1 : CHK_RET(SignalWait(signal, stream.ptr(), userRank, remoteUserRank, stage, timeOut, stream.IsMainStream()));
1279 :
1280 1 : return HCCL_SUCCESS;
1281 : }
1282 :
1283 0 : HcclResult DispatcherPub::AddRetryPreamble(Stream &stream)
1284 : {
1285 0 : return HCCL_SUCCESS;
1286 : }
1287 :
1288 0 : HcclResult DispatcherPub::WaitValue(hccl::Stream &stream, u64 waitAddr, u64 valueAddr, bool reset)
1289 : {
1290 0 : return HCCL_SUCCESS;
1291 : }
1292 0 : HcclResult DispatcherPub::WriteValue(hccl::Stream &stream, u64 writeAddr, u64 valueAddr)
1293 : {
1294 0 : return HCCL_SUCCESS;
1295 : }
1296 :
1297 0 : bool DispatcherPub::IsProfSubscribeAdditionInfo() {
1298 0 : u64 profConfig = GetProfConfig();
1299 0 : if (((profConfig & PROF_TASK_TIME_L1_MASK) != 0) || ((profConfig & PROF_HCCL_TRACE_MASK) != 0) || isForce_) {
1300 0 : return true;
1301 : }
1302 0 : return false;
1303 : }
1304 :
1305 0 : HcclResult DispatcherPub::StreamSync(Stream &stream)
1306 : {
1307 0 : HCCL_INFO("StreamSync is not supported");
1308 0 : return HCCL_SUCCESS;
1309 : }
1310 :
1311 11 : void DispatcherPub::SetHcclQos(u32 hcclQos)
1312 : {
1313 11 : HCCL_INFO("[DispatcherPub] [SetHcclQos] hcclQos = %u", hcclQos);
1314 : // 按区间映射HCCL QOS到SDMA QOS
1315 11 : if (hcclQos <= HCCL_QOS_LEVEL_1_LIMIT) {
1316 11 : hcclQos_ = SDMA_QOS_LOW;
1317 0 : } else if (hcclQos <= HCCL_QOS_LEVEL_2_LIMIT) {
1318 0 : hcclQos_ = SDMA_QOS_MIDDLE;
1319 0 : } else if (hcclQos <= HCCL_QOS_LEVEL_3_LIMIT) {
1320 0 : hcclQos_ = SDMA_QOS_HIGH;
1321 : } else {
1322 : // 超出有效范围，使用默认值（包括hcclQos < HCCL_QOS_MIN的异常情况）
1323 0 : hcclQos_ = SDMA_QOS_DEFAULT;
1324 : }
1325 11 : }
1326 :
1327 11 : void DispatcherPub::SetMpamid(u32 mPamid)
1328 : {
1329 11 : HCCL_INFO("[DispatcherPub] [SetMpamid] mPamid[%u]", mPamid);
1330 11 : mPamid_ = mPamid;
1331 11 : return;
1332 : }
1333 : #endif
|
