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