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 : #include "ub_mem_transport.h"
11 : #include "serializable.h"
12 : #include "exchange_ub_buffer_dto.h"
13 : #include "exchange_ub_conn_dto.h"
14 : #include "local_ub_rma_buffer.h"
15 : #include "dev_capability.h"
16 : #include "dev_buffer.h"
17 : #include "../../common/dlprof_func.h"
18 : #include "user_remote_mem_getter.h"
19 : #include "exception_util.h"
20 : #include "env_config/env_config.h"
21 :
22 : namespace Hccl {
23 : constexpr u32 FINISH_MSG_SIZE = 128;
24 : constexpr char_t FINISH_MSG[FINISH_MSG_SIZE] = "Ub Comm Pipe ready!";
25 : constexpr u32 ONE_MILLISECOND_OF_USLEEP = 1000;
26 :
27 105 : UbMemTransport::UbMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
28 : const Socket &socket, RdmaHandle rdmaHandle1, LocCntNotifyRes &locCntNotifyRes1,
29 105 : bool isRecvFirst)
30 105 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::UB), rdmaHandle(rdmaHandle1),
31 105 : locCntNotifyRes(locCntNotifyRes1), isRecvFirst_(isRecvFirst)
32 : {
33 315 : HCCL_INFO("source: %s", locCntNotifyRes.Describe().c_str());
34 105 : }
35 :
36 8 : UbMemTransport::UbMemTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
37 : const Socket &socket, RdmaHandle rdmaHandle1, LocCntNotifyRes &locCntNotifyRes1,
38 8 : std::function<void(u32 streamId, u32 taskId, const TaskParam &taskParam)> callback)
39 8 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::UB, callback), rdmaHandle(rdmaHandle1),
40 8 : locCntNotifyRes(locCntNotifyRes1)
41 : {
42 24 : HCCL_INFO("source: %s", locCntNotifyRes.Describe().c_str());
43 8 : }
44 :
45 10 : std::string UbMemTransport::Describe() const
46 : {
47 : string msg = StringFormat("UbMemTransport=[commonLocRes=%s, locCntNotifyRes=%s, ubStatus=%s, ",
48 30 : commonLocRes.Describe().c_str(), locCntNotifyRes.Describe().c_str(),
49 40 : ubStatus.Describe().c_str());
50 10 : msg += StringFormat("exchangeDataSize=%u, ", exchangeDataSize);
51 10 : msg += StringFormat("rmtNotifyNum=%zu, rmtCntNotifyVecNum=%zu]", rmtNotifyVec.size(), rmtCntNotifyVec.size());
52 10 : return msg;
53 0 : }
54 :
55 1 : HcclResult UbMemTransport::BuildDrainResource()
56 : {
57 : // notify作为read的落点
58 1 : if (drainNotify_ == nullptr) {
59 1 : bool devUsed = true;
60 1 : EXCEPTION_CATCH(
61 : drainNotify_ = std::make_unique<Hccl::UbLocalNotify>(rdmaHandle, devUsed),
62 : return HCCL_E_PTR
63 : );
64 3 : HCCL_INFO("[UbMemTransport][%s] drain notify created: %s", __func__, drainNotify_->Describe().c_str());
65 : }
66 :
67 : // 常量1内存供远端读取
68 1 : if (drainBuffer_ == nullptr) {
69 1 : u32 notifySize = Hccl::DevCapability::GetInstance().GetNotifySize();
70 :
71 1 : std::shared_ptr<Hccl::DevBuffer> constMem;
72 1 : EXCEPTION_CATCH(constMem = std::make_shared<Hccl::DevBuffer>(notifySize), return HCCL_E_PTR);
73 :
74 1 : Hccl::HrtMemcpy(reinterpret_cast<void *>(constMem->GetAddr()), constMem->GetSize(),
75 : &NORMAL_NOTIFY_VAL, sizeof(NORMAL_NOTIFY_VAL), RT_MEMCPY_HOST_TO_DEVICE);
76 :
77 1 : EXCEPTION_CATCH(
78 : drainBuffer_ = std::make_unique<Hccl::LocalUbRmaBuffer>(constMem, rdmaHandle),
79 : return HCCL_E_PTR
80 : );
81 3 : HCCL_INFO("[UbMemTransport][%s] drain buffer created: addr[0x%llx], size[%zu]",
82 : __func__, static_cast<unsigned long long>(drainBuffer_->GetAddr()), drainBuffer_->GetSize());
83 1 : }
84 :
85 1 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult UbMemTransport::Describe(std::string &dfxMsg)
89 : {
90 0 : HCCL_INFO("UbMemTransport Describe connNum[%u]", connNum);
91 0 : for (u32 i = 0; i < connNum; i++) {
92 0 : CHK_RET(commonLocRes.connVec[i]->Describe(dfxMsg));
93 : }
94 0 : return HCCL_SUCCESS;
95 : }
96 :
97 8 : MemoryBuffer UbMemTransport::GetLocMemBuffer(const RmaBufferSlice &locSlice) const
98 : {
99 8 : return MemoryBuffer(locSlice.addr, locSlice.size, locSlice.buf->GetMemHandle());
100 : }
101 :
102 8 : MemoryBuffer UbMemTransport::GetRmtMemBuffer(const RmtRmaBufferSlice &rmtSlice) const
103 : {
104 8 : return MemoryBuffer(rmtSlice.addr, rmtSlice.size, rmtSlice.buf->GetMemHandle());
105 : }
106 :
107 5 : MemoryBuffer UbMemTransport::GetRmtNotifyMemBuffer(u32 index)
108 : {
109 10 : return MemoryBuffer(rmtNotifyVec[index]->GetAddr(), rmtNotifyVec[index]->GetSize(),
110 15 : rmtNotifyVec[index]->GetMemHandle());
111 : }
112 :
113 4 : MemoryBuffer UbMemTransport::GetRmtCntNotifyMemBuffer(const WithNotifyIn &withNotify)
114 : {
115 4 : auto index = withNotify.index_;
116 8 : return MemoryBuffer(rmtCntNotifyVec[index]->GetAddr(), rmtCntNotifyVec[index]->GetSize(),
117 12 : rmtCntNotifyVec[index]->GetMemHandle());
118 : }
119 :
120 5 : static void SubmitTask(const TaskUbDbSend &ubSend, const Stream &stream)
121 : {
122 15 : HCCL_INFO("SubmitTask UbDbSend ");
123 : HrtUbDbInfo info;
124 5 : info.dbNum = 1;
125 5 : info.wrCqe = 0; // 默认值是0 不会cqe 如果传1，驱动分发，会给hccl cqe，用于维护ci指针。
126 5 : info.info[0].functionId = ubSend.GetFuncId();
127 5 : info.info[0].dieId = ubSend.GetDieId();
128 5 : info.info[0].jettyId = ubSend.GetJettyId();
129 5 : info.info[0].piValue = ubSend.GetPiVal();
130 5 : HrtUbDbSend(info, stream.GetPtr());
131 0 : }
132 :
133 1 : static void SubmitTask(const TaskUbDirectSend &ubDirectSend, const Stream &stream)
134 : {
135 3 : HCCL_INFO("SubmitTask UbDirectSend");
136 1 : if (ubDirectSend.GetDwqeSize() != DWQE_SIZE_64 && ubDirectSend.GetDwqeSize() != DWQE_SIZE_128) {
137 : std::string msg
138 0 : = StringFormat("dwqe size is not valid, cannot submit task, dwqeSize=%u", ubDirectSend.GetDwqeSize());
139 0 : THROW<InternalException>(msg);
140 0 : }
141 : HrtUbWqeInfo info;
142 1 : info.wrCqe = 0;
143 1 : info.functionId = ubDirectSend.GetFuncId();
144 1 : info.dieId = ubDirectSend.GetDieId();
145 1 : info.jettyId = ubDirectSend.GetJettyId();
146 1 : info.wqe = const_cast<u8 *>(ubDirectSend.GetDwqePtr());
147 1 : info.wqePtrLen = ubDirectSend.GetDwqeSize();
148 1 : info.wqeSize = info.wqePtrLen == DWQE_SIZE_64 ? 0 : 1;
149 1 : HrtUbDirectSend(info, stream.GetPtr());
150 0 : }
151 :
152 5 : static void SubmitTask(const TaskWriteValue &taskWriteValue, const Stream &stream)
153 : {
154 15 : HCCL_INFO("begin HrtWriteValue");
155 5 : HrtWriteValue(taskWriteValue.GetDbAddr(), taskWriteValue.GetPiVal(), stream.GetPtr());
156 0 : HCCL_INFO("finished HrtWriteValue");
157 0 : }
158 :
159 3 : template <typename TaskType> std::function<void(const BaseTask &, const Stream &)> GetSubmitUbTaskFunction()
160 : {
161 14 : return [](const BaseTask &task, const Stream &stream) {
162 11 : SubmitTask(static_cast<const TaskType &>(task), stream);
163 3 : };
164 : }
165 :
166 : std::map<TaskType, std::function<void(const BaseTask &, const Stream &)>> g_ubTaskSubmitRuleMap
167 : = {{TaskType::UB_SEND, GetSubmitUbTaskFunction<TaskUbDbSend>()},
168 : {TaskType::UB_DIRECT_SEND, GetSubmitUbTaskFunction<TaskUbDirectSend>()},
169 : {TaskType::WRITE_VALUE, GetSubmitUbTaskFunction<TaskWriteValue>()}};
170 :
171 13 : static void SubmitUbTask(unique_ptr<BaseTask> task, const Stream &stream)
172 : {
173 13 : if (task != nullptr) {
174 11 : g_ubTaskSubmitRuleMap.at(task->GetType())(*task.get(), stream);
175 : }
176 2 : }
177 :
178 5 : void UbMemTransport::SubmitNotify(const MemoryBuffer &rmtNotify, u64 data, const Stream &stream)
179 : {
180 5 : SqeConfig config;
181 10 : SubmitUbTask(commonLocRes.connVec[0]->PrepareInlineWrite(rmtNotify, data, config), stream);
182 0 : }
183 :
184 1 : void UbMemTransport::Post(u32 index, const Stream &stream)
185 : {
186 1 : TaskParam taskParam {};
187 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
188 :
189 1 : SubmitNotify(GetRmtNotifyMemBuffer(index), NORMAL_NOTIFY_VAL, stream);
190 :
191 0 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
192 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();;
193 0 : taskParam.taskPara.Notify.notifyID = rmtNotifyVec[index]->GetAddr();
194 0 : taskParam.taskPara.Notify.value = NORMAL_NOTIFY_VAL;
195 :
196 0 : SaveDfxTaskInfo(taskParam);
197 1 : }
198 :
199 1 : void UbMemTransport::Wait(u32 index, const Stream &stream, u32 timeout)
200 : {
201 1 : TaskParam taskParam {};
202 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
203 :
204 1 : commonLocRes.notifyVec[index]->Wait(stream, timeout);
205 :
206 1 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
207 1 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
208 1 : taskParam.taskPara.Notify.notifyID = commonLocRes.notifyVec[index]->GetNotify()->GetId();
209 1 : taskParam.taskPara.Notify.value = NORMAL_NOTIFY_VAL;
210 :
211 1 : SaveDfxTaskInfo(taskParam);
212 1 : }
213 :
214 1 : void UbMemTransport::Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
215 : {
216 1 : TaskParam taskParam {};
217 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
218 :
219 1 : SqeConfig config;
220 1 : config.wqeMode = WqeMode::DWQE;
221 2 : SubmitUbTask(commonLocRes.connVec[0]->PrepareRead(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config),
222 : stream);
223 :
224 0 : taskParam.taskType = TaskParamType::TASK_RDMA;
225 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
226 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
227 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
228 0 : taskParam.taskPara.DMA.size = rmtSlice.size;
229 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
230 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
231 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
232 0 : SaveDfxTaskInfo(taskParam);
233 1 : }
234 :
235 1 : void UbMemTransport::ReadReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
236 : const ReduceIn &reduceIn, const Stream &stream)
237 : {
238 1 : TaskParam taskParam {};
239 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
240 :
241 1 : SqeConfig config;
242 1 : config.wqeMode = WqeMode::DWQE;
243 2 : SubmitUbTask(commonLocRes.connVec[0]->PrepareReadReduce(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice),
244 : reduceIn.dataType, reduceIn.reduceOp, config),
245 : stream);
246 :
247 0 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
248 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
249 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
250 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
251 0 : taskParam.taskPara.DMA.size = rmtSlice.size;
252 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
253 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
254 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
255 :
256 0 : SaveDfxTaskInfo(taskParam);
257 1 : }
258 :
259 1 : void UbMemTransport::Write(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
260 : {
261 1 : TaskParam taskParam {};
262 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
263 :
264 1 : SqeConfig config;
265 1 : config.wqeMode = WqeMode::DWQE;
266 2 : SubmitUbTask(commonLocRes.connVec[0]->PrepareWrite(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config),
267 : stream);
268 0 : taskParam.taskType = TaskParamType::TASK_RDMA;
269 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
270 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
271 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
272 0 : taskParam.taskPara.DMA.size = locSlice.size;
273 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
274 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
275 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
276 :
277 0 : SaveDfxTaskInfo(taskParam);
278 1 : }
279 :
280 1 : void UbMemTransport::WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
281 : const ReduceIn &reduceIn, const Stream &stream)
282 : {
283 1 : TaskParam taskParam {};
284 1 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
285 :
286 1 : SqeConfig config;
287 1 : config.wqeMode = WqeMode::DWQE;
288 2 : SubmitUbTask(commonLocRes.connVec[0]->PrepareWriteReduce(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice),
289 : reduceIn.dataType, reduceIn.reduceOp, config),
290 : stream);
291 :
292 0 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
293 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
294 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(locSlice.addr);
295 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmtSlice.addr);
296 0 : taskParam.taskPara.DMA.size = locSlice.size;
297 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
298 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
299 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
300 :
301 0 : SaveDfxTaskInfo(taskParam);
302 1 : }
303 :
304 5 : void UbMemTransport::WriteWithNotify(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
305 : const WithNotifyIn &withNotify, const Stream &stream)
306 : {
307 5 : if (locSlice.size == 0) {
308 2 : return SubmitWriteEmptyWithNotify(withNotify, stream);
309 : }
310 :
311 3 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
312 1 : return SubmitWriteWithNotify(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), NORMAL_NOTIFY_VAL,
313 2 : GetRmtNotifyMemBuffer(withNotify.index_), stream);
314 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
315 1 : return SubmitWriteWithNotify(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), withNotify.userData_,
316 2 : GetRmtCntNotifyMemBuffer(withNotify), stream);
317 : } else {
318 1 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
319 1 : THROW<InternalException>(msg);
320 1 : }
321 : }
322 :
323 5 : void UbMemTransport::WriteReduceWithNotify(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
324 : const ReduceIn &reduceIn, const WithNotifyIn &withNotify,
325 : const Stream &stream)
326 : {
327 5 : if (locSlice.size == 0) {
328 2 : return SubmitWriteEmptyWithNotify(withNotify, stream);
329 : }
330 :
331 3 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
332 1 : SubmitWriteReduceWithNotify(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn, NORMAL_NOTIFY_VAL,
333 2 : GetRmtNotifyMemBuffer(withNotify.index_), stream);
334 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
335 1 : SubmitWriteReduceWithNotify(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), reduceIn, withNotify.userData_,
336 2 : GetRmtCntNotifyMemBuffer(withNotify), stream);
337 : } else {
338 1 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
339 1 : THROW<InternalException>(msg);
340 1 : }
341 : }
342 :
343 4 : void UbMemTransport::SubmitWriteEmptyWithNotify(const WithNotifyIn &withNotify, const Stream &stream)
344 : {
345 4 : TaskParam taskParam {};
346 4 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
347 4 : u32 value = NORMAL_NOTIFY_VAL;
348 :
349 4 : if (withNotify.notifyType_ == TransportNotifyType::NORMAL) {
350 2 : SubmitNotify(GetRmtNotifyMemBuffer(withNotify.index_), NORMAL_NOTIFY_VAL, stream);
351 2 : } else if (withNotify.notifyType_ == TransportNotifyType::COUNT) {
352 2 : SubmitNotify(GetRmtCntNotifyMemBuffer(withNotify), withNotify.userData_, stream);
353 0 : value = withNotify.userData_;
354 : } else {
355 0 : std::string msg = StringFormat("%s error", withNotify.Describe().c_str());
356 0 : THROW<InternalException>(msg);
357 0 : }
358 :
359 0 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
360 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
361 0 : taskParam.taskPara.Notify.notifyID = INVALID_VALUE_NOTIFYID;
362 0 : taskParam.taskPara.Notify.value = value;
363 :
364 0 : SaveDfxTaskInfo(taskParam);
365 4 : }
366 :
367 2 : void UbMemTransport::SubmitWriteWithNotify(const MemoryBuffer &rmt, const MemoryBuffer &loc, u64 data,
368 : const MemoryBuffer &rmtNotify, const Stream &stream)
369 : {
370 2 : TaskParam taskParam {};
371 2 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
372 :
373 2 : SqeConfig config;
374 2 : config.wqeMode = WqeMode::DWQE;
375 4 : SubmitUbTask(commonLocRes.connVec[0]->PrepareWriteWithNotify(rmt, loc, data, rmtNotify, config), stream);
376 :
377 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
378 0 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
379 0 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(loc.addr);
380 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmt.addr);
381 0 : taskParam.taskPara.DMA.size = loc.size;
382 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
383 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
384 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
385 :
386 0 : SaveDfxTaskInfo(taskParam);
387 2 : }
388 :
389 2 : void UbMemTransport::SubmitWriteReduceWithNotify(const MemoryBuffer &rmt, const MemoryBuffer &loc,
390 : const ReduceIn &reduceIn, u64 data, const MemoryBuffer &rmtNotify,
391 : const Stream &stream)
392 : {
393 2 : TaskParam taskParam {};
394 2 : taskParam.beginTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
395 :
396 2 : SqeConfig config;
397 2 : config.wqeMode = WqeMode::DWQE;
398 2 : SubmitUbTask(commonLocRes.connVec[0]->PrepareWriteReduceWithNotify(rmt, loc, reduceIn.dataType, reduceIn.reduceOp,
399 : data, rmtNotify, config),
400 : stream);
401 :
402 2 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
403 2 : taskParam.endTime = DlProfFunc::GetInstance().dlMsprofSysCycleTime();
404 2 : taskParam.taskPara.DMA.src = reinterpret_cast<const void*>(loc.addr);
405 2 : taskParam.taskPara.DMA.dst = reinterpret_cast<const void*>(rmt.addr);
406 2 : taskParam.taskPara.DMA.size = loc.size;
407 2 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
408 2 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
409 2 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
410 :
411 2 : SaveDfxTaskInfo(taskParam);
412 2 : }
413 :
414 5 : bool UbMemTransport::IsResReady()
415 : {
416 7 : for (auto &it : commonLocRes.connVec) {
417 3 : CHECK_NULLPTR(it,
418 6 : StringFormat("[UbMemTransport::%s] failed, connection pointer is nullptr", __func__));
419 :
420 3 : RmaConnType connType = it->GetRmaConnType();
421 3 : if (connType != RmaConnType::UB) {
422 0 : THROW<InternalException>("[UbMemTransport::%s] connection type[%s] is not ub",
423 0 : __func__, connType.Describe().c_str());
424 : }
425 :
426 3 : auto status = it->GetStatus();
427 6 : if (status != RmaConnStatus::EXCHANGEABLE &&
428 3 : status != RmaConnStatus::READY) {
429 1 : return false;
430 : }
431 : }
432 :
433 12 : HCCL_INFO("[UbMemTransport::IsResReady] all resources ready.");
434 4 : return true;
435 : }
436 :
437 4 : bool UbMemTransport::IsConnsReady()
438 : {
439 4 : for (u32 i = 0; i < connNum; i++) {
440 1 : if (commonLocRes.connVec[i]->GetStatus() != RmaConnStatus::READY) {
441 1 : return false;
442 : }
443 : }
444 9 : HCCL_INFO("conns are ready.");
445 3 : return true;
446 : }
447 :
448 15 : HcclResult UbMemTransport::StatusMachine()
449 : {
450 18 : TRY_CATCH_RETURN(
451 : if (socket == nullptr) {
452 : HCCL_ERROR("[UbMemTransport][StatusMachine]socket is nullptr, please check");
453 : return HcclResult::HCCL_E_INTERNAL;
454 : }
455 : SocketStatus socketStatus = isHost_ ? socket->GetStatus() : socket->GetAsyncStatus();
456 : if (socketStatus == Hccl::SocketStatus::INIT || socketStatus == Hccl::SocketStatus::TIMEOUT) {
457 : HCCL_ERROR("[UbMemTransport][StatusMachine]socket timeout or no link, please check");
458 : return HcclResult::HCCL_E_INTERNAL;
459 : }
460 :
461 : if (socketStatus != Hccl::SocketStatus::OK) {
462 : SaluSleep(ONE_MILLISECOND_OF_USLEEP); // 防止get sockets冲高CtrlCPU
463 : return HcclResult::HCCL_SUCCESS; // 操作成功，保持当前状态
464 : }
465 : switch (ubStatus) {
466 : case UbStatus::INIT:
467 : CHK_RET(HandleInitStatus());
468 : break;
469 : case UbStatus::SEND_DATA:
470 : CHK_RET(HandleSendAllStatus());
471 : break;
472 : case UbStatus::RECV_SIZE:
473 : CHK_RET(HandleRecvSizeStatus());
474 : break;
475 : case UbStatus::RECV_DATA:
476 : CHK_RET(HandleRecvDataStatus());
477 : break;
478 : case UbStatus::PROCESS_DATA:
479 : CHK_RET(HandleProcessDataStatus());
480 : break;
481 : case UbStatus::SEND_FIN:
482 : CHK_RET(HandleSendFinStatus());
483 : break;
484 : case UbStatus::RECV_FIN:
485 : CHK_RET(HandleRecvFinStatus());
486 : break;
487 : case UbStatus::SET_READY:
488 : CHK_RET(HandleSetReadyStatus());
489 : break;
490 : default:
491 : break;
492 : }
493 : );
494 14 : return HCCL_SUCCESS;
495 : }
496 :
497 4 : HcclResult UbMemTransport::HandleInitStatus()
498 : {
499 4 : ubStatus = isRecvFirst_ ? UbStatus::RECV_SIZE : UbStatus::SEND_DATA;
500 4 : baseStatus = TransportStatus::SOCKET_OK;
501 4 : return HCCL_SUCCESS;
502 : }
503 :
504 4 : HcclResult UbMemTransport::HandleSendAllStatus()
505 : {
506 4 : if (IsResReady()) {
507 3 : CHK_RET(SendAll());
508 3 : ubStatus = isRecvFirst_ ? UbStatus::PROCESS_DATA : UbStatus::RECV_SIZE;
509 : }
510 4 : return HCCL_SUCCESS;
511 : }
512 :
513 5 : HcclResult UbMemTransport::HandleRecvSizeStatus()
514 : {
515 8 : CHK_RET(RecvDataSize());
516 4 : ubStatus = UbStatus::RECV_DATA;
517 4 : return HCCL_SUCCESS;
518 : }
519 :
520 5 : HcclResult UbMemTransport::HandleRecvDataStatus()
521 : {
522 8 : CHK_RET(RecvExchangeData());
523 4 : ubStatus = isRecvFirst_ ? UbStatus::SEND_DATA : UbStatus::PROCESS_DATA;
524 4 : return HCCL_SUCCESS;
525 : }
526 :
527 5 : HcclResult UbMemTransport::HandleProcessDataStatus()
528 : {
529 5 : bool needSendFinish = false;
530 8 : CHK_RET(RecvDataProcess(needSendFinish));
531 4 : if (needSendFinish) {
532 2 : ubStatus = UbStatus::SEND_FIN;
533 : } else {
534 2 : SetBaseStatusReady();
535 2 : ubStatus = UbStatus::READY;
536 : }
537 4 : return HCCL_SUCCESS;
538 : }
539 :
540 4 : HcclResult UbMemTransport::HandleSendFinStatus()
541 : {
542 4 : if (IsConnsReady()) {
543 6 : CHK_RET(SendFinish());
544 2 : ubStatus = UbStatus::RECV_FIN;
545 : }
546 3 : return HCCL_SUCCESS;
547 : }
548 :
549 3 : HcclResult UbMemTransport::HandleRecvFinStatus()
550 : {
551 6 : CHK_RET(RecvFinish());
552 2 : ubStatus = UbStatus::SET_READY;
553 2 : return HCCL_SUCCESS;
554 : }
555 :
556 2 : HcclResult UbMemTransport::HandleSetReadyStatus()
557 : {
558 2 : SetBaseStatusReady();
559 2 : ubStatus = UbStatus::READY;
560 2 : return HCCL_SUCCESS;
561 : }
562 :
563 13 : TransportStatus UbMemTransport::GetStatus()
564 : {
565 13 : if (baseStatus == TransportStatus::READY
566 13 : || baseStatus == TransportStatus::CONNECT_FAILED
567 26 : || baseStatus == TransportStatus::SOCKET_TIMEOUT) {
568 0 : return baseStatus;
569 : }
570 :
571 13 : HcclResult ret = StatusMachine();
572 13 : if (ret != HCCL_SUCCESS) {
573 0 : HCCL_ERROR("[UbMemTransport::GetStatus] StatusMachine failed, ret=%d", ret);
574 0 : baseStatus = TransportStatus::CONNECT_FAILED;
575 : }
576 13 : return baseStatus;
577 : }
578 :
579 1 : HcclResult UbMemTransport::SendAll()
580 : {
581 1 : notifyNum = commonLocRes.notifyVec.size();
582 1 : bufferNum = commonLocRes.bufferVec.size();
583 1 : connNum = commonLocRes.connVec.size();
584 1 : cntNotifyNum = locCntNotifyRes.vec.size();
585 :
586 1 : cntNotifyDescSize = locCntNotifyRes.desc.size();
587 :
588 3 : HCCL_INFO("notifyNum=%u, bufferNum=%u, connNum=%u, cntNotifyNum=%u, cntNotifyDescSize=%u, isHost_[%d]",
589 : notifyNum, bufferNum, connNum, cntNotifyNum, cntNotifyDescSize, isHost_);
590 :
591 1 : BinaryStream binaryStream;
592 1 : HandshakeMsgPack(binaryStream);
593 1 : NotifyVecPack(binaryStream);
594 1 : BufferVecPack(binaryStream, commonLocRes.bufferVec);
595 1 : CntNotifyVecPack(binaryStream);
596 1 : CntNotifyDescPack(binaryStream);
597 1 : CHK_RET(DrainBufPack(binaryStream));
598 1 : ConnVecPack(binaryStream);
599 :
600 1 : sendDataPack_.resize(sizeof(u32));
601 1 : binaryStream.Dump(sendDataPack_);
602 1 : u32 dataSize = sendDataPack_.size() - sizeof(u32);
603 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sendDataPack_.data(), sizeof(u32), &dataSize, sizeof(u32)));
604 :
605 1 : bool ret = false;
606 1 : if (isHost_) {
607 0 : ret = socket->Send(sendDataPack_.data(), sendDataPack_.size());
608 : } else {
609 1 : socket->SendAsync(sendDataPack_.data(), sendDataPack_.size());
610 1 : ret = true;
611 : }
612 1 : if (!ret) {
613 0 : HCCL_ERROR("[UbMemTransport::SendAll] Send failed");
614 0 : return HCCL_E_INTERNAL;
615 : }
616 3 : HCCL_INFO("[UbMemTransport::%s] Send size[%zu] of data success.", __func__, sendDataPack_.size());
617 1 : return HCCL_SUCCESS;
618 1 : }
619 :
620 1 : HcclResult UbMemTransport::RecvDataSize()
621 : {
622 : // 接收数据包尺寸
623 1 : bool ret = false;
624 3 : HCCL_DEBUG("Starting to recv message size[%zu] bytes, isHost_[%d]", sizeof(exchangeDataSize), isHost_);
625 1 : if (isHost_) {
626 0 : ret = socket->Recv(&exchangeDataSize, sizeof(exchangeDataSize));
627 : } else {
628 1 : socket->RecvAsync(reinterpret_cast<u8 *>(&exchangeDataSize), sizeof(exchangeDataSize));
629 1 : ret = true;
630 : }
631 1 : if (!ret) {
632 0 : HCCL_ERROR("[UbMemTransport::RecvDataSize] Recv size failed");
633 0 : return HCCL_E_INTERNAL;
634 : }
635 3 : HCCL_INFO("[UbMemTransport::%s] Receive size[%u] of data success. [%zu] bytes received.",
636 : __func__, exchangeDataSize, sizeof(exchangeDataSize));
637 1 : return HCCL_SUCCESS;
638 : }
639 :
640 1 : HcclResult UbMemTransport::SendExchangeData()
641 : {
642 1 : bool ret = false;
643 3 : HCCL_DEBUG("Starting to send message size[%zu] bytes, isHost_[%d]", sendData.size(), isHost_);
644 1 : if (isHost_) {
645 0 : ret = socket->Send(sendData.data(), sendData.size());
646 : } else {
647 1 : socket->SendAsync(sendData.data(), sendData.size());
648 1 : ret = true;
649 : }
650 1 : if (!ret) {
651 0 : HCCL_ERROR("[UbMemTransport::SendExchangeData] Send data failed");
652 0 : return HCCL_E_INTERNAL;
653 : }
654 3 : HCCL_INFO("send data %s, size=%zu", GetLinkDescInfo().c_str(), sendData.size());
655 1 : return HCCL_SUCCESS;
656 : }
657 :
658 0 : HcclResult UbMemTransport::RecvExchangeData()
659 : {
660 0 : recvData.resize(exchangeDataSize);
661 0 : bool ret = false;
662 0 : HCCL_DEBUG("Starting to recv message size[%zu] bytes, isHost_[%d]", recvData.size(), isHost_);
663 0 : if (isHost_) {
664 0 : ret = socket->Recv(recvData.data(), recvData.size());
665 : } else {
666 0 : socket->RecvAsync(reinterpret_cast<u8 *>(recvData.data()), recvData.size());
667 0 : ret = true;
668 : }
669 0 : if (!ret) {
670 0 : HCCL_ERROR("[UbMemTransport::RecvExchangeData] Recv data failed");
671 0 : return HCCL_E_INTERNAL;
672 : }
673 :
674 0 : HCCL_INFO("recv data %s, size=%zu", GetLinkDescInfo().c_str(), recvData.size());
675 0 : return HCCL_SUCCESS;
676 : }
677 :
678 0 : HcclResult UbMemTransport::RecvDataProcess(bool &needSendFinish)
679 : {
680 0 : HCCL_INFO("RecvDataProcess: link=%s, size=%zu, exchangeDataSize=%u", GetLinkDescInfo().c_str(), recvData.size(),
681 : exchangeDataSize);
682 0 : BinaryStream binaryStream(recvData);
683 0 : HcclResult ret = HandshakeMsgUnpack(binaryStream);
684 0 : if (ret != HCCL_SUCCESS) {
685 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] HandshakeMsgUnpack failed, ret=%d", ret);
686 0 : return ret;
687 : }
688 :
689 0 : ret = RmtBufferVecUnpackProc(notifyNum, binaryStream, rmtNotifyVec, UbRmtBufType::NOTIFY);
690 0 : if (ret != HCCL_SUCCESS) {
691 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc notify failed, ret=%d", ret);
692 0 : return ret;
693 : }
694 :
695 0 : ret = RmtBufferVecUnpackProc(bufferNum, binaryStream, rmtBufferVec, UbRmtBufType::BUFFER);
696 0 : if (ret != HCCL_SUCCESS) {
697 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc buffer failed, ret=%d", ret);
698 0 : return ret;
699 : }
700 :
701 0 : ret = RmtBufferVecUnpackProc(cntNotifyNum, binaryStream, rmtCntNotifyVec, UbRmtBufType::CNT_NOTIFY);
702 0 : if (ret != HCCL_SUCCESS) {
703 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] RmtBufferVecUnpackProc cntNotify failed, ret=%d", ret);
704 0 : return ret;
705 : }
706 :
707 0 : ret = CntNotifyDescUnpack(binaryStream);
708 0 : if (ret != HCCL_SUCCESS) {
709 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] CntNotifyDescUnpack failed, ret=%d", ret);
710 0 : return ret;
711 : }
712 :
713 0 : ret = DrainBufUnpack(binaryStream);
714 0 : if (ret != HCCL_SUCCESS) {
715 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] DrainBufUnpack failed, ret=%d", ret);
716 0 : return ret;
717 : }
718 :
719 0 : ret = ConnVecUnpackProc(binaryStream, needSendFinish);
720 0 : if (ret != HCCL_SUCCESS) {
721 0 : HCCL_ERROR("[UbMemTransport::RecvDataProcess] ConnVecUnpackProc failed, ret=%d", ret);
722 0 : return ret;
723 : }
724 :
725 0 : return HCCL_SUCCESS;
726 0 : }
727 :
728 6 : void UbMemTransport::BufferVecPack(BinaryStream &binaryStream, std::vector<LocalRmaBuffer *> &bufferVec)
729 : {
730 6 : binaryStream << static_cast<u32>(bufferVec.size());
731 18 : HCCL_INFO("start pack %s bufferVec", transportType.Describe().c_str());
732 6 : u32 pos = 0;
733 13 : for (auto &it : bufferVec) {
734 7 : binaryStream << pos;
735 7 : if (it != nullptr) { // 非空的buffer，从buffer中获取 dto
736 7 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
737 7 : dto->Serialize(binaryStream);
738 21 : HCCL_INFO("pack buffer pos=%u dto %s", pos, dto->Describe().c_str());
739 7 : } else { // 空的buffer，dto所有字段为0(size=0)
740 0 : ExchangeUbBufferDto exchangeDto;
741 0 : exchangeDto.Serialize(binaryStream);
742 0 : HCCL_INFO("pack buffer pos=%u, dto is null %s", pos, exchangeDto.Describe().c_str());
743 0 : }
744 7 : pos++;
745 : }
746 6 : }
747 :
748 1 : void UbMemTransport::CntNotifyVecPack(BinaryStream &binaryStream)
749 : {
750 1 : binaryStream << cntNotifyNum;
751 3 : HCCL_INFO("pack UB cntNotify num=%u, %s", cntNotifyNum, GetLinkDescInfo().c_str());
752 1 : u32 pos = 0;
753 2 : for (auto &it : locCntNotifyRes.vec) {
754 1 : binaryStream << pos;
755 1 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
756 1 : dto->Serialize(binaryStream);
757 3 : HCCL_INFO("pack cntNotify pos=%u, dto %s", pos, dto->Describe().c_str());
758 1 : pos++;
759 1 : }
760 1 : }
761 :
762 1 : void UbMemTransport::CntNotifyDescPack(BinaryStream &binaryStream)
763 : {
764 1 : binaryStream << cntNotifyDescSize;
765 3 : HCCL_INFO("pack cntNotify desc size=%u %s", cntNotifyDescSize, GetLinkDescInfo().c_str());
766 3 : HCCL_INFO("pack cntNotify desc =%s", Bytes2hex(locCntNotifyRes.desc.data(), locCntNotifyRes.desc.size()).c_str());
767 3 : for (auto &it : locCntNotifyRes.desc) {
768 2 : binaryStream << it;
769 : }
770 1 : }
771 :
772 1 : HcclResult UbMemTransport::DrainBufPack(BinaryStream &binaryStream)
773 : {
774 : // 打包交换信息前，进行资源创建
775 1 : CHK_RET(BuildDrainResource());
776 :
777 : // 只需交换 常量buffer信息 供远端读
778 3 : HCCL_INFO("start pack drain buffer");
779 1 : if (drainBuffer_ != nullptr) { // 非空的buffer，从buffer中获取 dto
780 1 : std::unique_ptr<Serializable> dto = drainBuffer_->GetExchangeDto();
781 1 : dto->Serialize(binaryStream);
782 3 : HCCL_INFO("pack drain buffer dto %s", dto->Describe().c_str());
783 1 : } else { // 空的buffer，dto所有字段为0(size=0)
784 0 : ExchangeUbBufferDto exchangeDto;
785 0 : exchangeDto.Serialize(binaryStream);
786 0 : HCCL_INFO("pack drain buffer, dto is null %s", exchangeDto.Describe().c_str());
787 0 : }
788 :
789 1 : return HCCL_SUCCESS;
790 : }
791 :
792 0 : HcclResult UbMemTransport::DrainBufUnpack(BinaryStream &binaryStream)
793 : {
794 0 : HCCL_INFO("start unpack drain buffer");
795 0 : ExchangeUbBufferDto dto;
796 0 : dto.Deserialize(binaryStream);
797 :
798 0 : if (dto.size == 0) {
799 0 : rmtDrainBuffer_ = nullptr;
800 0 : HCCL_WARNING("unpack drain buffer dto is null");
801 : } else {
802 0 : rmtDrainBuffer_ = std::make_unique<RemoteUbRmaBuffer>(rdmaHandle, dto);
803 0 : HCCL_INFO("unpack drain buffer rmtDrainBuffer=%s", rmtDrainBuffer_->Describe().c_str());
804 : }
805 :
806 0 : return HCCL_SUCCESS;
807 0 : }
808 :
809 0 : HcclResult UbMemTransport::CntNotifyDescUnpack(BinaryStream &binaryStream)
810 : {
811 : u32 descSize;
812 0 : binaryStream >> descSize;
813 0 : if (descSize != cntNotifyDescSize) {
814 0 : HCCL_ERROR("[UbMemTransport::CntNotifyDescUnpack] size=%u is not equal to rmtNum=%u", descSize, cntNotifyDescSize);
815 0 : return HCCL_E_PARA;
816 : }
817 0 : rmtCntNotifyDesc.clear();
818 0 : u32 pos = 0;
819 0 : for (pos = 0; pos < descSize; pos++) {
820 : char c;
821 0 : binaryStream >> c;
822 0 : rmtCntNotifyDesc.push_back(c);
823 : }
824 0 : HCCL_INFO("unpack cntNotify Desc=%s", Bytes2hex(rmtCntNotifyDesc.data(), rmtCntNotifyDesc.size()).c_str());
825 0 : return HCCL_SUCCESS;
826 : }
827 :
828 3 : HcclResult UbMemTransport::RmtBufferVecUnpackProc(u32 locNum, BinaryStream &binaryStream, RemoteBufferVec &bufferVec,
829 : UbRmtBufType type)
830 : {
831 : u32 rmtNum;
832 3 : binaryStream >> rmtNum;
833 3 : if (UNLIKELY(type == UbRmtBufType::BUFFER && rmtNum > MAX_BUFFER_NUM)) {
834 0 : HCCL_ERROR("[UbMemTransport][RmtBufferVecUnpackProc] rmtNum[%u] exceeds limit[%u]",
835 : rmtNum, MAX_BUFFER_NUM);
836 0 : return HCCL_E_PARA;
837 : }
838 :
839 : // 允许本端和远端交换内存数量不一致
840 9 : HCCL_INFO("unpack %s %s, locNum=%u, rmtNum=%u", type.Describe().c_str(), GetLinkDescInfo().c_str(), locNum,
841 : rmtNum);
842 :
843 7 : for (u32 i = 0; i < rmtNum; i++) {
844 : u32 pos;
845 4 : binaryStream >> pos;
846 4 : ExchangeUbBufferDto dto;
847 4 : dto.Deserialize(binaryStream);
848 4 : if (bufferVec.size() > pos) {
849 : // 对于之前已经加过的资源，无需追加
850 0 : continue;
851 : }
852 :
853 12 : HCCL_INFO("unpack %s pos=%u, dto %s", type.Describe().c_str(), pos, dto.Describe().c_str());
854 4 : if (dto.size == 0) { // size为0，则为 remote 空buffer
855 0 : HCCL_INFO("unpack nullptr, pos=%u", pos);
856 0 : bufferVec.push_back(nullptr);
857 0 : FillRmtRmaBufferVec(nullptr, type);
858 : } else { // size非0，则构造一个remote buffer
859 4 : bufferVec.push_back(make_unique<RemoteUbRmaBuffer>(rdmaHandle, dto));
860 4 : FillRmtRmaBufferVec(bufferVec.back().get(), type);
861 12 : HCCL_INFO("unpack buffer pos=%u, rmtRmaBuffer=%s", pos, bufferVec.back()->Describe().c_str());
862 : }
863 4 : }
864 :
865 3 : return HCCL_SUCCESS;
866 : }
867 :
868 1 : HcclResult UbMemTransport::ConnVecUnpackProc(BinaryStream &binaryStream, bool &needSendFinish)
869 : {
870 : u32 rmtConnNum;
871 1 : binaryStream >> rmtConnNum;
872 3 : HCCL_INFO("start unpack conn %s connNum=%u, rmtConnNum=%u", GetLinkDescInfo().c_str(), connNum, rmtConnNum);
873 1 : if (connNum != rmtConnNum) {
874 0 : HCCL_ERROR("[UbMemTransport::ConnVecUnpackProc] connNum=%u is not equal to rmtConnNum=%u", connNum, rmtConnNum);
875 0 : return HCCL_E_PARA;
876 : }
877 :
878 1 : needSendFinish = false; // 不需要发送 finish
879 2 : for (u32 i = 0; i < rmtConnNum; i++) {
880 : u32 pos;
881 1 : binaryStream >> pos;
882 1 : ExchangeUbConnDto rmtDto;
883 1 : rmtDto.Deserialize(binaryStream);
884 3 : HCCL_INFO("unpack connection pos=%u dto %s", pos, rmtDto.Describe().c_str());
885 1 : if (commonLocRes.connVec[i]->GetStatus() != RmaConnStatus::READY) {
886 0 : HCCL_INFO("parse and import pos=%u, rmt dto to connection[%s]", pos,
887 : commonLocRes.connVec[i]->Describe().c_str());
888 0 : commonLocRes.connVec[i]->ParseRmtExchangeDto(rmtDto);
889 0 : commonLocRes.connVec[i]->ImportRmtDto();
890 0 : needSendFinish = true; // connection 建链，需要发送finish
891 : }
892 1 : }
893 1 : return HCCL_SUCCESS;
894 : }
895 :
896 4 : void UbMemTransport::FillRmtRmaBufferVec(RemoteRmaBuffer *rmaBuffer, UbRmtBufType type)
897 : {
898 4 : if (type == UbRmtBufType::BUFFER) {
899 4 : rmtRmaBufferVec.push_back(rmaBuffer);
900 : }
901 4 : }
902 :
903 2 : HcclResult UbMemTransport::SendFinish()
904 : {
905 6 : HCCL_INFO("start send Finish Msg %s [%s], isHost_[%d]", GetLinkDescInfo().c_str(), FINISH_MSG, isHost_);
906 2 : sendFinishMsg = std::vector<char>(FINISH_MSG, FINISH_MSG + FINISH_MSG_SIZE);
907 2 : bool ret = false;
908 2 : if (isHost_) {
909 0 : ret = socket->Send(sendFinishMsg.data(), FINISH_MSG_SIZE);
910 : } else {
911 2 : socket->SendAsync(sendFinishMsg.data(), FINISH_MSG_SIZE);
912 1 : ret = true;
913 : }
914 1 : if (!ret) {
915 0 : HCCL_ERROR("[UbMemTransport::SendFinish] Send finish msg failed");
916 0 : return HCCL_E_INTERNAL;
917 : }
918 3 : HCCL_INFO("end send Finish Msg %s [%s]", GetLinkDescInfo().c_str(), FINISH_MSG);
919 1 : return HCCL_SUCCESS;
920 : }
921 :
922 1 : HcclResult UbMemTransport::RecvFinish()
923 : {
924 1 : recvFinishMsg.resize(FINISH_MSG_SIZE);
925 3 : HCCL_INFO("start recv Finish Msg %s [%s], isHost_[%d]", GetLinkDescInfo().c_str(), FINISH_MSG, isHost_);
926 1 : bool ret = false;
927 1 : if (isHost_) {
928 0 : ret = socket->Recv(recvFinishMsg.data(), FINISH_MSG_SIZE);
929 : } else {
930 1 : socket->RecvAsync(reinterpret_cast<u8 *>(recvFinishMsg.data()), FINISH_MSG_SIZE);
931 1 : ret = true;
932 : }
933 1 : if (!ret) {
934 0 : HCCL_ERROR("[UbMemTransport::RecvFinish] Recv finish msg failed");
935 0 : return HCCL_E_INTERNAL;
936 : }
937 3 : HCCL_INFO("end recv Finish Msg %s [%s]", GetLinkDescInfo().c_str(), FINISH_MSG);
938 1 : return HCCL_SUCCESS;
939 : }
940 :
941 45 : std::vector<char> UbMemTransport::GetUniqueId()
942 : {
943 45 : if (baseStatus != TransportStatus::READY) {
944 4 : MACRO_THROW(InternalException, StringFormat("transport status is not ready, please check"));
945 : }
946 44 : u32 type = static_cast<u32>(transportType);
947 44 : BinaryStream binaryStream;
948 44 : binaryStream << type;
949 44 : binaryStream << notifyNum;
950 44 : binaryStream << bufferNum;
951 44 : binaryStream << static_cast<u32>(rmtBufferVec.size());
952 44 : binaryStream << connNum;
953 :
954 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
955 44 : auto notifyUniqueIds = GetNotifyUniqueIds();
956 44 : binaryStream << notifyUniqueIds;
957 :
958 44 : auto rmtNotifyUniqueIds = GetRmtBufferUniqueIds(rmtNotifyVec, UbRmtBufType::NOTIFY);
959 44 : binaryStream << rmtNotifyUniqueIds;
960 :
961 44 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds(rmtBufferVec, UbRmtBufType::BUFFER);
962 44 : binaryStream << rmtBufferUniqueIds;
963 :
964 44 : auto connUniqueIds = GetConnUniqueIds();
965 44 : binaryStream << connUniqueIds;
966 :
967 44 : std::vector<char> result;
968 44 : binaryStream.Dump(result);
969 44 : return result;
970 44 : }
971 :
972 0 : std::vector<char> UbMemTransport::GetUniqueIdV2()
973 : {
974 0 : if (baseStatus != TransportStatus::READY) {
975 0 : MACRO_THROW(InternalException, StringFormat("transport status[%d] is not ready[%d], please check.",
976 : baseStatus, TransportStatus::READY));
977 : }
978 0 : u32 type = static_cast<u32>(transportType);
979 0 : BinaryStream binaryStream;
980 0 : binaryStream << type;
981 0 : binaryStream << notifyNum;
982 0 : binaryStream << bufferNum;
983 0 : binaryStream << static_cast<u32>(rmtBufferVec.size());
984 0 : binaryStream << connNum;
985 :
986 0 : auto notifyUniqueIds = GetNotifyUniqueIds();
987 0 : binaryStream << notifyUniqueIds;
988 :
989 0 : auto rmtNotifyUniqueIds = GetRmtBufferUniqueIds(rmtNotifyVec, UbRmtBufType::NOTIFY);
990 0 : binaryStream << rmtNotifyUniqueIds;
991 :
992 0 : for (auto &it : commonLocRes.bufferVec) {
993 0 : locBufferVec.emplace_back(reinterpret_cast<LocalUbRmaBuffer *>(it));
994 : }
995 :
996 0 : auto locBufferUniqueIds = GetLocBufferUniqueIds(locBufferVec, UbRmtBufType::BUFFER);
997 0 : binaryStream << locBufferUniqueIds;
998 :
999 0 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds(rmtBufferVec, UbRmtBufType::BUFFER);
1000 0 : binaryStream << rmtBufferUniqueIds;
1001 :
1002 0 : auto drainUniqueIds = GetDrainUniqueIds();
1003 0 : binaryStream << drainUniqueIds;
1004 :
1005 0 : auto connUniqueIds = GetConnUniqueIds();
1006 0 : binaryStream << connUniqueIds;
1007 :
1008 0 : std::vector<char> result;
1009 0 : binaryStream.Dump(result);
1010 0 : return result;
1011 0 : }
1012 :
1013 0 : std::vector<char> UbMemTransport::PackConnData()
1014 : {
1015 0 : if (baseStatus != TransportStatus::READY) {
1016 0 : MACRO_THROW(InternalException, StringFormat("transport status[%d] is not ready[%d], please check.",
1017 : baseStatus, TransportStatus::READY));
1018 : }
1019 0 : u32 type = static_cast<u32>(transportType);
1020 0 : BinaryStream binaryStream;
1021 0 : binaryStream << type;
1022 0 : binaryStream << connNum;
1023 :
1024 0 : auto connUniqueIds = GetConnUniqueIds();
1025 0 : binaryStream << connUniqueIds;
1026 :
1027 0 : std::vector<char> result;
1028 0 : binaryStream.Dump(result);
1029 0 : return result;
1030 0 : }
1031 :
1032 0 : std::vector<char> UbMemTransport::GetSingleRmtBufferUniqueId(u64 addr, u64 size, u32 tokenId, u32 tokenValue,
1033 : u32 notifyId) const
1034 : {
1035 0 : BinaryStream binaryStream;
1036 0 : binaryStream << addr;
1037 0 : binaryStream << size;
1038 0 : binaryStream << tokenId;
1039 0 : binaryStream << tokenValue;
1040 0 : binaryStream << notifyId;
1041 0 : HCCL_INFO("UbMemTransport RmtBuffer[addr=0x%llx, size=0x%llx, notifyId=%u]", addr, size, notifyId);
1042 0 : std::vector<char> result;
1043 0 : binaryStream.Dump(result);
1044 0 : return result;
1045 0 : }
1046 :
1047 44 : std::vector<char> UbMemTransport::GetNotifyUniqueIds()
1048 : {
1049 132 : HCCL_INFO("start packing all notify uniqueIds");
1050 44 : std::vector<char> result(0);
1051 86 : for (auto &it : commonLocRes.notifyVec) {
1052 126 : HCCL_INFO("ubMemTransport Notify %s", it->Describe().c_str());
1053 42 : auto uniqueId = it->GetUniqueId();
1054 42 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1055 42 : }
1056 44 : return result;
1057 0 : }
1058 :
1059 0 : std::vector<char> UbMemTransport::GetDrainUniqueIds() const
1060 : {
1061 0 : HCCL_INFO("start packing drain resources uniqueIds");
1062 0 : std::vector<char> result(0);
1063 0 : std::vector<char> uniqueId;
1064 :
1065 : // pack drain notify
1066 0 : if (drainNotify_ != nullptr) {
1067 0 : auto dto = drainNotify_->GetExchangeDto();
1068 0 : ExchangeUbBufferDto* rawDto = static_cast<ExchangeUbBufferDto*>(dto.get());
1069 0 : uniqueId = GetSingleRmtBufferUniqueId(rawDto->addr, rawDto->size, rawDto->tokenId, rawDto->tokenValue, rawDto->notifyId);
1070 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, %s", drainNotify_->Describe().c_str());
1071 0 : } else {
1072 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1073 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, drainNotify_ null buffer");
1074 : }
1075 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1076 :
1077 : // pack drain rmtMem
1078 0 : if (rmtDrainBuffer_ != nullptr) {
1079 0 : uniqueId = GetSingleRmtBufferUniqueId(rmtDrainBuffer_->GetAddr(), rmtDrainBuffer_->GetSize(),
1080 0 : rmtDrainBuffer_->GetTokenId(), rmtDrainBuffer_->GetTokenValue(), rmtDrainBuffer_->GetNotifyId());
1081 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, %s", rmtDrainBuffer_->Describe().c_str());
1082 : } else {
1083 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1084 0 : HCCL_INFO("UbMemTransport::GetDrainUniqueIds, rmtDrainBuffer_ null buffer");
1085 : }
1086 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1087 :
1088 0 : return result;
1089 0 : }
1090 :
1091 88 : std::vector<char> UbMemTransport::GetRmtBufferUniqueIds(RemoteBufferVec &bufferVec, UbRmtBufType type) const
1092 : {
1093 264 : HCCL_INFO("start packing all remote buffer %s uniqueIds", type.Describe().c_str());
1094 88 : std::vector<char> result(0);
1095 88 : for (auto &it : bufferVec) {
1096 0 : std::vector<char> uniqueId;
1097 0 : if (it != nullptr) {
1098 0 : uniqueId = GetSingleRmtBufferUniqueId(it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue(),
1099 0 : it->GetNotifyId());
1100 0 : HCCL_INFO("UbMemTransport::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
1101 : } else {
1102 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1103 0 : HCCL_INFO("UbMemTransport::GetRmtBufferUniqueIds, null buffer");
1104 : }
1105 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1106 0 : }
1107 88 : return result;
1108 0 : }
1109 :
1110 0 : std::vector<char> UbMemTransport::GetLocBufferUniqueIds(LocalBufferVec &bufferVec, UbRmtBufType type) const
1111 : {
1112 0 : HCCL_INFO("start packing all local buffer %s uniqueIds", type.Describe().c_str());
1113 0 : std::vector<char> result(0);
1114 0 : for (auto &it : bufferVec) {
1115 0 : std::vector<char> uniqueId;
1116 0 : if (it != nullptr) {
1117 0 : uniqueId = GetSingleRmtBufferUniqueId(it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue(),
1118 0 : UINT32_MAX);
1119 0 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, %s", it->Describe().c_str());
1120 : } else {
1121 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
1122 0 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, null buffer");
1123 : }
1124 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1125 0 : }
1126 0 : return result;
1127 0 : }
1128 :
1129 44 : std::vector<char> UbMemTransport::GetConnUniqueIds()
1130 : {
1131 132 : HCCL_INFO("start packing all conn uniqueIds");
1132 44 : std::vector<char> result(0);
1133 86 : for (auto &it : commonLocRes.connVec) {
1134 126 : HCCL_INFO("[UbMemTransport::%s] conn[%s]", __func__, it->Describe().c_str());
1135 42 : auto uniqueId = it->GetUniqueId();
1136 42 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1137 42 : }
1138 44 : return result;
1139 0 : }
1140 :
1141 3 : void UbMemTransport::SaveDfxTaskInfo(const TaskParam &taskParam)
1142 : {
1143 : u32 taskId;
1144 : u32 streamId;
1145 3 : HrtGetTaskIdAndStreamID(taskId, streamId);
1146 :
1147 3 : callback(streamId, taskId, taskParam);
1148 3 : }
1149 :
1150 2 : HcclResult UbMemTransport::GetRemoteMems(uint32_t *memNum, CommMem **remoteMem, char ***memInfos)
1151 : {
1152 2 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
1153 2 : Hccl::RemoteMemCtx<std::unique_ptr<RemoteUbRmaBuffer>> remoteMemCtx{cacheValid_, rmtBufferVec,
1154 2 : remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
1155 2 : CHK_RET(GetRemoteUserMems(remoteMemCtx));
1156 2 : return HCCL_SUCCESS;
1157 2 : }
1158 :
1159 5 : HcclResult UbMemTransport::CheckSocketStatus(std::string socketOpreator)
1160 : {
1161 5 : CHK_PTR_NULL(socket);
1162 5 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
1163 5 : auto startTime = std::chrono::steady_clock::now();
1164 5 : uint32_t retryCount = 0;
1165 : while(true) {
1166 5 : SocketStatus socketStatus = socket->GetAsyncStatus();
1167 5 : if (socketStatus == SocketStatus::OK) {
1168 4 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
1169 8 : std::chrono::steady_clock::now() - startTime).count();
1170 12 : HCCL_INFO("[UbMemTransport][%s] socket transport operation[%s] success, elapsed[%lld]ms, retryCount[%u]",
1171 : __func__, socketOpreator.c_str(), elapsed, retryCount);
1172 4 : break;
1173 : }
1174 2 : if ((std::chrono::steady_clock::now() - startTime) >= timeout ||
1175 1 : socketStatus == Hccl::SocketStatus::TIMEOUT) {
1176 1 : auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
1177 2 : std::chrono::steady_clock::now() - startTime).count();
1178 3 : HCCL_ERROR("[UbMemTransport][%s] socket transport operation[%s] timeout after %lld sec, elapsed[%lld]ms, retryCount[%u]",
1179 : __func__, socketOpreator.c_str(), timeout, elapsed, retryCount);
1180 1 : return HCCL_E_TIMEOUT;
1181 : }
1182 0 : retryCount++;
1183 0 : }
1184 4 : return HCCL_SUCCESS;
1185 : }
1186 :
1187 3 : HcclResult UbMemTransport::UpdateMemInfo(std::vector<LocalRmaBuffer *> &bufferVecTemp)
1188 : {
1189 3 : if (bufferVecTemp.size() == 0) {
1190 3 : HCCL_WARNING("[UbMemTransport][UpdateMemInfo] bufferNum is 0.");
1191 1 : return HCCL_SUCCESS;
1192 : }
1193 6 : HCCL_INFO("[UbMemTransport][UpdateMemInfo] bufferNum[%zu]", bufferVecTemp.size());
1194 2 : sendData.clear();
1195 2 : BinaryStream sendStream;
1196 2 : std::vector<std::unique_ptr<RemoteUbRmaBuffer>> rmtBufferTemp{};
1197 84 : TRY_CATCH_RETURN(
1198 : [&]() -> void {
1199 : BufferVecPack(sendStream, bufferVecTemp);
1200 : sendStream.Dump(sendData);
1201 : u32 sendSize = sendData.size();
1202 : socket->SendAsync(&sendSize, sizeof(sendSize));
1203 : HCCL_INFO("[UbMemTransport][UpdateMemInfo] Send size[%u] of data success. [%zu] bytes sent.",
1204 : sendSize, sizeof(sendSize));
1205 : HcclResult result = CheckSocketStatus("SendDataSize");
1206 : CHK_RET_THROW(InternalException,
1207 : StringFormat("[UbMemTransport][UpdateMemInfo] failed to send dataSize."),
1208 : result);
1209 : RecvDataSize();
1210 : result = CheckSocketStatus("RecvDataSize");
1211 : CHK_RET_THROW(InternalException,
1212 : StringFormat("[UbMemTransport][UpdateMemInfo] failed to receive dataSize."),
1213 : result);
1214 : SendExchangeData();
1215 : result = CheckSocketStatus("SendExchangeData");
1216 : CHK_RET_THROW(InternalException,
1217 : StringFormat("[UbMemTransport][UpdateMemInfo] failed to send data."),
1218 : result);
1219 : RecvExchangeData();
1220 : result = CheckSocketStatus("RecvExchangeData");
1221 : CHK_RET_THROW(InternalException,
1222 : StringFormat("[UbMemTransport][UpdateMemInfo] failed to receive data."),
1223 : result);
1224 : BinaryStream recvStream(recvData);
1225 : RmtBufferVecUnpackProc(bufferNum, recvStream, rmtBufferTemp, UbRmtBufType::BUFFER);
1226 : }());
1227 1 : rmtBufferVec.insert(rmtBufferVec.end(), std::make_move_iterator(rmtBufferTemp.begin()),
1228 : std::make_move_iterator(rmtBufferTemp.end()));
1229 1 : commonLocRes.bufferVec.insert(commonLocRes.bufferVec.end(), bufferVecTemp.begin(), bufferVecTemp.end());
1230 1 : cacheValid_ = false;
1231 1 : return HCCL_SUCCESS;
1232 2 : }
1233 :
1234 0 : HcclResult UbMemTransport::Init()
1235 : {
1236 0 : for (auto& ubConn : commonLocRes.connVec) {
1237 0 : TRY_CATCH_RETURN(ubConn->Connect());
1238 : }
1239 :
1240 0 : return HCCL_SUCCESS;
1241 : }
1242 :
1243 0 : HcclResult UbMemTransport::DeInit() const
1244 : {
1245 0 : socket->Destroy();
1246 0 : return HCCL_SUCCESS;
1247 : }
1248 :
1249 0 : HcclResult UbMemTransport::GetRemoteSeg(const void* addr, u64 len, u64 *seg)
1250 : {
1251 0 : if (rmtBufferVec.empty()) {
1252 0 : HCCL_ERROR("[UbMemTransport::%s] rmtBufferVec is empty.", __func__);
1253 0 : return HCCL_E_INTERNAL;
1254 : }
1255 :
1256 0 : bool isAddrInRange = false;
1257 0 : for (auto &it : rmtBufferVec) {
1258 0 : Buffer iterBuf(it->GetAddr(), it->GetSize());
1259 0 : if (iterBuf.Contains(reinterpret_cast<uintptr_t>(addr), len)) {
1260 0 : *seg = it->GetSegVa();
1261 0 : isAddrInRange = true;
1262 0 : break;
1263 : }
1264 0 : }
1265 :
1266 0 : if (!isAddrInRange) {
1267 0 : return HCCL_E_INTERNAL;
1268 : }
1269 0 : return HCCL_SUCCESS;
1270 : }
1271 :
1272 : } // namespace Hccl
|
