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 2109 : DispatcherPub::~DispatcherPub()
52 : {
53 1603 : HcclResult ret = HCCL_SUCCESS;
54 : #ifndef HCCD
55 1603 : std::map<int32_t, void*>::iterator devMemIter;
56 1604 : std::unique_lock<std::mutex> lock(devMemMutex_);
57 1604 : 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 1601 : if (deviceLogicId_ != HOST_DEVICE_ID) {
67 1151 : 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 554 : fftsPubInfo_ = nullptr;
77 : }
78 : #endif
79 :
80 : // 清空task信息
81 1601 : if (hostNicTcpSendThread_ != nullptr) {
82 0 : WaitHostNicTcpSendThreadComplete();
83 : }
84 1603 : ClearHostNicRdmaParamsVec();
85 1600 : ClearHostNicTcpSendParamsVec();
86 1604 : ClearHostNicTcpRecvParamsVec();
87 :
88 1601 : 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 2107 : }
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 1603 : hostNicRdmaParamsVec_.clear();
147 1600 : }
148 :
149 1601 : void DispatcherPub::ClearHostNicTcpSendParamsVec()
150 : {
151 1601 : 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 1604 : }
162 :
163 1604 : void DispatcherPub::ClearHostNicTcpRecvParamsVec()
164 : {
165 1604 : 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 1601 : hostNicTcpRecvParamsVec_.clear();
175 1601 : }
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 1005 : 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 1005 : 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(void* stream, void*& devMem1, void*& devMem2)
544 : {
545 : #ifndef HCCD
546 : int32_t streamId;
547 : u32 blockSize;
548 0 : CHK_RET(hrtGetStreamId(stream, streamId));
549 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
550 :
551 0 : std::unique_lock<std::mutex> lock(devMemMutex_);
552 0 : if (devMemMap_.find(streamId) == devMemMap_.end()) {
553 0 : u32 devMemSize = blockSize + blockSize;
554 0 : CHK_RET(hrtMalloc(&devMem1, devMemSize));
555 0 : CHK_PTR_NULL(devMem1);
556 0 : CHK_RET(hrtMemSet(devMem1, devMemSize, devMemSize));
557 0 : devMem2 = static_cast<char*>(devMem1) + blockSize;
558 0 : devMemMap_[streamId] = devMem1;
559 : } else {
560 0 : devMem1 = devMemMap_[streamId];
561 0 : devMem2 = static_cast<char*>(devMem1) + blockSize;
562 : }
563 : #endif
564 0 : return HCCL_SUCCESS;
565 0 : }
566 :
567 0 : HcclResult DispatcherPub::JudgeIsTail(
568 : const void* src1, const void* src2, const void* dst, u64 count, const HcclDataType dataType, u64& headCount,
569 : u64& tailCount, void*& tailSrc1, void*& tailSrc2, void*& tailDst)
570 : {
571 : #ifndef HCCD
572 0 : CHK_PRT_RET(dataType >= HCCL_DATA_TYPE_RESERVED, HCCL_ERROR("dataType is failed."), HCCL_E_PARA);
573 0 : u32 blockSize = 0;
574 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
575 : // 获取总的数据量
576 0 : u64 dataSize = SIZE_TABLE[dataType] * count; // 计算总的字节数
577 0 : headCount = dataSize / blockSize * blockSize / SIZE_TABLE[dataType]; // 计算出32字节整倍数的数据数量
578 0 : tailCount = count - headCount;
579 :
580 0 : if (tailCount != 0) {
581 0 : tailSrc1 = static_cast<char*>(const_cast<void*>(src1)) + (headCount * SIZE_TABLE[dataType]);
582 0 : tailSrc2 = static_cast<char*>(const_cast<void*>(src2)) + (headCount * SIZE_TABLE[dataType]);
583 0 : tailDst = static_cast<char*>(const_cast<void*>(dst)) + (headCount * SIZE_TABLE[dataType]);
584 : }
585 : #endif
586 0 : return HCCL_SUCCESS;
587 : }
588 :
589 0 : HcclResult DispatcherPub::DealTbeReduce(
590 : const void* src1, const void* src2, u64 count, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
591 : 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(), overflowAddrs.size(), deviceLogicId_);
613 0 : CHK_PRT_RET(
614 : ret != HCCL_SUCCESS,
615 : HCCL_ERROR(
616 : "[DispatcherPub][ReduceAsync]errNo[0x%016llx] tbe vector Reduce fail,return[%d]. "
617 : "para: src1[%p] src2[%p] count_reduce[%llu] datatype[%s] op[%s] stream[%p] dst_reduce[%p].",
618 : HCCL_ERROR_CODE(ret), ret, src1, src2, count, GetDataTypeEnumStr(datatype).c_str(),
619 : GetReduceOpEnumStr(redOp).c_str(), stream.ptr(), dst),
620 : ret);
621 : }
622 : // 对不满足32字节整倍数的剩余数据进行reduce
623 0 : if (tailCount != 0) {
624 0 : void* devMem1 = nullptr;
625 0 : void* devMem2 = nullptr;
626 0 : u32 blockSize = 0;
627 0 : CHK_RET(HcclGetVectorBlockSize(&blockSize, deviceLogicId_));
628 0 : CHK_RET(DevMemMalloc(stream.ptr(), devMem1, devMem2));
629 0 : CHK_RET(hrtMemAsyncCopy(
630 : devMem1, blockSize, tailSrc1, tailCount * SIZE_TABLE[datatype],
631 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
632 0 : CHK_RET(hrtMemAsyncCopy(
633 : devMem2, blockSize, tailSrc2, tailCount * SIZE_TABLE[datatype],
634 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
635 0 : param.src1 = devMem1;
636 0 : param.src2 = devMem2;
637 0 : param.dst = devMem2;
638 0 : param.count = tailCount;
639 0 : ret = HcclTbeReduce(¶m, stream.ptr(), overflowAddrs.data(), overflowAddrs.size(), deviceLogicId_);
640 0 : CHK_PRT_RET(
641 : ret != HCCL_SUCCESS,
642 : HCCL_ERROR(
643 : "[DispatcherPub][ReduceAsync]errNo[0x%016llx] tbe vector Reduce fail,return[%d]. "
644 : "para: src1[%p] src2[%p] count_reduce[%llu] datatype[%s] op[%s] stream[%p] dst_reduce[%p].",
645 : HCCL_ERROR_CODE(ret), ret, src1, src2, count, GetDataTypeEnumStr(datatype).c_str(),
646 : GetReduceOpEnumStr(redOp).c_str(), stream.ptr(), dst),
647 : ret);
648 0 : CHK_RET(hrtMemAsyncCopy(
649 : tailDst, tailCount * SIZE_TABLE[datatype], devMem2, tailCount * SIZE_TABLE[datatype],
650 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
651 : }
652 : #endif
653 0 : return HCCL_SUCCESS;
654 0 : }
655 :
656 0 : HcclResult DispatcherPub::TbeReduceAsync(
657 : const void* src1, const void* src2, u64 count, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
658 : const void* dst)
659 : {
660 0 : HCCL_DEBUG(
661 : "Enter--para: src1[%p], src2[%p], count[%llu], datatype[%s], red_op[%s], dst[%p].", src1, src2, count,
662 : GetDataTypeEnumStr(datatype).c_str(), GetReduceOpEnumStr(redOp).c_str(), dst);
663 0 : uint64_t beginTime = GetMsprofSysCycleTime();
664 :
665 0 : if (count == 0) {
666 0 : HCCL_WARNING("count is 0, return success.");
667 0 : return HCCL_SUCCESS;
668 : }
669 : #ifndef HCCD
670 0 : CHK_RET(DealTbeReduce(src1, src2, count, datatype, redOp, stream, dst));
671 : #else
672 : HCCL_ERROR("[DispatcherPub][ReduceAsync] does not support this interface.");
673 : return HCCL_E_PARA;
674 : #endif
675 : // 调用回调来保存task信息
676 0 : if (callback_ != nullptr) {
677 0 : hccl::TaskParaReduce para(src1, dst, count, redOp, datatype, hccl::LinkType::LINK_ONCHIP);
678 0 : hccl::TaskPara taskPara(TaskType::TASK_REDUCE_TBE, para);
679 0 : taskPara.stream = stream.ptr();
680 0 : taskPara.beginTime = beginTime;
681 0 : taskPara.isMainStream = stream.IsMainStream();
682 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
683 0 : }
684 :
685 0 : return HCCL_SUCCESS;
686 : }
687 :
688 92 : HcclResult DispatcherPub::SetGlobalWorkSpace(std::vector<void*>& globalWorkSpaceAddr)
689 : {
690 : #ifndef HCCD
691 : DevType devType;
692 92 : CHK_RET(hrtGetDeviceType(devType));
693 92 : if (devType != DevType::DEV_TYPE_910 && devType != DevType::DEV_TYPE_310P3) {
694 4 : return HCCL_SUCCESS;
695 : }
696 :
697 88 : void* overflowAddr = nullptr;
698 88 : CHK_RET(hrtCtxGetOverflowAddr(&overflowAddr));
699 88 : globalWorkSpaceAddr.push_back(overflowAddr);
700 88 : if (globalWorkSpaceAddr.size() != 0) {
701 : // 第0位代表溢出检测
702 88 : overflowAddr_ = globalWorkSpaceAddr[static_cast<u32>(GlobalWorkSpaceType::OVERFLOW_DETECT_MODE)];
703 : }
704 : #else
705 : HCCL_ERROR("[DispatcherPub][SetGlobalWorkSpace] does not support this interface.");
706 : return HCCL_E_PARA;
707 : #endif
708 88 : return HCCL_SUCCESS;
709 : }
710 :
711 9 : HcclResult DispatcherPub::InlineReduceAsync(
712 : const void* src, u64 count, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream, void* dst,
713 : u32 remoteUserRank, hccl::LinkType inLinkType)
714 : {
715 9 : if (count == 0) {
716 0 : HCCL_WARNING("count is 0, return success.");
717 0 : return HCCL_SUCCESS;
718 : }
719 : /* 注意：profiling数据任务时间仍提供切分前整个任务时间 */
720 9 : uint64_t beginTime = GetMsprofSysCycleTime();
721 :
722 9 : CHK_PTR_NULL(stream.ptr());
723 :
724 9 : aclDataType runtimeDataType = ACL_DT_UNDEFINED;
725 9 : aclrtReduceKind rtReduceOp = ACL_RT_MEMCPY_SDMA_AUTOMATIC_EQUAL;
726 : try {
727 9 : runtimeDataType = HCCL_RT_DATA_TYPE_MAP.at(datatype);
728 9 : rtReduceOp = HCCL_RT_REDUCE_OP_MAP.at(redOp);
729 0 : } catch (...) {
730 0 : HCCL_ERROR(
731 : "[DispatcherPub][ReduceAsync]data type[%s] or reduceOp[%s] is not support",
732 : GetDataTypeEnumStr(datatype).c_str(), GetReduceOpEnumStr(redOp).c_str());
733 0 : return HCCL_E_PARA;
734 0 : }
735 :
736 : DevType deviceType;
737 9 : CHK_RET(hrtGetDeviceType(deviceType));
738 :
739 9 : uint64_t spiltLoop = 0;
740 9 : uint64_t addr_offset = 0;
741 9 : uint64_t contSplit = 0;
742 9 : uint64_t countSize = count * SIZE_TABLE[datatype];
743 9 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
744 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ? (countSize / HCCL_SDMA_MAX_COUNT_4GB) :
745 0 : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
746 0 : HCCL_INFO(
747 : "InlineReduceAsync SDMA task countSize is bigger than 4GB and do segmentation splitloop[%llu]", spiltLoop);
748 : }
749 18 : for (uint64_t index = 0; index <= spiltLoop; index++) {
750 9 : addr_offset = index * HCCL_SDMA_MAX_COUNT_4GB;
751 9 : contSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
752 9 : void* srcSplit = static_cast<void*>(static_cast<char*>(const_cast<void*>(src)) + addr_offset);
753 9 : void* dstSplit = static_cast<void*>(static_cast<char*>(dst) + addr_offset);
754 :
755 9 : CHK_RET(hrtReduceAsync(dstSplit, contSplit, srcSplit, contSplit, rtReduceOp, runtimeDataType, stream.ptr()));
756 :
757 : // 调用回调来保存 task 信息
758 9 : if (callback_ != nullptr) {
759 9 : hccl::TaskParaReduce para(srcSplit, dstSplit, contSplit, redOp, datatype, inLinkType, remoteUserRank);
760 9 : hccl::TaskPara taskPara(TaskType::TASK_REDUCE_INLINE, para);
761 9 : taskPara.stream = stream.ptr();
762 9 : taskPara.beginTime = beginTime;
763 9 : taskPara.isMainStream = stream.IsMainStream();
764 9 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
765 9 : }
766 :
767 9 : u32 taskID = 0;
768 9 : u32 streamID = 0;
769 9 : hrtGetTaskIdAndStreamID(taskID, streamID);
770 9 : PLF_CONFIG_INFO(
771 : PLF_TASK,
772 : "%s para: dst[%p] src[%p] count[%llu] rtReduceOp[%d] runtimeDataType[%d] taskID[%u] streamID[%u] "
773 : "remoteUserRank[%u] inLinkType[%d]",
774 : __func__, dstSplit, srcSplit, contSplit / SIZE_TABLE[datatype], redOp, runtimeDataType, taskID, streamID,
775 : remoteUserRank, inLinkType);
776 : }
777 :
778 9 : return HCCL_SUCCESS;
779 : }
780 :
781 0 : HcclResult DispatcherPub::ReduceAsync(
782 : const void* src, void* dst, u64 dataCount, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
783 : HcclReduceType reduceType)
784 : {
785 0 : return (reduceType == HcclReduceType::HCCL_INLINE_REDUCE) ?
786 0 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst) :
787 0 : TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
788 : }
789 :
790 0 : HcclResult DispatcherPub::SignalRecord(
791 : hccl::DeviceMem& dst, hccl::DeviceMem& src, hccl::Stream& stream, u32 remoteUserRank, hccl::LinkType inLinkType,
792 : u32 notifyId)
793 : {
794 0 : HCCL_ERROR("does not support this interface.");
795 0 : return HCCL_E_NOT_SUPPORT;
796 : }
797 :
798 0 : HcclResult DispatcherPub::RdmaRecord(
799 : u32 dbindex, u64 dbinfo, const struct SendWr& wr, hccl::Stream& stream, RdmaType rdmaType, u32 userRank, u64 offset,
800 : u32 notifyId)
801 : {
802 0 : HCCL_ERROR("does not support this interface.");
803 0 : return HCCL_E_NOT_SUPPORT;
804 : }
805 :
806 0 : HcclResult DispatcherPub::GetCallbackResult() { return g_callBackResult; }
807 :
808 0 : void HostNicTcpCallBackProfiling(RaSocketParams* params, std::chrono::microseconds duration)
809 : {
810 : hccl::TaskParaHost para(
811 0 : params->taskInfo.streamId, params->taskInfo.taskId, params->len, duration, params->taskInfo.tag);
812 0 : hccl::TaskPara taskPara(TaskType::TASK_HOST, para);
813 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
814 0 : params->callback(params->callBackUserPtr, (void*)&taskPara, sizeof(struct TaskPara));
815 0 : }
816 :
817 0 : void HostNicCallbackSendWr(void* fnData)
818 : {
819 0 : RaSendWrParams* params = static_cast<RaSendWrParams*>(fnData);
820 0 : unsigned int completeNum = 0;
821 0 : HcclUs startut = TIME_NOW();
822 0 : HcclResult ret = HrtRaSendWrlistExt(params->qpHandle, ¶ms->wr, ¶ms->opRsp, 1, &completeNum);
823 0 : HcclUs endtut = TIME_NOW();
824 0 : std::chrono::microseconds duration = DURATION_US(endtut - startut);
825 0 : if (ret != HCCL_SUCCESS) {
826 0 : HCCL_ERROR("[Send][Wr]host nic hrtRaSendWrlist failed");
827 0 : g_callBackResult = ret;
828 : }
829 :
830 : hccl::TaskParaHost para(
831 0 : params->taskInfo.streamId, params->taskInfo.taskId, params->wr.memList.len, duration, params->taskInfo.tag);
832 0 : hccl::TaskPara taskPara(TaskType::TASK_HOST, para);
833 0 : taskPara.profilerType = ProfilerType::TASK_PROFILING;
834 0 : params->callback(params->callBackUserPtr, (void*)&taskPara, sizeof(struct TaskPara));
835 :
836 : // 单算子场景内存需要及时释放
837 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
838 0 : DispatcherPub* tmpDispatcherPtr = static_cast<DispatcherPub*>(params->dispatcherPtr);
839 0 : ret = tmpDispatcherPtr->DelHostNICRdmaTask(params->taskInfo.streamId, params->taskInfo.taskId);
840 0 : if (ret != HCCL_SUCCESS) {
841 0 : HCCL_ERROR("[Send][Wr]Del Host NIC Task failed");
842 0 : g_callBackResult = ret;
843 : }
844 : }
845 0 : }
846 :
847 : // 一次callback，多次收发
848 0 : void HostNicCallbackTcpSend(void* fnData)
849 : {
850 0 : RaSocketParams* params = static_cast<RaSocketParams*>(fnData);
851 0 : u64 bufferSize = params->socketBufferLen;
852 0 : u64 sendCount = params->len / bufferSize + (params->len % bufferSize != 0); // 要发送buffer的次数
853 0 : u64 totalSentSize = 0; // 已发送大小
854 0 : HcclResult ret = HCCL_SUCCESS;
855 0 : ret = hrtSetDevice(params->deviceLogicId);
856 0 : if (ret != HCCL_SUCCESS) {
857 0 : HCCL_ERROR("[Socket][Send] set deviceId[%d] failed", params->deviceLogicId);
858 0 : g_callBackResult = ret;
859 0 : return;
860 : }
861 0 : HcclUs startut = TIME_NOW();
862 0 : for (u64 i = 0; i < sendCount; ++i) {
863 0 : u64 curSendSize = bufferSize;
864 0 : if (i == sendCount - 1 && totalSentSize + bufferSize > params->len) {
865 0 : curSendSize = params->len - totalSentSize;
866 : }
867 0 : ret = hrtMemSyncCopy(
868 : params->socketBufferPtr, curSendSize,
869 0 : static_cast<void*>(reinterpret_cast<char*>(params->ptr) + totalSentSize), curSendSize,
870 0 : (params->nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) ?
871 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST :
872 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE);
873 0 : if (ret != HCCL_SUCCESS) {
874 0 : HCCL_ERROR(
875 : "[Socket][Send]host nic hrtRaSocketBlockSend memcpy failed, tcp nicDeploy[%d]", params->nicDeploy);
876 0 : g_callBackResult = ret;
877 : }
878 0 : ret = hrtRaSocketBlockSend(params->socketFdHandle, params->socketBufferPtr, curSendSize);
879 0 : if (ret != HCCL_SUCCESS) {
880 0 : HCCL_ERROR("[Socket][Send]host nic hrtRaSocketBlockSend send failed");
881 0 : g_callBackResult = ret;
882 : }
883 0 : totalSentSize += curSendSize;
884 : }
885 0 : ret = hrtResetDevice(params->deviceLogicId);
886 0 : if (ret != HCCL_SUCCESS) {
887 0 : HCCL_ERROR("[Socket][Send] reset deviceId[%d] failed", params->deviceLogicId);
888 0 : g_callBackResult = ret;
889 : }
890 0 : HostNicTcpCallBackProfiling(params, DURATION_US(TIME_NOW() - startut));
891 : }
892 :
893 0 : void HostNicCallbackTcpRecv(void* fnData)
894 : {
895 0 : RaSocketParams* params = static_cast<RaSocketParams*>(fnData);
896 0 : u64 bufferSize = params->socketBufferLen;
897 0 : u64 recvCount = params->len / bufferSize + (params->len % bufferSize != 0); // 要接收buffer的次数
898 0 : u64 totalRecvSize = 0; // 已接收大小
899 0 : HcclResult ret = hrtSetDevice(params->deviceLogicId);
900 0 : if (ret != HCCL_SUCCESS) {
901 0 : HCCL_ERROR("[Socket][Recv] set deviceId[%d] failed", params->deviceLogicId);
902 0 : g_callBackResult = ret;
903 : }
904 0 : HcclUs startut = TIME_NOW();
905 0 : for (u64 i = 0; i < recvCount; ++i) {
906 0 : u64 curRecvSize = bufferSize;
907 0 : if (i == recvCount - 1 && totalRecvSize + bufferSize > params->len) {
908 0 : curRecvSize = params->len - totalRecvSize;
909 : }
910 0 : ret = hrtRaSocketBlockRecv(params->socketFdHandle, params->socketBufferPtr, curRecvSize);
911 0 : if (ret != HCCL_SUCCESS) {
912 0 : HCCL_ERROR("[Socket][Recv]host nic hrtRaSocketBlockRecv recv failed");
913 0 : g_callBackResult = ret;
914 : }
915 0 : ret = hrtMemSyncCopy(
916 0 : static_cast<void*>(reinterpret_cast<char*>(params->ptr) + totalRecvSize), curRecvSize,
917 0 : params->socketBufferPtr, curRecvSize,
918 0 : (params->nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) ?
919 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST :
920 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE);
921 0 : if (ret != HCCL_SUCCESS) {
922 0 : HCCL_ERROR(
923 : "[Socket][Recv]host nic hrtRaSocketBlockRecv memcpy failed, tcp nicDeploy[%d]", params->nicDeploy);
924 0 : g_callBackResult = ret;
925 : }
926 0 : totalRecvSize += curRecvSize;
927 : }
928 0 : ret = hrtResetDevice(params->deviceLogicId);
929 0 : if (ret != HCCL_SUCCESS) {
930 0 : HCCL_ERROR("[Socket][Reset] reset deviceId[%d] failed", params->deviceLogicId);
931 0 : g_callBackResult = ret;
932 : }
933 0 : HostNicTcpCallBackProfiling(params, DURATION_US(TIME_NOW() - startut));
934 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
935 0 : DispatcherPub* tmpDispatcherPtr = static_cast<DispatcherPub*>(params->dispatcherPtr);
936 0 : ret = tmpDispatcherPtr->DelHostNICTcpRecvTask(params->taskInfo.streamId, params->taskInfo.taskId);
937 0 : if (ret != HCCL_SUCCESS) {
938 0 : HCCL_ERROR("[Socket][Send]Del Host NIC Task failed");
939 0 : g_callBackResult = ret;
940 : }
941 : } // 单算子场景内存需要及时释放
942 0 : }
943 :
944 0 : void WaitHostNicTcpSendDone(void* dispatcher) { static_cast<DispatcherPub*>(dispatcher)->WaitHostNicTcpSendTaskDone(); }
945 :
946 0 : void StartHostNicTcpSendThread(void* fnData)
947 : {
948 0 : RaSocketParams* params = static_cast<RaSocketParams*>(fnData);
949 0 : DispatcherPub* tmpDispatcherPtr = static_cast<DispatcherPub*>(params->dispatcherPtr);
950 0 : HcclResult ret = tmpDispatcherPtr->SetHostNicTcpSendThreadPara(fnData);
951 : // 单算子场景内存需要及时释放
952 0 : if (params->workMode == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
953 0 : tmpDispatcherPtr = static_cast<DispatcherPub*>(params->dispatcherPtr);
954 0 : ret = tmpDispatcherPtr->DelHostNICTcpSendTask(params->taskInfo.streamId, params->taskInfo.taskId);
955 0 : if (ret != HCCL_SUCCESS) {
956 0 : HCCL_ERROR("[Socket][Send]Del Host NIC Task failed");
957 0 : g_callBackResult = ret;
958 : }
959 : }
960 0 : }
961 :
962 0 : HcclResult DispatcherPub::HostNicRdmaSend(
963 : QpHandle qpHandle, SendWrlistDataExt& wr, SendWrRsp& opRsp, hccl::Stream& stream, u32 userRank, u64 offset)
964 : {
965 0 : uint64_t beginTime = GetMsprofSysCycleTime();
966 0 : CHK_PTR_NULL(qpHandle);
967 0 : CHK_PTR_NULL(stream.ptr());
968 : (void)opRsp;
969 :
970 0 : if (wr.memList.len == 0) {
971 : // zero byte message 不需要进行通信
972 0 : return HCCL_SUCCESS;
973 : }
974 :
975 0 : u64 notifyID = userRank;
976 0 : notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
977 0 : u32 taskID = 0;
978 0 : u32 streamID = 0;
979 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
980 :
981 0 : std::unique_ptr<RaSendWrParams> params = nullptr;
982 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
983 0 : params.reset(new (std::nothrow) RaSendWrParams(
984 0 : qpHandle, wr, static_cast<void*>(this), streamID, taskID, notifyID, workflowMode, callback_, callBackUserPtr_));
985 0 : CHK_PTR_NULL(params);
986 :
987 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
988 0 : hostNicRdmaParamsVec_[streamID].push(move(params));
989 0 : lock.unlock();
990 :
991 0 : CHK_RET(hrtCallbackLaunch(HostNicCallbackSendWr, hostNicRdmaParamsVec_[streamID].back().get(), stream.ptr(), true));
992 :
993 0 : RdmaType rdmaType = (offset == 0xFFFFFFFFFFFFFFFF) ? RdmaType::RDMA_SEND_PAYLOAD : RdmaType::RDMA_SEND_NOTIFY;
994 :
995 : // 调用回调来保存task信息
996 0 : if (callback_ != nullptr) {
997 : hccl::TaskParaDMA para(
998 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.memList.addr)),
999 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.dstAddr)), wr.memList.len, notifyID,
1000 0 : hccl::LinkType::LINK_ROCE, rdmaType);
1001 0 : hccl::TaskPara taskPara;
1002 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
1003 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1004 0 : }
1005 :
1006 0 : return HCCL_SUCCESS;
1007 0 : }
1008 0 : HcclResult DispatcherPub::HostNicTcpSend(
1009 : FdHandle socketFdHandle, const void* socketBufferPtr, u64 socketBufferLen, const void* src, u64 len,
1010 : hccl::Stream& stream, const NICDeployment nicDeploy)
1011 : {
1012 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1013 0 : CHK_PTR_NULL(socketFdHandle);
1014 0 : CHK_PTR_NULL(stream.ptr());
1015 0 : u32 taskID = 0;
1016 0 : u32 streamID = 0;
1017 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
1018 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
1019 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1020 0 : params.reset(new (std::nothrow) RaSocketParams(
1021 : socketFdHandle, socketBufferPtr, socketBufferLen, src, len, static_cast<void*>(this), streamID, taskID,
1022 0 : workflowMode, deviceLogicId_, nicDeploy, callback_, callBackUserPtr_));
1023 0 : std::unique_lock<std::mutex> taskLock(hostNicMutex_);
1024 0 : hostNicTcpSendParamsVec_[streamID].push(move(params));
1025 0 : taskLock.unlock();
1026 :
1027 : // 下发callback task
1028 0 : CHK_RET(hrtCallbackLaunch(
1029 : StartHostNicTcpSendThread, hostNicTcpSendParamsVec_[streamID].back().get(), stream.ptr(), true));
1030 :
1031 : // 回调保存信息供profiling记录
1032 0 : if (callback_ != nullptr) {
1033 : hccl::TaskParaDMA para(
1034 0 : src, socketBufferPtr, len, INVALID_U64, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_TYPE_RESERVED);
1035 0 : hccl::TaskPara taskPara;
1036 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
1037 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1038 0 : }
1039 0 : return HCCL_SUCCESS;
1040 0 : }
1041 0 : HcclResult DispatcherPub::HostNicTcpRecv(
1042 : FdHandle socketFdHandle, const void* socketBufferPtr, u64 socketBufferLen, const void* src, u64 len,
1043 : hccl::Stream& stream, const NICDeployment nicDeploy)
1044 : {
1045 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1046 0 : CHK_PTR_NULL(socketFdHandle);
1047 0 : CHK_PTR_NULL(stream.ptr());
1048 :
1049 0 : u32 taskID = 0;
1050 0 : u32 streamID = 0;
1051 0 : CHK_RET(hrtGetTaskIdAndStreamID(taskID, streamID));
1052 0 : HcclWorkflowMode workflowMode = GetWorkflowMode();
1053 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1054 0 : params.reset(new (std::nothrow) RaSocketParams(
1055 : socketFdHandle, socketBufferPtr, socketBufferLen, src, len, static_cast<void*>(this), streamID, taskID,
1056 0 : workflowMode, deviceLogicId_, nicDeploy, callback_, callBackUserPtr_));
1057 0 : CHK_SMART_PTR_NULL(params);
1058 :
1059 0 : std::unique_lock<std::mutex> taskLock(hostNicMutex_);
1060 0 : hostNicTcpRecvParamsVec_[streamID].push(move(params));
1061 0 : taskLock.unlock();
1062 :
1063 : // 下发callback task
1064 0 : CHK_RET(
1065 : hrtCallbackLaunch(HostNicCallbackTcpRecv, hostNicTcpRecvParamsVec_[streamID].back().get(), stream.ptr(), true));
1066 :
1067 : // 回调保存信息供profiling记录
1068 0 : if (callback_ != nullptr) {
1069 : hccl::TaskParaDMA para(
1070 0 : src, socketBufferPtr, len, INVALID_U64, hccl::LinkType::LINK_ROCE, RdmaType::RDMA_TYPE_RESERVED);
1071 0 : hccl::TaskPara taskPara;
1072 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, ProfilerType::TASK_EXCEPTION, stream, beginTime, false);
1073 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1074 0 : }
1075 :
1076 0 : return HCCL_SUCCESS;
1077 0 : }
1078 :
1079 0 : HcclResult DispatcherPub::SetHostNicTcpSendThreadPara(void* fnData)
1080 : {
1081 0 : std::unique_ptr<RaSocketParams> params = nullptr;
1082 0 : auto tmpRaSocketParamsPtr = new (std::nothrow) RaSocketParams(*(static_cast<RaSocketParams*>(fnData)));
1083 0 : CHK_PTR_NULL(tmpRaSocketParamsPtr);
1084 0 : params.reset(tmpRaSocketParamsPtr);
1085 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1086 0 : if (hostNicTcpSendThreadParam_ == nullptr) {
1087 0 : hostNicTcpSendThreadParam_ = move(params);
1088 : } else {
1089 0 : HCCL_ERROR(
1090 : "last send task is not finished! stream[%u] task[%u]", hostNicTcpSendThreadParam_->taskInfo.streamId,
1091 : hostNicTcpSendThreadParam_->taskInfo.taskId);
1092 : }
1093 0 : return HCCL_SUCCESS;
1094 0 : }
1095 :
1096 0 : void DispatcherPub::HostNicTcpSendThreadTask()
1097 : {
1098 : // 给当前线程添加名字
1099 0 : SetThreadName("Hccl_HostNicTcp");
1100 :
1101 0 : while (hostNicTcpSendThreadState_) {
1102 0 : if (hostNicTcpSendThreadParam_ == nullptr) {
1103 0 : SaluSleep(TCP_SEND_THREAD_SLEEP_TWO_HUNDRED_MICROSECOND);
1104 : } else {
1105 0 : void* fnData = hostNicTcpSendThreadParam_.get();
1106 0 : if (fnData != nullptr) {
1107 0 : HostNicCallbackTcpSend(fnData);
1108 : }
1109 0 : hostNicTcpSendThreadParam_ = nullptr;
1110 : }
1111 : }
1112 0 : }
1113 :
1114 0 : HcclResult DispatcherPub::HostNicTcpWaitSendCompletion(hccl::Stream& stream)
1115 : {
1116 0 : CHK_RET(hrtCallbackLaunch(WaitHostNicTcpSendDone, this, stream.ptr(), true));
1117 0 : return HCCL_SUCCESS;
1118 : }
1119 0 : HcclResult DispatcherPub::DelHostNICRdmaTask(u32 streamID, u32 taskID)
1120 : {
1121 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1122 0 : CHK_PRT_RET(
1123 : (hostNicRdmaParamsVec_.find(streamID) == hostNicRdmaParamsVec_.end()),
1124 : HCCL_ERROR(
1125 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicRdmaParamsVec_",
1126 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID),
1127 : HCCL_E_PARA);
1128 :
1129 0 : CHK_PRT_RET(
1130 : (hostNicRdmaParamsVec_[streamID].size() == 0),
1131 : HCCL_ERROR(
1132 : "[DispatcherPub][DelHostNICTask]"
1133 : "errNo[0x%016llx] streamID[%u] task num is 0",
1134 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID),
1135 : HCCL_E_INTERNAL);
1136 :
1137 0 : CHK_PRT_RET(
1138 : (hostNicRdmaParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1139 : HCCL_ERROR(
1140 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]"
1141 : " is not equal to the front taskID[%u]",
1142 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID, taskID,
1143 : hostNicRdmaParamsVec_[streamID].front()->taskInfo.taskId),
1144 : HCCL_E_INTERNAL);
1145 :
1146 0 : hostNicRdmaParamsVec_[streamID].pop();
1147 0 : return HCCL_SUCCESS;
1148 0 : }
1149 0 : HcclResult DispatcherPub::DelHostNICTcpSendTask(u32 streamID, u32 taskID)
1150 : {
1151 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1152 0 : CHK_PRT_RET(
1153 : (hostNicTcpSendParamsVec_.find(streamID) == hostNicTcpSendParamsVec_.end()),
1154 : HCCL_ERROR(
1155 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicTcpSendParamsVec_",
1156 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID),
1157 : HCCL_E_PARA);
1158 :
1159 0 : CHK_PRT_RET(
1160 : (hostNicTcpSendParamsVec_[streamID].size() == 0),
1161 : HCCL_ERROR(
1162 : "[DispatcherPub][DelHostNICTask]"
1163 : "errNo[0x%016llx] streamID[%u] task num is 0",
1164 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID),
1165 : HCCL_E_INTERNAL);
1166 :
1167 0 : CHK_PRT_RET(
1168 : (hostNicTcpSendParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1169 : HCCL_ERROR(
1170 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]"
1171 : " is not equal to the front taskID[%u]",
1172 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID, taskID,
1173 : hostNicTcpSendParamsVec_[streamID].front()->taskInfo.taskId),
1174 : HCCL_E_INTERNAL);
1175 :
1176 0 : hostNicTcpSendParamsVec_[streamID].pop();
1177 0 : return HCCL_SUCCESS;
1178 0 : }
1179 0 : HcclResult DispatcherPub::DelHostNICTcpRecvTask(u32 streamID, u32 taskID)
1180 : {
1181 0 : std::unique_lock<std::mutex> lock(hostNicMutex_);
1182 0 : CHK_PRT_RET(
1183 : (hostNicTcpRecvParamsVec_.find(streamID) == hostNicTcpRecvParamsVec_.end()),
1184 : HCCL_ERROR(
1185 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] not found in hostNicTcpRecvParamsVec_",
1186 : HCCL_ERROR_CODE(HCCL_E_PARA), streamID),
1187 : HCCL_E_PARA);
1188 :
1189 0 : CHK_PRT_RET(
1190 : (hostNicTcpRecvParamsVec_[streamID].size() == 0),
1191 : HCCL_ERROR(
1192 : "[DispatcherPub][DelHostNICTask]"
1193 : "errNo[0x%016llx] streamID[%u] task num is 0",
1194 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID),
1195 : HCCL_E_INTERNAL);
1196 :
1197 0 : CHK_PRT_RET(
1198 : (hostNicTcpRecvParamsVec_[streamID].front()->taskInfo.taskId != taskID),
1199 : HCCL_ERROR(
1200 : "[DispatcherPub][DelHostNICTask]errNo[0x%016llx] streamID[%u] taskID[%u]"
1201 : " is not equal to the front taskID[%u]",
1202 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), streamID, taskID,
1203 : hostNicTcpRecvParamsVec_[streamID].front()->taskInfo.taskId),
1204 : HCCL_E_INTERNAL);
1205 :
1206 0 : hostNicTcpRecvParamsVec_[streamID].pop();
1207 0 : return HCCL_SUCCESS;
1208 0 : }
1209 : // 下沉模式下内部接口
1210 0 : HcclResult DispatcherPub::RdmaSend(
1211 : u32 qpn, u32 wqeIndex, const struct SendWr& wr, HcclRtStream stream, RdmaType rdmaType, u64 notifyID,
1212 : bool isMainStream)
1213 : {
1214 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1215 0 : if ((qpn == INVALID_UINT) && (wqeIndex == INVALID_UINT)) {
1216 : // zero byte message 不需要下发rdma send task
1217 0 : return HCCL_SUCCESS;
1218 : }
1219 :
1220 0 : CHK_RET(hrtRDMASend(qpn, wqeIndex, stream));
1221 :
1222 : // 调用回调来保存task信息
1223 0 : if (callback_ != nullptr) {
1224 : hccl::TaskParaDMA para(
1225 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.bufList[0].addr)),
1226 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.dstAddr)), wr.bufList[0].len, notifyID,
1227 0 : hccl::LinkType::LINK_ROCE, rdmaType);
1228 0 : hccl::TaskPara taskPara;
1229 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, stream, beginTime, isMainStream);
1230 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1231 0 : }
1232 :
1233 0 : u32 taskID = 0;
1234 0 : u32 streamID = 0;
1235 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
1236 0 : PLF_CONFIG_INFO(
1237 : PLF_TASK, "%s para: qpn[%u] wqeIndex[%u] rdmaType[%d] notifyId[0x%016llx] taskID[%u] streamID[%u]", __func__,
1238 : qpn, wqeIndex, rdmaType, notifyID, taskID, streamID);
1239 0 : return HCCL_SUCCESS;
1240 : }
1241 :
1242 : // 下沉模式下对外接口, 用于发送notify 信息
1243 : HcclResult
1244 0 : DispatcherPub::RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr& wr, hccl::Stream& stream, u32 userRank, u64 offset)
1245 : {
1246 0 : u64 NotifyID
1247 0 : = (static_cast<u64>(userRank) << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
1248 0 : return RdmaSend(qpn, wqeIndex, wr, stream.ptr(), RdmaType::RDMA_SEND_NOTIFY, NotifyID, stream.IsMainStream());
1249 : }
1250 :
1251 : // 下沉模式下对外接口, 用于发送payload 信息
1252 0 : HcclResult DispatcherPub::RdmaSend(u32 qpn, u32 wqeIndex, const struct SendWr& wr, hccl::Stream& stream, u32 userRank)
1253 : {
1254 0 : u64 NotifyID
1255 0 : = (static_cast<u64>(userRank) << 32) | (0x00000000FFFFFFFF); // 0x00000000FFFFFFFF usrrank位于notifyID的高32位
1256 0 : return RdmaSend(qpn, wqeIndex, wr, stream.ptr(), RdmaType::RDMA_SEND_PAYLOAD, NotifyID, stream.IsMainStream());
1257 : }
1258 :
1259 : // opbase 模式下内部接口
1260 0 : HcclResult DispatcherPub::RdmaSend(
1261 : u32 dbindex, u64 dbinfo, const struct SendWr& wr, HcclRtStream stream, RdmaType rdmaType, u64 notifyID, u64 offset,
1262 : bool isMainStream)
1263 : {
1264 0 : uint64_t beginTime = GetMsprofSysCycleTime();
1265 0 : if ((dbindex == INVALID_UINT) && (dbinfo == INVALID_U64)) {
1266 : // zero byte message 不需要下发rdma send task
1267 0 : return HCCL_SUCCESS;
1268 : }
1269 :
1270 0 : CHK_RET(hrtRDMADBSend(dbindex, dbinfo, stream));
1271 :
1272 : // 调用回调来保存task信息
1273 0 : if (callback_ != nullptr) {
1274 0 : notifyID = (notifyID << 32) | (offset & 0x00000000FFFFFFFF); // 0x00000000FFFFFFFF用于取offset的低32位
1275 : hccl::TaskParaDMA para(
1276 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.bufList[0].addr)),
1277 0 : reinterpret_cast<void*>(static_cast<uintptr_t>(wr.dstAddr)), wr.bufList[0].len, notifyID,
1278 0 : hccl::LinkType::LINK_ROCE, rdmaType);
1279 0 : hccl::TaskPara taskPara;
1280 0 : SetupTaskParaDma(taskPara, para, TaskType::TASK_RDMA, stream, beginTime, isMainStream);
1281 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1282 0 : }
1283 :
1284 0 : u32 taskID = 0;
1285 0 : u32 streamID = 0;
1286 0 : hrtGetTaskIdAndStreamID(taskID, streamID);
1287 0 : PLF_CONFIG_INFO(
1288 : PLF_TASK,
1289 : "%s para: dbindex[%u] dbinfo[%llu] rdmaType[%d] notifyId[0x%016llx] offset[%llu] taskID[%u] streamID[%u]",
1290 : __func__, dbindex, dbinfo, rdmaType, notifyID, offset, taskID, streamID);
1291 0 : return HCCL_SUCCESS;
1292 : }
1293 :
1294 : // opbase 模式下对外接口，用于发送notify 信息
1295 0 : HcclResult DispatcherPub::RdmaSend(
1296 : u32 dbindex, u64 dbinfo, const struct SendWr& wr, hccl::Stream& stream, u32 userRank, u64 offset, bool isCapture)
1297 : {
1298 0 : CHK_RET(RdmaSend(
1299 : dbindex, dbinfo, wr, stream.ptr(), RdmaType::RDMA_SEND_NOTIFY, userRank, offset, stream.IsMainStream()));
1300 :
1301 0 : return HCCL_SUCCESS;
1302 : }
1303 :
1304 : // opbase 模式下对外接口，用于发送payload 信息
1305 0 : HcclResult DispatcherPub::RdmaSend(
1306 : u32 dbindex, u64 dbinfo, const struct SendWr& wr, hccl::Stream& stream, u32 remoteUserRank, bool isCapture)
1307 : {
1308 0 : u64 offset = 0;
1309 0 : CHK_RET(RdmaSend(
1310 : dbindex, dbinfo, wr, stream.ptr(), RdmaType::RDMA_SEND_PAYLOAD, remoteUserRank, offset, stream.IsMainStream()));
1311 :
1312 0 : return HCCL_SUCCESS;
1313 : }
1314 :
1315 0 : HcclResult DispatcherPub::RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream& stream, RdmaTaskInfo& taskInfo)
1316 : {
1317 0 : HCCL_ERROR("does not support this interface."); // host暂不使用此接口，待后续归一
1318 0 : return HCCL_E_NOT_SUPPORT;
1319 : }
1320 :
1321 0 : HcclResult DispatcherPub::SignalRecord(
1322 : HcclRtNotify signal, Stream& stream, u32 userRank, u64 offset, s32 stage, bool inchip, u64 signalAddr, u32 notifyId)
1323 : {
1324 0 : CHK_RET(SignalRecord(signal, stream.ptr(), userRank, offset, stage, stream.IsMainStream()));
1325 :
1326 0 : return HCCL_SUCCESS;
1327 : }
1328 :
1329 1 : HcclResult DispatcherPub::SignalWait(
1330 : HcclRtNotify signal, Stream& stream, u32 userRank, u32 remoteUserRank, s32 stage, bool inchip, u32 notifyId,
1331 : u32 timeOut)
1332 : {
1333 : (void)notifyId;
1334 1 : CHK_RET(SignalWait(signal, stream.ptr(), userRank, remoteUserRank, stage, timeOut, stream.IsMainStream()));
1335 :
1336 1 : return HCCL_SUCCESS;
1337 : }
1338 :
1339 0 : HcclResult DispatcherPub::AddRetryPreamble(Stream& stream) { return HCCL_SUCCESS; }
1340 :
1341 0 : HcclResult DispatcherPub::WaitValue(hccl::Stream& stream, u64 waitAddr, u64 valueAddr, bool reset)
1342 : {
1343 0 : return HCCL_SUCCESS;
1344 : }
1345 0 : HcclResult DispatcherPub::WriteValue(hccl::Stream& stream, u64 writeAddr, u64 valueAddr) { return HCCL_SUCCESS; }
1346 :
1347 0 : bool DispatcherPub::IsProfSubscribeAdditionInfo()
1348 : {
1349 0 : u64 profConfig = GetProfConfig();
1350 0 : if (((profConfig & PROF_TASK_TIME_L1_MASK) != 0) || ((profConfig & PROF_HCCL_TRACE_MASK) != 0) || isForce_) {
1351 0 : return true;
1352 : }
1353 0 : return false;
1354 : }
1355 :
1356 0 : HcclResult DispatcherPub::StreamSync(Stream& stream)
1357 : {
1358 0 : HCCL_INFO("StreamSync is not supported");
1359 0 : return HCCL_SUCCESS;
1360 : }
1361 :
1362 11 : void DispatcherPub::SetHcclQos(u32 hcclQos)
1363 : {
1364 11 : HCCL_INFO("[DispatcherPub] [SetHcclQos] hcclQos = %u", hcclQos);
1365 : // 按区间映射HCCL QOS到SDMA QOS
1366 11 : if (hcclQos <= HCCL_QOS_LEVEL_1_LIMIT) {
1367 11 : hcclQos_ = SDMA_QOS_LOW;
1368 0 : } else if (hcclQos <= HCCL_QOS_LEVEL_2_LIMIT) {
1369 0 : hcclQos_ = SDMA_QOS_MIDDLE;
1370 0 : } else if (hcclQos <= HCCL_QOS_LEVEL_3_LIMIT) {
1371 0 : hcclQos_ = SDMA_QOS_HIGH;
1372 : } else {
1373 : // 超出有效范围，使用默认值（包括hcclQos < HCCL_QOS_MIN的异常情况）
1374 0 : hcclQos_ = SDMA_QOS_DEFAULT;
1375 : }
1376 11 : }
1377 :
1378 11 : void DispatcherPub::SetMpamid(u32 mPamid)
1379 : {
1380 11 : HCCL_INFO("[DispatcherPub] [SetMpamid] mPamid[%u]", mPamid);
1381 11 : mPamid_ = mPamid;
1382 11 : return;
1383 : }
1384 : #endif
|
